From 495425e2e5d6dcab7e75ea4934575587c7fb1e66 Mon Sep 17 00:00:00 2001 From: Seongmin Lee Date: Thu, 27 Aug 2026 13:14:37 +0900 Subject: [PATCH] spindle: deprecate `events` table Signed-off-by: Seongmin Lee --- orm/orm.go | 1 + spindle/db/db.go | 183 ++++++++++++ spindle/db/events.go | 123 ++++---- spindle/db/mill_state.go | 10 +- spindle/db/mill_state_test.go | 15 +- spindle/db/pipelines.go | 309 +++++++++++++-------- spindle/db/pipelines_test.go | 177 ++++++++++-- spindle/engine/engine.go | 4 +- spindle/engine/engine_test.go | 14 +- spindle/mill/auth_test.go | 8 +- spindle/mill/executor/observe.go | 22 +- spindle/mill/executor/outbox.go | 20 +- spindle/mill/executor/reserved_test.go | 4 +- spindle/mill/integration_test.go | 11 +- spindle/mill/mill.go | 6 +- spindle/mill/mill_test.go | 6 +- spindle/mill/restore.go | 3 +- spindle/mill/restore_test.go | 26 +- spindle/server.go | 7 +- spindle/stream.go | 142 ---------- spindle/tapclient.go | 2 +- spindle/xrpc/ci_pipeline_subscribe_logs.go | 27 +- spindle/xrpc/pipeline_cancel_pipeline.go | 4 +- spindle/xrpc/xrpc_test.go | 12 +- 24 files changed, 662 insertions(+), 474 deletions(-) delete mode 100644 spindle/stream.go diff --git a/orm/orm.go b/orm/orm.go index bc37ec48..92718378 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 de8de374..766b52bd 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" ) @@ -193,6 +196,10 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { name text unique ); `) + if err != nil { + return nil, err + } + if err := runMigrations(ctx, conn, logger); err != nil { return nil, err } @@ -322,6 +329,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 } @@ -391,3 +450,127 @@ func (d *DB) GetLastTimeUs() (int64, error) { err := row.Scan(&lastTimeUs) return lastTimeUs, err } + +// 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/events.go b/spindle/db/events.go index b30fc0b0..44dcb2d7 100644 --- a/spindle/db/events.go +++ b/spindle/db/events.go @@ -1,62 +1,63 @@ package db import ( - "encoding/json" + "database/sql" "time" - "tangled.org/core/api/tangled" - "tangled.org/core/eventstream" "tangled.org/core/notifier" "tangled.org/core/spindle/models" - "tangled.org/core/tid" ) -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.RFC3339), + ) + 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( @@ -66,64 +67,58 @@ 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 } func (d *DB) StatusPending(workflowId models.WorkflowId, n *notifier.Notifier) error { diff --git a/spindle/db/mill_state.go b/spindle/db/mill_state.go index 95310fcf..fe74c310 100644 --- a/spindle/db/mill_state.go +++ b/spindle/db/mill_state.go @@ -4,8 +4,8 @@ import ( "database/sql" "fmt" - "tangled.org/core/eventstream" "tangled.org/core/notifier" + "tangled.org/core/spindle/models" ) // enough to rebuild the fencing token and workflow identity after a restart @@ -254,12 +254,8 @@ type EventBatchTx struct { 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 eventstream.Insert(tx.tx, event, nil) +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 { diff --git a/spindle/db/mill_state_test.go b/spindle/db/mill_state_test.go index c4990b9d..0d6fe42f 100644 --- a/spindle/db/mill_state_test.go +++ b/spindle/db/mill_state_test.go @@ -7,6 +7,7 @@ import ( "testing" "tangled.org/core/notifier" + "tangled.org/core/spindle/models" ) func TestMillLeaseRoundTrip(t *testing.T) { @@ -106,8 +107,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, @@ -117,11 +117,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) @@ -137,8 +137,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, @@ -146,11 +145,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 028f13f8..7ceb5f6c 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, }) } @@ -221,12 +203,12 @@ func (d *DB) mapToCiPipeline(rkey string, created int64, raw tangled.Pipeline) ( return &tangled.CiPipeline{ Id: rkey, Commit: commitSha, - Repo: &repoDidStr, + Repo: repoDidStr, CreatedAt: &createdAtStr, 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 71e9bddc..20274f87 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 { @@ -97,8 +91,8 @@ func TestQueryPipelines_KindScopedToRepo(t *testing.T) { if total != 1 || len(pipelines) != 1 { t.Fatalf("total=%d len=%d, want exactly alice's single push pipeline", total, len(pipelines)) } - if pipelines[0].Repo == nil || *pipelines[0].Repo != "did:plc:alice" { - t.Errorf("returned pipeline repo = %v, want did:plc:alice", pipelines[0].Repo) + if pipelines[0].Repo != "did:plc:alice" { + t.Errorf("returned pipeline repo = %q, want did:plc:alice", pipelines[0].Repo) } } @@ -116,3 +110,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 7b06be72..e011fea9 100644 --- a/spindle/engine/engine.go +++ b/spindle/engine/engine.go @@ -129,8 +129,8 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, s } wg.Go(func() { - if st, err := db.GetStatus(wid); err == nil && models.StatusKind(st.Status).IsFinish() { - l.Info("skipping finished workflow", "wid", wid, "status", st.Status) + if st, err := db.GetStatus(wid); err == nil && st.IsFinish() { + l.Info("skipping finished workflow", "wid", wid, "status", st) return } var err error diff --git a/spindle/engine/engine_test.go b/spindle/engine/engine_test.go index 90155334..dedcbb68 100644 --- a/spindle/engine/engine_test.go +++ b/spindle/engine/engine_test.go @@ -141,17 +141,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) } } @@ -220,8 +220,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) } } @@ -256,8 +256,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 6876f791..218e6f16 100644 --- a/spindle/mill/auth_test.go +++ b/spindle/mill/auth_test.go @@ -276,8 +276,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{ @@ -298,8 +298,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 8ef595fb..df9ed09b 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" millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" "tangled.org/core/spindle/models" @@ -40,21 +40,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 @@ -67,7 +63,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() @@ -151,11 +147,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 20ee0650..d51cd9d7 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" millproto "tangled.org/core/spindle/mill/proto" millv1 "tangled.org/core/spindle/mill/proto/gen" @@ -103,7 +101,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) } @@ -116,11 +114,11 @@ 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 { return e.appendTerminalWithArtifact(leaseID, status, st, "", "") } -func (e *Executor) appendTerminalWithArtifact(leaseID, status string, st *tangled.PipelineStatus, ref, hash string) error { +func (e *Executor) appendTerminalWithArtifact(leaseID, status string, st *db.StatusRow, ref, hash string) error { var terminalStatus millv1.TerminalStatus switch status { case string(models.StatusKindSuccess): @@ -184,7 +182,7 @@ func (e *Executor) recoverPendingArtifacts() error { } } - st := &tangled.PipelineStatus{ + st := &db.StatusRow{ Status: p.Status, Error: &p.Error, ExitCode: &p.ExitCode, @@ -300,14 +298,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() @@ -319,7 +309,7 @@ func (e *Executor) deleteOutboxPrefix(upTo uint64) error { return err } -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 7fbc9816..d13da68e 100644 --- a/spindle/mill/executor/reserved_test.go +++ b/spindle/mill/executor/reserved_test.go @@ -328,8 +328,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), }) diff --git a/spindle/mill/integration_test.go b/spindle/mill/integration_test.go index 303eecc4..553706b2 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" @@ -295,21 +294,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 ad7e386b..319e55a9 100644 --- a/spindle/mill/mill.go +++ b/spindle/mill/mill.go @@ -948,9 +948,8 @@ 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 { - 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 } } @@ -979,9 +978,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 1cf88d02..e4a44461 100644 --- a/spindle/mill/mill_test.go +++ b/spindle/mill/mill_test.go @@ -582,9 +582,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) @@ -623,7 +623,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 4a74db42..d5470072 100644 --- a/spindle/mill/restore.go +++ b/spindle/mill/restore.go @@ -167,8 +167,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 4694173c..806dc8dd 100644 --- a/spindle/mill/restore_test.go +++ b/spindle/mill/restore_test.go @@ -118,8 +118,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() @@ -169,8 +169,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) { @@ -199,8 +199,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) } } @@ -253,8 +253,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) @@ -356,12 +356,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/server.go b/spindle/server.go index 4895a1fb..9a699aa8 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -468,9 +468,6 @@ func (s *Spindle) Router() http.Handler { return mux } - mux.HandleFunc("/events", s.Events) - mux.HandleFunc("/logs/{knot}/{rkey}/{name}", s.Logs) - // on a mill host, executors dial in here (plain ws, shared-secret auth) if s.mill != nil { mux.HandleFunc("/mill", s.mill.HandleExecutorConn) @@ -728,8 +725,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(repoDid, tpl, pipelineId, sourceRepo) return pipelineId, err diff --git a/spindle/stream.go b/spindle/stream.go deleted file mode 100644 index e8f2ca0c..00000000 --- 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 bfe9d90f..4bfeb52d 100644 --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -601,7 +601,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/xrpc/ci_pipeline_subscribe_logs.go b/spindle/xrpc/ci_pipeline_subscribe_logs.go index c839fa74..3627b645 100644 --- a/spindle/xrpc/ci_pipeline_subscribe_logs.go +++ b/spindle/xrpc/ci_pipeline_subscribe_logs.go @@ -39,33 +39,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.Error("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.Error("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.Error("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 { @@ -162,7 +143,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) @@ -183,7 +164,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 3801d44c..5ec26e6e 100644 --- a/spindle/xrpc/pipeline_cancel_pipeline.go +++ b/spindle/xrpc/pipeline_cancel_pipeline.go @@ -56,7 +56,7 @@ func (x *Xrpc) CancelPipeline(w http.ResponseWriter, r *http.Request) { fail(xrpcerr.GenericError(fmt.Errorf("failed to get pipeline: %w", err))) return } - if p.Repo == nil || *p.Repo != repoDid.String() { + if p.Repo != repoDid.String() { fail(xrpcerr.AccessControlError(actorDid.String())) return } @@ -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 fa2004e7..8255ff66 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