amadan
Add the scope computation for a scoped replication attestation, unwired
durabilityScope decides what a box may honestly attest about its replication ledger: every account is vouched for except the owners of unit-backed routes, of exec routes that record a database and do not hibernate, and of any route whose database the box-wide litestream config names. Nothing is attested when the replicator's state cannot be read, when the config names a database no route owns, or when the unit has been running since before its config was last written. Nothing in the agent calls it. No production box attests, and none will until the reader of the replicator's state and the agent's call exist behind a switch, have been reviewed, and have a scoped approval. The test the attestation plan required first is here: a box with a unit-backed route and the external replicator present publishes a day that is not complete for that route's owner, and is complete for the owner of a hibernating exec route. Also the move worker's checkpoint, which the requirements audit listed as untested: its requests are recorded against the route's owner under the checkpoint and lease purposes, and the unmetered fallback files an escape.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
3 files changed,
+467
−0
internal/host/durabilityscope.go+140 −0internal/host/durabilityscope_test.go+150 −0internal/host/movescheckpoint_test.go+177 −0
diff --git a/internal/host/durabilityscope.go b/internal/host/durabilityscope.go| new file mode 100644 |
| index 000000000..b35e32c7a |
| --- /dev/null |
| +++ b/internal/host/durabilityscope.go |
| @@ -0,0 +1,140 @@ |
| +package host |
| + |
| +import ( |
| + "fmt" |
| + "path/filepath" |
| + "sort" |
| + "strings" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +// The scoped attestation of the replication ledger: what this box may |
| +// honestly claim about who replicates through the ledger and who does not |
| +// (docs/superpowers/plans/2026-10-01-durability-attestation-path.md, path |
| +// A). NOTHING IN THE AGENT CALLS THIS YET. No production box attests, and |
| +// none will until the call is wired behind an explicit switch that has had |
| +// its own review and a scoped approval to set. |
| +// |
| +// A box has two replicators. The agent's own (the activator's embedded |
| +// Litestream, the move worker) records every request in the ledger. The |
| +// box-wide litestream unit, an external binary run as root from |
| +// /etc/litestream.yml, records nothing: it replicates every database that |
| +// is not a hibernating exec route's (ops.FleetDBs). An account can be |
| +// vouched for only when the external replicator touches none of its |
| +// databases, so the attestation names the accounts it does touch and |
| +// excludes them. |
| + |
| +// replicatorState is what the box can observe about the external |
| +// replicator. |
| +type replicatorState struct { |
| + // ConfigPresent is false when there is no config file at all: the |
| + // external replicator has nothing to replicate. |
| + ConfigPresent bool |
| + // Paths are the database paths the config names (ops.ParseLitestreamDBs). |
| + Paths []string |
| + // ConfigModified is the config file's modification time. |
| + ConfigModified time.Time |
| + // UnitActive and UnitStarted describe the unit that runs the external |
| + // replicator: whether it is running, and since when. |
| + UnitActive bool |
| + UnitStarted time.Time |
| + // Err is set when any of the above could not be read. |
| + Err error |
| +} |
| + |
| +// durabilityScope computes the attestation for this box from its routes |
| +// and the external replicator's observed state. reason says why nothing is |
| +// attested, when nothing is. |
| +// |
| +// An account is excluded when either says the external replicator may |
| +// touch its databases: |
| +// |
| +// - the registry: it owns a unit-backed route (whose database only the |
| +// box-wide replicator can reach, and which may sit in a hand-kept |
| +// config even when the route records no path), or an exec route that |
| +// records a database and does not hibernate (which the config |
| +// generator hands to the box-wide replicator). A route with no |
| +// database and no unit, a static site for one, excludes nobody; |
| +// - the config: it owns a route whose database path the live config |
| +// names, whatever kind of route the registry says it is. |
| +// |
| +// Nothing is attested, for anyone, when the box cannot tell what the |
| +// external replicator is doing: |
| +// |
| +// - its state could not be read; |
| +// - the config names a database no route owns and that is not the box's |
| +// own (the registry or the console database): somebody's data is being |
| +// replicated and nobody can be named as its owner; |
| +// - the unit has been running since before the config was last written: |
| +// it is replicating from an older config this code cannot see, which |
| +// may name databases the current one dropped. |
| +// |
| +// The box's own databases are platform data on the platform's prefix, not |
| +// tenant replication, and exclude nobody. |
| +// |
| +// What the caller owes before this may be wired: Paths must be every |
| +// database the config names. ops.ParseLitestreamDBs reads only the shape |
| +// the generator writes, so a reader built on it must refuse (Err) a config |
| +// it did not fully account for, or a hand-kept config would hide databases |
| +// from this function. And the console database's path follows the box's |
| +// configuration (ops.consoleDBPath), which the reader must resolve the |
| +// same way the generator does. |
| +func durabilityScope(routes []registry.Route, dataDir string, rs replicatorState) (usage.DurabilityAttestation, string) { |
| + if rs.Err != nil { |
| + return usage.DurabilityAttestation{}, fmt.Sprintf("the external replicator's state could not be read: %v", rs.Err) |
| + } |
| + if rs.ConfigPresent && rs.UnitActive && rs.UnitStarted.Before(rs.ConfigModified) { |
| + return usage.DurabilityAttestation{}, "the external replicator has been running since before its config was last written; what it replicates is not what the config says" |
| + } |
| + excluded := map[string]bool{} |
| + type owned struct { |
| + path string |
| + tree bool |
| + account string |
| + } |
| + var owners []owned |
| + for _, rt := range routes { |
| + acct := release.RouteAccount(rt) |
| + if rt.Backing == registry.BackingUnit || (rt.DBPath != "" && !rt.Hibernate) { |
| + excluded[acct] = true |
| + } |
| + if rt.DBPath != "" { |
| + owners = append(owners, owned{path: filepath.Clean(rt.DBPath), tree: rt.Backing == registry.BackingUnit, account: acct}) |
| + } |
| + } |
| + own := map[string]bool{ |
| + filepath.Clean(filepath.Join(dataDir, "registry.db")): true, |
| + filepath.Clean(filepath.Join(dataDir, "console", "console.db")): true, |
| + } |
| + if rs.ConfigPresent { |
| + for _, raw := range rs.Paths { |
| + path := filepath.Clean(raw) |
| + if own[path] { |
| + continue |
| + } |
| + account := "" |
| + for _, o := range owners { |
| + // A unit route's path may be a directory the generator |
| + // walks for databases (ops.treeDBs). |
| + if path == o.path || (o.tree && strings.HasPrefix(path, o.path+string(filepath.Separator))) { |
| + account = o.account |
| + break |
| + } |
| + } |
| + if account == "" { |
| + return usage.DurabilityAttestation{}, fmt.Sprintf("the external replicator replicates %s, which no route owns", path) |
| + } |
| + excluded[account] = true |
| + } |
| + } |
| + out := usage.DurabilityAttestation{Attested: true} |
| + for acct := range excluded { |
| + out.Unmetered = append(out.Unmetered, acct) |
| + } |
| + sort.Strings(out.Unmetered) |
| + return out, "" |
| +} |
diff --git a/internal/host/durabilityscope_test.go b/internal/host/durabilityscope_test.go| new file mode 100644 |
| index 000000000..edcf0c1d3 |
| --- /dev/null |
| +++ b/internal/host/durabilityscope_test.go |
| @@ -0,0 +1,150 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "reflect" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +func scopeRoutes() []registry.Route { |
| + return []registry.Route{ |
| + // A hibernating exec route: the activator replicates it, metered. |
| + {Host: "shop.example.com", Account: "bac", App: "shop", Hibernate: true, DBPath: "/data/shop.example.com.db"}, |
| + // An always-on exec route with a database: the box-wide replicator's. |
| + {Host: "api.example.com", Account: "bad", App: "api", DBPath: "/data/api.example.com.db"}, |
| + // A unit-backed route whose path is a tree. |
| + {Host: "kass.example.com", Account: "bae", App: "kass", Backing: registry.BackingUnit, Hibernate: true, DBPath: "/var/lib/carlos-app/kass.example.com"}, |
| + // A static site: no process, no database. |
| + {Host: "docs.example.com", Account: "baf", App: "docs"}, |
| + } |
| +} |
| + |
| +func scopeState(paths ...string) replicatorState { |
| + written := time.Date(2026, 9, 17, 8, 0, 0, 0, time.UTC) |
| + return replicatorState{ConfigPresent: true, Paths: paths, ConfigModified: written, UnitActive: true, UnitStarted: written.Add(time.Minute)} |
| +} |
| + |
| +func TestDurabilityScope(t *testing.T) { |
| + generated := []string{"/data/registry.db", "/data/console/console.db", "/data/api.example.com.db", "/var/lib/carlos-app/kass.example.com/registry.db", "/var/lib/carlos-app/kass.example.com/instances/a.db"} |
| + cases := []struct { |
| + name string |
| + routes []registry.Route |
| + state replicatorState |
| + attested bool |
| + unmetered []string |
| + }{ |
| + {"the generated config: always-on and unit owners are excluded, hibernating and static owners are vouched for", |
| + scopeRoutes(), scopeState(generated...), true, []string{"bad", "bae"}}, |
| + {"no config at all: the registry alone decides", |
| + scopeRoutes(), replicatorState{}, true, []string{"bad", "bae"}}, |
| + {"only hibernating exec routes: nobody is excluded", |
| + scopeRoutes()[:1], scopeState("/data/registry.db"), true, nil}, |
| + {"the config names a hibernating route's database: its owner is excluded on that evidence", |
| + scopeRoutes(), scopeState(append([]string{"/data/shop.example.com.db"}, generated...)...), true, []string{"bac", "bad", "bae"}}, |
| + {"the config names a database no route owns: nothing is attested", |
| + scopeRoutes(), scopeState(append([]string{"/data/leftover.db"}, generated...)...), false, nil}, |
| + {"a path that only shares a prefix with a unit's tree is not in the tree", |
| + scopeRoutes(), scopeState("/var/lib/carlos-app/kass.example.com.bak/x.db"), false, nil}, |
| + {"the unit has been running since before the config was written: nothing is attested", |
| + scopeRoutes(), func() replicatorState { |
| + s := scopeState(generated...) |
| + s.UnitStarted = s.ConfigModified.Add(-time.Hour) |
| + return s |
| + }(), false, nil}, |
| + {"the replicator's state could not be read: nothing is attested", |
| + scopeRoutes(), replicatorState{Err: errors.New("permission denied")}, false, nil}, |
| + {"a route with no valid account is the platform's: the platform account is excluded", |
| + []registry.Route{{Host: "console.example.com", DBPath: "/data/other.db"}}, scopeState("/data/other.db"), true, []string{release.OpsAccount}}, |
| + {"a unit route with no recorded path still excludes its owner", |
| + []registry.Route{{Host: "kass.example.com", Account: "bae", Backing: registry.BackingUnit}}, replicatorState{}, true, []string{"bae"}}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + got, reason := durabilityScope(c.routes, "/data", c.state) |
| + if got.Attested != c.attested || !reflect.DeepEqual(got.Unmetered, c.unmetered) { |
| + t.Fatalf("attested=%v unmetered=%v (reason %q), want attested=%v unmetered=%v", got.Attested, got.Unmetered, reason, c.attested, c.unmetered) |
| + } |
| + if !got.Attested && (reason == "" || len(got.Unmetered) != 0) { |
| + t.Fatalf("an unattested scope must say why and exclude nothing: %+v %q", got, reason) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// The test the attestation plan requires before any box may attest: a box |
| +// with a unit-backed route and the external replicator present publishes a |
| +// day that is NOT complete for that route's owner, and is complete for an |
| +// account whose only route hibernates under the activator. |
| +func TestScopedAttestationPublishesADayIncompleteForTheUnitOwner(t *testing.T) { |
| + // The box came up the evening before, so the 17th is wholly inside |
| + // this run. |
| + at := time.Date(2026, 9, 16, 23, 0, 0, 0, time.UTC) |
| + clock := at |
| + p, err := newDurabilityPublisher(t.TempDir(), "box1", func() time.Time { return clock }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), time.Second) |
| + defer cancel() |
| + if !p.stopped.Load() { |
| + _ = p.stop(ctx, nil) |
| + } |
| + _ = p.ledger.WaitLate(ctx) |
| + }) |
| + // As opened in production the ledger attests nothing. |
| + if p.ledger.Attestation().Attested { |
| + t.Fatal("a freshly opened production ledger attests") |
| + } |
| + scope, reason := durabilityScope(scopeRoutes(), "/data", scopeState("/data/registry.db", "/data/api.example.com.db", "/var/lib/carlos-app/kass.example.com/registry.db")) |
| + if !scope.Attested { |
| + t.Fatalf("fixture scope: %s", reason) |
| + } |
| + if err := p.ledger.SetAttestation(scope, at); err != nil { |
| + t.Fatal(err) |
| + } |
| + clock = time.Date(2026, 9, 17, 10, 0, 0, 0, time.UTC) |
| + rec := p.recorder(registry.Route{Account: "bac", App: "shop", Host: "shop.example.com"}, usage.PurposeReplicate) |
| + if rec == nil { |
| + t.Fatal("no recorder for the hibernating route") |
| + } |
| + clock = time.Date(2026, 9, 18, 0, 1, 0, 0, time.UTC) // past midnight: the day can close |
| + store := release.NewDirStore(t.TempDir()) |
| + if err := p.publish(context.Background(), store); err != nil { |
| + t.Fatal(err) |
| + } |
| + raw, _, err := store.Get(context.Background(), usage.DurabilityDayKey("box1", "2026-09-17")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + day, err := usage.ParseDurabilityDay(raw, "box1", "2026-09-17") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + from, to := time.Date(2026, 9, 17, 0, 0, 0, 0, time.UTC), time.Date(2026, 9, 18, 0, 0, 0, 0, time.UTC) |
| + ok, unmetered, why := day.Scope(from, to) |
| + if !ok { |
| + t.Fatalf("the day proves nothing for anyone: %s", why) |
| + } |
| + if !reflect.DeepEqual(unmetered, []string{"bad", "bae"}) { |
| + t.Fatalf("excluded %v, want the always-on owner and the unit owner", unmetered) |
| + } |
| + for _, snap := range day.Epochs { |
| + c, err := snap.Decode() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if complete, _ := c.CompletenessFor("bae", from, to); complete { |
| + t.Fatal("the day reads complete for the unit-backed route's owner") |
| + } |
| + if complete, why := c.CompletenessFor("bac", from, to); !complete { |
| + t.Fatalf("the day does not read complete for the hibernating route's owner: %s", why) |
| + } |
| + } |
| +} |
diff --git a/internal/host/movescheckpoint_test.go b/internal/host/movescheckpoint_test.go| new file mode 100644 |
| index 000000000..4143f6821 |
| --- /dev/null |
| +++ b/internal/host/movescheckpoint_test.go |
| @@ -0,0 +1,177 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "database/sql" |
| + "io" |
| + "net/http" |
| + "net/http/httptest" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +// checkpointS3 is enough of S3 for a leased litestream sync: an empty |
| +// listing, absent objects, and writes that succeed. It records each write |
| +// by path. |
| +type checkpointS3 struct { |
| + mu sync.Mutex |
| + puts []string |
| +} |
| + |
| +func (f *checkpointS3) serve(w http.ResponseWriter, r *http.Request) { |
| + _, _ = io.Copy(io.Discard, r.Body) |
| + w.Header().Set("Content-Type", "application/xml") |
| + switch { |
| + case r.Method == http.MethodGet && r.URL.Query().Has("list-type"): |
| + _, _ = io.WriteString(w, `<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><KeyCount>0</KeyCount><IsTruncated>false</IsTruncated></ListBucketResult>`) |
| + case r.Method == http.MethodGet, r.Method == http.MethodHead: |
| + w.WriteHeader(http.StatusNotFound) |
| + _, _ = io.WriteString(w, `<?xml version="1.0" encoding="UTF-8"?><Error><Code>NoSuchKey</Code><Message>no such key</Message></Error>`) |
| + case r.Method == http.MethodPut: |
| + f.mu.Lock() |
| + f.puts = append(f.puts, r.URL.Path) |
| + f.mu.Unlock() |
| + w.Header().Set("ETag", `"etag"`) |
| + case r.Method == http.MethodDelete: |
| + w.WriteHeader(http.StatusNoContent) |
| + default: |
| + w.WriteHeader(http.StatusMethodNotAllowed) |
| + } |
| +} |
| + |
| +func (f *checkpointS3) replicaPuts() int { |
| + f.mu.Lock() |
| + defer f.mu.Unlock() |
| + n := 0 |
| + for _, p := range f.puts { |
| + if !strings.HasSuffix(p, "/lock.json") { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +func checkpointFixture(t *testing.T) (*checkpointS3, *durabilityPublisher, string, provision.Record) { |
| + t.Helper() |
| + f := &checkpointS3{} |
| + server := httptest.NewServer(http.HandlerFunc(f.serve)) |
| + t.Cleanup(server.Close) |
| + t.Setenv("CARLOS_S3_ENDPOINT", server.URL) |
| + t.Setenv("AWS_ACCESS_KEY_ID", "test-access") |
| + t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret") |
| + t.Setenv("AWS_SESSION_TOKEN", "") |
| + t.Setenv("AWS_PROFILE", "") |
| + t.Setenv("AWS_DEFAULT_PROFILE", "") |
| + t.Setenv("AWS_CONFIG_FILE", filepath.Join(t.TempDir(), "absent")) |
| + t.Setenv("AWS_SHARED_CREDENTIALS_FILE", filepath.Join(t.TempDir(), "absent")) |
| + t.Setenv("AWS_EC2_METADATA_DISABLED", "true") |
| + dbPath := filepath.Join(t.TempDir(), "app.db") |
| + db, err := sql.Open("sqlite", dbPath) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := db.Exec(`PRAGMA journal_mode=WAL; CREATE TABLE t(x); INSERT INTO t VALUES(1)`); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := db.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + p, err := newDurabilityPublisher(t.TempDir(), "box1", time.Now) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| + defer cancel() |
| + if !p.stopped.Load() { |
| + if err := p.stop(ctx, nil); err != nil { |
| + t.Error(err) |
| + } |
| + } |
| + if err := p.ledger.WaitLate(ctx); err != nil { |
| + t.Error(err) |
| + } |
| + }) |
| + prev := armedMovesWorker.Load() |
| + t.Cleanup(func() { armedMovesWorker.Store(prev) }) |
| + armMovesPass(activator.Config{Bucket: "own-bucket", Region: "eu-west-1", Prefix: "hib"}, t.TempDir(), nil, nil, p) |
| + return f, p, dbPath, provision.Record{Account: "bac", App: "app", Host: "app.example.com"} |
| +} |
| + |
| +// The checkpoint a move takes before its cut runs through the ledger: its |
| +// S3 requests are recorded against the route's own account, app and host, |
| +// its replica writes under the checkpoint purpose and its lease under the |
| +// lease purpose. Neither is a continuous replication write, so neither is |
| +// billed to the tenant (usage.BillableReplicationWrites); they are |
| +// recorded so the box-day can read complete. |
| +func TestMovesPassCheckpointIsMeteredToTheRouteOwner(t *testing.T) { |
| + f, p, dbPath, rec := checkpointFixture(t) |
| + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) |
| + defer cancel() |
| + if err := armedMovesWorker.Load().cfg.CheckpointFor(ctx, rec, dbPath); err != nil { |
| + t.Fatal(err) |
| + } |
| + if f.replicaPuts() == 0 { |
| + t.Fatal("fixture: the checkpoint wrote nothing to the replica") |
| + } |
| + if p.bindingFailed.Load() { |
| + t.Fatal("a metered checkpoint filed an escape") |
| + } |
| + deadline := time.Now().Add(5 * time.Second) |
| + for { |
| + c, err := p.ledger.Snapshot(time.Now().UTC().Format("2006-01-02")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + writes := map[string]int64{} |
| + for _, rows := range c.Hours { |
| + for _, row := range rows { |
| + if row.Account != rec.Account || row.App != rec.App || row.Host != rec.Host { |
| + t.Fatalf("a checkpoint request lost its owner: %+v", row) |
| + } |
| + if row.Purpose != string(usage.PurposeCheckpoint) && row.Purpose != string(usage.PurposeLease) { |
| + t.Fatalf("a checkpoint request under purpose %q: %+v", row.Purpose, row) |
| + } |
| + if row.Operation == "PutObject" { |
| + writes[row.Purpose] += row.Attempts |
| + } |
| + } |
| + } |
| + if writes[string(usage.PurposeCheckpoint)] >= int64(f.replicaPuts()) && writes[string(usage.PurposeLease)] > 0 { |
| + return |
| + } |
| + if time.Now().After(deadline) { |
| + t.Fatalf("ledger writes %+v; the store saw %d replica writes", writes, f.replicaPuts()) |
| + } |
| + time.Sleep(20 * time.Millisecond) |
| + } |
| +} |
| + |
| +// With no recorder to bind (the publisher has stopped), the checkpoint |
| +// still runs: a move must not strand a database over metering. It runs |
| +// unmetered, and that is filed as an escape, so the box-day cannot read |
| +// complete. |
| +func TestMovesPassCheckpointFallsBackUnmeteredAndFilesAnEscape(t *testing.T) { |
| + f, p, dbPath, rec := checkpointFixture(t) |
| + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) |
| + defer cancel() |
| + if err := p.stop(ctx, nil); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := armedMovesWorker.Load().cfg.CheckpointFor(ctx, rec, dbPath); err != nil { |
| + t.Fatal(err) |
| + } |
| + if f.replicaPuts() == 0 { |
| + t.Fatal("the unmetered fallback did not replicate") |
| + } |
| + if !p.bindingFailed.Load() { |
| + t.Fatal("an unmetered checkpoint filed no escape") |
| + } |
| +} |