Create TTL in the UpdateWorkflowExecution cycles.

common/dynamicconfig/constants.go

@@ -1840,6 +1840,13 @@ const (
 	EnableShardIDMetrics
 	// LastBoolKey must be the last one in this const group
 	LastBoolKey
+
+	// EnableExecutionTTL is which domains are allowed to have workflow executions with a TTL
+	// KeyName: system.enableExecutionTTL
+	// Value type: Bool
+	// Default value: false
+	// Allowed filters: DomainID
+	EnableExecutionTTL
 )
 
 const (
@@ -3930,6 +3937,11 @@ var BoolKeys = map[BoolKey]DynamicBool{
 		Description:  "Enable shardId metrics in persistence client",
 		DefaultValue: true,
 	},
+	EnableExecutionTTL: DynamicBool{
+		KeyName:      "system.enableExecutionTTL",
+		Description:  "EnableExecutionTTL is which domains are allowed to have workflow executions with a TTL",
+		DefaultValue: false,
+	},
 }
 
 var FloatKeys = map[FloatKey]DynamicFloat{

common/persistence/config.go

@@ -31,6 +31,7 @@ type (
 		EnableCassandraAllConsistencyLevelDelete dynamicconfig.BoolPropertyFn
 		PersistenceSampleLoggingRate             dynamicconfig.IntPropertyFn
 		EnableShardIDMetrics                     dynamicconfig.BoolPropertyFn
+		EnableExecutionTTL                       dynamicconfig.BoolPropertyFnWithDomainIDFilter
 	}
 )
 
@@ -41,5 +42,6 @@ func NewDynamicConfiguration(dc *dynamicconfig.Collection) *DynamicConfiguration
 		EnableCassandraAllConsistencyLevelDelete: dc.GetBoolProperty(dynamicconfig.EnableCassandraAllConsistencyLevelDelete),
 		PersistenceSampleLoggingRate:             dc.GetIntProperty(dynamicconfig.SampleLoggingRate),
 		EnableShardIDMetrics:                     dc.GetBoolProperty(dynamicconfig.EnableShardIDMetrics),
+		EnableExecutionTTL:                       dc.GetBoolPropertyFilteredByDomainID(dynamicconfig.EnableExecutionTTL),
 	}
 }

common/persistence/dataManagerInterfaces.go

@@ -1024,8 +1024,9 @@ type (
 		ReplicationTasks  []Task
 		TimerTasks        []Task
 
-		Condition int64
-		Checksum  checksum.Checksum
+		Condition    int64
+		TTLInSeconds int64
+		Checksum     checksum.Checksum
 	}
 
 	// WorkflowSnapshot is used as generic workflow execution state snapshot

common/persistence/dataStoreInterfaces.go

@@ -460,8 +460,8 @@ type (
 		CrossClusterTasks []Task
 		TimerTasks        []Task
 		ReplicationTasks  []Task
-
-		Condition int64
+		TTLInSeconds      int64
+		Condition         int64
 
 		Checksum checksum.Checksum
 	}

common/persistence/executionManager.go

@@ -329,7 +329,6 @@ func (m *executionManagerImpl) UpdateWorkflowExecution(
 			return nil, err
 		}
 	}
-
 	newRequest := &InternalUpdateWorkflowExecutionRequest{
 		RangeID: request.RangeID,
 
@@ -661,9 +660,9 @@ func (m *executionManagerImpl) SerializeWorkflowMutation(
 		CrossClusterTasks: input.CrossClusterTasks,
 		ReplicationTasks:  input.ReplicationTasks,
 		TimerTasks:        input.TimerTasks,
-
-		Condition: input.Condition,
-		Checksum:  input.Checksum,
+		TTLInSeconds:      input.TTLInSeconds,
+		Condition:         input.Condition,
+		Checksum:          input.Checksum,
 	}, nil
 }
 

common/persistence/nosql/nosqlExecutionStore.go

@@ -467,7 +467,6 @@ func (d *nosqlExecutionStore) ConflictResolveWorkflowExecution(
 		ShardID: d.shardID,
 		RangeID: request.RangeID,
 	}
-
 	err = d.db.UpdateWorkflowExecutionWithTasks(
 		ctx, currentWorkflowWriteReq,
 		mutateExecution, insertExecution, resetExecution,

common/persistence/nosql/nosqlExecutionStoreUtil.go

@@ -85,9 +85,10 @@ func (d *nosqlExecutionStore) prepareResetWorkflowExecutionRequestWithMapsAndEve
 	versionHistories := resetWorkflow.VersionHistories
 	nowTimestamp := time.Now()
 
+	//TTLInSeconds is invalid in this case so passing a default value of 0.
 	executionRequest, err := d.prepareUpdateWorkflowExecutionTxn(
 		executionInfo, versionHistories, checkSum,
-		nowTimestamp, lastWriteVersion,
+		nowTimestamp, lastWriteVersion, 0,
 	)
 	if err != nil {
 		return nil, err
@@ -127,13 +128,14 @@ func (d *nosqlExecutionStore) prepareUpdateWorkflowExecutionRequestWithMapsAndEv
 ) (*nosqlplugin.WorkflowExecutionRequest, error) {
 	executionInfo := workflowMutation.ExecutionInfo
 	lastWriteVersion := workflowMutation.LastWriteVersion
+	ttlInSeconds := workflowMutation.TTLInSeconds
 	checkSum := workflowMutation.Checksum
 	versionHistories := workflowMutation.VersionHistories
 	nowTimestamp := time.Now()
 
 	executionRequest, err := d.prepareUpdateWorkflowExecutionTxn(
 		executionInfo, versionHistories, checkSum,
-		nowTimestamp, lastWriteVersion,
+		nowTimestamp, lastWriteVersion, ttlInSeconds,
 	)
 	if err != nil {
 		return nil, err
@@ -607,6 +609,7 @@ func (d *nosqlExecutionStore) prepareUpdateWorkflowExecutionTxn(
 	checksum checksum.Checksum,
 	nowTimestamp time.Time,
 	lastWriteVersion int64,
+	ttlInSeconds int64,
 ) (*nosqlplugin.WorkflowExecutionRequest, error) {
 	// validate workflow state & close status
 	if err := p.ValidateUpdateWorkflowStateCloseStatus(
@@ -636,6 +639,7 @@ func (d *nosqlExecutionStore) prepareUpdateWorkflowExecutionTxn(
 		VersionHistories:              versionHistories,
 		Checksums:                     &checksum,
 		LastWriteVersion:              lastWriteVersion,
+		TTLInSeconds:                  ttlInSeconds,
 	}, nil
 }
 

common/persistence/nosql/nosqlplugin/cassandra/workflowCql.go

@@ -315,8 +315,18 @@ workflow_state = ? ` +
 		`WHERE shard_id = ? ` +
 		`and type = ?`
 
-	templateUpdateWorkflowExecutionWithVersionHistoriesQuery = `UPDATE executions ` +
-		`SET execution = ` + templateWorkflowExecutionType +
+	templateUpdateWorkflowExecutionWithVersionHistoriesQueryPart1 = `INSERT INTO executions (` +
+		`domain_id, ` +
+		`run_id, ` +
+		`shard_id, ` +
+		`task_id, ` +
+		`type, ` +
+		`visibility_ts, ` +
+		`workflow_id ` +
+		`) VALUES (?, ?, ?, ?, ?, ?, ?) USING TTL ?`
+
+	templateUpdateWorkflowExecutionWithVersionHistoriesQueryPart2 = `UPDATE executions ` +
+		`USING TTL ? SET execution = ` + templateWorkflowExecutionType +
 		`, next_event_id = ? ` +
 		`, version_histories = ? ` +
 		`, version_histories_encoding = ? ` +

common/persistence/nosql/nosqlplugin/cassandra/workflowUtils.go

@@ -1078,7 +1078,6 @@ func (db *cdb) resetWorkflowExecutionAndMapsAndEventBuffer(
 	if execution.MapsWriteMode != nosqlplugin.WorkflowExecutionMapsWriteModeReset {
 		return fmt.Errorf("should only support WorkflowExecutionMapsWriteModeReset")
 	}
-
 	err = db.resetActivityInfos(batch, shardID, domainID, workflowID, execution.RunID, execution.ActivityInfos)
 	if err != nil {
 		return err
@@ -1168,7 +1167,6 @@ func (db *cdb) updateWorkflowExecutionAndEventBufferWithMergeAndDeleteMaps(
 			return err
 		}
 	}
-
 	if execution.MapsWriteMode != nosqlplugin.WorkflowExecutionMapsWriteModeUpdate {
 		return fmt.Errorf("should only support WorkflowExecutionMapsWriteModeUpdate")
 	}
@@ -1205,8 +1203,26 @@ func (db *cdb) updateWorkflowExecution(
 ) error {
 	execution.StartTimestamp = db.convertToCassandraTimestamp(execution.StartTimestamp)
 	execution.LastUpdatedTimestamp = db.convertToCassandraTimestamp(execution.LastUpdatedTimestamp)
-
-	batch.Query(templateUpdateWorkflowExecutionWithVersionHistoriesQuery,
+	//default TTL Value. 0 TTL means no ttl is set, hence your records will persist forever unless explicitly deleted.
+	ttlInSeconds := 0
+	//Only fires when the workflow is closing.
+	if execution.CloseStatus == 1 {
+		if db.dc.EnableExecutionTTL(domainID) {
+			ttlInSeconds = int(execution.TTLInSeconds)
+		}
+		batch.Query(templateUpdateWorkflowExecutionWithVersionHistoriesQueryPart1,
+			domainID,
+			execution.RunID,
+			shardID,
+			rowTypeExecutionTaskID,
+			rowTypeExecution,
+			defaultVisibilityTimestamp,
+			workflowID,
+			ttlInSeconds,
+		)
+	}
+	batch.Query(templateUpdateWorkflowExecutionWithVersionHistoriesQueryPart2,
+		ttlInSeconds,
 		domainID,
 		workflowID,
 		execution.RunID,

common/persistence/nosql/nosqlplugin/types.go

@@ -39,6 +39,7 @@ type (
 		VersionHistories *persistence.DataBlob
 		Checksums        *checksum.Checksum
 		LastWriteVersion int64
+		TTLInSeconds     int64
 		// condition checking for updating execution info
 		PreviousNextEventIDCondition *int64
 

common/persistence/persistence-tests/executionManagerTest.go

@@ -609,6 +609,92 @@ func (s *ExecutionManagerSuite) TestUpdateWorkflowExecutionStateCloseStatus() {
 	}
 }
 
+// TestUpdateWorkflowExecutionTTL test
+func (s *ExecutionManagerSuite) TestUpdateWorkflowExecutionTTL() {
+	//TODO: Update the test when the TTLbuffer becomes configurable.
+	ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout)
+	defer cancel()
+	if s.ExecutionManager.GetName() != "cassandra" {
+		// TTL API is only supported in cassandra"
+		return
+	}
+
+	domainID := uuid.New()
+	workflowID := "update-workflow-test-with-ttl"
+	workflowExecution := types.WorkflowExecution{
+		WorkflowID: workflowID,
+		RunID:      uuid.New(),
+	}
+	tasklist := "some random tasklist"
+	workflowType := "some random workflow type"
+	workflowTimeout := int32(10)
+	decisionTimeout := int32(14)
+	lastProcessedEventID := int64(0)
+	nextEventID := int64(3)
+	csum := s.newRandomChecksum()
+	versionHistory := p.NewVersionHistory([]byte{}, []*p.VersionHistoryItem{
+		{
+			EventID: nextEventID,
+			Version: common.EmptyVersion,
+		},
+	})
+	versionHistories := p.NewVersionHistories(versionHistory)
+
+	// create and update a workflow to make it completed
+	req := &p.CreateWorkflowExecutionRequest{
+		NewWorkflowSnapshot: p.WorkflowSnapshot{
+			ExecutionInfo: &p.WorkflowExecutionInfo{
+				CreateRequestID:             uuid.New(),
+				DomainID:                    domainID,
+				WorkflowID:                  workflowExecution.GetWorkflowID(),
+				RunID:                       workflowExecution.GetRunID(),
+				FirstExecutionRunID:         workflowExecution.GetRunID(),
+				TaskList:                    tasklist,
+				WorkflowTypeName:            workflowType,
+				WorkflowTimeout:             workflowTimeout,
+				DecisionStartToCloseTimeout: decisionTimeout,
+				NextEventID:                 nextEventID,
+				LastProcessedEvent:          lastProcessedEventID,
+				State:                       p.WorkflowStateRunning,
+				CloseStatus:                 p.WorkflowCloseStatusNone,
+			},
+			ExecutionStats:   &p.ExecutionStats{},
+			Checksum:         csum,
+			VersionHistories: versionHistories,
+		},
+		RangeID: s.ShardInfo.RangeID,
+		Mode:    p.CreateWorkflowModeBrandNew,
+	}
+	_, err := s.ExecutionManager.CreateWorkflowExecution(ctx, req)
+	s.Nil(err)
+	currentRunID, err := s.GetCurrentWorkflowRunID(ctx, domainID, workflowID)
+	s.Nil(err)
+	s.Equal(workflowExecution.GetRunID(), currentRunID)
+
+	info, err := s.GetWorkflowExecutionInfo(ctx, domainID, workflowExecution)
+	s.Nil(err)
+
+	updatedInfo := copyWorkflowExecutionInfo(info.ExecutionInfo)
+	updateStats := copyExecutionStats(info.ExecutionStats)
+	updatedInfo.State = p.WorkflowStateCompleted
+	updatedInfo.CloseStatus = p.WorkflowCloseStatusCompleted
+	_, err = s.ExecutionManager.UpdateWorkflowExecution(ctx, &p.UpdateWorkflowExecutionRequest{
+		UpdateWorkflowMutation: p.WorkflowMutation{
+			ExecutionInfo:    updatedInfo,
+			ExecutionStats:   updateStats,
+			Condition:        nextEventID,
+			TTLInSeconds:     1,
+			VersionHistories: versionHistories,
+		},
+		RangeID: s.ShardInfo.RangeID,
+		Mode:    p.UpdateWorkflowModeUpdateCurrent,
+	})
+	s.NoError(err)
+	time.Sleep(2 * time.Second)
+	info, err = s.GetWorkflowExecutionInfo(ctx, domainID, workflowExecution)
+	s.IsType(&types.EntityNotExistsError{}, err)
+}
+
 // TestUpdateWorkflowExecutionWithZombieState test
 func (s *ExecutionManagerSuite) TestUpdateWorkflowExecutionWithZombieState() {
 	ctx, cancel := context.WithTimeout(context.Background(), testContextTimeout)

common/persistence/persistence-tests/persistenceTestBase.go

@@ -142,6 +142,7 @@ func NewTestBaseWithNoSQL(options *TestBaseOptions) TestBase {
 		EnableCassandraAllConsistencyLevelDelete: dynamicconfig.GetBoolPropertyFn(true),
 		PersistenceSampleLoggingRate:             dynamicconfig.GetIntPropertyFn(100),
 		EnableShardIDMetrics:                     dynamicconfig.GetBoolPropertyFn(true),
+		EnableExecutionTTL:                       dynamicconfig.GetBoolPropertyFnFilteredByDomainID(true),
 	}
 	params := TestBaseParams{
 		DefaultTestCluster:    testCluster,

config/dynamicconfig/development.yaml

@@ -7,6 +7,9 @@ system.minRetentionDays:
 history.EnableConsistentQueryByDomain:
 - value: true
   constraints: {}
+system.enableExecutionTTL:
+    - value: true
+      constraints: {}
 history.enableCrossClusterOperations:
 - value: true
   constraints: {}

service/history/execution/context.go

@@ -45,6 +45,8 @@ import (
 
 const (
 	defaultRemoteCallTimeout = 30 * time.Second
+	ttlBufferDays            = 15
+	dayToSecondMultiplier    = 86400
 )
 
 type conflictError struct {
@@ -1214,7 +1216,6 @@ func (c *contextImpl) updateWorkflowExecutionWithRetry(
 		resp, err = c.shard.UpdateWorkflowExecution(ctx, request)
 		return err
 	}
-
 	isRetryable := func(err error) bool {
 		if _, ok := err.(*persistence.TimeoutError); ok {
 			// timeout error is not retryable for update workflow execution

service/history/execution/mutable_state_builder.go

@@ -4066,6 +4066,11 @@ func (e *mutableStateBuilder) CloseTransactionAsMutation(
 	// impact the checksum calculation
 	checksum := e.generateChecksum()
 
+	TTLInSeconds, err := e.calculateTTL()
+	if err != nil {
+		e.logError("TTL calculation failed")
+	}
+
 	workflowMutation := &persistence.WorkflowMutation{
 		ExecutionInfo:    e.executionInfo,
 		VersionHistories: e.versionHistories,
@@ -4090,8 +4095,9 @@ func (e *mutableStateBuilder) CloseTransactionAsMutation(
 		ReplicationTasks:  e.insertReplicationTasks,
 		TimerTasks:        e.insertTimerTasks,
 
-		Condition: e.nextEventIDInDB,
-		Checksum:  checksum,
+		Condition:    e.nextEventIDInDB,
+		Checksum:     checksum,
+		TTLInSeconds: int64(TTLInSeconds),
 	}
 
 	e.checksum = checksum
@@ -4807,3 +4813,27 @@ func (e *mutableStateBuilder) logDataInconsistency() {
 		tag.WorkflowRunID(runID),
 	)
 }
+func (e *mutableStateBuilder) calculateTTL() (int, error) {
+	domainID := e.executionInfo.DomainID
+	//Calculating the TTL for workflow Execution.
+
+	domainObj, err := e.shard.GetDomainCache().GetDomainByID(domainID)
+	if err != nil {
+		return 0, err
+	}
+	config := domainObj.GetConfig()
+	retention := time.Duration(config.Retention)
+	daysInSeconds := int((retention + ttlBufferDays) * dayToSecondMultiplier)
+	//Default state of TTL, means there is no TTL attached.
+	TTLInSeconds := 0
+	startTime := e.executionInfo.StartTimestamp
+	//Handles Cron and Delaystart. For Cron workflows the StartTimestamp does not show up until the wf has started.
+	//default value os TTL ie. 0 will be passed down in this case. The TTL is calculated only if the startTime is non zero.
+	if !time.Time.IsZero(startTime) {
+		CalculateTTLInSeconds := int(e.executionInfo.WorkflowTimeout) - int(time.Now().Sub(startTime).Seconds()) + daysInSeconds
+		if CalculateTTLInSeconds >= 0 {
+			return CalculateTTLInSeconds, nil
+		}
+	}
+	return TTLInSeconds, nil
+}

service/history/execution/mutable_state_builder_test.go

@@ -97,6 +97,7 @@ func (s *mutableStateSuite) SetupTest() {
 	s.logger = s.mockShard.GetLogger()
 
 	s.mockShard.Resource.DomainCache.EXPECT().GetDomainID(constants.TestDomainName).Return(constants.TestDomainID, nil).AnyTimes()
+	s.mockShard.Resource.DomainCache.EXPECT().GetDomainByID(constants.TestDomainID).Return(constants.TestLocalDomainEntry, nil).AnyTimes()
 
 	s.msBuilder = newMutableStateBuilder(s.mockShard, s.logger, constants.TestLocalDomainEntry)
 }