Revert "Create TTL in the UpdateWorkflowExecution cycles. (uber#5243)" (

uber#5380)

This reverts commit bc8bebe.

common/dynamicconfig/constants.go

@@ -1867,13 +1867,6 @@ 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 (
@@ -4030,12 +4023,6 @@ var BoolKeys = map[BoolKey]DynamicBool{
 		Description:  "Enable shardId metrics in persistence client",
 		DefaultValue: true,
 	},
-	EnableExecutionTTL: DynamicBool{
-		KeyName:      "system.enableExecutionTTL",
-		Filters:      []Filter{DomainID},
-		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,7 +31,6 @@ type (
 		EnableCassandraAllConsistencyLevelDelete dynamicconfig.BoolPropertyFn
 		PersistenceSampleLoggingRate             dynamicconfig.IntPropertyFn
 		EnableShardIDMetrics                     dynamicconfig.BoolPropertyFn
-		EnableExecutionTTL                       dynamicconfig.BoolPropertyFnWithDomainIDFilter
 	}
 )
 
@@ -42,6 +41,5 @@ 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,9 +1024,8 @@ type (
 		ReplicationTasks  []Task
 		TimerTasks        []Task
 
-		Condition    int64
-		TTLInSeconds int64
-		Checksum     checksum.Checksum
+		Condition 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
-		TTLInSeconds      int64
-		Condition         int64
+
+		Condition int64
 
 		Checksum checksum.Checksum
 	}

common/persistence/executionManager.go

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

common/persistence/nosql/nosqlExecutionStore.go

@@ -467,6 +467,7 @@ 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,10 +85,9 @@ 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, 0,
+		nowTimestamp, lastWriteVersion,
 	)
 	if err != nil {
 		return nil, err
@@ -128,14 +127,13 @@ 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, ttlInSeconds,
+		nowTimestamp, lastWriteVersion,
 	)
 	if err != nil {
 		return nil, err
@@ -609,7 +607,6 @@ 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(
@@ -639,7 +636,6 @@ func (d *nosqlExecutionStore) prepareUpdateWorkflowExecutionTxn(
 		VersionHistories:              versionHistories,
 		Checksums:                     &checksum,
 		LastWriteVersion:              lastWriteVersion,
-		TTLInSeconds:                  ttlInSeconds,
 	}, nil
 }
 

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

@@ -315,18 +315,8 @@ workflow_state = ? ` +
 		`WHERE shard_id = ? ` +
 		`and type = ?`
 
-	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 +
+	templateUpdateWorkflowExecutionWithVersionHistoriesQuery = `UPDATE executions ` +
+		`SET execution = ` + templateWorkflowExecutionType +
 		`, next_event_id = ? ` +
 		`, version_histories = ? ` +
 		`, version_histories_encoding = ? ` +

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

@@ -1078,6 +1078,7 @@ 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
@@ -1167,6 +1168,7 @@ func (db *cdb) updateWorkflowExecutionAndEventBufferWithMergeAndDeleteMaps(
 			return err
 		}
 	}
+
 	if execution.MapsWriteMode != nosqlplugin.WorkflowExecutionMapsWriteModeUpdate {
 		return fmt.Errorf("should only support WorkflowExecutionMapsWriteModeUpdate")
 	}
@@ -1203,26 +1205,8 @@ func (db *cdb) updateWorkflowExecution(
 ) error {
 	execution.StartTimestamp = db.convertToCassandraTimestamp(execution.StartTimestamp)
 	execution.LastUpdatedTimestamp = db.convertToCassandraTimestamp(execution.LastUpdatedTimestamp)
-	//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.State == persistence.WorkflowStateCompleted || execution.State == persistence.WorkflowStateCorrupted {
-		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,
+
+	batch.Query(templateUpdateWorkflowExecutionWithVersionHistoriesQuery,
 		domainID,
 		workflowID,
 		execution.RunID,

common/persistence/nosql/nosqlplugin/types.go

@@ -39,7 +39,6 @@ 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,92 +609,6 @@ 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,7 +142,6 @@ 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,9 +7,6 @@ system.minRetentionDays:
 history.EnableConsistentQueryByDomain:
 - value: true
   constraints: {}
-system.enableExecutionTTL:
-    - value: true
-      constraints: {}
 history.enableCrossClusterOperations:
 - value: true
   constraints: {}

service/history/execution/context.go

@@ -45,8 +45,6 @@ import (
 
 const (
 	defaultRemoteCallTimeout = 30 * time.Second
-	ttlBufferDays            = 15
-	dayToSecondMultiplier    = 86400
 )
 
 type conflictError struct {
@@ -1216,6 +1214,7 @@ 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,11 +4066,6 @@ 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,
@@ -4095,9 +4090,8 @@ func (e *mutableStateBuilder) CloseTransactionAsMutation(
 		ReplicationTasks:  e.insertReplicationTasks,
 		TimerTasks:        e.insertTimerTasks,
 
-		Condition:    e.nextEventIDInDB,
-		Checksum:     checksum,
-		TTLInSeconds: int64(TTLInSeconds),
+		Condition: e.nextEventIDInDB,
+		Checksum:  checksum,
 	}
 
 	e.checksum = checksum
@@ -4813,27 +4807,3 @@ 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,7 +97,6 @@ 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)
 }