diff --git a/services/core/README.md b/services/core/README.md index b0e14ab..e052959 100644 --- a/services/core/README.md +++ b/services/core/README.md @@ -198,3 +198,18 @@ NomadCode Core의 작업 라이프사이클 관리는 다음과 같은 책임 - **Lifecycle**: 작업 상태 전이(Status Transitions) 및 메타데이터 병합 규칙(Metadata Merge Rules)의 Canonical State 처리를 독점적으로 소유합니다. 상태 변경은 반드시 Lifecycle 헬퍼를 경유해야 합니다. - **Scheduler Worker**: 작업의 실행 단계만 담당합니다. 비동기 큐에서 작업을 꺼내어 `StartTask`를 호출하고, 실제 Agent/Model 실행을 수행한 후, 성공 시 `CompleteTask`, 실패 시 `FailTask`를 호출하여 실행 단계를 마무리합니다. - **Provider Projection**: 외부 협업 도구(Plane, Mattermost 등)로 상태나 코멘트를 반영(Projection)하는 동작의 실패가 Core의 정규 작업 상태(Canonical Task State)를 롤백해서는 안 됩니다. 외부 투영 실패는 재시도 가능한 독립적인 부수 효과로 처리됩니다. + +## 최소 재시도 및 타임아웃 정책 (Minimum Retry and Timeout Policy) + +NomadCode Core의 비동기 작업 재시도 및 타임아웃 처리는 다음과 같은 규칙에 따라 작동합니다. + +- **단일 시도 타임아웃 (Timeout wraps a single worker execution attempt)**: + - 개별 작업 실행 시도는 `WORKFLOW_TASK_TIMEOUT_SEC` (기본값: 300초) 설정 범위 내에서 래핑되어 실행됩니다. + - 이 제한을 초과하면 단일 실행이 실패한 것으로 간주하여 에러를 반환하고, 작업 상태를 `failed`로 변경합니다. 이때 실패 분류 메타데이터로 `failure_type=timeout`, `status_reason=timeout`이 기록됩니다. +- **최대 시도 횟수 (Max Attempts)**: + - River 비동기 작업 큐는 실패한 작업 시도를 최대 `workflow.DefaultTaskMaxAttempts` (기본값: 3회)까지 자동으로 다시 시도합니다. + - 각 시도가 시작될 때마다 작업 메타데이터의 `attempt` 카운트가 1씩 증가하여 기록되며, 실패 시 메타데이터의 `retryable` 필드에 남은 재시도 가능 여부(`attempt < DefaultTaskMaxAttempts`)가 기록됩니다. +- **재-Enqueue 가능 여부 (Re-Enqueue Behavior)**: + - 최종적으로 실패 상태(`failed`)인 작업은 다시 `queued`로 Enqueue하여 다시 처음부터 실행할 수 있습니다. + - 반면 완료(`completed`) 또는 취소(`canceled`)된 터미널 상태의 작업은 다시 시작(Restart/Re-enqueue)할 수 없습니다. + diff --git a/services/core/internal/scheduler/jobs_test.go b/services/core/internal/scheduler/jobs_test.go index bb2db54..8d70ac2 100644 --- a/services/core/internal/scheduler/jobs_test.go +++ b/services/core/internal/scheduler/jobs_test.go @@ -263,6 +263,9 @@ func TestWorkMarksTimeoutFailure(t *testing.T) { if err == nil { t.Fatal("expected Work to return timeout error") } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("expected context.DeadlineExceeded error, got: %v", err) + } if len(fakeLifecycle.started) != 1 || fakeLifecycle.started[0] != "task-1" { t.Fatalf("expected started to have task-1, got %v", fakeLifecycle.started) diff --git a/services/core/internal/scheduler/river.go b/services/core/internal/scheduler/river.go index deb47a3..164912a 100644 --- a/services/core/internal/scheduler/river.go +++ b/services/core/internal/scheduler/river.go @@ -14,6 +14,7 @@ import ( "github.com/nomadcode/nomadcode-core/internal/agent" "github.com/nomadcode/nomadcode-core/internal/model" "github.com/nomadcode/nomadcode-core/internal/notification" + "github.com/nomadcode/nomadcode-core/internal/workflow" ) type Client struct { @@ -40,7 +41,7 @@ func New(pool *pgxpool.Pool, lifecycle TaskLifecycle, notifications *notificatio river.QueueDefault: {MaxWorkers: 2}, }, Workers: workers, - MaxAttempts: 3, + MaxAttempts: workflow.DefaultTaskMaxAttempts, }) if err != nil { return nil, err diff --git a/services/core/internal/workflow/lifecycle.go b/services/core/internal/workflow/lifecycle.go index ab2e006..9893f7c 100644 --- a/services/core/internal/workflow/lifecycle.go +++ b/services/core/internal/workflow/lifecycle.go @@ -96,6 +96,30 @@ func nextTaskAttempt(metadata json.RawMessage) (int, error) { return attempt, nil } +func currentTaskAttempt(metadata json.RawMessage) int { + if len(metadata) == 0 || string(metadata) == "null" { + return 0 + } + var currentMap map[string]any + if err := json.Unmarshal(metadata, ¤tMap); err != nil { + return 0 + } + if currentMap != nil { + if val, ok := currentMap[MetadataKeyAttempt]; ok { + switch v := val.(type) { + case float64: + return int(v) + case int: + return v + case int64: + return int(v) + } + } + } + return 0 +} + + func mergeTaskMetadata(current json.RawMessage, updates map[string]any) (json.RawMessage, error) { var currentMap map[string]any if len(current) > 0 && string(current) != "null" { @@ -231,12 +255,19 @@ func (l *Lifecycle) FailTaskWithMetadata(ctx context.Context, id string, input F failType = FailureTypeExecution } + attempt := currentTaskAttempt(task.Metadata) + if attempt == 0 { + attempt = 1 + } + retryable := attempt < DefaultTaskMaxAttempts + updates := map[string]any{ MetadataKeyAgentRunState: "failed", MetadataKeyStatusReason: msg, MetadataKeyWaitType: nil, MetadataKeyFailureType: string(failType), MetadataKeyFailedAt: time.Now().UTC().Format(time.RFC3339), + MetadataKeyRetryable: retryable, } newMeta, err := mergeTaskMetadata(task.Metadata, updates) diff --git a/services/core/internal/workflow/model.go b/services/core/internal/workflow/model.go index dcb5c68..d85d045 100644 --- a/services/core/internal/workflow/model.go +++ b/services/core/internal/workflow/model.go @@ -36,6 +36,8 @@ const ( FailureTypeEnqueue FailureType = "enqueue" ) +const DefaultTaskMaxAttempts = 3 + const ( MetadataKeyAgentRunState = "agent_run_state" MetadataKeyAgentPhase = "agent_phase" @@ -46,5 +48,7 @@ const ( MetadataKeyLastHeartbeat = "last_heartbeat_at" MetadataKeyPlanRef = "plan_ref" MetadataKeyAttempt = "attempt" + MetadataKeyRetryable = "retryable" ) + diff --git a/services/core/internal/workflow/service_test.go b/services/core/internal/workflow/service_test.go index 3c2191e..62642d4 100644 --- a/services/core/internal/workflow/service_test.go +++ b/services/core/internal/workflow/service_test.go @@ -540,6 +540,30 @@ func TestLifecycleFailTaskStoresFailureMetadataAndError(t *testing.T) { if _, ok := meta[MetadataKeyFailedAt]; !ok { t.Errorf("expected failed_at to be set in metadata") } + if meta[MetadataKeyRetryable] != true { + t.Errorf("expected retryable to be true, got: %v", meta[MetadataKeyRetryable]) + } + + // Test when attempt is equal to DefaultTaskMaxAttempts (not retryable) + store.tasks["task-fail-max-attempts"] = storage.Task{ + ID: "task-fail-max-attempts", + Status: string(StatusRunning), + Metadata: json.RawMessage(`{"attempt":3,"agent_run_state":"running"}`), + } + taskMax, err := l.FailTaskWithMetadata(ctx, "task-fail-max-attempts", FailureInput{ + Message: "max attempts reached", + Type: FailureTypeExecution, + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + var metaMax map[string]any + if err := json.Unmarshal(taskMax.Metadata, &metaMax); err != nil { + t.Fatalf("failed to unmarshal metadata: %v", err) + } + if metaMax[MetadataKeyRetryable] != false { + t.Errorf("expected retryable to be false, got: %v", metaMax[MetadataKeyRetryable]) + } // Test trimming and empty/default behavior store.tasks["task-fail-empty"] = storage.Task{