From e1b3c64edf59eed1ee27a2a66c9983d23ad22a25 Mon Sep 17 00:00:00 2001 From: Eugene Blikh Date: Wed, 22 Jul 2026 14:07:36 +0300 Subject: [PATCH] =?UTF-8?q?feat:=20service=20=E2=80=94=20wiring,=20space?= =?UTF-8?q?=20lifecycle,=20read=20paths,=20push=20validation,=20reconciler?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 1 of the implementation plan: the orchestration layer's read, validate and repair halves. The write plane (propose/merge) is Phase 3 and is absent. - Config/New assemble a Service from the shared config.ini, reporting every missing key in one message; TokenStore adapts db/ to authn.TokenStore, mapping db.ErrNotFound onto authn.ErrUnknownToken so an unknown credential is a 401 rather than a 503. - CreateSpace writes the repository first and the row second, removing the repository again if the insert fails; a crash between the two leaves content that is merely unlisted rather than a phantom space. - ReadDocument/ListDocuments/Policy/ResolveRev resolve the approved head or a pinned revision through one code path. - ValidatePush checks the refs rule first and unconditionally, then frontmatter and document-id uniqueness, which --push-option=skip-validation waives. The rejection is a structured, terminal-shaped message naming the document. - PlanRepairs is the repair table as a pure, table-tested function; Reconcile gathers the facts and applies them, listing stale-index spaces for Phase 2. Two departures from the design's repair table, both to stop the reconciler destroying live state, documented at their definitions: an open row with no branch is left alone inside a grace window (every propose passes through that state), and a branch still sitting on its recorded base is never treated as merged (its tip is trivially an ancestor of the approved head). --- service/doc.go | 44 +++ service/fixture_test.go | 378 ++++++++++++++++++++++++ service/integration_test.go | 313 ++++++++++++++++++++ service/push.go | 440 ++++++++++++++++++++++++++++ service/push_test.go | 349 ++++++++++++++++++++++ service/read.go | 177 +++++++++++ service/read_test.go | 213 ++++++++++++++ service/reconcile.go | 570 ++++++++++++++++++++++++++++++++++++ service/reconcile_test.go | 295 +++++++++++++++++++ service/service.go | 247 ++++++++++++++++ service/service_test.go | 184 ++++++++++++ service/space.go | 150 ++++++++++ service/space_test.go | 107 +++++++ service/token.go | 75 +++++ 14 files changed, 3542 insertions(+) create mode 100644 service/doc.go create mode 100644 service/fixture_test.go create mode 100644 service/integration_test.go create mode 100644 service/push.go create mode 100644 service/push_test.go create mode 100644 service/read.go create mode 100644 service/read_test.go create mode 100644 service/reconcile.go create mode 100644 service/reconcile_test.go create mode 100644 service/service.go create mode 100644 service/service_test.go create mode 100644 service/space.go create mode 100644 service/space_test.go create mode 100644 service/token.go diff --git a/service/doc.go b/service/doc.go new file mode 100644 index 0000000000000000000000000000000000000000..e9f73a3d936ecb15841eafb8991dea0aeafddbbb --- /dev/null +++ b/service/doc.go @@ -0,0 +1,44 @@ +// Package service is spec.sr.ht's orchestration layer: the single seam that +// REST, MCP, GraphQL and the web UI all call, and the only place where git, +// Postgres and identity are allowed to meet. +// +// Dependency direction is strictly downward. This package imports core, gitx, +// db and authn; nothing above it may touch gitx or db directly. That rule is +// what keeps the three agent-facing surfaces behaviourally identical — they +// share these functions rather than each re-deriving the rules. +// +// # What Phase 1 contains +// +// Reads, validation and repair. Concretely: +// +// - Wiring: [Config] and [New] assemble a [Service] from the shared +// SourceHut config.ini, failing loudly at startup on a missing key rather +// than deep inside the first request. +// - Space lifecycle: [Service.CreateSpace], [Service.OpenSpace], +// [Service.ListSpaces]. +// - Reads: [Service.ReadDocument], [Service.ListDocuments], +// [Service.Policy] — each resolving either the approved head or a pinned +// revision through the same code path, because there is one storage tier +// and no checkout. +// - Push validation: [Service.ValidatePush], the function the `update` hook +// calls over RPC before a ref moves. +// - Repair: [Reconcile] and [Service.RunReconciler]. +// +// The write plane — propose, If-Match resolution, merge, auto-merge policy, +// digest bookkeeping — is Phase 3 and is deliberately absent. +// +// # One storage tier +// +// Every read resolves a git revision and reads blobs. The approved head, a +// pinned ?rev= and a proposal branch are the same call with a different +// revision string; an empty revision means "the approved head", which is what +// makes "reads default to the approved revision" a property of this layer +// rather than of each caller. +// +// # Errors +// +// Callers above this layer must not need gitx or db to interpret a failure, so +// every error that leaves this package satisfies one of the sentinels declared +// here ([ErrNotFound], [ErrSpaceExists], [ErrPushRejected], ...) as well as the +// lower-level class it came from. Multi-%w wrapping keeps both. +package service diff --git a/service/fixture_test.go b/service/fixture_test.go new file mode 100644 index 0000000000000000000000000000000000000000..46aca7c0e479583cc90bf204050108f9f3e8ca81 --- /dev/null +++ b/service/fixture_test.go @@ -0,0 +1,378 @@ +package service + +import ( + "context" + "crypto/rand" + "database/sql" + "encoding/hex" + "fmt" + "net/url" + "os" + "sort" + "strings" + "testing" + "time" + + "github.com/go-git/go-git/v5" + "github.com/go-git/go-git/v5/plumbing" + "github.com/go-git/go-git/v5/plumbing/filemode" + "github.com/go-git/go-git/v5/plumbing/object" + "github.com/go-git/go-git/v5/plumbing/storer" + _ "github.com/lib/pq" + "github.com/vaughan0/go-ini" + + "sourcecraft.dev/bigbes/sr-ht-spec/authn" + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/gitx" +) + +// testEnv names the DSN env var that gates the Postgres-backed tests, spelled +// exactly as db/ spells it so one variable turns on the integration tests of +// the whole module. When it is unset those tests skip; everything expressible +// without a database — config validation, the token-store error contract, the +// push rejection message, the reconciler's decision table — runs regardless. +const testEnv = "SPECSRHT_TEST_PG" + +var fxSpace = core.SpaceRef{Owner: "bigbes", Name: "rfcs"} + +func fxTime(n int) time.Time { + return time.Date(2026, 7, 22, 12, 0, 0, 0, time.UTC).Add(time.Duration(n) * time.Minute) +} + +// testInstance is the identity half of the config, matching what +// authn.InstanceFromConfig would derive from the origin below. +func testInstance() authn.Instance { + return authn.Instance{ + OwnerName: "bigbes", + OwnerEmail: "bigbes@gmail.com", + AgentEmail: "agent@spec.srht.bigb.es", + } +} + +func testConfig(t *testing.T, repos string) Config { + t.Helper() + return Config{ + Repos: repos, + Cache: t.TempDir(), + Origin: "https://spec.srht.bigb.es", + ConnectionString: "postgresql://specsrht@localhost/spec.sr.ht?sslmode=disable", + Instance: testInstance(), + } +} + +// testIni renders a config.ini with every key service/ requires, minus the ones +// named in omit, so a test can knock out exactly one key and see what is said +// about it. +func testIni(t *testing.T, repos string, omit ...string) ini.File { + t.Helper() + keys := []struct{ section, key, value string }{ + {"sr.ht", "owner-name", "bigbes"}, + {"sr.ht", "owner-email", "bigbes@gmail.com"}, + {ConfigSection, "origin", "https://spec.srht.bigb.es"}, + {ConfigSection, "repos", repos}, + {ConfigSection, "cache", "/var/cache/spec"}, + {ConfigSection, "connection-string", "postgresql://specsrht@localhost/spec.sr.ht?sslmode=disable"}, + } + dropped := make(map[string]bool, len(omit)) + for _, o := range omit { + dropped[o] = true + } + var b strings.Builder + section := "" + for _, k := range keys { + if dropped[k.key] { + continue + } + if k.section != section { + fmt.Fprintf(&b, "[%s]\n", k.section) + section = k.section + } + fmt.Fprintf(&b, "%s=%s\n", k.key, k.value) + } + conf, err := ini.Load(strings.NewReader(b.String())) + if err != nil { + t.Fatalf("load synthesized ini: %v", err) + } + return conf +} + +// deadDB is a database handle that resolves and then fails to connect. It gives +// the tests a *sql.DB whose every query errors without a server, which is what +// New requires (a nil handle is refused) and what the compensating-delete test +// in CreateSpace needs. +func deadDB(t *testing.T) *sql.DB { + t.Helper() + pool, err := sql.Open("postgres", + "postgres://nobody@127.0.0.1:1/nothing?sslmode=disable&connect_timeout=1") + if err != nil { + t.Fatalf("open dead pool: %v", err) + } + t.Cleanup(func() { pool.Close() }) + return pool +} + +// newService builds a Service over a repos root and a database that cannot be +// reached, for the tests that exercise git and pure logic only. Any accidental +// query fails loudly instead of passing silently. +func newService(t *testing.T) (*Service, string) { + t.Helper() + root := t.TempDir() + svc, err := New(testConfig(t, root), deadDB(t)) + if err != nil { + t.Fatalf("New: %v", err) + } + return svc, root +} + +// newSpace creates a bare repository for fxSpace under root and returns it as a +// Space with the row id a test supplies. It bypasses Service.CreateSpace on +// purpose: most tests here have no database, and the repository is the space. +func newSpace(t *testing.T, root string, id int) *Space { + t.Helper() + repo, err := gitx.Create(context.Background(), root, fxSpace, gitx.CreateOptions{ + Owner: gitx.Signature{Name: "bigbes", Email: "bigbes@gmail.com", When: fxTime(0)}, + }) + if err != nil { + t.Fatalf("gitx.Create: %v", err) + } + return &Space{Ref: fxSpace, ID: id, Created: fxTime(0), Repo: repo} +} + +// cutBranch creates a proposal branch at base, the way an agent's first write +// does. Committing onto a branch that was never cut would produce an orphan +// history, which no push in this system can create. +func cutBranch(t *testing.T, sp *Space, branch, base string) { + t.Helper() + if _, err := sp.Repo.CreateProposalBranch(context.Background(), branch, base); err != nil { + t.Fatalf("CreateProposalBranch(%q, %q): %v", branch, base, err) + } +} + +// doc renders a minimal valid document: the three required keys, then a body. +func doc(id, title, body string) []byte { + return []byte(fmt.Sprintf("---\nid: %s\ntitle: %s\nstatus: draft\n---\n\n%s\n", id, title, body)) +} + +// commitFiles writes files onto a branch, preserving everything already there, +// and returns the new commit. It stands in for a human push through +// receive-pack, which is the only way content reaches the approved branch. +// +// A nil value deletes the path. +func commitFiles(t *testing.T, sp *Space, branch string, n int, files map[string][]byte) plumbing.Hash { + t.Helper() + repo, err := git.PlainOpen(sp.Repo.Dir()) + if err != nil { + t.Fatalf("PlainOpen %s: %v", sp.Repo.Dir(), err) + } + st := repo.Storer + name := plumbing.NewBranchReferenceName(branch) + + var parents []plumbing.Hash + entries := map[string]plumbing.Hash{} + if ref, err := repo.Reference(name, false); err == nil { + parents = []plumbing.Hash{ref.Hash()} + commit, err := repo.CommitObject(ref.Hash()) + if err != nil { + t.Fatalf("commit %s: %v", ref.Hash(), err) + } + iter, err := commit.Files() + if err != nil { + t.Fatalf("files of %s: %v", ref.Hash(), err) + } + if err := iter.ForEach(func(f *object.File) error { + entries[f.Name] = f.Blob.Hash + return nil + }); err != nil { + t.Fatalf("walk %s: %v", ref.Hash(), err) + } + } + + for path, data := range files { + if data == nil { + delete(entries, path) + continue + } + entries[path] = writeBlob(t, st, data) + } + + tree := writeTree(t, st, entries) + sig := object.Signature{Name: "bigbes", Email: "bigbes@gmail.com", When: fxTime(n)} + commit := &object.Commit{ + Author: sig, Committer: sig, + Message: fmt.Sprintf("commit %d\n", n), + TreeHash: tree, + ParentHashes: parents, + } + obj := st.NewEncodedObject() + if err := commit.Encode(obj); err != nil { + t.Fatalf("encode commit: %v", err) + } + hash, err := st.SetEncodedObject(obj) + if err != nil { + t.Fatalf("store commit: %v", err) + } + if err := st.SetReference(plumbing.NewHashReference(name, hash)); err != nil { + t.Fatalf("set %s: %v", name, err) + } + return hash +} + +func writeBlob(t *testing.T, st storer.EncodedObjectStorer, data []byte) plumbing.Hash { + t.Helper() + obj := st.NewEncodedObject() + obj.SetType(plumbing.BlobObject) + obj.SetSize(int64(len(data))) + w, err := obj.Writer() + if err != nil { + t.Fatalf("blob writer: %v", err) + } + if _, err := w.Write(data); err != nil { + t.Fatalf("write blob: %v", err) + } + if err := w.Close(); err != nil { + t.Fatalf("close blob: %v", err) + } + hash, err := st.SetEncodedObject(obj) + if err != nil { + t.Fatalf("store blob: %v", err) + } + return hash +} + +// writeTree builds a tree from a flat path->blob map, recursing on directories. +// Entries are sorted the way git sorts them (directories compare as if they +// ended in '/'), so the objects this produces are byte-identical to git's. +func writeTree(t *testing.T, st storer.EncodedObjectStorer, files map[string]plumbing.Hash) plumbing.Hash { + t.Helper() + blobs := map[string]plumbing.Hash{} + dirs := map[string]map[string]plumbing.Hash{} + for path, hash := range files { + name, rest, nested := strings.Cut(path, "/") + if !nested { + blobs[name] = hash + continue + } + if dirs[name] == nil { + dirs[name] = map[string]plumbing.Hash{} + } + dirs[name][rest] = hash + } + + var entries []object.TreeEntry + for name, hash := range blobs { + entries = append(entries, object.TreeEntry{Name: name, Mode: filemode.Regular, Hash: hash}) + } + for name, sub := range dirs { + entries = append(entries, object.TreeEntry{ + Name: name, Mode: filemode.Dir, Hash: writeTree(t, st, sub), + }) + } + sortKey := func(e object.TreeEntry) string { + if e.Mode == filemode.Dir { + return e.Name + "/" + } + return e.Name + } + sort.Slice(entries, func(i, j int) bool { return sortKey(entries[i]) < sortKey(entries[j]) }) + + tree := &object.Tree{Entries: entries} + obj := st.NewEncodedObject() + if err := tree.Encode(obj); err != nil { + t.Fatalf("encode tree: %v", err) + } + hash, err := st.SetEncodedObject(obj) + if err != nil { + t.Fatalf("store tree: %v", err) + } + return hash +} + +// newTestService connects to the Postgres pointed at by SPECSRHT_TEST_PG, +// applies schema.sql into an isolated scratch schema, and returns a Service +// bound to it plus its repos root. Skips when the variable is unset. The shape +// is db/'s newTestStore, adapted: a per-test schema needs no CREATE DATABASE +// privilege and cleans up with one DROP SCHEMA ... CASCADE. +func newTestService(t *testing.T) (*Service, string) { + t.Helper() + base := os.Getenv(testEnv) + if base == "" { + t.Skipf("%s not set; skipping Postgres-backed test (set it to a DSN to run)", testEnv) + } + + admin, err := sql.Open("postgres", base) + if err != nil { + t.Fatalf("open admin pool: %v", err) + } + if err := admin.Ping(); err != nil { + admin.Close() + t.Fatalf("ping %s: %v", testEnv, err) + } + + schema := "specsrht_test_" + randToken() + if _, err := admin.Exec(`CREATE SCHEMA "` + schema + `"`); err != nil { + admin.Close() + t.Fatalf("create schema %s: %v", schema, err) + } + drop := func() { + if _, err := admin.Exec(`DROP SCHEMA "` + schema + `" CASCADE`); err != nil { + t.Errorf("drop schema %s: %v", schema, err) + } + admin.Close() + } + + scopedDSN, err := withSearchPath(base, schema) + if err != nil { + drop() + t.Fatalf("build scoped dsn: %v", err) + } + pool, err := sql.Open("postgres", scopedDSN) + if err != nil { + drop() + t.Fatalf("open scoped pool: %v", err) + } + ddl, err := os.ReadFile("../schema.sql") + if err != nil { + pool.Close() + drop() + t.Fatalf("read schema.sql: %v", err) + } + if _, err := pool.Exec(string(ddl)); err != nil { + pool.Close() + drop() + t.Fatalf("apply schema.sql: %v", err) + } + t.Cleanup(func() { + pool.Close() + drop() + }) + + root := t.TempDir() + svc, err := New(testConfig(t, root), pool) + if err != nil { + t.Fatalf("New: %v", err) + } + return svc, root +} + +func withSearchPath(base, schema string) (string, error) { + opt := "-c search_path=" + schema + if strings.Contains(base, "://") { + u, err := url.Parse(base) + if err != nil { + return "", err + } + q := u.Query() + q.Set("options", opt) + u.RawQuery = q.Encode() + return u.String(), nil + } + return base + " options='" + opt + "'", nil +} + +func randToken() string { + b := make([]byte, 8) + if _, err := rand.Read(b); err != nil { + panic(err) + } + return hex.EncodeToString(b) +} diff --git a/service/integration_test.go b/service/integration_test.go new file mode 100644 index 0000000000000000000000000000000000000000..33efb5498c092e82840bdfb052f5acfca4e2eb06 --- /dev/null +++ b/service/integration_test.go @@ -0,0 +1,313 @@ +package service + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/go-git/go-git/v5" + "github.com/go-git/go-git/v5/plumbing" + + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/db" +) + +// These exercise the paths that genuinely need Postgres: the global ID registry +// behind push validation, and the reconciler's repairs. Everything they do not +// cover is covered without a database elsewhere in this package. + +func TestValidatePushAcceptsAGoodPush(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + sp := mustCreateSpace(t, svc) + + old, err := sp.Repo.ApprovedHead(ctx) + if err != nil { + t.Fatalf("approved head: %v", err) + } + head := commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007.md": doc("SPEC-0007", "Storage", "body"), + }) + + err = svc.ValidatePush(ctx, PushRequest{ + Space: fxSpace, Principal: owner(), Ref: "refs/heads/main", + Old: old.String(), New: head.String(), + }) + if err != nil { + t.Fatalf("ValidatePush: %v", err) + } +} + +func TestValidatePushRejectsAGlobalIDCollision(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + sp := mustCreateSpace(t, svc) + + // SPEC-0007 already lives in another space. + other, err := svc.CreateSpace(ctx, core.SpaceRef{Owner: "bigbes", Name: "notes"}) + if err != nil { + t.Fatalf("CreateSpace: %v", err) + } + id, err := core.ParseDocID("SPEC-0007") + if err != nil { + t.Fatal(err) + } + if _, err := svc.Store().RegisterDocID(ctx, other.ID, + db.DocRef{ID: id, Path: "rfcs/0007.md"}, "0000000000000000000000000000000000000000"); err != nil { + t.Fatalf("RegisterDocID: %v", err) + } + + old, err := sp.Repo.ApprovedHead(ctx) + if err != nil { + t.Fatal(err) + } + head := commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007.md": doc("SPEC-0007", "Storage", "body"), + }) + + err = svc.ValidatePush(ctx, PushRequest{ + Space: fxSpace, Principal: owner(), Ref: "refs/heads/main", + Old: old.String(), New: head.String(), + }) + if !errors.Is(err, ErrPushRejected) { + t.Fatalf("err = %v, want ErrPushRejected", err) + } + var rej *PushRejection + if !errors.As(err, &rej) { + t.Fatalf("err is not a *PushRejection: %T", err) + } + if len(rej.Problems) != 1 || rej.Problems[0].Kind != ProblemIDCollision { + t.Fatalf("problems = %+v", rej.Problems) + } + msg := err.Error() + for _, want := range []string{"specs/0007.md", "SPEC-0007", "~bigbes/notes", "rfcs/0007.md"} { + if !strings.Contains(msg, want) { + t.Errorf("message does not name %q:\n%s", want, msg) + } + } + if !rej.Skippable { + t.Error("an id collision must be skippable; the escape hatch exists for exactly this") + } +} + +// The escape hatch waives frontmatter and id validation. It never waives the +// refs rule. +func TestValidatePushSkipValidationCoversContentButNotTheRefsRule(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + sp := mustCreateSpace(t, svc) + + old, err := sp.Repo.ApprovedHead(ctx) + if err != nil { + t.Fatal(err) + } + head := commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007.md": []byte("no frontmatter here\n"), + }) + + req := PushRequest{ + Space: fxSpace, Principal: owner(), Ref: "refs/heads/main", + Old: old.String(), New: head.String(), + } + if err := svc.ValidatePush(ctx, req); !errors.Is(err, ErrPushRejected) { + t.Fatalf("err = %v, want the malformed document rejected", err) + } + req.SkipValidation = true + if err := svc.ValidatePush(ctx, req); err != nil { + t.Fatalf("skip-validation did not let the push through: %v", err) + } + + // Same flag, agent principal, approved branch: still refused. + req.Principal = agent() + err = svc.ValidatePush(ctx, req) + if !errors.Is(err, ErrPushRejected) { + t.Fatalf("err = %v, want the refs rule to refuse an agent", err) + } + var rej *PushRejection + if !errors.As(err, &rej) { + t.Fatalf("err is not a *PushRejection: %T", err) + } + if rej.Skippable || rej.Problems[0].Kind != ProblemRefsRule { + t.Fatalf("rejection = %+v, want an unskippable refs-rule refusal", rej) + } +} + +func TestReconcileRepairsTheRepairTable(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + sp := mustCreateSpace(t, svc) + + // Every row inserted below is "in flight" until the grace window passes; + // move the clock rather than the rows. + svc.now = func() time.Time { return time.Now().Add(2 * DefaultReconcileGrace) } + + base, err := sp.Repo.ApprovedHead(ctx) + if err != nil { + t.Fatal(err) + } + + // 1. Crash between the row insert and the branch write. + rowOnly := mustOpenProposal(t, svc, sp.ID, "row only", base.String()) + + // 2. Crash between the merge commit and the row update: a branch with + // commits of its own, already reachable from the approved head. + merged := mustOpenProposal(t, svc, sp.ID, "merged", base.String()) + cutBranch(t, sp, merged.Branch, sp.ApprovedBranch()) + commitFiles(t, sp, merged.Branch, 1, map[string][]byte{ + "specs/0007.md": doc("SPEC-0007", "Storage", "body"), + }) + branchHead, err := sp.Repo.BranchHead(ctx, merged.Branch) + if err != nil { + t.Fatal(err) + } + fastForwardApproved(t, sp, branchHead) + + // 3. An ordinary open proposal: a branch cut at its recorded base, not yet + // committed to. Its tip is trivially an ancestor of the approved head, + // and it must survive anyway — this is the state every propose passes + // through between cutting the branch and the agent's first write. + live := mustOpenProposal(t, svc, sp.ID, "live", base.String()) + cutBranch(t, sp, live.Branch, live.BaseRev) + + // 4. An orphan ref with no row at all. + cutBranch(t, sp, "proposals/9999", sp.ApprovedBranch()) + + rep, err := svc.Reconcile(ctx) + if err != nil { + t.Fatalf("Reconcile: %v", err) + } + if len(rep.Failures) != 0 { + t.Fatalf("failures = %v", rep.Failures) + } + if rep.Spaces != 1 { + t.Errorf("spaces = %d", rep.Spaces) + } + + got := map[RepairKind]int{} + for _, r := range rep.Repaired { + got[r.Kind]++ + } + want := map[RepairKind]int{RepairDeleteRow: 1, RepairDeleteRef: 1, RepairMarkMerged: 1} + for kind, n := range want { + if got[kind] != n { + t.Errorf("%s applied %d times, want %d (all: %+v)", kind, got[kind], n, rep.Repaired) + } + } + if len(rep.Reindex) != 1 { + t.Errorf("reindex = %+v, want the one never-indexed space", rep.Reindex) + } + + if _, err := svc.Store().GetProposal(ctx, rowOnly.ID); !errors.Is(err, db.ErrNotFound) { + t.Errorf("the branchless row survived: %v", err) + } + got2, err := svc.Store().GetProposal(ctx, merged.ID) + if err != nil { + t.Fatalf("GetProposal: %v", err) + } + if got2.State != core.StateMerged { + t.Errorf("state = %q, want merged", got2.State) + } + if got2.Approval != core.ApprovalPolicy { + t.Errorf("approval = %q; a repaired merge must never claim human approval", + got2.Approval) + } + stillOpen, err := svc.Store().GetProposal(ctx, live.ID) + if err != nil { + t.Fatalf("GetProposal: %v", err) + } + if stillOpen.State != core.StateOpen { + t.Errorf("a proposal whose branch was cut but never committed to was resolved as %q", + stillOpen.State) + } + + branches, err := sp.Repo.ListProposalBranches(ctx) + if err != nil { + t.Fatal(err) + } + var names []string + for _, b := range branches { + names = append(names, b.Name) + } + if len(names) != 2 { + t.Fatalf("branches = %v, want the merged and the live one", names) + } + for _, name := range names { + if name == "proposals/9999" { + t.Error("the orphan ref survived") + } + } + + // A second pass must find nothing left to do apart from the reindex flag, + // which only Phase 2 can clear. + again, err := svc.Reconcile(ctx) + if err != nil { + t.Fatalf("second Reconcile: %v", err) + } + if len(again.Repaired) != 0 || len(again.Failures) != 0 { + t.Errorf("second pass repaired %+v, failed %v", again.Repaired, again.Failures) + } +} + +// A proposal younger than the grace window is in flight, not abandoned: every +// live propose passes through "row inserted, branch not yet written". +func TestReconcileLeavesAnInFlightProposalAlone(t *testing.T) { + svc, _ := newTestService(t) + ctx := context.Background() + sp := mustCreateSpace(t, svc) + + base, err := sp.Repo.ApprovedHead(ctx) + if err != nil { + t.Fatal(err) + } + p := mustOpenProposal(t, svc, sp.ID, "in flight", base.String()) + + rep, err := svc.Reconcile(ctx) + if err != nil { + t.Fatalf("Reconcile: %v", err) + } + for _, r := range rep.Repaired { + if r.Kind == RepairDeleteRow { + t.Fatalf("deleted a proposal opened moments ago: %v", r) + } + } + if _, err := svc.Store().GetProposal(ctx, p.ID); err != nil { + t.Fatalf("the in-flight row is gone: %v", err) + } +} + +func mustCreateSpace(t *testing.T, svc *Service) *Space { + t.Helper() + sp, err := svc.CreateSpace(context.Background(), fxSpace) + if err != nil { + t.Fatalf("CreateSpace: %v", err) + } + return sp +} + +func mustOpenProposal(t *testing.T, svc *Service, spaceID int, title, base string) *db.Proposal { + t.Helper() + p, err := svc.Store().OpenProposal(context.Background(), &db.Proposal{ + SpaceID: spaceID, Title: title, BaseRev: base, + Agent: "claude-code/spec-writer", AgentSession: "8fb9c9a4", + }) + if err != nil { + t.Fatalf("OpenProposal %q: %v", title, err) + } + return p +} + +// fastForwardApproved moves the approved branch to head, standing in for the +// merge whose row update never happened. +func fastForwardApproved(t *testing.T, sp *Space, head plumbing.Hash) { + t.Helper() + repo, err := git.PlainOpen(sp.Repo.Dir()) + if err != nil { + t.Fatalf("PlainOpen: %v", err) + } + name := plumbing.NewBranchReferenceName(sp.ApprovedBranch()) + if err := repo.Storer.SetReference(plumbing.NewHashReference(name, head)); err != nil { + t.Fatalf("set %s: %v", name, err) + } +} diff --git a/service/push.go b/service/push.go new file mode 100644 index 0000000000000000000000000000000000000000..78d7c9fbca2ce3e410f08807076aa0e5af803403 --- /dev/null +++ b/service/push.go @@ -0,0 +1,440 @@ +package service + +import ( + "context" + "encoding/hex" + "fmt" + "sort" + "strings" + + "github.com/go-git/go-git/v5/plumbing" + + "sourcecraft.dev/bigbes/sr-ht-spec/authn" + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/db" + "sourcecraft.dev/bigbes/sr-ht-spec/gitx" +) + +// PushRequest is one proposed ref update, as the `update` hook sees it and +// forwards it to the daemon over the localhost RPC. +// +// Object names are hex strings rather than plumbing.Hash so hooks/ can build +// this straight from the hook's argv without importing gitx, and so an +// unparseable value is a rejection rather than something that silently becomes +// the zero hash — which the refs rule would read as "creating a branch". +type PushRequest struct { + // Space is the space being pushed to. + Space core.SpaceRef + + // Principal is who is pushing, as authn resolved the SSH key or token the + // forced command was invoked with. An anonymous principal is refused: there + // is no unauthenticated write path. + Principal authn.Principal + + // Ref is the full ref name, "refs/heads/main". + Ref string + + // Old is the value the ref currently holds. Empty or the all-zero object + // name means the ref is being created. + Old string + + // New is the value proposed. Empty or the all-zero object name means the + // ref is being deleted. + New string + + // SkipValidation carries `--push-option=skip-validation`. + // + // It waives frontmatter and document-id validation and nothing else. The + // refs rule is never skippable: the escape hatch exists so a hook bug or a + // bad schema cannot lock the owner out of their own repository, not so that + // an agent can reach the approved branch. + SkipValidation bool +} + +// PushProblemKind classifies one reason a push was refused, so the API and the +// review UI can branch on it without parsing the message. +type PushProblemKind string + +const ( + // ProblemRefsRule is the refs rule refusing this principal this ref. Never + // skippable. + ProblemRefsRule PushProblemKind = "refs-rule" + + // ProblemFrontmatter is a document whose frontmatter is missing, + // unparseable, or fails the space's schema. + ProblemFrontmatter PushProblemKind = "frontmatter" + + // ProblemDuplicateID is two documents in the pushed tree carrying one id. + ProblemDuplicateID PushProblemKind = "duplicate-id" + + // ProblemIDCollision is a document id already registered to another space. + ProblemIDCollision PushProblemKind = "id-collision" +) + +// PushProblem is one reason a push was refused. +type PushProblem struct { + // Kind is the class of failure. + Kind PushProblemKind + + // Path is the offending document, empty for a problem that is not about + // one document (the refs rule). + Path string + + // Detail is one line naming what is wrong, written for the human reading + // their terminal after a rejected `git push`. + Detail string +} + +// PushRejection is the structured refusal the `update` hook prints and exits +// non-zero on. It wraps ErrPushRejected, so callers match the class with +// errors.Is and type-assert only when they want the detail. +type PushRejection struct { + Space core.SpaceRef + Ref string + Problems []PushProblem + + // Skippable reports whether `--push-option=skip-validation` would let this + // push through. It is false whenever any problem is a refs-rule violation, + // and the message says so rather than suggesting a flag that will not help. + Skippable bool +} + +func (e *PushRejection) Is(target error) bool { return target == ErrPushRejected } + +// Error renders the rejection as the message a human sees on their terminal +// after `git push`. Every line is short enough to survive git's "remote: " +// prefix in an 80-column terminal, and the offending document is always named +// first, because "which file" is the first thing anybody wants to know. +func (e *PushRejection) Error() string { + var b strings.Builder + fmt.Fprintf(&b, "spec.sr.ht rejected this push.\n\n") + fmt.Fprintf(&b, " space: %s\n", e.Space) + fmt.Fprintf(&b, " ref: %s\n\n", e.Ref) + for _, p := range e.Problems { + if p.Path != "" { + fmt.Fprintf(&b, " %s\n %s\n", p.Path, p.Detail) + continue + } + fmt.Fprintf(&b, " %s\n", p.Detail) + } + fmt.Fprintf(&b, "\n%s. Nothing was written; the ref still points where it did.\n", + plural(len(e.Problems), "problem")) + if e.Skippable { + b.WriteString("Re-push with --push-option=skip-validation to bypass frontmatter\n") + b.WriteString("and document-id validation.\n") + } else { + b.WriteString("The refs rule cannot be bypassed: --push-option=skip-validation\n") + b.WriteString("waives frontmatter and document-id validation only.\n") + } + return b.String() +} + +func plural(n int, what string) string { + if n == 1 { + return fmt.Sprintf("1 %s", what) + } + return fmt.Sprintf("%d %ss", n, what) +} + +// ValidatePush is what the `update` hook calls, per ref, before the ref moves. +// +// It answers in two parts, and the split is the design's: +// +// 1. The refs rule — may this principal move this ref? Always checked, never +// skippable, and checked first so that a rejection for the right reason is +// not preceded by pages of schema complaints. +// 2. Frontmatter and document-id validation of everything this push changes. +// Waived by SkipValidation, because a hook bug or a bad schema must never +// be able to lock the owner out of their own repository. +// +// A nil return means the push may proceed. A *PushRejection means it must not, +// and its Error() is the text to print. Any other error is an infrastructure +// failure — Postgres down, repository unreadable — and the hook must fail +// closed on it: a rejected push is recoverable in one command, while a silently +// unvalidated one is a corruption discovered much later. +func (s *Service) ValidatePush(ctx context.Context, req PushRequest) error { + sp, err := s.OpenSpace(ctx, req.Space) + if err != nil { + return err + } + + oldHash, err := parseObjectName("old", req.Old) + if err != nil { + return err + } + newHash, err := parseObjectName("new", req.New) + if err != nil { + return err + } + + if problem := s.checkRefsRule(ctx, sp, req, oldHash, newHash); problem != nil { + return &PushRejection{ + Space: req.Space, + Ref: req.Ref, + Problems: []PushProblem{*problem}, + Skippable: false, + } + } + + // A deletion leaves no tree to validate, and skip-validation waives + // everything that is left. Both still went through the refs rule above. + if newHash.IsZero() || req.SkipValidation { + return nil + } + + problems, err := s.validateContent(ctx, sp, oldHash, newHash) + if err != nil { + return err + } + if len(problems) > 0 { + return &PushRejection{ + Space: req.Space, + Ref: req.Ref, + Problems: problems, + Skippable: true, + } + } + return nil +} + +// checkRefsRule applies gitx.CheckRefUpdate, computing the fast-forward fact it +// cannot compute itself. It returns nil when the update is permitted. +func (s *Service) checkRefsRule(ctx context.Context, sp *Space, req PushRequest, old, new plumbing.Hash) *PushProblem { + kind, err := principalKind(req.Principal) + if err != nil { + return &PushProblem{Kind: ProblemRefsRule, Detail: err.Error()} + } + + // Ancestry is only meaningful when both ends name a commit. A creation has + // no old value and a deletion has no new one; gitx treats a creation as a + // fast-forward and ignores the flag entirely for a deletion. + fastForward := old.IsZero() + if !old.IsZero() && !new.IsZero() { + ff, err := sp.Repo.IsAncestor(ctx, old, new) + if err != nil { + // Not knowing whether this is a fast-forward is not permission to + // assume it is: an unreadable object must refuse the push, not + // wave through a force-update of the approved branch. + return &PushProblem{ + Kind: ProblemRefsRule, + Detail: fmt.Sprintf("cannot determine whether %s..%s is a fast-forward: %v", old, new, err), + } + } + fastForward = ff + } + + err = gitx.CheckRefUpdate(kind, sp.ApprovedBranch(), gitx.RefUpdate{ + Ref: req.Ref, + Old: old, + New: new, + FastForward: fastForward, + }) + if err != nil { + return &PushProblem{Kind: ProblemRefsRule, Detail: err.Error()} + } + return nil +} + +// validateContent validates the frontmatter of every document this push changes +// and checks document-id uniqueness, both within the pushed tree and against +// the global registry. +func (s *Service) validateContent(ctx context.Context, sp *Space, old, new plumbing.Hash) ([]PushProblem, error) { + all, changed, err := s.changedDocuments(ctx, sp, old, new) + if err != nil { + return nil, err + } + + // The schema is read at the *new* revision, so a push that edits .spec.yml + // is validated against the policy it is installing. Validating against the + // old one would make a schema change and the documents that satisfy it + // impossible to land in a single push. + policy, err := s.Policy(ctx, sp, new.String()) + if err != nil { + return nil, err + } + + problems, refs := validateDocuments(all, changed, policy.Schema) + + collisions, err := s.store.CheckDocIDCollisions(ctx, sp.ID, refs) + if err != nil { + return nil, fmt.Errorf("service: check document id collisions for %s: %w", sp.Ref, err) + } + byID := make(map[string]string, len(refs)) + for _, r := range refs { + byID[r.ID.String()] = r.Path + } + for _, c := range collisions { + owner, err := s.store.GetSpaceByID(ctx, c.Existing.SpaceID) + if err != nil { + return nil, fmt.Errorf("service: resolve space %d holding document id %s: %w", + c.Existing.SpaceID, c.DocID, err) + } + problems = append(problems, PushProblem{ + Kind: ProblemIDCollision, + Path: byID[c.DocID.String()], + Detail: fmt.Sprintf("id %s is already registered to %s at %s", + c.DocID, owner.Ref, c.Existing.Path), + }) + } + sort.SliceStable(problems, func(i, j int) bool { return problems[i].Path < problems[j].Path }) + return problems, nil +} + +// changedDocuments returns every document at the new revision, and the subset +// of them this push changes. +// +// The baseline is the ref's old value, or — when the ref is being created — the +// space's approved head. A brand-new proposal branch is cut from the approved +// branch, so comparing it against nothing would revalidate the entire space and +// let one document that was pushed with --push-option=skip-validation block +// every future proposal branch. +// +// Only the changed subset is schema-validated, for the same reason. Malformed +// documents already on a branch are tolerated rather than fatal: a single typo +// must not become an outage that blocks every later push. +func (s *Service) changedDocuments(ctx context.Context, sp *Space, old, new plumbing.Hash) (all, changed []Document, err error) { + all, err = s.ListDocuments(ctx, sp, new.String()) + if err != nil { + return nil, nil, err + } + + baseline := old.String() + if old.IsZero() { + baseline = ApprovedRev + } + before, err := s.ListDocuments(ctx, sp, baseline) + if err != nil { + return nil, nil, err + } + prior := make(map[string]string, len(before)) + for _, d := range before { + prior[d.Path] = d.Blob + } + + for _, d := range all { + if prior[d.Path] != d.Blob { + changed = append(changed, d) + } + } + return all, changed, nil +} + +// validateDocuments is the whole of push validation that needs neither git nor +// Postgres: schema conformance of the changed documents, and id uniqueness +// within the pushed tree. It returns the problems it found and the (id, path) +// refs of the changed documents, which is what the registry check runs against. +// +// A duplicate id is reported only when at least one of the documents carrying +// it is part of this push. Two colliding documents that were both already there +// are somebody's earlier skip-validation typo; rejecting every subsequent push +// until they are fixed would turn a cosmetic error into a lockout, and the fix +// itself would be unpushable. +func validateDocuments(all, changed []Document, schema core.Schema) ([]PushProblem, []db.DocRef) { + byID := make(map[string][]string) + for _, d := range all { + fm, _, err := core.ParseDocument(d.Data) + if err != nil { + continue // reported below if this document is part of the push + } + if id, err := core.ParseDocID(fm.ID); err == nil { + byID[id.String()] = append(byID[id.String()], d.Path) + } + } + + var problems []PushProblem + var refs []db.DocRef + for _, d := range changed { + fm, _, err := core.ParseDocument(d.Data) + if err != nil { + problems = append(problems, PushProblem{ + Kind: ProblemFrontmatter, + Path: d.Path, + Detail: err.Error(), + }) + continue + } + if err := schema.ValidateFrontmatter(fm); err != nil { + problems = append(problems, PushProblem{ + Kind: ProblemFrontmatter, + Path: d.Path, + Detail: err.Error(), + }) + continue + } + id, err := core.ParseDocID(fm.ID) + if err != nil { + // Reachable only when the space's schema does not require `id`. + // Such a document is unregistrable but not malformed, so it is not + // a problem — it simply contributes nothing to the registry. + continue + } + if others := without(byID[id.String()], d.Path); len(others) > 0 { + problems = append(problems, PushProblem{ + Kind: ProblemDuplicateID, + Path: d.Path, + Detail: fmt.Sprintf("id %s is also carried by %s", + id, strings.Join(others, ", ")), + }) + } + refs = append(refs, db.DocRef{ID: id, Path: d.Path}) + } + sort.SliceStable(problems, func(i, j int) bool { return problems[i].Path < problems[j].Path }) + return problems, refs +} + +// without returns paths with one occurrence of self removed. +func without(paths []string, self string) []string { + out := make([]string, 0, len(paths)) + dropped := false + for _, p := range paths { + if p == self && !dropped { + dropped = true + continue + } + out = append(out, p) + } + if len(out) == 0 { + return nil + } + return out +} + +// principalKind maps a resolved identity onto the two principals the refs rule +// knows about. An anonymous principal is refused rather than mapped to either: +// there is no unauthenticated write path, and defaulting it to "agent" would +// give an unidentified pusher the proposal namespace. +func principalKind(p authn.Principal) (gitx.PrincipalKind, error) { + switch { + case p.IsOwner(): + return gitx.PrincipalHuman, nil + case p.IsAgent(): + return gitx.PrincipalAgent, nil + default: + return "", fmt.Errorf("no credential identifies this push; %s may not write any ref", p) + } +} + +// zeroObjectName is git's "this ref does not exist" sentinel as the hook spells +// it on the command line. +const zeroObjectName = "0000000000000000000000000000000000000000" + +// parseObjectName converts a hook's hex argument into an object name. The empty +// string and the all-zero name both mean "absent". +// +// plumbing.NewHash is deliberately not used: it maps anything unparseable to +// the zero hash, which the refs rule would read as a branch creation or a +// deletion. A malformed argument is a bug in whatever built the request, and it +// fails here rather than becoming a permitted force-push. +func parseObjectName(which, s string) (plumbing.Hash, error) { + if s == "" || s == zeroObjectName { + return plumbing.ZeroHash, nil + } + if len(s) != len(zeroObjectName) { + return plumbing.ZeroHash, fmt.Errorf("service: %s object name %q is not %d hex digits", + which, s, len(zeroObjectName)) + } + if _, err := hex.DecodeString(s); err != nil { + return plumbing.ZeroHash, fmt.Errorf("service: %s object name %q is not hex: %w", which, s, err) + } + return plumbing.NewHash(s), nil +} diff --git a/service/push_test.go b/service/push_test.go new file mode 100644 index 0000000000000000000000000000000000000000..30991827005c1e580146c5866db158fd80d67062 --- /dev/null +++ b/service/push_test.go @@ -0,0 +1,349 @@ +package service + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/go-git/go-git/v5/plumbing" + + "sourcecraft.dev/bigbes/sr-ht-spec/authn" + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/gitx" +) + +func owner() authn.Principal { + return authn.Principal{Kind: authn.KindOwner, Owner: "bigbes"} +} + +func agent() authn.Principal { + return authn.Principal{ + Kind: authn.KindAgent, Owner: "bigbes", + Agent: "claude-code/spec-writer", Session: "8fb9c9a4", + } +} + +func TestValidateDocuments(t *testing.T) { + schema := core.DefaultSchema() + spec7 := Document{Path: "specs/0007.md", Blob: "a", Data: doc("SPEC-0007", "Storage", "body")} + spec8 := Document{Path: "specs/0008.md", Blob: "b", Data: doc("SPEC-0008", "Other", "body")} + dupe := Document{Path: "notes/copy.md", Blob: "c", Data: doc("SPEC-0007", "Copy", "body")} + noStatus := Document{Path: "specs/0009.md", Blob: "d", + Data: []byte("---\nid: SPEC-0009\ntitle: No status\n---\n\nbody\n")} + noFrontmatter := Document{Path: "specs/0010.md", Blob: "e", Data: []byte("# just a heading\n")} + badID := Document{Path: "specs/0011.md", Blob: "f", + Data: doc("spec-11", "Lowercase id", "body")} + + tests := []struct { + name string + all []Document + changed []Document + wantKind []PushProblemKind + wantIn []string + wantRefs []string + }{ + { + name: "clean push", + all: []Document{spec7, spec8}, + changed: []Document{spec8}, + wantRefs: []string{"SPEC-0008"}, + }, + { + name: "missing required key", + all: []Document{noStatus}, + changed: []Document{noStatus}, + wantKind: []PushProblemKind{ProblemFrontmatter}, + wantIn: []string{"status"}, + }, + { + name: "no frontmatter at all", + all: []Document{noFrontmatter}, + changed: []Document{noFrontmatter}, + wantKind: []PushProblemKind{ProblemFrontmatter}, + wantIn: []string{"frontmatter"}, + }, + { + name: "malformed id", + all: []Document{badID}, + changed: []Document{badID}, + wantKind: []PushProblemKind{ProblemFrontmatter}, + wantIn: []string{"document id"}, + }, + { + // The failure mode the whole escape hatch exists to guard: a + // duplicated id: corrupts the global registry. + name: "duplicate id introduced by this push", + all: []Document{dupe, spec7}, + changed: []Document{dupe}, + wantKind: []PushProblemKind{ProblemDuplicateID}, + wantIn: []string{"SPEC-0007", "specs/0007.md"}, + wantRefs: []string{"SPEC-0007"}, + }, + { + // Both halves of the collision were already there, pushed under + // skip-validation. Rejecting every later push would turn a + // cosmetic error into a lockout — and the fix unpushable. + name: "pre-existing duplicate untouched by this push", + all: []Document{dupe, spec7, spec8}, + changed: []Document{spec8}, + wantRefs: []string{"SPEC-0008"}, + }, + { + // A malformed document already on the branch is tolerated, not + // fatal: it must not block every future push. + name: "malformed document not part of this push", + all: []Document{noFrontmatter, spec8}, + changed: []Document{spec8}, + wantRefs: []string{"SPEC-0008"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + problems, refs := validateDocuments(tc.all, tc.changed, schema) + if len(problems) != len(tc.wantKind) { + t.Fatalf("problems = %+v, want %d", problems, len(tc.wantKind)) + } + for i, kind := range tc.wantKind { + if problems[i].Kind != kind { + t.Errorf("problem %d kind = %q, want %q", i, problems[i].Kind, kind) + } + } + for _, want := range tc.wantIn { + found := false + for _, p := range problems { + if strings.Contains(p.Detail, want) { + found = true + } + } + if !found { + t.Errorf("no problem mentions %q: %+v", want, problems) + } + } + var got []string + for _, r := range refs { + got = append(got, r.ID.String()) + } + if strings.Join(got, ",") != strings.Join(tc.wantRefs, ",") { + t.Errorf("refs = %v, want %v", got, tc.wantRefs) + } + }) + } +} + +// The message is what a human reads in their terminal after a rejected push, so +// it is a product surface and is asserted as one. +func TestPushRejectionMessage(t *testing.T) { + rej := &PushRejection{ + Space: fxSpace, + Ref: "refs/heads/main", + Problems: []PushProblem{ + {Kind: ProblemFrontmatter, Path: "specs/0007.md", + Detail: `missing required frontmatter field: "status"`}, + {Kind: ProblemIDCollision, Path: "notes/x.md", + Detail: "id RFC-0001 is already registered to ~bigbes/notes at rfcs/0001.md"}, + }, + Skippable: true, + } + msg := rej.Error() + for _, want := range []string{ + "~bigbes/rfcs", "refs/heads/main", + "specs/0007.md", `missing required frontmatter field: "status"`, + "notes/x.md", "RFC-0001", "~bigbes/notes", + "2 problems", "--push-option=skip-validation", + } { + if !strings.Contains(msg, want) { + t.Errorf("message does not contain %q:\n%s", want, msg) + } + } + if !errors.Is(rej, ErrPushRejected) { + t.Error("rejection does not satisfy errors.Is(err, ErrPushRejected)") + } + for _, line := range strings.Split(msg, "\n") { + if len(line) > 72 { + t.Errorf("line is %d chars, too long once git prefixes it with %q: %s", + len(line), "remote: ", line) + } + } +} + +// A refs-rule rejection must not suggest a flag that will not help. +func TestPushRejectionNeverOffersToSkipTheRefsRule(t *testing.T) { + rej := &PushRejection{ + Space: fxSpace, + Ref: "refs/heads/main", + Problems: []PushProblem{{Kind: ProblemRefsRule, Detail: "agents may only write proposals/*"}}, + } + msg := rej.Error() + if strings.Contains(msg, "Re-push with") { + t.Errorf("refs-rule rejection offers the escape hatch:\n%s", msg) + } + if !strings.Contains(msg, "cannot be bypassed") { + t.Errorf("refs-rule rejection does not say the rule is absolute:\n%s", msg) + } + if !strings.Contains(msg, "1 problem") { + t.Errorf("singular problem count is wrong:\n%s", msg) + } +} + +func TestParseObjectName(t *testing.T) { + tests := []struct { + in string + wantErr bool + wantNil bool // resolves to the zero object name + }{ + {in: "", wantNil: true}, + {in: zeroObjectName, wantNil: true}, + {in: "1f0c1d1a1e2b3c4d5e6f708192a3b4c5d6e7f809"}, + {in: "1f0c1d1a", wantErr: true}, + {in: "1f0c1d1a1e2b3c4d5e6f708192a3b4c5d6e7f80z", wantErr: true}, + {in: "refs/heads/main", wantErr: true}, + } + for _, tc := range tests { + got, err := parseObjectName("new", tc.in) + if tc.wantErr != (err != nil) { + t.Errorf("parseObjectName(%q) err = %v, wantErr %v", tc.in, err, tc.wantErr) + continue + } + if err == nil && got.IsZero() != tc.wantNil { + t.Errorf("parseObjectName(%q) = %s, wantZero %v", tc.in, got, tc.wantNil) + } + } +} + +// Anonymous must not default to either principal: there is no unauthenticated +// write path, and mapping it to "agent" would hand it the proposal namespace. +func TestPrincipalKind(t *testing.T) { + if got, err := principalKind(owner()); err != nil || got != gitx.PrincipalHuman { + t.Errorf("owner -> %q, %v", got, err) + } + if got, err := principalKind(agent()); err != nil || got != gitx.PrincipalAgent { + t.Errorf("agent -> %q, %v", got, err) + } + if _, err := principalKind(authn.Anonymous()); err == nil { + t.Error("anonymous resolved to a writing principal") + } +} + +func TestCheckRefsRule(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + first := commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007.md": doc("SPEC-0007", "Storage", "one"), + }) + second := commitFiles(t, sp, sp.ApprovedBranch(), 2, map[string][]byte{ + "specs/0007.md": doc("SPEC-0007", "Storage", "two"), + }) + cutBranch(t, sp, "proposals/1", sp.ApprovedBranch()) + branch := commitFiles(t, sp, "proposals/1", 3, map[string][]byte{ + "specs/0008.md": doc("SPEC-0008", "Other", "one"), + }) + + tests := []struct { + name string + principal authn.Principal + ref string + old, new string + wantOK bool + }{ + {"owner fast-forwards the approved branch", owner(), "refs/heads/main", + first.String(), second.String(), true}, + {"owner force-pushes the approved branch", owner(), "refs/heads/main", + second.String(), first.String(), false}, + {"owner deletes the approved branch", owner(), "refs/heads/main", + second.String(), zeroObjectName, false}, + {"agent moves the approved branch", agent(), "refs/heads/main", + first.String(), second.String(), false}, + {"agent writes a proposal branch", agent(), "refs/heads/proposals/1", + zeroObjectName, branch.String(), true}, + {"agent force-updates its own proposal branch", agent(), "refs/heads/proposals/1", + branch.String(), first.String(), true}, + {"owner pushes a tag", owner(), "refs/tags/v1", zeroObjectName, second.String(), false}, + {"owner pushes an unrelated branch", owner(), "refs/heads/scratch", + zeroObjectName, second.String(), false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + old, err := parseObjectName("old", tc.old) + if err != nil { + t.Fatal(err) + } + newHash, err := parseObjectName("new", tc.new) + if err != nil { + t.Fatal(err) + } + problem := svc.checkRefsRule(ctx, sp, PushRequest{ + Principal: tc.principal, Ref: tc.ref, + }, old, newHash) + if tc.wantOK != (problem == nil) { + t.Fatalf("problem = %+v, wantOK %v", problem, tc.wantOK) + } + if problem != nil && problem.Kind != ProblemRefsRule { + t.Errorf("kind = %q", problem.Kind) + } + }) + } +} + +// A new proposal branch is compared against the approved head, not against +// nothing: comparing against nothing would revalidate the whole space and let +// one skip-validation typo block every future proposal branch. +func TestChangedDocumentsBaselinesANewBranchOnTheApprovedHead(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007.md": doc("SPEC-0007", "Storage", "one"), + "specs/0009.md": []byte("not a document at all\n"), + }) + cutBranch(t, sp, "proposals/1", sp.ApprovedBranch()) + branch := commitFiles(t, sp, "proposals/1", 2, map[string][]byte{ + "specs/0008.md": doc("SPEC-0008", "Other", "one"), + }) + + all, changed, err := svc.changedDocuments(ctx, sp, zeroHash(t), branch) + if err != nil { + t.Fatalf("changedDocuments: %v", err) + } + if len(all) != 3 { + t.Errorf("all = %d documents, want 3", len(all)) + } + if len(changed) != 1 || changed[0].Path != "specs/0008.md" { + t.Fatalf("changed = %+v, want only specs/0008.md", changed) + } +} + +func TestChangedDocumentsUsesTheOldRefValueWhenThereIsOne(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + old := commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007.md": doc("SPEC-0007", "Storage", "one"), + "specs/0008.md": doc("SPEC-0008", "Other", "one"), + }) + newHead := commitFiles(t, sp, sp.ApprovedBranch(), 2, map[string][]byte{ + "specs/0008.md": doc("SPEC-0008", "Other", "two"), + }) + + _, changed, err := svc.changedDocuments(ctx, sp, old, newHead) + if err != nil { + t.Fatalf("changedDocuments: %v", err) + } + if len(changed) != 1 || changed[0].Path != "specs/0008.md" { + t.Fatalf("changed = %+v, want only specs/0008.md", changed) + } +} + +func zeroHash(t *testing.T) plumbing.Hash { + t.Helper() + h, err := parseObjectName("old", "") + if err != nil { + t.Fatal(err) + } + return h +} diff --git a/service/read.go b/service/read.go new file mode 100644 index 0000000000000000000000000000000000000000..9bdb593f4e92a5f5c501efaa8087a4e98a6263f5 --- /dev/null +++ b/service/read.go @@ -0,0 +1,177 @@ +package service + +import ( + "context" + "errors" + "fmt" + + "github.com/go-git/go-git/v5/plumbing" + + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/gitx" +) + +// ApprovedRev is the revision string meaning "the space's approved head". It is +// the empty string so that a caller which simply forwards an absent ?rev= gets +// the approved revision by default, which is the read contract: reads default +// to the approved revision, because serving drafts by default would poison +// every downstream agent context with unreviewed text. +const ApprovedRev = "" + +// Document is one document as it exists at a revision. +// +// Blob and Rev are hex object names rather than plumbing.Hash so that api/, +// mcpsrv/, graph/ and web/ can carry them without importing gitx — the layering +// rule is that nothing above service/ touches the git layer, and a leaked +// plumbing type would break it on the first struct field. +type Document struct { + // Path is the document's path in the tree. + Path string + + // Blob is the sha of the document's blob — the render cache key. It is + // content-addressed, so a cache entry keyed by it can never go stale. + Blob string + + // Rev is the commit the read resolved to. For a read at the approved head + // this is the value to hand back as the pinned ?rev=, and it is the same + // value an agent sends as If-Match. + Rev string + + // Data is the whole document: frontmatter and body. + Data []byte +} + +// ReadDocument reads one document by path, at the approved head when rev is +// ApprovedRev and at a pinned revision otherwise. +// +// This is the same code path for both. There is one storage tier and no +// checkout, so "the approved text of SPEC-0007" and "SPEC-0007 at +// 1f0c1d1a" differ only in which revision is resolved. +func (s *Service) ReadDocument(ctx context.Context, sp *Space, rev, path string) (Document, error) { + commit, resolved, err := s.resolveRev(ctx, sp, rev) + if err != nil { + return Document{}, err + } + doc, err := sp.Repo.ReadDocument(ctx, resolved, path) + if err != nil { + return Document{}, readErr(err, "read %s at %s in %s", path, resolved, sp.Ref) + } + return Document{ + Path: doc.Path, + Blob: doc.Blob.String(), + Rev: commit.String(), + Data: doc.Data, + }, nil +} + +// ListDocuments returns every document in a space at a revision, in tree order. +// +// Bodies are included: they come off the same tree walk, the volume is tens of +// documents a day, and every caller that lists documents (the indexer, the +// review page, the ID map a push validation builds) needs the frontmatter, +// which is not separable from the blob. +func (s *Service) ListDocuments(ctx context.Context, sp *Space, rev string) ([]Document, error) { + commit, resolved, err := s.resolveRev(ctx, sp, rev) + if err != nil { + return nil, err + } + docs, err := sp.Repo.ListDocuments(ctx, resolved) + if err != nil { + return nil, readErr(err, "list documents at %s in %s", resolved, sp.Ref) + } + out := make([]Document, 0, len(docs)) + for _, d := range docs { + out = append(out, Document{ + Path: d.Path, + Blob: d.Blob.String(), + Rev: commit.String(), + Data: d.Data, + }) + } + return out, nil +} + +// Policy reads the space's effective .spec.yml at a revision. +// +// A space with no .spec.yml gets core.DefaultPolicy: the house frontmatter +// contract and nothing auto-merged. That is the fail-closed direction — a space +// that has not said anything about review must not be quietly laundering +// unreviewed agent output onto the approved branch — and it is a defined +// default rather than a fallback, which is why an absent file is not an error +// but an unparseable one is. +// +// Reading it at a revision rather than from configuration is what makes policy +// changes reviewable like any other change, and it is why a push that edits +// .spec.yml is validated against the policy it is installing. +func (s *Service) Policy(ctx context.Context, sp *Space, rev string) (core.Policy, error) { + _, resolved, err := s.resolveRev(ctx, sp, rev) + if err != nil { + return core.Policy{}, err + } + data, _, err := sp.Repo.ReadBlob(ctx, resolved, core.PolicyFile) + if err != nil { + if errors.Is(err, gitx.ErrNotFound) { + return core.DefaultPolicy(), nil + } + return core.Policy{}, readErr(err, "read %s at %s in %s", + core.PolicyFile, resolved, sp.Ref) + } + pol, err := core.ParsePolicy(data) + if err != nil { + return core.Policy{}, fmt.Errorf("service: %s at %s in %s: %w", + core.PolicyFile, resolved, sp.Ref, err) + } + return pol, nil +} + +// ResolveRev resolves a revision string against a space, returning the commit +// it names as a hex object name. ApprovedRev resolves to the approved head. +// +// Callers use it to pin: the review UI turns "the approved head right now" into +// an immutable ?rev= before it renders anything, so a merge landing mid-render +// cannot make one page describe two revisions. +func (s *Service) ResolveRev(ctx context.Context, sp *Space, rev string) (string, error) { + commit, _, err := s.resolveRev(ctx, sp, rev) + if err != nil { + return "", err + } + return commit.String(), nil +} + +// resolveRev turns a caller's revision string into both the commit it names and +// the string to pass back down to gitx. +// +// Both are returned because they are not interchangeable: the hash is what a +// caller pins and compares, while the original string is what the read is +// issued against. Re-issuing reads against the resolved hash instead would be +// one extra object lookup per read for no gain, and would lose the branch name +// from error messages. +func (s *Service) resolveRev(ctx context.Context, sp *Space, rev string) (plumbing.Hash, string, error) { + if sp == nil || sp.Repo == nil { + return plumbing.ZeroHash, "", errors.New("service: space has no open repository") + } + resolved := rev + if resolved == ApprovedRev { + resolved = sp.Repo.ApprovedBranch() + } + commit, err := sp.Repo.ResolveRev(ctx, resolved) + if err != nil { + return plumbing.ZeroHash, "", readErr(err, "resolve revision %q in %s", resolved, sp.Ref) + } + return commit, resolved, nil +} + +// readErr maps a gitx failure onto this package's sentinels so callers above +// service/ can branch on it without importing gitx. ErrNotFound and ErrBadRev +// both become ErrNotFound at this boundary — a crafted revision must not be +// able to tell "malformed" from "absent" by probing — while the original class +// stays in the chain for logs and for gitx-aware callers. +func readErr(err error, format string, args ...any) error { + what := fmt.Sprintf(format, args...) + switch { + case errors.Is(err, gitx.ErrNotFound), errors.Is(err, gitx.ErrBadRev): + return fmt.Errorf("%w: %s: %w", ErrNotFound, what, err) + default: + return fmt.Errorf("service: %s: %w", what, err) + } +} diff --git a/service/read_test.go b/service/read_test.go new file mode 100644 index 0000000000000000000000000000000000000000..d0aa915709e48b590cec8c2a440159b186f30af1 --- /dev/null +++ b/service/read_test.go @@ -0,0 +1,213 @@ +package service + +import ( + "context" + "errors" + "strings" + "testing" + + "sourcecraft.dev/bigbes/sr-ht-spec/core" +) + +func TestReadDocumentResolvesTheApprovedHeadByDefault(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + first := commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007-storage.md": doc("SPEC-0007", "Storage model", "first"), + }) + second := commitFiles(t, sp, sp.ApprovedBranch(), 2, map[string][]byte{ + "specs/0007-storage.md": doc("SPEC-0007", "Storage model", "second"), + }) + + got, err := svc.ReadDocument(ctx, sp, ApprovedRev, "specs/0007-storage.md") + if err != nil { + t.Fatalf("ReadDocument: %v", err) + } + if !contains(got.Data, "second") { + t.Errorf("approved read returned %q, want the newest revision", got.Data) + } + if got.Rev != second.String() { + t.Errorf("Rev = %s, want the resolved approved head %s", got.Rev, second) + } + if got.Blob == "" { + t.Error("Blob is empty; it is the render cache key") + } + + // The same call with a pinned revision is the same code path with a + // different revision — one storage tier, no checkout. + pinned, err := svc.ReadDocument(ctx, sp, first.String(), "specs/0007-storage.md") + if err != nil { + t.Fatalf("pinned ReadDocument: %v", err) + } + if !contains(pinned.Data, "first") { + t.Errorf("pinned read returned %q, want the pinned revision", pinned.Data) + } + if pinned.Blob == got.Blob { + t.Error("two revisions of one document share a blob sha") + } +} + +// A draft must never be served by default: doing so would poison every +// downstream agent context with unreviewed text. +func TestReadDocumentDoesNotServeAProposalByDefault(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007-storage.md": doc("SPEC-0007", "Storage model", "approved"), + }) + cutBranch(t, sp, "proposals/1", sp.ApprovedBranch()) + commitFiles(t, sp, "proposals/1", 2, map[string][]byte{ + "specs/0007-storage.md": doc("SPEC-0007", "Storage model", "draft"), + }) + + approved, err := svc.ReadDocument(ctx, sp, ApprovedRev, "specs/0007-storage.md") + if err != nil { + t.Fatalf("ReadDocument: %v", err) + } + if !contains(approved.Data, "approved") { + t.Fatalf("default read returned the draft: %q", approved.Data) + } + draft, err := svc.ReadDocument(ctx, sp, "proposals/1", "specs/0007-storage.md") + if err != nil { + t.Fatalf("ReadDocument at the proposal branch: %v", err) + } + if !contains(draft.Data, "draft") { + t.Errorf("proposal read returned %q", draft.Data) + } +} + +func TestReadDocumentMapsAbsenceOntoErrNotFound(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + tests := []struct { + name string + rev string + path string + }{ + {"missing document", ApprovedRev, "specs/nope.md"}, + {"unknown revision", "deadbeefdeadbeefdeadbeefdeadbeefdeadbeef", "specs/nope.md"}, + // Revision arithmetic is not a usable revision; it must not be + // distinguishable from an absent one by probing. + {"revision arithmetic", "main^2", "specs/nope.md"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if _, err := svc.ReadDocument(ctx, sp, tc.rev, tc.path); !errors.Is(err, ErrNotFound) { + t.Fatalf("err = %v, want ErrNotFound", err) + } + }) + } +} + +func TestListDocumentsReturnsEveryDocumentAtARevision(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007-storage.md": doc("SPEC-0007", "Storage", "body"), + "notes/scratch.md": doc("NOTE-0001", "Scratch", "body"), + "attachment.png": []byte("\x89PNG not a document"), + core.PolicyFile: []byte("review:\n auto_merge: [notes/**]\n"), + }) + + docs, err := svc.ListDocuments(ctx, sp, ApprovedRev) + if err != nil { + t.Fatalf("ListDocuments: %v", err) + } + var paths []string + for _, d := range docs { + paths = append(paths, d.Path) + } + want := []string{"notes/scratch.md", "specs/0007-storage.md"} + if len(paths) != len(want) { + t.Fatalf("paths = %v, want %v (documents only)", paths, want) + } + for i := range want { + if paths[i] != want[i] { + t.Fatalf("paths = %v, want %v", paths, want) + } + } +} + +func TestPolicyReadsTheVersionedSpecYml(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + // A space with no .spec.yml gets the house contract and nothing + // auto-merged: the fail-closed direction. + pol, err := svc.Policy(ctx, sp, ApprovedRev) + if err != nil { + t.Fatalf("Policy: %v", err) + } + if pol.AutoMerges("notes/x.md") { + t.Error("a space with no policy auto-merged a path") + } + if len(pol.Schema.Required) == 0 { + t.Error("default policy carries no required keys") + } + + before := commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + core.PolicyFile: []byte("review:\n auto_merge: [notes/**]\n"), + }) + pol, err = svc.Policy(ctx, sp, ApprovedRev) + if err != nil { + t.Fatalf("Policy: %v", err) + } + if !pol.AutoMerges("notes/x.md") || pol.AutoMerges("specs/x.md") { + t.Errorf("auto_merge = %v", pol.Review.AutoMerge) + } + + // Policy is read at a revision, which is what makes a policy change + // reviewable like any other change. + commitFiles(t, sp, sp.ApprovedBranch(), 2, map[string][]byte{ + core.PolicyFile: []byte("review:\n auto_merge: []\n"), + }) + pinned, err := svc.Policy(ctx, sp, before.String()) + if err != nil { + t.Fatalf("pinned Policy: %v", err) + } + if !pinned.AutoMerges("notes/x.md") { + t.Error("a pinned policy read returned the newer policy") + } +} + +func TestPolicyFailsOnAnUnparseableSpecYml(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + + commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + core.PolicyFile: []byte("review:\n auto-merge: [notes/**]\n"), + }) + _, err := svc.Policy(context.Background(), sp, ApprovedRev) + if !errors.Is(err, core.ErrInvalidPolicy) { + t.Fatalf("err = %v, want core.ErrInvalidPolicy — a typo that silently does nothing is worse", err) + } +} + +func TestResolveRevPinsTheApprovedHead(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + + head := commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007-storage.md": doc("SPEC-0007", "Storage", "body"), + }) + got, err := svc.ResolveRev(context.Background(), sp, ApprovedRev) + if err != nil { + t.Fatalf("ResolveRev: %v", err) + } + if got != head.String() { + t.Errorf("ResolveRev = %s, want %s", got, head) + } +} + +func contains(data []byte, want string) bool { + return strings.Contains(string(data), want) +} diff --git a/service/reconcile.go b/service/reconcile.go new file mode 100644 index 0000000000000000000000000000000000000000..4911e84eb3947b4444b9d4b09c6bc1ddae2451f4 --- /dev/null +++ b/service/reconcile.go @@ -0,0 +1,570 @@ +package service + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/go-git/go-git/v5" + "github.com/go-git/go-git/v5/plumbing" + + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/db" + "sourcecraft.dev/bigbes/sr-ht-spec/gitx" +) + +// Three systems are touched by a merge — git refs, the bleve index and +// Postgres — and none of it is transactional. The rule that makes it tractable: +// +// Git refs are the source of truth for whether a proposal has merged. The +// Postgres row is the source of truth for that a proposal exists and what it +// is. The index and the render cache are pure caches. +// +// The reconciler is the backstop that repairs divergence, and it implements +// exactly four repairs — no more, because every additional guess about what a +// half-finished write meant is a way to invent state nobody wrote. + +const ( + // DefaultReconcileInterval is how often the reconciler runs after startup. + DefaultReconcileInterval = 15 * time.Minute + + // DefaultReconcileGrace is how long a proposal row with no branch is left + // alone before it is deleted. + // + // The design's repair table has no grace period, and without one the + // reconciler is actively destructive during ordinary operation: a proposal + // is opened row-first, so every live propose passes through the exact state + // ("open row, no branch") that the table says to delete. The window is + // milliseconds wide and the reconciler runs on a timer, so it would be rare + // — which makes it worse, not better, since it would destroy an agent's + // work at random and never in a test. + // + // A row younger than this is therefore assumed to be in flight rather than + // abandoned. It costs one extra reconcile cycle before a genuinely crashed + // proposal is cleaned up, which nothing is waiting on. + DefaultReconcileGrace = 5 * time.Minute +) + +// RepairKind names one of the four repairs. +type RepairKind string + +const ( + // RepairDeleteRow removes an `open` proposal row whose branch does not + // exist: the daemon died between the row insert and the branch write. The + // row holds no content, and the agent still holds the document it wanted to + // write, so it re-proposes. + RepairDeleteRow RepairKind = "delete-proposal-row" + + // RepairDeleteRef removes a proposals/* ref with no row. It is unreferenced + // — the id is a Postgres serial, and title, rationale, base_rev, agent and + // agent_session live nowhere in a ref — so its content is unrecoverable + // anyway and recreating the row would mean inventing every field. + RepairDeleteRef RepairKind = "delete-proposal-ref" + + // RepairMarkMerged transitions a row still `open` whose branch has merged + // into the approved head. The ref is truth for merged-ness. + RepairMarkMerged RepairKind = "mark-proposal-merged" + + // RepairReindex flags a space whose index stamp differs from its approved + // head. Phase 2 owns the rebuild; the reconciler only reports the list. + RepairReindex RepairKind = "reindex-space" +) + +// Repair is one repair the reconciler decided on. +type Repair struct { + Kind RepairKind + Space core.SpaceRef + SpaceID int + + // ProposalID is the proposal row's id, and zero for RepairReindex and for + // an orphan ref whose name carries no parseable id. + ProposalID int + + // Branch is the proposal branch this repair is about, empty for + // RepairReindex. + Branch string + + // Rev is the revision the repair records: the merge revision for + // RepairMarkMerged, the approved head for RepairReindex. + Rev string + + // Approval is the approval kind RepairMarkMerged records. See + // PlanRepairs for why it is always core.ApprovalPolicy. + Approval core.Approval + + // Reason is a human-readable sentence for the log. + Reason string +} + +func (r Repair) String() string { + return fmt.Sprintf("%s %s: %s", r.Kind, r.Space, r.Reason) +} + +// ProposalFact is what the reconciler observed about one proposal — its row, +// its branch, or both. Facts are gathered by I/O and consumed by PlanRepairs, +// which is pure so that the decision table can be exhaustively tested without a +// repository or a database. +type ProposalFact struct { + // ID is the proposal row id, and the id parsed out of the branch name when + // there is no row. Zero when the branch name carries no parseable id. + ID int + + // Branch is the proposal branch name, "proposals/42". + Branch string + + // HasRow and HasBranch record which halves exist. Both false is not a fact. + HasRow bool + HasBranch bool + + // State is the row's state, meaningless when HasRow is false. + State core.ProposalState + + // Created is when the row was inserted, for the grace window. + Created time.Time + + // MergedIntoApproved reports whether the branch tip is reachable from the + // approved head. Computed by I/O (it needs the object database) and passed + // in, exactly as gitx.RefUpdate.FastForward is. + MergedIntoApproved bool + + // BranchHead is the branch tip. + BranchHead string + + // BaseRev is the proposal's recorded base, resolved to an object name, and + // empty when it could not be resolved. See PlanRepairs for why a branch + // still sitting on its base is not a merge. + BaseRev string +} + +// SpaceFacts is everything the reconciler observed about one space. +type SpaceFacts struct { + Space core.SpaceRef + SpaceID int + + // ApprovedHead is the current tip of the approved branch. + ApprovedHead string + + // IndexRev is the revision the global index currently reflects for this + // space, empty when the space has never been indexed. Empty is stale by + // construction, which is why a missing stamp is not treated as up to date. + IndexRev string + + Proposals []ProposalFact + + // Now and Grace parameterize the grace window, so it is an input to the + // decision rather than a clock read inside it. + Now time.Time + Grace time.Duration +} + +// PlanRepairs is the repair table, as a pure function. +// +// It implements exactly the four rows of the design's "Consistency and +// recovery" table and nothing else. States it does not name — a merged row +// whose branch is no longer an ancestor of the approved head, a rejected row +// whose branch still exists — are deliberately left alone: neither is a +// half-finished write, and repairing them would mean deciding something the +// design did not. +// +// "Merged" needs one qualification the design's table does not state, and +// without it the reconciler corrupts state during ordinary operation. A +// proposal branch is cut *at* the approved head, so between the cut and the +// agent's first commit its tip is trivially an ancestor of that head — and the +// literal rule "branch merged into the approved head, row still open" fires on +// a proposal that has not merged and has no content at all. The same holds +// forever after for a proposal whose agent never committed. A branch still +// sitting on its recorded base is therefore never treated as merged, and a base +// that could not be resolved is treated the same way: repairing on facts we +// could not establish is worse than leaving the row open for a human to see. +// +// RepairMarkMerged always records core.ApprovalPolicy. The ref proves the +// merge happened and nothing proves how it was authorized — the approval kind +// existed only in the memory of the process that died. Of the two available +// lies, "policy" is the safe one: recording "human" would launder unreviewed +// content as blessed, which is the exact failure the bimodal decision exists to +// prevent, while recording "policy" understates the review and puts the +// proposal in the policy-merged digest, where a human sees it again. Erring +// toward visibility is the whole point of the digest. +func PlanRepairs(f SpaceFacts) []Repair { + var repairs []Repair + base := Repair{Space: f.Space, SpaceID: f.SpaceID} + + for _, p := range f.Proposals { + r := base + r.ProposalID = p.ID + r.Branch = p.Branch + + switch { + case p.HasRow && p.HasBranch && p.State == core.StateOpen && p.MergedIntoApproved && + p.BaseRev != "" && p.BranchHead != p.BaseRev: + r.Kind = RepairMarkMerged + r.Rev = f.ApprovedHead + r.Approval = core.ApprovalPolicy + r.Reason = fmt.Sprintf("branch %s has merged into the approved head %s but the row is still open", + p.Branch, short(f.ApprovedHead)) + repairs = append(repairs, r) + + case p.HasRow && !p.HasBranch && p.State == core.StateOpen: + if f.Now.Sub(p.Created) < f.Grace { + continue // in flight: the row is written before the branch + } + r.Kind = RepairDeleteRow + r.Reason = fmt.Sprintf("row is open but branch %s does not exist; the agent re-proposes", p.Branch) + repairs = append(repairs, r) + + case !p.HasRow && p.HasBranch: + r.Kind = RepairDeleteRef + r.Reason = fmt.Sprintf("branch %s has no row; its content is unrecoverable", p.Branch) + repairs = append(repairs, r) + } + } + + if f.ApprovedHead != "" && f.IndexRev != f.ApprovedHead { + r := base + r.Kind = RepairReindex + r.Rev = f.ApprovedHead + r.Reason = fmt.Sprintf("index stamp %s differs from the approved head %s", + stampOrNever(f.IndexRev), short(f.ApprovedHead)) + repairs = append(repairs, r) + } + return repairs +} + +func short(rev string) string { + if len(rev) > 8 { + return rev[:8] + } + return rev +} + +func stampOrNever(rev string) string { + if rev == "" { + return "(never indexed)" + } + return short(rev) +} + +// ReconcileFailure is one thing the reconciler could not do. Failures never +// abort the run: a space with an unreadable repository must not stop the other +// spaces from being repaired. +type ReconcileFailure struct { + Space core.SpaceRef + Repair *Repair // nil when the whole space could not be examined + Err error +} + +func (f ReconcileFailure) Error() string { + if f.Repair != nil { + return fmt.Sprintf("%s: %v", f.Repair, f.Err) + } + return fmt.Sprintf("%s: %v", f.Space, f.Err) +} + +// ReconcileReport is what one reconciler pass did. +type ReconcileReport struct { + // Spaces is how many spaces were examined. + Spaces int + + // Repaired lists the repairs that were applied. + Repaired []Repair + + // Reindex lists spaces whose index is stale. They are reported, not + // repaired: bleve is single-writer and Phase 2 owns the index. A caller + // that has an indexer drives it from this list. + Reindex []Repair + + // Failures lists what could not be examined or could not be repaired. + Failures []ReconcileFailure +} + +// Reconcile runs one pass: scan proposals/* refs and each space's approved +// head, compare against rows and index stamps, repair divergence. +// +// The read order is load-bearing. Refs are listed for every space *before* any +// proposal row is read, so a proposal opened concurrently can only ever look +// like "row with no branch" — which the grace window protects — and never like +// "branch with no row", which would delete a live agent's work. Reversing the +// two reads turns an ordinary concurrent propose into data loss. +func (s *Service) Reconcile(ctx context.Context) (*ReconcileReport, error) { + spaces, err := s.ListSpaces(ctx) + if err != nil { + return nil, err + } + + rep := &ReconcileReport{} + + // Pass one: every space's repository, approved head and proposal branches. + type observed struct { + space *Space + head plumbing.Hash + branches []gitx.Branch + } + seen := make([]observed, 0, len(spaces)) + for _, sp := range spaces { + repo, err := s.openRepo(sp.Ref) + if err != nil { + rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err}) + continue + } + sp.Repo = repo + head, err := repo.ApprovedHead(ctx) + if err != nil { + rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err}) + continue + } + branches, err := repo.ListProposalBranches(ctx) + if err != nil { + rep.Failures = append(rep.Failures, ReconcileFailure{Space: sp.Ref, Err: err}) + continue + } + seen = append(seen, observed{space: sp, head: head, branches: branches}) + } + + // Pass two: the rows, read strictly after every ref listing above. + open, err := s.store.ListProposalsByState(ctx, core.StateOpen, 0) + if err != nil { + return nil, fmt.Errorf("service: list open proposals: %w", err) + } + openBySpace := make(map[int][]*db.Proposal, len(spaces)) + for _, p := range open { + openBySpace[p.SpaceID] = append(openBySpace[p.SpaceID], p) + } + + for _, o := range seen { + rep.Spaces++ + facts, err := s.spaceFacts(ctx, o.space, o.head, o.branches, openBySpace[o.space.ID]) + if err != nil { + rep.Failures = append(rep.Failures, ReconcileFailure{Space: o.space.Ref, Err: err}) + continue + } + for _, r := range PlanRepairs(facts) { + if r.Kind == RepairReindex { + rep.Reindex = append(rep.Reindex, r) + continue + } + if err := s.applyRepair(ctx, o.space, r); err != nil { + repair := r + rep.Failures = append(rep.Failures, ReconcileFailure{ + Space: o.space.Ref, Repair: &repair, Err: err, + }) + continue + } + rep.Repaired = append(rep.Repaired, r) + } + } + return rep, nil +} + +// spaceFacts turns one space's refs and rows into the facts PlanRepairs +// consumes, resolving the two things only I/O can answer: whether a branch has +// merged into the approved head, and whether a branch without an *open* row has +// any row at all. +func (s *Service) spaceFacts(ctx context.Context, sp *Space, head plumbing.Hash, + branches []gitx.Branch, openRows []*db.Proposal) (SpaceFacts, error) { + + facts := SpaceFacts{ + Space: sp.Ref, + SpaceID: sp.ID, + ApprovedHead: head.String(), + Now: s.now(), + Grace: s.grace, + } + + stamp, err := s.store.GetIndexStamp(ctx, sp.ID) + switch { + case err == nil: + facts.IndexRev = stamp.Rev + case errors.Is(err, db.ErrNotFound): + // Never indexed. Left empty, which PlanRepairs reads as stale. + default: + return SpaceFacts{}, fmt.Errorf("service: read index stamp for %s: %w", sp.Ref, err) + } + + byBranch := make(map[string]gitx.Branch, len(branches)) + for _, b := range branches { + byBranch[b.Name] = b + } + + rowBranches := make(map[string]bool, len(openRows)) + for _, row := range openRows { + rowBranches[row.Branch] = true + fact := ProposalFact{ + ID: row.ID, + Branch: row.Branch, + HasRow: true, + State: row.State, + Created: row.Created, + } + if b, ok := byBranch[row.Branch]; ok { + fact.HasBranch = true + fact.BranchHead = b.Head.String() + merged, err := sp.Repo.IsAncestor(ctx, b.Head, head) + if err != nil { + return SpaceFacts{}, fmt.Errorf("service: ancestry of %s in %s: %w", row.Branch, sp.Ref, err) + } + fact.MergedIntoApproved = merged + // Resolved rather than compared as a string: base_rev is whatever + // the agent sent as If-Match, and an abbreviated spelling of the + // branch tip would otherwise read as "the branch has commits". + // + // A base that is no longer in the repository leaves this empty, + // which PlanRepairs reads as "do not repair" — the row stays open + // where a human can see it. Any other failure is a real read error + // and is surfaced rather than silently disarming the check. + base, err := sp.Repo.ResolveRev(ctx, row.BaseRev) + switch { + case err == nil: + fact.BaseRev = base.String() + case errors.Is(err, gitx.ErrNotFound), errors.Is(err, gitx.ErrBadRev): + default: + return SpaceFacts{}, fmt.Errorf("service: resolve base %q of %s in %s: %w", + row.BaseRev, row.Branch, sp.Ref, err) + } + } + facts.Proposals = append(facts.Proposals, fact) + } + + for _, b := range branches { + if rowBranches[b.Name] { + continue + } + // No *open* row claims this branch. A merged or rejected proposal keeps + // its branch and its row, so before calling the ref an orphan we ask + // whether any row owns it. Getting this wrong deletes the branch of an + // already-merged proposal. + id, ok := gitx.ParseProposalBranch(b.Name) + if ok { + row, err := s.store.GetProposal(ctx, int(id)) + switch { + case err == nil && row.SpaceID == sp.ID && row.Branch == b.Name: + facts.Proposals = append(facts.Proposals, ProposalFact{ + ID: row.ID, Branch: b.Name, HasRow: true, HasBranch: true, + State: row.State, Created: row.Created, BranchHead: b.Head.String(), + }) + continue + case err == nil, errors.Is(err, db.ErrNotFound): + // Either no row at all, or a row that belongs to another space + // or another branch — both mean this ref is unreferenced. + default: + return SpaceFacts{}, fmt.Errorf("service: look up proposal %d for %s: %w", id, sp.Ref, err) + } + } + facts.Proposals = append(facts.Proposals, ProposalFact{ + ID: int(idOrZero(b.Name)), Branch: b.Name, HasBranch: true, BranchHead: b.Head.String(), + }) + } + return facts, nil +} + +func idOrZero(branch string) int64 { + id, ok := gitx.ParseProposalBranch(branch) + if !ok { + return 0 + } + return id +} + +// applyRepair executes one repair. RepairReindex never reaches it — Phase 2 +// owns the index — and an unknown kind is an error rather than a no-op, so a +// repair added to PlanRepairs without an implementation fails loudly. +func (s *Service) applyRepair(ctx context.Context, sp *Space, r Repair) error { + switch r.Kind { + case RepairMarkMerged: + return s.store.MarkProposalMerged(ctx, r.ProposalID, r.Approval, r.Rev) + case RepairDeleteRow: + return s.deleteOpenProposalRow(ctx, r.ProposalID) + case RepairDeleteRef: + return s.deleteProposalRef(ctx, sp, r.Branch) + default: + return fmt.Errorf("service: no implementation for repair %q", r.Kind) + } +} + +// deleteOpenProposalRow removes a proposal row that never got a branch. +// +// The SQL is written here rather than called on db.Store because db/ exposes no +// delete — the corrected repair table needs one and the persistence layer was +// built before the table was corrected. It belongs in db/ as DeleteOpenProposal +// and should move there; it lives in this file so that the repair the design +// specifies actually happens rather than silently not happening. +// +// The `state = 'open'` guard is in the statement, not in Go: a check-then-write +// would let a merge land in between and delete the row of a proposal that had +// just succeeded. +func (s *Service) deleteOpenProposalRow(ctx context.Context, id int) error { + const q = `DELETE FROM proposal WHERE id = $1 AND state = 'open'` + res, err := s.q.ExecContext(ctx, q, id) + if err != nil { + return fmt.Errorf("service: delete proposal %d: %w", id, err) + } + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("service: delete proposal %d: rows affected: %w", id, err) + } + if n == 0 { + return fmt.Errorf("service: proposal %d is no longer open; not deleted", id) + } + return nil +} + +// deleteProposalRef removes an unreferenced proposals/* branch. +// +// Like the row delete above, this primitive is missing from the layer that +// should own it: gitx exposes branch creation and the merge but no delete. It +// is written here against go-git directly, under gitx's own per-space write +// lock, so it still excludes the merge path and a concurrent in-process write — +// but it should move into gitx, where the compare-and-swap retry logic for the +// two-writer problem already lives. +// +// The namespace check is not defence in depth, it is the only thing standing +// between a caller bug and a deleted approved branch. +func (s *Service) deleteProposalRef(ctx context.Context, sp *Space, branch string) error { + if !gitx.IsProposalBranch(branch) { + return fmt.Errorf("service: refusing to delete %q: only %s* branches are deletable", + branch, gitx.ProposalPrefix) + } + return sp.Repo.WithLock(ctx, func(ctx context.Context) error { + repo, err := git.PlainOpen(sp.Repo.Dir()) + if err != nil { + return fmt.Errorf("service: open %s: %w", sp.Repo.Dir(), err) + } + name := plumbing.NewBranchReferenceName(branch) + if err := repo.Storer.RemoveReference(name); err != nil { + return fmt.Errorf("service: delete %s in %s: %w", name, sp.Ref, err) + } + return nil + }) +} + +// RunReconciler runs the reconciler at startup and then on a ticker, until ctx +// is cancelled. report is called with the outcome of every pass; a nil report +// callback discards it. +// +// Running at startup is the half that matters: a daemon killed mid-merge +// repairs itself on the next boot with no manual intervention. The ticker +// catches the rest — a crash that leaves the daemon running, or a repair that +// failed once and succeeds later. +func (s *Service) RunReconciler(ctx context.Context, interval time.Duration, report func(*ReconcileReport, error)) { + if interval <= 0 { + interval = DefaultReconcileInterval + } + run := func() { + rep, err := s.Reconcile(ctx) + if report != nil { + report(rep, err) + } + } + run() + + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + run() + } + } +} diff --git a/service/reconcile_test.go b/service/reconcile_test.go new file mode 100644 index 0000000000000000000000000000000000000000..6d80e2b9ab6824f3dd031a09068a7e6871d76d38 --- /dev/null +++ b/service/reconcile_test.go @@ -0,0 +1,295 @@ +package service + +import ( + "context" + "strings" + "testing" + "time" + + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/gitx" +) + +// facts builds a SpaceFacts with the clock and grace fixed, so a case only has +// to say what it is about. +func facts(head, indexRev string, proposals ...ProposalFact) SpaceFacts { + return SpaceFacts{ + Space: fxSpace, + SpaceID: 1, + ApprovedHead: head, + IndexRev: indexRev, + Proposals: proposals, + Now: fxTime(60), + Grace: DefaultReconcileGrace, + } +} + +const ( + headRev = "1f0c1d1a1e2b3c4d5e6f708192a3b4c5d6e7f809" + branchRev = "2a1b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d" + baseRev = "3b2c3d4e5f60718293a4b5c6d7e8f90a1b2c3d4e" +) + +// The four rows of the design's repair table, plus every state that must be +// left alone. Nothing here touches git or Postgres. +func TestPlanRepairs(t *testing.T) { + old := fxTime(0) // an hour before Now: well past the grace window + fresh := fxTime(59) // one minute before Now: still in flight + + tests := []struct { + name string + facts SpaceFacts + want []RepairKind + }{ + { + name: "nothing to do", + facts: facts(headRev, headRev), + }, + { + // Crash between the row insert and the branch write. The row holds + // no content and the agent still has the document it wanted. + name: "open row with no branch", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, + State: core.StateOpen, Created: old, + }), + want: []RepairKind{RepairDeleteRow}, + }, + { + // The same state a live propose passes through. Deleting it would + // destroy an agent's work at random and never in a test. + name: "open row with no branch, still inside the grace window", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, + State: core.StateOpen, Created: fresh, + }), + }, + { + // The id is a Postgres serial and title, rationale, base_rev and + // provenance live nowhere in a ref, so the row cannot be rebuilt. + name: "orphan ref with no row", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasBranch: true, BranchHead: branchRev, + }), + want: []RepairKind{RepairDeleteRef}, + }, + { + name: "orphan ref whose name carries no id", + facts: facts(headRev, headRev, ProposalFact{ + Branch: "proposals/scratch", HasBranch: true, BranchHead: branchRev, + }), + want: []RepairKind{RepairDeleteRef}, + }, + { + // Crash between the merge commit and the row update. The ref is + // truth for merged-ness. + name: "open row whose branch merged into the approved head", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, HasBranch: true, + State: core.StateOpen, Created: old, MergedIntoApproved: true, + BranchHead: branchRev, BaseRev: baseRev, + }), + want: []RepairKind{RepairMarkMerged}, + }, + { + // A branch still sitting on its base has no content and cannot have + // merged, even though its tip is trivially an ancestor of the head. + // This is the state every propose passes through between cutting + // the branch and the agent's first commit. + name: "open row whose branch was cut but never committed to", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, HasBranch: true, + State: core.StateOpen, Created: old, MergedIntoApproved: true, + BranchHead: baseRev, BaseRev: baseRev, + }), + }, + { + // Repairing on facts we could not establish is worse than leaving + // the row open where a human can see it. + name: "open row whose base could not be resolved", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, HasBranch: true, + State: core.StateOpen, Created: old, MergedIntoApproved: true, + BranchHead: branchRev, + }), + }, + { + name: "ordinary open proposal", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, HasBranch: true, + State: core.StateOpen, Created: old, + }), + }, + { + // A merged or rejected proposal keeps its branch and its row. + // Neither is a half-finished write. + name: "merged proposal that still has its branch", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, HasBranch: true, + State: core.StateMerged, Created: old, MergedIntoApproved: true, + BranchHead: branchRev, BaseRev: baseRev, + }), + }, + { + name: "rejected proposal that still has its branch", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, HasBranch: true, + State: core.StateRejected, Created: old, + }), + }, + { + name: "rejected proposal whose branch is gone", + facts: facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, + State: core.StateRejected, Created: old, + }), + }, + { + name: "index stamp behind the approved head", + facts: facts(headRev, branchRev), + want: []RepairKind{RepairReindex}, + }, + { + // A missing stamp is stale, never "up to date". + name: "space that has never been indexed", + facts: facts(headRev, ""), + want: []RepairKind{RepairReindex}, + }, + { + name: "several repairs in one space", + facts: facts(headRev, branchRev, + ProposalFact{ID: 1, Branch: "proposals/1", HasRow: true, + State: core.StateOpen, Created: old}, + ProposalFact{ID: 2, Branch: "proposals/2", HasBranch: true, BranchHead: branchRev}, + ProposalFact{ID: 3, Branch: "proposals/3", HasRow: true, HasBranch: true, + State: core.StateOpen, Created: old, MergedIntoApproved: true, + BranchHead: branchRev, BaseRev: baseRev}, + ), + want: []RepairKind{RepairDeleteRow, RepairDeleteRef, RepairMarkMerged, RepairReindex}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := PlanRepairs(tc.facts) + if len(got) != len(tc.want) { + t.Fatalf("repairs = %v, want %v", kinds(got), tc.want) + } + for i, want := range tc.want { + if got[i].Kind != want { + t.Fatalf("repairs = %v, want %v", kinds(got), tc.want) + } + if got[i].Space != fxSpace || got[i].SpaceID != 1 { + t.Errorf("repair %d lost its space: %+v", i, got[i]) + } + if got[i].Reason == "" { + t.Errorf("repair %d has no reason for the log", i) + } + } + }) + } +} + +// The approval kind existed only in the memory of the process that died. Of the +// two available lies, "policy" is the safe one: "human" would launder +// unreviewed content as blessed, while "policy" puts the proposal in the digest +// where a human sees it again. +func TestPlanRepairsRecordsAPolicyApprovalOnARepairedMerge(t *testing.T) { + got := PlanRepairs(facts(headRev, headRev, ProposalFact{ + ID: 42, Branch: "proposals/42", HasRow: true, HasBranch: true, + State: core.StateOpen, Created: fxTime(0), MergedIntoApproved: true, + BranchHead: branchRev, BaseRev: baseRev, + })) + if len(got) != 1 { + t.Fatalf("repairs = %v", kinds(got)) + } + if got[0].Approval != core.ApprovalPolicy { + t.Errorf("approval = %q, want %q", got[0].Approval, core.ApprovalPolicy) + } + if got[0].Rev != headRev { + t.Errorf("merged rev = %q, want the approved head", got[0].Rev) + } +} + +func TestPlanRepairsIgnoresAnUnknownApprovedHead(t *testing.T) { + if got := PlanRepairs(facts("", "")); len(got) != 0 { + t.Fatalf("repairs = %v, want none when the approved head is unknown", kinds(got)) + } +} + +func kinds(rs []Repair) []RepairKind { + out := make([]RepairKind, 0, len(rs)) + for _, r := range rs { + out = append(out, r.Kind) + } + return out +} + +func TestDeleteProposalRefRemovesOnlyProposalBranches(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + ctx := context.Background() + + commitFiles(t, sp, sp.ApprovedBranch(), 1, map[string][]byte{ + "specs/0007.md": doc("SPEC-0007", "Storage", "one"), + }) + cutBranch(t, sp, "proposals/1", sp.ApprovedBranch()) + + if err := svc.deleteProposalRef(ctx, sp, "proposals/1"); err != nil { + t.Fatalf("deleteProposalRef: %v", err) + } + branches, err := sp.Repo.ListProposalBranches(ctx) + if err != nil { + t.Fatalf("ListProposalBranches: %v", err) + } + if len(branches) != 0 { + t.Fatalf("branches = %+v, want none", branches) + } + + // The namespace check is the only thing between a caller bug and a deleted + // approved branch. + err = svc.deleteProposalRef(ctx, sp, sp.ApprovedBranch()) + if err == nil { + t.Fatal("deleted the approved branch") + } + if !strings.Contains(err.Error(), gitx.ProposalPrefix) { + t.Errorf("err = %v, want it to name the only deletable namespace", err) + } + if _, err := sp.Repo.ApprovedHead(ctx); err != nil { + t.Fatalf("approved branch is gone: %v", err) + } +} + +func TestApplyRepairRefusesAnUnimplementedKind(t *testing.T) { + svc, root := newService(t) + sp := newSpace(t, root, 1) + err := svc.applyRepair(context.Background(), sp, Repair{Kind: "invented"}) + if err == nil { + t.Fatal("an unknown repair kind was silently ignored") + } +} + +func TestRunReconcilerStopsWithItsContext(t *testing.T) { + svc, _ := newService(t) + ctx, cancel := context.WithCancel(context.Background()) + + done := make(chan struct{}) + var passes int + go func() { + defer close(done) + svc.RunReconciler(ctx, time.Hour, func(*ReconcileReport, error) { + passes++ + cancel() + }) + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("RunReconciler did not return after its context was cancelled") + } + // The startup pass is the half that matters: a daemon killed mid-merge + // repairs itself on the next boot with no manual intervention. + if passes != 1 { + t.Errorf("passes = %d, want the one startup pass", passes) + } +} diff --git a/service/service.go b/service/service.go new file mode 100644 index 0000000000000000000000000000000000000000..3567b295b882e8d7460e8b134643e50e09113256 --- /dev/null +++ b/service/service.go @@ -0,0 +1,247 @@ +package service + +import ( + "errors" + "fmt" + "net/url" + "path/filepath" + "strings" + "time" + + "github.com/vaughan0/go-ini" + + "sourcecraft.dev/bigbes/sr-ht-spec/authn" + "sourcecraft.dev/bigbes/sr-ht-spec/db" +) + +// ConfigSection is our config section, the literal "spec.sr.ht". The ".sr.ht" +// suffix is what puts us in the nav network list and what other services look +// us up by, so it is taken from authn rather than re-spelled here. +const ConfigSection = authn.ConfigSection + +// Sentinel errors. Callers compare with errors.Is. Every error leaving this +// package carries one of these in addition to the gitx or db class it came +// from, so api/, mcpsrv/, graph/ and web/ can map a failure to a status code +// without importing the layers below service. +var ( + // ErrIncompleteConfig is returned by LoadConfig when the instance config + // omits a key this service needs. It is a startup failure: the daemon + // should print it and exit rather than serve a request that will fail + // deeper down with a worse message. + ErrIncompleteConfig = errors.New("service: incomplete configuration") + + // ErrNotFound marks a missing space, revision, document or row. + ErrNotFound = errors.New("service: not found") + + // ErrSpaceExists marks a create that would clobber an existing space, + // either its repository on disk or its row. + ErrSpaceExists = errors.New("service: space already exists") + + // ErrPushRejected marks a push the update hook must refuse. Type-assert to + // *PushRejection for the message to print to the pushing client. + ErrPushRejected = errors.New("service: push rejected") +) + +// Config is everything service/ needs from the instance config.ini. It is a +// value so the daemon can build it once, log it, and hand copies around. +type Config struct { + // Repos is [spec.sr.ht] repos: the root under which every space's bare + // repository lives as /~/. Must be absolute — gitx + // keys its per-space write lock by directory, and two spellings of one + // directory would be two locks that do not exclude each other. + Repos string + + // Cache is [spec.sr.ht] cache: the bleve index and the blob-sha-keyed + // render cache. Pure cache, safe to delete at any time; Phase 2 owns what + // goes in it, Phase 1 only insists it is configured and absolute. + Cache string + + // Origin is [spec.sr.ht] origin, without a trailing slash. It is the base + // of every proposal URL an agent hands a human, and the host half of it is + // where authn derives the synthetic agent mailbox from. + Origin string + + // ConnectionString is [spec.sr.ht] connection-string. This package does not + // open the pool — the daemon does, so core-go's database middleware and the + // reconciler share one — but a service whose DSN is missing cannot work at + // all, so it is validated here with the rest. + ConnectionString string + + // Instance carries [sr.ht] owner-name / owner-email and the derived agent + // mailbox: the identities stamped on every commit this service makes. + Instance authn.Instance +} + +// LoadConfig reads and validates every key service/ needs out of the instance +// config, reporting all missing keys at once. +// +// Reporting them together is deliberate, and copied from compare.sr.ht's +// validateConfig: an operator fixes the config in one pass instead of +// discovering each gap on a separate restart. +// +// It validates only what this package reads. The daemon is still responsible +// for the keys core-go itself fatals on — [sr.ht] network-key and [webhooks] +// private-key, both required by crypto.InitCrypto — because those belong to +// server.New's contract, not to ours, and duplicating them here would give the +// instance two lists to keep in sync. +func LoadConfig(conf ini.File) (Config, error) { + var missing []string + get := func(section, key string) string { + v, ok := conf.Get(section, key) + if v = strings.TrimSpace(v); !ok || v == "" { + missing = append(missing, fmt.Sprintf("[%s] %s", section, key)) + return "" + } + return v + } + + cfg := Config{ + Repos: get(ConfigSection, "repos"), + Cache: get(ConfigSection, "cache"), + Origin: get(ConfigSection, "origin"), + ConnectionString: get(ConfigSection, "connection-string"), + } + // One canonical spelling of the origin, so a proposal URL built from it + // never grows a double slash and never differs between two callers. + cfg.Origin = strings.TrimSuffix(cfg.Origin, "/") + + // Read for their presence only; authn.InstanceFromConfig is what turns them + // into identities, and it must not be reached with a key missing or it + // reports one gap where we want to report all of them. + get("sr.ht", "owner-name") + get("sr.ht", "owner-email") + + if len(missing) > 0 { + return Config{}, fmt.Errorf("%w; missing required keys:\n\t%s", + ErrIncompleteConfig, strings.Join(missing, "\n\t")) + } + + inst, err := authn.InstanceFromConfig(conf) + if err != nil { + return Config{}, fmt.Errorf("%w: %w", ErrIncompleteConfig, err) + } + cfg.Instance = inst + + if err := cfg.Validate(); err != nil { + return Config{}, err + } + return cfg, nil +} + +// Validate reports whether the configuration is usable. It is exported so a +// daemon that builds a Config from somewhere other than an ini file — a test, +// or a future flag — is held to the same rules. +func (c Config) Validate() error { + var problems []string + requireAbs := func(key, path string) { + switch { + case path == "": + problems = append(problems, fmt.Sprintf("[%s] %s is empty", ConfigSection, key)) + case !filepath.IsAbs(path): + problems = append(problems, fmt.Sprintf("[%s] %s must be an absolute path, got %q", + ConfigSection, key, path)) + } + } + requireAbs("repos", c.Repos) + requireAbs("cache", c.Cache) + + switch u, err := url.Parse(c.Origin); { + case c.Origin == "": + problems = append(problems, fmt.Sprintf("[%s] origin is empty", ConfigSection)) + case err != nil: + problems = append(problems, fmt.Sprintf("[%s] origin %q is not a URL: %v", + ConfigSection, c.Origin, err)) + case u.Hostname() == "": + problems = append(problems, fmt.Sprintf("[%s] origin %q has no host", ConfigSection, c.Origin)) + case u.Scheme != "http" && u.Scheme != "https": + problems = append(problems, fmt.Sprintf("[%s] origin %q must be http or https", + ConfigSection, c.Origin)) + } + + if c.ConnectionString == "" { + problems = append(problems, fmt.Sprintf("[%s] connection-string is empty", ConfigSection)) + } + if err := c.Instance.Validate(); err != nil { + problems = append(problems, err.Error()) + } + + if len(problems) > 0 { + return fmt.Errorf("%w:\n\t%s", ErrIncompleteConfig, strings.Join(problems, "\n\t")) + } + return nil +} + +// Service is the orchestration layer. One per daemon; safe for concurrent use. +type Service struct { + cfg Config + q db.Querier + store *db.Store + tokens *TokenStore + resolver *authn.Resolver + + // grace is how long a proposal row with no branch is left alone before the + // reconciler deletes it. See DefaultReconcileGrace. + grace time.Duration + + // now is the clock, injectable so the reconciler's grace window is + // testable without sleeping. + now func() time.Time +} + +// New assembles a Service over a database handle. +// +// q is normally the *sql.DB the daemon opened from Config.ConnectionString and +// handed to core-go's database middleware, so request-scoped queries and the +// reconciler's background queries share one pool. A nil handle is refused +// rather than tolerated: every agent token would then resolve as unknown, which +// looks exactly like a mass revocation and is a miserable thing to debug. +func New(cfg Config, q db.Querier) (*Service, error) { + if err := cfg.Validate(); err != nil { + return nil, err + } + if q == nil { + return nil, errors.New("service: nil database handle") + } + store := db.NewStore(q) + tokens := NewTokenStore(store) + resolver, err := authn.NewResolver(cfg.Instance.OwnerName, tokens) + if err != nil { + return nil, fmt.Errorf("service: build resolver: %w", err) + } + return &Service{ + cfg: cfg, + q: q, + store: store, + tokens: tokens, + resolver: resolver, + grace: DefaultReconcileGrace, + now: time.Now, + }, nil +} + +// Config returns the configuration this service was built from. +func (s *Service) Config() Config { return s.cfg } + +// ReposRoot is [spec.sr.ht] repos, the root every space's bare repository lives +// under. +func (s *Service) ReposRoot() string { return s.cfg.Repos } + +// CacheDir is [spec.sr.ht] cache. Phase 2 owns its contents. +func (s *Service) CacheDir() string { return s.cfg.Cache } + +// Origin is our external origin, without a trailing slash. +func (s *Service) Origin() string { return s.cfg.Origin } + +// Instance returns the commit identities: the instance owner and the derived +// agent mailbox. +func (s *Service) Instance() authn.Instance { return s.cfg.Instance } + +// Store exposes the persistence layer. It is here for the daemon's own +// bookkeeping (token minting, migrations tooling); handlers above this layer +// call Service methods instead, because the dependency rule says nothing above +// service/ may touch db/ directly. +func (s *Service) Store() *db.Store { return s.store } + +// Resolver turns a request into an authn.Principal. The daemon installs +// Resolver().Middleware() on its router. +func (s *Service) Resolver() *authn.Resolver { return s.resolver } diff --git a/service/service_test.go b/service/service_test.go new file mode 100644 index 0000000000000000000000000000000000000000..7cb0c71e7a7ee37549a4f9bdd8a2cdeaf12665b6 --- /dev/null +++ b/service/service_test.go @@ -0,0 +1,184 @@ +package service + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "sourcecraft.dev/bigbes/sr-ht-spec/authn" + "sourcecraft.dev/bigbes/sr-ht-spec/db" +) + +func TestLoadConfigAcceptsACompleteConfig(t *testing.T) { + cfg, err := LoadConfig(testIni(t, "/var/lib/spec")) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.Repos != "/var/lib/spec" || cfg.Cache != "/var/cache/spec" { + t.Errorf("paths = %q, %q", cfg.Repos, cfg.Cache) + } + if cfg.Origin != "https://spec.srht.bigb.es" { + t.Errorf("origin = %q", cfg.Origin) + } + if cfg.Instance.AgentEmail != "agent@spec.srht.bigb.es" { + t.Errorf("agent email = %q, want it derived from our own origin", cfg.Instance.AgentEmail) + } + if cfg.Instance.OwnerName != "bigbes" { + t.Errorf("owner = %q", cfg.Instance.OwnerName) + } +} + +func TestLoadConfigTrimsTheOriginsTrailingSlash(t *testing.T) { + conf := testIni(t, "/var/lib/spec", "origin") + conf["spec.sr.ht"]["origin"] = "https://spec.srht.bigb.es/" + cfg, err := LoadConfig(conf) + if err != nil { + t.Fatalf("LoadConfig: %v", err) + } + if cfg.Origin != "https://spec.srht.bigb.es" { + t.Errorf("origin = %q, want the trailing slash gone", cfg.Origin) + } +} + +// Every missing key must be named in one message: an operator fixes the config +// in one pass instead of discovering each gap on a separate restart. +func TestLoadConfigNamesEveryMissingKeyAtOnce(t *testing.T) { + conf := testIni(t, "/var/lib/spec", "repos", "cache", "owner-email") + _, err := LoadConfig(conf) + if !errors.Is(err, ErrIncompleteConfig) { + t.Fatalf("err = %v, want ErrIncompleteConfig", err) + } + for _, want := range []string{"[spec.sr.ht] repos", "[spec.sr.ht] cache", "[sr.ht] owner-email"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("message does not name %q:\n%s", want, err) + } + } + if strings.Contains(err.Error(), "connection-string") { + t.Errorf("message names a key that was present:\n%s", err) + } +} + +func TestLoadConfigRejectsUnusableValues(t *testing.T) { + tests := []struct { + name string + section string + key string + value string + want string + }{ + {"relative repos", ConfigSection, "repos", "spec", "absolute path"}, + {"relative cache", ConfigSection, "cache", "./cache", "absolute path"}, + {"origin with no host", ConfigSection, "origin", "spec.srht.bigb.es", "no host"}, + {"origin with a bad scheme", ConfigSection, "origin", "ftp://spec.srht.bigb.es", "http or https"}, + {"blank owner", "sr.ht", "owner-name", " ", "[sr.ht] owner-name"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + conf := testIni(t, "/var/lib/spec", tc.key) + conf[tc.section][tc.key] = tc.value + _, err := LoadConfig(conf) + if !errors.Is(err, ErrIncompleteConfig) { + t.Fatalf("err = %v, want ErrIncompleteConfig", err) + } + if !strings.Contains(err.Error(), tc.want) { + t.Errorf("message does not mention %q:\n%s", tc.want, err) + } + }) + } +} + +// A nil handle would make every agent token resolve as unknown, which looks +// exactly like a mass revocation. +func TestNewRefusesANilDatabaseHandle(t *testing.T) { + if _, err := New(testConfig(t, t.TempDir()), nil); err == nil { + t.Fatal("New with a nil handle succeeded") + } +} + +func TestNewValidatesTheConfig(t *testing.T) { + cfg := testConfig(t, "relative/path") + if _, err := New(cfg, deadDB(t)); !errors.Is(err, ErrIncompleteConfig) { + t.Fatalf("err = %v, want ErrIncompleteConfig", err) + } +} + +// fakeTokens is an agentTokenLookup that answers from a script. +type fakeTokens struct { + row *db.AgentToken + err error +} + +func (f fakeTokens) AgentTokenByHash(context.Context, []byte) (*db.AgentToken, error) { + return f.row, f.err +} + +// The whole of the adapter is this error contract: db/ says ErrNotFound, authn +// demands ErrUnknownToken, and an unmapped pass-through would turn a bad +// credential into a 503 telling the agent to retry forever. +func TestTokenStoreMapsAMissingRowToUnknownToken(t *testing.T) { + ts := NewTokenStore(fakeTokens{err: db.ErrNotFound}) + _, err := ts.LookupAgentToken(context.Background(), []byte("hash")) + if !errors.Is(err, authn.ErrUnknownToken) { + t.Fatalf("err = %v, want authn.ErrUnknownToken", err) + } + if !authn.IsAuthFailure(err) { + t.Error("an unknown token must be a permanent auth failure, not a transient one") + } +} + +func TestTokenStoreKeepsOtherFailuresTransient(t *testing.T) { + boom := errors.New("connection refused") + ts := NewTokenStore(fakeTokens{err: boom}) + _, err := ts.LookupAgentToken(context.Background(), []byte("hash")) + if !errors.Is(err, boom) { + t.Fatalf("err = %v, want it to wrap the store failure", err) + } + if authn.IsAuthFailure(err) { + t.Error("a store outage must never read as a bad credential") + } +} + +// A revoked row is returned rather than refused, so authn can say "revoked" +// instead of "unknown". +func TestTokenStoreReturnsARevokedRow(t *testing.T) { + revoked := fxTime(1) + ts := NewTokenStore(fakeTokens{row: &db.AgentToken{ + ID: 7, Name: "cron", Hash: []byte("h"), Created: fxTime(0), Revoked: &revoked, + }}) + tok, err := ts.LookupAgentToken(context.Background(), []byte("h")) + if err != nil { + t.Fatalf("LookupAgentToken: %v", err) + } + if !tok.IsRevoked() || tok.ID != 7 || tok.Name != "cron" { + t.Fatalf("token = %+v", tok) + } +} + +func TestTokenStoreRefusesANilRowWithNoError(t *testing.T) { + ts := NewTokenStore(fakeTokens{}) + if _, err := ts.LookupAgentToken(context.Background(), []byte("h")); err == nil { + t.Fatal("a nil row with no error authenticated") + } +} + +func TestServiceExposesItsWiring(t *testing.T) { + svc, root := newService(t) + if svc.ReposRoot() != root { + t.Errorf("ReposRoot = %q, want %q", svc.ReposRoot(), root) + } + if svc.Origin() != "https://spec.srht.bigb.es" { + t.Errorf("Origin = %q", svc.Origin()) + } + if svc.Resolver() == nil || svc.Resolver().Owner() != "bigbes" { + t.Errorf("resolver = %v", svc.Resolver()) + } + if svc.Store() == nil || svc.TokenStore() == nil { + t.Error("store or token store is nil") + } + if svc.grace != DefaultReconcileGrace || svc.now == nil { + t.Errorf("reconciler defaults not wired: grace=%v", svc.grace) + } + var _ time.Duration = DefaultReconcileInterval +} diff --git a/service/space.go b/service/space.go new file mode 100644 index 0000000000000000000000000000000000000000..97b1c082d6ddbb1dba4b5fab65d0d9f3897506ae --- /dev/null +++ b/service/space.go @@ -0,0 +1,150 @@ +package service + +import ( + "context" + "errors" + "fmt" + "os" + "time" + + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/db" + "sourcecraft.dev/bigbes/sr-ht-spec/gitx" +) + +// Space is one space, resolved: its reference, its row and its repository. +// +// The repository is the space — "spaces exist as repos" — and the row is +// bookkeeping that makes listing and index staleness cheap. ID is the row's +// primary key, which every other table references the space by; it is zero only +// for a space whose repository exists but whose row does not, a state +// CreateSpace is written to avoid and OpenSpace refuses to invent. +type Space struct { + Ref core.SpaceRef + ID int + Created time.Time + Repo *gitx.Repo +} + +// ApprovedBranch is the branch a document must be reachable from to be +// approved. Read from the repository's HEAD, so there is exactly one place a +// space records it. +func (sp *Space) ApprovedBranch() string { return sp.Repo.ApprovedBranch() } + +// CreateSpace creates a space: the bare repository first, then the row. +// +// The order is not arbitrary and it is the opposite of the proposal path's. +// A proposal must be row-first because its branch name derives from the row's +// serial id; a space has no such dependency, and git is authoritative for +// content, so the repository leads: +// +// - repository, then row — a crash in between leaves content on disk that is +// merely unlisted, and owner and name are recoverable from the directory +// name alone. +// - row, then repository — a crash in between leaves a phantom space that +// lists fine and 404s on every read, and db/ exposes no way to delete it. +// +// If the row insert fails, the repository this call just created is removed +// again and the insert's error is returned. That is safe precisely because it +// is seconds old, empty apart from gitx's initial commit, and named nowhere +// yet: nothing can have pushed to it. A cleanup failure is reported alongside +// the original error rather than swallowed. +func (s *Service) CreateSpace(ctx context.Context, ref core.SpaceRef) (*Space, error) { + if err := core.ValidateOwner(ref.Owner); err != nil { + return nil, err + } + if err := core.ValidateSpaceName(ref.Name); err != nil { + return nil, err + } + + repo, err := gitx.Create(ctx, s.cfg.Repos, ref, gitx.CreateOptions{ + Owner: gitx.Signature{ + Name: s.cfg.Instance.OwnerName, + Email: s.cfg.Instance.OwnerEmail, + When: s.now().UTC(), + }, + }) + if err != nil { + if errors.Is(err, gitx.ErrExists) { + return nil, fmt.Errorf("%w: %w", ErrSpaceExists, err) + } + return nil, fmt.Errorf("service: create repository for %s: %w", ref, err) + } + + row, err := s.store.CreateSpace(ctx, ref) + if err != nil { + rmErr := os.RemoveAll(repo.Dir()) + if rmErr != nil { + return nil, fmt.Errorf("service: create row for %s: %w "+ + "(the repository at %s could not be removed either: %v — remove it by hand before retrying)", + ref, err, repo.Dir(), rmErr) + } + if errors.Is(err, db.ErrSpaceExists) { + return nil, fmt.Errorf("%w: %w", ErrSpaceExists, err) + } + return nil, fmt.Errorf("service: create row for %s: %w", ref, err) + } + + return &Space{Ref: ref, ID: row.ID, Created: row.Created, Repo: repo}, nil +} + +// OpenSpace resolves a space by reference: its row and its repository, both +// required. +// +// A row with no repository, or a repository with no row, is a half-created +// space rather than a space, and is reported as such. Neither half is invented: +// serving reads from a repository with no row would give every document a +// space_id of zero in the index, and returning a row whose repository is +// missing would answer "the space exists" to every question and fail on each +// individual document. +func (s *Service) OpenSpace(ctx context.Context, ref core.SpaceRef) (*Space, error) { + row, err := s.store.GetSpace(ctx, ref) + if err != nil { + if errors.Is(err, db.ErrNotFound) { + return nil, fmt.Errorf("%w: space %s", ErrNotFound, ref) + } + return nil, fmt.Errorf("service: look up space %s: %w", ref, err) + } + repo, err := s.openRepo(ref) + if err != nil { + return nil, err + } + return &Space{Ref: ref, ID: row.ID, Created: row.Created, Repo: repo}, nil +} + +// openRepo opens a space's bare repository, mapping gitx's absence onto this +// package's. Split out because the reconciler resolves repositories for spaces +// it already has rows for. +func (s *Service) openRepo(ref core.SpaceRef) (*gitx.Repo, error) { + repo, err := gitx.Open(s.cfg.Repos, ref) + if err != nil { + if errors.Is(err, gitx.ErrNotFound) { + return nil, fmt.Errorf("%w: repository for space %s at %s", + ErrNotFound, ref, gitx.DiskPath(s.cfg.Repos, ref)) + } + return nil, fmt.Errorf("service: open repository for %s: %w", ref, err) + } + return repo, nil +} + +// ListSpaces returns every space, ordered by owner then name. +// +// There is one human on this instance and no visibility levels, so there is +// nothing to filter by — the list is the whole corpus, which is also exactly +// what the meta-project (a filter that excludes nothing) needs. +// +// Repositories are not opened. Listing is a page of names, and opening N bare +// repositories to render it would make the cheapest view in the service the +// most expensive; callers that need a repository call OpenSpace. Space.Repo is +// therefore nil in every element returned here. +func (s *Service) ListSpaces(ctx context.Context) ([]*Space, error) { + rows, err := s.store.ListSpaces(ctx) + if err != nil { + return nil, fmt.Errorf("service: list spaces: %w", err) + } + out := make([]*Space, 0, len(rows)) + for _, row := range rows { + out = append(out, &Space{Ref: row.Ref, ID: row.ID, Created: row.Created}) + } + return out, nil +} diff --git a/service/space_test.go b/service/space_test.go new file mode 100644 index 0000000000000000000000000000000000000000..53c90c3897593e9db2b2188ea1b6e0c236f26bed --- /dev/null +++ b/service/space_test.go @@ -0,0 +1,107 @@ +package service + +import ( + "context" + "errors" + "os" + "testing" + + "sourcecraft.dev/bigbes/sr-ht-spec/core" + "sourcecraft.dev/bigbes/sr-ht-spec/gitx" +) + +// A crash between the repository and the row leaves content that is merely +// unlisted; a *failure* of the row insert leaves nothing at all, because the +// repository is seconds old, empty and named nowhere yet. +func TestCreateSpaceRemovesTheRepositoryWhenTheRowFails(t *testing.T) { + svc, root := newService(t) // its database handle cannot be reached + dir := gitx.DiskPath(root, fxSpace) + + _, err := svc.CreateSpace(context.Background(), fxSpace) + if err == nil { + t.Fatal("CreateSpace succeeded with an unreachable database") + } + if _, statErr := os.Stat(dir); !os.IsNotExist(statErr) { + t.Fatalf("repository at %s survived a failed row insert (stat: %v)", dir, statErr) + } +} + +func TestCreateSpaceRejectsUnsafeNames(t *testing.T) { + svc, _ := newService(t) + ctx := context.Background() + for _, ref := range []core.SpaceRef{ + {Owner: "..", Name: "rfcs"}, + {Owner: "bigbes", Name: "../../etc"}, + {Owner: "bigbes", Name: ""}, + {Owner: "", Name: "rfcs"}, + } { + if _, err := svc.CreateSpace(ctx, ref); err == nil { + t.Errorf("CreateSpace(%+v) succeeded", ref) + } + } +} + +func TestSpaceLifecycle(t *testing.T) { + svc, root := newTestService(t) + ctx := context.Background() + + sp, err := svc.CreateSpace(ctx, fxSpace) + if err != nil { + t.Fatalf("CreateSpace: %v", err) + } + if sp.ID == 0 { + t.Error("space has no row id") + } + if sp.Repo.Dir() != gitx.DiskPath(root, fxSpace) { + t.Errorf("dir = %q", sp.Repo.Dir()) + } + // The approved head resolves from the moment the space exists, so no + // reader, reconciler or merge has to special-case an unborn branch. + if _, err := sp.Repo.ApprovedHead(ctx); err != nil { + t.Errorf("approved head: %v", err) + } + + if _, err := svc.CreateSpace(ctx, fxSpace); !errors.Is(err, ErrSpaceExists) { + t.Errorf("second CreateSpace err = %v, want ErrSpaceExists", err) + } + + opened, err := svc.OpenSpace(ctx, fxSpace) + if err != nil { + t.Fatalf("OpenSpace: %v", err) + } + if opened.ID != sp.ID || opened.Repo == nil { + t.Errorf("opened = %+v", opened) + } + + if _, err := svc.OpenSpace(ctx, core.SpaceRef{Owner: "bigbes", Name: "nope"}); !errors.Is(err, ErrNotFound) { + t.Errorf("OpenSpace of a missing space err = %v, want ErrNotFound", err) + } + + spaces, err := svc.ListSpaces(ctx) + if err != nil { + t.Fatalf("ListSpaces: %v", err) + } + if len(spaces) != 1 || spaces[0].Ref != fxSpace { + t.Fatalf("spaces = %+v", spaces) + } + if spaces[0].Repo != nil { + t.Error("ListSpaces opened repositories; listing must stay cheap") + } +} + +// A row whose repository is missing is a half-created space, not a space. It is +// reported rather than served: every document read would fail individually. +func TestOpenSpaceRefusesARowWithNoRepository(t *testing.T) { + svc, root := newTestService(t) + ctx := context.Background() + + if _, err := svc.CreateSpace(ctx, fxSpace); err != nil { + t.Fatalf("CreateSpace: %v", err) + } + if err := os.RemoveAll(gitx.DiskPath(root, fxSpace)); err != nil { + t.Fatalf("remove repository: %v", err) + } + if _, err := svc.OpenSpace(ctx, fxSpace); !errors.Is(err, ErrNotFound) { + t.Fatalf("err = %v, want ErrNotFound", err) + } +} diff --git a/service/token.go b/service/token.go new file mode 100644 index 0000000000000000000000000000000000000000..e54528996f41894e2739a8e660e54cfda257ac57 --- /dev/null +++ b/service/token.go @@ -0,0 +1,75 @@ +package service + +import ( + "context" + "errors" + "fmt" + + "sourcecraft.dev/bigbes/sr-ht-spec/authn" + "sourcecraft.dev/bigbes/sr-ht-spec/db" +) + +// AgentTokenLookup is the one db/ method the token adapter needs. It is an +// interface rather than a *db.Store so the error mapping below — the part that +// actually carries a contract — can be tested against a fake instead of a +// Postgres instance. *db.Store satisfies it. +type AgentTokenLookup interface { + AgentTokenByHash(ctx context.Context, hash []byte) (*db.AgentToken, error) +} + +// TokenStore adapts db/'s agent_token queries to authn.TokenStore. It exists +// because authn must not import db: authn declares the sliver of persistence it +// needs, and service/ — the layer that is allowed to know about both — wires +// them together. +// +// The whole of the adaptation is the error contract, and it is not cosmetic. +// authn's contract is that "no such token" is an error satisfying +// errors.Is(err, authn.ErrUnknownToken) and that everything else is transient. +// db/ spells the same condition ErrNotFound, which authn has never heard of, so +// an unmapped pass-through would make an unknown token look like a Postgres +// outage: a 503 telling an agent to retry a credential that will never work. +type TokenStore struct { + lookup AgentTokenLookup +} + +// NewTokenStore wires a db.Store in as authn's TokenStore. +func NewTokenStore(lookup AgentTokenLookup) *TokenStore { + return &TokenStore{lookup: lookup} +} + +// TokenStore returns the adapter the resolver authenticates agents through. +func (s *Service) TokenStore() *TokenStore { return s.tokens } + +// LookupAgentToken implements authn.TokenStore. +// +// A revoked row is returned rather than refused: authn is what turns it into a +// refusal, so the refusal can say "revoked" instead of "unknown" and an +// operator can tell a token they deliberately killed from one that never +// existed. Any other failure is returned wrapped and unclassified, which authn +// reads as transient — the fail-closed direction, since a store outage must +// never read as a valid credential. +func (t *TokenStore) LookupAgentToken(ctx context.Context, hash []byte) (authn.AgentToken, error) { + if t.lookup == nil { + return authn.AgentToken{}, errors.New("service: TokenStore has no backing store") + } + row, err := t.lookup.AgentTokenByHash(ctx, hash) + if err != nil { + if errors.Is(err, db.ErrNotFound) { + return authn.AgentToken{}, fmt.Errorf("%w: no agent_token row matches the presented token", + authn.ErrUnknownToken) + } + return authn.AgentToken{}, fmt.Errorf("service: look up agent token: %w", err) + } + if row == nil { + // db/ never returns (nil, nil); a store that did would otherwise + // authenticate a nil row as a valid token. + return authn.AgentToken{}, errors.New("service: agent token lookup returned no row and no error") + } + return authn.AgentToken{ + ID: int64(row.ID), + Name: row.Name, + Hash: row.Hash, + Created: row.Created, + Revoked: row.Revoked, + }, nil +}