fix: harden task runtime across restart and terminal mirror

Clear open run segments on process start without imputing downtime.
Mirror terminal/pause runtime fields into memory before hydration, make
SetPaused finalize segments, and add DB lifecycle tests.
This commit is contained in:
artex committed 2026-10-27 14:14:56 +08:00
1 parent 41c967f7ad
commit d8bda59d7d
4 files changed
+274 -24

No files matched your search

+172
View File
@@ -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)
}
}
+67 -10
View File
@@ -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
+34 -13
View File
@@ -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
}
+1 -1
View File
@@ -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));