Store status for closed workflow executions (#164)

This is a first step to being able to query closed workflows by status in visibility APIs.
Issue #123

Author: Tamer Eldeeb

common/persistence/cassandraPersistence.go

@@ -71,6 +71,7 @@ const (
 		`decision_task_timeout: ?, ` +
 		`execution_context: ?, ` +
 		`state: ?, ` +
+		`close_status: ?, ` +
 		`next_event_id: ?, ` +
 		`last_processed_event: ?, ` +
 		`start_time: ?, ` +
@@ -645,6 +646,7 @@ func (d *cassandraPersistence) CreateWorkflowExecutionWithinBatch(request *Creat
 		request.DecisionTimeoutValue,
 		request.ExecutionContext,
 		WorkflowStateCreated,
+		WorkflowCloseStatusNone,
 		request.NextEventID,
 		request.LastProcessedEvent,
 		cqlNowTimestamp,
@@ -720,6 +722,7 @@ func (d *cassandraPersistence) UpdateWorkflowExecution(request *UpdateWorkflowEx
 		executionInfo.DecisionTimeoutValue,
 		executionInfo.ExecutionContext,
 		executionInfo.State,
+		executionInfo.CloseStatus,
 		executionInfo.NextEventID,
 		executionInfo.LastProcessedEvent,
 		executionInfo.StartTimestamp,
@@ -830,6 +833,7 @@ func (d *cassandraPersistence) DeleteWorkflowExecution(request *DeleteWorkflowEx
 		info.DecisionTimeoutValue,
 		info.ExecutionContext,
 		info.State,
+		info.CloseStatus,
 		info.NextEventID,
 		info.LastProcessedEvent,
 		info.StartTimestamp,

common/persistence/dataInterfaces.go

@@ -20,6 +20,17 @@ const (
 	WorkflowStateCompleted
 )
 
+// Workflow execution close status
+const (
+	WorkflowCloseStatusNone = iota
+	WorkflowCloseStatusCompleted
+	WorkflowCloseStatusFailed
+	WorkflowCloseStatusCanceled
+	WorkflowCloseStatusTerminated
+	WorkflowCloseStatusContinuedAsNew
+	WorkflowCloseStatusTimedOut
+)
+
 // Types of task lists
 const (
 	TaskListTypeDecision = iota
@@ -83,6 +94,7 @@ type (
 		DecisionTimeoutValue int32
 		ExecutionContext     []byte
 		State                int
+		CloseStatus          int
 		NextEventID          int64
 		LastProcessedEvent   int64
 		StartTimestamp       time.Time

schema/workflow_test.cql

@@ -25,6 +25,7 @@ CREATE TYPE workflow_execution (
   decision_task_timeout  int,
   execution_context      blob,
   state                  int,  -- enum WorkflowState {Created, Running, Completed}
+  close_status           int,  -- enum WorkflowCloseStatus {None, Completed, Failed, Canceled, Terminated, ContinuedAsNew, TimedOut}
   next_event_id          bigint,
   last_processed_event   bigint,
   start_time             timestamp,

service/history/mutableStateBuilder.go

@@ -69,6 +69,7 @@ func newMutableStateBuilder(logger bark.Logger) *mutableStateBuilder {
 	s.executionInfo = &persistence.WorkflowExecutionInfo{
 		NextEventID:        firstEventID,
 		State:              persistence.WorkflowStateCreated,
+		CloseStatus:        persistence.WorkflowCloseStatusNone,
 		LastProcessedEvent: emptyEventID,
 	}
 
@@ -330,6 +331,7 @@ func (e *mutableStateBuilder) AddWorkflowExecutionStartedEvent(domainID string,
 	e.executionInfo.DecisionTimeoutValue = request.GetTaskStartToCloseTimeoutSeconds()
 
 	e.executionInfo.State = persistence.WorkflowStateCreated
+	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusNone
 	e.executionInfo.LastProcessedEvent = emptyEventID
 	e.executionInfo.CreateRequestID = request.GetRequestId()
 	e.executionInfo.DecisionScheduleID = emptyEventID
@@ -591,6 +593,7 @@ func (e *mutableStateBuilder) AddCompletedWorkflowEvent(decisionCompletedEventID
 	}
 
 	e.executionInfo.State = persistence.WorkflowStateCompleted
+	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusCompleted
 	return e.hBuilder.AddCompletedWorkflowEvent(decisionCompletedEventID, attributes)
 }
 
@@ -602,6 +605,7 @@ func (e *mutableStateBuilder) AddFailWorkflowEvent(decisionCompletedEventID int6
 	}
 
 	e.executionInfo.State = persistence.WorkflowStateCompleted
+	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusFailed
 	return e.hBuilder.AddFailWorkflowEvent(decisionCompletedEventID, attributes)
 }
 
@@ -628,6 +632,7 @@ func (e *mutableStateBuilder) AddWorkflowExecutionCanceledEvent(decisionTaskComp
 	}
 
 	e.executionInfo.State = persistence.WorkflowStateCompleted
+	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusCanceled
 	return e.hBuilder.AddWorkflowExecutionCanceledEvent(decisionTaskCompletedEventID, attributes)
 }
 
@@ -735,6 +740,7 @@ func (e *mutableStateBuilder) AddWorkflowExecutionTerminatedEvent(
 	}
 
 	e.executionInfo.State = persistence.WorkflowStateCompleted
+	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusTerminated
 	return e.hBuilder.AddWorkflowExecutionTerminatedEvent(request)
 }
 
@@ -759,6 +765,7 @@ func (e *mutableStateBuilder) AddContinueAsNewEvent(decisionCompletedEventID int
 	}
 
 	e.executionInfo.State = persistence.WorkflowStateCompleted
+	e.executionInfo.CloseStatus = persistence.WorkflowCloseStatusContinuedAsNew
 	newExecution := workflow.WorkflowExecution{
 		WorkflowId: common.StringPtr(e.executionInfo.WorkflowID),
 		RunId:      common.StringPtr(newRunID),