amadan
WIP: the metered fleet store, refusing anything it cannot meter
Ties the metered litestream store to the owner of each database. It starts only when every database the box replicates has an owner and a recorder; one with neither stops the whole store and is named. Unopened, unwired, and it attests nothing: that needs the replica credentials held by nothing else on the box, which is a change to the box. Not gated, not reviewed.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2 files changed,
+160
−0
internal/ops/litestream_metered.go+71 −0internal/ops/litestream_metered_test.go+89 −0
diff --git a/internal/ops/litestream_metered.go b/internal/ops/litestream_metered.go| new file mode 100644 |
| index 000000000..a752607e7 |
| --- /dev/null |
| +++ b/internal/ops/litestream_metered.go |
| @@ -0,0 +1,71 @@ |
| +package ops |
| + |
| +import ( |
| + "errors" |
| + "fmt" |
| + "strings" |
| + |
| + "github.com/benbjohnson/litestream" |
| + |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// ErrUnmeterable: the box replicates a database that cannot be metered |
| +// against an account. The platform's replicator refuses to start over one, |
| +// because starting would mean either leaving it unreplicated or streaming |
| +// it uncounted, and a box that does the second cannot say every write went |
| +// through its ledger. |
| +var ErrUnmeterable = errors.New("ops: a replicated database has no owner to meter it against") |
| + |
| +// MeteredFleetStore builds the platform's own replicator for this box: one |
| +// litestream store over every database FleetDBs lists, each replica client |
| +// metered against the database's owner (path B in |
| +// docs/superpowers/plans/2026-10-01-durability-attestation-path.md). |
| +// |
| +// meter returns the recorder for one database, bound by the caller to the |
| +// box's durability ledger. It is asked once for each database. A nil |
| +// recorder refuses the whole store: there is no unmetered fallback. |
| +// |
| +// prefix is the replica lineage this replicator writes. It must not be one |
| +// the external binary has written to: the two do not share a format or a |
| +// metadata directory, and a restore from a prefix both have touched is not |
| +// something anyone has tested. |
| +// |
| +// The store is returned unopened, and nothing here attests anything. A box |
| +// may say that every replica write was counted only when this is the only |
| +// thing on it that holds the replica bucket's credentials, which is a |
| +// change to the box and not to this code. |
| +func MeteredFleetStore(reg *registry.Registry, dataDir, bucket, prefix, region string, meter func(OwnedDB) lease.RequestRecorder) (*litestream.Store, []OwnedDB, error) { |
| + if meter == nil { |
| + return nil, nil, fmt.Errorf("%w: no meter", ErrUnmeterable) |
| + } |
| + owned, unowned, _, err := FleetDBOwners(reg, dataDir) |
| + if err != nil { |
| + return nil, nil, err |
| + } |
| + if len(unowned) > 0 { |
| + return nil, nil, fmt.Errorf("%w: %s", ErrUnmeterable, strings.Join(unowned, ", ")) |
| + } |
| + if len(owned) == 0 { |
| + return nil, nil, errors.New("ops: no databases to replicate") |
| + } |
| + prefix = strings.Trim(prefix, "/") |
| + dbs := make([]lease.MeteredDB, 0, len(owned)) |
| + for _, o := range owned { |
| + rec := meter(o) |
| + if rec == nil { |
| + return nil, nil, fmt.Errorf("%w: no recorder for %s (%s/%s)", ErrUnmeterable, o.Path, o.Account, o.App) |
| + } |
| + key := replicaKey(o.Path) |
| + if prefix != "" { |
| + key = prefix + "/" + key |
| + } |
| + dbs = append(dbs, lease.MeteredDB{Path: o.Path, Prefix: key, Meter: rec}) |
| + } |
| + store, err := lease.NewMeteredStore(bucket, region, dbs) |
| + if err != nil { |
| + return nil, nil, err |
| + } |
| + return store, owned, nil |
| +} |
diff --git a/internal/ops/litestream_metered_test.go b/internal/ops/litestream_metered_test.go| new file mode 100644 |
| index 000000000..305b12248 |
| --- /dev/null |
| +++ b/internal/ops/litestream_metered_test.go |
| @@ -0,0 +1,89 @@ |
| +package ops |
| + |
| +import ( |
| + "errors" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/lease" |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +type nopRecorder struct{} |
| + |
| +func (nopRecorder) Start(string, time.Time) (uint64, error) { return 1, nil } |
| +func (nopRecorder) Finish(uint64, lease.RequestOutcome) error { return nil } |
| +func (nopRecorder) Abandon(uint64, time.Time) {} |
| +func (nopRecorder) Incomplete(time.Time) {} |
| +func (nopRecorder) Hold(time.Time) func() { return func() {} } |
| + |
| +// The platform's replicator starts only when every database it would |
| +// stream has an owner and a recorder. One that has neither stops the whole |
| +// store: it is not left out quietly and it is not streamed uncounted. |
| +func TestMeteredFleetStoreRefusesADatabaseItCannotMeter(t *testing.T) { |
| + dir := t.TempDir() |
| + reg, err := registry.Open(filepath.Join(dir, "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + owned := filepath.Join(dir, "a.example.com.db") |
| + if err := os.WriteFile(owned, []byte("x"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.Upsert(registry.Route{Host: "a.example.com", Account: "bac", App: "one", DBPath: owned}); err != nil { |
| + t.Fatal(err) |
| + } |
| + var asked []OwnedDB |
| + meter := func(o OwnedDB) lease.RequestRecorder { |
| + asked = append(asked, o) |
| + return nopRecorder{} |
| + } |
| + store, dbs, err := MeteredFleetStore(reg, dir, "replica-bucket", "/metered/v1/", "eu-west-1", meter) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(dbs) != 2 || len(asked) != 2 || len(store.DBs()) != 2 { |
| + t.Fatalf("store over %d databases, %d owners, meter asked %d times: want the route's database and the registry", len(store.DBs()), len(dbs), len(asked)) |
| + } |
| + for _, o := range asked { |
| + if o.Path == owned && (o.Account != "bac" || o.App != "one" || o.Platform) { |
| + t.Errorf("the route's database was metered against %+v", o) |
| + } |
| + if o.Path != owned && !o.Platform { |
| + t.Errorf("the box's own database was not metered against the platform: %+v", o) |
| + } |
| + } |
| + // A recorder that cannot be bound: nothing is replicated unmetered. |
| + _, _, err = MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", func(o OwnedDB) lease.RequestRecorder { |
| + if o.Path == owned { |
| + return nil |
| + } |
| + return nopRecorder{} |
| + }) |
| + if !errors.Is(err, ErrUnmeterable) || !strings.Contains(err.Error(), owned) { |
| + t.Fatalf("a database with no recorder: %v, want ErrUnmeterable naming it", err) |
| + } |
| + if _, _, err := MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", nil); !errors.Is(err, ErrUnmeterable) { |
| + t.Fatalf("no meter at all: %v", err) |
| + } |
| + // A database two accounts claim. |
| + shared := filepath.Join(dir, "shared.example.com.db") |
| + if err := os.WriteFile(shared, []byte("x"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, r := range []registry.Route{ |
| + {Host: "shared.example.com", Account: "bac", App: "one", DBPath: shared}, |
| + {Host: "shared2.example.com", Account: "bae", App: "two", DBPath: shared}, |
| + } { |
| + if err := reg.Upsert(r); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + _, _, err = MeteredFleetStore(reg, dir, "replica-bucket", "metered/v1", "eu-west-1", meter) |
| + if !errors.Is(err, ErrUnmeterable) || !strings.Contains(err.Error(), shared) { |
| + t.Fatalf("a database with no single owner: %v, want ErrUnmeterable naming it", err) |
| + } |
| +} |