From d1290885484a108c2ff346b20850b852e32aabe8 Mon Sep 17 00:00:00 2001 From: dawn Date: Thu, 27 Aug 2026 13:14:37 +0900 Subject: [PATCH] spindle: replace events with pipeline and workflow status tables Signed-off-by: dawn --- orm/orm.go | 1 + spindle/db/db.go | 179 ++++++++++++ spindle/db/delete_test.go | 46 ++- spindle/db/events.go | 238 +++++++--------- spindle/db/events_metrics_test.go | 20 +- spindle/db/jobs.go | 22 +- spindle/db/jobs_test.go | 4 +- spindle/db/mill_state.go | 40 +-- spindle/db/mill_state_test.go | 14 +- spindle/db/pipelines.go | 307 +++++++++++++-------- spindle/db/pipelines_test.go | 176 ++++++++++-- spindle/engine/engine.go | 4 +- spindle/engine/engine_test.go | 18 +- spindle/mill/auth_test.go | 8 +- spindle/mill/executor/observe.go | 22 +- spindle/mill/executor/outbox.go | 20 +- spindle/mill/executor/reserved_test.go | 10 +- spindle/mill/integration_test.go | 11 +- spindle/mill/mill.go | 10 +- spindle/mill/mill_test.go | 6 +- spindle/mill/restore.go | 3 +- spindle/mill/restore_test.go | 26 +- spindle/observability/metrics.go | 11 +- spindle/observability/metrics_test.go | 32 +++ spindle/server.go | 7 +- spindle/stream.go | 142 ---------- spindle/tapclient.go | 2 +- spindle/wipe.go | 2 +- spindle/wipe_test.go | 19 +- spindle/xrpc/ci_pipeline_subscribe_logs.go | 27 +- spindle/xrpc/pipeline_cancel_pipeline.go | 2 +- spindle/xrpc/xrpc_test.go | 12 +- 32 files changed, 804 insertions(+), 637 deletions(-) delete mode 100644 spindle/stream.go diff --git a/orm/orm.go b/orm/orm.go index bc37ec48b..927183787 100644 --- a/orm/orm.go +++ b/orm/orm.go @@ -83,6 +83,7 @@ func newFilter(key, cmp string, arg any) Filter { func FilterEq(key string, arg any) Filter { return newFilter(key, "=", arg) } func FilterNotEq(key string, arg any) Filter { return newFilter(key, "<>", arg) } func FilterGte(key string, arg any) Filter { return newFilter(key, ">=", arg) } +func FilterLt(key string, arg any) Filter { return newFilter(key, "<", arg) } func FilterLte(key string, arg any) Filter { return newFilter(key, "<=", arg) } func FilterIs(key string, arg any) Filter { return newFilter(key, "is", arg) } func FilterIsNot(key string, arg any) Filter { return newFilter(key, "is not", arg) } diff --git a/spindle/db/db.go b/spindle/db/db.go index f02874609..162e9ca2b 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -3,11 +3,14 @@ package db import ( "context" "database/sql" + "encoding/json" "log/slog" "slices" "strings" + "time" _ "github.com/mattn/go-sqlite3" + "tangled.org/core/api/tangled" "tangled.org/core/log" "tangled.org/core/orm" ) @@ -809,6 +812,58 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } + if err := orm.RunMigration(conn, logger, "pipelines-and-workflow-statuses", func(tx *sql.Tx) error { + if _, err := tx.Exec(` + drop table if exists workflows; + drop table if exists pipelines; + `); err != nil { + return err + } + + if _, err := tx.Exec(` + create table pipelines ( + id integer primary key autoincrement, + rkey text not null unique, + knot text not null, + repo_did text not null, + commit_sha text not null, + kind text not null, + payload text not null + ); + create index idx_pipelines_repo_id on pipelines(repo_did, id); + create index idx_pipelines_repo_commit on pipelines(repo_did, commit_sha); + + create table workflow_statuses ( + id integer primary key autoincrement, + rkey text not null, + workflow text not null, + status text not null, + error text, + exit_code integer, + created_at text not null + ); + create index idx_workflow_statuses_lookup on workflow_statuses(rkey, workflow, id); + `); err != nil { + return err + } + + if err := migratePipelines(tx, logger); err != nil { + return err + } + if err := migrateWorkflowStatuses(tx, logger); err != nil { + return err + } + + _, err := tx.Exec(` + drop index if exists idx_events_pipeline_lookup; + drop index if exists idx_events_pipeline_status; + alter table events rename to events_legacy; + `) + return err + }); err != nil { + return err + } + return nil } @@ -882,3 +937,127 @@ func (d *DB) GetRepoOwnerAndDid(knot, rkey string) (owner string, repoDid string err = d.QueryRow("SELECT owner, repo_did FROM repos WHERE knot = ? AND rkey = ?", knot, rkey).Scan(&owner, &repoDid) return } + +// migratePipelines converts the legacy sh.tangled.pipeline event rows into +// sh.tangled.ci.pipeline payloads. ordered by created so the new rowids follow +// historical order, which is what the pagination cursor reads. +func migratePipelines(tx *sql.Tx, logger *slog.Logger) error { + rows, err := tx.Query( + `select rkey, event, created from events where nsid = ? order by created asc`, + tangled.PipelineNSID, + ) + if err != nil { + return err + } + defer rows.Close() + + type converted struct { + rkey, knot, repoDid, commitSha, kind string + payload []byte + } + var out []converted + var skipped int + + for rows.Next() { + var rkey, eventJson string + var created int64 + if err := rows.Scan(&rkey, &eventJson, &created); err != nil { + return err + } + + var raw tangled.Pipeline + if err := json.Unmarshal([]byte(eventJson), &raw); err != nil { + skipped++ + continue + } + if raw.TriggerMetadata == nil { + skipped++ + continue + } + + p, kind := mapToCiPipeline(rkey, time.Unix(0, created), raw) + payload, err := json.Marshal(p) + if err != nil { + skipped++ + continue + } + var knot string + if raw.TriggerMetadata.Repo != nil { + knot = raw.TriggerMetadata.Repo.Knot + } + out = append(out, converted{rkey, knot, p.Repo, p.Commit, string(kind), payload}) + } + if err := rows.Err(); err != nil { + return err + } + + for _, c := range out { + if _, err := tx.Exec( + `insert into pipelines (rkey, knot, repo_did, commit_sha, kind, payload) values (?, ?, ?, ?, ?, ?)`, + c.rkey, c.knot, c.repoDid, c.commitSha, c.kind, string(c.payload), + ); err != nil { + return err + } + } + + logger.Info("backfilled pipelines", "converted", len(out), "skipped", skipped) + return nil +} + +// migrateWorkflowStatuses flattens the legacy status event log. +func migrateWorkflowStatuses(tx *sql.Tx, logger *slog.Logger) error { + rows, err := tx.Query( + `select event from events where nsid = ? order by created asc`, + tangled.PipelineStatusNSID, + ) + if err != nil { + return err + } + defer rows.Close() + + type converted struct { + rkey, workflow, status, createdAt string + wfError *string + exitCode *int64 + } + var out []converted + var skipped int + + for rows.Next() { + var eventJson string + if err := rows.Scan(&eventJson); err != nil { + return err + } + + var st tangled.PipelineStatus + if err := json.Unmarshal([]byte(eventJson), &st); err != nil { + skipped++ + continue + } + + idx := strings.LastIndex(st.Pipeline, "/") + if idx < 0 || idx == len(st.Pipeline)-1 { + skipped++ + continue + } + rkey := st.Pipeline[idx+1:] + + out = append(out, converted{rkey, st.Workflow, st.Status, st.CreatedAt, st.Error, st.ExitCode}) + } + if err := rows.Err(); err != nil { + return err + } + + for _, c := range out { + if _, err := tx.Exec( + `insert into workflow_statuses (rkey, workflow, status, error, exit_code, created_at) + values (?, ?, ?, ?, ?, ?)`, + c.rkey, c.workflow, c.status, c.wfError, c.exitCode, c.createdAt, + ); err != nil { + return err + } + } + + logger.Info("backfilled workflow statuses", "converted", len(out), "skipped", skipped) + return nil +} diff --git a/spindle/db/delete_test.go b/spindle/db/delete_test.go index 7839e9bb5..dfff5cb61 100644 --- a/spindle/db/delete_test.go +++ b/spindle/db/delete_test.go @@ -2,6 +2,7 @@ package db import ( "context" + "encoding/json" "fmt" "testing" @@ -167,24 +168,20 @@ func TestDeleteEventsByRepo(t *testing.T) { repo := "did:plc:repo1" other := "did:plc:repo2" - pipelineEvent := func(did, knot string) string { - return fmt.Sprintf(`{"triggerMetadata":{"repo":{"repoDid":"%s","knot":"%s"}}}`, did, knot) - } - statusEvent := func(knot, rkey string) string { - return fmt.Sprintf(`{"pipeline":"at://did:web:%s/sh.tangled.pipeline/%s"}`, knot, rkey) - } - - ins := []struct{ nsid, rkey, event string }{ - {"sh.tangled.pipeline", "p1", pipelineEvent(repo, "k1")}, - {"sh.tangled.pipeline", "p2", pipelineEvent(other, "k1")}, - {"sh.tangled.pipeline.status", "s1", statusEvent("k1", "p1")}, - {"sh.tangled.pipeline.status", "s2", statusEvent("k1", "p2")}, - {"sh.tangled.pipeline.status", "s3", statusEvent("k9", "p1")}, - {"sh.tangled.pipeline.status", "s4", statusEvent("k7", "p9")}, - {"sh.tangled.repo", "x1", `{"whatever":1}`}, + insertPipeline := func(rkey, repoDid, knot string) { + p := &tangled.CiPipeline{Id: rkey, Repo: repoDid} + payload, err := json.Marshal(p) + if err != nil { + t.Fatal(err) + } + if _, err := d.Exec(`insert into pipelines (rkey, knot, repo_did, commit_sha, kind, payload) values (?, ?, ?, '', '', ?)`, rkey, knot, repoDid, string(payload)); err != nil { + t.Fatal(err) + } } - for _, e := range ins { - if _, err := d.Exec(`insert into events (nsid, rkey, event, created) values (?, ?, ?, 1)`, e.nsid, e.rkey, e.event); err != nil { + insertPipeline("p1", repo, "k1") + insertPipeline("p2", other, "k1") + for _, rkey := range []string{"p1", "p2", "p9"} { + if _, err := d.Exec(`insert into workflow_statuses (rkey, workflow, status, created_at) values (?, 'build', 'pending', 'now')`, rkey); err != nil { t.Fatal(err) } } @@ -194,16 +191,13 @@ func TestDeleteEventsByRepo(t *testing.T) { } var n int - if err := d.QueryRow(`select count(*) from events`).Scan(&n); err != nil { - t.Fatal(err) - } - if n != 4 { - t.Fatalf("expected 4 events left, got %d", n) + if err := d.QueryRow(`select count(*) from pipelines where rkey = 'p1'`).Scan(&n); err != nil || n != 0 { + t.Fatalf("expected repo pipeline gone, got %d err %v", n, err) } - if err := d.QueryRow(`select count(*) from events where rkey in ('p1', 's1', 's4')`).Scan(&n); err != nil || n != 0 { - t.Fatalf("expected wiped events gone, got %d err %v", n, err) + if err := d.QueryRow(`select count(*) from workflow_statuses where rkey in ('p1', 'p9')`).Scan(&n); err != nil || n != 0 { + t.Fatalf("expected wiped statuses gone, got %d err %v", n, err) } - if err := d.QueryRow(`select count(*) from events where rkey = 's3'`).Scan(&n); err != nil || n != 1 { - t.Fatalf("expected status on unrecorded knot to survive, got %d err %v", n, err) + if err := d.QueryRow(`select count(*) from pipelines where rkey = 'p2'`).Scan(&n); err != nil || n != 1 { + t.Fatalf("expected other pipeline to survive, got %d err %v", n, err) } } diff --git a/spindle/db/events.go b/spindle/db/events.go index c89b62208..cd38c2053 100644 --- a/spindle/db/events.go +++ b/spindle/db/events.go @@ -4,61 +4,63 @@ import ( "context" "database/sql" "encoding/json" - "fmt" + "time" + "tangled.org/core/api/tangled" - "tangled.org/core/eventstream" "tangled.org/core/notifier" "tangled.org/core/spindle/models" - "tangled.org/core/tid" - "time" ) -func (d *DB) insertEvent(event eventstream.Event, n *notifier.Notifier) error { - return eventstream.Insert(d, event, n) +// StatusRow is one append-only workflow status transition. Its id is the cursor +// the mill executor's observe loop follows. +type StatusRow struct { + Id int64 + Pipeline string // pipeline id + Workflow string // workflow name + Status string + Error *string + ExitCode *int64 } -func (d *DB) GetEvents(cursor int64, limit int) ([]eventstream.Event, error) { - return eventstream.List(d, cursor, limit) +func insertStatusTx(tx DBTX, wid models.WorkflowId, kind models.StatusKind, workflowError *string, exitCode *int64) error { + _, err := tx.Exec( + `insert into workflow_statuses (rkey, workflow, status, error, exit_code, created_at) + values (?, ?, ?, ?, ?, ?)`, + wid.PipelineId.Rkey, wid.Name, string(kind), workflowError, exitCode, + time.Now().Format(time.RFC3339Nano), + ) + return err } func (d *DB) EventHighWater() (int64, error) { - return eventstream.HighWater(d) + var id int64 + err := d.QueryRow(`select coalesce(max(id), 0) from workflow_statuses`).Scan(&id) + return id, err } -func (d *DB) CreatePipelineEvent(rkey string, pipeline tangled.Pipeline, n *notifier.Notifier) error { - eventJson, err := json.Marshal(pipeline) +func (d *DB) GetEvents(cursor int64, limit int) ([]StatusRow, error) { + rows, err := d.Query( + `select id, rkey, workflow, status, error, exit_code + from workflow_statuses + where id > ? + order by id asc + limit ?`, + cursor, limit, + ) if err != nil { - return err - } - event := eventstream.Event{ - Rkey: rkey, - Nsid: tangled.PipelineNSID, - EventJson: eventJson, - } - return d.insertEvent(event, n) -} - -// the envelope Created stays zero so insertEvent stamps the local clock, the record's CreatedAt is separate -func statusEvent(pipelineAtUri, workflow, status string, workflowError *string, exitCode *int64) (eventstream.Event, error) { - s := tangled.PipelineStatus{ - CreatedAt: time.Now().Format(time.RFC3339), - Error: workflowError, - ExitCode: exitCode, - Pipeline: pipelineAtUri, - Workflow: workflow, - Status: status, + return nil, err } + defer rows.Close() - eventJson, err := json.Marshal(s) - if err != nil { - return eventstream.Event{}, err + var out []StatusRow + for rows.Next() { + var r StatusRow + if err := rows.Scan(&r.Id, &r.Pipeline, &r.Workflow, &r.Status, &r.Error, &r.ExitCode); err != nil { + return nil, err + } + out = append(out, r) } - - return eventstream.Event{ - Rkey: tid.TID(), - Nsid: tangled.PipelineStatusNSID, - EventJson: eventJson, - }, nil + return out, rows.Err() } func (d *DB) createStatusEvent( @@ -68,113 +70,110 @@ func (d *DB) createStatusEvent( exitCode *int64, n *notifier.Notifier, ) error { - event, err := statusEvent(string(workflowId.PipelineId.AtUri()), workflowId.Name, string(statusKind), workflowError, exitCode) - if err != nil { + if err := insertStatusTx(d, workflowId, statusKind, workflowError, exitCode); err != nil { return err } - return d.insertEvent(event, n) + n.NotifyAll() + return nil } // deleting the lease in the same transaction prevents the terminal event // from replaying func (d *DB) CompleteMillLease( leaseID string, - pipelineAtUri string, - workflow string, + workflowId models.WorkflowId, status string, workflowError *string, exitCode *int64, n *notifier.Notifier, ) error { return d.ApplyEventBatch(n, func(tx *EventBatchTx) error { - if err := tx.InsertStatusEvent(pipelineAtUri, workflow, status, workflowError, exitCode); err != nil { + if err := tx.InsertStatusEvent(workflowId, status, workflowError, exitCode); err != nil { return err } return tx.DeleteLease(leaseID) }) } -func (d *DB) GetStatus(workflowId models.WorkflowId) (*tangled.PipelineStatus, error) { - pipelineAtUri := workflowId.PipelineId.AtUri() +func (d *DB) GetStatus(workflowId models.WorkflowId) (models.StatusKind, error) { + pipelineId := workflowId.PipelineId.Rkey - var eventJson string + var status string err := d.QueryRow( ` select - event from events + status from workflow_statuses where - nsid = ? - and json_extract(event, '$.pipeline') = ? - and json_extract(event, '$.workflow') = ? + rkey = ? + and workflow = ? order by - created desc + id desc limit 1 `, - tangled.PipelineStatusNSID, - string(pipelineAtUri), + pipelineId, workflowId.Name, - ).Scan(&eventJson) - + ).Scan(&status) if err != nil { - return nil, err - } - - var status tangled.PipelineStatus - if err := json.Unmarshal([]byte(eventJson), &status); err != nil { - return nil, err + if err == sql.ErrNoRows { + return "", nil + } + return "", err } - return &status, nil + return models.StatusKind(status), nil } type statusQueryer interface { QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row } -func workflowStartupDelay(ctx context.Context, q statusQueryer, pipelineAtURI, workflow string) (time.Duration, bool, error) { - var pending, running sql.NullInt64 +func workflowStartupDelay(ctx context.Context, q statusQueryer, wid models.WorkflowId) (time.Duration, bool, error) { + var pending, running sql.NullString err := q.QueryRowContext(ctx, ` select - min(case when json_extract(event, '$.status') = 'pending' then created end), - min(case when json_extract(event, '$.status') = 'running' then created end) - from events - where nsid = 'sh.tangled.pipeline.status' - and json_extract(event, '$.pipeline') = ? - and json_extract(event, '$.workflow') = ? - `, pipelineAtURI, workflow).Scan(&pending, &running) + min(case when status = 'pending' then created_at end), + min(case when status = 'running' then created_at end) + from workflow_statuses + where rkey = ? and workflow = ? + `, wid.PipelineId.Rkey, wid.Name).Scan(&pending, &running) if err != nil { return 0, false, err } if !pending.Valid || !running.Valid { return 0, false, nil } - delay := time.Duration(running.Int64 - pending.Int64) + pendingAt, err := time.Parse(time.RFC3339Nano, pending.String) + if err != nil { + return 0, false, err + } + runningAt, err := time.Parse(time.RFC3339Nano, running.String) + if err != nil { + return 0, false, err + } + delay := runningAt.Sub(pendingAt) if delay < 0 { delay = 0 } return delay, true, nil } -func (d *DB) WorkflowStartupDelay(ctx context.Context, workflowID models.WorkflowId) (time.Duration, bool, error) { - return workflowStartupDelay(ctx, d, string(workflowID.PipelineId.AtUri()), workflowID.Name) +func (d *DB) WorkflowStartupDelay(ctx context.Context, wid models.WorkflowId) (time.Duration, bool, error) { + return workflowStartupDelay(ctx, d, wid) } -func (tx *EventBatchTx) WorkflowStartupDelay(ctx context.Context, pipelineAtURI, workflow string) (time.Duration, bool, error) { - return workflowStartupDelay(ctx, tx.tx, pipelineAtURI, workflow) +func (tx *EventBatchTx) WorkflowStartupDelay(ctx context.Context, wid models.WorkflowId) (time.Duration, bool, error) { + return workflowStartupDelay(ctx, tx.tx, wid) } -func (tx *EventBatchTx) HasWorkflowStatus(ctx context.Context, pipelineAtURI, workflow, status string) (bool, error) { +func (tx *EventBatchTx) HasWorkflowStatus(ctx context.Context, wid models.WorkflowId, status string) (bool, error) { var present bool err := tx.tx.QueryRowContext(ctx, ` select exists( - select 1 from events - where nsid = 'sh.tangled.pipeline.status' - and json_extract(event, '$.pipeline') = ? - and json_extract(event, '$.workflow') = ? - and json_extract(event, '$.status') = ? + select 1 from workflow_statuses + where rkey = ? and workflow = ? and status = ? ) - `, pipelineAtURI, workflow, status).Scan(&present) + `, wid.PipelineId.Rkey, wid.Name, status).Scan(&present) return present, err } @@ -209,13 +208,7 @@ type PipelineWorkflow struct { } func (d *DB) ListPipelineWorkflows(repoDid string) ([]PipelineWorkflow, error) { - rows, err := d.Query( - `select rkey, event from events - where nsid = 'sh.tangled.pipeline' - and coalesce(json_extract(event, '$.triggerMetadata.repo.repoDid'), - json_extract(event, '$.triggerMetadata.repo.did')) = ?`, - repoDid, - ) + rows, err := d.Query(`select rkey, knot, payload from pipelines where repo_did = ?`, repoDid) if err != nil { return nil, err } @@ -223,20 +216,18 @@ func (d *DB) ListPipelineWorkflows(repoDid string) ([]PipelineWorkflow, error) { var out []PipelineWorkflow for rows.Next() { - var rkey, raw string - if err := rows.Scan(&rkey, &raw); err != nil { + var rkey, knot, raw string + if err := rows.Scan(&rkey, &knot, &raw); err != nil { return nil, err } - var p tangled.Pipeline + var p tangled.CiPipeline if err := json.Unmarshal([]byte(raw), &p); err != nil { continue } - knot := "" - if p.TriggerMetadata != nil && p.TriggerMetadata.Repo != nil { - knot = p.TriggerMetadata.Repo.Knot - } for _, wf := range p.Workflows { - out = append(out, PipelineWorkflow{Knot: knot, Rkey: rkey, Name: wf.Name}) + if wf != nil { + out = append(out, PipelineWorkflow{Knot: knot, Rkey: rkey, Name: wf.Name}) + } } } return out, rows.Err() @@ -255,58 +246,41 @@ func (d *DB) DeleteEventsByRepo(repoDid string, extra []PipelineKey) error { } defer tx.Rollback() - // use knots recorded in pipeline events, not the repo's current knots - rows, err := tx.Query( - `select coalesce(json_extract(event, '$.triggerMetadata.repo.knot'), ''), rkey from events - where nsid = 'sh.tangled.pipeline' - and coalesce(json_extract(event, '$.triggerMetadata.repo.repoDid'), - json_extract(event, '$.triggerMetadata.repo.did')) = ?`, - repoDid, - ) + rows, err := tx.Query(`select rkey from pipelines where repo_did = ?`, repoDid) if err != nil { return err } - seen := map[PipelineKey]bool{} - var keys []PipelineKey - add := func(knot, rkey string) { - k := PipelineKey{Knot: knot, Rkey: rkey} - if k.Knot != "" && k.Rkey != "" && !seen[k] { - seen[k] = true - keys = append(keys, k) + seen := make(map[string]bool) + var rkeys []string + add := func(rkey string) { + if rkey != "" && !seen[rkey] { + seen[rkey] = true + rkeys = append(rkeys, rkey) } } for rows.Next() { - var knot, rkey string - if err := rows.Scan(&knot, &rkey); err != nil { + var rkey string + if err := rows.Scan(&rkey); err != nil { rows.Close() return err } - add(knot, rkey) + add(rkey) } if err := rows.Err(); err != nil { rows.Close() return err } rows.Close() - for _, k := range extra { - add(k.Knot, k.Rkey) + for _, key := range extra { + add(key.Rkey) } - for _, k := range keys { - if _, err := tx.Exec( - `delete from events where nsid = 'sh.tangled.pipeline.status' and json_extract(event, '$.pipeline') = ?`, - fmt.Sprintf("at://did:web:%s/sh.tangled.pipeline/%s", k.Knot, k.Rkey), - ); err != nil { + for _, rkey := range rkeys { + if _, err := tx.Exec(`delete from workflow_statuses where rkey = ?`, rkey); err != nil { return err } } - if _, err := tx.Exec( - `delete from events - where nsid = 'sh.tangled.pipeline' - and coalesce(json_extract(event, '$.triggerMetadata.repo.repoDid'), - json_extract(event, '$.triggerMetadata.repo.did')) = ?`, - repoDid, - ); err != nil { + if _, err := tx.Exec(`delete from pipelines where repo_did = ?`, repoDid); err != nil { return err } return tx.Commit() diff --git a/spindle/db/events_metrics_test.go b/spindle/db/events_metrics_test.go index 4df51445e..e73c0aa92 100644 --- a/spindle/db/events_metrics_test.go +++ b/spindle/db/events_metrics_test.go @@ -26,8 +26,8 @@ func TestWorkflowStartupDelay(t *testing.T) { if err := database.StatusPending(wid, &n); err != nil { t.Fatal(err) } - pendingAt := time.Now().Add(-5 * time.Second).UnixNano() - if _, err := database.Exec(`update events set created = ? where json_extract(event, '$.status') = 'pending'`, pendingAt); err != nil { + pendingAt := time.Now().Add(-5 * time.Second).Format(time.RFC3339Nano) + if _, err := database.Exec(`update workflow_statuses set created_at = ? where status = 'pending'`, pendingAt); err != nil { t.Fatal(err) } if err := database.StatusRunning(wid, &n); err != nil { @@ -56,13 +56,11 @@ func TestWorkflowStartupDelayUsesStatusIndex(t *testing.T) { rows, err := database.Query(` explain query plan select - min(case when json_extract(event, '$.status') = 'pending' then created end), - min(case when json_extract(event, '$.status') = 'running' then created end) - from events - where nsid = 'sh.tangled.pipeline.status' - and json_extract(event, '$.pipeline') = ? - and json_extract(event, '$.workflow') = ? - `, "at://knot.example.com/sh.tangled.pipeline/pipeline", "build") + min(case when status = 'pending' then created_at end), + min(case when status = 'running' then created_at end) + from workflow_statuses + where rkey = ? and workflow = ? + `, "pipeline", "build") if err != nil { t.Fatal(err) } @@ -75,7 +73,7 @@ func TestWorkflowStartupDelayUsesStatusIndex(t *testing.T) { if err := rows.Scan(&id, &parent, &unused, &detail); err != nil { t.Fatal(err) } - if strings.Contains(detail, "idx_events_pipeline_status") { + if strings.Contains(detail, "idx_workflow_statuses_lookup") { usedIndex = true } } @@ -83,6 +81,6 @@ func TestWorkflowStartupDelayUsesStatusIndex(t *testing.T) { t.Fatal(err) } if !usedIndex { - t.Fatal("workflow startup query did not use idx_events_pipeline_status") + t.Fatal("workflow startup query did not use idx_workflow_statuses_lookup") } } diff --git a/spindle/db/jobs.go b/spindle/db/jobs.go index fa54132fd..9b627dc6e 100644 --- a/spindle/db/jobs.go +++ b/spindle/db/jobs.go @@ -8,7 +8,6 @@ import ( "time" "tangled.org/core/api/tangled" - "tangled.org/core/eventstream" "tangled.org/core/notifier" "tangled.org/core/spindle/models" ) @@ -66,7 +65,7 @@ func (d *DB) enqueueJob( return err } - enqueue := func(insert eventstream.Inserter) error { + enqueue := func() error { tx, err := d.BeginTx(ctx, nil) if err != nil { return err @@ -110,17 +109,8 @@ func (d *DB) enqueueJob( if workflow == nil { continue } - event, err := statusEvent( - string(pipelineID.AtUri()), - workflow.Name, - string(models.StatusKindPending), - nil, - nil, - ) - if err != nil { - return err - } - if err := insert(tx, event); err != nil { + wid := models.WorkflowId{PipelineId: pipelineID, Name: workflow.Name} + if err := insertStatusTx(tx, wid, models.StatusKindPending, nil, nil); err != nil { return err } } @@ -128,11 +118,7 @@ func (d *DB) enqueueJob( return tx.Commit() } - if markPending { - err = eventstream.WithClock(enqueue) - } else { - err = enqueue(nil) - } + err = enqueue() if err != nil { return err } diff --git a/spindle/db/jobs_test.go b/spindle/db/jobs_test.go index a0775dda9..e05a03276 100644 --- a/spindle/db/jobs_test.go +++ b/spindle/db/jobs_test.go @@ -78,8 +78,8 @@ func TestEnqueueJobWithPendingCommitsJobAndStatusesTogether(t *testing.T) { if err != nil { t.Fatal(err) } - if status.Status != string(models.StatusKindPending) { - t.Fatalf("status = %q, want pending", status.Status) + if status != models.StatusKindPending { + t.Fatalf("status = %q, want pending", status) } job, err := database.DequeueJob(ctx) if err != nil { diff --git a/spindle/db/mill_state.go b/spindle/db/mill_state.go index 7009a5376..9d8e8cd54 100644 --- a/spindle/db/mill_state.go +++ b/spindle/db/mill_state.go @@ -4,7 +4,6 @@ import ( "database/sql" "fmt" - "tangled.org/core/eventstream" "tangled.org/core/notifier" "tangled.org/core/spindle/models" ) @@ -263,17 +262,12 @@ func (d *DB) GetOutboxState() (string, uint64, error) { } type EventBatchTx struct { - tx *sql.Tx - db *DB - insert eventstream.Inserter + tx *sql.Tx + db *DB } -func (tx *EventBatchTx) InsertStatusEvent(pipelineAtUri, workflow, status string, workflowError *string, exitCode *int64) error { - event, err := statusEvent(pipelineAtUri, workflow, status, workflowError, exitCode) - if err != nil { - return err - } - return tx.insert(tx.tx, event) +func (tx *EventBatchTx) InsertStatusEvent(wid models.WorkflowId, status string, workflowError *string, exitCode *int64) error { + return insertStatusTx(tx.tx, wid, models.StatusKind(status), workflowError, exitCode) } func (tx *EventBatchTx) DeleteLease(leaseID string) error { @@ -419,26 +413,18 @@ func (d *DB) ListPendingArtifacts() ([]PendingArtifact, error) { } func (d *DB) ApplyEventBatch(n *notifier.Notifier, fn func(tx *EventBatchTx) error) error { - err := eventstream.WithClock(func(insert eventstream.Inserter) error { - tx, err := d.Begin() - if err != nil { - return err - } - defer tx.Rollback() - - batchTx := &EventBatchTx{ - tx: tx, - db: d, - insert: insert, - } - if err := fn(batchTx); err != nil { - return err - } - return tx.Commit() - }) + tx, err := d.Begin() if err != nil { return err } + defer tx.Rollback() + + if err := fn(&EventBatchTx{tx: tx, db: d}); err != nil { + return err + } + if err := tx.Commit(); err != nil { + return err + } if n != nil { n.NotifyAll() } diff --git a/spindle/db/mill_state_test.go b/spindle/db/mill_state_test.go index 39adb1ddd..5884ed7eb 100644 --- a/spindle/db/mill_state_test.go +++ b/spindle/db/mill_state_test.go @@ -112,8 +112,7 @@ func TestCompleteMillLeaseIsAtomic(t *testing.T) { defer n.Unsubscribe(notifications) err := d.CompleteMillLease( "lease-1", - "at://knot.example/sh.tangled.pipeline/rkey1", - "build", + models.WorkflowId{PipelineId: models.PipelineId{Rkey: "rkey1"}, Name: "build"}, "failed", nil, nil, @@ -123,11 +122,11 @@ func TestCompleteMillLeaseIsAtomic(t *testing.T) { t.Fatal("CompleteMillLease succeeded despite forced lease deletion failure") } var eventCount int - if err := d.QueryRow(`select count(*) from events`).Scan(&eventCount); err != nil { + if err := d.QueryRow(`select count(*) from workflow_statuses`).Scan(&eventCount); err != nil { t.Fatalf("count events after rollback: %v", err) } if eventCount != 0 { - t.Fatalf("terminal event count after rollback = %d, want 0", eventCount) + t.Fatalf("terminal status count after rollback = %d, want 0", eventCount) } if leases, listErr := d.ListMillLeases(); listErr != nil || len(leases) != 1 { t.Fatalf("leases after rollback = %+v, err = %v; want original lease", leases, listErr) @@ -143,8 +142,7 @@ func TestCompleteMillLeaseIsAtomic(t *testing.T) { } if err := d.CompleteMillLease( "lease-1", - "at://knot.example/sh.tangled.pipeline/rkey1", - "build", + models.WorkflowId{PipelineId: models.PipelineId{Rkey: "rkey1"}, Name: "build"}, "failed", nil, nil, @@ -152,11 +150,11 @@ func TestCompleteMillLeaseIsAtomic(t *testing.T) { ); err != nil { t.Fatalf("CompleteMillLease retry: %v", err) } - if err := d.QueryRow(`select count(*) from events`).Scan(&eventCount); err != nil { + if err := d.QueryRow(`select count(*) from workflow_statuses`).Scan(&eventCount); err != nil { t.Fatalf("count committed events: %v", err) } if eventCount != 1 { - t.Fatalf("terminal event count after commit = %d, want 1", eventCount) + t.Fatalf("terminal status count after commit = %d, want 1", eventCount) } if leases, listErr := d.ListMillLeases(); listErr != nil || len(leases) != 0 { t.Fatalf("leases after commit = %+v, err = %v; want none", leases, listErr) diff --git a/spindle/db/pipelines.go b/spindle/db/pipelines.go index d93208fd9..7ceb5f6ce 100644 --- a/spindle/db/pipelines.go +++ b/spindle/db/pipelines.go @@ -8,6 +8,7 @@ import ( "time" "tangled.org/core/api/tangled" + "tangled.org/core/orm" "tangled.org/core/spindle/models" "tangled.org/core/workflow" ) @@ -17,126 +18,120 @@ func (d *DB) QueryPipelines(ctx context.Context, repoDid string, commits []strin limit = 30 } - var query string - var args []any - query = ` - select - rkey, event, created from events - where - nsid = 'sh.tangled.pipeline' - and coalesce(json_extract(event, '$.triggerMetadata.repo.repoDid'), json_extract(event, '$.triggerMetadata.repo.did')) = ? - ` - args = append(args, repoDid) - + filters := []orm.Filter{orm.FilterEq("repo_did", repoDid)} + // only filter when asked: FilterIn compiles an empty slice to `1 = 0` if len(commits) > 0 { - placeholders := make([]string, len(commits)) - for i := range commits { - placeholders[i] = "?" - args = append(args, commits[i]) - } - query += ` and coalesce( - json_extract(event, '$.triggerMetadata.push.newSha'), - json_extract(event, '$.triggerMetadata.pullRequest.sourceSha'), - json_extract(event, '$.triggerMetadata.manual.sha') - ) in (` + strings.Join(placeholders, ",") + ")" + filters = append(filters, orm.FilterIn("commit_sha", commits)) } - if len(kinds) > 0 { - placeholders := make([]string, len(kinds)) - for i := range kinds { - placeholders[i] = "?" - args = append(args, kinds[i]) - } - query += " and json_extract(event, '$.triggerMetadata.kind') in (" + strings.Join(placeholders, ",") + ")" + filters = append(filters, orm.FilterIn("kind", kinds)) } - if cursor != "" { - if cVal, err := strconv.ParseInt(cursor, 10, 64); err == nil { - query += " and created < ?" - args = append(args, cVal) - } + var conditions []string + var args []any + for _, filter := range filters { + conditions = append(conditions, filter.Condition()) + args = append(args, filter.Arg()...) } + whereClause := " where " + strings.Join(conditions, " and ") - // First get total count var total int64 - countQuery := "select count(*) from (" + query + ")" - if err := d.QueryRowContext(ctx, countQuery, args...).Scan(&total); err != nil { + if err := d.QueryRowContext(ctx, + `select count(*) from pipelines`+whereClause, args...).Scan(&total); err != nil { return nil, "", 0, err } - query += " order by created desc limit ?" + // the cursor bounds the page but not the total, so it joins after the count + if cursor != "" { + if cVal, err := strconv.ParseInt(cursor, 10, 64); err == nil { + filter := orm.FilterLt("id", cVal) + conditions = append(conditions, filter.Condition()) + args = append(args, filter.Arg()...) + whereClause = " where " + strings.Join(conditions, " and ") + } + } args = append(args, limit) - rows, err := d.QueryContext(ctx, query, args...) + rows, err := d.QueryContext(ctx, + `select id, payload from pipelines`+whereClause+` order by id desc limit ?`, args...) if err != nil { return nil, "", 0, err } defer rows.Close() var pipelines []*tangled.CiPipeline - var lastCreated int64 + var lastId int64 for rows.Next() { - var rkey, eventJson string - var created int64 - if err := rows.Scan(&rkey, &eventJson, &created); err != nil { + var id int64 + var payload string + if err := rows.Scan(&id, &payload); err != nil { return nil, "", 0, err } - lastCreated = created - - var rawPipeline tangled.Pipeline - if err := json.Unmarshal([]byte(eventJson), &rawPipeline); err != nil { + var p tangled.CiPipeline + if err := json.Unmarshal([]byte(payload), &p); err != nil { continue } + lastId = id + pipelines = append(pipelines, &p) + } + if err := rows.Err(); err != nil { + return nil, "", 0, err + } - p, err := d.mapToCiPipeline(rkey, created, rawPipeline) - if err != nil { - return nil, "", 0, err - } - pipelines = append(pipelines, p) + if err := d.applyStatuses(ctx, pipelines); err != nil { + return nil, "", 0, err } nextCursor := "" if len(pipelines) == limit { - nextCursor = strconv.FormatInt(lastCreated, 10) + nextCursor = strconv.FormatInt(lastId, 10) } return pipelines, nextCursor, total, nil } -func (d *DB) GetPipeline(ctx context.Context, rkey string) (*tangled.CiPipeline, error) { - var eventJson string - var created int64 - err := d.QueryRowContext(ctx, - ` - select - event, created from events - where - nsid = 'sh.tangled.pipeline' - and rkey = ? - `, - rkey, - ).Scan(&eventJson, &created) +// GetPipelineWithKnot also returns the knot the pipeline was created on, which +// subscribePipelineLogs needs to locate the workflow log file. +func (d *DB) GetPipelineWithKnot(ctx context.Context, rkey string) (*tangled.CiPipeline, string, error) { + var payload, knot string + if err := d.QueryRowContext(ctx, + `select payload, knot from pipelines where rkey = ?`, rkey, + ).Scan(&payload, &knot); err != nil { + return nil, "", err + } - if err != nil { - return nil, err + var p tangled.CiPipeline + if err := json.Unmarshal([]byte(payload), &p); err != nil { + return nil, "", err } - var rawPipeline tangled.Pipeline - if err := json.Unmarshal([]byte(eventJson), &rawPipeline); err != nil { - return nil, err + if err := d.applyStatuses(ctx, []*tangled.CiPipeline{&p}); err != nil { + return nil, "", err } + return &p, knot, nil +} - return d.mapToCiPipeline(rkey, created, rawPipeline) +func (d *DB) GetPipeline(ctx context.Context, rkey string) (*tangled.CiPipeline, error) { + p, _, err := d.GetPipelineWithKnot(ctx, rkey) + return p, err } -func (d *DB) mapToCiPipeline(rkey string, created int64, raw tangled.Pipeline) (*tangled.CiPipeline, error) { - createdAtStr := time.Unix(0, created).Format(time.RFC3339) +// mapToCiPipeline converts a compiled legacy pipeline into the sh.tangled.ci.pipeline +// payload we store, plus the columns we index on. workflow statuses in the returned +// payload are always "pending": the read path overwrites them from workflow_statuses, +// so the payload never has to be rewritten as a pipeline progresses. +// +// kind is taken straight from raw.TriggerMetadata.Kind, which already holds the +// workflow.TriggerKind vocabulary the queryPipelines `kinds` param uses. Producing it +// in the same switch as the trigger union keeps the two from disagreeing. +func mapToCiPipeline(rkey string, createdAt time.Time, raw tangled.Pipeline) (*tangled.CiPipeline, workflow.TriggerKind) { + createdAtStr := createdAt.Format(time.RFC3339) var repoDidStr string if raw.TriggerMetadata != nil && raw.TriggerMetadata.Repo != nil { - if raw.TriggerMetadata.Repo.RepoDid != nil { - repoDidStr = *raw.TriggerMetadata.Repo.RepoDid + if rd := raw.TriggerMetadata.Repo.RepoDid; rd != nil && *rd != "" { + repoDidStr = *rd } else { repoDidStr = raw.TriggerMetadata.Repo.Did } @@ -144,9 +139,11 @@ func (d *DB) mapToCiPipeline(rkey string, created int64, raw tangled.Pipeline) ( commitSha := "" var trigger tangled.CiPipeline_Trigger + var kind workflow.TriggerKind if raw.TriggerMetadata != nil { - switch workflow.TriggerKind(raw.TriggerMetadata.Kind) { + kind = workflow.TriggerKind(raw.TriggerMetadata.Kind) + switch kind { case workflow.TriggerKindPush: if raw.TriggerMetadata.Push != nil { commitSha = raw.TriggerMetadata.Push.NewSha @@ -181,35 +178,20 @@ func (d *DB) mapToCiPipeline(rkey string, created int64, raw tangled.Pipeline) ( } } - var workflows []*tangled.CiPipeline_Workflow + workflows := make([]*tangled.CiPipeline_Workflow, 0, len(raw.Workflows)) for _, wf := range raw.Workflows { - status := "pending" - var startedAt, finishedAt, wfError *string - - if raw.TriggerMetadata != nil && raw.TriggerMetadata.Repo != nil { - wfId := models.WorkflowId{ - PipelineId: models.PipelineId{ - Knot: raw.TriggerMetadata.Repo.Knot, - Rkey: rkey, - }, - Name: wf.Name, - } - - wfStatus, err := d.GetStatus(wfId) - if err == nil && wfStatus != nil { - status = wfStatus.Status - startedAt, finishedAt = d.GetWorkflowTimes(wfId) - wfError = wfStatus.Error - } + if wf == nil { + continue } + // NOTE: workflow statuses will be filled from caller workflows = append(workflows, &tangled.CiPipeline_Workflow{ Id: wf.Name, Name: wf.Name, - Status: status, - StartedAt: startedAt, - FinishedAt: finishedAt, - Error: wfError, + Status: string(models.StatusKindPending), + StartedAt: nil, + FinishedAt: nil, + Error: nil, }) } @@ -226,7 +208,7 @@ func (d *DB) mapToCiPipeline(rkey string, created int64, raw tangled.Pipeline) ( Trigger: &trigger, Workflows: workflows, SourceRepo: sourceRepo, - }, nil + }, kind } func pipelinePairsToCiTriggerPairs(inputs []*tangled.Pipeline_Pair) []*tangled.CiTrigger_Pair { @@ -246,24 +228,109 @@ func pipelinePairsToCiTriggerPairs(inputs []*tangled.Pipeline_Pair) []*tangled.C return pairs } -func (d *DB) GetWorkflowTimes(workflowId models.WorkflowId) (startedAt, finishedAt *string) { - pipelineAtUri := workflowId.PipelineId.AtUri() - - _ = d.QueryRow( - ` - select - min(case when json_extract(event, '$.status') = 'running' then json_extract(event, '$.createdAt') end), - max(case when json_extract(event, '$.status') in ('success', 'failed', 'timeout', 'cancelled') then json_extract(event, '$.createdAt') end) - from events - where - nsid = ? - and json_extract(event, '$.pipeline') = ? - and json_extract(event, '$.workflow') = ? - `, - tangled.PipelineStatusNSID, - string(pipelineAtUri), - workflowId.Name, - ).Scan(&startedAt, &finishedAt) - - return +// knot is stored only because the workflow log path still embeds it; it is not +// part of pipeline identity and nothing resolves or dials it. +func (d *DB) CreatePipeline(id models.PipelineId, raw tangled.Pipeline) error { + p, kind := mapToCiPipeline(id.Rkey, time.Now(), raw) + payload, err := json.Marshal(p) + if err != nil { + return err + } + _, err = d.Exec( + `insert into pipelines (rkey, knot, repo_did, commit_sha, kind, payload) values (?, ?, ?, ?, ?, ?)`, + id.Rkey, id.Knot, p.Repo, p.Commit, string(kind), string(payload), + ) + return err +} + +// applyStatuses overlays live workflow status onto stored payloads with a single +// query for the whole page, replacing the old per-workflow GetStatus + +// GetWorkflowTimes fan-out. CiPipeline.Id is the rkey, so no parallel slice. +func (d *DB) applyStatuses(ctx context.Context, pipelines []*tangled.CiPipeline) error { + if len(pipelines) == 0 { + return nil + } + + rkeys := make([]string, 0, len(pipelines)) + for _, p := range pipelines { + rkeys = append(rkeys, p.Id) + } + + statuses, err := d.workflowStatuses(ctx, rkeys) + if err != nil { + return err + } + + for _, p := range pipelines { + for _, wf := range p.Workflows { + if wf == nil { + continue + } + st, ok := statuses[wfKey{p.Id, wf.Name}] + if !ok { + continue + } + wf.Status = st.Status + wf.Error = st.Error + wf.StartedAt = st.StartedAt + wf.FinishedAt = st.FinishedAt + } + } + return nil +} + +type wfKey struct { + Rkey string + Workflow string +} + +type wfStatus struct { + Status string + Error *string + StartedAt *string + FinishedAt *string +} + +func (d *DB) workflowStatuses(ctx context.Context, rkeys []string) (map[wfKey]wfStatus, error) { + filter := orm.FilterIn("rkey", rkeys) + rows, err := d.QueryContext(ctx, + `select rkey, workflow, status, error, created_at + from workflow_statuses + where `+filter.Condition()+` + order by id asc`, filter.Arg()...) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make(map[wfKey]wfStatus) + for rows.Next() { + var rkey, wfName, status, createdAt string + var wfError *string + if err := rows.Scan(&rkey, &wfName, &status, &wfError, &createdAt); err != nil { + return nil, err + } + + k := wfKey{rkey, wfName} + st := out[k] + + // rows arrive in insertion order, so the last one wins for the current status + st.Status = status + st.Error = wfError + + switch kind := models.StatusKind(status); { + case kind == models.StatusKindRunning: + // the first running row is when the workflow actually started + if st.StartedAt == nil { + at := createdAt + st.StartedAt = &at + } + case kind.IsFinish(): + at := createdAt + st.FinishedAt = &at + } + + out[k] = st + } + return out, rows.Err() } diff --git a/spindle/db/pipelines_test.go b/spindle/db/pipelines_test.go index 640ecf382..d2fdebb25 100644 --- a/spindle/db/pipelines_test.go +++ b/spindle/db/pipelines_test.go @@ -8,9 +8,12 @@ import ( "time" "tangled.org/core/api/tangled" + "tangled.org/core/notifier" + "tangled.org/core/spindle/models" + "tangled.org/core/workflow" ) -func seedPipelineEvent(t *testing.T, d *DB, rkey, repoDid, kind string, created int64) { +func seedPipeline(t *testing.T, d *DB, rkey, repoDid, kind string) { t.Helper() repo := repoDid tm := &tangled.Pipeline_TriggerMetadata{ @@ -29,15 +32,8 @@ func seedPipelineEvent(t *testing.T, d *DB, rkey, repoDid, kind string, created TriggerMetadata: tm, Workflows: []*tangled.Pipeline_Workflow{{Name: "ci.yml"}}, } - eventJson, err := json.Marshal(raw) - if err != nil { - t.Fatalf("marshal pipeline: %v", err) - } - if _, err := d.Exec( - `insert into events (rkey, nsid, event, created) values (?, 'sh.tangled.pipeline', ?, ?)`, - rkey, string(eventJson), created, - ); err != nil { - t.Fatalf("seed event %s: %v", rkey, err) + if err := d.CreatePipeline(models.PipelineId{Knot: "knot.test", Rkey: rkey}, raw); err != nil { + t.Fatalf("seed pipeline %s: %v", rkey, err) } } @@ -45,11 +41,10 @@ func TestQueryPipelines_FilterByKind(t *testing.T) { d := newTestDB(t) ctx := context.Background() repo := "did:plc:boltless" - base := time.Now().UnixNano() - seedPipelineEvent(t, d, "p-push", repo, "push", base+1) - seedPipelineEvent(t, d, "p-pull", repo, "pull_request", base+2) - seedPipelineEvent(t, d, "p-manual", repo, "manual", base+3) + seedPipeline(t, d, "p-push", repo, "push") + seedPipeline(t, d, "p-pull", repo, "pull_request") + seedPipeline(t, d, "p-manual", repo, "manual") cases := []struct { kinds []string @@ -85,10 +80,9 @@ func TestQueryPipelines_FilterByKind(t *testing.T) { func TestQueryPipelines_KindScopedToRepo(t *testing.T) { d := newTestDB(t) ctx := context.Background() - base := time.Now().UnixNano() - seedPipelineEvent(t, d, "a-push", "did:plc:alice", "push", base+1) - seedPipelineEvent(t, d, "b-push", "did:plc:bob", "push", base+2) + seedPipeline(t, d, "a-push", "did:plc:alice", "push") + seedPipeline(t, d, "b-push", "did:plc:bob", "push") pipelines, _, total, err := d.QueryPipelines(ctx, "did:plc:alice", nil, "", []string{"push"}, 30) if err != nil { @@ -98,7 +92,8 @@ func TestQueryPipelines_KindScopedToRepo(t *testing.T) { t.Fatalf("total=%d len=%d, want exactly alice's single push pipeline", total, len(pipelines)) } if pipelines[0].Repo != "did:plc:alice" { - t.Errorf("returned pipeline repo = %v, want did:plc:alice", pipelines[0].Repo) + t.Errorf("returned pipeline repo = %q, want did:plc:alice", pipelines[0].Repo) + } } @@ -116,3 +111,148 @@ func triggerKindOf(p *tangled.CiPipeline) string { } return "" } + +func TestQueryPipelines_WorkflowStatuses(t *testing.T) { + d := newTestDB(t) + ctx := context.Background() + repo := "did:plc:boltless" + n := notifier.New() + + raw := tangled.Pipeline{ + TriggerMetadata: &tangled.Pipeline_TriggerMetadata{ + Kind: "push", + Repo: &tangled.Pipeline_TriggerRepo{Knot: "knot.test", RepoDid: &repo, Did: repo}, + Push: &tangled.Pipeline_PushTriggerData{NewSha: "sha1", Ref: "refs/heads/main"}, + }, + Workflows: []*tangled.Pipeline_Workflow{{Name: "a"}, {Name: "b"}}, + } + if err := d.CreatePipeline(models.PipelineId{Knot: "knot.test", Rkey: "pl1"}, raw); err != nil { + t.Fatalf("CreatePipeline: %v", err) + } + + widA := models.WorkflowId{PipelineId: models.PipelineId{Rkey: "pl1"}, Name: "a"} + widB := models.WorkflowId{PipelineId: models.PipelineId{Rkey: "pl1"}, Name: "b"} + + for _, step := range []func() error{ + func() error { return d.StatusPending(widA, &n) }, + func() error { return d.StatusRunning(widA, &n) }, + func() error { return d.StatusSuccess(widA, &n) }, + func() error { return d.StatusPending(widB, &n) }, + } { + if err := step(); err != nil { + t.Fatalf("seed status: %v", err) + } + } + + pipelines, _, total, err := d.QueryPipelines(ctx, repo, nil, "", nil, 30) + if err != nil { + t.Fatalf("QueryPipelines: %v", err) + } + if total != 1 || len(pipelines) != 1 { + t.Fatalf("total = %d, len = %d; want 1, 1", total, len(pipelines)) + } + + byName := map[string]*tangled.CiPipeline_Workflow{} + for _, wf := range pipelines[0].Workflows { + byName[wf.Name] = wf + } + + a, ok := byName["a"] + if !ok { + t.Fatal("workflow a missing") + } + if a.Status != string(models.StatusKindSuccess) { + t.Errorf("a status = %q, want success", a.Status) + } + if a.StartedAt == nil { + t.Error("a startedAt = nil, want the running transition's timestamp") + } + if a.FinishedAt == nil { + t.Error("a finishedAt = nil, want the terminal transition's timestamp") + } + + b, ok := byName["b"] + if !ok { + t.Fatal("workflow b missing") + } + if b.Status != string(models.StatusKindPending) { + t.Errorf("b status = %q, want pending", b.Status) + } + if b.StartedAt != nil || b.FinishedAt != nil { + t.Errorf("b times = (%q, %q), want both unset", derefStr(b.StartedAt), derefStr(b.FinishedAt)) + } +} + +func derefStr(s *string) string { + if s == nil { + return "" + } + return *s +} + +func TestToCiPipeline_TriggerKinds(t *testing.T) { + repo := "did:plc:boltless" + sha := "1111111111111111111111111111111111111111" + + tests := []struct { + kind workflow.TriggerKind + meta func(*tangled.Pipeline_TriggerMetadata) + wantUnion func(*tangled.CiPipeline_Trigger) bool + }{ + { + kind: workflow.TriggerKindPush, + meta: func(m *tangled.Pipeline_TriggerMetadata) { + m.Push = &tangled.Pipeline_PushTriggerData{NewSha: sha, Ref: "refs/heads/main"} + }, + wantUnion: func(tr *tangled.CiPipeline_Trigger) bool { return tr.CiTrigger_Push != nil }, + }, + { + kind: workflow.TriggerKindPullRequest, + meta: func(m *tangled.Pipeline_TriggerMetadata) { + m.PullRequest = &tangled.Pipeline_PullRequestTriggerData{ + SourceSha: sha, SourceBranch: "feature", TargetBranch: "main", + } + }, + wantUnion: func(tr *tangled.CiPipeline_Trigger) bool { return tr.CiTrigger_PullRequest != nil }, + }, + { + kind: workflow.TriggerKindManual, + meta: func(m *tangled.Pipeline_TriggerMetadata) { + m.Manual = &tangled.Pipeline_ManualTriggerData{Sha: sha} + }, + wantUnion: func(tr *tangled.CiPipeline_Trigger) bool { return tr.CiTrigger_Manual != nil }, + }, + } + + for _, tt := range tests { + t.Run(string(tt.kind), func(t *testing.T) { + meta := &tangled.Pipeline_TriggerMetadata{ + Kind: string(tt.kind), + Repo: &tangled.Pipeline_TriggerRepo{Knot: "knot.test", RepoDid: &repo, Did: repo}, + } + tt.meta(meta) + + p, kind := mapToCiPipeline("rk1", time.Now(), tangled.Pipeline{ + TriggerMetadata: meta, + Workflows: []*tangled.Pipeline_Workflow{{Name: "ci.yml"}}, + }) + + if kind != tt.kind { + t.Errorf("kind = %q, want %q", kind, tt.kind) + } + if p.Repo != repo { + t.Errorf("p.Repo = %q, want %q", p.Repo, repo) + } + if p.Commit != sha { + t.Errorf("p.Commit = %q, want %q", p.Commit, sha) + } + if !tt.wantUnion(p.Trigger) { + t.Errorf("wrong trigger union populated for kind %q", tt.kind) + } + // the payload must be marshalable, CreatePipeline stores it as JSON + if _, err := json.Marshal(p); err != nil { + t.Errorf("marshal payload: %v", err) + } + }) + } +} diff --git a/spindle/engine/engine.go b/spindle/engine/engine.go index e25b0704a..a2d33dd4d 100644 --- a/spindle/engine/engine.go +++ b/spindle/engine/engine.go @@ -273,8 +273,8 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q "workflow_id", wid.String(), ) - if st, err := db.GetStatus(wid); err == nil && models.StatusKind(st.Status).IsFinish() { - wl.Info("skipping finished workflow", "wid", wid, "status", st.Status) + if st, err := db.GetStatus(wid); err == nil && st.IsFinish() { + wl.Info("skipping finished workflow", "wid", wid, "status", st) return } diff --git a/spindle/engine/engine_test.go b/spindle/engine/engine_test.go index de7ce8c47..8134c4c49 100644 --- a/spindle/engine/engine_test.go +++ b/spindle/engine/engine_test.go @@ -164,17 +164,17 @@ func TestStartWorkflows_CollisionRejection(t *testing.T) { widUnique := models.WorkflowId{PipelineId: pipelineId, Name: "unique_job"} status1, err := testDB.GetStatus(widColliding1) - if err != nil || status1.Status != string(models.StatusKindFailed) { + if err != nil || status1 != models.StatusKindFailed { t.Fatalf("expected colliding1 status to be failed, got status=%v err=%v", status1, err) } status2, err := testDB.GetStatus(widColliding2) - if err != nil || status2.Status != string(models.StatusKindFailed) { + if err != nil || status2 != models.StatusKindFailed { t.Fatalf("expected colliding2 status to be failed, got status=%v err=%v", status2, err) } statusUnique, err := testDB.GetStatus(widUnique) - if err != nil || statusUnique.Status != string(models.StatusKindSuccess) { + if err != nil || statusUnique != models.StatusKindSuccess { t.Fatalf("expected unique status to be success, got status=%v err=%v", statusUnique, err) } } @@ -243,8 +243,8 @@ func TestCancelWorkflow_NotOverwritten(t *testing.T) { if err != nil { t.Fatalf("GetStatus error = %v", err) } - if st.Status != string(models.StatusKindCancelled) { - t.Fatalf("expected status to be cancelled, got %s", st.Status) + if st != models.StatusKindCancelled { + t.Fatalf("expected status to be cancelled, got %s", st) } } @@ -319,8 +319,8 @@ func TestStartWorkflows_FlushesLogBeforeTerminalStatus(t *testing.T) { if err != nil { t.Fatal(err) } - if status.Status != string(models.StatusKindFailed) { - t.Fatalf("expected failed status, got %s", status.Status) + if status != models.StatusKindFailed { + t.Fatalf("expected failed status, got %s", status) } log, err := os.ReadFile(models.LogFilePath(logDir, wid)) @@ -370,8 +370,8 @@ func TestSetupTimeout_ReportsTimeout(t *testing.T) { if err != nil { t.Fatalf("GetStatus error = %v", err) } - if st.Status != string(models.StatusKindTimeout) { - t.Fatalf("expected status to be timeout, got %s", st.Status) + if st != models.StatusKindTimeout { + t.Fatalf("expected status to be timeout, got %s", st) } if len(eng.runStepCalls) != 0 { diff --git a/spindle/mill/auth_test.go b/spindle/mill/auth_test.go index 6464d1dff..47843d789 100644 --- a/spindle/mill/auth_test.go +++ b/spindle/mill/auth_test.go @@ -299,8 +299,8 @@ func TestOnStatusEventOwnership(t *testing.T) { }, }, }) - if _, err := bdb.GetStatus(foreign.wid); err == nil { - t.Fatal("status stream for a foreign lease authored a status row; an executor forged another pipeline's status") + if st, err := bdb.GetStatus(foreign.wid); err != nil || st != "" { + t.Fatalf("status stream for a foreign lease authored status %q (err %v); an executor forged another pipeline's status", st, err) } _ = m.onEventBatch(sessZ, &millv1.EventBatch{ @@ -321,8 +321,8 @@ func TestOnStatusEventOwnership(t *testing.T) { if err != nil { t.Fatalf("owned status stream did not author a status row: %v", err) } - if st.Status != "running" { - t.Fatalf("owned status = %q, want %q", st.Status, "running") + if st != "running" { + t.Fatalf("owned status = %q, want %q", st, "running") } } diff --git a/spindle/mill/executor/observe.go b/spindle/mill/executor/observe.go index c21f89af0..b88c54ffe 100644 --- a/spindle/mill/executor/observe.go +++ b/spindle/mill/executor/observe.go @@ -12,7 +12,7 @@ import ( "time" "github.com/hpcloud/tail" - "tangled.org/core/api/tangled" + "tangled.org/core/spindle/db" "tangled.org/core/spindle/engine" millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" @@ -41,21 +41,17 @@ func (e *Executor) drainEvents(cursor *int64) { e.l.Error("drain status events failed", "err", err) return } - for _, ev := range events { - if ev.Created > *cursor { - *cursor = ev.Created + for _, event := range events { + if event.Id > *cursor { + *cursor = event.Id } - st, ok := parseStatus(ev.EventJson) - if !ok { - continue - } - if err := e.onStatusRow(st); err != nil { + if err := e.onStatusRow(&event); err != nil { e.l.Error("process status row failed", "err", err) } } } -func (e *Executor) onStatusRow(st *tangled.PipelineStatus) error { +func (e *Executor) onStatusRow(st *db.StatusRow) error { res := e.reservationFor(st.Pipeline, st.Workflow) if res == nil { return nil @@ -68,7 +64,7 @@ func (e *Executor) onStatusRow(st *tangled.PipelineStatus) error { return e.appendStatus(res.leaseID, st) } -func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) error { +func (e *Executor) finishJob(res *reservation, st *db.StatusRow) error { e.mu.Lock() if e.active[res.leaseID] != res { e.mu.Unlock() @@ -175,11 +171,11 @@ func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) error return nil } -func (e *Executor) reservationFor(pipelineAturi, workflow string) *reservation { +func (e *Executor) reservationFor(rkey, workflow string) *reservation { e.mu.Lock() defer e.mu.Unlock() for _, res := range e.active { - if string(res.wid.PipelineId.AtUri()) == pipelineAturi && res.wid.Name == workflow { + if res.wid.PipelineId.Rkey == rkey && res.wid.Name == workflow { return res } } diff --git a/spindle/mill/executor/outbox.go b/spindle/mill/executor/outbox.go index a01fca2db..5bf45a596 100644 --- a/spindle/mill/executor/outbox.go +++ b/spindle/mill/executor/outbox.go @@ -4,13 +4,11 @@ import ( "context" "crypto/rand" "encoding/hex" - "encoding/json" "fmt" "os" "time" "google.golang.org/protobuf/proto" - "tangled.org/core/api/tangled" "tangled.org/core/spindle/db" "tangled.org/core/spindle/engine" millproto "tangled.org/core/spindle/mill/proto" @@ -127,7 +125,7 @@ func (e *Executor) appendAndSend(leaseID string, payload any, control bool) erro return nil } -func (e *Executor) appendStatus(leaseID string, st *tangled.PipelineStatus) error { +func (e *Executor) appendStatus(leaseID string, st *db.StatusRow) error { if st.Status != string(models.StatusKindRunning) { return fmt.Errorf("unsupported nonterminal status %q", st.Status) } @@ -140,7 +138,7 @@ func (e *Executor) appendStatus(leaseID string, st *tangled.PipelineStatus) erro return e.appendAndSend(leaseID, payload, true) } -func (e *Executor) appendTerminal(leaseID, status string, st *tangled.PipelineStatus) error { +func (e *Executor) appendTerminal(leaseID, status string, st *db.StatusRow) error { failureClass, failureReason := defaultTerminalAttribution(status) return e.appendTerminalWithArtifact(leaseID, status, st, "", "", failureClass, failureReason, true) } @@ -160,7 +158,7 @@ func defaultTerminalAttribution(status string) (string, string) { func (e *Executor) appendTerminalWithArtifact( leaseID, status string, - st *tangled.PipelineStatus, + st *db.StatusRow, ref, hash, failureClass, failureReason string, millRecordsTerminalMetrics bool, ) error { @@ -233,7 +231,7 @@ func (e *Executor) recoverPendingArtifacts(parent context.Context) error { } } - st := &tangled.PipelineStatus{ + st := &db.StatusRow{ Status: p.Status, Error: &p.Error, ExitCode: &p.ExitCode, @@ -357,14 +355,6 @@ func (e *Executor) subtractOutboxBytes(deleted db.OutboxDeletion) { e.outboxBytes = max(0, e.outboxBytes-deleted.Bytes) } -func parseStatus(raw json.RawMessage) (*tangled.PipelineStatus, bool) { - var st tangled.PipelineStatus - if err := json.Unmarshal(raw, &st); err != nil { - return nil, false - } - return &st, true -} - func (e *Executor) deleteOutboxPrefix(upTo uint64) error { e.eventMu.Lock() defer e.eventMu.Unlock() @@ -401,7 +391,7 @@ func (e *Executor) outboxIdle() <-chan struct{} { return idle } -func parseStatusExitAndError(st *tangled.PipelineStatus) (int64, string) { +func parseStatusExitAndError(st *db.StatusRow) (int64, string) { if st == nil { return 0, "" } diff --git a/spindle/mill/executor/reserved_test.go b/spindle/mill/executor/reserved_test.go index efed26d76..27e14bb94 100644 --- a/spindle/mill/executor/reserved_test.go +++ b/spindle/mill/executor/reserved_test.go @@ -303,7 +303,7 @@ func TestTerminalOutboxEntrySuppressesLaterEvents(t *testing.T) { } e.handleCancel("lease-1") - if err := e.appendStatus("lease-1", &tangled.PipelineStatus{Status: string(models.StatusKindRunning)}); err != nil { + if err := e.appendStatus("lease-1", &db.StatusRow{Status: string(models.StatusKindRunning)}); err != nil { t.Fatal(err) } @@ -826,8 +826,8 @@ func TestFinishJobReportsCancelledReservationAsCancelled(t *testing.T) { t.Fatal(err) } - e.finishJob(res, &tangled.PipelineStatus{ - Pipeline: string(res.wid.PipelineId.AtUri()), + e.finishJob(res, &db.StatusRow{ + Pipeline: res.wid.PipelineId.Rkey, Workflow: res.wid.Name, Status: string(models.StatusKindFailed), }) @@ -872,8 +872,8 @@ func TestFinishJobWaitsForEngineCleanup(t *testing.T) { done := make(chan error, 1) go func() { - done <- e.finishJob(res, &tangled.PipelineStatus{ - Pipeline: string(res.wid.PipelineId.AtUri()), + done <- e.finishJob(res, &db.StatusRow{ + Pipeline: res.wid.PipelineId.Rkey, Workflow: res.wid.Name, Status: string(models.StatusKindSuccess), }) diff --git a/spindle/mill/integration_test.go b/spindle/mill/integration_test.go index 50d36fa51..beb3049eb 100644 --- a/spindle/mill/integration_test.go +++ b/spindle/mill/integration_test.go @@ -2,7 +2,6 @@ package mill import ( "context" - "encoding/json" "io" "log/slog" "net/http" @@ -351,21 +350,13 @@ func waitForSessionLabels(t *testing.T, m *Mill, nodeID string, want []string) b func waitForStatus(t *testing.T, d *db.DB, wid models.WorkflowId, want string) bool { t.Helper() deadline := time.Now().Add(3 * time.Second) - aturi := string(wid.PipelineId.AtUri()) for time.Now().Before(deadline) { evs, err := d.GetEvents(0, 1000) if err != nil { t.Fatalf("GetEvents: %v", err) } for _, ev := range evs { - if ev.Nsid != tangled.PipelineStatusNSID { - continue - } - var st tangled.PipelineStatus - if err := json.Unmarshal(ev.EventJson, &st); err != nil { - continue - } - if st.Pipeline == aturi && st.Workflow == wid.Name && st.Status == want { + if ev.Pipeline == wid.PipelineId.Rkey && ev.Workflow == wid.Name && ev.Status == want { return true } } diff --git a/spindle/mill/mill.go b/spindle/mill/mill.go index 9d5b3e311..299f47744 100644 --- a/spindle/mill/mill.go +++ b/spindle/mill/mill.go @@ -1401,17 +1401,16 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { if c := ev.GetExitCode(); c != 0 { exitCode = &c } - pipelineAtUri := string(lease.wid.PipelineId.AtUri()) if tx != nil { - alreadyRunning, err := tx.HasWorkflowStatus(context.Background(), pipelineAtUri, lease.wid.Name, statusStr) + alreadyRunning, err := tx.HasWorkflowStatus(context.Background(), lease.wid, statusStr) if err != nil { return err } - if err := tx.InsertStatusEvent(pipelineAtUri, lease.wid.Name, statusStr, errMsg, exitCode); err != nil { + if err := tx.InsertStatusEvent(lease.wid, statusStr, errMsg, exitCode); err != nil { return err } if !alreadyRunning { - delay, ok, err := tx.WorkflowStartupDelay(context.Background(), pipelineAtUri, lease.wid.Name) + delay, ok, err := tx.WorkflowStartupDelay(context.Background(), lease.wid) if err != nil { return err } @@ -1448,9 +1447,8 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { exitCode = &c } finishedInBatch[lease.id] = struct{}{} - pipelineAtUri := string(lease.wid.PipelineId.AtUri()) if tx != nil { - if err := tx.InsertStatusEvent(pipelineAtUri, lease.wid.Name, statusStr, errMsg, exitCode); err != nil { + if err := tx.InsertStatusEvent(lease.wid, statusStr, errMsg, exitCode); err != nil { return err } if err := tx.DeleteLease(lease.id); err != nil { diff --git a/spindle/mill/mill_test.go b/spindle/mill/mill_test.go index 4c1501e11..4132eceda 100644 --- a/spindle/mill/mill_test.go +++ b/spindle/mill/mill_test.go @@ -637,9 +637,9 @@ func TestAtomicBatchRollback(t *testing.T) { if _, err := bdb.Exec(` create trigger reject_status_event - before insert on events + before insert on workflow_statuses begin - select raise(abort, 'forced status event failure'); + select raise(abort, 'forced status insert failure'); end `); err != nil { t.Fatalf("failed to create fail trigger: %v", err) @@ -678,7 +678,7 @@ func TestTerminalBeforeACK(t *testing.T) { if msg.GetAck() != nil { wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} st, err := bdb.GetStatus(wid) - if err != nil || st.Status != "success" { + if err != nil || st != "success" { t.Errorf("expected terminal status success at ACK time, got status: %v, err: %v", st, err) } close(ackSent) diff --git a/spindle/mill/restore.go b/spindle/mill/restore.go index 5a4f12740..b895d8105 100644 --- a/spindle/mill/restore.go +++ b/spindle/mill/restore.go @@ -203,8 +203,7 @@ func (m *Mill) completeLeaseRow(lease *RemoteLease, status string, errMsg *strin } return m.db.CompleteMillLease( lease.id, - string(lease.wid.PipelineId.AtUri()), - lease.wid.Name, + lease.wid, status, errMsg, exitCode, diff --git a/spindle/mill/restore_test.go b/spindle/mill/restore_test.go index d0b3f3017..fe023bd5d 100644 --- a/spindle/mill/restore_test.go +++ b/spindle/mill/restore_test.go @@ -119,8 +119,8 @@ func TestOrphanTerminalAuthorsStatusRow(t *testing.T) { if err != nil { t.Fatalf("GetStatus after orphan terminal: %v", err) } - if st.Status != string(models.StatusKindSuccess) { - t.Fatalf("orphan terminal authored status %q, want success", st.Status) + if st != models.StatusKindSuccess { + t.Fatalf("orphan terminal authored status %q, want success", st) } m.mu.Lock() @@ -170,8 +170,8 @@ func TestSnapshotReconciliationFailsDroppedOrphans(t *testing.T) { if err != nil { t.Fatalf("GetStatus for dropped orphan: %v", err) } - if st.Status != string(models.StatusKindFailed) { - t.Fatalf("dropped orphan authored status %q, want failed", st.Status) + if st != models.StatusKindFailed { + t.Fatalf("dropped orphan authored status %q, want failed", st) } } func TestSnapshotReconciliationPreservesRequestedCancellation(t *testing.T) { @@ -200,8 +200,8 @@ func TestSnapshotReconciliationPreservesRequestedCancellation(t *testing.T) { if err != nil { t.Fatalf("GetStatus: %v", err) } - if st.Status != string(models.StatusKindCancelled) { - t.Fatalf("reconciled status = %q, want cancelled", st.Status) + if st != models.StatusKindCancelled { + t.Fatalf("reconciled status = %q, want cancelled", st) } } @@ -271,8 +271,8 @@ func TestSweepFailsOrphansOfAbsentExecutors(t *testing.T) { if err != nil { t.Fatalf("GetStatus after sweep: %v", err) } - if st.Status != string(models.StatusKindFailed) { - t.Fatalf("sweep authored status %q, want failed", st.Status) + if st != models.StatusKindFailed { + t.Fatalf("sweep authored status %q, want failed", st) } if rows, _ := bdb.ListMillLeases(); len(rows) != 0 { t.Fatalf("swept orphan still persisted: %+v", rows) @@ -374,12 +374,12 @@ func TestOrphanTerminalFailureKeepsLeaseAndSeqnoRetryable(t *testing.T) { if rows, err := bdb.ListMillLeases(); err != nil || len(rows) != 1 { t.Fatalf("durable leases after transaction rollback = %+v, err = %v; want retained lease", rows, err) } - var events int - if err := bdb.QueryRow(`select count(*) from events`).Scan(&events); err != nil { - t.Fatalf("count events: %v", err) + var statuses int + if err := bdb.QueryRow(`select count(*) from workflow_statuses`).Scan(&statuses); err != nil { + t.Fatalf("count workflow statuses: %v", err) } - if events != 0 { - t.Fatalf("terminal events after transaction rollback = %d, want 0", events) + if statuses != 0 { + t.Fatalf("terminal statuses after transaction rollback = %d, want 0", statuses) } if _, err := bdb.Exec(`drop trigger reject_orphan_lease_delete`); err != nil { diff --git a/spindle/observability/metrics.go b/spindle/observability/metrics.go index 728be908a..9bb2d8e1a 100644 --- a/spindle/observability/metrics.go +++ b/spindle/observability/metrics.go @@ -1060,15 +1060,12 @@ func (c *dbCollector) refresh(ctx context.Context) { rows, err := c.db.QueryContext(ctx, ` WITH latest AS ( SELECT - json_extract(event, '$.status') AS status, + status, row_number() OVER ( - PARTITION BY - json_extract(event, '$.pipeline'), - json_extract(event, '$.workflow') - ORDER BY created DESC, rowid DESC + PARTITION BY rkey, workflow + ORDER BY id DESC ) AS rank - FROM events - WHERE nsid = 'sh.tangled.pipeline.status' + FROM workflow_statuses ) SELECT CASE status diff --git a/spindle/observability/metrics_test.go b/spindle/observability/metrics_test.go index c404a65c3..c230a2a26 100644 --- a/spindle/observability/metrics_test.go +++ b/spindle/observability/metrics_test.go @@ -7,6 +7,7 @@ import ( "net/http" "net/http/httptest" "os" + "path/filepath" "regexp" "strings" "sync" @@ -20,6 +21,7 @@ import ( "go.opentelemetry.io/otel" sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" + "tangled.org/core/spindle/db" ) var forbiddenLabelValues = []string{ @@ -907,3 +909,33 @@ func TestWorkflowQueueAndExecutorMetrics(t *testing.T) { t.Fatalf("spindle_executor_seats = %v, want 8", seats) } } + +func TestDBCollectorUsesLatestNormalizedWorkflowStatus(t *testing.T) { + ctx := context.Background() + database, err := db.Make(ctx, filepath.Join(t.TempDir(), "metrics.db")) + if err != nil { + t.Fatal(err) + } + defer database.Close() + for _, row := range []struct { + pipeline, workflow, status string + }{ + {"p1", "build", "pending"}, + {"p1", "build", "running"}, + {"p2", "test", "failed"}, + {"p1", "build", "success"}, + } { + if _, err := database.Exec(`insert into workflow_statuses (rkey, workflow, status, created_at) values (?, ?, ?, ?)`, row.pipeline, row.workflow, row.status, time.Now().Format(time.RFC3339)); err != nil { + t.Fatal(err) + } + } + collector := &dbCollector{ + db: database, + collectionFailures: prometheus.NewCounter(prometheus.CounterOpts{Name: "test_collection_failures_total"}), + workflowStatus: make(map[string]int), + } + collector.refresh(ctx) + if collector.workflowStatus["success"] != 1 || collector.workflowStatus["failed"] != 1 || len(collector.workflowStatus) != 2 { + t.Fatalf("workflow statuses = %#v", collector.workflowStatus) + } +} diff --git a/spindle/server.go b/spindle/server.go index 059801ce2..4cc5f65e5 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -719,9 +719,6 @@ func (s *Spindle) Router() http.Handler { _, _ = io.WriteString(w, "{\"status\":\"ok\"}\n") }) if s.cfg.Role != config.RoleExecutor { - mux.HandleFunc("/events", s.Events) - mux.HandleFunc("/logs/{knot}/{rkey}/{name}", s.Logs) - if s.mill != nil { mux.HandleFunc("/mill", s.mill.HandleExecutorConn) } @@ -864,8 +861,8 @@ func (s *Spindle) runPipeline(ctx context.Context, repoDid syntax.DID, trigger t Knot: trigger.Repo.Knot, Rkey: tid.TID(), } - if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { - return models.PipelineId{}, fmt.Errorf("creating pipeline event: %w", err) + if err := s.db.CreatePipeline(pipelineId, tpl); err != nil { + return models.PipelineId{}, fmt.Errorf("creating pipeline: %w", err) } err = s.processPipeline(ctx, repoDid, tpl, pipelineId, sourceRepo) return pipelineId, err diff --git a/spindle/stream.go b/spindle/stream.go deleted file mode 100644 index e8f2ca0c0..000000000 --- a/spindle/stream.go +++ /dev/null @@ -1,142 +0,0 @@ -package spindle - -import ( - "context" - "errors" - "fmt" - "net/http" - "time" - - "tangled.org/core/eventstream" - "tangled.org/core/log" - "tangled.org/core/spindle/logview" - "tangled.org/core/spindle/models" - - "github.com/go-chi/chi/v5" - "github.com/gorilla/websocket" -) - -var upgrader = websocket.Upgrader{ - ReadBufferSize: 1024, - WriteBufferSize: 1024, -} - -func (s *Spindle) Events(w http.ResponseWriter, r *http.Request) { - l := log.SubLogger(s.l, "eventstream") - l.Debug("received new connection") - - err := eventstream.Stream(w, r, eventstream.StreamConfig{ - Backend: s.db, - Notifier: s.n, - Logger: l, - }) - if err != nil && !errors.Is(err, eventstream.ErrDrainCap) { - l.Error("event stream ended with error", "err", err) - } -} - -func (s *Spindle) Logs(w http.ResponseWriter, r *http.Request) { - wid, err := getWorkflowID(r) - if err != nil { - http.Error(w, err.Error(), http.StatusBadRequest) - return - } - - l := s.l.With("handler", "Logs") - l = s.l.With("wid", wid) - - conn, err := upgrader.Upgrade(w, r, nil) - if err != nil { - l.Error("websocket upgrade failed", "err", err) - http.Error(w, "failed to upgrade", http.StatusInternalServerError) - return - } - defer func() { - _ = conn.WriteControl( - websocket.CloseMessage, - websocket.FormatCloseMessage(websocket.CloseNormalClosure, "log stream complete"), - time.Now().Add(time.Second), - ) - conn.Close() - }() - l.Debug("upgraded http to wss") - - ctx, cancel := context.WithCancel(r.Context()) - defer cancel() - - go func() { - for { - if _, _, err := conn.NextReader(); err != nil { - l.Debug("client disconnected", "err", err) - cancel() - return - } - } - }() - - if err := s.streamLogsFromDisk(ctx, conn, wid); err != nil { - l.Info("log stream ended", "err", err) - } - - l.Info("logs connection closed") -} - -func (s *Spindle) streamLogsFromDisk(ctx context.Context, conn *websocket.Conn, wid models.WorkflowId) error { - status, err := s.db.GetStatus(wid) - if err != nil { - return err - } - isFinished := models.StatusKind(status.Status).IsFinish() - - lines, stop, err := logview.Follow(ctx, s.db, s.reader, s.cfg.Server.LogDir, wid, isFinished) - if err != nil { - return fmt.Errorf("failed to follow workflow log: %w", err) - } - defer stop() - for { - select { - case <-ctx.Done(): - return ctx.Err() - case line, ok := <-lines: - if !ok && isFinished { - return fmt.Errorf("log completed") - } - if !ok { - return fmt.Errorf("log channel closed unexpectedly") - } - if line == nil { - continue - } - if line.Err != nil { - return fmt.Errorf("error following workflow log: %w", line.Err) - } - - if err := conn.WriteMessage(websocket.TextMessage, []byte(line.Text)); err != nil { - return fmt.Errorf("failed to write to websocket: %w", err) - } - case <-time.After(30 * time.Second): - // send a keep-alive - if err := conn.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(time.Second)); err != nil { - return fmt.Errorf("failed to write control: %w", err) - } - } - } -} - -func getWorkflowID(r *http.Request) (models.WorkflowId, error) { - knot := chi.URLParam(r, "knot") - rkey := chi.URLParam(r, "rkey") - name := chi.URLParam(r, "name") - - if knot == "" || rkey == "" || name == "" { - return models.WorkflowId{}, fmt.Errorf("missing required parameters") - } - - return models.WorkflowId{ - PipelineId: models.PipelineId{ - Knot: knot, - Rkey: rkey, - }, - Name: name, - }, nil -} diff --git a/spindle/tapclient.go b/spindle/tapclient.go index 02d1ffd73..c356961ac 100644 --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -628,7 +628,7 @@ func (s *Spindle) triggerPullRequestPipeline(ctx context.Context, l *slog.Logger Knot: tpl.TriggerMetadata.Repo.Knot, Rkey: tid.TID(), } - if err := s.db.CreatePipelineEvent(pipelineId.Rkey, tpl, s.n); err != nil { + if err := s.db.CreatePipeline(pipelineId, tpl); err != nil { l.Error("failed to create pipeline event", "err", err) return nil } diff --git a/spindle/wipe.go b/spindle/wipe.go index f3a29ff1a..b8168032e 100644 --- a/spindle/wipe.go +++ b/spindle/wipe.go @@ -174,7 +174,7 @@ func (s *Spindle) cancelRepoWorkflows(ctx context.Context, repoDid syntax.DID, r var errs []error for _, wid := range wids { st, err := s.db.GetStatus(wid) - if err == nil && st != nil && models.StatusKind(st.Status).IsFinish() { + if err == nil && st.IsFinish() { continue } if err := s.db.StatusCancelled(wid, reason, -1, s.n); err != nil { diff --git a/spindle/wipe_test.go b/spindle/wipe_test.go index e5c2d0c78..289bcf8d3 100644 --- a/spindle/wipe_test.go +++ b/spindle/wipe_test.go @@ -2,7 +2,7 @@ package spindle import ( "context" - "fmt" + "encoding/json" "os" "path/filepath" "testing" @@ -104,12 +104,19 @@ func seedWipeRepo(t *testing.T, s *Spindle, repoDid, owner syntax.DID) { if err := s.vault.AddSecret(ctx, secrets.UnlockedSecret{Key: "api_key", Value: "v", Repo: secrets.RepoIdentifier(repoDid.String()), CreatedBy: owner}); err != nil { t.Fatal(err) } - pipelineEvent := fmt.Sprintf(`{"triggerMetadata":{"repo":{"repoDid":"%s"}}}`, repoDid) - if _, err := s.db.Exec(`insert into events (nsid, rkey, event, created) values ('sh.tangled.pipeline', ?, ?, 1)`, "p"+sfx, pipelineEvent); err != nil { + pipeline := &tangled.CiPipeline{ + Id: "p" + sfx, + Repo: repoDid.String(), + Workflows: []*tangled.CiPipeline_Workflow{{Id: "w" + sfx, Name: "w" + sfx, Status: "pending"}}, + } + payload, err := json.Marshal(pipeline) + if err != nil { + t.Fatal(err) + } + if _, err := s.db.Exec(`insert into pipelines (rkey, knot, repo_did, commit_sha, kind, payload) values (?, 'knot.example.com', ?, '', '', ?)`, pipeline.Id, repoDid.String(), string(payload)); err != nil { t.Fatal(err) } - statusEvent := fmt.Sprintf(`{"pipeline":"at://did:web:knot.example.com/sh.tangled.pipeline/%s"}`, "p"+sfx) - if _, err := s.db.Exec(`insert into events (nsid, rkey, event, created) values ('sh.tangled.pipeline.status', ?, ?, 1)`, "s"+sfx, statusEvent); err != nil { + if _, err := s.db.Exec(`insert into workflow_statuses (rkey, workflow, status, created_at) values (?, ?, 'pending', 'now')`, pipeline.Id, "w"+sfx); err != nil { t.Fatal(err) } @@ -145,7 +152,7 @@ func TestWipeRepoRemovesAllState(t *testing.T) { } var n int - for _, table := range []string{"repos", "jobs", "mill_leases", "mill_artifacts", "quota_allocations", "repo_collaborators", "events"} { + for _, table := range []string{"repos", "jobs", "mill_leases", "mill_artifacts", "quota_allocations", "repo_collaborators", "pipelines", "workflow_statuses"} { if err := s.db.QueryRow(`select count(*) from ` + table).Scan(&n); err != nil { t.Fatal(err) } diff --git a/spindle/xrpc/ci_pipeline_subscribe_logs.go b/spindle/xrpc/ci_pipeline_subscribe_logs.go index 4a6d763fa..b7c7b3126 100644 --- a/spindle/xrpc/ci_pipeline_subscribe_logs.go +++ b/spindle/xrpc/ci_pipeline_subscribe_logs.go @@ -41,33 +41,14 @@ var wsUpgrader = websocket.Upgrader{ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipeline syntax.TID, workflows []string) { l := x.Logger.With("pipeline", pipeline, "workflows", workflows) - // 1. query the event from database to get the knot - var eventJson string - err := x.Db.QueryRow( - `select event from events where nsid = ? and rkey = ?`, - tangled.PipelineNSID, - pipeline.String(), - ).Scan(&eventJson) + // 1. query the pipeline from database to get the knot. knot is used to locate the workflow log files. + tpl, knot, err := x.Db.GetPipelineWithKnot(r.Context(), pipeline.String()) if err != nil { l.ErrorContext(r.Context(), "failed to find pipeline event", "err", err) writeJson(w, http.StatusNotFound, atclient.ErrorBody{Name: "NotFound", Message: fmt.Sprintf("pipeline not found: %s", pipeline.String())}) return } - var tpl tangled.Pipeline - if err := json.Unmarshal([]byte(eventJson), &tpl); err != nil { - l.ErrorContext(r.Context(), "failed to unmarshal pipeline event", "err", err) - writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalError", Message: "failed to parse pipeline event"}) - return - } - - if tpl.TriggerMetadata == nil || tpl.TriggerMetadata.Repo == nil { - l.ErrorContext(r.Context(), "pipeline event trigger metadata is incomplete") - writeJson(w, http.StatusInternalServerError, atclient.ErrorBody{Name: "InternalError", Message: "pipeline event trigger metadata is incomplete"}) - return - } - knot := tpl.TriggerMetadata.Repo.Knot - // 2. if workflows is empty, default to all workflows defined in the pipeline if len(workflows) == 0 { for _, wf := range tpl.Workflows { @@ -164,7 +145,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel var isFinished bool status, err := x.Db.GetStatus(wid) if err == nil { - isFinished = models.StatusKind(status.Status).IsFinish() + isFinished = status.IsFinish() } lines, stop, err := logview.Follow(ctx, x.Db, x.ArtifactReader, x.Config.Server.LogDir, wid, isFinished) @@ -189,7 +170,7 @@ func (x *Xrpc) handleSubscribeLogs(w http.ResponseWriter, r *http.Request, pipel return case <-ticker.C: status, err := x.Db.GetStatus(wid) - if err == nil && models.StatusKind(status.Status).IsFinish() { + if err == nil && status.IsFinish() { stop() return } diff --git a/spindle/xrpc/pipeline_cancel_pipeline.go b/spindle/xrpc/pipeline_cancel_pipeline.go index 71dce36e6..2a8f6844f 100644 --- a/spindle/xrpc/pipeline_cancel_pipeline.go +++ b/spindle/xrpc/pipeline_cancel_pipeline.go @@ -89,7 +89,7 @@ func (x *Xrpc) CancelPipeline(w http.ResponseWriter, r *http.Request) { // dont cancel a workflow that already finished st, err := x.Db.GetStatus(wid) - if err == nil && models.StatusKind(st.Status).IsFinish() { + if err == nil && st.IsFinish() { continue } diff --git a/spindle/xrpc/xrpc_test.go b/spindle/xrpc/xrpc_test.go index fa2004e76..8255ff66a 100644 --- a/spindle/xrpc/xrpc_test.go +++ b/spindle/xrpc/xrpc_test.go @@ -183,19 +183,17 @@ func TestCancelPipeline_RBAC(t *testing.T) { Knot: "knot.test", Did: repoOwnerDid.String(), }, + Manual: &tangled.Pipeline_ManualTriggerData{ + Sha: "1111111111111111111111111111111111111111", + }, }, Workflows: []*tangled.Pipeline_Workflow{ {Name: "test-workflow"}, }, } - err = d.CreatePipelineEvent(pipelineTid, tpl, nil) - if err != nil { - t.Fatalf("CreatePipelineEvent: %v", err) - } - - _, err = d.Exec(`UPDATE pipelines SET repo_did = ? WHERE id = ?`, repoDid.String(), pipelineTid) + err = d.CreatePipeline(models.PipelineId{Knot: "knot.test", Rkey: pipelineTid}, tpl) if err != nil { - t.Fatalf("Update pipeline repo association: %v", err) + t.Fatalf("CreatePipeline: %v", err) } x := &Xrpc{ -- 2.51.2