diff --git a/spindle/artifactstore/artifactstore.go b/spindle/artifactstore/artifactstore.go index fcb1de0c..855a6c51 100644 --- a/spindle/artifactstore/artifactstore.go +++ b/spindle/artifactstore/artifactstore.go @@ -23,9 +23,11 @@ type Reader interface { Open(ctx context.Context, ref string) (io.ReadCloser, error) } +// missing refs are not errors type Store interface { Writer Reader + Delete(ctx context.Context, ref string) error } type DiskStore struct { @@ -83,13 +85,25 @@ func (s *DiskStore) Open(_ context.Context, ref string) (io.ReadCloser, error) { return f, nil } +func (s *DiskStore) Delete(_ context.Context, ref string) error { + path, err := s.resolve(ref) + if err != nil { + return err + } + if err := os.Remove(path); err != nil && !os.IsNotExist(err) { + return fmt.Errorf("delete disk artifact %q: %w", path, err) + } + return nil +} + func (s *DiskStore) resolve(ref string) (string, error) { if ref == "" || filepath.IsAbs(ref) { return "", fmt.Errorf("invalid disk artifact ref %q", ref) } path := filepath.Join(s.root, filepath.Clean(ref)) rel, err := filepath.Rel(s.root, path) - if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { + // reject the store root + if err != nil || rel == "." || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { return "", fmt.Errorf("artifact ref %q escapes disk root %q", ref, s.root) } return path, nil @@ -98,6 +112,7 @@ func (s *DiskStore) resolve(ref string) (string, error) { type s3API interface { PutObject(ctx context.Context, params *s3.PutObjectInput, optFns ...func(*s3.Options)) (*s3.PutObjectOutput, error) GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error) + DeleteObject(ctx context.Context, params *s3.DeleteObjectInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) } type S3Store struct { @@ -144,6 +159,20 @@ func (s *S3Store) Open(ctx context.Context, ref string) (io.ReadCloser, error) { return res.Body, nil } +func (s *S3Store) Delete(ctx context.Context, ref string) error { + if err := validateObjectRef(ref); err != nil { + return err + } + _, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{ + Bucket: aws.String(s.bucket), + Key: aws.String(ref), + }) + if err != nil { + return fmt.Errorf("s3 delete object: %w", err) + } + return nil +} + func validateObjectRef(ref string) error { if ref == "" || strings.HasPrefix(ref, "/") || strings.Contains(ref, "://") { return fmt.Errorf("invalid artifact ref %q", ref) @@ -238,3 +267,13 @@ func (s *Stores) PutFile(ctx context.Context, ref, sourcePath string) []error { } return errs } + +func (s *Stores) Delete(ctx context.Context, ref string) []error { + var errs []error + for _, name := range s.order { + if err := s.stores[name].Delete(ctx, ref); err != nil { + errs = append(errs, fmt.Errorf("%s: %w", name, err)) + } + } + return errs +} diff --git a/spindle/artifactstore/artifactstore_test.go b/spindle/artifactstore/artifactstore_test.go index a16479e0..57b58690 100644 --- a/spindle/artifactstore/artifactstore_test.go +++ b/spindle/artifactstore/artifactstore_test.go @@ -35,6 +35,14 @@ func (m *mockS3Client) PutObject(ctx context.Context, params *s3.PutObjectInput, return &s3.PutObjectOutput{}, nil } +func (m *mockS3Client) DeleteObject(ctx context.Context, params *s3.DeleteObjectInput, optFns ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) { + m.mu.Lock() + defer m.mu.Unlock() + key := *params.Bucket + "/" + *params.Key + delete(m.store, key) + return &s3.DeleteObjectOutput{}, nil +} + func (m *mockS3Client) GetObject(ctx context.Context, params *s3.GetObjectInput, optFns ...func(*s3.Options)) (*s3.GetObjectOutput, error) { m.mu.Lock() defer m.mu.Unlock() @@ -128,3 +136,47 @@ func TestS3Store(t *testing.T) { t.Fatalf("got content %q, want %q", string(got), content) } } + +func TestStoreDelete(t *testing.T) { + ctx := context.Background() + + t.Run("disk", func(t *testing.T) { + store, err := NewDiskStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err := store.Put(ctx, "logs/l1.log", strings.NewReader("data")); err != nil { + t.Fatal(err) + } + if err := store.Delete(ctx, "logs/l1.log"); err != nil { + t.Fatal(err) + } + if _, err := store.Open(ctx, "logs/l1.log"); err == nil { + t.Fatal("expected open to fail after delete") + } + // deleting a missing ref is fine + if err := store.Delete(ctx, "logs/l1.log"); err != nil { + t.Fatal(err) + } + }) + + t.Run("s3", func(t *testing.T) { + mock := newMockS3Client() + store, err := NewS3Store(mock, "bucket") + if err != nil { + t.Fatal(err) + } + if err := store.Put(ctx, "logs/l1.log", strings.NewReader("data")); err != nil { + t.Fatal(err) + } + if err := store.Delete(ctx, "logs/l1.log"); err != nil { + t.Fatal(err) + } + if _, err := store.Open(ctx, "logs/l1.log"); err == nil { + t.Fatal("expected open to fail after delete") + } + if err := store.Delete(ctx, "logs/l1.log"); err != nil { + t.Fatal(err) + } + }) +} diff --git a/spindle/db/artifacts.go b/spindle/db/artifacts.go index 059da9da..599cff81 100644 --- a/spindle/db/artifacts.go +++ b/spindle/db/artifacts.go @@ -22,11 +22,11 @@ func (d *DB) GetFinishedLog(workflow string) (*FinishedLog, error) { return &fl, nil } -func (d *DB) SaveArtifactRef(leaseID, workflow, ref, hash string) error { +func (d *DB) SaveArtifactRef(leaseID, repoDid, workflow, ref, hash string) error { _, err := d.Exec( - `insert into mill_artifacts (lease_id, workflow, ref, hash) - values (?, ?, ?, ?)`, - leaseID, workflow, ref, hash, + `insert into mill_artifacts (lease_id, repo_did, workflow, ref, hash) + values (?, ?, ?, ?, ?)`, + leaseID, repoDid, workflow, ref, hash, ) return err } diff --git a/spindle/db/bans.go b/spindle/db/bans.go new file mode 100644 index 00000000..b279f23f --- /dev/null +++ b/spindle/db/bans.go @@ -0,0 +1,79 @@ +package db + +import ( + "database/sql" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +// bans match either the repo or owner did +type BanEntry struct { + SubjectDid syntax.DID + CreatedAt string +} + +func (d *DB) PutBan(ban BanEntry) error { + _, err := d.Exec( + `insert into bans (subject_did) values (?) on conflict(subject_did) do nothing`, + ban.SubjectDid.String(), + ) + return err +} + +func (d *DB) DeleteBan(subject syntax.DID) (bool, error) { + res, err := d.Exec(`delete from bans where subject_did = ?`, subject.String()) + if err != nil { + return false, err + } + n, err := res.RowsAffected() + return n > 0, err +} + +func (d *DB) GetBan(subject syntax.DID) (*BanEntry, error) { + row := d.QueryRow( + `select subject_did, created_at from bans where subject_did = ?`, + subject.String(), + ) + return scanBan(row) +} + +func (d *DB) BanList() ([]BanEntry, error) { + rows, err := d.Query(`select subject_did, created_at from bans order by subject_did`) + if err != nil { + return nil, err + } + defer rows.Close() + + var bans []BanEntry + for rows.Next() { + b, err := scanBan(rows) + if err != nil { + return nil, err + } + bans = append(bans, *b) + } + return bans, rows.Err() +} + +// check both roles for a preemptive ban +func (d *DB) IsBanned(repoDid, ownerDid syntax.DID) (*BanEntry, error) { + row := d.QueryRow( + `select subject_did, created_at from bans + where subject_did in (?, ?) limit 1`, + repoDid.String(), ownerDid.String(), + ) + return scanBan(row) +} + +func scanBan(row interface{ Scan(...any) error }) (*BanEntry, error) { + var b BanEntry + var subject string + if err := row.Scan(&subject, &b.CreatedAt); err != nil { + if err == sql.ErrNoRows { + return nil, nil + } + return nil, err + } + b.SubjectDid = syntax.DID(subject) + return &b, nil +} diff --git a/spindle/db/bans_test.go b/spindle/db/bans_test.go new file mode 100644 index 00000000..501da07e --- /dev/null +++ b/spindle/db/bans_test.go @@ -0,0 +1,84 @@ +package db + +import ( + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" +) + +func TestBans_RoundTrip(t *testing.T) { + d := newTestDB(t) + repo := syntax.DID("did:plc:repo1") + + b, err := d.GetBan(repo) + if err != nil || b != nil { + t.Fatalf("GetBan on empty store: %v, %v", b, err) + } + + if err := d.PutBan(BanEntry{SubjectDid: repo}); err != nil { + t.Fatalf("PutBan: %v", err) + } + b, err = d.GetBan(repo) + if err != nil || b == nil || b.CreatedAt == "" { + t.Fatalf("GetBan after put: %+v, %v", b, err) + } + + first := *b + if err := d.PutBan(BanEntry{SubjectDid: repo}); err != nil { + t.Fatalf("PutBan idempotent: %v", err) + } + b, _ = d.GetBan(repo) + if b.CreatedAt != first.CreatedAt { + t.Fatalf("re-ban clobbered timestamp: %+v", b) + } + + ok, err := d.DeleteBan(repo) + if err != nil || !ok { + t.Fatalf("DeleteBan present: %v, %v", ok, err) + } + ok, err = d.DeleteBan(repo) + if err != nil || ok { + t.Fatalf("DeleteBan absent should report false: %v, %v", ok, err) + } +} + +func TestBans_EffectiveBan(t *testing.T) { + d := newTestDB(t) + repo := syntax.DID("did:plc:repo2") + owner := syntax.DID("did:plc:owner2") + + if err := d.PutBan(BanEntry{SubjectDid: owner}); err != nil { + t.Fatal(err) + } + b, _ := d.IsBanned(repo, owner) + if b == nil || b.SubjectDid != owner { + t.Fatalf("IsBanned should find the owner ban: %+v", b) + } + // a ban matches either role + b, _ = d.IsBanned(owner, "did:plc:someoneelse") + if b == nil || b.SubjectDid != owner { + t.Fatalf("IsBanned should find the ban under the repo role too: %+v", b) + } + + b, _ = d.IsBanned("did:plc:cleanrepo", "did:plc:cleanowner") + if b != nil { + t.Fatalf("IsBanned false positive: %+v", b) + } +} + +func TestBans_List(t *testing.T) { + d := newTestDB(t) + if err := d.PutBan(BanEntry{SubjectDid: "did:plc:b"}); err != nil { + t.Fatal(err) + } + if err := d.PutBan(BanEntry{SubjectDid: "did:plc:a"}); err != nil { + t.Fatal(err) + } + bans, err := d.BanList() + if err != nil { + t.Fatal(err) + } + if len(bans) != 2 || bans[0].SubjectDid != "did:plc:a" || bans[1].SubjectDid != "did:plc:b" { + t.Fatalf("BanList order/content: %+v", bans) + } +} diff --git a/spindle/db/db.go b/spindle/db/db.go index 764a98eb..cdd50211 100644 --- a/spindle/db/db.go +++ b/spindle/db/db.go @@ -183,6 +183,7 @@ func Make(ctx context.Context, dbPath string) (*DB, error) { create table if not exists mill_artifacts ( id integer primary key autoincrement, lease_id text not null, + repo_did text not null, workflow text not null, ref text not null, hash text not null @@ -585,6 +586,18 @@ func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error return err } + if err := orm.RunMigration(conn, logger, "bans-schema", func(tx *sql.Tx) error { + _, err := tx.Exec(` + create table if not exists bans ( + subject_did text primary key, + created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) + ); + `) + return err + }); err != nil { + return err + } + return nil } diff --git a/spindle/db/delete_test.go b/spindle/db/delete_test.go new file mode 100644 index 00000000..7bc29821 --- /dev/null +++ b/spindle/db/delete_test.go @@ -0,0 +1,209 @@ +package db + +import ( + "context" + "fmt" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/api/tangled" + "tangled.org/core/spindle/models" +) + +func TestDeleteReposByDid(t *testing.T) { + d := newTestDB(t) + repoDid := syntax.DID("did:plc:repo1") + other := syntax.DID("did:plc:repo2") + + for _, r := range []Repo{ + {Knot: "k1", Owner: syntax.DID("did:plc:o1"), Rkey: syntax.RecordKey("a"), RepoDid: repoDid}, + {Knot: "k2", Owner: syntax.DID("did:plc:o1"), Rkey: syntax.RecordKey("b"), RepoDid: repoDid}, + {Knot: "k1", Owner: syntax.DID("did:plc:o2"), Rkey: syntax.RecordKey("c"), RepoDid: other}, + } { + if err := d.AddRepo(r); err != nil { + t.Fatal(err) + } + } + + repos, err := d.ReposByDid(repoDid) + if err != nil { + t.Fatal(err) + } + if len(repos) != 2 { + t.Fatalf("expected 2 sibling rows, got %d", len(repos)) + } + + if err := d.DeleteReposByDid(repoDid); err != nil { + t.Fatal(err) + } + if repos, _ := d.ReposByDid(repoDid); len(repos) != 0 { + t.Fatalf("expected rows gone, got %d", len(repos)) + } + if repos, _ := d.ReposByDid(other); len(repos) != 1 { + t.Fatalf("expected other repo untouched, got %d", len(repos)) + } +} + +func TestDeleteJobsByRepo(t *testing.T) { + d := newTestDB(t) + ctx := context.Background() + pid := models.PipelineId{Knot: "k1", Rkey: "p1"} + if err := d.EnqueueJob(ctx, "did:plc:repo1", pid, nil, tangled.Pipeline{}, "", ""); err != nil { + t.Fatal(err) + } + if err := d.EnqueueJob(ctx, "did:plc:repo2", pid, nil, tangled.Pipeline{}, "", ""); err != nil { + t.Fatal(err) + } + + if err := d.DeleteJobsByRepo(ctx, "did:plc:repo1"); err != nil { + t.Fatal(err) + } + var n int + if err := d.QueryRow(`select count(*) from jobs`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("expected 1 job left, got %d", n) + } + if err := d.QueryRow(`select count(*) from jobs where repo_did = 'did:plc:repo2'`).Scan(&n); err != nil || n != 1 { + t.Fatalf("expected surviving job to be repo2's, count=%d err=%v", n, err) + } +} + +func TestDeleteMillLeasesByRepo(t *testing.T) { + d := newTestDB(t) + for _, l := range []MillLease{ + {LeaseID: "l1", NodeID: "n1", Epoch: "e1", Engine: "microvm", Knot: "k1", Rkey: "r1", Workflow: "w1", State: "active", RepoDID: "did:plc:repo1"}, + {LeaseID: "l2", NodeID: "n1", Epoch: "e1", Engine: "microvm", Knot: "k1", Rkey: "r1", Workflow: "w2", State: "active", RepoDID: "did:plc:repo1"}, + {LeaseID: "l3", NodeID: "n1", Epoch: "e1", Engine: "microvm", Knot: "k1", Rkey: "r9", Workflow: "w1", State: "active", RepoDID: "did:plc:repo2"}, + } { + if err := d.SaveMillLease(l); err != nil { + t.Fatal(err) + } + if _, err := d.Exec(`insert into mill_artifacts (lease_id, repo_did, workflow, ref, hash) values (?, ?, ?, ?, ?)`, l.LeaseID, l.RepoDID, l.Workflow, "logs/"+l.LeaseID+".log", "h"); err != nil { + t.Fatal(err) + } + if _, err := d.Exec(`insert into executor_pending_artifacts (lease_id, workflow, status, ref, hash) values (?, ?, 'done', 'r', 'h')`, l.LeaseID, l.Workflow); err != nil { + t.Fatal(err) + } + } + + removed, err := d.DeleteMillLeasesByRepo("did:plc:repo1") + if err != nil { + t.Fatal(err) + } + if len(removed) != 2 { + t.Fatalf("expected 2 lease ids, got %v", removed) + } + + leases, err := d.ListMillLeases() + if err != nil { + t.Fatal(err) + } + if len(leases) != 1 || leases[0].LeaseID != "l3" { + t.Fatalf("expected only l3 left, got %+v", leases) + } + if err := d.DeleteArtifactRefsByRepo("did:plc:repo1"); err != nil { + t.Fatal(err) + } + var n int + if err := d.QueryRow(`select count(*) from mill_artifacts`).Scan(&n); err != nil || n != 1 { + t.Fatalf("expected 1 artifact row left, got %d err %v", n, err) + } + if err := d.QueryRow(`select count(*) from executor_pending_artifacts`).Scan(&n); err != nil || n != 1 { + t.Fatalf("expected 1 pending artifact row left, got %d err %v", n, err) + } +} + +func TestDeleteQuotaStateForRepo(t *testing.T) { + d := newTestDB(t) + ctx := context.Background() + repo := "did:plc:repo1" + other := "did:plc:repo2" + + ins := []string{ + fmt.Sprintf(`insert into quota_reservations (id, resource, kind, key, amount, repo_did, owner_did, phase, created_at) values ('r1', 'compute', 'generic', 'k', 1, '%s', 'did:plc:o1', 'active', 1)`, repo), + fmt.Sprintf(`insert into quota_reservations (id, resource, kind, key, amount, repo_did, owner_did, phase, created_at) values ('r2', 'compute', 'generic', 'k', 1, '%s', 'did:plc:o1', 'active', 1)`, other), + fmt.Sprintf(`insert into quota_allocations (repo_did, resource, kind, key, amount) values ('%s', 'compute', 'generic', 'k', 2)`, repo), + fmt.Sprintf(`insert into quota_allocations (repo_did, resource, kind, key, amount) values ('%s', 'compute', 'generic', 'k', 2)`, other), + fmt.Sprintf(`insert into quota_limits (did, resource, max_amount) values ('%s', 'compute', 5)`, repo), + `insert into quota_limits (did, resource, max_amount) values ('did:plc:o1', 'compute', 5)`, + fmt.Sprintf(`insert into quota_repo_owners (repo_did, owner_did) values ('%s', 'did:plc:o1')`, repo), + } + for _, q := range ins { + if _, err := d.Exec(q); err != nil { + t.Fatalf("fixture %q: %v", q, err) + } + } + + if err := d.DeleteQuotaStateForRepo(ctx, repo); err != nil { + t.Fatal(err) + } + + checks := []struct { + q, want string + }{ + {`select id from quota_reservations`, "r2"}, + {`select repo_did from quota_allocations`, other}, + {`select did from quota_limits`, "did:plc:o1"}, + } + for _, c := range checks { + var got string + if err := d.QueryRow(c.q).Scan(&got); err != nil { + t.Fatalf("%s: %v", c.q, err) + } + if got != c.want { + t.Fatalf("%s: surviving row = %s, want %s", c.q, got, c.want) + } + } + var n int + if err := d.QueryRow(`select count(*) from quota_repo_owners`).Scan(&n); err != nil || n != 0 { + t.Fatalf("expected repo owner rows gone, got %d err %v", n, err) + } +} + +func TestDeleteEventsByRepo(t *testing.T) { + d := newTestDB(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}`}, + } + 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 { + t.Fatal(err) + } + } + + if err := d.DeleteEventsByRepo(repo, []PipelineKey{{Knot: "k7", Rkey: "p9"}}); err != nil { + t.Fatal(err) + } + + 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 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 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) + } +} diff --git a/spindle/db/events.go b/spindle/db/events.go index efa72a75..cbd8c891 100644 --- a/spindle/db/events.go +++ b/spindle/db/events.go @@ -2,13 +2,13 @@ package db import ( "encoding/json" - "time" - + "fmt" "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 { @@ -165,3 +165,113 @@ func (d *DB) StatusSuccess(workflowId models.WorkflowId, n *notifier.Notifier) e func (d *DB) StatusTimeout(workflowId models.WorkflowId, n *notifier.Notifier) error { return d.createStatusEvent(workflowId, models.StatusKindTimeout, nil, nil, n) } + +type PipelineWorkflow struct { + Knot string + Rkey string + Name string +} + +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, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []PipelineWorkflow + for rows.Next() { + var rkey, raw string + if err := rows.Scan(&rkey, &raw); err != nil { + return nil, err + } + var p tangled.Pipeline + 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}) + } + } + return out, rows.Err() +} + +// PipelineKey is one exact pipeline at-uri identity +type PipelineKey struct { + Knot string + Rkey string +} + +func (d *DB) DeleteEventsByRepo(repoDid string, extra []PipelineKey) error { + tx, err := d.Begin() + if err != nil { + return err + } + 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, + ) + 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) + } + } + for rows.Next() { + var knot, rkey string + if err := rows.Scan(&knot, &rkey); err != nil { + rows.Close() + return err + } + add(knot, rkey) + } + if err := rows.Err(); err != nil { + rows.Close() + return err + } + rows.Close() + for _, k := range extra { + add(k.Knot, k.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 { + 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 { + return err + } + return tx.Commit() +} diff --git a/spindle/db/jobs.go b/spindle/db/jobs.go index b0ac90d2..9951d1f2 100644 --- a/spindle/db/jobs.go +++ b/spindle/db/jobs.go @@ -4,6 +4,8 @@ import ( "context" "database/sql" "encoding/json" + "errors" + "tangled.org/core/api/tangled" "tangled.org/core/spindle/models" ) @@ -19,16 +21,44 @@ type JobRow struct { Tracestate string } +// the atomic check rejects a banned job before admission +var ErrBannedSubject = errors.New("subject is banned") + func (d *DB) EnqueueJob(ctx context.Context, repoDid string, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo, tpl tangled.Pipeline, traceparent, tracestate string) error { tplJson, err := json.Marshal(tpl) if err != nil { return err } - _, err = d.ExecContext(ctx, ` + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + + var owner sql.NullString + if err := tx.QueryRowContext(ctx, `select owner from repos where repo_did = ?`, repoDid).Scan(&owner); err != nil && !errors.Is(err, sql.ErrNoRows) { + return err + } + var banned bool + if err := tx.QueryRowContext(ctx, ` + select exists( + select 1 from bans + where subject_did in (?, ?) + )`, repoDid, owner.String).Scan(&banned); err != nil { + return err + } + if banned { + return ErrBannedSubject + } + + _, err = tx.ExecContext(ctx, ` insert into jobs (repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl, traceparent, tracestate) values (?, ?, ?, ?, ?, ?, ?) `, repoDid, pipelineId.Knot, pipelineId.Rkey, string(sourceRepoJson(sourceRepo)), string(tplJson), traceparent, tracestate) - return err + if err != nil { + return err + } + return tx.Commit() } func (d *DB) DequeueJob(ctx context.Context) (*JobRow, error) { var row JobRow @@ -68,3 +98,8 @@ func sourceRepoJson(sr *tangled.Pipeline_TriggerRepo) []byte { b, _ := json.Marshal(sr) return b } + +func (d *DB) DeleteJobsByRepo(ctx context.Context, repoDid string) error { + _, err := d.ExecContext(ctx, `delete from jobs where repo_did = ?`, repoDid) + return err +} diff --git a/spindle/db/mill_state.go b/spindle/db/mill_state.go index 1cfe5014..8d04cfdc 100644 --- a/spindle/db/mill_state.go +++ b/spindle/db/mill_state.go @@ -325,11 +325,11 @@ func (d *DB) ClearPendingArtifacts() error { return err } -func (tx *EventBatchTx) InsertArtifactRef(leaseID, workflow, ref, hash string) error { +func (tx *EventBatchTx) InsertArtifactRef(leaseID, repoDid, workflow, ref, hash string) error { _, err := tx.tx.Exec( - `insert into mill_artifacts (lease_id, workflow, ref, hash) - values (?, ?, ?, ?)`, - leaseID, workflow, ref, hash, + `insert into mill_artifacts (lease_id, repo_did, workflow, ref, hash) + values (?, ?, ?, ?, ?)`, + leaseID, repoDid, workflow, ref, hash, ) return err } @@ -408,3 +408,73 @@ func (d *DB) ApplyEventBatch(n *notifier.Notifier, fn func(tx *EventBatchTx) err } return nil } + +func (d *DB) DeleteMillLeasesByRepo(repoDid string) ([]string, error) { + tx, err := d.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() + + rows, err := tx.Query(`select lease_id from mill_leases where repo_did = ?`, repoDid) + if err != nil { + return nil, err + } + var leaseIDs []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + rows.Close() + return nil, err + } + leaseIDs = append(leaseIDs, id) + } + if err := rows.Err(); err != nil { + rows.Close() + return nil, err + } + rows.Close() + + for _, id := range leaseIDs { + if _, err := tx.Exec(`delete from executor_pending_artifacts where lease_id = ?`, id); err != nil { + return nil, err + } + } + if _, err := tx.Exec(`delete from mill_leases where repo_did = ?`, repoDid); err != nil { + return nil, err + } + // return ids only after the deletion commits + if err := tx.Commit(); err != nil { + return nil, err + } + return leaseIDs, nil +} + +func (d *DB) ListArtifactRefsByRepo(repoDid string) ([]string, error) { + rows, err := d.Query(` + select ref from mill_artifacts where repo_did = ? + union + select pa.ref from executor_pending_artifacts pa + join mill_leases ml on ml.lease_id = pa.lease_id where ml.repo_did = ?`, + repoDid, repoDid, + ) + if err != nil { + return nil, err + } + defer rows.Close() + + var refs []string + for rows.Next() { + var ref string + if err := rows.Scan(&ref); err != nil { + return nil, err + } + refs = append(refs, ref) + } + return refs, rows.Err() +} + +func (d *DB) DeleteArtifactRefsByRepo(repoDid string) error { + _, err := d.Exec(`delete from mill_artifacts where repo_did = ?`, repoDid) + return err +} diff --git a/spindle/db/quota.go b/spindle/db/quota.go index d1d65c29..4278a9bb 100644 --- a/spindle/db/quota.go +++ b/spindle/db/quota.go @@ -841,3 +841,30 @@ func (d *QuotaStore) Recover(ctx context.Context, liveIDs []string) error { return err }) } + +// DeleteQuotaStateForRepo drops every quota row keyed by repo DID (cache +// state lives in quota tables since quota-schema); the repo's limit +// overrides go too, other dids' rows stay. +func (d *DB) DeleteQuotaStateForRepo(ctx context.Context, repoDid string) error { + tx, err := d.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() + + stmts := []struct { + query string + args []any + }{ + {`delete from quota_reservations where repo_did = ?`, []any{repoDid}}, + {`delete from quota_allocations where repo_did = ?`, []any{repoDid}}, + {`delete from quota_limits where did = ?`, []any{repoDid}}, + {`delete from quota_repo_owners where repo_did = ?`, []any{repoDid}}, + } + for _, s := range stmts { + if _, err := tx.ExecContext(ctx, s.query, s.args...); err != nil { + return err + } + } + return tx.Commit() +} diff --git a/spindle/db/repos.go b/spindle/db/repos.go index fa8bc0de..23d6864f 100644 --- a/spindle/db/repos.go +++ b/spindle/db/repos.go @@ -161,3 +161,40 @@ func (d *DB) DeleteRepoByOwnerRkey(owner syntax.DID, rkey syntax.RecordKey) erro _, err := d.Exec(`delete from repos where owner = ? and rkey = ?`, owner.String(), rkey.String()) return err } + +// one repo did can have several rows after renames +func (d *DB) ReposByDid(repoDid syntax.DID) ([]Repo, error) { + rows, err := d.Query( + `select knot, owner, rkey, repo_did, coalesce(created_at, '') from repos where repo_did = ?`, + repoDid.String(), + ) + if err != nil { + return nil, err + } + defer rows.Close() + + var repos []Repo + for rows.Next() { + var r Repo + var owner, rkey, did string + if err := rows.Scan(&r.Knot, &owner, &rkey, &did, &r.CreatedAt); err != nil { + return nil, err + } + r.Owner = syntax.DID(owner) + r.Rkey = syntax.RecordKey(rkey) + r.RepoDid = syntax.DID(did) + repos = append(repos, r) + } + return repos, rows.Err() +} + +func (d *DB) DeleteReposByDid(repoDid syntax.DID) error { + _, err := d.Exec(`delete from repos where repo_did = ?`, repoDid.String()) + return err +} + +func (d *DB) CountReposByOwner(owner syntax.DID) (int, error) { + var n int + err := d.QueryRow(`select count(*) from repos where owner = ?`, owner.String()).Scan(&n) + return n, err +} diff --git a/spindle/engine/engine.go b/spindle/engine/engine.go index ad31f84c..035201b0 100644 --- a/spindle/engine/engine.go +++ b/spindle/engine/engine.go @@ -359,7 +359,7 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q } }) } - defer archiveWorkflowLog(wl, stores, db, cfg.Server.LogDir, wid) + defer archiveWorkflowLog(wl, stores, db, cfg.Server.LogDir, wid, repoDID) defer closeLog() } @@ -548,7 +548,7 @@ func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, q l.Info("all workflows completed") } -func archiveWorkflowLog(l *slog.Logger, stores *artifactstore.Stores, database *db.DB, logDir string, wid models.WorkflowId) { +func archiveWorkflowLog(l *slog.Logger, stores *artifactstore.Stores, database *db.DB, logDir string, wid models.WorkflowId, repoDID string) { if stores == nil { return } @@ -577,7 +577,7 @@ func archiveWorkflowLog(l *slog.Logger, stores *artifactstore.Stores, database * return } digest := "sha256:" + hex.EncodeToString(hash.Sum(nil)) - if err := database.SaveArtifactRef(wid.String(), wid.Name, ref, digest); err != nil { + if err := database.SaveArtifactRef(wid.String(), repoDID, wid.Name, ref, digest); err != nil { l.Error("save workflow log artifact", "wid", wid, "err", err) } } diff --git a/spindle/mill/executor/observe.go b/spindle/mill/executor/observe.go index d11caa2c..8d7dc989 100644 --- a/spindle/mill/executor/observe.go +++ b/spindle/mill/executor/observe.go @@ -252,5 +252,8 @@ func (v *memVault) GetSecretsLocked(ctx context.Context, repo secrets.RepoIdenti } func (v *memVault) AddSecret(ctx context.Context, s secrets.UnlockedSecret) error { return nil } func (v *memVault) RemoveSecret(ctx context.Context, s secrets.Secret[any]) error { return nil } +func (v *memVault) RemoveAllSecrets(ctx context.Context, repo secrets.RepoIdentifier) error { + return nil +} var _ secrets.Manager = (*memVault)(nil) diff --git a/spindle/mill/mill.go b/spindle/mill/mill.go index 58bb3bf6..8584e91b 100644 --- a/spindle/mill/mill.go +++ b/spindle/mill/mill.go @@ -1293,11 +1293,17 @@ func (m *Mill) onEventBatch(sess *millSession, batch *millv1.EventBatch) error { if a.GetRef() == "" { return protoErrf("empty log artifact ref") } + // the ref names an object the mill deletes on wipe; + // pin it to the derived log key so an executor cannot + // point cleanup at someone else's objects + if a.GetRef() != "logs/"+lease.id+".log" { + return protoErrf("log artifact ref %q does not match lease %s", a.GetRef(), lease.id) + } if !strings.HasPrefix(a.GetHash(), "sha256:") { return protoErrf("invalid log artifact hash %q", a.GetHash()) } if tx != nil { - if err := tx.InsertArtifactRef(lease.id, lease.wid.Name, a.GetRef(), a.GetHash()); err != nil { + if err := tx.InsertArtifactRef(lease.id, lease.repoDID, lease.wid.Name, a.GetRef(), a.GetHash()); err != nil { return err } } diff --git a/spindle/secrets/manager.go b/spindle/secrets/manager.go index 1c43301b..f38d9af7 100644 --- a/spindle/secrets/manager.go +++ b/spindle/secrets/manager.go @@ -29,6 +29,7 @@ type UnlockedSecret = Secret[string] type Manager interface { AddSecret(ctx context.Context, secret UnlockedSecret) error RemoveSecret(ctx context.Context, secret Secret[any]) error + RemoveAllSecrets(ctx context.Context, repo RepoIdentifier) error GetSecretsLocked(ctx context.Context, repo RepoIdentifier) ([]LockedSecret, error) GetSecretsUnlocked(ctx context.Context, repo RepoIdentifier) ([]UnlockedSecret, error) } diff --git a/spindle/secrets/openbao.go b/spindle/secrets/openbao.go index 381429ae..4ee8c753 100644 --- a/spindle/secrets/openbao.go +++ b/spindle/secrets/openbao.go @@ -2,6 +2,7 @@ package secrets import ( "context" + "errors" "fmt" "log/slog" "path" @@ -323,3 +324,36 @@ func (v *OpenBaoManager) buildRepoPath(repo RepoIdentifier) string { func (v *OpenBaoManager) buildSecretPath(repo RepoIdentifier, key string) string { return path.Join(v.buildRepoPath(repo), key) } + +func (v *OpenBaoManager) RemoveAllSecrets(ctx context.Context, repo RepoIdentifier) error { + repoPath := v.buildRepoPath(repo) + + secretsList, err := v.client.Logical().ListWithContext(ctx, fmt.Sprintf("%s/metadata/%s", v.mountPath, repoPath)) + if err != nil { + if strings.Contains(err.Error(), "no secret found") || strings.Contains(err.Error(), "no handler for route") { + return nil + } + return fmt.Errorf("failed to list secrets for removal: %w", err) + } + if secretsList == nil { + return nil + } + + keys, ok := secretsList.Data["keys"].([]interface{}) + if !ok { + return nil + } + var errs []error + for _, k := range keys { + key, ok := k.(string) + if !ok { + continue + } + // one undeletable key must not strand the rest of the repo's secrets + if err := v.client.KVv2(v.mountPath).DeleteMetadata(ctx, v.buildSecretPath(repo, key)); err != nil { + errs = append(errs, fmt.Errorf("failed to delete secret %s from openbao: %w", key, err)) + } + } + v.logger.Info("removed all secrets for repo", "repo", repo, "count", len(keys)) + return errors.Join(errs...) +} diff --git a/spindle/secrets/openbao_test.go b/spindle/secrets/openbao_test.go index fadae843..f638c77b 100644 --- a/spindle/secrets/openbao_test.go +++ b/spindle/secrets/openbao_test.go @@ -64,6 +64,20 @@ func (m *MockOpenBaoManager) RemoveSecret(ctx context.Context, secret Secret[any return nil } +func (m *MockOpenBaoManager) RemoveAllSecrets(ctx context.Context, repo RepoIdentifier) error { + if m.shouldError { + return m.errorToReturn + } + + // match on the stored repo, not a key prefix: "a" is a prefix of "a_b" + for key, secret := range m.secrets { + if secret.Repo == repo { + delete(m.secrets, key) + } + } + return nil +} + func (m *MockOpenBaoManager) GetSecretsLocked(ctx context.Context, repo RepoIdentifier) ([]LockedSecret, error) { if m.shouldError { return nil, m.errorToReturn diff --git a/spindle/secrets/sqlite.go b/spindle/secrets/sqlite.go index 73c6654a..ff240590 100644 --- a/spindle/secrets/sqlite.go +++ b/spindle/secrets/sqlite.go @@ -174,3 +174,9 @@ func (s *SqliteManager) GetSecretsUnlocked(ctx context.Context, didSlashRepo Rep return ls, nil } + +func (s *SqliteManager) RemoveAllSecrets(ctx context.Context, repo RepoIdentifier) error { + query := fmt.Sprintf(`delete from %s where repo = ?;`, s.tableName) + _, err := s.db.ExecContext(ctx, query, repo) + return err +} diff --git a/spindle/secrets/sqlite_test.go b/spindle/secrets/sqlite_test.go index 25cbb839..f1f477de 100644 --- a/spindle/secrets/sqlite_test.go +++ b/spindle/secrets/sqlite_test.go @@ -588,3 +588,43 @@ func TestSqliteManager_StopperInterface(t *testing.T) { _, ok := interface{}(manager).(Stopper) assert.False(t, ok, "SqliteManager should NOT implement Stopper interface") } + +func TestSqliteManager_RemoveAllSecrets(t *testing.T) { + manager := createInMemoryDB(t) + defer manager.db.Close() + + for _, s := range []UnlockedSecret{ + createTestSecret("did:plc:repo1", "api_key", "v1", "did:plc:example123"), + createTestSecret("did:plc:repo1", "other_key", "v2", "did:plc:example123"), + createTestSecret("did:plc:repo2", "api_key", "v3", "did:plc:example123"), + } { + if err := manager.AddSecret(context.Background(), s); err != nil { + t.Fatalf("Failed to setup secret: %v", err) + } + } + + if err := manager.RemoveAllSecrets(context.Background(), RepoIdentifier("did:plc:repo1")); err != nil { + t.Fatalf("RemoveAllSecrets: %v", err) + } + + left, err := manager.GetSecretsUnlocked(context.Background(), RepoIdentifier("did:plc:repo1")) + if err != nil { + t.Fatal(err) + } + if len(left) != 0 { + t.Fatalf("expected no secrets left for repo1, got %d", len(left)) + } + + other, err := manager.GetSecretsUnlocked(context.Background(), RepoIdentifier("did:plc:repo2")) + if err != nil { + t.Fatal(err) + } + if len(other) != 1 { + t.Fatalf("expected repo2 secrets untouched, got %d", len(other)) + } + + // wipes can run repeatedly, so removal must tolerate an empty repo + if err := manager.RemoveAllSecrets(context.Background(), RepoIdentifier("did:plc:repo1")); err != nil { + t.Fatalf("RemoveAllSecrets on empty repo: %v", err) + } +} diff --git a/spindle/server.go b/spindle/server.go index 9ec553e1..0bc7816d 100644 --- a/spindle/server.go +++ b/spindle/server.go @@ -316,34 +316,28 @@ func (s *Spindle) DB() *db.DB { return s.db } -// Engines returns the map of available engines. func (s *Spindle) Engines() map[string]models.Engine { return s.engs } -// Vault returns the secrets manager instance. func (s *Spindle) Vault() secrets.Manager { return s.vault } -// Notifier returns the notifier instance. func (s *Spindle) Notifier() *notifier.Notifier { return s.n } -// Enforcer returns the RBAC enforcer instance. func (s *Spindle) Enforcer() *rbac.Enforcer { return s.e } -// SetMotdContent sets custom MOTD content, replacing the embedded default. func (s *Spindle) SetMotdContent(content []byte) { s.motdMu.Lock() defer s.motdMu.Unlock() s.motd = content } -// GetMotdContent returns the current MOTD content. func (s *Spindle) GetMotdContent() []byte { s.motdMu.RLock() defer s.motdMu.RUnlock() @@ -366,24 +360,7 @@ func (s *Spindle) Start(ctx context.Context) error { return fmt.Errorf("starting metrics listener: %w", err) } - // only standalone runs the local queue. mill hosts place directly onto - // executors, and executors only run jobs explicitly assigned by a mill - workersCtx, cancelWorkers := context.WithCancel(runCtx) - s.StartJobWorkers(workersCtx) - defer func() { - cancelWorkers() - done := make(chan struct{}) - go func() { - s.jobWorkers.Wait() - close(done) - }() - select { - case <-done: - case <-time.After(5 * time.Second): - s.l.Warn("timed out waiting for job workers to stop") - } - }() - + // an executor dials out to its mill and takes work from it var execDone chan struct{} if s.exec != nil { execDone = make(chan struct{}) @@ -432,6 +409,9 @@ func (s *Spindle) Start(ctx context.Context) error { defer tapCancel() } + // tap must be ready before owner wipes start + s.resumeWipes(ctx) + go func() { s.l.Info("starting knot event consumer") s.ks.Start(runCtx) @@ -441,6 +421,23 @@ func (s *Spindle) Start(ctx context.Context) error { s.tap.Start(tapCtx) } + // start workers after resuming wipes so banned jobs cannot run + workersCtx, cancelWorkers := context.WithCancel(runCtx) + s.StartJobWorkers(workersCtx) + defer func() { + cancelWorkers() + done := make(chan struct{}) + go func() { + s.jobWorkers.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + s.l.Warn("timed out waiting for job workers to stop") + } + }() + server := &http.Server{ Addr: s.cfg.Server.ListenAddr, Handler: s.Router(), @@ -784,6 +781,13 @@ func (s *Spindle) processKnotStreamInner(ctx context.Context, src eventconsumer. return fmt.Errorf("unknown repoDid %s: %w", repoDid, err) } + if ban, err := s.db.IsBanned(repoDid, repo.Owner); err != nil { + return fmt.Errorf("checking bans: %w", err) + } else if ban != nil { + l.Warn("dropping push for banned repo") + return nil + } + if src.Host != repo.Knot { return fmt.Errorf("repo knot does not match event source: %s != %s", src.Host, repo.Knot) } @@ -931,7 +935,6 @@ func (s *Spindle) resolvePipelineSourceRepo(ctx context.Context, trigger *tangle return s.resolveSourceRepoInfo(ctx, repoDid) } -// resolveSourceRepoInfo resolves trigger-repo metadata for a source repo DID. func (s *Spindle) resolveSourceRepoInfo(ctx context.Context, repoDid syntax.DID) (*tangled.Pipeline_TriggerRepo, error) { repo, err := s.db.GetRepoByDid(repoDid) if err == nil { @@ -950,9 +953,6 @@ func (s *Spindle) resolveSourceRepoInfo(ctx context.Context, repoDid syntax.DID) return s.buildTriggerRepoFrom(ctx, res.KnotURL.Host(), ownership.OwnerDid.String(), ownership.Rkey.String(), repoDid.String()), nil } -// runPipeline compiles and enqueues the pipeline for the given revision. -// sourceRepo is the resolved repo the code was checked out from, forwarded to -// processPipeline for env vars. func (s *Spindle) runPipeline(ctx context.Context, repoDid syntax.DID, trigger tangled.Pipeline_TriggerMetadata, changedFiles []string, repoCloneUri, repoPath, rev string, only []string, sourceRepo *tangled.Pipeline_TriggerRepo) (models.PipelineId, error) { l := log.FromContext(ctx) @@ -996,7 +996,6 @@ func (s *Spindle) runPipeline(ctx context.Context, repoDid syntax.DID, trigger t return pipelineId, err } -// filterWorkflows filters workflows to the requested names func filterWorkflows(workflows []*tangled.Pipeline_Workflow, only []string) []*tangled.Pipeline_Workflow { allowed := make(map[string]struct{}, len(only)) for _, n := range only { @@ -1022,6 +1021,12 @@ func (s *Spindle) TriggerManual(ctx context.Context, repoDid syntax.DID, sha, re return "", fmt.Errorf("unknown repoDid %s: %w", repoDid, err) } + if ban, err := s.db.IsBanned(repoDid, repo.Owner); err != nil { + return "", fmt.Errorf("checking bans: %w", err) + } else if ban != nil { + return "", fmt.Errorf("repo %s is banned", repoDid) + } + triggerRepo, err := s.buildTriggerRepo(ctx, repo) if err != nil { return "", fmt.Errorf("building trigger repo: %w", err) @@ -1061,6 +1066,13 @@ func (s *Spindle) TriggerManual(ctx context.Context, repoDid syntax.DID, sha, re if sourceInfo == nil { sourceInfo = triggerRepo } else { + // reject a banned source before running its workflow + sourceOwner, _ := syntax.ParseDID(sourceInfo.Did) + if ban, err := s.db.IsBanned(sourceRepo, sourceOwner); err != nil { + return "", fmt.Errorf("checking bans: %w", err) + } else if ban != nil { + return "", fmt.Errorf("source repo %s is banned", sourceRepo) + } sourceRepoStr := sourceRepo.String() trigger.SourceRepo = &sourceRepoStr } @@ -1177,8 +1189,6 @@ func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev strin return rawPipeline, nil } -// newRepoPath creates a path to store repository by its did and rkey. -// The path format would be: `/data/repos/did:plc:foo/sh.tangled.repo/repo-rkey func (s *Spindle) newRepoPath(repo syntax.DID) string { return filepath.Join(s.cfg.Server.RepoDir, repo.String()) } @@ -1359,7 +1369,6 @@ func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) { }, pipelineId) } -// enqueues the workflows in tpl. func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo) error { traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx) if err := s.db.EnqueueJob( diff --git a/spindle/tapclient.go b/spindle/tapclient.go index b2f16aed..927c61b1 100644 --- a/spindle/tapclient.go +++ b/spindle/tapclient.go @@ -81,6 +81,13 @@ func (t *Tap) AddOwnerDIDs(ctx context.Context, dids []syntax.DID) error { return t.tap.AddRepos(ctx, dids) } +func (t *Tap) RemoveOwnerDIDs(ctx context.Context, dids []syntax.DID) error { + if len(dids) == 0 { + return nil + } + return t.tap.RemoveRepos(ctx, dids) +} + func (t *Tap) Start(connCtx context.Context) { go t.tap.Connect(connCtx, &tapc.SimpleIndexer{ EventHandler: t.processEvent, @@ -142,6 +149,19 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error return nil } + if ban, err := t.spindle.db.IsBanned(repoDid, ownerDid); err != nil { + return fmt.Errorf("checking bans: %w", err) + } else if ban != nil { + l.Warn("refusing banned repo") + // wiping is best-effort and retried by the next record or + // resumeWipes; nacking would poison redelivery on any + // persistent cleanup failure + if err := t.spindle.WipeRepo(ctx, repoDid, "banned"); err != nil { + l.Error("wipe of banned repo failed", "err", err) + } + return nil + } + isMember, err := t.spindle.e.IsSpindleMember(ownerDid.String(), rbac.ThisServer) if err != nil { return fmt.Errorf("checking spindle membership: %w", err) @@ -151,7 +171,7 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error return nil } - // check if this repo DID is already owned by someone else + // a repo did already owned by someone else is a hijack attempt existingRepo, err := t.spindle.db.GetRepoByDid(repoDid) if err == nil { if existingRepo.Owner != ownerDid { @@ -162,6 +182,15 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error return fmt.Errorf("lookup existing repo by DID: %w", err) } + // recheck at the commit point: a ban landing during the checks above + // would otherwise register the repo after its wipe already ran + if ban, err := t.spindle.db.IsBanned(repoDid, ownerDid); err != nil { + return fmt.Errorf("rechecking bans: %w", err) + } else if ban != nil { + l.Warn("repo banned mid-registration") + return nil + } + if err := t.spindle.e.AddRepo(ownerDid.String(), rbac.ThisServer, repoDid.String()); err != nil { l.Error("failed to add repo policy", "err", err) return fmt.Errorf("add repo policy: %w", err) @@ -183,7 +212,6 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error return fmt.Errorf("add repo: %w", err) } - // setup sparse sync repoCloneUri := t.spindle.newRepoCloneUrl(repo.Knot, repo.RepoDid) repoPath := t.spindle.newRepoPath(repo.RepoDid) if err := gitutil.SparseSync(ctx, repoCloneUri, repoPath, "", sparseWorkflowDir); err != nil { @@ -225,32 +253,10 @@ func (t *Tap) processRepo(ctx context.Context, evt *tapc.RecordEventData) error func (t *Tap) teardownRepo(l *slog.Logger, repo *db.Repo, ownerDid syntax.DID, rkey syntax.RecordKey) error { if repo.RepoDid != "" { - collabs, err := t.spindle.db.ListCollaboratorsByRepoDid(repo.RepoDid) - if err != nil { - l.Error("failed to list collaborators for cleanup", "err", err) - return fmt.Errorf("list collaborators: %w", err) - } - for _, c := range collabs { - if err := t.spindle.e.RemoveCollaborator(c.Subject.String(), rbac.ThisServer, repo.RepoDid.String()); err != nil { - l.Error("failed to remove collaborator policy", "subject", c.Subject, "err", err) - return fmt.Errorf("remove collaborator policy: %w", err) - } - } - if err := t.spindle.db.DeleteRepoCollaboratorsByRepoDid(repo.RepoDid); err != nil { - l.Error("failed to clear collaborator rows", "err", err) - return err - } - if err := t.spindle.e.RemoveRepo(ownerDid.String(), rbac.ThisServer, repo.RepoDid.String()); err != nil { - l.Error("failed to remove repo policy", "err", err) - return fmt.Errorf("remove repo policy: %w", err) - } + return t.spindle.WipeRepo(context.Background(), repo.RepoDid, "repo record removed") } - if err := t.spindle.db.DeleteRepoByOwnerRkey(ownerDid, rkey); err != nil { - l.Error("failed to delete repo row", "err", err) - return fmt.Errorf("delete repo row: %w", err) - } - // TODO: clear sparse-synced git repo - return nil + // rows without a repo did never got past registration + return t.spindle.db.DeleteRepoByOwnerRkey(ownerDid, rkey) } func (t *Tap) processCollaborator(ctx context.Context, evt *tapc.RecordEventData) error { @@ -519,6 +525,13 @@ func (s *Spindle) triggerPullRequestPipeline(ctx context.Context, l *slog.Logger return fmt.Errorf("target repo is unknown") } + if ban, err := s.db.IsBanned(repo.RepoDid, repo.Owner); err != nil { + return fmt.Errorf("checking bans: %w", err) + } else if ban != nil { + l.Warn("dropping pull trigger for banned repo", "repo", repo.RepoDid) + return nil + } + // authorize the actor against the target repo allowed, err := s.isPullTriggerAuthorized(eventDid, pullDid, repo.RepoDid.String()) if err != nil { diff --git a/spindle/tapclient_test.go b/spindle/tapclient_test.go index 4bcaf738..58e4aa64 100644 --- a/spindle/tapclient_test.go +++ b/spindle/tapclient_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "log/slog" + "path/filepath" "strings" "tangled.org/core/jetstream" "testing" @@ -14,9 +15,13 @@ import ( "tangled.org/core/api/tangled" "tangled.org/core/eventconsumer" "tangled.org/core/idresolver" + "tangled.org/core/notifier" "tangled.org/core/rbac" + "tangled.org/core/spindle/artifactstore" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" + "tangled.org/core/spindle/models" + "tangled.org/core/spindle/secrets" "tangled.org/core/workflow" "tangled.org/core/tapc" @@ -432,17 +437,33 @@ func TestTeardownRepo_RBAC(t *testing.T) { cfg := &config.Config{} cfg.Server.Hostname = "spindle.test" + cfg.Server.RepoDir = t.TempDir() + cfg.Server.LogDir = t.TempDir() jc, jcerr := jetstream.NewJetstreamClient("", "", nil, nil, slog.Default(), nil, false, false) if jcerr != nil { t.Fatalf("NewJetstreamClient: %v", jcerr) } + vault, verr := secrets.NewSQLiteManager(filepath.Join(t.TempDir(), "secrets.db")) + if verr != nil { + t.Fatalf("NewSQLiteManager: %v", verr) + } + stores, serr := artifactstore.NewStores(config.ArtifactStores{ + Disk: config.ArtifactStoreDisk{Dir: t.TempDir()}, + }, "", "") + if serr != nil { + t.Fatalf("NewStores: %v", serr) + } s := &Spindle{ db: d, e: e, l: slog.Default(), cfg: cfg, jc: jc, + vault: vault, + stores: stores, + n: ptr(notifier.New()), + engs: map[string]models.Engine{}, rootCtx: context.Background(), } @@ -486,6 +507,14 @@ func TestTeardownRepo_RBAC(t *testing.T) { t.Fatalf("AddCollaborator policy: %v", err) } + // entities the delete path only reaches through DeleteAllData + if err := vault.AddSecret(context.Background(), secrets.UnlockedSecret{Key: "api_key", Value: "v", Repo: secrets.RepoIdentifier(repoDid.String()), CreatedBy: ownerDid}); err != nil { + t.Fatalf("AddSecret: %v", err) + } + if err := d.EnqueueJob(context.Background(), repoDid.String(), models.PipelineId{Knot: "knot.test", Rkey: "p1"}, nil, tangled.Pipeline{}, "", ""); err != nil { + t.Fatalf("EnqueueJob: %v", err) + } + err = tap.processRepo(context.Background(), &tapc.RecordEventData{ Live: true, Did: ownerDid, @@ -519,6 +548,15 @@ func TestTeardownRepo_RBAC(t *testing.T) { if err != nil || ok { t.Fatal("collaborator policy remained in Casbin after delete") } + + if got, kerr := vault.GetSecretsUnlocked(context.Background(), secrets.RepoIdentifier(repoDid.String())); kerr != nil || len(got) > 0 { + t.Fatalf("secrets remained in vault after delete: %v %v", got, kerr) + } + + var njobs int + if err := d.QueryRow(`select count(*) from jobs where repo_did = ?`, repoDid.String()).Scan(&njobs); err != nil || njobs != 0 { + t.Fatalf("jobs remained in DB after delete: %d %v", njobs, err) + } } func TestProcessRepo_ForgeDeleteRejection(t *testing.T) { diff --git a/spindle/wipe.go b/spindle/wipe.go new file mode 100644 index 00000000..f3a29ff1 --- /dev/null +++ b/spindle/wipe.go @@ -0,0 +1,247 @@ +package spindle + +import ( + "context" + "errors" + "fmt" + "os" + + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/rbac" + "tangled.org/core/spindle/db" + "tangled.org/core/spindle/engine" + "tangled.org/core/spindle/models" + "tangled.org/core/spindle/secrets" +) + +// cleanup continues after individual steps fail +func (s *Spindle) WipeRepo(ctx context.Context, repoDid syntax.DID, reason string) error { + l := s.l.With("repo", repoDid, "reason", reason) + l.Warn("wiping repo") + + repos, err := s.db.ReposByDid(repoDid) + if err != nil { + return fmt.Errorf("list repo rows: %w", err) + } + + var errs []error + fail := func(step string, err error) { + if err == nil { + return + } + l.Error("wipe step failed", "step", step, "err", err) + errs = append(errs, fmt.Errorf("%s: %w", step, err)) + } + + // cancel work before deleting the rows it references + wids, err := s.cancelRepoWorkflows(ctx, repoDid, repos, reason) + if err != nil { + return fmt.Errorf("cancel workflows: %w", err) + } + + // collect refs before deleting their rows + refs, err := s.db.ListArtifactRefsByRepo(repoDid.String()) + fail("list artifact refs", err) + + // leave rows as the retry manifest if object cleanup fails + objectsFailed := err != nil + refSet := map[string]bool{} + for _, ref := range refs { + refSet[ref] = true + for _, err := range s.stores.Delete(ctx, ref) { + if err != nil { + objectsFailed = true + fail("delete artifact "+ref, err) + } + } + } + if !objectsFailed { + fail("delete artifact refs", s.db.DeleteArtifactRefsByRepo(repoDid.String())) + } + + leaseIDs, err := s.db.DeleteMillLeasesByRepo(repoDid.String()) + fail("delete mill leases", err) + + fail("delete queued jobs", s.db.DeleteJobsByRepo(ctx, repoDid.String())) + fail("delete quota state", s.db.DeleteQuotaStateForRepo(ctx, repoDid.String())) + fail("delete secrets", s.vault.RemoveAllSecrets(ctx, secrets.RepoIdentifier(repoDid.String()))) + + collabs, err := s.db.ListCollaboratorsByRepoDid(repoDid) + fail("list collaborators", err) + for _, c := range collabs { + fail("remove collaborator acl", s.e.RemoveCollaborator(c.Subject.String(), rbac.ThisServer, repoDid.String())) + } + fail("delete collaborator rows", s.db.DeleteRepoCollaboratorsByRepoDid(repoDid)) + + for _, r := range repos { + fail("remove repo acl", s.e.RemoveRepo(r.Owner.String(), rbac.ThisServer, repoDid.String())) + } + + // use the recorded knot and rkey for status events + seen := map[db.PipelineKey]bool{} + var keys []db.PipelineKey + for _, wid := range wids { + k := db.PipelineKey{Knot: wid.PipelineId.Knot, Rkey: wid.PipelineId.Rkey} + if !seen[k] { + seen[k] = true + keys = append(keys, k) + } + } + fail("delete events", s.db.DeleteEventsByRepo(repoDid.String(), keys)) + + // keep the repo row when cleanup needs a retry + if len(errs) > 0 { + l.Error("keeping repo row for retry: earlier steps failed") + } else { + fail("delete repo rows", s.db.DeleteReposByDid(repoDid)) + } + + for _, leaseID := range leaseIDs { + ref := "logs/" + leaseID + ".log" + if refSet[ref] { + continue + } + for _, err := range s.stores.Delete(ctx, ref) { + fail("delete log "+leaseID, err) + } + } + + fail("delete clone", os.RemoveAll(s.newRepoPath(repoDid))) + + for _, r := range repos { + fail("release owner interest", s.releaseOwnerInterest(ctx, r.Owner)) + } + + if len(errs) > 0 { + return fmt.Errorf("wiped %s with %d step errors: %w", repoDid, len(errs), errors.Join(errs...)) + } + l.Info("repo wiped", "rows", len(repos)) + return nil +} + +func (s *Spindle) WipeOwner(ctx context.Context, ownerDid syntax.DID, reason string) error { + repos, err := s.db.AllRepos() + if err != nil { + return fmt.Errorf("list repos: %w", err) + } + var errs []error + for _, r := range repos { + if r.Owner != ownerDid || r.RepoDid == "" { + continue + } + if err := s.WipeRepo(ctx, r.RepoDid, reason); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +func (s *Spindle) cancelRepoWorkflows(ctx context.Context, repoDid syntax.DID, repos []db.Repo, reason string) ([]models.WorkflowId, error) { + var wids []models.WorkflowId + seen := map[models.WorkflowId]bool{} + add := func(knot, rkey, name string) { + wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: knot, Rkey: rkey}, Name: name} + if !seen[wid] { + seen[wid] = true + wids = append(wids, wid) + } + } + + pipes, err := s.db.ListPipelineWorkflows(repoDid.String()) + if err != nil { + return nil, fmt.Errorf("list pipeline workflows: %w", err) + } + for _, p := range pipes { + if p.Knot != "" { + add(p.Knot, p.Rkey, p.Name) + continue + } + for _, r := range repos { + add(r.Knot, p.Rkey, p.Name) + } + } + + leases, err := s.db.ListMillLeases() + if err != nil { + return nil, fmt.Errorf("list mill leases: %w", err) + } + for _, lease := range leases { + if lease.RepoDID == repoDid.String() { + add(lease.Knot, lease.Rkey, lease.Workflow) + } + } + + var errs []error + for _, wid := range wids { + st, err := s.db.GetStatus(wid) + if err == nil && st != nil && models.StatusKind(st.Status).IsFinish() { + continue + } + if err := s.db.StatusCancelled(wid, reason, -1, s.n); err != nil { + errs = append(errs, err) + } + engine.CancelWorkflow(wid) + for _, eng := range s.engs { + if err := eng.DestroyWorkflow(ctx, wid); err != nil { + errs = append(errs, err) + } + } + if err := os.Remove(models.LogFilePath(s.cfg.Server.LogDir, wid)); err != nil && !os.IsNotExist(err) { + errs = append(errs, fmt.Errorf("remove live log for %s: %w", wid, err)) + } + } + return wids, errors.Join(errs...) +} + +func (s *Spindle) releaseOwnerInterest(ctx context.Context, ownerDid syntax.DID) error { + if ownerDid == "" { + return nil + } + n, err := s.db.CountReposByOwner(ownerDid) + if err != nil { + return err + } + if n > 0 { + return nil + } + members, err := db.CountSpindleMembersBySubject(s.db, ownerDid.String()) + if err != nil { + return err + } + if members > 0 { + return nil + } + // the server owner inherits membership via casbin without a members row + if ok, err := s.e.IsSpindleMember(ownerDid.String(), rbac.ThisServer); err != nil { + return err + } else if ok { + return nil + } + if err := db.RemoveDid(s.db, ownerDid.String()); err != nil { + return err + } + s.jc.RemoveDid(ownerDid.String()) + if s.tap != nil { + return s.tap.RemoveOwnerDIDs(ctx, []syntax.DID{ownerDid}) + } + return nil +} + +// retry wipes recorded by bans +func (s *Spindle) resumeWipes(ctx context.Context) { + bans, err := s.db.BanList() + if err != nil { + s.l.Error("failed to list bans for reconcile", "err", err) + return + } + for _, b := range bans { + // try both ban roles + err := s.WipeRepo(ctx, b.SubjectDid, "banned") + if oerr := s.WipeOwner(ctx, b.SubjectDid, "banned"); oerr != nil { + err = errors.Join(err, oerr) + } + if err != nil { + s.l.Error("ban reconcile failed", "subject", b.SubjectDid, "err", err) + } + } +} diff --git a/spindle/wipe_test.go b/spindle/wipe_test.go new file mode 100644 index 00000000..80eab1a1 --- /dev/null +++ b/spindle/wipe_test.go @@ -0,0 +1,301 @@ +package spindle + +import ( + "context" + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/bluesky-social/indigo/atproto/syntax" + "tangled.org/core/api/tangled" + "tangled.org/core/jetstream" + "tangled.org/core/log" + "tangled.org/core/notifier" + "tangled.org/core/rbac" + "tangled.org/core/spindle/artifactstore" + "tangled.org/core/spindle/config" + "tangled.org/core/spindle/db" + "tangled.org/core/spindle/models" + "tangled.org/core/spindle/secrets" +) + +func newWipeTestSpindle(t *testing.T) *Spindle { + t.Helper() + tmp := t.TempDir() + ctx := context.Background() + + d, err := db.Make(ctx, filepath.Join(tmp, "spindle.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { d.Close() }) + + e, err := rbac.NewEnforcer(filepath.Join(tmp, "rbac.db")) + if err != nil { + t.Fatal(err) + } + + vault, err := secrets.NewSQLiteManager(filepath.Join(tmp, "secrets.db")) + if err != nil { + t.Fatal(err) + } + + stores, err := artifactstore.NewStores(config.ArtifactStores{ + Disk: config.ArtifactStoreDisk{Dir: filepath.Join(tmp, "artifacts")}, + }, "", "") + if err != nil { + t.Fatal(err) + } + + jc, err := jetstream.NewJetstreamClient("", "test", nil, nil, log.New("test"), d, false, false) + if err != nil { + t.Fatal(err) + } + + return &Spindle{ + l: log.New("test"), + db: d, + e: e, + vault: vault, + stores: stores, + jc: jc, + n: ptr(notifier.New()), + engs: map[string]models.Engine{}, + cfg: &config.Config{ + Server: config.Server{ + Hostname: "spindle.example.com", + RepoDir: filepath.Join(tmp, "repos"), + LogDir: filepath.Join(tmp, "logs"), + }, + }, + } +} + +func seedWipeRepo(t *testing.T, s *Spindle, repoDid, owner syntax.DID) { + t.Helper() + ctx := context.Background() + + sfx := repoDid.String()[len(repoDid.String())-1:] + + if err := s.db.AddRepo(db.Repo{Knot: "knot.example.com", Owner: owner, Rkey: syntax.RecordKey("r" + sfx), RepoDid: repoDid}); err != nil { + t.Fatal(err) + } + if err := db.AddDid(s.db, owner.String()); err != nil { + t.Fatal(err) + } + if err := s.db.AddRepoCollaborator(db.RepoCollaborator{OwnerDid: owner, Rkey: syntax.RecordKey("c" + sfx), Subject: "did:plc:collab", RepoDid: repoDid}); err != nil { + t.Fatal(err) + } + if err := s.db.EnqueueJob(ctx, repoDid.String(), models.PipelineId{Knot: "knot.example.com", Rkey: "p" + sfx}, nil, tangled.Pipeline{}, "", ""); err != nil { + t.Fatal(err) + } + if err := s.db.SaveMillLease(db.MillLease{LeaseID: "l" + sfx, NodeID: "n1", Epoch: "e1", Engine: "microvm", Knot: "knot.example.com", Rkey: "p" + sfx, Workflow: "w" + sfx, State: "active", RepoDID: repoDid.String()}); err != nil { + t.Fatal(err) + } + if _, err := s.db.Exec(`insert into mill_artifacts (lease_id, repo_did, workflow, ref, hash) values (?, ?, ?, ?, 'h')`, "l"+sfx, repoDid.String(), "w"+sfx, "out/l"+sfx+".bin"); err != nil { + t.Fatal(err) + } + if _, err := s.db.Exec(`insert into quota_allocations (repo_did, resource, kind, key, amount) values (?, 'compute', 'generic', 'k', 1)`, repoDid.String()); err != nil { + t.Fatal(err) + } + 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 { + 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 { + t.Fatal(err) + } + + if err := os.MkdirAll(s.newRepoPath(repoDid), 0755); err != nil { + t.Fatal(err) + } + for _, ref := range []string{"out/l" + sfx + ".bin", "logs/l" + sfx + ".log"} { + if err := s.stores.PutFile(ctx, ref, writeTempFile(t, "data")); err != nil && len(err) > 0 { + t.Fatal(err) + } + } +} + +func writeTempFile(t *testing.T, content string) string { + t.Helper() + p := filepath.Join(t.TempDir(), "src") + if err := os.WriteFile(p, []byte(content), 0644); err != nil { + t.Fatal(err) + } + return p +} + +func TestWipeRepoRemovesAllState(t *testing.T) { + s := newWipeTestSpindle(t) + ctx := context.Background() + repoDid := syntax.DID("did:plc:repo1") + owner := syntax.DID("did:plc:owner1") + + seedWipeRepo(t, s, repoDid, owner) + + if err := s.WipeRepo(ctx, repoDid, "test"); err != nil { + t.Fatalf("WipeRepo: %v", err) + } + + var n int + for _, table := range []string{"repos", "jobs", "mill_leases", "mill_artifacts", "quota_allocations", "repo_collaborators", "events"} { + if err := s.db.QueryRow(`select count(*) from ` + table).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("%s: expected empty, got %d rows", table, n) + } + } + + got, err := s.vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(repoDid.String())) + if err != nil { + t.Fatal(err) + } + if len(got) != 0 { + t.Fatalf("expected secrets gone, got %d", len(got)) + } + + if _, err := os.Stat(s.newRepoPath(repoDid)); !os.IsNotExist(err) { + t.Fatalf("expected clone dir gone, stat err: %v", err) + } + if _, err := s.stores.Open(ctx, "out/l1.bin"); err == nil { + t.Fatal("expected artifact gone") + } + if _, err := s.stores.Open(ctx, "logs/l1.log"); err == nil { + t.Fatal("expected log artifact gone") + } + + dids, err := s.db.GetAllDids() + if err != nil { + t.Fatal(err) + } + if len(dids) != 0 { + t.Fatalf("expected owner did released, got %v", dids) + } + + if err := s.WipeRepo(ctx, repoDid, "test again"); err != nil { + t.Fatalf("second WipeRepo: %v", err) + } +} + +func TestWipeRepoKeepsMemberInterest(t *testing.T) { + s := newWipeTestSpindle(t) + ctx := context.Background() + repoDid := syntax.DID("did:plc:repo1") + owner := syntax.DID("did:plc:owner1") + + seedWipeRepo(t, s, repoDid, owner) + if err := db.AddSpindleMember(s.db, db.SpindleMember{ + Did: owner, + Rkey: "m1", + Instance: "spindle.example.com", + Subject: owner, + }); err != nil { + t.Fatal(err) + } + + if err := s.WipeRepo(ctx, repoDid, "test"); err != nil { + t.Fatalf("WipeRepo: %v", err) + } + + dids, err := s.db.GetAllDids() + if err != nil { + t.Fatal(err) + } + if len(dids) != 1 || dids[0] != owner.String() { + t.Fatalf("expected member did kept, got %v", dids) + } +} + +func TestWipeRepoKeepsCasbinMemberInterest(t *testing.T) { + s := newWipeTestSpindle(t) + repoDid := syntax.DID("did:plc:repo1") + owner := syntax.DID("did:plc:owner1") + + seedWipeRepo(t, s, repoDid, owner) + // the server owner inherits membership via casbin without a members row + if err := s.e.AddSpindleMember(rbac.ThisServer, owner.String()); err != nil { + t.Fatal(err) + } + + if err := s.WipeRepo(context.Background(), repoDid, "test"); err != nil { + t.Fatalf("WipeRepo: %v", err) + } + + dids, err := s.db.GetAllDids() + if err != nil { + t.Fatal(err) + } + if len(dids) != 1 || dids[0] != owner.String() { + t.Fatalf("expected casbin member did kept, got %v", dids) + } +} + +func TestWipeRepoRemovesLocalEngineArtifacts(t *testing.T) { + s := newWipeTestSpindle(t) + ctx := context.Background() + repoDid := syntax.DID("did:plc:repo1") + owner := syntax.DID("did:plc:owner1") + + seedWipeRepo(t, s, repoDid, owner) + + wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "knot.example.com", Rkey: "p1"}, Name: "w1"} + localRef := "logs/" + wid.String() + ".log" + if err := s.db.SaveArtifactRef(wid.String(), repoDid.String(), wid.Name, localRef, "h"); err != nil { + t.Fatal(err) + } + if err := s.stores.PutFile(ctx, localRef, writeTempFile(t, "log")); err != nil && len(err) > 0 { + t.Fatal(err) + } + + if err := s.WipeRepo(ctx, repoDid, "test"); err != nil { + t.Fatalf("WipeRepo: %v", err) + } + + if _, err := s.stores.Open(ctx, localRef); err == nil { + t.Fatal("expected local engine artifact gone") + } + var n int + if err := s.db.QueryRow(`select count(*) from mill_artifacts`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 0 { + t.Fatalf("expected mill_artifacts empty, got %d rows", n) + } +} + +func TestWipeOwnerWipesAllOwnedRepos(t *testing.T) { + s := newWipeTestSpindle(t) + ctx := context.Background() + owner := syntax.DID("did:plc:owner1") + other := syntax.DID("did:plc:owner2") + + seedWipeRepo(t, s, "did:plc:repo1", owner) + seedWipeRepo(t, s, "did:plc:repo2", owner) + seedWipeRepo(t, s, "did:plc:repo3", other) + + if err := s.WipeOwner(ctx, owner, "test"); err != nil { + t.Fatalf("WipeOwner: %v", err) + } + + var n int + if err := s.db.QueryRow(`select count(*) from repos`).Scan(&n); err != nil { + t.Fatal(err) + } + if n != 1 { + t.Fatalf("expected only the other owner's repo left, got %d", n) + } + repos, err := s.db.ReposByDid("did:plc:repo3") + if err != nil || len(repos) != 1 { + t.Fatalf("expected repo3 intact, got %v err %v", repos, err) + } +} + +func ptr[T any](v T) *T { return &v } + +