func logMessages(msgs []SagaMessage, log SagaLog) error {
	var err error
	if len(msgs) == 1 {
		err = log.LogMessage(msgs[0])
	} else {
		err = log.LogBatchMessages(msgs)
	}
	return err
}
logMessages durably stores the messages in sagalog
if the message(s) is invalid, returns an error and an unmodified saga state
package saga
import (
	"fmt"
	"sync"
	"github.com/twitter/scoot/common"
	"github.com/twitter/scoot/common/stats"
)
type Saga struct {
	id       string
	log      SagaLog
	state    *SagaState
	updateCh chan sagaUpdate
	mutex    sync.RWMutex  
	chMutex  sync.RWMutex  
	closed   bool
	stat     stats.StatsReceiver
}
func newSaga(sagaId string, job []byte, log SagaLog, stat stats.StatsReceiver) (*Saga, error) {
	state, err := makeSagaState(sagaId, job)
	if err != nil {
		return nil, err
	}
	err = log.StartSaga(sagaId, job)
	if err != nil {
		return nil, err
	}
	updateCh := make(chan sagaUpdate, common.DefaultSagaUpdateChSize)
	s := &Saga{
		id:       sagaId,
		log:      log,
		state:    state,
		updateCh: updateCh,
		mutex:    sync.RWMutex{},
		chMutex:  sync.RWMutex{},
		stat:     stat,
	}
	go s.updateSagaStateLoop()
	return s, nil
}
func rehydrateSaga(sagaId string, state *SagaState, log SagaLog, stat stats.StatsReceiver) *Saga {
	updateCh := make(chan sagaUpdate, common.DefaultSagaUpdateChSize)
	s := &Saga{
		id:       sagaId,
		log:      log,
		state:    state,
		updateCh: updateCh,
		mutex:    sync.RWMutex{},
		chMutex:  sync.RWMutex{},
		stat:     stat,
	}
	if !state.IsSagaCompleted() {
		go s.updateSagaStateLoop()
	}
	return s
}
func (s *Saga) ID() string {
	return s.id
}
func (s *Saga) GetState() *SagaState {
	s.mutex.RLock()
	defer s.mutex.RUnlock()
	return copySagaState(s.state)
}
func (s *Saga) EndSaga() error {
	return s.updateSagaState([]SagaMessage{MakeEndSagaMessage(s.id)})
}
func (s *Saga) AbortSaga() error {
	return s.updateSagaState([]SagaMessage{MakeAbortSagaMessage(s.id)})
}
func (s *Saga) StartTask(taskId string, data []byte) error {
	defer s.stat.Latency(stats.SagaStartOrEndTaskLatency_ms).Time().Stop()
	return s.updateSagaState([]SagaMessage{MakeStartTaskMessage(s.id, taskId, data)})
}
func (s *Saga) EndTask(taskId string, results []byte) error {
	defer s.stat.Latency(stats.SagaStartOrEndTaskLatency_ms).Time().Stop()
	return s.updateSagaState([]SagaMessage{MakeEndTaskMessage(s.id, taskId, results)})
}
func (s *Saga) StartCompensatingTask(taskId string, data []byte) error {
	return s.updateSagaState([]SagaMessage{MakeStartCompTaskMessage(s.id, taskId, data)})
}
func (s *Saga) EndCompensatingTask(taskId string, results []byte) error {
	return s.updateSagaState([]SagaMessage{MakeEndCompTaskMessage(s.id, taskId, results)})
}
func (s *Saga) BulkMessage(messages []SagaMessage) error {
	return s.updateSagaState(messages)
}
func (s *Saga) updateSagaState(msgs []SagaMessage) error {
	resultCh := make(chan error, 0)
	s.chMutex.RLock()
	if s.closed {
		return fmt.Errorf("Trying to update saga %s that has closed channel", s.id)
	}
	s.updateCh <- sagaUpdate{
		msgs:     msgs,
		resultCh: resultCh,
	}
	s.chMutex.RUnlock()
	result := <-resultCh
	for _, msg := range msgs {
		if msg.MsgType == EndSaga {
			s.chMutex.Lock()
			close(s.updateCh)
			s.closed = true
			s.chMutex.Unlock()
			break
		}
	}
	return result
}
func (s *Saga) updateSagaStateLoop() {
	var loopLatency = s.stat.Latency(stats.SagaUpdateStateLoopLatency_ms).Time()
	updates := []sagaUpdate{}
	for update := range s.updateCh {
		updates = append(updates, update)
		if len(s.updateCh) == 0 || len(updates) == common.DefaultSagaUpdateChSize {
			loopLatency.Stop()
			s.stat.Histogram(stats.SagaNumUpdatesProcessed).Update(int64(len(updates)))
			s.updateSaga(updates)
			updates = []sagaUpdate{}
			loopLatency.Time()
		}
	}
}
func (s *Saga) updateSaga(updates []sagaUpdate) {
	defer s.stat.Latency(stats.SagaUpdateStateLatency_ms).Time().Stop()
	s.mutex.Lock()
	defer s.mutex.Unlock()
	oldState := copySagaState(s.state)
	var err error
	var validUpdates []sagaUpdate
	var validUpdateMsgs []SagaMessage
	for _, update := range updates {
		err = nil
		if len(update.msgs) == 1 {
			err = updateSagaState(s.state, update.msgs[0])
		} else {
			s.state, err = bulkUpdateSagaState(s.state, update.msgs)
		}
		if err != nil {
			update.resultCh <- err
			continue
		}
		validUpdates = append(validUpdates, update)
		validUpdateMsgs = append(validUpdateMsgs, update.msgs...)
	}
	if len(validUpdateMsgs) == 0 {
		return
	}
	err = logMessages(validUpdateMsgs, s.log)
	if err != nil {
		s.state = oldState
	}
	for _, update := range validUpdates {
		update.resultCh <- err
	}
}
type sagaUpdate struct {
	msgs     []SagaMessage
	resultCh chan error
}