diff --git a/db/task_runtime_test.go b/db/task_runtime_test.go new file mode 100644 index 0000000..492343f --- /dev/null +++ b/db/task_runtime_test.go @@ -0,0 +1,172 @@ +package db + +import ( + "database/sql" + "testing" +) + +func TestTaskRunSegmentLifecycle(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer d.Close() + + task, err := d.CreateTask("runtime lifecycle", "runtime lifecycle test", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = d.DeleteTask(task.ID) }) + + active, started, err := d.StartTaskRunSegment(task.ID) + if err != nil { + t.Fatal(err) + } + if started == nil { + t.Fatal("StartTaskRunSegment did not open a segment") + } + if active != 0 { + t.Fatalf("new task active seconds=%d, want 0", active) + } + if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil { + t.Fatal(err) + } + + active, started, err = d.FinalizeTaskRunSegment(task.ID) + if err != nil { + t.Fatal(err) + } + if active <= 0 || started != nil { + t.Fatalf("first finalize active=%d started=%v, want positive and NULL", active, started) + } + firstActive := active + + // Pause folds the open segment exactly once and closes it. Repeated + // finalization after wall-clock advancement remains a no-op. + if _, _, err := d.StartTaskRunSegment(task.ID); err != nil { + t.Fatal(err) + } + if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil { + t.Fatal(err) + } + active, started, err = d.SetPausedRuntime(task.ID, true) + if err != nil { + t.Fatal(err) + } + if active <= firstActive || started != nil { + t.Fatalf("pause active=%d started=%v, want increase and NULL", active, started) + } + pausedActive := active + if active, started, err = d.FinalizeTaskRunSegment(task.ID); err != nil { + t.Fatal(err) + } else if active != pausedActive || started != nil { + t.Fatalf("repeated finalize after pause active=%d started=%v, want %d and NULL", active, started, pausedActive) + } + + // Resuming opens a fresh segment, whose elapsed time is added on top of the + // paused total rather than replacing it. + if _, _, err := d.SetPausedRuntime(task.ID, false); err != nil { + t.Fatal(err) + } + if _, started, err = d.StartTaskRunSegment(task.ID); err != nil { + t.Fatal(err) + } else if started == nil { + t.Fatal("resume did not open a segment") + } + if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil { + t.Fatal(err) + } + active, started, err = d.FinalizeTaskRunSegment(task.ID) + if err != nil { + t.Fatal(err) + } + if active <= pausedActive || started != nil { + t.Fatalf("resume finalize active=%d started=%v, want increase and NULL", active, started) + } + + // Both terminal status paths fold an open segment and leave no live clock. + if _, started, err = d.StartTaskRunSegment(task.ID); err != nil { + t.Fatal(err) + } else if started == nil { + t.Fatal("terminal status setup did not open a segment") + } + if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil { + t.Fatal(err) + } + beforeTerminal := active + active, started, err = d.SetStatusRuntime(task.ID, "done") + if err != nil { + t.Fatal(err) + } + if active <= beforeTerminal || started != nil { + t.Fatalf("SetStatusRuntime terminal active=%d started=%v, want increase and NULL", active, started) + } + + if _, _, err = d.SetStatusRuntime(task.ID, "running"); err != nil { + t.Fatal(err) + } + if _, started, err = d.StartTaskRunSegment(task.ID); err != nil { + t.Fatal(err) + } else if started == nil { + t.Fatal("guarded terminal setup did not open a segment") + } + if _, err := d.Exec(`UPDATE tasks SET run_segment_started_at=now()-interval '3 seconds' WHERE id=$1`, task.ID); err != nil { + t.Fatal(err) + } + beforeGuarded := active + active, started, won, err := d.SetTerminalStatusGuardedRuntime(task.ID, "failed") + if err != nil || !won { + t.Fatalf("SetTerminalStatusGuardedRuntime: active=%d started=%v won=%v err=%v", active, started, won, err) + } + if active <= beforeGuarded || started != nil { + t.Fatalf("guarded terminal active=%d started=%v, want increase and NULL", active, started) + } + finalActive := active + active, started, won, err = d.SetTerminalStatusGuardedRuntime(task.ID, "timeout") + if err != nil || won { + t.Fatalf("second guarded terminal: active=%d started=%v won=%v err=%v", active, started, won, err) + } + if active != finalActive || started != nil { + t.Fatalf("second guarded terminal returned active=%d started=%v, want unchanged active=%d and NULL segment", active, started, finalActive) + } + got, err := d.GetTask(task.ID) + if err != nil { + t.Fatal(err) + } + if got == nil || got.ActiveRunSeconds != finalActive || got.RunSegmentStartedAt != nil { + t.Fatalf("persisted terminal runtime=%+v, want active=%d and NULL segment", got, finalActive) + } +} + +func TestClearOpenTaskRunSegmentsPreservesAccumulatedRuntime(t *testing.T) { + d, err := Open(testDSN(t)) + if err != nil { + t.Skipf("postgres unavailable (%v) — skipping", err) + } + defer d.Close() + + task, err := d.CreateTask("runtime restart clear", "runtime restart clear test", nil, 0, 0) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = d.DeleteTask(task.ID) }) + if _, err := d.Exec(`UPDATE tasks SET active_run_seconds=17, run_segment_started_at=now()-interval '1 hour' WHERE id=$1`, task.ID); err != nil { + t.Fatal(err) + } + + cleared, err := d.ClearOpenTaskRunSegments() + if err != nil { + t.Fatal(err) + } + if cleared < 1 { + t.Fatalf("ClearOpenTaskRunSegments cleared %d rows, want at least 1", cleared) + } + var active int64 + var started sql.NullTime + if err := d.QueryRow(`SELECT active_run_seconds, run_segment_started_at FROM tasks WHERE id=$1`, task.ID).Scan(&active, &started); err != nil { + t.Fatal(err) + } + if active != 17 || started.Valid { + t.Fatalf("restart clear changed runtime: active=%d started=%v, want 17 and NULL", active, started) + } +} diff --git a/db/tasks.go b/db/tasks.go index 330a4cd..ac5c19e 100644 --- a/db/tasks.go +++ b/db/tasks.go @@ -495,10 +495,29 @@ func (d *DB) GetTask(id int64) (*Task, error) { // SetPaused persists a task's paused flag. func (d *DB) SetPaused(id int64, paused bool) error { - _, err := d.Exec(`UPDATE tasks SET paused=$1 WHERE id=$2`, paused, id) + _, _, err := d.SetPausedRuntime(id, paused) return err } +// SetPausedRuntime persists a task's paused flag and, when pausing, folds any +// open execution segment into active_run_seconds. Keeping this in the DB helper +// makes direct callers safe even when they do not go through Manager pause +// orchestration. +func (d *DB) SetPausedRuntime(id int64, paused bool) (int64, *time.Time, error) { + var active int64 + var started *time.Time + err := d.QueryRow(`UPDATE tasks +SET paused=$1, + active_run_seconds=active_run_seconds + CASE WHEN $1 AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END, + run_segment_started_at=CASE WHEN $1 THEN NULL ELSE run_segment_started_at END +WHERE id=$2 +RETURNING active_run_seconds, run_segment_started_at`, paused, id).Scan(&active, &started) + if err == sql.ErrNoRows { + return 0, nil, nil + } + return active, started, err +} + // StartTaskRunSegment begins the currently active execution segment. It is // idempotent and refuses to start while a task is paused or waiting in the // concurrency queue. @@ -581,14 +600,27 @@ func (d *DB) SetQueued(id int64, queued bool) error { // stable); moving back to a non-terminal state clears it, so a re-run has no stale // finish time. func (d *DB) SetStatus(id int64, status string) error { - _, err := d.Exec(` + _, _, err := d.SetStatusRuntime(id, status) + return err +} + +// SetStatusRuntime updates a task status and returns the committed runtime +// fields from the same UPDATE. Terminal transitions finalize any open segment. +func (d *DB) SetStatusRuntime(id int64, status string) (int64, *time.Time, error) { + var active int64 + var started *time.Time + err := d.QueryRow(` UPDATE tasks SET status = $1, completed_at = CASE WHEN $1 IN ('done','failed','timeout') THEN COALESCE(completed_at, now()) ELSE NULL END, active_run_seconds = active_run_seconds + CASE WHEN $1 IN ('done','failed','timeout') AND run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END, run_segment_started_at = CASE WHEN $1 IN ('done','failed','timeout') THEN NULL ELSE run_segment_started_at END - WHERE id = $2`, status, id) - return err + WHERE id = $2 + RETURNING active_run_seconds, run_segment_started_at`, status, id).Scan(&active, &started) + if err == sql.ErrNoRows { + return 0, nil, nil + } + return active, started, err } // SetTerminalStatusGuarded sets a terminal status only when the task is NOT already @@ -596,18 +628,43 @@ UPDATE tasks // Returns won=false (no error) when another terminal status already stuck — the // caller then leaves it alone. Non-terminal transitions / re-run still use SetStatus. func (d *DB) SetTerminalStatusGuarded(id int64, status string) (won bool, err error) { - res, err := d.Exec(` + _, _, won, err = d.SetTerminalStatusGuardedRuntime(id, status) + return won, err +} + +// SetTerminalStatusGuardedRuntime is the runtime-returning form used by the +// Manager so it can mirror the terminal clock before any follow-up hydration. +func (d *DB) SetTerminalStatusGuardedRuntime(id int64, status string) (active int64, started *time.Time, won bool, err error) { + err = d.QueryRow(` UPDATE tasks SET status = $1, completed_at = COALESCE(completed_at, now()), active_run_seconds = active_run_seconds + CASE WHEN run_segment_started_at IS NOT NULL THEN GREATEST(0, FLOOR(EXTRACT(EPOCH FROM (now() - run_segment_started_at)))::bigint) ELSE 0 END, run_segment_started_at = NULL - WHERE id = $2 AND status NOT IN ('done','failed','timeout')`, status, id) - if err != nil { - return false, err + WHERE id = $2 AND status NOT IN ('done','failed','timeout') + RETURNING active_run_seconds, run_segment_started_at`, status, id).Scan(&active, &started) + if err == sql.ErrNoRows { + // Preserve the authoritative runtime fields for the losing terminal + // writer. The guarded transition itself is still a no-op. + err = d.QueryRow(`SELECT active_run_seconds, run_segment_started_at FROM tasks WHERE id=$1`, id).Scan(&active, &started) + if err == sql.ErrNoRows { + return 0, nil, false, nil + } + return active, started, false, err } - n, _ := res.RowsAffected() - return n > 0, nil + return active, started, err == nil, err +} + +// ClearOpenTaskRunSegments drops execution segments that were live when the +// process stopped. Their wall time is unknowable after a crash or deploy, so +// startup deliberately begins with no live segment; the engine starts a fresh +// segment only when it actually resumes work. +func (d *DB) ClearOpenTaskRunSegments() (int64, error) { + res, err := d.Exec(`UPDATE tasks SET run_segment_started_at=NULL WHERE run_segment_started_at IS NOT NULL`) + if err != nil { + return 0, err + } + return res.RowsAffected() } // StampFirstRun records a task's first-real-run moment and computes its absolute diff --git a/server/manager.go b/server/manager.go index 9907f49..1aa7b70 100644 --- a/server/manager.go +++ b/server/manager.go @@ -532,6 +532,13 @@ func NewManager(dir, proxyAddr string) (*Manager, error) { if err != nil { return nil, err } + // A live segment cannot be trusted across a process restart: its end time + // could be deployment/crash downtime. Clear it before tasks are restored so + // the next engine admission starts a fresh segment at actual execution time. + if _, err := pg.ClearOpenTaskRunSegments(); err != nil { + _ = pg.Close() + return nil, fmt.Errorf("clear stale task run segments: %w", err) + } if err := pg.RecoverFindingRetests(); err != nil { pg.Close() return nil, fmt.Errorf("recover finding retests: %w", err) @@ -1297,13 +1304,16 @@ func (m *Manager) SetTaskPaused(id string, paused bool) error { if err != nil { return err } - if err := m.pg.SetPaused(n, paused); err != nil { + active, started, err := m.pg.SetPausedRuntime(n, paused) + if err != nil { return err } m.mu.Lock() if t := m.tasks[id]; t != nil { t.updateLifecycle(func(state *taskLifecycleState) { state.Paused = paused + state.ActiveRunSeconds = active + state.RunSegmentStartedAt = unixOrZero(started) }) } m.mu.Unlock() @@ -1662,23 +1672,30 @@ func (m *Manager) SetTaskStatusGuarded(id, status string) (won bool, err error) if err != nil { return false, err } - won, err = m.pg.SetTerminalStatusGuarded(n, status) + active, started, won, err := m.pg.SetTerminalStatusGuardedRuntime(n, status) if err != nil || !won { return won, err } - if err := m.syncTaskRuntime(id, n); err != nil { - return false, err - } + // Mirror the committed terminal transition before any optional hydration. + // This keeps the in-memory clock stopped even if GetTask later fails. m.mu.Lock() if t := m.tasks[id]; t != nil { t.updateLifecycle(func(state *taskLifecycleState) { state.Status = status - if state.CompletedAt == 0 { - state.CompletedAt = time.Now().Unix() + state.ActiveRunSeconds = active + state.RunSegmentStartedAt = unixOrZero(started) + if pgdb.IsTerminal(status) { + state.RunSegmentStartedAt = 0 + if state.CompletedAt == 0 { + state.CompletedAt = time.Now().Unix() + } } }) } m.mu.Unlock() + if err := m.syncTaskRuntime(id, n); err != nil { + return true, err + } return true, nil } @@ -1691,19 +1708,20 @@ func (m *Manager) SetTaskStatus(id, status string) error { if err != nil { return err } - if err := m.pg.SetStatus(n, status); err != nil { - return err - } - if err := m.syncTaskRuntime(id, n); err != nil { + active, started, err := m.pg.SetStatusRuntime(n, status) + if err != nil { return err } + // Mirror the successful UPDATE immediately. A later GetTask hydration is + // best-effort and must not reopen a terminal clock in memory. m.mu.Lock() if t := m.tasks[id]; t != nil { t.updateLifecycle(func(state *taskLifecycleState) { state.Status = status - // Mirror the DB's completed_at stamp on the live handle so the DTO shows - // the finish time without a reload (terminal -> stamp once; else clear). + state.ActiveRunSeconds = active + state.RunSegmentStartedAt = unixOrZero(started) if pgdb.IsTerminal(status) { + state.RunSegmentStartedAt = 0 if state.CompletedAt == 0 { state.CompletedAt = time.Now().Unix() } @@ -1713,6 +1731,9 @@ func (m *Manager) SetTaskStatus(id, status string) error { }) } m.mu.Unlock() + if err := m.syncTaskRuntime(id, n); err != nil { + return err + } return nil } diff --git a/web/src/app/(main)/function/tasks/page.tsx b/web/src/app/(main)/function/tasks/page.tsx index fedf22b..df88c10 100644 --- a/web/src/app/(main)/function/tasks/page.tsx +++ b/web/src/app/(main)/function/tasks/page.tsx @@ -191,7 +191,7 @@ function fmtDuration(sec: number): string { // taskDuration counts only active execution segments. A live segment ticks locally; // paused, queued, and terminal tasks stay fixed at their persisted active total. function taskDuration(task: Task, nowSec: number): number { - const active = Math.max(0, Math.floor(task.active_run_seconds ?? task.run_seconds ?? 0)); + const active = Math.max(0, Math.floor(task.active_run_seconds ?? 0)); const segmentStart = task.run_segment_started_unix ?? 0; if (task.status === "running" && !task.paused && !task.queued && segmentStart > 0) { return active + Math.max(0, Math.floor(nowSec - segmentStart));