Revert "Optimize activity task timeouts (#8470)" (#8530)

This reverts commit c88fecc.

## What changed?
Revert activity timeout optimization

## Why?
to fix hb timeouts

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

Author: prathyushpv

service/history/interfaces/mutable_state.go

@@ -298,8 +298,6 @@ type (
 		GetWorkflowTaskScheduleToStartTimeoutTask() *tasks.WorkflowTaskTimeoutTask
 		GetWorkflowTaskStartToCloseTimeoutTask() *tasks.WorkflowTaskTimeoutTask
 
-		StoreActivityTimeoutTask(task *tasks.ActivityTimeoutTask)
-
 		IsDirty() bool
 		IsTransitionHistoryEnabled() bool
 		// StartTransaction sets up the mutable state for transacting.

service/history/interfaces/mutable_state_mock.go

@@ -120,14 +120,6 @@ var (
 var emptyTasks = []tasks.Task{}
 
 type (
-	// activityTimeoutTasks holds references to all timeout tasks for a single activity.
-	activityTimeoutTasks struct {
-		ScheduleToStartTimeoutTask *tasks.ActivityTimeoutTask
-		ScheduleToCloseTimeoutTask *tasks.ActivityTimeoutTask
-		StartToCloseTimeoutTask    *tasks.ActivityTimeoutTask
-		HeartbeatTimeoutTask       *tasks.ActivityTimeoutTask
-	}
-
 	MutableStateImpl struct {
 		pendingActivityTimerHeartbeats map[int64]time.Time                    // Scheduled Event ID -> LastHeartbeatTimeoutVisibilityInSeconds.
 		pendingActivityInfoIDs         map[int64]*persistencespb.ActivityInfo // Scheduled Event ID -> Activity Info.
@@ -235,10 +227,6 @@ type (
 		wftScheduleToStartTimeoutTask *tasks.WorkflowTaskTimeoutTask
 		wftStartToCloseTimeoutTask    *tasks.WorkflowTaskTimeoutTask
 
-		// In-memory storage for activity timeout tasks. These are set when timeout tasks are generated and used to
-		// delete them when the activity completes/fails. Map key is the activity's scheduled event ID.
-		activityTimeoutTasks map[int64]*activityTimeoutTasks
-
 		// Do not rely on this, this is only updated on
 		// Load() and closeTransactionXXX methods. So when
 		// a transaction is in progress, this value will be
@@ -288,7 +276,6 @@ func NewMutableState(
 		pendingActivityIDToEventID:     make(map[string]int64),
 		deleteActivityInfos:            make(map[int64]struct{}),
 		syncActivityTasks:              make(map[int64]struct{}),
-		activityTimeoutTasks:           make(map[int64]*activityTimeoutTasks),
 
 		pendingTimerInfoIDs:     make(map[string]*persistencespb.TimerInfo),
 		pendingTimerEventIDToID: make(map[int64]string),
@@ -1864,9 +1851,6 @@ func (ms *MutableStateImpl) UpdateActivityProgress(
 			metrics.OperationTag(metrics.HistoryRecordActivityTaskHeartbeatScope),
 			metrics.NamespaceTag(ms.namespaceEntry.Name().String()))
 	}
-
-	// Delete Heartbeat timeout task since heartbeat has been received
-	ms.deleteActivityTimeoutTask(ai.ScheduledEventId, enumspb.TIMEOUT_TYPE_HEARTBEAT, ai)
 }
 
 // UpdateActivityInfo applies the necessary activity information
@@ -1973,7 +1957,6 @@ func (ms *MutableStateImpl) DeleteActivity(
 			// log data inconsistency instead of returning an error
 			ms.logDataInconsistency()
 		}
-		ms.recordActivityTimeoutTasksForDeletion(scheduledEventID, activityInfo)
 	} else {
 		ms.logError(
 			fmt.Sprintf("unable to find activity event id: %v in mutable state", scheduledEventID),
@@ -3652,8 +3635,6 @@ func (ms *MutableStateImpl) AddActivityTaskStartedEvent(
 		if err := ms.ApplyActivityTaskStartedEvent(event); err != nil {
 			return nil, err
 		}
-		// Delete ScheduleToStart timeout task since activity has now started
-		ms.deleteActivityTimeoutTask(scheduledEventID, enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START, ai)
 		return event, nil
 	}
 
@@ -3677,8 +3658,6 @@ func (ms *MutableStateImpl) AddActivityTaskStartedEvent(
 		return nil, err
 	}
 	ms.syncActivityTasks[ai.ScheduledEventId] = struct{}{}
-	// Delete ScheduleToStart timeout task since activity has now started
-	ms.deleteActivityTimeoutTask(scheduledEventID, enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START, ai)
 	return nil, nil
 }
 
@@ -6209,118 +6188,6 @@ func (ms *MutableStateImpl) GetWorkflowTaskStartToCloseTimeoutTask() *tasks.Work
 	return ms.wftStartToCloseTimeoutTask
 }
 
-// StoreActivityTimeoutTask stores an activity timeout task reference in memory.
-func (ms *MutableStateImpl) StoreActivityTimeoutTask(
-	task *tasks.ActivityTimeoutTask,
-) {
-	timeoutTasks, exists := ms.activityTimeoutTasks[task.EventID]
-	if !exists {
-		timeoutTasks = &activityTimeoutTasks{}
-		ms.activityTimeoutTasks[task.EventID] = timeoutTasks
-	}
-
-	switch task.TimeoutType {
-	case enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START:
-		timeoutTasks.ScheduleToStartTimeoutTask = task
-	case enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE:
-		timeoutTasks.ScheduleToCloseTimeoutTask = task
-	case enumspb.TIMEOUT_TYPE_START_TO_CLOSE:
-		timeoutTasks.StartToCloseTimeoutTask = task
-	case enumspb.TIMEOUT_TYPE_HEARTBEAT:
-		timeoutTasks.HeartbeatTimeoutTask = task
-	default:
-		// No-op for unhandled timeout types
-	}
-}
-
-// deleteActivityTimeoutTask marks a specific activity timeout task to be deleted.
-// This is used to delete timeout tasks at specific lifecycle points (e.g., when activity starts, when heartbeat is received).
-// The task will be regenerated with updated deadline on the next transaction close if still needed.
-func (ms *MutableStateImpl) deleteActivityTimeoutTask(
-	scheduledEventID int64,
-	timeoutType enumspb.TimeoutType,
-	activityInfo *persistencespb.ActivityInfo,
-) {
-	if activityInfo == nil {
-		return
-	}
-
-	timeoutTasks, exists := ms.activityTimeoutTasks[scheduledEventID]
-	if !exists {
-		return
-	}
-
-	var task *tasks.ActivityTimeoutTask
-
-	switch timeoutType {
-	case enumspb.TIMEOUT_TYPE_SCHEDULE_TO_START:
-		task = timeoutTasks.ScheduleToStartTimeoutTask
-		timeoutTasks.ScheduleToStartTimeoutTask = nil
-	case enumspb.TIMEOUT_TYPE_HEARTBEAT:
-		task = timeoutTasks.HeartbeatTimeoutTask
-		timeoutTasks.HeartbeatTimeoutTask = nil
-	default:
-		return
-	}
-
-	if task == nil {
-		return
-	}
-
-	key := task.GetKey()
-	ms.BestEffortDeleteTasks[tasks.CategoryTimer] = append(
-		ms.BestEffortDeleteTasks[tasks.CategoryTimer],
-		key,
-	)
-}
-
-// recordActivityTimeoutTasksForDeletion records activity timeout tasks for deletion. This is called when an activity
-// completes or fails. It deletes all remaining timeout tasks for the activity.
-func (ms *MutableStateImpl) recordActivityTimeoutTasksForDeletion(
-	scheduledEventID int64,
-	activityInfo *persistencespb.ActivityInfo,
-) {
-	if activityInfo == nil {
-		return
-	}
-
-	timeoutTasks, exists := ms.activityTimeoutTasks[scheduledEventID]
-	if !exists {
-		return
-	}
-
-	// Delete all timeout tasks for this activity
-	if task := timeoutTasks.ScheduleToStartTimeoutTask; task != nil {
-		ms.BestEffortDeleteTasks[tasks.CategoryTimer] = append(
-			ms.BestEffortDeleteTasks[tasks.CategoryTimer],
-			task.GetKey(),
-		)
-	}
-
-	if task := timeoutTasks.ScheduleToCloseTimeoutTask; task != nil {
-		ms.BestEffortDeleteTasks[tasks.CategoryTimer] = append(
-			ms.BestEffortDeleteTasks[tasks.CategoryTimer],
-			task.GetKey(),
-		)
-	}
-
-	if task := timeoutTasks.StartToCloseTimeoutTask; task != nil {
-		ms.BestEffortDeleteTasks[tasks.CategoryTimer] = append(
-			ms.BestEffortDeleteTasks[tasks.CategoryTimer],
-			task.GetKey(),
-		)
-	}
-
-	if task := timeoutTasks.HeartbeatTimeoutTask; task != nil {
-		ms.BestEffortDeleteTasks[tasks.CategoryTimer] = append(
-			ms.BestEffortDeleteTasks[tasks.CategoryTimer],
-			task.GetKey(),
-		)
-	}
-
-	delete(ms.activityTimeoutTasks, scheduledEventID)
-}
-
 func (ms *MutableStateImpl) GetWorkflowStateStatus() (enumsspb.WorkflowExecutionState, enumspb.WorkflowExecutionStatus) {
 	return ms.executionState.State, ms.executionState.Status
 }