diff --git a/spindle/db/db.go b/spindle/db/db.go index bc508e5be..96ad17e8a 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -195,6 +195,8 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { create table if not exists executor_pending_artifacts ( lease_id text primary key, + knot text not null default '', + rkey text not null default '', workflow text not null, status text not null, error text not null default '', @@ -730,6 +732,29 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } + if err := orm.RunMigration(conn, logger, "pending-artifact-workflow-identity", func(tx *sql.Tx) error { + for _, column := range []string{"knot", "rkey"} { + var present int + if err := tx.QueryRow( + `select count(*) from pragma_table_info('executor_pending_artifacts') where name = ?`, + column, + ).Scan(&present); err != nil { + return err + } + if present != 0 { + continue + } + if _, err := tx.Exec( + `alter table executor_pending_artifacts add column ` + column + ` text not null default ''`, + ); err != nil { + return err + } + } + return nil + }); err != nil { + return err + } + if err := orm.RunMigration(conn, logger, "pending-artifact-terminal-metric-authority", func(tx *sql.Tx) error { var present int if err := tx.QueryRow( diff --git a/spindle/db/mill_state.go b/spindle/db/mill_state.go index 32fba498d..7009a5376 100644 --- a/spindle/db/mill_state.go +++ b/spindle/db/mill_state.go @@ -344,6 +344,8 @@ func (tx *EventBatchTx) InsertArtifactRef(leaseID, repoDid string, wid models.Wo type PendingArtifact struct { LeaseID string + Knot string + Rkey string Workflow string Status string Error string @@ -356,17 +358,21 @@ type PendingArtifact struct { } func (d *DB) SavePendingArtifact( - leaseID, workflow, status, errStr string, + leaseID string, + wid models.WorkflowId, + status, errStr string, exitCode int64, ref, hash, failureClass, failureReason string, millRecordsTerminalMetrics bool, ) error { _, err := d.Exec( `insert into executor_pending_artifacts ( - lease_id, workflow, status, error, exit_code, ref, hash, + lease_id, knot, rkey, workflow, status, error, exit_code, ref, hash, failure_class, failure_reason, mill_records_terminal_metrics - ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) on conflict(lease_id) do update set + knot = excluded.knot, + rkey = excluded.rkey, workflow = excluded.workflow, status = excluded.status, error = excluded.error, @@ -376,7 +382,7 @@ func (d *DB) SavePendingArtifact( failure_class = excluded.failure_class, failure_reason = excluded.failure_reason, mill_records_terminal_metrics = excluded.mill_records_terminal_metrics`, - leaseID, workflow, status, errStr, exitCode, ref, hash, + leaseID, wid.Knot, wid.Rkey, wid.Name, status, errStr, exitCode, ref, hash, failureClass, failureReason, millRecordsTerminalMetrics, ) return err @@ -389,7 +395,7 @@ func (d *DB) RemovePendingArtifact(leaseID string) error { func (d *DB) ListPendingArtifacts() ([]PendingArtifact, error) { rows, err := d.Query(` - select lease_id, workflow, status, error, exit_code, ref, hash, + select lease_id, knot, rkey, workflow, status, error, exit_code, ref, hash, failure_class, failure_reason, mill_records_terminal_metrics from executor_pending_artifacts `) @@ -402,7 +408,7 @@ func (d *DB) ListPendingArtifacts() ([]PendingArtifact, error) { for rows.Next() { var p PendingArtifact if err := rows.Scan( - &p.LeaseID, &p.Workflow, &p.Status, &p.Error, &p.ExitCode, &p.Ref, &p.Hash, + &p.LeaseID, &p.Knot, &p.Rkey, &p.Workflow, &p.Status, &p.Error, &p.ExitCode, &p.Ref, &p.Hash, &p.FailureClass, &p.FailureReason, &p.MillRecordsTerminalMetrics, ); err != nil { return nil, err diff --git a/spindle/db/mill_state_test.go b/spindle/db/mill_state_test.go index d11339f62..39adb1ddd 100644 --- a/spindle/db/mill_state_test.go +++ b/spindle/db/mill_state_test.go @@ -2,12 +2,14 @@ package db import ( "context" + "database/sql" "fmt" "path/filepath" "sync" "testing" "tangled.org/core/notifier" + "tangled.org/core/spindle/models" ) func TestMillLeaseRoundTrip(t *testing.T) { @@ -489,16 +491,71 @@ func TestExecutorCursorResetHelpers(t *testing.T) { } } +func TestPendingArtifactWorkflowIdentityMigration(t *testing.T) { + path := filepath.Join(t.TempDir(), "spindle.db") + legacy, err := sql.Open("sqlite3", path) + if err != nil { + t.Fatal(err) + } + if _, err := legacy.Exec(` + create table executor_pending_artifacts ( + lease_id text primary key, + workflow text not null, + status text not null, + error text not null default '', + exit_code integer not null default 0, + ref text not null, + hash text not null, + failure_class text not null default '', + failure_reason text not null default '', + mill_records_terminal_metrics integer not null default 0 + ); + insert into executor_pending_artifacts (lease_id, workflow, status, ref, hash) + values ('lease-1', 'build', 'success', 'logs/lease-1.log', 'sha256:test'); + `); err != nil { + legacy.Close() + t.Fatal(err) + } + if err := legacy.Close(); err != nil { + t.Fatal(err) + } + + d, err := Make(context.Background(), path) + if err != nil { + t.Fatal(err) + } + defer d.Close() + rows, err := d.ListPendingArtifacts() + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 || rows[0].Knot != "" || rows[0].Rkey != "" || rows[0].Workflow != "build" { + t.Fatalf("migrated pending artifact = %+v", rows) + } + var identityColumns int + if err := d.QueryRow(` + select count(*) from pragma_table_info('executor_pending_artifacts') + where name in ('knot', 'rkey') + `).Scan(&identityColumns); err != nil { + t.Fatal(err) + } + if identityColumns != 2 { + t.Fatalf("pending artifact identity columns = %d, want 2", identityColumns) + } +} + func TestClearPendingArtifacts(t *testing.T) { d := newTestDB(t) - if err := d.SavePendingArtifact("lease-1", "build", "success", "", 0, "ref", "sha256:x", "none", "success", true); err != nil { + wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "knot.example", Rkey: "rkey1"}, Name: "build"} + if err := d.SavePendingArtifact("lease-1", wid, "success", "", 0, "ref", "sha256:x", "none", "success", true); err != nil { t.Fatalf("SavePendingArtifact: %v", err) } rows, err := d.ListPendingArtifacts() if err != nil { t.Fatal(err) } - if len(rows) != 1 || rows[0].FailureClass != "none" || rows[0].FailureReason != "success" || !rows[0].MillRecordsTerminalMetrics { + if len(rows) != 1 || rows[0].Knot != wid.Knot || rows[0].Rkey != wid.Rkey || rows[0].Workflow != wid.Name || + rows[0].FailureClass != "none" || rows[0].FailureReason != "success" || !rows[0].MillRecordsTerminalMetrics { t.Fatalf("pending artifact attribution = %+v", rows) } if err := d.ClearPendingArtifacts(); err != nil { diff --git a/spindle/mill/executor/executor.go b/spindle/mill/executor/executor.go index 051d08182..e96804394 100644 --- a/spindle/mill/executor/executor.go +++ b/spindle/mill/executor/executor.go @@ -155,6 +155,15 @@ func New(cfg *config.Config, engines map[string]models.Engine, d *db.DB, n *noti func (e *Executor) Connect(ctx context.Context) { e.lifecycleCtx = ctx + recoveryDone := make(chan struct{}) + go func() { + defer close(recoveryDone) + if err := e.recoverPendingArtifacts(ctx); err != nil && ctx.Err() == nil { + e.l.Warn("recover pending artifacts failed", "err", err) + } + }() + defer func() { <-recoveryDone }() + sub := e.n.Subscribe() cursor, err := e.db.EventHighWater() if err != nil { diff --git a/spindle/mill/executor/observe.go b/spindle/mill/executor/observe.go index b4f047f01..c21f89af0 100644 --- a/spindle/mill/executor/observe.go +++ b/spindle/mill/executor/observe.go @@ -134,7 +134,7 @@ func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) error // persist pending artifact state so a restart can retry the upload if e.db != nil { _ = e.db.SavePendingArtifact( - res.leaseID, res.wid.Name, terminalStatus, errStr, exitCode, ref, hash, + res.leaseID, res.wid, terminalStatus, errStr, exitCode, ref, hash, failureClass, failureReason, millRecordsTerminalMetrics, ) } diff --git a/spindle/mill/executor/outbox.go b/spindle/mill/executor/outbox.go index 9938a6252..a01fca2db 100644 --- a/spindle/mill/executor/outbox.go +++ b/spindle/mill/executor/outbox.go @@ -30,6 +30,10 @@ func generateEpoch() string { } func (e *Executor) initOutbox() error { + return e.loadOutbox() +} + +func (e *Executor) loadOutbox() error { e.eventMu.Lock() defer e.eventMu.Unlock() @@ -62,7 +66,6 @@ func (e *Executor) initOutbox() error { if e.outboxBytes > 0 { e.outboxIdleCh = make(chan struct{}) } - _ = e.recoverPendingArtifacts() return nil } @@ -195,7 +198,7 @@ func (e *Executor) appendTerminalWithArtifact( return e.appendAndSend(leaseID, payload, true) } -func (e *Executor) recoverPendingArtifacts() error { +func (e *Executor) recoverPendingArtifacts(parent context.Context) error { if e.db == nil { return nil } @@ -203,25 +206,28 @@ func (e *Executor) recoverPendingArtifacts() error { if err != nil || len(pending) == 0 { return err } + ctx, cancel := context.WithTimeout(parent, 2*time.Minute) + defer cancel() for _, p := range pending { if p.Ref != "" { if e.writer == nil { continue } - ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) var logDir string if e.cfg != nil { logDir = e.cfg.Server.LogDir } - logPath := models.LogFilePath(logDir, models.WorkflowId{Name: p.Workflow}) + wid := models.WorkflowId{ + PipelineId: models.PipelineId{Knot: p.Knot, Rkey: p.Rkey}, + Name: p.Workflow, + } + logPath := models.LogFilePath(logDir, wid) f, openErr := os.Open(logPath) if openErr != nil { - cancel() continue } uploadErr := e.writer.Put(ctx, p.Ref, f) _ = f.Close() - cancel() if uploadErr != nil { continue } diff --git a/spindle/mill/executor/reserved_test.go b/spindle/mill/executor/reserved_test.go index 7ed1c8de1..efed26d76 100644 --- a/spindle/mill/executor/reserved_test.go +++ b/spindle/mill/executor/reserved_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "github.com/bluesky-social/indigo/atproto/syntax" "github.com/gorilla/websocket" "google.golang.org/protobuf/proto" @@ -11,6 +12,7 @@ import ( "log/slog" "net/http" "net/http/httptest" + "os" "path/filepath" "strings" "testing" @@ -46,6 +48,12 @@ type blockingEncoder struct { release chan struct{} } +type artifactWriterFunc func(context.Context, string, io.Reader) error + +func (f artifactWriterFunc) Put(ctx context.Context, ref string, r io.Reader) error { + return f(ctx, ref, r) +} + func (e *blockingEncoder) Encode(*millproto.Message) error { select { case e.started <- struct{}{}: @@ -350,6 +358,187 @@ func TestTerminalGuardSurvivesExecutorRestart(t *testing.T) { } } +func TestPendingArtifactRecoveryCompletesWithoutDeadlock(t *testing.T) { + d := testDB(t) + logDir := t.TempDir() + wid := models.WorkflowId{ + PipelineId: models.PipelineId{Knot: "knot.example", Rkey: "3abc"}, + Name: "build", + } + logPath := models.LogFilePath(logDir, wid) + if err := os.MkdirAll(filepath.Dir(logPath), 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(logPath, []byte("finished"), 0600); err != nil { + t.Fatal(err) + } + if err := d.SavePendingArtifact( + "lease-1", wid, string(models.StatusKindSuccess), "", 0, + "logs/lease-1.log", "sha256:test", string(engine.FailureClassNone), + string(engine.FailureReasonSuccess), true, + ); err != nil { + t.Fatal(err) + } + + uploaded := make(chan string, 1) + e := &Executor{ + db: d, + cfg: &config.Config{Server: config.Server{LogDir: logDir}}, + l: slog.New(slog.NewTextHandler(io.Discard, nil)), + active: make(map[string]*reservation), + writer: artifactWriterFunc(func(_ context.Context, ref string, r io.Reader) error { + if _, err := io.ReadAll(r); err != nil { + return err + } + uploaded <- ref + return nil + }), + maxOutboxBytes: 10 * 1024 * 1024, + } + if err := e.initOutbox(); err != nil { + t.Fatalf("initOutbox: %v", err) + } + done := make(chan error, 1) + go func() { done <- e.recoverPendingArtifacts(context.Background()) }() + select { + case err := <-done: + if err != nil { + t.Fatalf("recoverPendingArtifacts: %v", err) + } + case <-time.After(time.Second): + t.Fatal("pending artifact recovery deadlocked") + } + select { + case ref := <-uploaded: + if ref != "logs/lease-1.log" { + t.Fatalf("uploaded ref = %q", ref) + } + default: + t.Fatal("pending artifact was not uploaded") + } + pending, err := d.ListPendingArtifacts() + if err != nil { + t.Fatal(err) + } + if len(pending) != 0 { + t.Fatalf("pending artifacts after recovery = %d, want 0", len(pending)) + } + rows, err := d.ListOutboxRows() + if err != nil { + t.Fatal(err) + } + if len(rows) != 1 { + t.Fatalf("outbox rows after recovery = %d, want 1", len(rows)) + } +} + +func TestPendingArtifactRecoveryKeepsRowWhenLogIsMissing(t *testing.T) { + d := testDB(t) + wid := models.WorkflowId{ + PipelineId: models.PipelineId{Knot: "knot.example", Rkey: "3abc"}, + Name: "build", + } + if err := d.SavePendingArtifact( + "lease-1", wid, string(models.StatusKindSuccess), "", 0, + "logs/lease-1.log", "sha256:test", string(engine.FailureClassNone), + string(engine.FailureReasonSuccess), true, + ); err != nil { + t.Fatal(err) + } + e := &Executor{ + db: d, + cfg: &config.Config{Server: config.Server{LogDir: t.TempDir()}}, + l: slog.New(slog.NewTextHandler(io.Discard, nil)), + active: make(map[string]*reservation), + writer: artifactWriterFunc(func(context.Context, string, io.Reader) error { + t.Fatal("writer called without a log file") + return nil + }), + maxOutboxBytes: 10 * 1024 * 1024, + } + if err := e.initOutbox(); err != nil { + t.Fatalf("initOutbox: %v", err) + } + if err := e.recoverPendingArtifacts(context.Background()); err != nil { + t.Fatalf("recoverPendingArtifacts: %v", err) + } + pending, err := d.ListPendingArtifacts() + if err != nil { + t.Fatal(err) + } + if len(pending) != 1 { + t.Fatalf("pending artifacts with missing log = %d, want 1", len(pending)) + } + rows, err := d.ListOutboxRows() + if err != nil { + t.Fatal(err) + } + if len(rows) != 0 { + t.Fatalf("outbox rows with missing log = %d, want 0", len(rows)) + } +} + +func TestPendingArtifactRecoveryUsesAggregateDeadline(t *testing.T) { + d := testDB(t) + logDir := t.TempDir() + wids := []models.WorkflowId{ + {PipelineId: models.PipelineId{Knot: "knot.example", Rkey: "3abc"}, Name: "build"}, + {PipelineId: models.PipelineId{Knot: "knot.example", Rkey: "3def"}, Name: "test"}, + } + for i, wid := range wids { + logPath := models.LogFilePath(logDir, wid) + if err := os.MkdirAll(filepath.Dir(logPath), 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(logPath, []byte("finished"), 0600); err != nil { + t.Fatal(err) + } + leaseID := fmt.Sprintf("lease-%d", i+1) + if err := d.SavePendingArtifact( + leaseID, wid, string(models.StatusKindSuccess), "", 0, + "logs/"+leaseID+".log", "sha256:test", string(engine.FailureClassNone), + string(engine.FailureReasonSuccess), true, + ); err != nil { + t.Fatal(err) + } + } + + var deadlines []time.Time + e := &Executor{ + db: d, + cfg: &config.Config{Server: config.Server{LogDir: logDir}}, + l: slog.New(slog.NewTextHandler(io.Discard, nil)), + active: make(map[string]*reservation), + writer: artifactWriterFunc(func(ctx context.Context, _ string, _ io.Reader) error { + deadline, ok := ctx.Deadline() + if !ok { + t.Fatal("artifact recovery context has no deadline") + } + deadlines = append(deadlines, deadline) + return errors.New("unavailable") + }), + maxOutboxBytes: 10 * 1024 * 1024, + } + if err := e.initOutbox(); err != nil { + t.Fatalf("initOutbox: %v", err) + } + started := time.Now() + if err := e.recoverPendingArtifacts(context.Background()); err != nil { + t.Fatalf("recoverPendingArtifacts: %v", err) + } + finished := time.Now() + if len(deadlines) != len(wids) { + t.Fatalf("upload attempts = %d, want %d", len(deadlines), len(wids)) + } + if !deadlines[0].Equal(deadlines[1]) { + t.Fatalf("artifact recovery deadlines differ: %v, %v", deadlines[0], deadlines[1]) + } + wantTimeout := 2 * time.Minute + if deadlines[0].Before(started.Add(wantTimeout)) || deadlines[0].After(finished.Add(wantTimeout)) { + t.Fatalf("artifact recovery deadline = %v, want %v after startup", deadlines[0], wantTimeout) + } +} + func TestHandleCommitPreservesPreauthorizedSecrets(t *testing.T) { d := testDB(t) n := notifier.New() @@ -410,6 +599,110 @@ func TestHandleCommitPreservesPreauthorizedSecrets(t *testing.T) { e.jobsWG.Wait() } +func TestConnectHandshakesWhilePendingArtifactRecoveryIsBlocked(t *testing.T) { + handshake := make(chan error, 1) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Upgrade(w, r, nil, 1024, 1024) + if err != nil { + handshake <- err + return + } + defer conn.Close() + stream := millproto.NewWSStream(conn) + dec := millproto.NewDecoder(stream) + enc := millproto.NewEncoder(stream) + helloMsg, err := dec.Decode() + if err != nil { + handshake <- err + return + } + hello := helloMsg.GetHello() + if hello == nil { + handshake <- errors.New("first executor message was not Hello") + return + } + if err := enc.Encode(&millproto.Message{Resume: &millv1.Resume{Epoch: hello.GetEpoch()}}); err != nil { + handshake <- err + return + } + snapshotMsg, err := dec.Decode() + if err != nil { + handshake <- err + return + } + if snapshotMsg.GetNodeSnapshot() == nil { + handshake <- errors.New("executor did not send NodeSnapshot after Resume") + return + } + handshake <- nil + for { + if _, err := dec.Decode(); err != nil { + return + } + } + })) + t.Cleanup(srv.Close) + + e := testSessionExecutor(t, "ws"+strings.TrimPrefix(srv.URL, "http")) + n := notifier.New() + e.n = &n + logDir := t.TempDir() + e.cfg.Server.LogDir = logDir + wid := models.WorkflowId{ + PipelineId: models.PipelineId{Knot: "knot.example", Rkey: "3abc"}, + Name: "build", + } + logPath := models.LogFilePath(logDir, wid) + if err := os.MkdirAll(filepath.Dir(logPath), 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(logPath, []byte("finished"), 0600); err != nil { + t.Fatal(err) + } + if err := e.db.SavePendingArtifact( + "lease-1", wid, string(models.StatusKindSuccess), "", 0, + "logs/lease-1.log", "sha256:test", string(engine.FailureClassNone), + string(engine.FailureReasonSuccess), true, + ); err != nil { + t.Fatal(err) + } + recoveryStarted := make(chan struct{}) + e.writer = artifactWriterFunc(func(ctx context.Context, _ string, _ io.Reader) error { + close(recoveryStarted) + <-ctx.Done() + return ctx.Err() + }) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + e.Connect(ctx) + close(done) + }() + select { + case <-recoveryStarted: + case <-time.After(time.Second): + cancel() + t.Fatal("pending artifact recovery did not start") + } + select { + case err := <-handshake: + if err != nil { + cancel() + t.Fatalf("mill handshake: %v", err) + } + case <-time.After(time.Second): + cancel() + t.Fatal("blocked artifact recovery delayed the mill handshake") + } + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("Connect did not stop after cancellation") + } +} + func TestRunSessionCancellationClosesStalledWebsocket(t *testing.T) { connected := make(chan struct{}) release := make(chan struct{})