From 33823ac9e0ab0b6b678281936ae9a44dd9e584df Mon Sep 17 00:00:00 2001 From: ddbaron Date: Tue, 8 Sep 2026 08:30:25 -0500 Subject: [PATCH 1/3] fix(storage): honor execution activity in workflow reaper --- .../internal/storage/execution_records.go | 28 +++-- .../storage/stale_execution_reaper_test.go | 106 ++++++++++++++++++ 2 files changed, 125 insertions(+), 9 deletions(-) diff --git a/control-plane/internal/storage/execution_records.go b/control-plane/internal/storage/execution_records.go index ea533463b..be63ea0f0 100644 --- a/control-plane/internal/storage/execution_records.go +++ b/control-plane/internal/storage/execution_records.go @@ -1295,8 +1295,10 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time } // MarkStaleWorkflowExecutions updates workflow executions stuck in non-terminal states -// when their updated_at timestamp exceeds the staleAfter threshold. This catches orphaned -// child executions whose parent failed without cascading cancellation. +// when both the workflow and its paired execution activity timestamps exceed the +// staleAfter threshold. This catches orphaned child executions whose parent failed +// without cascading cancellation while allowing activity recorded in either table +// to keep the execution alive. // // See MarkStaleExecutions for the updated_at invariant, the COALESCE fallback // rationale, and why a row with a non-terminal child is skipped rather than reaped. @@ -1311,21 +1313,29 @@ func (ls *LocalStorage) MarkStaleWorkflowExecutions(ctx context.Context, staleAf cutoff := time.Now().UTC().Add(-staleAfter) db := ls.requireSQLDB() - tsExpr := ls.staleTimestampExpr("COALESCE(updated_at, created_at, started_at)") + workflowTSExpr := ls.staleTimestampExpr("COALESCE(w.updated_at, w.created_at, w.started_at)") + executionTSExpr := ls.staleTimestampExpr("COALESCE(e.updated_at, e.created_at, e.started_at)") cutoffExpr := ls.staleTimestampExpr("?") + // The legacy reaper runs first and makes its row terminal while updating + // updated_at. Terminal rows must not shield their still-active workflow row + // from this reaper, or the two tables could remain out of sync forever. rows, err := db.QueryContext(ctx, ` - SELECT execution_id, started_at + SELECT w.execution_id, w.started_at FROM workflow_executions w - WHERE status IN ('running', 'pending', 'queued', 'waiting') - AND `+tsExpr+` <= `+cutoffExpr+` - AND COALESCE(approval_status, '') != 'pending' + LEFT JOIN executions e + ON e.execution_id = w.execution_id + AND e.status IN ('running', 'pending', 'queued', 'waiting') + WHERE w.status IN ('running', 'pending', 'queued', 'waiting') + AND `+workflowTSExpr+` <= `+cutoffExpr+` + AND (e.execution_id IS NULL OR `+executionTSExpr+` <= `+cutoffExpr+`) + AND COALESCE(w.approval_status, '') != 'pending' AND NOT EXISTS ( SELECT 1 FROM workflow_executions c WHERE c.parent_execution_id = w.execution_id AND c.status IN ('running', 'pending', 'queued', 'waiting') ) - ORDER BY `+tsExpr+` ASC - LIMIT ?`, cutoff, limit) + ORDER BY `+workflowTSExpr+` ASC + LIMIT ?`, cutoff, cutoff, limit) if err != nil { return 0, fmt.Errorf("query stale workflow executions: %w", err) } diff --git a/control-plane/internal/storage/stale_execution_reaper_test.go b/control-plane/internal/storage/stale_execution_reaper_test.go index 361ee907c..0fd59138a 100644 --- a/control-plane/internal/storage/stale_execution_reaper_test.go +++ b/control-plane/internal/storage/stale_execution_reaper_test.go @@ -259,6 +259,112 @@ func TestMarkStaleWorkflowExecutions_ReapsWaitingState(t *testing.T) { require.Equal(t, "timeout", record.Status) } +func TestMarkStaleWorkflowExecutions_ExecutionActivityProtectsWorkflow(t *testing.T) { + ls, ctx := setupTestLocalStorage(t) + now := time.Now().UTC() + + workflow := &types.WorkflowExecution{ + WorkflowID: "wf-heartbeat", + ExecutionID: "exec-heartbeat", + AgentFieldRequestID: "req-heartbeat", + AgentNodeID: "agent-1", + ReasonerID: "reasoner.coder", + Status: "running", + StartedAt: now.Add(-2 * time.Hour), + CreatedAt: now.Add(-2 * time.Hour), + UpdatedAt: now.Add(-1 * time.Hour), + WorkflowTags: []string{}, + InputData: json.RawMessage("{}"), + OutputData: json.RawMessage("{}"), + } + require.NoError(t, ls.StoreWorkflowExecution(ctx, workflow)) + + execution := &types.Execution{ + ExecutionID: "exec-heartbeat", + RunID: "run-heartbeat", + AgentNodeID: "agent-1", + ReasonerID: "reasoner.coder", + NodeID: "node-1", + Status: "running", + StartedAt: now.Add(-2 * time.Hour), + } + require.NoError(t, ls.CreateExecutionRecord(ctx, execution)) + backdateExecutionUpdatedAt(t, ls, "executions", execution.ExecutionID, now.Add(-1*time.Hour)) + + // A heartbeat-style note updates only executions.updated_at. The workflow + // row remains old, but the execution is still actively making progress. + _, err := ls.UpdateExecutionRecord(ctx, execution.ExecutionID, func(current *types.Execution) (*types.Execution, error) { + current.Notes = append(current.Notes, types.ExecutionNote{ + Message: "heartbeat", + Timestamp: now, + }) + return current, nil + }) + require.NoError(t, err) + + reaped, err := ls.MarkStaleWorkflowExecutions(ctx, 30*time.Minute, 100) + require.NoError(t, err) + require.Equal(t, 0, reaped, "recent activity in executions must protect the workflow row") + + workflowRecord, err := ls.GetWorkflowExecution(ctx, workflow.ExecutionID) + require.NoError(t, err) + require.Equal(t, "running", workflowRecord.Status) + + executionRecord, err := ls.GetExecutionRecord(ctx, execution.ExecutionID) + require.NoError(t, err) + require.Equal(t, "running", executionRecord.Status) +} + +func TestMarkStaleWorkflowExecutions_SilentRowsReapAndSyncExecution(t *testing.T) { + ls, ctx := setupTestLocalStorage(t) + now := time.Now().UTC() + + workflow := &types.WorkflowExecution{ + WorkflowID: "wf-silent-paired", + ExecutionID: "exec-silent-paired", + AgentFieldRequestID: "req-silent-paired", + AgentNodeID: "agent-1", + ReasonerID: "reasoner.coder", + Status: "running", + StartedAt: now.Add(-2 * time.Hour), + CreatedAt: now.Add(-2 * time.Hour), + UpdatedAt: now.Add(-1 * time.Hour), + WorkflowTags: []string{}, + InputData: json.RawMessage("{}"), + OutputData: json.RawMessage("{}"), + } + require.NoError(t, ls.StoreWorkflowExecution(ctx, workflow)) + + execution := &types.Execution{ + ExecutionID: "exec-silent-paired", + RunID: "run-silent-paired", + AgentNodeID: "agent-1", + ReasonerID: "reasoner.coder", + NodeID: "node-1", + Status: "running", + StartedAt: now.Add(-2 * time.Hour), + } + require.NoError(t, ls.CreateExecutionRecord(ctx, execution)) + backdateExecutionUpdatedAt(t, ls, "executions", execution.ExecutionID, now.Add(-1*time.Hour)) + + reaped, err := ls.MarkStaleWorkflowExecutions(ctx, 30*time.Minute, 100) + require.NoError(t, err) + require.Equal(t, 1, reaped, "rows silent on both clocks must still be reaped") + + workflowRecord, err := ls.GetWorkflowExecution(ctx, workflow.ExecutionID) + require.NoError(t, err) + require.Equal(t, "timeout", workflowRecord.Status) + require.NotNil(t, workflowRecord.CompletedAt) + require.Contains(t, *workflowRecord.ErrorMessage, "no activity") + + // The legacy execution record must keep the existing terminal sync behavior. + executionRecord, err := ls.GetExecutionRecord(ctx, execution.ExecutionID) + require.NoError(t, err) + require.Equal(t, "timeout", executionRecord.Status) + require.NotNil(t, executionRecord.CompletedAt) + require.Contains(t, *executionRecord.ErrorMessage, "no activity") +} + func TestMarkStaleWorkflowExecutions_MultipleStuckExecutions(t *testing.T) { ls, ctx := setupTestLocalStorage(t) now := time.Now().UTC() From bcdfb064b012d7024d443a57c53c35ba0077381f Mon Sep 17 00:00:00 2001 From: ddbaron Date: Wed, 9 Sep 2026 09:16:15 -0500 Subject: [PATCH 2/3] fix(storage): guard stale workflow update against activity race --- .../internal/storage/execution_records.go | 37 +++++++++++- .../storage/stale_execution_reaper_test.go | 57 +++++++++++++++++++ 2 files changed, 92 insertions(+), 2 deletions(-) diff --git a/control-plane/internal/storage/execution_records.go b/control-plane/internal/storage/execution_records.go index be63ea0f0..2cd4faf75 100644 --- a/control-plane/internal/storage/execution_records.go +++ b/control-plane/internal/storage/execution_records.go @@ -1303,6 +1303,10 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time // See MarkStaleExecutions for the updated_at invariant, the COALESCE fallback // rationale, and why a row with a non-terminal child is skipped rather than reaped. func (ls *LocalStorage) MarkStaleWorkflowExecutions(ctx context.Context, staleAfter time.Duration, limit int) (int, error) { + return ls.markStaleWorkflowExecutions(ctx, staleAfter, limit, nil) +} + +func (ls *LocalStorage) markStaleWorkflowExecutions(ctx context.Context, staleAfter time.Duration, limit int, afterCandidateSelection func()) (int, error) { if limit <= 0 { return 0, nil } @@ -1362,6 +1366,12 @@ func (ls *LocalStorage) MarkStaleWorkflowExecutions(ctx context.Context, staleAf return 0, nil } + // Package tests use this seam to make the candidate-selection-to-update + // interleaving deterministic; production callers leave it nil. + if afterCandidateSelection != nil { + afterCandidateSelection() + } + tx, err := db.BeginTx(ctx, nil) if err != nil { return 0, fmt.Errorf("begin stale workflow execution transaction: %w", err) @@ -1369,9 +1379,30 @@ func (ls *LocalStorage) MarkStaleWorkflowExecutions(ctx context.Context, staleAf defer rollbackTx(tx, "MarkStaleWorkflowExecutions") updateStmt, err := tx.PrepareContext(ctx, ` - UPDATE workflow_executions + UPDATE workflow_executions AS w SET status = ?, error_message = ?, completed_at = ?, duration_ms = ?, updated_at = ? - WHERE execution_id = ? AND status IN ('running', 'pending', 'queued', 'waiting')`) + WHERE w.execution_id = ? + AND w.status IN ('running', 'pending', 'queued', 'waiting') + AND `+workflowTSExpr+` <= `+cutoffExpr+` + AND ( + NOT EXISTS ( + SELECT 1 FROM executions e + WHERE e.execution_id = w.execution_id + AND e.status IN ('running', 'pending', 'queued', 'waiting') + ) + OR EXISTS ( + SELECT 1 FROM executions e + WHERE e.execution_id = w.execution_id + AND e.status IN ('running', 'pending', 'queued', 'waiting') + AND `+executionTSExpr+` <= `+cutoffExpr+` + ) + ) + AND COALESCE(w.approval_status, '') != 'pending' + AND NOT EXISTS ( + SELECT 1 FROM workflow_executions c + WHERE c.parent_execution_id = w.execution_id + AND c.status IN ('running', 'pending', 'queued', 'waiting') + )`) if err != nil { return 0, fmt.Errorf("prepare stale workflow execution update: %w", err) } @@ -1409,6 +1440,8 @@ func (ls *LocalStorage) MarkStaleWorkflowExecutions(ctx context.Context, staleAf durationMS, now, rec.id, + cutoff, + cutoff, ) if err != nil { return 0, fmt.Errorf("update stale workflow execution %s: %w", rec.id, err) diff --git a/control-plane/internal/storage/stale_execution_reaper_test.go b/control-plane/internal/storage/stale_execution_reaper_test.go index 0fd59138a..d984f566a 100644 --- a/control-plane/internal/storage/stale_execution_reaper_test.go +++ b/control-plane/internal/storage/stale_execution_reaper_test.go @@ -315,6 +315,63 @@ func TestMarkStaleWorkflowExecutions_ExecutionActivityProtectsWorkflow(t *testin require.Equal(t, "running", executionRecord.Status) } +func TestMarkStaleWorkflowExecutions_ActivityAfterSelectionSkipsUpdate(t *testing.T) { + ls, ctx := setupTestLocalStorage(t) + now := time.Now().UTC() + + workflow := &types.WorkflowExecution{ + WorkflowID: "wf-selection-race", + ExecutionID: "exec-selection-race", + AgentFieldRequestID: "req-selection-race", + AgentNodeID: "agent-1", + ReasonerID: "reasoner.coder", + Status: "running", + StartedAt: now.Add(-2 * time.Hour), + CreatedAt: now.Add(-2 * time.Hour), + UpdatedAt: now.Add(-1 * time.Hour), + WorkflowTags: []string{}, + InputData: json.RawMessage("{}"), + OutputData: json.RawMessage("{}"), + } + require.NoError(t, ls.StoreWorkflowExecution(ctx, workflow)) + + execution := &types.Execution{ + ExecutionID: "exec-selection-race", + RunID: "run-selection-race", + AgentNodeID: "agent-1", + ReasonerID: "reasoner.coder", + NodeID: "node-1", + Status: "running", + StartedAt: now.Add(-2 * time.Hour), + } + require.NoError(t, ls.CreateExecutionRecord(ctx, execution)) + backdateExecutionUpdatedAt(t, ls, "executions", execution.ExecutionID, now.Add(-1*time.Hour)) + + var heartbeatErr error + reaped, err := ls.markStaleWorkflowExecutions(ctx, 30*time.Minute, 100, func() { + // This runs after candidate selection and before the conditional update. + _, heartbeatErr = ls.UpdateExecutionRecord(ctx, execution.ExecutionID, func(current *types.Execution) (*types.Execution, error) { + current.Notes = append(current.Notes, types.ExecutionNote{ + Message: "heartbeat", + Timestamp: now, + }) + return current, nil + }) + }) + require.NoError(t, heartbeatErr) + require.NoError(t, err) + require.Equal(t, 0, reaped, "activity after selection must prevent the final timeout update") + + workflowRecord, err := ls.GetWorkflowExecution(ctx, workflow.ExecutionID) + require.NoError(t, err) + require.Equal(t, "running", workflowRecord.Status) + + executionRecord, err := ls.GetExecutionRecord(ctx, execution.ExecutionID) + require.NoError(t, err) + require.Equal(t, "running", executionRecord.Status) + require.Len(t, executionRecord.Notes, 1) +} + func TestMarkStaleWorkflowExecutions_SilentRowsReapAndSyncExecution(t *testing.T) { ls, ctx := setupTestLocalStorage(t) now := time.Now().UTC() From 8c2395ecf3f6144cc33b4816f28326de67076e32 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Wed, 9 Sep 2026 11:01:26 -0400 Subject: [PATCH 3/3] fix(storage): re-check staleness in the execution reaper update MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MarkStaleWorkflowExecutions now repeats its candidate predicates in the conditional UPDATE, but MarkStaleExecutions still only re-checked status. A heartbeat that lands between its candidate selection and that UPDATE therefore still flips a live execution row to timeout — the same false timeout the workflow reaper just stopped producing, through a narrower window (its candidate query reads the clock the heartbeat writes, so the race is the millisecond gap between the two statements rather than the whole run). Give it the same treatment: the conditional UPDATE re-evaluates the activity clock against the sweep cutoff and the non-terminal-child guard, and the body moves behind the same post-selection seam the workflow reaper uses so the interleaving is testable without sleeps. Tests: a real execution-note write landing in that window leaves the row running with its note intact; a seam that writes nothing still reaps the silent row with the existing "no activity" message. Co-Authored-By: Claude Opus 5 --- .../internal/storage/execution_records.go | 25 ++++++- .../storage/stale_execution_reaper_test.go | 68 +++++++++++++++++++ 2 files changed, 91 insertions(+), 2 deletions(-) diff --git a/control-plane/internal/storage/execution_records.go b/control-plane/internal/storage/execution_records.go index 2cd4faf75..14882c682 100644 --- a/control-plane/internal/storage/execution_records.go +++ b/control-plane/internal/storage/execution_records.go @@ -1185,7 +1185,13 @@ func parseTimeString(value string) (time.Time, error) { // is reaped first, which makes its parent childless and eligible on the next // sweep, and so on up. Nothing is stuck forever; it just takes one sweep per // level. +// The conditional UPDATE re-evaluates the staleness predicates so a row that +// gains activity after selection is left alone. func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time.Duration, limit int) (int, error) { + return ls.markStaleExecutions(ctx, staleAfter, limit, nil) +} + +func (ls *LocalStorage) markStaleExecutions(ctx context.Context, staleAfter time.Duration, limit int, afterCandidateSelection func()) (int, error) { if limit <= 0 { return 0, nil } @@ -1197,6 +1203,7 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time db := ls.requireSQLDB() tsExpr := ls.staleTimestampExpr("COALESCE(updated_at, created_at, started_at)") + executionUpdateTSExpr := ls.staleTimestampExpr("COALESCE(e.updated_at, e.created_at, e.started_at)") cutoffExpr := ls.staleTimestampExpr("?") rows, err := db.QueryContext(ctx, ` SELECT execution_id, started_at @@ -1236,6 +1243,12 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time return 0, nil } + // Package tests use this seam to make the candidate-selection-to-update + // interleaving deterministic; production callers leave it nil. + if afterCandidateSelection != nil { + afterCandidateSelection() + } + tx, err := db.BeginTx(ctx, nil) if err != nil { return 0, fmt.Errorf("begin stale execution transaction: %w", err) @@ -1243,9 +1256,16 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time defer rollbackTx(tx, "MarkStaleExecutions") updateStmt, err := tx.PrepareContext(ctx, ` - UPDATE executions + UPDATE executions AS e SET status = ?, error_message = ?, completed_at = ?, duration_ms = ?, updated_at = ? - WHERE execution_id = ? AND status IN ('running', 'pending', 'queued')`) + WHERE e.execution_id = ? + AND e.status IN ('running', 'pending', 'queued') + AND `+executionUpdateTSExpr+` <= `+cutoffExpr+` + AND NOT EXISTS ( + SELECT 1 FROM executions c + WHERE c.parent_execution_id = e.execution_id + AND c.status IN ('running', 'pending', 'queued') + )`) if err != nil { return 0, fmt.Errorf("prepare stale execution update: %w", err) } @@ -1273,6 +1293,7 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time durationMS, now, rec.id, + cutoff, ) if err != nil { return 0, fmt.Errorf("update stale execution %s: %w", rec.id, err) diff --git a/control-plane/internal/storage/stale_execution_reaper_test.go b/control-plane/internal/storage/stale_execution_reaper_test.go index d984f566a..dbf405772 100644 --- a/control-plane/internal/storage/stale_execution_reaper_test.go +++ b/control-plane/internal/storage/stale_execution_reaper_test.go @@ -372,6 +372,74 @@ func TestMarkStaleWorkflowExecutions_ActivityAfterSelectionSkipsUpdate(t *testin require.Len(t, executionRecord.Notes, 1) } +func TestMarkStaleExecutions_ActivityAfterSelectionSkipsUpdate(t *testing.T) { + ls, ctx := setupTestLocalStorage(t) + now := time.Now().UTC() + + execution := &types.Execution{ + ExecutionID: "exec-selection-race-legacy", + RunID: "run-selection-race-legacy", + AgentNodeID: "agent-1", + ReasonerID: "reasoner.coder", + NodeID: "node-1", + Status: "running", + StartedAt: now.Add(-2 * time.Hour), + } + require.NoError(t, ls.CreateExecutionRecord(ctx, execution)) + backdateExecutionUpdatedAt(t, ls, "executions", execution.ExecutionID, now.Add(-1*time.Hour)) + + var heartbeatErr error + reaped, err := ls.markStaleExecutions(ctx, 30*time.Minute, 100, func() { + // This runs after candidate selection and before the conditional update. + _, heartbeatErr = ls.UpdateExecutionRecord(ctx, execution.ExecutionID, func(current *types.Execution) (*types.Execution, error) { + current.Notes = append(current.Notes, types.ExecutionNote{ + Message: "heartbeat", + Timestamp: now, + }) + return current, nil + }) + }) + require.NoError(t, heartbeatErr) + require.NoError(t, err) + require.Equal(t, 0, reaped, "activity after selection must prevent the final timeout update") + + executionRecord, err := ls.GetExecutionRecord(ctx, execution.ExecutionID) + require.NoError(t, err) + require.Equal(t, "running", executionRecord.Status) + require.Len(t, executionRecord.Notes, 1) +} + +func TestMarkStaleExecutions_SeamWithoutActivityStillReaps(t *testing.T) { + ls, ctx := setupTestLocalStorage(t) + now := time.Now().UTC() + + execution := &types.Execution{ + ExecutionID: "exec-selection-race-control", + RunID: "run-selection-race-control", + AgentNodeID: "agent-1", + ReasonerID: "reasoner.coder", + NodeID: "node-1", + Status: "running", + StartedAt: now.Add(-2 * time.Hour), + } + require.NoError(t, ls.CreateExecutionRecord(ctx, execution)) + backdateExecutionUpdatedAt(t, ls, "executions", execution.ExecutionID, now.Add(-1*time.Hour)) + + hookRan := false + reaped, err := ls.markStaleExecutions(ctx, 30*time.Minute, 100, func() { + hookRan = true + }) + require.NoError(t, err) + require.Equal(t, 1, reaped, "a silent row must still be reaped when the seam writes nothing") + require.True(t, hookRan) + + executionRecord, err := ls.GetExecutionRecord(ctx, execution.ExecutionID) + require.NoError(t, err) + require.Equal(t, "timeout", executionRecord.Status) + require.NotNil(t, executionRecord.ErrorMessage) + require.Contains(t, *executionRecord.ErrorMessage, "no activity") +} + func TestMarkStaleWorkflowExecutions_SilentRowsReapAndSyncExecution(t *testing.T) { ls, ctx := setupTestLocalStorage(t) now := time.Now().UTC()