Skip to content

Commit 56410a6

Browse files
AbirAbbasclaudesantoshkumarradha
authored
fix: reap stale workflow executions and use updated_at for staleness (#262)
* fix: reap stale workflow executions and use updated_at for staleness detection The existing MarkStaleExecutions only covered the executions table and used started_at to detect staleness, which missed orphaned workflow executions entirely and could incorrectly timeout legitimately long-running executions. This change: - Switches staleness detection from started_at to updated_at so only executions with no recent activity are reaped - Adds MarkStaleWorkflowExecutions to handle the workflow_executions table where orphaned child executions get permanently stuck in running state when their parent fails - Wires both into the existing ExecutionCleanupService background loop Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * test: add functional tests for stale execution reaper with real SQLite Tests run against a real database (no mocks) covering: - Stuck executions reaped while active ones are preserved - Long-running executions with recent activity NOT incorrectly reaped - Orphaned workflow children reaped when parent already failed - Waiting-state executions reaped after inactivity - Batch limit respected across multiple reaper passes - End-to-end scenario: parent fails, children stuck in both tables Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: use COALESCE fallback for NULL updated_at in stale reaper queries - Use COALESCE(updated_at, created_at, started_at) in both MarkStaleExecutions and MarkStaleWorkflowExecutions to handle rows where updated_at was never set - Add invariant comment documenting that updated_at must be bumped on every meaningful activity for staleness detection to work - Add tests for NULL updated_at scenario on both execution types --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com> Co-authored-by: Santosh <santosh@agentfield.ai>
1 parent a0df125 commit 56410a6

7 files changed

Lines changed: 732 additions & 3 deletions

File tree

‎control-plane/internal/handlers/execute_test.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,11 @@ func (m *MockStorageProvider) MarkStaleExecutions(ctx context.Context, staleAfte
5050
return args.Int(0), args.Error(1)
5151
}
5252

53+
func (m *MockStorageProvider) MarkStaleWorkflowExecutions(ctx context.Context, staleAfter time.Duration, limit int) (int, error) {
54+
args := m.Called(ctx, staleAfter, limit)
55+
return args.Int(0), args.Error(1)
56+
}
57+
5358
// Add other required methods as no-ops for the interface
5459
func (m *MockStorageProvider) Initialize(ctx context.Context, config interface{}) error { return nil }
5560
func (m *MockStorageProvider) Close(ctx context.Context) error { return nil }

‎control-plane/internal/handlers/execution_cleanup.go‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,16 @@ func (ecs *ExecutionCleanupService) performCleanup(ctx context.Context) {
141141
Dur("stale_timeout", ecs.config.StaleExecutionTimeout).
142142
Msg("marked stale executions as timed out")
143143
}
144+
145+
wfTimedOut, err := ecs.storage.MarkStaleWorkflowExecutions(cleanupCtx, ecs.config.StaleExecutionTimeout, ecs.config.BatchSize)
146+
if err != nil {
147+
logger.Logger.Error().Err(err).Msg("failed to mark stale workflow executions as timed out")
148+
} else if wfTimedOut > 0 {
149+
logger.Logger.Debug().
150+
Int("timed_out", wfTimedOut).
151+
Dur("stale_timeout", ecs.config.StaleExecutionTimeout).
152+
Msg("marked stale workflow executions as timed out")
153+
}
144154
}
145155

146156
for {

‎control-plane/internal/handlers/execution_cleanup_test.go‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,9 @@ type cleanupStoreMock struct {
4242

4343
markStaleCalls []markStaleCall
4444
markStaleResponses []cleanupResponse
45+
46+
markStaleWfCalls []markStaleCall
47+
markStaleWfResponses []cleanupResponse
4548
}
4649

4750
func (m *cleanupStoreMock) CleanupOldExecutions(ctx context.Context, retentionPeriod time.Duration, batchSize int) (int, error) {
@@ -98,6 +101,33 @@ func (m *cleanupStoreMock) getMarkStaleCalls() []markStaleCall {
98101
return out
99102
}
100103

104+
func (m *cleanupStoreMock) MarkStaleWorkflowExecutions(ctx context.Context, staleAfter time.Duration, limit int) (int, error) {
105+
m.mu.Lock()
106+
defer m.mu.Unlock()
107+
108+
callIndex := len(m.markStaleWfCalls)
109+
m.markStaleWfCalls = append(m.markStaleWfCalls, markStaleCall{
110+
ctx: ctx,
111+
staleAfter: staleAfter,
112+
limit: limit,
113+
})
114+
115+
if callIndex < len(m.markStaleWfResponses) {
116+
return m.markStaleWfResponses[callIndex].count, m.markStaleWfResponses[callIndex].err
117+
}
118+
119+
return 0, nil
120+
}
121+
122+
func (m *cleanupStoreMock) getMarkStaleWfCalls() []markStaleCall {
123+
m.mu.Lock()
124+
defer m.mu.Unlock()
125+
126+
out := make([]markStaleCall, len(m.markStaleWfCalls))
127+
copy(out, m.markStaleWfCalls)
128+
return out
129+
}
130+
101131
type syncBuffer struct {
102132
mu sync.Mutex
103133
buf bytes.Buffer
@@ -443,6 +473,66 @@ func TestExecutionCleanupService_PerformCleanup_StopsWhenContextIsCancelled(t *t
443473
}
444474
}
445475

476+
func TestExecutionCleanupService_PerformCleanup_MarksStaleWorkflowExecutions(t *testing.T) {
477+
logBuffer := setupExecutionCleanupTestLogger(t)
478+
store := &cleanupStoreMock{
479+
markStaleResponses: []cleanupResponse{{count: 1}},
480+
markStaleWfResponses: []cleanupResponse{{count: 3}},
481+
cleanupResponses: []cleanupResponse{{count: 0}},
482+
}
483+
484+
cfg := testExecutionCleanupConfig(10)
485+
service := NewExecutionCleanupService(store, cfg)
486+
service.performCleanup(context.Background())
487+
488+
markCalls := store.getMarkStaleCalls()
489+
if len(markCalls) != 1 {
490+
t.Fatalf("expected 1 mark stale call, got %d", len(markCalls))
491+
}
492+
493+
wfCalls := store.getMarkStaleWfCalls()
494+
if len(wfCalls) != 1 {
495+
t.Fatalf("expected 1 mark stale workflow call, got %d", len(wfCalls))
496+
}
497+
if wfCalls[0].staleAfter != cfg.StaleExecutionTimeout {
498+
t.Fatalf("expected stale timeout %v, got %v", cfg.StaleExecutionTimeout, wfCalls[0].staleAfter)
499+
}
500+
if wfCalls[0].limit != cfg.BatchSize {
501+
t.Fatalf("expected batch size %d, got %d", cfg.BatchSize, wfCalls[0].limit)
502+
}
503+
504+
logs := logBuffer.String()
505+
if !strings.Contains(logs, "marked stale executions as timed out") {
506+
t.Fatalf("expected stale execution log, got logs: %s", logs)
507+
}
508+
if !strings.Contains(logs, "marked stale workflow executions as timed out") {
509+
t.Fatalf("expected stale workflow execution log, got logs: %s", logs)
510+
}
511+
}
512+
513+
func TestExecutionCleanupService_PerformCleanup_ContinuesWhenMarkStaleWorkflowFails(t *testing.T) {
514+
logBuffer := setupExecutionCleanupTestLogger(t)
515+
store := &cleanupStoreMock{
516+
markStaleResponses: []cleanupResponse{{count: 0}},
517+
markStaleWfResponses: []cleanupResponse{{err: errors.New("workflow stale failed")}},
518+
cleanupResponses: []cleanupResponse{{count: 0}},
519+
}
520+
521+
cfg := testExecutionCleanupConfig(5)
522+
service := NewExecutionCleanupService(store, cfg)
523+
service.performCleanup(context.Background())
524+
525+
// Cleanup should still proceed despite workflow stale-marking failure
526+
if len(store.getCleanupCalls()) != 1 {
527+
t.Fatalf("expected cleanup to continue after workflow stale-marking failure")
528+
}
529+
530+
logs := logBuffer.String()
531+
if !strings.Contains(logs, "failed to mark stale workflow executions as timed out") {
532+
t.Fatalf("expected workflow stale-mark failure log, got logs: %s", logs)
533+
}
534+
}
535+
446536
func TestExecutionCleanupService_CleanupLoop_StopsOnContextCancellation(t *testing.T) {
447537
logBuffer := setupExecutionCleanupTestLogger(t)
448538
store := &cleanupStoreMock{}

‎control-plane/internal/server/server_routes_test.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,9 @@ func (s *stubStorage) CleanupOldExecutions(ctx context.Context, retentionPeriod
119119
func (s *stubStorage) MarkStaleExecutions(ctx context.Context, staleAfter time.Duration, limit int) (int, error) {
120120
return 0, nil
121121
}
122+
func (s *stubStorage) MarkStaleWorkflowExecutions(ctx context.Context, staleAfter time.Duration, limit int) (int, error) {
123+
return 0, nil
124+
}
122125
func (s *stubStorage) CleanupWorkflow(ctx context.Context, workflowID string, dryRun bool) (*types.WorkflowCleanupResult, error) {
123126
return &types.WorkflowCleanupResult{
124127
Success: true,

‎control-plane/internal/storage/execution_records.go‎

Lines changed: 116 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -930,6 +930,12 @@ func parseTimeString(value string) (time.Time, error) {
930930
}
931931

932932
// MarkStaleExecutions updates executions stuck in non-terminal states beyond the provided timeout.
933+
// Staleness is determined by updated_at (last activity) rather than started_at, so legitimately
934+
// long-running executions that are still making progress are not incorrectly timed out.
935+
//
936+
// INVARIANT: callers must ensure updated_at is bumped on every meaningful execution activity.
937+
// If updated_at is not maintained, active executions may be incorrectly reaped.
938+
// Uses COALESCE(updated_at, created_at, started_at) to handle rows where updated_at may be NULL.
933939
func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time.Duration, limit int) (int, error) {
934940
if limit <= 0 {
935941
return 0, nil
@@ -945,8 +951,8 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time
945951
SELECT execution_id, started_at
946952
FROM executions
947953
WHERE status IN ('running', 'pending', 'queued')
948-
AND started_at <= ?
949-
ORDER BY started_at ASC
954+
AND COALESCE(updated_at, created_at, started_at) <= ?
955+
ORDER BY COALESCE(updated_at, created_at, started_at) ASC
950956
LIMIT ?`, cutoff, limit)
951957
if err != nil {
952958
return 0, fmt.Errorf("query stale executions: %w", err)
@@ -990,7 +996,7 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time
990996
defer updateStmt.Close()
991997

992998
now := time.Now().UTC()
993-
timeoutMessage := "execution timed out"
999+
timeoutMessage := "execution timed out (no activity)"
9941000

9951001
updated := 0
9961002
for _, rec := range stale {
@@ -1032,6 +1038,113 @@ func (ls *LocalStorage) MarkStaleExecutions(ctx context.Context, staleAfter time
10321038
return updated, nil
10331039
}
10341040

1041+
// MarkStaleWorkflowExecutions updates workflow executions stuck in non-terminal states
1042+
// when their updated_at timestamp exceeds the staleAfter threshold. This catches orphaned
1043+
// child executions whose parent failed without cascading cancellation.
1044+
//
1045+
// See MarkStaleExecutions for the updated_at invariant and COALESCE fallback rationale.
1046+
func (ls *LocalStorage) MarkStaleWorkflowExecutions(ctx context.Context, staleAfter time.Duration, limit int) (int, error) {
1047+
if limit <= 0 {
1048+
return 0, nil
1049+
}
1050+
if err := ctx.Err(); err != nil {
1051+
return 0, fmt.Errorf("context cancelled before marking stale workflow executions: %w", err)
1052+
}
1053+
1054+
cutoff := time.Now().UTC().Add(-staleAfter)
1055+
1056+
db := ls.requireSQLDB()
1057+
rows, err := db.QueryContext(ctx, `
1058+
SELECT execution_id, started_at
1059+
FROM workflow_executions
1060+
WHERE status IN ('running', 'pending', 'queued', 'waiting')
1061+
AND COALESCE(updated_at, created_at, started_at) <= ?
1062+
ORDER BY COALESCE(updated_at, created_at, started_at) ASC
1063+
LIMIT ?`, cutoff, limit)
1064+
if err != nil {
1065+
return 0, fmt.Errorf("query stale workflow executions: %w", err)
1066+
}
1067+
defer rows.Close()
1068+
1069+
type staleRecord struct {
1070+
id string
1071+
startedAt time.Time
1072+
}
1073+
1074+
var stale []staleRecord
1075+
for rows.Next() {
1076+
var rec staleRecord
1077+
if err := rows.Scan(&rec.id, &rec.startedAt); err != nil {
1078+
return 0, fmt.Errorf("scan stale workflow execution: %w", err)
1079+
}
1080+
stale = append(stale, rec)
1081+
}
1082+
if err := rows.Err(); err != nil {
1083+
return 0, fmt.Errorf("iterate stale workflow executions: %w", err)
1084+
}
1085+
1086+
if len(stale) == 0 {
1087+
return 0, nil
1088+
}
1089+
1090+
tx, err := db.BeginTx(ctx, nil)
1091+
if err != nil {
1092+
return 0, fmt.Errorf("begin stale workflow execution transaction: %w", err)
1093+
}
1094+
defer rollbackTx(tx, "MarkStaleWorkflowExecutions")
1095+
1096+
updateStmt, err := tx.PrepareContext(ctx, `
1097+
UPDATE workflow_executions
1098+
SET status = ?, error_message = ?, completed_at = ?, duration_ms = ?, updated_at = ?
1099+
WHERE execution_id = ? AND status IN ('running', 'pending', 'queued', 'waiting')`)
1100+
if err != nil {
1101+
return 0, fmt.Errorf("prepare stale workflow execution update: %w", err)
1102+
}
1103+
defer updateStmt.Close()
1104+
1105+
now := time.Now().UTC()
1106+
timeoutMessage := "execution timed out (no activity)"
1107+
1108+
updated := 0
1109+
for _, rec := range stale {
1110+
duration := now.Sub(rec.startedAt)
1111+
if duration < 0 {
1112+
duration = 0
1113+
}
1114+
durationMS := int(duration.Milliseconds())
1115+
if durationMS < 0 {
1116+
durationMS = 0
1117+
}
1118+
1119+
result, err := updateStmt.ExecContext(
1120+
ctx,
1121+
types.ExecutionStatusTimeout,
1122+
timeoutMessage,
1123+
now,
1124+
durationMS,
1125+
now,
1126+
rec.id,
1127+
)
1128+
if err != nil {
1129+
return 0, fmt.Errorf("update stale workflow execution %s: %w", rec.id, err)
1130+
}
1131+
1132+
rowsAffected, err := result.RowsAffected()
1133+
if err != nil {
1134+
return 0, fmt.Errorf("rows affected for workflow execution %s: %w", rec.id, err)
1135+
}
1136+
if rowsAffected > 0 {
1137+
updated++
1138+
}
1139+
}
1140+
1141+
if err := tx.Commit(); err != nil {
1142+
return 0, fmt.Errorf("commit stale workflow execution transaction: %w", err)
1143+
}
1144+
1145+
return updated, nil
1146+
}
1147+
10351148
func scanExecution(scanner interface {
10361149
Scan(dest ...interface{}) error
10371150
}) (*types.Execution, error) {

0 commit comments

Comments
 (0)