amadan
Add the reader of the box-wide replicator's state, unwired
The scoped attestation needs to know what the external litestream unit replicates. replicatorProbe reads its config, the config's modification time, and from systemd whether the unit is running and since when. It will not speak for anything it cannot fully account for. A config that is not, line for line, what the generator writes is refused whole (ops.GeneratedLitestreamDBs): the existing parser reads only the generated shape and would silently miss a database written any other way. A running unit with no config, a unit between running and stopped, a start time systemd did not give in unix form, and a config or unit that changed while being read are all refused. A refusal attests nothing. Nothing in the agent calls it. The attestation plan records it, and a prerequisite it uncovered: the flagship's config is kept by hand, so the flagship attests nothing until that config is cut over to the generated one. Also a comment on the teardown guard naming the one thing it can leave behind, a scanner plan on a bucket that was shared. Built after review round five began on daa32cb6, so it is unreviewed.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
6 files changed,
+385
−1
docs/superpowers/plans/2026-10-01-durability-attestation-path.md+5 −1internal/host/durabilityreplicator.go+150 −0internal/host/durabilityreplicator_test.go+115 −0internal/objstore/walk.go+7 −0internal/ops/litestream.go+56 −0internal/ops/litestream_generated_test.go+52 −0
CI passed — run details
diff --git a/docs/superpowers/plans/2026-10-01-durability-attestation-path.md b/docs/superpowers/plans/2026-10-01-durability-attestation-path.md| index 4e84e93b4..577bb6dfd 100644 |
| --- a/docs/superpowers/plans/2026-10-01-durability-attestation-path.md |
| +++ b/docs/superpowers/plans/2026-10-01-durability-attestation-path.md |
| @@ -41,9 +41,13 @@ Built and tested on the branch, with no effect in production: |
| - **The scope computation** (`internal/host/durabilityscope.go`, `durabilityScope`). It excludes the owner of every unit-backed route, of every exec route that records a database and does not hibernate, and of every route whose database the live config names. It attests nothing 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. `TestDurabilityScope` pins each rule. |
| - **The test this plan required first.** `TestScopedAttestationPublishesADayIncompleteForTheUnitOwner`: 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. |
| +- **The reader of the replicator's state** (`internal/host/durabilityreplicator.go`, `replicatorProbe`, with `ops.GeneratedLitestreamDBs`). It reads the config, its modification time, and from systemd whether the unit is running and since when. It refuses a config that is not, line for line, what the generator writes; a running unit with no config; a unit in any state between running and stopped; and a config or unit that changed while it was being read. `TestGeneratedLitestreamDBsRefusesAnythingElse`, `TestParseSystemdUnitState` and `TestReplicatorProbe` pin it. Built after round five of the review began, so it is unreviewed. |
| + |
| +A prerequisite this uncovered: **the flagship's `/etc/litestream.yml` is kept by hand** (`internal/ops/litestream.go` says so). The reader will not speak for a hand-kept config, so the flagship attests nothing until its config is cut over to the generated one. That cutover is its own piece of work with its own risk (`ops.CoverageShortfall` exists for it) and is not part of this plan. |
| + |
| Not built, in the order it should be: |
| -1. The reader of the replicator's state: the config's paths, its modification time, and the unit's start time from systemd. It must refuse a config it did not fully account for (`ops.ParseLitestreamDBs` reads only the generated shape), and resolve the console database path the way the generator does. |
| +1. The reader must still resolve the console database path the way the generator does (`ops.consoleDBPath`); today the scope assumes `<data>/console/console.db`. |
| 2. The agent's call, on open and before every seal: compute the scope, and exclude any account that has newly appeared from the earliest instant the external replicator could have known its database (the config's modification time), through `DurabilityLedger.Escaped`. The call must be behind an explicit switch that is off by default. |
| 3. An independent review of 1 and 2. This is the code that makes attestation read true. |
| 4. A scoped approval to set the switch on the flagship, after a dev box has run with it for a week and its published days have been read by hand. |
diff --git a/internal/host/durabilityreplicator.go b/internal/host/durabilityreplicator.go| new file mode 100644 |
| index 000000000..423c026fd |
| --- /dev/null |
| +++ b/internal/host/durabilityreplicator.go |
| @@ -0,0 +1,150 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "os/exec" |
| + "strconv" |
| + "strings" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/ops" |
| +) |
| + |
| +// The reader of the external replicator's state, for durabilityScope. |
| +// Like the scope, NOTHING IN THE AGENT CALLS THIS YET. |
| + |
| +const ( |
| + externalReplicatorUnit = "carlos-litestream.service" |
| + externalReplicatorConfig = "/etc/litestream.yml" |
| +) |
| + |
| +// replicatorProbe reads what the box can observe about the box-wide |
| +// litestream unit. Its three functions are seams; newReplicatorProbe wires |
| +// the real ones. |
| +type replicatorProbe struct { |
| + configPath string |
| + unit string |
| + readFile func(string) ([]byte, error) |
| + modTime func(string) (time.Time, error) |
| + // unitState reports whether the unit is active and since when. |
| + unitState func(ctx context.Context, unit string) (active bool, since time.Time, err error) |
| +} |
| + |
| +func newReplicatorProbe() replicatorProbe { |
| + return replicatorProbe{ |
| + configPath: externalReplicatorConfig, |
| + unit: externalReplicatorUnit, |
| + readFile: os.ReadFile, |
| + modTime: func(path string) (time.Time, error) { |
| + fi, err := os.Stat(path) |
| + if err != nil { |
| + return time.Time{}, err |
| + } |
| + return fi.ModTime(), nil |
| + }, |
| + unitState: systemdUnitState, |
| + } |
| +} |
| + |
| +// read observes the replicator. Anything it cannot establish is an Err, and |
| +// an Err attests nothing (durabilityScope). |
| +// |
| +// The unit is read before and after the config. If it changed in between |
| +// (a restart landed while this was reading), the config read may belong to |
| +// neither state, so nothing is concluded this time. |
| +// |
| +// A config that is not, line for line, what the generator writes is an |
| +// Err: this code will not speak for a hand-kept config, because it cannot |
| +// know it has seen every database in it. A box whose config is still kept |
| +// by hand must be cut over to the generated one before it can attest. |
| +func (p replicatorProbe) read(ctx context.Context) replicatorState { |
| + active, since, err := p.unitState(ctx, p.unit) |
| + if err != nil { |
| + return replicatorState{Err: fmt.Errorf("unit %s: %w", p.unit, err)} |
| + } |
| + out := replicatorState{UnitActive: active, UnitStarted: since} |
| + raw, err := p.readFile(p.configPath) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + // No config: the external replicator has nothing to replicate. A |
| + // unit that is nevertheless running is replicating from something |
| + // this code cannot see. |
| + if active { |
| + return replicatorState{Err: fmt.Errorf("unit %s is running and %s does not exist", p.unit, p.configPath)} |
| + } |
| + return out |
| + case err != nil: |
| + return replicatorState{Err: fmt.Errorf("read %s: %w", p.configPath, err)} |
| + } |
| + modified, err := p.modTime(p.configPath) |
| + if err != nil { |
| + return replicatorState{Err: fmt.Errorf("stat %s: %w", p.configPath, err)} |
| + } |
| + paths, err := ops.GeneratedLitestreamDBs(string(raw)) |
| + if err != nil { |
| + return replicatorState{Err: fmt.Errorf("%s: %w", p.configPath, err)} |
| + } |
| + // The config again, and the unit again: both must be what they were. |
| + again, err := p.readFile(p.configPath) |
| + if err != nil || string(again) != string(raw) { |
| + return replicatorState{Err: fmt.Errorf("%s changed while it was being read", p.configPath)} |
| + } |
| + active2, since2, err := p.unitState(ctx, p.unit) |
| + if err != nil { |
| + return replicatorState{Err: fmt.Errorf("unit %s: %w", p.unit, err)} |
| + } |
| + if active2 != active || !since2.Equal(since) { |
| + return replicatorState{Err: fmt.Errorf("unit %s changed state while its config was being read", p.unit)} |
| + } |
| + out.ConfigPresent, out.Paths, out.ConfigModified = true, paths, modified |
| + return out |
| +} |
| + |
| +// systemdUnitState asks systemd whether unit is active and when it became |
| +// so. The timestamp is requested in unix form: the default rendering |
| +// carries a zone abbreviation, which cannot be parsed back reliably. |
| +func systemdUnitState(ctx context.Context, unit string) (bool, time.Time, error) { |
| + out, err := exec.CommandContext(ctx, "systemctl", "show", unit, "--timestamp=unix", "-p", "LoadState", "-p", "ActiveState", "-p", "ActiveEnterTimestamp").Output() |
| + if err != nil { |
| + return false, time.Time{}, err |
| + } |
| + return parseSystemdUnitState(string(out)) |
| +} |
| + |
| +// parseSystemdUnitState reads `systemctl show --timestamp=unix` output for |
| +// LoadState, ActiveState and ActiveEnterTimestamp. Only two answers are |
| +// accepted: a unit that is active with a start time, and a unit that is |
| +// plainly not running (inactive or failed, or not installed at all). Any |
| +// state in between (activating, deactivating, reloading), a missing |
| +// property or a timestamp in another form is an error. |
| +func parseSystemdUnitState(out string) (bool, time.Time, error) { |
| + props := map[string]string{} |
| + for _, line := range strings.Split(strings.TrimSpace(out), "\n") { |
| + if k, v, ok := strings.Cut(line, "="); ok { |
| + props[k] = strings.TrimSpace(v) |
| + } |
| + } |
| + load, state := props["LoadState"], props["ActiveState"] |
| + switch { |
| + case load == "not-found" && state == "inactive": |
| + return false, time.Time{}, nil |
| + case load != "loaded": |
| + return false, time.Time{}, fmt.Errorf("load state %q", load) |
| + case state == "inactive", state == "failed": |
| + return false, time.Time{}, nil |
| + case state != "active": |
| + return false, time.Time{}, fmt.Errorf("active state %q", state) |
| + } |
| + stamp, ok := strings.CutPrefix(props["ActiveEnterTimestamp"], "@") |
| + if !ok { |
| + return false, time.Time{}, fmt.Errorf("start time %q is not in unix form", props["ActiveEnterTimestamp"]) |
| + } |
| + secs, err := strconv.ParseInt(stamp, 10, 64) |
| + if err != nil || secs <= 0 { |
| + return false, time.Time{}, fmt.Errorf("start time %q", props["ActiveEnterTimestamp"]) |
| + } |
| + return true, time.Unix(secs, 0).UTC(), nil |
| +} |
diff --git a/internal/host/durabilityreplicator_test.go b/internal/host/durabilityreplicator_test.go| new file mode 100644 |
| index 000000000..068ce8731 |
| --- /dev/null |
| +++ b/internal/host/durabilityreplicator_test.go |
| @@ -0,0 +1,115 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "os" |
| + "reflect" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/ops" |
| +) |
| + |
| +func TestParseSystemdUnitState(t *testing.T) { |
| + cases := []struct { |
| + name string |
| + out string |
| + active bool |
| + since int64 |
| + fails bool |
| + }{ |
| + {"running", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=@1790817621\n", true, 1790817621, false}, |
| + {"stopped", "LoadState=loaded\nActiveState=inactive\nActiveEnterTimestamp=\n", false, 0, false}, |
| + {"failed", "LoadState=loaded\nActiveState=failed\nActiveEnterTimestamp=@1790817621\n", false, 0, false}, |
| + {"not installed", "LoadState=not-found\nActiveState=inactive\nActiveEnterTimestamp=\n", false, 0, false}, |
| + {"starting: neither running nor stopped", "LoadState=loaded\nActiveState=activating\nActiveEnterTimestamp=\n", false, 0, true}, |
| + {"stopping", "LoadState=loaded\nActiveState=deactivating\nActiveEnterTimestamp=@1790817621\n", false, 0, true}, |
| + {"a start time with a zone abbreviation", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=Thu 2026-10-01 01:40:21 IST\n", false, 0, true}, |
| + {"running with no start time", "LoadState=loaded\nActiveState=active\nActiveEnterTimestamp=\n", false, 0, true}, |
| + {"masked", "LoadState=masked\nActiveState=inactive\nActiveEnterTimestamp=\n", false, 0, true}, |
| + {"nothing at all", "", false, 0, true}, |
| + } |
| + for _, c := range cases { |
| + active, since, err := parseSystemdUnitState(c.out) |
| + if (err != nil) != c.fails || active != c.active || (c.since != 0 && since.Unix() != c.since) { |
| + t.Errorf("%s: active=%v since=%v err=%v", c.name, active, since, err) |
| + } |
| + } |
| +} |
| + |
| +func TestReplicatorProbe(t *testing.T) { |
| + ctx := context.Background() |
| + written := time.Date(2026, 9, 17, 8, 0, 0, 0, time.UTC) |
| + generated := ops.LitestreamConfig([]string{"/data/registry.db", "/data/api.example.com.db"}, "bucket", "prefix", "eu-west-1") |
| + probe := func(cfg *string, active bool, since time.Time) replicatorProbe { |
| + return replicatorProbe{configPath: "/etc/litestream.yml", unit: externalReplicatorUnit, |
| + readFile: func(string) ([]byte, error) { |
| + if cfg == nil { |
| + return nil, os.ErrNotExist |
| + } |
| + return []byte(*cfg), nil |
| + }, |
| + modTime: func(string) (time.Time, error) { return written, nil }, |
| + unitState: func(context.Context, string) (bool, time.Time, error) { return active, since, nil }, |
| + } |
| + } |
| + // The ordinary box: a generated config, the unit restarted after it. |
| + got := probe(&generated, true, written.Add(time.Minute)).read(ctx) |
| + if got.Err != nil || !got.ConfigPresent || !got.UnitActive || !got.ConfigModified.Equal(written) || |
| + !reflect.DeepEqual(got.Paths, []string{"/data/api.example.com.db", "/data/registry.db"}) { |
| + t.Fatalf("generated config: %+v", got) |
| + } |
| + // No config and no unit: nothing external replicates. |
| + if got := probe(nil, false, time.Time{}).read(ctx); got.Err != nil || got.ConfigPresent { |
| + t.Fatalf("no config, no unit: %+v", got) |
| + } |
| + // A running unit with no config is replicating from somewhere unseen. |
| + if got := probe(nil, true, written).read(ctx); got.Err == nil { |
| + t.Fatal("a running unit with no config was accepted") |
| + } |
| + // A hand-kept config is refused whole. |
| + hand := "dbs:\n - path: /data/registry.db\n replicas:\n - url: s3://bucket/x\n" |
| + if got := probe(&hand, true, written.Add(time.Minute)).read(ctx); !errors.Is(got.Err, ops.ErrConfigNotGenerated) || len(got.Paths) != 0 { |
| + t.Fatalf("a hand-kept config: %+v", got) |
| + } |
| + // The unit restarts while the config is being read. |
| + calls := 0 |
| + racing := probe(&generated, true, written.Add(time.Minute)) |
| + racing.unitState = func(context.Context, string) (bool, time.Time, error) { |
| + calls++ |
| + return true, written.Add(time.Duration(calls) * time.Minute), nil |
| + } |
| + if got := racing.read(ctx); got.Err == nil { |
| + t.Fatal("a unit that restarted mid-read was accepted") |
| + } |
| + // The config is rewritten while it is being read. |
| + reads := 0 |
| + rewriting := probe(&generated, true, written.Add(time.Minute)) |
| + rewriting.readFile = func(string) ([]byte, error) { |
| + reads++ |
| + if reads > 1 { |
| + return []byte(generated + " - path: /data/new.db\n"), nil |
| + } |
| + return []byte(generated), nil |
| + } |
| + if got := rewriting.read(ctx); got.Err == nil { |
| + t.Fatal("a config rewritten mid-read was accepted") |
| + } |
| + // systemd cannot be asked. |
| + deaf := probe(&generated, true, written) |
| + deaf.unitState = func(context.Context, string) (bool, time.Time, error) { |
| + return false, time.Time{}, errors.New("systemctl: not found") |
| + } |
| + if got := deaf.read(ctx); got.Err == nil { |
| + t.Fatal("an unreadable unit state was accepted") |
| + } |
| + // What the probe reads feeds the scope: an unreadable state attests |
| + // nothing, a readable one attests with its exclusions. |
| + if scope, _ := durabilityScope(scopeRoutes(), "/data", deaf.read(ctx)); scope.Attested { |
| + t.Fatal("an unreadable replicator attested") |
| + } |
| + if scope, reason := durabilityScope(scopeRoutes(), "/data", probe(&generated, true, written.Add(time.Minute)).read(ctx)); !scope.Attested { |
| + t.Fatalf("a readable replicator did not attest: %s", reason) |
| + } |
| +} |
diff --git a/internal/objstore/walk.go b/internal/objstore/walk.go| index bd4ec4094..b133ea732 100644 |
| --- a/internal/objstore/walk.go |
| +++ b/internal/objstore/walk.go |
| @@ -566,6 +566,13 @@ func Teardown(ctx context.Context, s release.Store, p Provisioner, sc Scanner, a |
| // and the delete below (the walk has no cross-process lock on a bucket |
| // name), and a holder provisioned before ledgers existed whose interval |
| // was never written. |
| + // |
| + // What it can leave behind: a scanner plan. If this app enabled scanning |
| + // on a bucket it shared and the last holder's record does not say so, |
| + // that holder's teardown deletes the bucket without disabling the plan, |
| + // and nothing names the plan afterwards. Leaving another app's bucket |
| + // unprotected would be worse, and the case needs an operator to have |
| + // shared an adopted bucket and scanned it under one app only. |
| sharedWith := "" |
| if bucket != "" && ValidBucketName(bucket) { |
| existing, _, lerr := ReadBucketOwnership(ctx, s, bucket) |
diff --git a/internal/ops/litestream.go b/internal/ops/litestream.go| index 21396ef01..8fa88badf 100644 |
| --- a/internal/ops/litestream.go |
| +++ b/internal/ops/litestream.go |
| @@ -22,10 +22,12 @@ package ops |
| import ( |
| "bytes" |
| + "errors" |
| "fmt" |
| "os" |
| "os/exec" |
| "path/filepath" |
| + "regexp" |
| "sort" |
| "strconv" |
| "strings" |
| @@ -184,6 +186,60 @@ func ParseLitestreamDBs(cfg string) []string { |
| return dbs |
| } |
| +// ErrConfigNotGenerated means a config is not, line for line, what |
| +// LitestreamConfig writes. |
| +var ErrConfigNotGenerated = errors.New("litestream config is not the generated shape") |
| + |
| +var generatedConfigLines = []*regexp.Regexp{ |
| + regexp.MustCompile(`^#.*$`), |
| + regexp.MustCompile(`^snapshot:$`), |
| + regexp.MustCompile(`^ interval: \S+$`), |
| + regexp.MustCompile(`^ retention: \S+$`), |
| + regexp.MustCompile(`^dbs:$`), |
| + regexp.MustCompile(`^ replicas:$`), |
| + regexp.MustCompile(`^ - url: s3://\S+$`), |
| + regexp.MustCompile(`^ region: \S+$`), |
| +} |
| + |
| +var generatedConfigPath = regexp.MustCompile(`^ - path: (/\S+)$`) |
| + |
| +// GeneratedLitestreamDBs is ParseLitestreamDBs for a caller that must know |
| +// it has seen EVERY database the config names, not only the ones written |
| +// the way the generator writes them. It accounts for every line of the |
| +// file: each must be one LitestreamConfig produces, and a file with any |
| +// other line (a hand-kept config, a second replica, a directory entry, an |
| +// include, a tab) is refused with ErrConfigNotGenerated. It is not a YAML |
| +// parser and does not try to be; a config it cannot fully account for is |
| +// one it will not speak for. |
| +func GeneratedLitestreamDBs(cfg string) ([]string, error) { |
| + if !strings.HasPrefix(cfg, "# GENERATED by `carlos ops litestream`") { |
| + return nil, ErrConfigNotGenerated |
| + } |
| + var dbs []string |
| + seen := map[string]bool{} |
| + for n, line := range strings.Split(strings.TrimSuffix(cfg, "\n"), "\n") { |
| + if m := generatedConfigPath.FindStringSubmatch(line); m != nil { |
| + if seen[m[1]] { |
| + return nil, fmt.Errorf("%w: line %d names %s twice", ErrConfigNotGenerated, n+1, m[1]) |
| + } |
| + seen[m[1]] = true |
| + dbs = append(dbs, m[1]) |
| + continue |
| + } |
| + known := false |
| + for _, re := range generatedConfigLines { |
| + if re.MatchString(line) { |
| + known = true |
| + break |
| + } |
| + } |
| + if !known { |
| + return nil, fmt.Errorf("%w: line %d", ErrConfigNotGenerated, n+1) |
| + } |
| + } |
| + return dbs, nil |
| +} |
| + |
| // CoverageShortfall names every database the EXISTING config replicates that |
| // the proposed set does not — the databases a write would stop replicating. |
| // |
diff --git a/internal/ops/litestream_generated_test.go b/internal/ops/litestream_generated_test.go| new file mode 100644 |
| index 000000000..5ff612dda |
| --- /dev/null |
| +++ b/internal/ops/litestream_generated_test.go |
| @@ -0,0 +1,52 @@ |
| +package ops |
| + |
| +import ( |
| + "errors" |
| + "reflect" |
| + "strings" |
| + "testing" |
| +) |
| + |
| +// The strict reader accounts for every line the generator writes, with and |
| +// without a region, and returns exactly the generator's databases. |
| +func TestGeneratedLitestreamDBsRoundTrip(t *testing.T) { |
| + paths := []string{"/data/registry.db", "/data/console/console.db", "/var/lib/carlos-app/kass.example.com/instances/a.db"} |
| + for _, region := range []string{"eu-west-1", ""} { |
| + cfg := LitestreamConfig(paths, "bucket", "prefix", region) |
| + got, err := GeneratedLitestreamDBs(cfg) |
| + if err != nil { |
| + t.Fatalf("region %q: %v\n%s", region, err, cfg) |
| + } |
| + want := ParseLitestreamDBs(cfg) |
| + if !reflect.DeepEqual(got, want) || len(got) != len(paths) { |
| + t.Fatalf("region %q: got %v, the loose parser reads %v", region, got, want) |
| + } |
| + } |
| + if got, err := GeneratedLitestreamDBs(LitestreamConfig(nil, "bucket", "prefix", "eu-west-1")); err != nil || len(got) != 0 { |
| + t.Fatalf("an empty generated config: %v %v", got, err) |
| + } |
| +} |
| + |
| +// Anything the generator does not write is refused: the loose parser would |
| +// read such a file and silently miss what it does not recognise. |
| +func TestGeneratedLitestreamDBsRefusesAnythingElse(t *testing.T) { |
| + base := LitestreamConfig([]string{"/data/registry.db"}, "bucket", "prefix", "eu-west-1") |
| + cases := map[string]string{ |
| + "a hand-kept config with no marker": strings.SplitN(base, "\n", 2)[1], |
| + "a database written in flow style": base + " - {path: /data/hidden.db}\n", |
| + "a database indented differently": base + " - path: /data/hidden.db\n", |
| + "a directory entry": base + " - dir: /var/lib/tenants\n", |
| + "a second replica": base + " - type: sftp\n", |
| + "a relative path": base + " - path: data/relative.db\n", |
| + "a path with a trailing comment": base + " - path: /data/x.db # and more\n", |
| + "the same database twice": base + " - path: /data/registry.db\n", |
| + "a tab": strings.Replace(base, " - path:", "\t- path:", 1), |
| + "a key the generator never writes": base + "addr: :9090\n", |
| + "an empty file": "", |
| + } |
| + for name, cfg := range cases { |
| + if got, err := GeneratedLitestreamDBs(cfg); !errors.Is(err, ErrConfigNotGenerated) { |
| + t.Errorf("%s: got %v %v, want ErrConfigNotGenerated", name, got, err) |
| + } |
| + } |
| +} |