fix: stop archived automation-run tasks from pinning max_concurrent_runs (#1632)
This commit is contained in:
@@ -426,6 +426,7 @@ test-scripts:
|
||||
@python3 scripts/opencode-code-review.test.py
|
||||
@python3 scripts/lint-harness-files.test.py
|
||||
@bash scripts/release-desktop.test.sh
|
||||
@node --test apps/desktop/e2e/desktop-launch-smoke.test.mjs
|
||||
|
||||
.PHONY: test-e2e
|
||||
test-e2e: build-backend build-web
|
||||
|
||||
@@ -45,6 +45,14 @@ const (
|
||||
RunStatusSucceeded RunStatus = "succeeded"
|
||||
RunStatusFailed RunStatus = "failed"
|
||||
RunStatusSkipped RunStatus = "skipped"
|
||||
// RunStatusCancelled is a read-time-derived terminal status: it is never
|
||||
// written by a trigger-firing code path, only computed by ListRuns (and
|
||||
// implied by CountActiveRuns' filter) when a task_created run's
|
||||
// generated task has been archived or no longer exists. The task's
|
||||
// outcome is unknown at that point — the user closed it out some other
|
||||
// way — so this is deliberately distinct from succeeded/failed rather
|
||||
// than guessing one.
|
||||
RunStatusCancelled RunStatus = "cancelled"
|
||||
)
|
||||
|
||||
// ExecutionMode controls whether an automation firing creates a visible
|
||||
|
||||
@@ -197,3 +197,60 @@ func TestFireTrigger_ConcurrencyCapCheckError_DoesNotAdvanceLastEvaluatedAt(t *t
|
||||
t.Fatal("expected LastEvaluatedAt to remain nil after a concurrency-cap check error, got non-nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestFireTrigger_ArchivedTaskRun_DoesNotBlockConcurrencyCap reproduces the
|
||||
// reported bug end to end: an automation-generated task that gets archived
|
||||
// (manually, via auto-archive, or via cascade) before its run is otherwise
|
||||
// finalized left the run stuck at task_created forever, permanently pinning
|
||||
// max_concurrent_runs=1 automations at their cap with no way to schedule
|
||||
// again. A run whose task is archived no longer represents outstanding
|
||||
// work and must not count toward the cap.
|
||||
func TestFireTrigger_ArchivedTaskRun_DoesNotBlockConcurrencyCap(t *testing.T) {
|
||||
svc := newTestService(t)
|
||||
createTasksTable(t, svc.store)
|
||||
ctx := context.Background()
|
||||
|
||||
a := &Automation{
|
||||
WorkspaceID: "ws-1",
|
||||
Name: "Daily rebase",
|
||||
WorkflowID: "wf-1",
|
||||
WorkflowStepID: "s-1",
|
||||
Enabled: true,
|
||||
MaxConcurrentRuns: 1,
|
||||
}
|
||||
if err := svc.store.CreateAutomation(ctx, a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg, _ := json.Marshal(ScheduledTriggerConfig{CronExpression: "@daily"})
|
||||
trig := &AutomationTrigger{AutomationID: a.ID, Type: TriggerTypeScheduled, Config: cfg, Enabled: true}
|
||||
if err := svc.store.CreateTrigger(ctx, trig); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
insertTask(t, svc.store, "task-archived", true)
|
||||
if err := svc.store.CreateRun(ctx, &AutomationRun{
|
||||
AutomationID: a.ID,
|
||||
TriggerID: trig.ID,
|
||||
TriggerType: TriggerTypeScheduled,
|
||||
Status: RunStatusTaskCreated,
|
||||
TaskID: "task-archived",
|
||||
DedupKey: "prior-run",
|
||||
TriggerData: json.RawMessage(`{}`),
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if err := svc.FireTrigger(ctx, a.ID, trig.ID, TriggerTypeScheduled, json.RawMessage(`{}`), "new-run"); err != nil {
|
||||
t.Fatalf("FireTrigger returned error: %v", err)
|
||||
}
|
||||
|
||||
runs, err := svc.store.ListRuns(ctx, a.ID, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, r := range runs {
|
||||
if r.Status == RunStatusSkipped {
|
||||
t.Fatalf("expected the new trigger to fire instead of skip for the cap; runs=%+v", runs)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -12,6 +12,8 @@ import (
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jmoiron/sqlx"
|
||||
|
||||
"github.com/kandev/kandev/internal/db"
|
||||
)
|
||||
|
||||
// Store provides SQLite persistence for automations.
|
||||
@@ -435,18 +437,60 @@ func (s *Store) updateRunTerminalStatus(ctx context.Context, taskID string, stat
|
||||
return err
|
||||
}
|
||||
|
||||
// ListRuns returns recent runs for an automation.
|
||||
// ListRuns returns recent runs for an automation. A task_created run whose
|
||||
// generated task has been archived or no longer exists is reported as
|
||||
// cancelled — the task's outcome is unknown and it's no longer outstanding
|
||||
// work — without touching runs that already reached a real terminal
|
||||
// status. Falls back to the raw stored status when the tasks table isn't
|
||||
// present (isolated automation-only tests; production always has it,
|
||||
// migrated by the task repository before automation triggers can fire).
|
||||
func (s *Store) ListRuns(ctx context.Context, automationID string, limit int) ([]*AutomationRun, error) {
|
||||
if limit <= 0 {
|
||||
limit = 50
|
||||
}
|
||||
runs, err := s.listRunsWithTaskState(ctx, automationID, limit)
|
||||
if db.IsMissingTableError(err) {
|
||||
runs, err = s.listRunsRaw(ctx, automationID, limit)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, r := range runs {
|
||||
r.TriggerData = json.RawMessage(r.TriggerDataJSON)
|
||||
}
|
||||
return runs, nil
|
||||
}
|
||||
|
||||
func (s *Store) listRunsWithTaskState(ctx context.Context, automationID string, limit int) ([]*AutomationRun, error) {
|
||||
// Assumes a task_created run always carries a non-empty ar.task_id: the
|
||||
// sole production write path (orchestrator's recordSuccessRun) sets
|
||||
// TaskID and Status together in the same INSERT. If that's ever
|
||||
// violated, the LEFT JOIN never matches an empty task_id against a
|
||||
// real task row, so the run falls into the "no live task" branch below
|
||||
// and displays as cancelled rather than its raw stored status —
|
||||
// reachable today only through the e2e run-seeding endpoint, never in
|
||||
// production.
|
||||
var runs []*AutomationRun
|
||||
err := s.ro.SelectContext(ctx, &runs, `
|
||||
SELECT ar.id, ar.automation_id, ar.trigger_id, ar.trigger_type, ar.task_id,
|
||||
CASE
|
||||
WHEN ar.status = ? AND (t.id IS NULL OR t.archived_at IS NOT NULL) THEN ?
|
||||
ELSE ar.status
|
||||
END AS status,
|
||||
ar.dedup_key, ar.trigger_data, ar.error_message, ar.created_at
|
||||
FROM automation_runs ar
|
||||
LEFT JOIN tasks t ON t.id = ar.task_id
|
||||
WHERE ar.automation_id = ?
|
||||
ORDER BY ar.created_at DESC LIMIT ?`,
|
||||
string(RunStatusTaskCreated), string(RunStatusCancelled), automationID, limit)
|
||||
return runs, err
|
||||
}
|
||||
|
||||
func (s *Store) listRunsRaw(ctx context.Context, automationID string, limit int) ([]*AutomationRun, error) {
|
||||
var runs []*AutomationRun
|
||||
err := s.ro.SelectContext(ctx, &runs,
|
||||
`SELECT * FROM automation_runs WHERE automation_id = ? ORDER BY created_at DESC LIMIT ?`,
|
||||
automationID, limit)
|
||||
for _, r := range runs {
|
||||
r.TriggerData = json.RawMessage(r.TriggerDataJSON)
|
||||
}
|
||||
return runs, err
|
||||
}
|
||||
|
||||
@@ -462,8 +506,36 @@ func (s *Store) HasRunWithDedupKey(ctx context.Context, automationID, dedupKey s
|
||||
return count > 0, err
|
||||
}
|
||||
|
||||
// CountActiveRuns returns the number of runs with task_created status for an automation.
|
||||
// CountActiveRuns returns the number of runs with task_created status for
|
||||
// an automation whose generated task is still open. A task_created run
|
||||
// whose task was archived or deleted no longer represents outstanding
|
||||
// work — the user closed it out some other way — so it must not keep
|
||||
// counting against max_concurrent_runs forever. Falls back to a plain
|
||||
// count when the tasks table isn't present (isolated automation-only
|
||||
// tests; production always has it).
|
||||
func (s *Store) CountActiveRuns(ctx context.Context, automationID string) (int, error) {
|
||||
count, err := s.countActiveRunsWithTaskState(ctx, automationID)
|
||||
if db.IsMissingTableError(err) {
|
||||
return s.countActiveRunsRaw(ctx, automationID)
|
||||
}
|
||||
return count, err
|
||||
}
|
||||
|
||||
func (s *Store) countActiveRunsWithTaskState(ctx context.Context, automationID string) (int, error) {
|
||||
// Same non-empty-task_id assumption as listRunsWithTaskState above: an
|
||||
// empty ar.task_id never matches a real task row either, so such a run
|
||||
// silently falls out of the active count below instead of erroring.
|
||||
var count int
|
||||
err := s.ro.GetContext(ctx, &count, `
|
||||
SELECT COUNT(*) FROM automation_runs ar
|
||||
LEFT JOIN tasks t ON t.id = ar.task_id
|
||||
WHERE ar.automation_id = ? AND ar.status = ?
|
||||
AND t.id IS NOT NULL AND t.archived_at IS NULL`,
|
||||
automationID, string(RunStatusTaskCreated))
|
||||
return count, err
|
||||
}
|
||||
|
||||
func (s *Store) countActiveRunsRaw(ctx context.Context, automationID string) (int, error) {
|
||||
var count int
|
||||
err := s.ro.GetContext(ctx, &count,
|
||||
`SELECT COUNT(*) FROM automation_runs WHERE automation_id = ? AND status = ?`,
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/jmoiron/sqlx"
|
||||
_ "github.com/mattn/go-sqlite3"
|
||||
@@ -298,3 +299,185 @@ func TestDeleteAllRuns(t *testing.T) {
|
||||
t.Errorf("expected 0 runs after delete, got %d", len(runs))
|
||||
}
|
||||
}
|
||||
|
||||
// createTasksTable adds a minimal shadow of the task repository's `tasks`
|
||||
// table (id, archived_at) to the store's DB. The automation package never
|
||||
// owns this table — apps/backend/internal/task/repository/sqlite is the
|
||||
// canonical owner — so only tests that exercise the CountActiveRuns/ListRuns
|
||||
// task-state join create it; production always has the real table already
|
||||
// migrated by the task repository before automation triggers can fire.
|
||||
func createTasksTable(t *testing.T, store *Store) {
|
||||
t.Helper()
|
||||
if _, err := store.db.Exec(`CREATE TABLE tasks (id TEXT PRIMARY KEY, archived_at DATETIME)`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func insertTask(t *testing.T, store *Store, id string, archived bool) {
|
||||
t.Helper()
|
||||
var archivedAt interface{}
|
||||
if archived {
|
||||
archivedAt = time.Now().UTC()
|
||||
}
|
||||
if _, err := store.db.Exec(`INSERT INTO tasks (id, archived_at) VALUES (?, ?)`, id, archivedAt); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCountActiveRuns_ExcludesArchivedOrMissingTask reproduces the reported
|
||||
// bug: an automation-generated task that gets archived (manually, via
|
||||
// auto-archive, or via cascade) leaves its automation run stuck at
|
||||
// task_created forever unless CountActiveRuns checks the task's current
|
||||
// state. A run whose task is archived, gone entirely, or never recorded
|
||||
// (empty task_id — see countActiveRunsWithTaskState's docstring in
|
||||
// store.go) no longer represents outstanding work and must not count
|
||||
// against max_concurrent_runs.
|
||||
func TestCountActiveRuns_ExcludesArchivedOrMissingTask(t *testing.T) {
|
||||
store := setupTestStore(t)
|
||||
createTasksTable(t, store)
|
||||
ctx := context.Background()
|
||||
|
||||
a := &Automation{WorkspaceID: "ws-1", Name: "A", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true}
|
||||
if err := store.CreateAutomation(ctx, a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
insertTask(t, store, "task-active", false)
|
||||
insertTask(t, store, "task-archived", true)
|
||||
// "task-missing" deliberately has no row in tasks at all; "" (the
|
||||
// task_id column default) exercises the same no-live-task branch
|
||||
// through a different route.
|
||||
|
||||
for _, tid := range []string{"task-active", "task-archived", "task-missing", ""} {
|
||||
if err := store.CreateRun(ctx, &AutomationRun{
|
||||
AutomationID: a.ID,
|
||||
TriggerType: TriggerTypeScheduled,
|
||||
Status: RunStatusTaskCreated,
|
||||
TaskID: tid,
|
||||
TriggerData: json.RawMessage(`{}`),
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
count, err := store.CountActiveRuns(ctx, a.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 1 {
|
||||
t.Fatalf("expected 1 active run (only task-active is open), got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCountActiveRuns_FallsBackWhenTasksTableAbsent guards the isolated
|
||||
// automation-only test DB (and any other DB where the task repository
|
||||
// hasn't initialised yet): CountActiveRuns must keep counting by status
|
||||
// alone rather than error when the tasks table doesn't exist.
|
||||
func TestCountActiveRuns_FallsBackWhenTasksTableAbsent(t *testing.T) {
|
||||
store := setupTestStore(t)
|
||||
ctx := context.Background()
|
||||
|
||||
a := &Automation{WorkspaceID: "ws-1", Name: "A", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true}
|
||||
if err := store.CreateAutomation(ctx, a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.CreateRun(ctx, &AutomationRun{
|
||||
AutomationID: a.ID,
|
||||
TriggerType: TriggerTypeScheduled,
|
||||
Status: RunStatusTaskCreated,
|
||||
TaskID: "task-xyz",
|
||||
TriggerData: json.RawMessage(`{}`),
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
count, err := store.CountActiveRuns(ctx, a.ID)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if count != 1 {
|
||||
t.Fatalf("expected 1 active run when tasks table is absent, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
// TestListRuns_ArchivedOrMissingTaskShowsCancelled ensures the "Recent Runs"
|
||||
// list the settings UI reads stops labeling a run "Running" once its
|
||||
// generated task is archived, gone, or never recorded (empty task_id — see
|
||||
// listRunsWithTaskState's docstring in store.go), without touching the
|
||||
// stored status of runs that already reached a real terminal outcome.
|
||||
func TestListRuns_ArchivedOrMissingTaskShowsCancelled(t *testing.T) {
|
||||
store := setupTestStore(t)
|
||||
createTasksTable(t, store)
|
||||
ctx := context.Background()
|
||||
|
||||
a := &Automation{WorkspaceID: "ws-1", Name: "A", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true}
|
||||
if err := store.CreateAutomation(ctx, a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
insertTask(t, store, "task-active", false)
|
||||
insertTask(t, store, "task-archived", true)
|
||||
|
||||
active := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-active", TriggerData: json.RawMessage(`{}`)}
|
||||
archived := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-archived", TriggerData: json.RawMessage(`{}`)}
|
||||
missing := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "task-missing", TriggerData: json.RawMessage(`{}`)}
|
||||
emptyTaskID := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusTaskCreated, TaskID: "", TriggerData: json.RawMessage(`{}`)}
|
||||
succeededOnArchived := &AutomationRun{AutomationID: a.ID, TriggerType: TriggerTypeScheduled, Status: RunStatusSucceeded, TaskID: "task-archived", TriggerData: json.RawMessage(`{}`)}
|
||||
for _, r := range []*AutomationRun{active, archived, missing, emptyTaskID, succeededOnArchived} {
|
||||
if err := store.CreateRun(ctx, r); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
got, err := store.ListRuns(ctx, a.ID, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
statusByID := map[string]RunStatus{}
|
||||
for _, r := range got {
|
||||
statusByID[r.ID] = r.Status
|
||||
}
|
||||
if s := statusByID[active.ID]; s != RunStatusTaskCreated {
|
||||
t.Errorf("active task run: expected task_created, got %q", s)
|
||||
}
|
||||
if s := statusByID[archived.ID]; s != RunStatusCancelled {
|
||||
t.Errorf("archived task run: expected cancelled, got %q", s)
|
||||
}
|
||||
if s := statusByID[missing.ID]; s != RunStatusCancelled {
|
||||
t.Errorf("missing task run: expected cancelled, got %q", s)
|
||||
}
|
||||
if s := statusByID[emptyTaskID.ID]; s != RunStatusCancelled {
|
||||
t.Errorf("empty task_id run: expected cancelled, got %q", s)
|
||||
}
|
||||
if s := statusByID[succeededOnArchived.ID]; s != RunStatusSucceeded {
|
||||
t.Errorf("already-succeeded run on now-archived task: expected succeeded preserved, got %q", s)
|
||||
}
|
||||
}
|
||||
|
||||
// TestListRuns_FallsBackWhenTasksTableAbsent mirrors
|
||||
// TestCountActiveRuns_FallsBackWhenTasksTableAbsent for the display path.
|
||||
func TestListRuns_FallsBackWhenTasksTableAbsent(t *testing.T) {
|
||||
store := setupTestStore(t)
|
||||
ctx := context.Background()
|
||||
|
||||
a := &Automation{WorkspaceID: "ws-1", Name: "A", WorkflowID: "wf-1", WorkflowStepID: "s-1", Enabled: true}
|
||||
if err := store.CreateAutomation(ctx, a); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.CreateRun(ctx, &AutomationRun{
|
||||
AutomationID: a.ID,
|
||||
TriggerType: TriggerTypeScheduled,
|
||||
Status: RunStatusTaskCreated,
|
||||
TaskID: "task-xyz",
|
||||
TriggerData: json.RawMessage(`{}`),
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
got, err := store.ListRuns(ctx, a.ID, 10)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 1 || got[0].Status != RunStatusTaskCreated {
|
||||
t.Fatalf("expected 1 run with status task_created when tasks table is absent, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -261,6 +261,10 @@ func handleE2ECreateAutomation(svc *automation.Service, log *logger.Logger) gin.
|
||||
type e2eCreateAutomationRunRequest struct {
|
||||
AutomationID string `json:"automation_id"`
|
||||
Status string `json:"status"`
|
||||
// TaskID optionally links the seeded run to a real task so E2E tests can
|
||||
// exercise task-lifecycle interactions (e.g. archiving the task and
|
||||
// asserting the run's displayed status/concurrency accounting reacts).
|
||||
TaskID string `json:"task_id"`
|
||||
}
|
||||
|
||||
// handleE2ECreateAutomationRun seeds an automation run row for E2E tests.
|
||||
@@ -279,6 +283,7 @@ func handleE2ECreateAutomationRun(svc *automation.Service, log *logger.Logger) g
|
||||
AutomationID: body.AutomationID,
|
||||
TriggerType: automation.TriggerTypeScheduled,
|
||||
Status: status,
|
||||
TaskID: body.TaskID,
|
||||
TriggerData: []byte(`{}`),
|
||||
}
|
||||
if err := svc.RecordRun(c.Request.Context(), run); err != nil {
|
||||
@@ -286,6 +291,8 @@ func handleE2ECreateAutomationRun(svc *automation.Service, log *logger.Logger) g
|
||||
c.JSON(http.StatusInternalServerError, gin.H{errKey: err.Error()})
|
||||
return
|
||||
}
|
||||
c.JSON(http.StatusCreated, gin.H{"id": run.ID, "automation_id": run.AutomationID, statusKey: run.Status})
|
||||
c.JSON(http.StatusCreated, gin.H{
|
||||
"id": run.ID, "automation_id": run.AutomationID, statusKey: run.Status, taskIDPayloadKey: run.TaskID,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ const (
|
||||
postgresDuplicateColumn = "42701"
|
||||
postgresDuplicateTable = "42P07"
|
||||
postgresDuplicateObject = "42710"
|
||||
postgresUndefinedTable = "42P01"
|
||||
sqliteAlreadyExistsText = " already exists"
|
||||
)
|
||||
|
||||
@@ -54,3 +55,20 @@ func isSQLiteDuplicateObjectMessage(s string) bool {
|
||||
strings.HasPrefix(s, "trigger ") && strings.Contains(s, sqliteAlreadyExistsText) ||
|
||||
strings.HasPrefix(s, "view ") && strings.Contains(s, sqliteAlreadyExistsText)
|
||||
}
|
||||
|
||||
// IsMissingTableError reports whether err means a query referenced a table
|
||||
// that does not exist in this database — e.g. a cross-domain JOIN against a
|
||||
// table owned by another package's schema that hasn't been initialised yet
|
||||
// (isolated unit tests that only set up one domain's tables). Callers use
|
||||
// this to fall back to a query that doesn't need the missing table, rather
|
||||
// than failing outright.
|
||||
func IsMissingTableError(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
var pgErr *pgconn.PgError
|
||||
if errors.As(err, &pgErr) {
|
||||
return pgErr.Code == postgresUndefinedTable
|
||||
}
|
||||
return strings.Contains(err.Error(), "no such table")
|
||||
}
|
||||
|
||||
@@ -143,3 +143,55 @@ func TestIsAlreadyExistsError(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestIsMissingTableError(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
err error
|
||||
want bool
|
||||
}{
|
||||
{
|
||||
name: "sqlite no such table",
|
||||
err: errors.New("no such table: tasks"),
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "wrapped sqlite no such table",
|
||||
err: fmt.Errorf("count active runs: %w", errors.New("no such table: tasks")),
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "postgres undefined table",
|
||||
err: &pgconn.PgError{Code: postgresUndefinedTable},
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "wrapped postgres undefined table",
|
||||
err: fmt.Errorf("count active runs: %w", &pgconn.PgError{Code: postgresUndefinedTable}),
|
||||
want: true,
|
||||
},
|
||||
{
|
||||
name: "postgres undefined column is not a missing table",
|
||||
err: &pgconn.PgError{Code: "42703"},
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "unrelated",
|
||||
err: errors.New("duplicate column name: branch_slug"),
|
||||
want: false,
|
||||
},
|
||||
{
|
||||
name: "nil",
|
||||
err: nil,
|
||||
want: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := IsMissingTableError(tt.err); got != tt.want {
|
||||
t.Fatalf("IsMissingTableError() = %v, want %v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,10 +11,22 @@ const __dirname = dirname(fileURLToPath(import.meta.url));
|
||||
const desktopRoot = resolve(__dirname, "..");
|
||||
const repoRoot = resolve(desktopRoot, "../..");
|
||||
|
||||
if (process.argv[2] === "--fake-runtime") {
|
||||
await runFakeRuntime(process.argv[3], process.argv.slice(4));
|
||||
} else {
|
||||
await runSmoke();
|
||||
// Keep above HEALTH_TIMEOUT in apps/desktop/src-tauri/src/backend.rs (60s): the Rust side
|
||||
// retries GET /health every 250ms until that deadline, so this must give it enough room to
|
||||
// finish, or a launcher Rust would still consider "starting" gets killed here first.
|
||||
// desktop-launch-smoke.test.mjs asserts the relationship so the two stay in sync.
|
||||
export const HEALTH_REQUESTED_TIMEOUT_MS = 90_000;
|
||||
export const ROOT_REQUESTED_TIMEOUT_MS = 60_000;
|
||||
|
||||
// Only run the CLI behavior when this file is executed directly (`node desktop-launch-smoke.mjs`
|
||||
// or the fake-runtime re-exec below) — not when desktop-launch-smoke.test.mjs imports it.
|
||||
const isEntryPoint = resolve(process.argv[1] ?? "") === fileURLToPath(import.meta.url);
|
||||
if (isEntryPoint) {
|
||||
if (process.argv[2] === "--fake-runtime") {
|
||||
await runFakeRuntime(process.argv[3], process.argv.slice(4));
|
||||
} else {
|
||||
await runSmoke();
|
||||
}
|
||||
}
|
||||
|
||||
async function runSmoke() {
|
||||
@@ -60,10 +72,21 @@ async function runSmoke() {
|
||||
throw new Error(`desktop app exited early with code ${child.exitCode}\n${stdout}\n${stderr}`);
|
||||
}
|
||||
};
|
||||
const describeChild = () => `[stdout]\n${stdout}\n[stderr]\n${stderr}`;
|
||||
|
||||
try {
|
||||
await waitForFile(join(stateDir, "health-requested"), 45_000, failIfExited);
|
||||
await waitForFile(join(stateDir, "root-requested"), 45_000, failIfExited);
|
||||
await waitForFile(
|
||||
join(stateDir, "health-requested"),
|
||||
HEALTH_REQUESTED_TIMEOUT_MS,
|
||||
failIfExited,
|
||||
describeChild,
|
||||
);
|
||||
await waitForFile(
|
||||
join(stateDir, "root-requested"),
|
||||
ROOT_REQUESTED_TIMEOUT_MS,
|
||||
failIfExited,
|
||||
describeChild,
|
||||
);
|
||||
} finally {
|
||||
await stopProcess(child);
|
||||
}
|
||||
@@ -151,7 +174,7 @@ async function runFakeRuntime(stateDir, args) {
|
||||
process.on("SIGINT", stop);
|
||||
}
|
||||
|
||||
async function waitForFile(path, timeoutMs, tick) {
|
||||
export async function waitForFile(path, timeoutMs, tick, describeDetail) {
|
||||
const deadline = Date.now() + timeoutMs;
|
||||
while (Date.now() < deadline) {
|
||||
tick?.();
|
||||
@@ -160,7 +183,8 @@ async function waitForFile(path, timeoutMs, tick) {
|
||||
}
|
||||
await new Promise((resolveWait) => setTimeout(resolveWait, 250));
|
||||
}
|
||||
throw new Error(`Timed out waiting for ${path}`);
|
||||
const detail = describeDetail?.();
|
||||
throw new Error(`Timed out waiting for ${path}${detail ? `\n\n${detail}` : ""}`);
|
||||
}
|
||||
|
||||
async function stopProcess(child) {
|
||||
|
||||
@@ -0,0 +1,98 @@
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { dirname, join, resolve } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "node:test";
|
||||
|
||||
import {
|
||||
HEALTH_REQUESTED_TIMEOUT_MS,
|
||||
ROOT_REQUESTED_TIMEOUT_MS,
|
||||
waitForFile,
|
||||
} from "./desktop-launch-smoke.mjs";
|
||||
|
||||
const __dirname = dirname(fileURLToPath(import.meta.url));
|
||||
const backendRsPath = resolve(__dirname, "../src-tauri/src/backend.rs");
|
||||
|
||||
async function withTempDir(run) {
|
||||
const dir = await mkdtemp(join(tmpdir(), "wait-for-file-"));
|
||||
try {
|
||||
return await run(dir);
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
test("waitForFile resolves once the target file appears", async () => {
|
||||
await withTempDir(async (dir) => {
|
||||
const target = join(dir, "marker");
|
||||
const write = new Promise((r) => setTimeout(r, 50)).then(() => writeFile(target, "1"));
|
||||
await Promise.all([waitForFile(target, 2_000), write]);
|
||||
});
|
||||
});
|
||||
|
||||
test("waitForFile throws a timeout error including the describeDetail() text", async () => {
|
||||
await withTempDir(async (dir) => {
|
||||
const target = join(dir, "never-appears");
|
||||
await assert.rejects(
|
||||
() => waitForFile(target, 100, undefined, () => "custom diagnostic detail"),
|
||||
(err) => {
|
||||
assert.match(err.message, /Timed out waiting for/);
|
||||
assert.match(err.message, /custom diagnostic detail/);
|
||||
return true;
|
||||
},
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
test("waitForFile omits the detail block when describeDetail is not provided", async () => {
|
||||
await withTempDir(async (dir) => {
|
||||
const target = join(dir, "never-appears");
|
||||
await assert.rejects(
|
||||
() => waitForFile(target, 100),
|
||||
(err) => {
|
||||
assert.equal(err.message, `Timed out waiting for ${target}`);
|
||||
return true;
|
||||
},
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
test("waitForFile calls tick on every poll and surfaces a tick failure immediately", async () => {
|
||||
await withTempDir(async (dir) => {
|
||||
const target = join(dir, "never-appears");
|
||||
let calls = 0;
|
||||
await assert.rejects(
|
||||
() =>
|
||||
waitForFile(target, 5_000, () => {
|
||||
calls += 1;
|
||||
if (calls >= 2) {
|
||||
throw new Error("boom");
|
||||
}
|
||||
}),
|
||||
/boom/,
|
||||
);
|
||||
assert.ok(calls >= 2, `expected at least 2 tick() calls, got ${calls}`);
|
||||
});
|
||||
});
|
||||
|
||||
test("health-requested timeout stays above the Rust backend's own HEALTH_TIMEOUT", async () => {
|
||||
const source = await readFile(backendRsPath, "utf8");
|
||||
const match = source.match(/const HEALTH_TIMEOUT: Duration = Duration::from_secs\((\d+)\);/);
|
||||
assert.ok(
|
||||
match,
|
||||
"could not find HEALTH_TIMEOUT in backend.rs — update this test if it moved or was renamed",
|
||||
);
|
||||
|
||||
const rustHealthTimeoutMs = Number(match[1]) * 1_000;
|
||||
assert.ok(
|
||||
HEALTH_REQUESTED_TIMEOUT_MS > rustHealthTimeoutMs,
|
||||
`HEALTH_REQUESTED_TIMEOUT_MS (${HEALTH_REQUESTED_TIMEOUT_MS}ms) must exceed backend.rs's HEALTH_TIMEOUT ` +
|
||||
`(${rustHealthTimeoutMs}ms) — otherwise this smoke test can kill a launcher the Rust side would still ` +
|
||||
"consider healthy-pending, turning a legitimately slow CI run into a spurious failure.",
|
||||
);
|
||||
assert.ok(
|
||||
ROOT_REQUESTED_TIMEOUT_MS > 0 && Number.isInteger(ROOT_REQUESTED_TIMEOUT_MS),
|
||||
"ROOT_REQUESTED_TIMEOUT_MS must be a positive integer",
|
||||
);
|
||||
});
|
||||
@@ -37,6 +37,10 @@ const STATUS_BADGE: Record<
|
||||
succeeded: { variant: "default", label: "Succeeded" },
|
||||
failed: { variant: "destructive", label: "Failed" },
|
||||
skipped: { variant: "outline", label: "Skipped" },
|
||||
// The generating task was archived or no longer exists — its outcome is
|
||||
// unknown, so this is deliberately distinct from succeeded/failed rather
|
||||
// than guessing one. See internal/automation.RunStatusCancelled.
|
||||
cancelled: { variant: "outline", label: "Cancelled" },
|
||||
};
|
||||
|
||||
type RunRowProps = {
|
||||
|
||||
@@ -1961,10 +1961,12 @@ export class ApiClient {
|
||||
async seedAutomationRun(
|
||||
automationId: string,
|
||||
status = "skipped",
|
||||
): Promise<{ id: string; automation_id: string; status: string }> {
|
||||
taskId?: string,
|
||||
): Promise<{ id: string; automation_id: string; status: string; task_id: string }> {
|
||||
return this.request("POST", "/api/v1/e2e/automation-runs", {
|
||||
automation_id: automationId,
|
||||
status,
|
||||
task_id: taskId ?? "",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -349,4 +349,60 @@ test.describe("Automations settings page", () => {
|
||||
// Table should show the empty state.
|
||||
await expect(testPage.getByText("No runs yet")).toBeVisible({ timeout: 5_000 });
|
||||
});
|
||||
|
||||
test("archived task's run shows Cancelled instead of Running", async ({
|
||||
testPage,
|
||||
seedData,
|
||||
apiClient,
|
||||
}) => {
|
||||
// Regression test: an automation-generated task that gets archived
|
||||
// (manually, via auto-archive, or via cascade) before its run is
|
||||
// otherwise finalized used to leave the run stuck at "task_created"
|
||||
// forever, displayed as "Running" and permanently pinned against
|
||||
// max_concurrent_runs. See internal/automation.Store.CountActiveRuns /
|
||||
// ListRuns.
|
||||
const automation = await apiClient.seedAutomation({
|
||||
workspaceId: seedData.workspaceId,
|
||||
name: "Archived Task Run Test",
|
||||
workflowId: seedData.workflowId,
|
||||
workflowStepId: seedData.startStepId,
|
||||
});
|
||||
|
||||
const archivedTask = await apiClient.createTask(
|
||||
seedData.workspaceId,
|
||||
"Archived Automation Task",
|
||||
{ workflow_id: seedData.workflowId, workflow_step_id: seedData.startStepId },
|
||||
);
|
||||
const openTask = await apiClient.createTask(seedData.workspaceId, "Open Automation Task", {
|
||||
workflow_id: seedData.workflowId,
|
||||
workflow_step_id: seedData.startStepId,
|
||||
});
|
||||
await apiClient.seedAutomationRun(automation.id, "task_created", archivedTask.id);
|
||||
await apiClient.seedAutomationRun(automation.id, "task_created", openTask.id);
|
||||
await apiClient.archiveTask(archivedTask.id);
|
||||
|
||||
await testPage.goto(`/settings/workspace/${seedData.workspaceId}/automations/${automation.id}`);
|
||||
await testPage.getByTestId("automation-editor").waitFor({ state: "visible", timeout: 15_000 });
|
||||
|
||||
const scrollContainer = testPage.getByTestId("settings-scroll-container");
|
||||
await scrollContainer.evaluate((el) => (el.scrollTop = el.scrollHeight));
|
||||
|
||||
const recentRunsButton = testPage.locator("button", { hasText: /Recent Runs/ });
|
||||
await recentRunsButton.waitFor({ state: "visible", timeout: 10_000 });
|
||||
await recentRunsButton.click();
|
||||
|
||||
const tbody = testPage.locator("table tbody");
|
||||
await tbody.waitFor({ state: "visible", timeout: 5_000 });
|
||||
await expect(tbody.locator("tr")).toHaveCount(2, { timeout: 10_000 });
|
||||
|
||||
// The archived task's run is no longer outstanding work.
|
||||
const archivedRow = testPage.locator("table tbody tr", {
|
||||
hasText: archivedTask.id.slice(0, 8),
|
||||
});
|
||||
await expect(archivedRow.getByText("Cancelled", { exact: true })).toBeVisible();
|
||||
|
||||
// The still-open task's run is unaffected.
|
||||
const openRow = testPage.locator("table tbody tr", { hasText: openTask.id.slice(0, 8) });
|
||||
await expect(openRow.getByText("Running", { exact: true })).toBeVisible();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -2,7 +2,13 @@
|
||||
|
||||
export type TriggerType = "scheduled" | "github_pr" | "github_push" | "github_ci" | "webhook";
|
||||
|
||||
export type RunStatus = "triggered" | "task_created" | "succeeded" | "failed" | "skipped";
|
||||
export type RunStatus =
|
||||
| "triggered"
|
||||
| "task_created"
|
||||
| "succeeded"
|
||||
| "failed"
|
||||
| "skipped"
|
||||
| "cancelled";
|
||||
|
||||
// ExecutionMode controls whether an automation firing creates a visible
|
||||
// kanban task ("task", the default) or an ephemeral run hidden from the
|
||||
|
||||
Reference in New Issue
Block a user