Saga Recovery

saga/inMemorySagaLog.go

@@ -27,45 +27,61 @@ func MakeInMemorySaga() Saga {
 }
 
 func (log *inMemorySagaLog) LogMessage(msg sagaMessage) error {
-	fmt.Println(fmt.Sprintf("Saga %s: %s %s", msg.sagaId, msg.msgType.String(), msg.taskId))
 
+	log.mutex.Lock()
+	defer log.mutex.Unlock()
+
+	fmt.Println(fmt.Sprintf("Saga %s: %s %s", msg.sagaId, msg.msgType.String(), msg.taskId))
 	sagaId := msg.sagaId
-	var err error
 
-	log.mutex.Lock()
 	msgs, ok := log.sagas[sagaId]
 	if !ok {
 		return errors.New(fmt.Sprintf("Saga: %s is not Started yet.", msg.sagaId))
 	}
 
 	log.sagas[sagaId] = append(msgs, msg)
-	log.mutex.Unlock()
-	return err
+	return nil
 }
 
 func (log *inMemorySagaLog) StartSaga(sagaId string, job []byte) error {
 
 	log.mutex.Lock()
+	defer log.mutex.Unlock()
 
 	fmt.Println(fmt.Sprintf("Start Saga %s", sagaId))
 
 	startMsg := MakeStartSagaMessage(sagaId, job)
 	log.sagas[sagaId] = []sagaMessage{startMsg}
 
-	log.mutex.Unlock()
-
 	return nil
 }
 
 func (log *inMemorySagaLog) GetMessages(sagaId string) ([]sagaMessage, error) {
 
 	log.mutex.RLock()
+	defer log.mutex.RUnlock()
+
 	msgs, ok := log.sagas[sagaId]
-	log.mutex.RUnlock()
 
 	if ok {
 		return msgs, nil
 	} else {
 		return nil, nil
 	}
 }
+
+/*
+ * Returns all Sagas Started since this InMemory Saga was created
+ */
+func (log *inMemorySagaLog) GetActiveSagas() ([]string, error) {
+	log.mutex.RLock()
+	defer log.mutex.RUnlock()
+
+	keys := make([]string, 0, len(log.sagas))
+
+	for key, _ := range log.sagas {
+		keys = append(keys, key)
+	}
+
+	return keys, nil
+}

saga/saga.go

@@ -1,22 +1,5 @@
 package saga
 
-type SagaRecoveryType int
-
-/*
- * Saga Recovery Types define how to interpret SagaState in RecoveryMode.
- *
- * ForwardRecovery: all tasks in the saga must be executed at least once.
- *                  tasks MUST BE idempotent
- *
- * RollbackRecovery: if Saga is Aborted or in unsafe state, compensating
- *                   tasks for all started tasks need to be executed.
- *                   compensating tasks MUST BE idempotent.
- */
-const (
-	BackwardRecovery SagaRecoveryType = iota
-	ForwardRecovery
-)
-
 /*
  * Saga Object which provides all Saga Functionality
  * Implementations of SagaLog should provide a factory method
@@ -56,26 +39,6 @@ func (s Saga) StartSaga(sagaId string, job []byte) (*SagaState, error) {
 	return state, nil
 }
 
-/*
- * logs the specified message durably to the SagaLog & updates internal state if its a valid state transition
- */
-func (s Saga) logMessage(state *SagaState, msg sagaMessage) (*SagaState, error) {
-
-	//verify that the applied message results in a valid state
-	newState, err := updateSagaState(state, msg)
-	if err != nil {
-		return nil, err
-	}
-
-	//try durably storing the message
-	err = s.log.LogMessage(msg)
-	if err != nil {
-		return nil, err
-	}
-
-	return newState, nil
-}
-
 /*
  * Log an End Saga Message to the log, returns updated SagaState
  * Returns the resulting SagaState or an error if it fails
@@ -148,3 +111,48 @@ func (s Saga) StartCompensatingTask(state *SagaState, taskId string, data []byte
 func (s Saga) EndCompensatingTask(state *SagaState, taskId string, results []byte) (*SagaState, error) {
 	return s.logMessage(state, MakeEndCompTaskMessage(state.sagaId, taskId, results))
 }
+
+/*
+ * Should be called at Saga Creation time.
+ * Returns a Slice of In Progress SagaIds
+ */
+func (s Saga) Startup() ([]string, error) {
+
+	ids, err := s.log.GetActiveSagas()
+	if err != nil {
+		return nil, err
+	}
+
+	return ids, nil
+}
+
+/*
+ * Recovers SagaState by reading all logged messages from the log.
+ * Utilizes the specified recoveryType to determine if Saga needs to be
+ * Aborted or can proceed safely.
+ *
+ * Returns the current SagaState
+ */
+func (s Saga) RecoverSagaState(sagaId string, recoveryType SagaRecoveryType) (*SagaState, error) {
+	return recoverState(sagaId, s, recoveryType)
+}
+
+/*
+ * logs the specified message durably to the SagaLog & updates internal state if its a valid state transition
+ */
+func (s Saga) logMessage(state *SagaState, msg sagaMessage) (*SagaState, error) {
+
+	//verify that the applied message results in a valid state
+	newState, err := updateSagaState(state, msg)
+	if err != nil {
+		return nil, err
+	}
+
+	//try durably storing the message
+	err = s.log.LogMessage(msg)
+	if err != nil {
+		return nil, err
+	}
+
+	return newState, nil
+}

saga/sagaRecovery.go

@@ -0,0 +1,101 @@
+package saga
+
+import (
+	"fmt"
+)
+
+type SagaRecoveryType int
+
+/*
+ * Saga Recovery Types define how to interpret SagaState in RecoveryMode.
+ *
+ * ForwardRecovery: all tasks in the saga must be executed at least once.
+ *                  tasks MUST BE idempotent
+ *
+ * RollbackRecovery: if Saga is Aborted or in unsafe state, compensating
+ *                   tasks for all started tasks need to be executed.
+ *                   compensating tasks MUST BE idempotent.
+ */
+const (
+	RollbackRecovery SagaRecoveryType = iota
+	ForwardRecovery
+)
+
+/*
+ * Recovers SagaState from SagaLog messages
+ */
+func recoverState(sagaId string, saga Saga, recoveryType SagaRecoveryType) (*SagaState, error) {
+
+	// Get Logged Messages For this Saga from the Log.
+	msgs, err := saga.log.GetMessages(sagaId)
+	if err != nil {
+		return nil, err
+	}
+
+	if msgs == nil || len(msgs) == 0 {
+		return nil, nil
+	}
+
+	// Reconstruct Saga State from Logged Messages
+	startMsg := msgs[0]
+	if startMsg.msgType != StartSaga {
+		return nil, fmt.Errorf("InvalidMessages: first message must be StartSaga")
+	}
+
+	state, err := makeSagaState(sagaId, startMsg.data)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, msg := range msgs {
+		state, err = updateSagaState(state, msg)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// Check if we can safely proceed forward based on recovery method
+	// RollbackRecovery must check if in a SafeState,
+	// ForwardRecovery can always make progress
+	switch recoveryType {
+
+	case RollbackRecovery:
+
+		// if Saga is not in a safe state we must abort the saga
+		// And compensating tasks should start
+		if !isSagaInSafeState(state) {
+			state, err = saga.AbortSaga(state)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+	case ForwardRecovery:
+
+	}
+
+	return state, nil
+}
+
+/*
+ * Returns true if saga is in a safe state, i.e. execution can pick up where
+ * it left off.  This is only used in RollbackRecovery
+ *
+ * A Saga is in a Safe State if all StartedTasks also have EndTask Messages
+ * A Saga is also in a Safe State if the Saga has been aborted and compensating
+ * actions have started to be applied.
+ */
+func isSagaInSafeState(state *SagaState) bool {
+
+	if state.IsSagaAborted() {
+		return true
+	}
+
+	for taskId, _ := range state.taskState {
+		if state.IsTaskStarted(taskId) && !state.IsTaskCompleted(taskId) {
+			return false
+		}
+	}
+
+	return true
+}