amadan

carlos / platform Public

WIP: a metered litestream store with no lease, the first piece of the platform's own replicator

Path B of the attestation plan: the platform does the replica writes itself,
through clients that count. This is a litestream store over any number of
databases in which every replica client carries a request meter bound to that
database's owner. It refuses a database with no meter: there is no unmetered
fallback. Tested against the fake S3: every request lands on its owner's
meter and nothing reaches the bucket that no meter saw.

Nothing calls it. It does not replace the external binary, change a unit or
attest anything. Not gated, not reviewed.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude a4fb842a4939c42fff5050031d185debea94a7ba parent 72b4f59
2 files changed, +193 −0
  • internal/lease/meteredstore.go +103 −0
  • internal/lease/meteredstore_test.go +90 −0
diff --git a/internal/lease/meteredstore.go b/internal/lease/meteredstore.go
new file mode 100644
index 000000000..032357bcd
--- /dev/null
+++ b/internal/lease/meteredstore.go
@@ -0,0 +1,103 @@
+package lease
+
+import (
+ "errors"
+ "fmt"
+ "path/filepath"
+ "time"
+
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+ "github.com/benbjohnson/litestream"
+)
+
+// The platform's own replicator, first piece (path B in
+// docs/superpowers/plans/2026-10-01-durability-attestation-path.md).
+//
+// Why. Replication writes are billed, and a box may attest its day only if
+// every write to the replica bucket went through the durability ledger.
+// Exec-backed routes are replicated by the activator through this package
+// and are metered. Unit-backed routes and the box's own databases are
+// replicated by an external litestream binary that the ledger cannot see,
+// which is what keeps every box unattested. Inspecting that binary from
+// outside cannot prove what it writes (path A was parked for that). The
+// way to know is for the platform to do the writing itself, through a
+// client that counts.
+//
+// What this is. A litestream store over any number of databases, with no
+// hibernation lease (these databases are not hibernated; the lease belongs
+// to the activator's wake and sleep), in which every replica client carries
+// a request meter bound to that database's owner.
+//
+// What it refuses. A database with no meter. There is no unmetered
+// fallback here, on purpose: a write that is not counted is exactly what
+// the attestation would be lying about.
+//
+// What it is not, yet. Nothing calls it. It does not replace the external
+// binary, change a unit, or attest anything. Replacing the binary also
+// needs the replica credentials to stop being readable by anything else on
+// the box, a restore path that does not shell out to the external binary,
+// and a replica prefix the old binary's files do not share. None of that is
+// built.
+
+// MeteredDB is one database for the platform's replicator.
+type MeteredDB struct {
+ // Path is the SQLite file.
+ Path string
+ // Prefix is the replica's path inside the bucket.
+ Prefix string
+ // Meter records every request the replica client makes. It is bound
+ // by the caller to the database's owner. Required.
+ Meter RequestRecorder
+ // SyncInterval is litestream's replica sync tick. Zero keeps its
+ // default.
+ SyncInterval time.Duration
+}
+
+// NewMeteredStore builds one litestream store over dbs, each replica client
+// metered by its own recorder. The store is returned unopened. It refuses
+// an empty list, a database with no path, prefix or meter, and two
+// databases that share a file or a replica prefix (two owners must never
+// be billed for one stream, nor one stream written twice).
+func NewMeteredStore(bucket, region string, dbs []MeteredDB) (*litestream.Store, error) {
+ if bucket == "" {
+ return nil, errors.New("lease: metered store: no bucket")
+ }
+ if len(dbs) == 0 {
+ return nil, errors.New("lease: metered store: no databases")
+ }
+ paths, prefixes := map[string]bool{}, map[string]bool{}
+ built := make([]*litestream.DB, 0, len(dbs))
+ for _, d := range dbs {
+ switch {
+ case d.Path == "" || !filepath.IsAbs(d.Path):
+ return nil, fmt.Errorf("lease: metered store: database path %q is not absolute", d.Path)
+ case d.Prefix == "":
+ return nil, fmt.Errorf("lease: metered store: %s has no replica prefix", d.Path)
+ case d.Meter == nil:
+ return nil, fmt.Errorf("lease: metered store: %s: %w", d.Path, errRequestMeterRequired)
+ case paths[filepath.Clean(d.Path)]:
+ return nil, fmt.Errorf("lease: metered store: %s is listed twice", d.Path)
+ case prefixes[d.Prefix]:
+ return nil, fmt.Errorf("lease: metered store: replica prefix %q is used by two databases", d.Prefix)
+ }
+ paths[filepath.Clean(d.Path)], prefixes[d.Prefix] = true, true
+ option, err := requestMeterOption(d.Meter)
+ if err != nil {
+ return nil, fmt.Errorf("lease: metered store: %s: %w", d.Path, err)
+ }
+ db := litestream.NewDB(d.Path)
+ // A metadata directory of its own: sharing one with another
+ // replicator (the activator's, or the external binary's default)
+ // corrupts shadow state, even across sequential runs.
+ db.SetMetaPath(filepath.Join(filepath.Dir(d.Path), ".replicator-"+filepath.Base(d.Path)+"-litestream"))
+ db.MonitorInterval = dbMonitorInterval
+ db.Replica = litestream.NewReplicaWithClient(db, newReplicaClient(bucket, d.Prefix, region, []func(*awss3.Options){option}...))
+ if d.SyncInterval > 0 {
+ db.Replica.SyncInterval = d.SyncInterval
+ }
+ built = append(built, db)
+ }
+ store := litestream.NewStore(built, compactionLevels)
+ store.L0RetentionCheckInterval = l0RetentionCheckInterval
+ return store, nil
+}
diff --git a/internal/lease/meteredstore_test.go b/internal/lease/meteredstore_test.go
new file mode 100644
index 000000000..8ebc5ac62
--- /dev/null
+++ b/internal/lease/meteredstore_test.go
@@ -0,0 +1,90 @@
+package lease
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "path/filepath"
+ "testing"
+)
+
+// Two databases with two owners in one store. Every request each replica
+// client makes lands on that database's own meter, and nothing reaches the
+// bucket that no meter saw. That last line is the property an attestation
+// would rest on.
+func TestMeteredStoreCountsEveryRequestOnItsOwnersMeter(t *testing.T) {
+ f, endpoint := newFakeS3(t)
+ fakeCreds(t)
+ t.Setenv("CARLOS_S3_ENDPOINT", endpoint)
+ dir := t.TempDir()
+ one, two := &memoryRequestRecorder{}, &memoryRequestRecorder{}
+ store, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{
+ {Path: filepath.Join(dir, "one.db"), Prefix: "replicas/bab/one", Meter: one},
+ {Path: filepath.Join(dir, "two.db"), Prefix: "replicas/bac/two", Meter: two},
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ dbs := store.DBs()
+ if len(dbs) != 2 {
+ t.Fatalf("%d databases in the store, want 2", len(dbs))
+ }
+ if dbs[0].MetaPath() == dbs[1].MetaPath() || filepath.Base(dbs[0].MetaPath()) != ".replicator-one.db-litestream" {
+ t.Fatalf("metadata paths %q and %q: want one each, in the replicator's own namespace", dbs[0].MetaPath(), dbs[1].MetaPath())
+ }
+ ctx := context.Background()
+ if _, err := dbs[0].Replica.Client.WriteLTXFile(ctx, 0, 2, 2, bytes.NewReader(mustLTXBytes(t))); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := dbs[1].Replica.Client.WriteLTXFile(ctx, 0, 2, 2, bytes.NewReader(mustLTXBytes(t))); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := dbs[1].Replica.Client.WriteLTXFile(ctx, 0, 3, 3, bytes.NewReader(mustLTXBytes(t))); err != nil {
+ t.Fatal(err)
+ }
+ one.mu.Lock()
+ defer one.mu.Unlock()
+ two.mu.Lock()
+ defer two.mu.Unlock()
+ if len(one.requests) != 1 || len(two.requests) != 2 {
+ t.Fatalf("meters saw %d and %d requests, want 1 and 2", len(one.requests), len(two.requests))
+ }
+ for _, r := range append(append([]recordedRequest{}, one.requests...), two.requests...) {
+ if r.operation != "PutObject" || !r.finished || !r.outcome.Answered {
+ t.Fatalf("a request was not recorded whole: %+v", r)
+ }
+ }
+ if got := len(f.requests()); got != len(one.requests)+len(two.requests) {
+ t.Fatalf("%d requests reached the bucket and %d were metered: %v", got, len(one.requests)+len(two.requests), f.requests())
+ }
+}
+
+// There is no unmetered way in: a database with no meter is refused, and so
+// is anything that would bill two owners for one stream.
+func TestMeteredStoreRefusesWhatItCannotAttribute(t *testing.T) {
+ dir := t.TempDir()
+ m := &memoryRequestRecorder{}
+ ok := MeteredDB{Path: filepath.Join(dir, "a.db"), Prefix: "replicas/a", Meter: m}
+ for name, dbs := range map[string][]MeteredDB{
+ "no databases": nil,
+ "no meter": {{Path: ok.Path, Prefix: ok.Prefix}},
+ "no prefix": {{Path: ok.Path, Meter: m}},
+ "a relative path": {{Path: "a.db", Prefix: ok.Prefix, Meter: m}},
+ "one file listed twice": {ok, {Path: filepath.Join(dir, ".", "a.db"), Prefix: "replicas/b", Meter: m}},
+ "one prefix for two": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: ok.Prefix, Meter: m}},
+ "a meter missing in many": {ok, {Path: filepath.Join(dir, "b.db"), Prefix: "replicas/b"}},
+ } {
+ if store, err := NewMeteredStore("test-bucket", "eu-west-1", dbs); err == nil || store != nil {
+ t.Errorf("%s: accepted", name)
+ }
+ }
+ if _, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{{Path: ok.Path, Prefix: ok.Prefix}}); !errors.Is(err, errRequestMeterRequired) {
+ t.Errorf("no meter: %v, want the meter-required error", err)
+ }
+ if _, err := NewMeteredStore("", "eu-west-1", []MeteredDB{ok}); err == nil {
+ t.Error("no bucket: accepted")
+ }
+ if store, err := NewMeteredStore("test-bucket", "eu-west-1", []MeteredDB{ok}); err != nil || store == nil {
+ t.Errorf("a well-formed database: %v", err)
+ }
+}