diff --git a/docs/architecture.md b/docs/architecture.md index b05483a9..794b9fc0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -393,13 +393,40 @@ heartbeat-sweep tick that handles offline-worker cleanup, controlled by └──────────┘ └──────────┘ └──────────────┘ ``` -Transitions are validated — only the arrows above are permitted. Any other -transition returns an error from `store.TransitionTask`. The auto-retry -re-queue described below (`running` → `ready` on a transient failure) is a -**separate, policy-driven store call** (`RequeueTaskForRetry`), not a -`TransitionTask` arrow — the diagram above still reflects every state a task -can be validated *into* via `TransitionTask`; auto-retry sends a task back to -`ready` directly instead of landing it on `failed` at all. +Transitions are validated by `store.ValidateTaskTransition` +(`internal/store/statemachine.go`) and enforced by `UpdateTaskStatus` in both +store implementations: the SQLite store reads the current status and writes the +new one inside a single transaction, so the check cannot race a concurrent +writer, and the in-memory fake does the same under its mutex. A transition +outside the permitted set returns `store.ErrInvalidTransition` and leaves the +row unchanged. + +Two rules keep enforcement safe given that task status arrives over JetStream +(at-least-once delivery): + +- **Writing a task's current status is a no-op, not an error** — a redelivered + message must not fail. +- **The consumer acks an invalid transition instead of Nak'ing it.** A message + describing a state the task has already left cannot become legal on + redelivery, so Nak'ing would loop forever. It is discarded with a warning, + the same treatment a malformed payload gets. + +Cancellation follows the same principle. `CancelTask` checks for a terminal +status before writing, but the check and the write are separate operations, so +a task can finish in between and the state machine then rejects the cancel. +That is treated as the no-op it would have been had the check seen the newer +value — canceling a completed task is not an error, regardless of which side +of the race the caller landed on. Other store failures still propagate. + +Two arrows deserve note. `assigned` → `succeeded`/`failed` is permitted even +though it appears to skip `running`: the worker publishes `running` first, but +that publish is best-effort and gives up after `MaxRetries`, so it can be lost +while the task still runs to completion. Rejecting the terminal message would +strand finished work. Separately, the auto-retry re-queue described below +(`running` → `ready` on a transient failure) is a **policy-driven store call** +(`RequeueTaskForRetry`) with its own guarded SQL, as are the other bulk paths +(`RetryTasks`, `TransitionStepPendingTasks`, `CancelJobTasks`, and the reclaim +sweeps); none of them route through `UpdateTaskStatus`. ### Auto-retry on worker-reported failure diff --git a/internal/api/tasks_test.go b/internal/api/tasks_test.go index 625e6fc2..be2a01c4 100644 --- a/internal/api/tasks_test.go +++ b/internal/api/tasks_test.go @@ -53,7 +53,25 @@ func (f *fakeTaskCanceler) RetryTask(ctx context.Context, id string) error { if status == "" { status = store.TaskStatusReady } - return f.retryStore.UpdateTaskStatus(ctx, id, status) + // Revive through RetryTasks, the same store call the real scheduler + // makes. UpdateTaskStatus would be wrong here: it enforces the task + // state machine, and failed → ready is not an arrow — production + // revives failed → pending inside RetryTasks and then promotes to ready + // via dependency resolution. Using UpdateTaskStatus made this double + // exercise a transition the real store rejects. + task, err := f.retryStore.GetTask(ctx, id) + if err != nil { + return err + } + if _, err := f.retryStore.RetryTasks(ctx, task.JobID, []string{id}, time.Now()); err != nil { + return err + } + if status != store.TaskStatusPending { + // RetryTasks lands on pending; walk the legal pending → ready arrow + // when the test wants the post-resolution status. + return f.retryStore.UpdateTaskStatus(ctx, id, status) + } + return nil } return nil } diff --git a/internal/openjd/statemachine.go b/internal/openjd/statemachine.go index 2d458f8f..b75decfe 100644 --- a/internal/openjd/statemachine.go +++ b/internal/openjd/statemachine.go @@ -9,72 +9,17 @@ import ( "github.com/uberware/sqi/internal/store" ) -// ErrInvalidTransition is returned when a requested status transition is not -// permitted by the task or step state machine. +// ErrInvalidTransition is returned when a requested step status transition is +// not permitted by the step state machine. // // Use errors.Is to test: // -// err := ValidateTaskTransition(from, to) +// err := ValidateStepTransition(from, to) // if errors.Is(err, ErrInvalidTransition) { ... } -var ErrInvalidTransition = errors.New("openjd: invalid state transition") - -// ── Task state machine ──────────────────────────────────────────────────────── -// -// Permitted task status transitions: // -// pending → ready dependency resolution: all dependency steps completed -// pending → canceled job canceled before step dependencies were satisfied -// ready → assigned scheduler assigns task to a worker -// ready → canceled task canceled while waiting for a worker -// assigned → running worker confirms execution has started -// assigned → ready reassignment: assigned worker disconnected or timed out -// assigned → canceled task canceled after assignment but before confirmation -// running → succeeded worker reports clean exit (exit code 0) -// running → failed worker reports non-zero exit or fatal error -// running → ready reassignment: running worker became unreachable -// running → canceled task canceled while executing -// -// Terminal states (succeeded, failed, canceled) have no outgoing transitions. - -var validTaskTransitions = map[store.TaskStatus]map[store.TaskStatus]struct{}{ - store.TaskStatusPending: { - store.TaskStatusReady: {}, - store.TaskStatusCanceled: {}, - }, - store.TaskStatusReady: { - store.TaskStatusAssigned: {}, - store.TaskStatusCanceled: {}, - }, - store.TaskStatusAssigned: { - store.TaskStatusRunning: {}, - store.TaskStatusReady: {}, - store.TaskStatusCanceled: {}, - }, - store.TaskStatusRunning: { - store.TaskStatusSucceeded: {}, - store.TaskStatusFailed: {}, - store.TaskStatusReady: {}, - store.TaskStatusCanceled: {}, - }, - // Terminal states — no outgoing transitions. - store.TaskStatusSucceeded: {}, - store.TaskStatusFailed: {}, - store.TaskStatusCanceled: {}, -} - -// ValidateTaskTransition returns nil if transitioning a task from old to new -// status is permitted by the state machine, or a descriptive error wrapping -// [ErrInvalidTransition] otherwise. -func ValidateTaskTransition(from, to store.TaskStatus) error { - targets, known := validTaskTransitions[from] - if !known { - return fmt.Errorf("%w: unknown task status %q", ErrInvalidTransition, from) - } - if _, ok := targets[to]; ok { - return nil - } - return fmt.Errorf("%w: task %q → %q not permitted", ErrInvalidTransition, from, to) -} +// The task state machine lives in package store, which enforces it on every +// status write, and carries its own [store.ErrInvalidTransition]. +var ErrInvalidTransition = errors.New("openjd: invalid state transition") // ── Step state machine ──────────────────────────────────────────────────────── // diff --git a/internal/openjd/statemachine_test.go b/internal/openjd/statemachine_test.go index 1d6a9355..2e810b7e 100644 --- a/internal/openjd/statemachine_test.go +++ b/internal/openjd/statemachine_test.go @@ -14,68 +14,6 @@ import ( "github.com/uberware/sqi/internal/store" ) -// ── Task transitions ────────────────────────────────────────────────────────── - -func TestValidateTaskTransition(t *testing.T) { - legal := []struct { - from store.TaskStatus - to store.TaskStatus - }{ - {store.TaskStatusPending, store.TaskStatusReady}, - {store.TaskStatusPending, store.TaskStatusCanceled}, - {store.TaskStatusReady, store.TaskStatusAssigned}, - {store.TaskStatusReady, store.TaskStatusCanceled}, - {store.TaskStatusAssigned, store.TaskStatusRunning}, - {store.TaskStatusAssigned, store.TaskStatusReady}, - {store.TaskStatusAssigned, store.TaskStatusCanceled}, - {store.TaskStatusRunning, store.TaskStatusSucceeded}, - {store.TaskStatusRunning, store.TaskStatusFailed}, - {store.TaskStatusRunning, store.TaskStatusReady}, - {store.TaskStatusRunning, store.TaskStatusCanceled}, - } - for _, tc := range legal { - if err := openjd.ValidateTaskTransition(tc.from, tc.to); err != nil { - t.Errorf("expected legal transition %q→%q, got error: %v", tc.from, tc.to, err) - } - } - - illegal := []struct { - from store.TaskStatus - to store.TaskStatus - }{ - {store.TaskStatusPending, store.TaskStatusRunning}, - {store.TaskStatusPending, store.TaskStatusSucceeded}, - {store.TaskStatusReady, store.TaskStatusRunning}, - {store.TaskStatusReady, store.TaskStatusSucceeded}, - {store.TaskStatusSucceeded, store.TaskStatusRunning}, - {store.TaskStatusSucceeded, store.TaskStatusFailed}, - {store.TaskStatusFailed, store.TaskStatusRunning}, - {store.TaskStatusFailed, store.TaskStatusSucceeded}, - {store.TaskStatusCanceled, store.TaskStatusRunning}, - {store.TaskStatusCanceled, store.TaskStatusSucceeded}, - } - for _, tc := range illegal { - err := openjd.ValidateTaskTransition(tc.from, tc.to) - if err == nil { - t.Errorf("expected error for illegal transition %q→%q, got nil", tc.from, tc.to) - continue - } - if !errors.Is(err, openjd.ErrInvalidTransition) { - t.Errorf("transition %q→%q: error %v should wrap ErrInvalidTransition", tc.from, tc.to, err) - } - } -} - -func TestValidateTaskTransition_UnknownStatus(t *testing.T) { - err := openjd.ValidateTaskTransition("bogus", store.TaskStatusReady) - if err == nil { - t.Fatal("expected error for unknown status, got nil") - } - if !errors.Is(err, openjd.ErrInvalidTransition) { - t.Errorf("expected ErrInvalidTransition, got %v", err) - } -} - // ── Step transitions ────────────────────────────────────────────────────────── func TestValidateStepTransition(t *testing.T) { diff --git a/internal/scheduler/cancellation.go b/internal/scheduler/cancellation.go index 941660aa..0d701e5d 100644 --- a/internal/scheduler/cancellation.go +++ b/internal/scheduler/cancellation.go @@ -114,7 +114,10 @@ func (s *Scheduler) CancelJob(ctx context.Context, jobID string) error { // task.cancel. signal to the assigned worker. // // If the task is already in a terminal state (succeeded, failed, canceled), -// CancelTask returns nil without modifying any state. +// CancelTask returns nil without modifying any state. That holds whether the +// terminal state was visible up front or the task reached it mid-cancel: the +// guard below and the status write are separate operations, and losing that +// race is reported the same way as never having had it. func (s *Scheduler) CancelTask(ctx context.Context, taskID string) error { now := time.Now().UTC() @@ -148,6 +151,24 @@ func (s *Scheduler) CancelTask(ctx context.Context, taskID string) error { } if err = s.store.UpdateTaskStatus(ctx, taskID, store.TaskStatusCanceled); err != nil { + // Losing a race to completion is a no-op, not a failure. The terminal + // guard above is a separate read, so the task can finish between that + // read and this write; the state machine then rejects it. Canceling an + // already-terminal task is documented to return nil, and which side of + // the race the caller landed on must not change that. + // + // Narrow by construction: every non-terminal status has a legal arrow + // to canceled, so ErrInvalidTransition on *this* write can only mean + // the task is already terminal. Any other store failure still + // propagates. + if errors.Is(err, store.ErrInvalidTransition) { + s.logger.DebugContext( + ctx, "scheduler: cancel task — reached terminal state first", + slog.String("task_id", taskID), + slog.Any("error", err), + ) + return nil + } return fmt.Errorf("scheduler: transition task %s to canceled: %w", taskID, err) } diff --git a/internal/scheduler/cancellation_race_test.go b/internal/scheduler/cancellation_race_test.go new file mode 100644 index 00000000..77492c3d --- /dev/null +++ b/internal/scheduler/cancellation_race_test.go @@ -0,0 +1,111 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package scheduler + +// CancelTask's "already terminal" guard is a read followed by a separate write, +// so a task can reach a terminal state in between: the guard sees `running` and +// lets the cancel through, but by the time UpdateTaskStatus runs the row is +// `succeeded` and the state machine rejects the write. +// +// The documented contract is that canceling an already-terminal task is a +// silent no-op. Losing that race must therefore behave the same way it would +// have if the guard had seen the newer value — return nil — not surface a 500 +// to the caller. +// +// staleReadStore reproduces the race deterministically. The underlying task is +// already terminal; the wrapper's GetTask hands back the pre-completion status, +// standing in for a guard that read a moment too early. No sleeps, no +// goroutines, no flakiness. + +import ( + "context" + "errors" + "testing" + + "github.com/uberware/sqi/internal/store" + "github.com/uberware/sqi/internal/store/fake" +) + +type staleReadStore struct { + store.Store + + taskID string + staleStatus store.TaskStatus +} + +func (s *staleReadStore) GetTask(ctx context.Context, id string) (store.Task, error) { + task, err := s.Store.GetTask(ctx, id) + if err != nil || id != s.taskID { + return task, err + } + task.Status = s.staleStatus // what the guard would have read pre-completion + return task, nil +} + +func TestCancelTask_LosesRaceToCompletion_IsNoOp(t *testing.T) { + for _, terminal := range []store.TaskStatus{ + store.TaskStatusSucceeded, + store.TaskStatusFailed, + } { + t.Run(string(terminal), func(t *testing.T) { + st := fake.New() + bus := &stubBus{} + job := seedCancelJob(t, st) + tk := seedTaskForJob(t, st, job, "w1", terminal) + + // The guard reads "running"; the row is already terminal. + s := newTestScheduler(&staleReadStore{ + Store: st, + taskID: tk.ID, + staleStatus: store.TaskStatusRunning, + }, bus) + + if err := s.CancelTask(t.Context(), tk.ID); err != nil { + t.Fatalf("CancelTask losing the race to completion = %v, want nil (no-op)", err) + } + + stored, err := st.GetTask(t.Context(), tk.ID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if stored.Status != terminal { + t.Errorf("status = %q, want %q — a completed task must not be overwritten by a losing cancel", + stored.Status, terminal) + } + }) + } +} + +// TestCancelTask_RealErrorStillPropagates guards against the fix being written +// as a blanket "swallow every error from UpdateTaskStatus". +func TestCancelTask_RealErrorStillPropagates(t *testing.T) { + st := fake.New() + bus := &stubBus{} + job := seedCancelJob(t, st) + tk := seedTaskForJob(t, st, job, "w1", store.TaskStatusRunning) + + s := newTestScheduler(&failingUpdateStore{Store: st, taskID: tk.ID}, bus) + + err := s.CancelTask(t.Context(), tk.ID) + if err == nil { + t.Fatal("CancelTask = nil, want the underlying store error to propagate") + } + if errors.Is(err, store.ErrInvalidTransition) { + t.Errorf("error = %v, want the store failure, not ErrInvalidTransition", err) + } +} + +var errStoreUnavailable = errors.New("store unavailable") + +type failingUpdateStore struct { + store.Store + + taskID string +} + +func (s *failingUpdateStore) UpdateTaskStatus(ctx context.Context, id string, status store.TaskStatus) error { + if id == s.taskID { + return errStoreUnavailable + } + return s.Store.UpdateTaskStatus(ctx, id, status) +} diff --git a/internal/scheduler/failure_test.go b/internal/scheduler/failure_test.go index 7663ed78..c092baf4 100644 --- a/internal/scheduler/failure_test.go +++ b/internal/scheduler/failure_test.go @@ -192,8 +192,37 @@ func (h *failureHarness) reportFailedWithMessage(taskID, message string) { // reassignAndReportFailed simulates a worker re-leasing the retried task: it // opens a new attempt on workerID, then reports that attempt failed. +// reassignAndReportFailed models a genuine second attempt: a task sitting in +// ready after a retry is assigned to a worker and starts running before it +// fails again. The assigned/running steps are not decoration — UpdateTaskStatus +// enforces the state machine, and ready → failed is not an arrow. Skipping them +// would have this helper exercise a transition the store rejects (and rightly: +// a "failed" landing on a ready task means a stale attempt's message was +// redelivered after a retry already revived the task, which must not re-fail +// it). func (h *failureHarness) reassignAndReportFailed(taskID, workerID string) { h.t.Helper() + ctx := h.t.Context() + task, err := h.st.GetTask(ctx, taskID) + if err != nil { + h.t.Fatalf("reassignAndReportFailed: GetTask: %v", err) + } + var path []store.TaskStatus + switch task.Status { + case store.TaskStatusReady: + path = []store.TaskStatus{store.TaskStatusAssigned, store.TaskStatusRunning} + case store.TaskStatusAssigned: + path = []store.TaskStatus{store.TaskStatusRunning} + case store.TaskStatusRunning: + // already executing; the caller is driving a repeat failure + default: + h.t.Fatalf("reassignAndReportFailed: cannot reassign from %q", task.Status) + } + for _, st := range path { + if err := h.st.UpdateTaskStatus(ctx, taskID, st); err != nil { + h.t.Fatalf("reassignAndReportFailed: UpdateTaskStatus(%q): %v", st, err) + } + } h.newAttempt(taskID, workerID) h.reportFailed(taskID) } diff --git a/internal/scheduler/taskstatus.go b/internal/scheduler/taskstatus.go index 51856f0e..1e651088 100644 --- a/internal/scheduler/taskstatus.go +++ b/internal/scheduler/taskstatus.go @@ -79,6 +79,21 @@ func (s *Scheduler) handleTaskStatusMessage(msg jetstream.Msg) { } if err := s.processTaskStatus(ctx, m); err != nil { + // An illegal transition is permanent: the task has moved on (retried, + // canceled, already terminal) and this message describes a past state. + // Redelivering cannot make it legal, so discard it rather than Nak into + // an infinite loop — the same reasoning as a malformed payload above. + if errors.Is(err, store.ErrInvalidTransition) { + s.logger.WarnContext( + ctx, "scheduler: task status rejected by state machine — discarding", + slog.String("task_id", m.TaskID), + slog.String("attempt_id", m.AttemptID), + slog.String("status", m.Status), + slog.Any("error", err), + ) + s.ackMsg(ctx, msg) + return + } s.logger.WarnContext( ctx, "scheduler: process task status failed — will redeliver", slog.String("task_id", m.TaskID), diff --git a/internal/scheduler/taskstatus_transition_test.go b/internal/scheduler/taskstatus_transition_test.go new file mode 100644 index 00000000..ca297730 --- /dev/null +++ b/internal/scheduler/taskstatus_transition_test.go @@ -0,0 +1,99 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package scheduler + +// The task status consumer must ACK a message whose transition the store +// rejects, never Nak it. +// +// UpdateTaskStatus enforces the task state machine, so a stale or out-of-order +// worker message can now legitimately fail. task.status is a JetStream subject +// (at-least-once), and handleTaskStatusMessage Naks on error — so treating an +// invalid transition as retryable would redeliver the same doomed message +// forever. Redelivery cannot fix a transition that is illegal, exactly as it +// cannot fix a malformed payload. + +import ( + "testing" + + "github.com/uberware/sqi/internal/store" + "github.com/uberware/sqi/internal/store/fake" + "github.com/uberware/sqi/internal/worker/protocol" +) + +// TestHandleTaskStatusMessage_InvalidTransitionIsAcked drives a "failed" +// message at a task that has already succeeded — the shape of a redelivered +// message arriving after the task reached a terminal state. +func TestHandleTaskStatusMessage_InvalidTransitionIsAcked(t *testing.T) { + st := fake.New() + s := newStatusTestScheduler(st) + s.ctx = t.Context() + + _, _, task, attempt := seedStatusFixture(t, st, store.TaskStatusRunning) + + // Drive the task to a terminal state first. + if err := st.UpdateTaskStatus(t.Context(), task.ID, store.TaskStatusSucceeded); err != nil { + t.Fatalf("UpdateTaskStatus(running → succeeded): %v", err) + } + + msg := &fakeJSMsg{ + data: taskStatusMsgJSON(t, protocol.TaskStatusMsg{ + TaskID: task.ID, + AttemptID: attempt.ID, + Status: "failed", + }), + } + s.handleTaskStatusMessage(msg) + + if msg.nacked { + t.Error("invalid transition was Naked — it would redeliver forever") + } + if !msg.acked { + t.Error("invalid transition should be acked (discarded); redelivery cannot fix it") + } + + // The terminal status must be untouched by the rejected message. + stored, err := st.GetTask(t.Context(), task.ID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if stored.Status != store.TaskStatusSucceeded { + t.Errorf("status = %q after rejected message, want succeeded (unchanged)", stored.Status) + } +} + +// TestHandleTaskStatusMessage_DuplicateRunningIsAcked covers the ordinary +// at-least-once case: the same "running" message delivered twice must ack both +// times, because a same-status write is a no-op rather than an error. +func TestHandleTaskStatusMessage_DuplicateRunningIsAcked(t *testing.T) { + st := fake.New() + s := newStatusTestScheduler(st) + s.ctx = t.Context() + + _, _, task, attempt := seedStatusFixture(t, st, store.TaskStatusAssigned) + + for i := range 2 { + msg := &fakeJSMsg{ + data: taskStatusMsgJSON(t, protocol.TaskStatusMsg{ + TaskID: task.ID, + AttemptID: attempt.ID, + Status: "running", + }), + } + s.handleTaskStatusMessage(msg) + + if msg.nacked { + t.Fatalf("delivery %d: running message was Naked", i+1) + } + if !msg.acked { + t.Fatalf("delivery %d: running message should be acked", i+1) + } + } + + stored, err := st.GetTask(t.Context(), task.ID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if stored.Status != store.TaskStatusRunning { + t.Errorf("status = %q, want running", stored.Status) + } +} diff --git a/internal/store/fake/store_methods_test.go b/internal/store/fake/store_methods_test.go index 5f43d6f3..6b5449d6 100644 --- a/internal/store/fake/store_methods_test.go +++ b/internal/store/fake/store_methods_test.go @@ -308,7 +308,9 @@ func TestUnschedulableReason_ClearedOnUpdateStatus(t *testing.T) { t.Fatalf("SetTaskUnschedulableReason: %v", err) } - if err := s.UpdateTaskStatus(ctx(), "t1", store.TaskStatusRunning); err != nil { + // ready → assigned is the legal arrow out of ready; the point of the test is + // that any status change clears the reason, not which status it lands on. + if err := s.UpdateTaskStatus(ctx(), "t1", store.TaskStatusAssigned); err != nil { t.Fatalf("UpdateTaskStatus: %v", err) } diff --git a/internal/store/fake/task.go b/internal/store/fake/task.go index 429197f5..a8f7b58b 100644 --- a/internal/store/fake/task.go +++ b/internal/store/fake/task.go @@ -5,6 +5,7 @@ package fake import ( "cmp" "context" + "fmt" "slices" "time" @@ -63,6 +64,13 @@ func (s *Store) ListTasks(_ context.Context, opts store.ListTasksOptions) (store // unschedulable_reason is only meaningful while a task is ready (set by the // scheduler sweep), so it is cleared here regardless of the destination // status — harmless when it was already empty. +// +// Enforces the task state machine ([store.ValidateTaskTransition]) under the +// same lock that performs the write, matching the SQLite store's transaction. +// Writing the status a task already holds is a no-op, not an error, so +// at-least-once redelivery stays idempotent. Keeping the two implementations in +// step matters: tests inject this fake, and a permissive fake would green-light +// transitions production rejects. func (s *Store) UpdateTaskStatus(_ context.Context, id string, status store.TaskStatus) error { s.mu.Lock() defer s.mu.Unlock() @@ -72,6 +80,13 @@ func (s *Store) UpdateTaskStatus(_ context.Context, id string, status store.Task return store.ErrNotFound } + if task.Status == status { + return nil // idempotent redelivery; nothing to write + } + if err := store.ValidateTaskTransition(task.Status, status); err != nil { + return fmt.Errorf("fake: task %s: %w", id, err) + } + task.Status = status task.UnschedulableReason = "" task.UpdatedAt = time.Now() diff --git a/internal/store/fake/task_transition_test.go b/internal/store/fake/task_transition_test.go new file mode 100644 index 00000000..124a8084 --- /dev/null +++ b/internal/store/fake/task_transition_test.go @@ -0,0 +1,106 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package fake + +// Parity tests: the fake must enforce the task state machine exactly as the +// SQLite store does. Handler and scheduler tests inject this fake, so a fake +// that accepted transitions SQLite rejects would let those tests pass on +// behavior that fails in production. + +import ( + "context" + "errors" + "testing" + + "github.com/uberware/sqi/internal/store" +) + +// fakeTaskAt returns a fake store holding one task walked to want along legal +// arrows only. +func fakeTaskAt(t *testing.T, want store.TaskStatus) (*Store, string) { + t.Helper() + ctx := context.Background() + s := New() + t.Cleanup(func() { _ = s.Close() }) + + if _, err := s.CreateTask(ctx, store.Task{ + ID: "t1", + JobID: "j1", + StepID: "s1", + Name: "t1", + Status: store.TaskStatusPending, + }); err != nil { + t.Fatalf("CreateTask: %v", err) + } + + var path []store.TaskStatus + switch want { + case store.TaskStatusPending: + case store.TaskStatusReady: + path = []store.TaskStatus{store.TaskStatusReady} + case store.TaskStatusAssigned: + path = []store.TaskStatus{store.TaskStatusReady, store.TaskStatusAssigned} + case store.TaskStatusRunning: + path = []store.TaskStatus{store.TaskStatusReady, store.TaskStatusAssigned, store.TaskStatusRunning} + default: + path = []store.TaskStatus{ + store.TaskStatusReady, store.TaskStatusAssigned, store.TaskStatusRunning, want, + } + } + for _, st := range path { + if err := s.UpdateTaskStatus(ctx, "t1", st); err != nil { + t.Fatalf("walk to %q: UpdateTaskStatus(%q): %v", want, st, err) + } + } + return s, "t1" +} + +func TestFakeUpdateTaskStatus_RejectsIllegalTransition(t *testing.T) { + s, id := fakeTaskAt(t, store.TaskStatusSucceeded) + + err := s.UpdateTaskStatus(context.Background(), id, store.TaskStatusReady) + if !errors.Is(err, store.ErrInvalidTransition) { + t.Fatalf("UpdateTaskStatus(succeeded → ready) = %v, want ErrInvalidTransition", err) + } + + got, getErr := s.GetTask(context.Background(), id) + if getErr != nil { + t.Fatalf("GetTask: %v", getErr) + } + if got.Status != store.TaskStatusSucceeded { + t.Errorf("status = %q after rejected transition, want succeeded (unchanged)", got.Status) + } +} + +func TestFakeUpdateTaskStatus_AllowsLegalTransition(t *testing.T) { + s, id := fakeTaskAt(t, store.TaskStatusRunning) + + if err := s.UpdateTaskStatus(context.Background(), id, store.TaskStatusSucceeded); err != nil { + t.Fatalf("UpdateTaskStatus(running → succeeded): %v", err) + } +} + +func TestFakeUpdateTaskStatus_SameStatusIsNoOp(t *testing.T) { + s, id := fakeTaskAt(t, store.TaskStatusRunning) + + if err := s.UpdateTaskStatus(context.Background(), id, store.TaskStatusRunning); err != nil { + t.Errorf("UpdateTaskStatus(running → running) = %v, want nil (no-op)", err) + } +} + +func TestFakeUpdateTaskStatus_AssignedToTerminal(t *testing.T) { + s, id := fakeTaskAt(t, store.TaskStatusAssigned) + + if err := s.UpdateTaskStatus(context.Background(), id, store.TaskStatusSucceeded); err != nil { + t.Errorf("UpdateTaskStatus(assigned → succeeded) = %v, want nil", err) + } +} + +func TestFakeUpdateTaskStatus_NotFound(t *testing.T) { + s, _ := fakeTaskAt(t, store.TaskStatusReady) + + err := s.UpdateTaskStatus(context.Background(), "no-such-task", store.TaskStatusAssigned) + if !errors.Is(err, store.ErrNotFound) { + t.Errorf("UpdateTaskStatus(missing) = %v, want ErrNotFound", err) + } +} diff --git a/internal/store/sqlite/store_test.go b/internal/store/sqlite/store_test.go index 5ec6f74f..ac94c594 100644 --- a/internal/store/sqlite/store_test.go +++ b/internal/store/sqlite/store_test.go @@ -131,6 +131,38 @@ func insertTask(t *testing.T, s *sqlite.Store, id, jobID, stepID string) store.T return task } +// walkTaskTo drives a freshly-inserted (pending) task to the wanted status +// along legal arrows only, so the fixture itself never depends on the +// enforcement being absent. +func walkTaskTo(t *testing.T, s *sqlite.Store, taskID string, want store.TaskStatus) { + t.Helper() + ctx := context.Background() + + var path []store.TaskStatus + switch want { + case store.TaskStatusPending: + return + case store.TaskStatusReady: + path = []store.TaskStatus{store.TaskStatusReady} + case store.TaskStatusAssigned: + path = []store.TaskStatus{store.TaskStatusReady, store.TaskStatusAssigned} + case store.TaskStatusRunning: + path = []store.TaskStatus{store.TaskStatusReady, store.TaskStatusAssigned, store.TaskStatusRunning} + case store.TaskStatusSucceeded, store.TaskStatusFailed, store.TaskStatusCanceled: + path = []store.TaskStatus{ + store.TaskStatusReady, store.TaskStatusAssigned, store.TaskStatusRunning, want, + } + default: + t.Fatalf("walkTaskTo: unsupported target status %q", want) + } + + for _, st := range path { + if err := s.UpdateTaskStatus(ctx, taskID, st); err != nil { + t.Fatalf("walkTaskTo(%q): UpdateTaskStatus(%q): %v", want, st, err) + } + } +} + // ── Farm CRUD ───────────────────────────────────────────────────────────────── func TestFarm_CreateAndGet(t *testing.T) { @@ -1098,9 +1130,7 @@ func TestTask_UpdateStatus(t *testing.T) { insertStep(t, s, "s1", "j1", "S1", 0) insertTask(t, s, "t1", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "t1", store.TaskStatusRunning); err != nil { - t.Fatalf("UpdateTaskStatus: %v", err) - } + walkTaskTo(t, s, "t1", store.TaskStatusRunning) task, err := s.GetTask(ctx, "t1") if err != nil { t.Fatalf("GetTask: %v", err) @@ -1733,9 +1763,7 @@ func TestUnschedulableReason_ClearedOnUpdateStatus(t *testing.T) { t.Fatalf("SetTaskUnschedulableReason: %v", err) } - if err := s.UpdateTaskStatus(ctx, "t1", store.TaskStatusRunning); err != nil { - t.Fatalf("UpdateTaskStatus running: %v", err) - } + walkTaskTo(t, s, "t1", store.TaskStatusRunning) got, err := s.GetTask(ctx, "t1") if err != nil { diff --git a/internal/store/sqlite/task.go b/internal/store/sqlite/task.go index 95d32229..7a6fc74f 100644 --- a/internal/store/sqlite/task.go +++ b/internal/store/sqlite/task.go @@ -5,6 +5,7 @@ package sqlite import ( "context" "database/sql" + "errors" "fmt" "strings" "time" @@ -34,6 +35,11 @@ RETURNING ` + taskCols sqlUpdateTaskStatus = ` UPDATE tasks SET status = ?, updated_at = ?, unschedulable_reason = '' WHERE id = ?` + // sqlSelectTaskStatus reads the current status inside UpdateTaskStatus's + // transaction so the state-machine check and the write are indivisible. + sqlSelectTaskStatus = ` +SELECT status FROM tasks WHERE id = ?` + sqlSetTaskUnschedulableReason = ` UPDATE tasks SET unschedulable_reason = ?, updated_at = ? WHERE id = ?` @@ -428,12 +434,55 @@ func (s *Store) ListTasks(ctx context.Context, opts store.ListTasksOptions) (sto } // UpdateTaskStatus implements [store.TaskStore]. +// +// The write is gated by the task state machine +// ([store.ValidateTaskTransition]): a transition the machine does not permit is +// rejected with [store.ErrInvalidTransition] and leaves the row untouched. +// +// Writing the status a task already holds is a no-op, not an error. Task status +// arrives over JetStream, which is at-least-once, so a redelivered message must +// not fail — the consumer would Nak it and redeliver forever. +// +// The read and the write share a transaction, serialized by the +// single-connection pool (SetMaxOpenConns(1)), so no other goroutine can move +// the task between the check and the update. Mirrors [Store.TryClaimSlots]. func (s *Store) UpdateTaskStatus(ctx context.Context, id string, status store.TaskStatus) error { - res, err := s.stmtUpdateTaskStatus.ExecContext(ctx, string(status), timeToText(time.Now().UTC()), id) + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return fmt.Errorf("sqlite: begin tx for update task status: %w", err) + } + defer func() { _ = tx.Rollback() }() //nolint:errcheck // rollback is best-effort after commit + + var current string + if err = tx.QueryRowContext(ctx, sqlSelectTaskStatus, id).Scan(¤t); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return store.ErrNotFound + } + return fmt.Errorf("sqlite: select task status: %w", mapErr(err)) + } + + if store.TaskStatus(current) == status { + return nil // idempotent redelivery; nothing to write + } + if err = store.ValidateTaskTransition(store.TaskStatus(current), status); err != nil { + return fmt.Errorf("sqlite: task %s: %w", id, err) + } + + // Raw SQL rather than the prepared s.stmtUpdateTaskStatus: a statement + // bound into a transaction with tx.StmtContext must itself be closed, and + // the other transactional writers here use tx.ExecContext for the same + // reason. + res, err := tx.ExecContext(ctx, sqlUpdateTaskStatus, string(status), timeToText(time.Now().UTC()), id) if err != nil { return mapErr(err) } - return checkRowsAffected(res) + if err := checkRowsAffected(res); err != nil { + return err + } + if err = tx.Commit(); err != nil { + return fmt.Errorf("sqlite: commit update task status: %w", err) + } + return nil } // SetTaskUnschedulableReason implements [store.TaskStore]. diff --git a/internal/store/sqlite/task_test.go b/internal/store/sqlite/task_test.go index 8fca98f6..5835e94a 100644 --- a/internal/store/sqlite/task_test.go +++ b/internal/store/sqlite/task_test.go @@ -27,17 +27,11 @@ func TestRetryTasks_SQLite(t *testing.T) { t.Fatalf("UpdateStepStatus failed: %v", err) } insertTask(t, s, "t-failed", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "t-failed", store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus t-failed: %v", err) - } + walkTaskTo(t, s, "t-failed", store.TaskStatusFailed) insertTask(t, s, "t-canceled", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "t-canceled", store.TaskStatusCanceled); err != nil { - t.Fatalf("UpdateTaskStatus t-canceled: %v", err) - } + walkTaskTo(t, s, "t-canceled", store.TaskStatusCanceled) insertTask(t, s, "t-ok", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "t-ok", store.TaskStatusSucceeded); err != nil { - t.Fatalf("UpdateTaskStatus t-ok: %v", err) - } + walkTaskTo(t, s, "t-ok", store.TaskStatusSucceeded) revived, err := s.RetryTasks(ctx, "j1", nil, time.Now().UTC()) if err != nil { @@ -110,9 +104,7 @@ func TestRetryTasks_EmptySliceRevivesNothing(t *testing.T) { t.Fatalf("UpdateStepStatus: %v", err) } insertTask(t, s, "t-failed", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "t-failed", store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus t-failed: %v", err) - } + walkTaskTo(t, s, "t-failed", store.TaskStatusFailed) // Non-nil but empty slice: "filter to exactly these (zero) IDs" → revive nothing. revived, err := s.RetryTasks(ctx, "j1", []string{}, time.Now().UTC()) @@ -152,13 +144,9 @@ func TestRetryTasks_MixedStateStep(t *testing.T) { t.Fatalf("UpdateStepStatus: %v", err) } insertTask(t, s, "ta", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "ta", store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus ta: %v", err) - } + walkTaskTo(t, s, "ta", store.TaskStatusFailed) insertTask(t, s, "tb", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "tb", store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus tb: %v", err) - } + walkTaskTo(t, s, "tb", store.TaskStatusFailed) // Retry only "ta" from the subset. revived, err := s.RetryTasks(ctx, "j1", []string{"ta"}, time.Now().UTC()) @@ -217,9 +205,7 @@ func TestRetryTasks_ResetsFailureCounters(t *testing.T) { } insertStep(t, s, "s1", "j1", "S1", 0) insertTask(t, s, "t1", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "t1", store.TaskStatusRunning); err != nil { - t.Fatalf("UpdateTaskStatus: %v", err) - } + walkTaskTo(t, s, "t1", store.TaskStatusRunning) // Drive genuine-failure bookkeeping: a failed attempt bumps both counters // and stamps a backoff; enough failures park the job with a reason. @@ -237,9 +223,7 @@ func TestRetryTasks_ResetsFailureCounters(t *testing.T) { // Drive the task and job to the terminal states RetryTasks operates on // (park leaves the job paused, not terminal — so move both to failed // directly, as the production failure sweep would eventually do). - if err := s.UpdateTaskStatus(ctx, "t1", store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus(failed): %v", err) - } + walkTaskTo(t, s, "t1", store.TaskStatusFailed) if err := s.UpdateJobStatus(ctx, "j1", store.JobStatusFailed); err != nil { t.Fatalf("UpdateJobStatus(failed): %v", err) } @@ -291,9 +275,7 @@ func TestRetryTasks_ClearsFailureReason(t *testing.T) { } insertStep(t, s, "s1", "j1", "S1", 0) insertTask(t, s, "t1", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "t1", store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus: %v", err) - } + walkTaskTo(t, s, "t1", store.TaskStatusFailed) if err := s.SetTaskFailureReason(ctx, "t1", "boom"); err != nil { t.Fatalf("SetTaskFailureReason: %v", err) } @@ -538,9 +520,7 @@ func TestRequeueTaskForRetry_GuardedToInFlight(t *testing.T) { t.Run(string(tc.status), func(t *testing.T) { id := "t" + string(rune('1'+i)) insertTask(t, s, id, "j1", "s1") - if err := s.UpdateTaskStatus(ctx, id, tc.status); err != nil { - t.Fatalf("UpdateTaskStatus: %v", err) - } + walkTaskTo(t, s, id, tc.status) if tc.reason != "" { if err := s.SetTaskFailureReason(ctx, id, tc.reason); err != nil { t.Fatalf("SetTaskFailureReason: %v", err) @@ -730,9 +710,7 @@ func TestRetryTasks_UnparksAutoParkedJob(t *testing.T) { t.Fatalf("RecordTaskFailure: %v", err) } // The tripping task went terminal-failed and the job parked. - if err := s.UpdateTaskStatus(ctx, "t1", store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus(failed): %v", err) - } + walkTaskTo(t, s, "t1", store.TaskStatusFailed) if err := s.ParkJob(ctx, "j1", "failure limit reached (1)", now); err != nil { t.Fatalf("ParkJob: %v", err) } @@ -760,9 +738,7 @@ func TestRetryTasks_UnparksAutoParkedJob(t *testing.T) { func TestRetryTasks_LeavesManualPauseAlone(t *testing.T) { s, ctx, now := recordFailureFixture(t) - if err := s.UpdateTaskStatus(ctx, "t1", store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus(failed): %v", err) - } + walkTaskTo(t, s, "t1", store.TaskStatusFailed) if err := s.UpdateJobStatus(ctx, "j1", store.JobStatusPaused); err != nil { t.Fatalf("UpdateJobStatus(paused): %v", err) } @@ -918,9 +894,7 @@ func failedTaskWithReason(t *testing.T, s *sqlite.Store, id, jobID, stepID, reas t.Helper() insertTask(t, s, id, jobID, stepID) ctx := context.Background() - if err := s.UpdateTaskStatus(ctx, id, store.TaskStatusFailed); err != nil { - t.Fatalf("UpdateTaskStatus(%q): %v", id, err) - } + walkTaskTo(t, s, id, store.TaskStatusFailed) if err := s.SetTaskFailureReason(ctx, id, reason); err != nil { t.Fatalf("SetTaskFailureReason(%q): %v", id, err) } @@ -985,9 +959,7 @@ func TestFailureReasonSummary_Empty(t *testing.T) { insertJob(t, s, "j1", "f1", "q1") insertStep(t, s, "s1", "j1", "S1", 0) insertTask(t, s, "t0", "j1", "s1") - if err := s.UpdateTaskStatus(ctx, "t0", store.TaskStatusSucceeded); err != nil { - t.Fatalf("UpdateTaskStatus: %v", err) - } + walkTaskTo(t, s, "t0", store.TaskStatusSucceeded) sum, err := s.FailureReasonSummary(ctx, "j1") if err != nil { diff --git a/internal/store/sqlite/task_transition_test.go b/internal/store/sqlite/task_transition_test.go new file mode 100644 index 00000000..8b95f901 --- /dev/null +++ b/internal/store/sqlite/task_transition_test.go @@ -0,0 +1,166 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package sqlite_test + +// Tests that UpdateTaskStatus enforces the task state machine +// ([store.ValidateTaskTransition]) rather than writing whatever status it is +// handed. The check and the write share one transaction, so a concurrent +// writer cannot slip a task between the read and the update. + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/uberware/sqi/internal/store" + "github.com/uberware/sqi/internal/store/sqlite" +) + +// transitionFixture returns a store with one task, walked to from. +func transitionFixture(t *testing.T, from store.TaskStatus) (*sqlite.Store, string) { + t.Helper() + s := openTestStore(t) + insertFarm(t, s, "f1", "F1") + insertQueue(t, s, "q1", "f1", "Q1") + insertJob(t, s, "j1", "f1", "q1") + insertStep(t, s, "s1", "j1", "S1", 0) + insertTask(t, s, "t1", "j1", "s1") + walkTaskTo(t, s, "t1", from) + return s, "t1" +} + +func TestUpdateTaskStatus_RejectsIllegalTransition(t *testing.T) { + tests := []struct { + name string + from store.TaskStatus + to store.TaskStatus + }{ + {"succeeded is terminal", store.TaskStatusSucceeded, store.TaskStatusReady}, + {"failed cannot succeed", store.TaskStatusFailed, store.TaskStatusSucceeded}, + {"canceled is terminal", store.TaskStatusCanceled, store.TaskStatusRunning}, + {"ready cannot start running", store.TaskStatusReady, store.TaskStatusRunning}, + {"pending cannot be assigned", store.TaskStatusPending, store.TaskStatusAssigned}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + s, id := transitionFixture(t, tc.from) + ctx := context.Background() + + err := s.UpdateTaskStatus(ctx, id, tc.to) + if !errors.Is(err, store.ErrInvalidTransition) { + t.Fatalf("UpdateTaskStatus(%q → %q) = %v, want ErrInvalidTransition", tc.from, tc.to, err) + } + + // The row must be untouched. + got, getErr := s.GetTask(ctx, id) + if getErr != nil { + t.Fatalf("GetTask: %v", getErr) + } + if got.Status != tc.from { + t.Errorf("status = %q after rejected transition, want %q (unchanged)", got.Status, tc.from) + } + }) + } +} + +func TestUpdateTaskStatus_AllowsLegalTransition(t *testing.T) { + s, id := transitionFixture(t, store.TaskStatusRunning) + ctx := context.Background() + + if err := s.UpdateTaskStatus(ctx, id, store.TaskStatusSucceeded); err != nil { + t.Fatalf("UpdateTaskStatus(running → succeeded): %v", err) + } + got, err := s.GetTask(ctx, id) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if got.Status != store.TaskStatusSucceeded { + t.Errorf("status = %q, want succeeded", got.Status) + } +} + +// TestUpdateTaskStatus_AssignedToTerminal covers the arrow that exists because +// a worker's "running" publish can be dropped after MaxRetries while the task +// still completes. +func TestUpdateTaskStatus_AssignedToTerminal(t *testing.T) { + s, id := transitionFixture(t, store.TaskStatusAssigned) + ctx := context.Background() + + if err := s.UpdateTaskStatus(ctx, id, store.TaskStatusSucceeded); err != nil { + t.Fatalf("UpdateTaskStatus(assigned → succeeded): %v", err) + } + got, err := s.GetTask(ctx, id) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if got.Status != store.TaskStatusSucceeded { + t.Errorf("status = %q, want succeeded", got.Status) + } +} + +// TestUpdateTaskStatus_SameStatusIsNoOp pins idempotency. Task status arrives +// over JetStream, which is at-least-once: a redelivered "running" message must +// not turn into an error, or the consumer would Nak it and redeliver forever. +func TestUpdateTaskStatus_SameStatusIsNoOp(t *testing.T) { + s, id := transitionFixture(t, store.TaskStatusRunning) + ctx := context.Background() + + if err := s.UpdateTaskStatus(ctx, id, store.TaskStatusRunning); err != nil { + t.Fatalf("UpdateTaskStatus(running → running) = %v, want nil (no-op)", err) + } + got, err := s.GetTask(ctx, id) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if got.Status != store.TaskStatusRunning { + t.Errorf("status = %q, want running", got.Status) + } +} + +func TestUpdateTaskStatus_NotFound(t *testing.T) { + s, _ := transitionFixture(t, store.TaskStatusReady) + + err := s.UpdateTaskStatus(context.Background(), "no-such-task", store.TaskStatusAssigned) + if !errors.Is(err, store.ErrNotFound) { + t.Errorf("UpdateTaskStatus(missing) = %v, want ErrNotFound", err) + } +} + +// TestUpdateTaskStatus_ConcurrentRacesToOneWinner is the reason the check and +// the write share a transaction. Two goroutines race assigned → running and +// assigned → canceled; exactly one must win and the other must be rejected, +// never both applied. +func TestUpdateTaskStatus_ConcurrentRacesToOneWinner(t *testing.T) { + s, id := transitionFixture(t, store.TaskStatusRunning) + ctx := context.Background() + + var wg sync.WaitGroup + errs := make([]error, 2) + targets := []store.TaskStatus{store.TaskStatusSucceeded, store.TaskStatusFailed} + + wg.Add(2) + for i, target := range targets { + go func() { + defer wg.Done() + errs[i] = s.UpdateTaskStatus(ctx, id, target) + }() + } + wg.Wait() + + okCount := 0 + for _, err := range errs { + switch { + case err == nil: + okCount++ + case errors.Is(err, store.ErrInvalidTransition): + // expected loser + default: + t.Fatalf("unexpected error: %v", err) + } + } + if okCount != 1 { + t.Errorf("%d of 2 concurrent terminal transitions succeeded, want exactly 1", okCount) + } +} diff --git a/internal/store/statemachine.go b/internal/store/statemachine.go new file mode 100644 index 00000000..ee47d1a9 --- /dev/null +++ b/internal/store/statemachine.go @@ -0,0 +1,94 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package store + +import ( + "errors" + "fmt" +) + +// ErrInvalidTransition is returned when a requested status transition is not +// permitted by the task state machine. +// +// Use errors.Is to test: +// +// err := ValidateTaskTransition(from, to) +// if errors.Is(err, ErrInvalidTransition) { ... } +var ErrInvalidTransition = errors.New("store: invalid state transition") + +// ── Task state machine ──────────────────────────────────────────────────────── +// +// Permitted task status transitions: +// +// pending → ready dependency resolution: all dependency steps completed +// pending → canceled job canceled before step dependencies were satisfied +// ready → assigned scheduler assigns task to a worker +// ready → canceled task canceled while waiting for a worker +// assigned → running worker confirms execution has started +// assigned → ready reassignment: assigned worker disconnected or timed out +// assigned → canceled task canceled after assignment but before confirmation +// assigned → succeeded worker's "running" message was dropped (see below) +// assigned → failed worker's "running" message was dropped (see below) +// running → succeeded worker reports clean exit (exit code 0) +// running → failed worker reports non-zero exit or fatal error +// running → ready reassignment: running worker became unreachable +// running → canceled task canceled while executing +// +// Terminal states (succeeded, failed, canceled) have no outgoing transitions. +// +// assigned → succeeded/failed look like skipped states but are reachable in +// normal operation: a worker publishes "running" before any terminal status, +// but status.Publisher.publishWithRetry gives up after MaxRetries and returns, +// so that message can be lost for good while the task still runs to completion. +// The terminal message then lands on a row still in 'assigned'. Rejecting it +// would strand finished work until the heartbeat sweep reclaimed it. +// +// A transition from a status to itself is not listed here and is not valid: +// callers that must tolerate duplicate delivery treat same-status writes as a +// no-op before consulting this table (see [TaskStore.UpdateTaskStatus]). +var validTaskTransitions = map[TaskStatus]map[TaskStatus]struct{}{ + TaskStatusPending: { + TaskStatusReady: {}, + TaskStatusCanceled: {}, + }, + TaskStatusReady: { + TaskStatusAssigned: {}, + TaskStatusCanceled: {}, + }, + TaskStatusAssigned: { + TaskStatusRunning: {}, + TaskStatusReady: {}, + TaskStatusCanceled: {}, + TaskStatusSucceeded: {}, + TaskStatusFailed: {}, + }, + TaskStatusRunning: { + TaskStatusSucceeded: {}, + TaskStatusFailed: {}, + TaskStatusReady: {}, + TaskStatusCanceled: {}, + }, + // Terminal states — no outgoing transitions. + TaskStatusSucceeded: {}, + TaskStatusFailed: {}, + TaskStatusCanceled: {}, +} + +// ValidateTaskTransition returns nil if transitioning a task from one status to +// another is permitted by the state machine, or a descriptive error wrapping +// [ErrInvalidTransition] otherwise. +// +// This lives in package store rather than package openjd because the store is +// what enforces it on every write; openjd imports store, so the store cannot +// import openjd. [github.com/uberware/sqi/internal/openjd.ValidateTaskTransition] +// delegates here. +func ValidateTaskTransition(from, to TaskStatus) error { + targets, known := validTaskTransitions[from] + if !known { + return fmt.Errorf("%w: unknown task status %q", ErrInvalidTransition, from) + } + if _, ok := targets[to]; ok { + return nil + } + return fmt.Errorf("%w: task %q → %q not permitted", ErrInvalidTransition, from, to) +} diff --git a/internal/store/statemachine_test.go b/internal/store/statemachine_test.go new file mode 100644 index 00000000..8409fd0d --- /dev/null +++ b/internal/store/statemachine_test.go @@ -0,0 +1,106 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package store_test + +// Tests for statemachine.go — the task state machine that [store.Store] +// implementations enforce on every status write. +// +// The task machine moved here from internal/openjd so the store can enforce it +// without an import cycle (openjd imports store). openjd.ValidateTaskTransition +// now delegates here and keeps its own tests. + +import ( + "errors" + "testing" + + "github.com/uberware/sqi/internal/store" +) + +func TestValidateTaskTransition_Legal(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + from store.TaskStatus + to store.TaskStatus + }{ + {"pending to ready", store.TaskStatusPending, store.TaskStatusReady}, + {"pending to canceled", store.TaskStatusPending, store.TaskStatusCanceled}, + {"ready to assigned", store.TaskStatusReady, store.TaskStatusAssigned}, + {"ready to canceled", store.TaskStatusReady, store.TaskStatusCanceled}, + {"assigned to running", store.TaskStatusAssigned, store.TaskStatusRunning}, + {"assigned to ready", store.TaskStatusAssigned, store.TaskStatusReady}, + {"assigned to canceled", store.TaskStatusAssigned, store.TaskStatusCanceled}, + {"running to succeeded", store.TaskStatusRunning, store.TaskStatusSucceeded}, + {"running to failed", store.TaskStatusRunning, store.TaskStatusFailed}, + {"running to ready", store.TaskStatusRunning, store.TaskStatusReady}, + {"running to canceled", store.TaskStatusRunning, store.TaskStatusCanceled}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if err := store.ValidateTaskTransition(tc.from, tc.to); err != nil { + t.Errorf("ValidateTaskTransition(%q, %q) = %v, want nil", tc.from, tc.to, err) + } + }) + } +} + +// TestValidateTaskTransition_AssignedToTerminal pins the arrows added when +// store-level enforcement was introduced. A worker publishes "running" before +// it publishes a terminal status, but status.Publisher.publishWithRetry gives +// up after MaxRetries and returns — so the "running" message can be dropped +// permanently while the task still completes. The terminal message then arrives +// with the task row still in 'assigned'. Rejecting that would strand finished +// work in 'assigned' until the heartbeat sweep reclaimed it. +func TestValidateTaskTransition_AssignedToTerminal(t *testing.T) { + t.Parallel() + + for _, to := range []store.TaskStatus{store.TaskStatusSucceeded, store.TaskStatusFailed} { + t.Run(string(to), func(t *testing.T) { + t.Parallel() + if err := store.ValidateTaskTransition(store.TaskStatusAssigned, to); err != nil { + t.Errorf("ValidateTaskTransition(assigned, %q) = %v, want nil", to, err) + } + }) + } +} + +func TestValidateTaskTransition_Illegal(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + from store.TaskStatus + to store.TaskStatus + }{ + {"pending skips ready", store.TaskStatusPending, store.TaskStatusAssigned}, + {"pending to running", store.TaskStatusPending, store.TaskStatusRunning}, + {"ready to running", store.TaskStatusReady, store.TaskStatusRunning}, + {"ready to succeeded", store.TaskStatusReady, store.TaskStatusSucceeded}, + {"succeeded is terminal", store.TaskStatusSucceeded, store.TaskStatusReady}, + {"failed is terminal", store.TaskStatusFailed, store.TaskStatusRunning}, + {"canceled is terminal", store.TaskStatusCanceled, store.TaskStatusReady}, + {"failed cannot succeed", store.TaskStatusFailed, store.TaskStatusSucceeded}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + err := store.ValidateTaskTransition(tc.from, tc.to) + if !errors.Is(err, store.ErrInvalidTransition) { + t.Errorf("ValidateTaskTransition(%q, %q) = %v, want ErrInvalidTransition", tc.from, tc.to, err) + } + }) + } +} + +func TestValidateTaskTransition_UnknownStatus(t *testing.T) { + t.Parallel() + + err := store.ValidateTaskTransition("bogus", store.TaskStatusReady) + if !errors.Is(err, store.ErrInvalidTransition) { + t.Errorf("ValidateTaskTransition(bogus, ready) = %v, want ErrInvalidTransition", err) + } +}