amadan
Fold replication writes and AWS-billed store transfer into statements
Neither durability_writes_1k nor store_egress_gb was ever written into a statement, and MeterSource named no source for either, so every monthly freeze was refused with nothing to say which meter held it. durability_writes_1k (console/durabilityfold.go): each box's published replication ledger folds per account as a lower bound, and a day counts for the new source mapping only when every expected box published a day its ledger attests box-wide, closed, with one generation and no gap, conflict, recording failure or unknown outcome. What counts as a billable write is one function, usage.BillableReplicationWrites: answered 2xx write operations for the tenant's own replication or checkpoint; leases, restores, moves, platform buckets, reads, lists, deletes, failures and unanswered attempts are never billed. No production box attests, so the source reads zero days there and the month holds. store_egress_gb (console/storeegressfold.go): the month's AWS Cost and Usage Report delivery, read from a configured prefix of the deployment bucket, folds each account's charged, owned bucket-hours. The source is whole-month and complete only when a delivery exists, AWS has invoiced it, no charged line was unclassifiable and every charged bucket-hour has one owner or is a declared platform bucket. A configured report that is not yet provable holds the month open, because a closed statement is never rewritten. With no report configured nothing changes from before. Month evidence: bucket ownership ledgers are pinned by their projection onto the month, so a later teardown does not change what the month proved and a change inside the month does; discovery of a new durability box follows the same lifecycle rule as metrics and edge. Verified: go build, go vet, gofmt and go test ./... plus the vendored Litestream S3 package pass outside the sandbox (60 packages ok); the new fold tests were mutation-checked (removing the unowned-bucket hold and the attestation condition each fails a test). Not verified: any real AWS report or real ledger from a production box; no infrastructure for the report exists.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
13 files changed,
+1014
−34
cmd/console/main.go+24 −19internal/console/durabilityfold.go+91 −0internal/console/economicspass.go+26 −0internal/console/meterfolds_test.go+368 −0internal/console/server.go+11 −0internal/console/storeegressfold.go+214 −0internal/usage/billable_test.go+106 −0internal/usage/durability.go+41 −0internal/usage/durabilityledger_test.go+5 −2internal/usage/evidence.go+16 −10internal/usage/evidence_month.go+95 −0internal/usage/passes_test.go+2 −0internal/usage/usage.go+15 −3
diff --git a/cmd/console/main.go b/cmd/console/main.go| index 6b2395038..acbeebaef 100644 |
| --- a/cmd/console/main.go |
| +++ b/cmd/console/main.go |
| @@ -276,25 +276,30 @@ func serve() { |
| // opens it with — /fleet/architecture draws the box -> bucket edge |
| // from it. Empty on a CARLOS_DEPLOYMENT_DIR deployment, which |
| // correctly draws no bucket node. |
| - Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"), |
| - HomeRegion: strings.TrimSpace(os.Getenv("CONSOLE_HOME_REGION")), |
| - ReleasePub: pub, |
| - Key: key, |
| - EdgeIPs: edgeIPs(), |
| - SecretsPub: secretsPub(), |
| - AppsDomain: appsDomain(), |
| - AppsDomainAliasAccounts: strings.FieldsFunc(os.Getenv("CONSOLE_APPS_DOMAIN_ALIAS_ACCOUNTS"), func(r rune) bool { return r == ',' || r == ' ' }), |
| - SteeringHost: steeringHost(), |
| - RedirectHosts: redirectHosts(), |
| - DenyDomains: denyDomains(), |
| - StoreProvisioner: storeProvisioner, |
| - StoreScanner: storeScanner, |
| - ScanSeedFromDeprecatedEnv: seedAntivirus, |
| - ScanBudget: envDuration("CARLOS_STORE_SCAN_BUDGET", time.Minute), |
| - ScanInterval: envDuration("CARLOS_STORE_SCAN_INTERVAL", 24*time.Hour), |
| - ScanMaxSigAge: envDuration("CARLOS_CLAMD_MAX_SIG_AGE", 168*time.Hour), |
| - StoreDeployment: storeDeployment, |
| - StoreOperators: storeOperators(), |
| + Bucket: os.Getenv("CARLOS_DEPLOYMENT_BUCKET"), |
| + HomeRegion: strings.TrimSpace(os.Getenv("CONSOLE_HOME_REGION")), |
| + ReleasePub: pub, |
| + Key: key, |
| + EdgeIPs: edgeIPs(), |
| + SecretsPub: secretsPub(), |
| + AppsDomain: appsDomain(), |
| + AppsDomainAliasAccounts: strings.FieldsFunc(os.Getenv("CONSOLE_APPS_DOMAIN_ALIAS_ACCOUNTS"), func(r rune) bool { return r == ',' || r == ' ' }), |
| + // The AWS Cost and Usage Report prefix in the deployment bucket, and |
| + // the platform's own buckets, for the store-transfer meter. Unset |
| + // means the meter holds (console/storeegressfold.go). |
| + StoreEgressReportPrefix: strings.TrimSpace(os.Getenv("CONSOLE_STORE_EGRESS_REPORT_PREFIX")), |
| + StoreEgressPlatformBuckets: strings.FieldsFunc(os.Getenv("CONSOLE_STORE_EGRESS_PLATFORM_BUCKETS"), func(r rune) bool { return r == ',' || r == ' ' }), |
| + SteeringHost: steeringHost(), |
| + RedirectHosts: redirectHosts(), |
| + DenyDomains: denyDomains(), |
| + StoreProvisioner: storeProvisioner, |
| + StoreScanner: storeScanner, |
| + ScanSeedFromDeprecatedEnv: seedAntivirus, |
| + ScanBudget: envDuration("CARLOS_STORE_SCAN_BUDGET", time.Minute), |
| + ScanInterval: envDuration("CARLOS_STORE_SCAN_INTERVAL", 24*time.Hour), |
| + ScanMaxSigAge: envDuration("CARLOS_CLAMD_MAX_SIG_AGE", 168*time.Hour), |
| + StoreDeployment: storeDeployment, |
| + StoreOperators: storeOperators(), |
| // CARLOS_ACME_DELEGATION_ZONE — the same variable the edge reads |
| // (internal/edge/wildcard.go's newWildcardFromEnv) for the same |
| // delegated-ACME story; NewServer normalizes it (lowercase, trim). |
diff --git a/internal/console/durabilityfold.go b/internal/console/durabilityfold.go| new file mode 100644 |
| index 000000000..440d50d82 |
| --- /dev/null |
| +++ b/internal/console/durabilityfold.go |
| @@ -0,0 +1,91 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "log" |
| + "os" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +// The durability_writes_1k fold (pricing spec P6). |
| +// |
| +// Every box publishes usage/durability/<box>/days/<date>.json from its |
| +// replication-request ledger (usage/durability.go). sweepUsage asks the |
| +// same two questions of it as of metrics and edge: what did the box record |
| +// (folded whatever the answer to the second question is, as a lower bound), |
| +// and does what it published PROVE it recorded every request. |
| +// |
| +// The proof is the ledger's own contract (usage.DurabilityDay.Scope) over |
| +// the box's expected span of the day, and it is strict: exactly one ledger |
| +// generation, no conflict, no gap, no recording failure, no unknown |
| +// outcome, the day closed, and coverage attested for EVERY account. A day |
| +// attested for everyone except some accounts does not count either, |
| +// because a statement's SourceDays is one number shared by every account, |
| +// and it cannot say "complete for all but these". That errs toward holding. |
| +// |
| +// No production box attests today (the root litestream unit replicates |
| +// unit-backed routes outside the ledger), so this source reads zero days |
| +// in production and the month holds, which is the intended state until the |
| +// attestation path in docs/superpowers/plans lands. |
| + |
| +// foldDurabilityDay folds one UTC day of every candidate box's published |
| +// ledger into the sweep's per-account meters and reports whether the day is |
| +// complete for billing. It never sets the sweep's covered-day flag: that |
| +// legacy number is metrics, edge and storage only. |
| +func (s *Server) foldDurabilityDay(ctx context.Context, cands []sourceCandidate, d time.Time, date string, lifecycleOK bool, sweep *usageSweep, touch func(string) map[string]float64) bool { |
| + // No candidate proves nothing: a deployment that has never published a |
| + // ledger has no evidence that nothing replicated. |
| + complete := lifecycleOK && len(cands) > 0 |
| + for _, cand := range cands { |
| + box := cand.Box |
| + wFrom, wTo, expected := cand.window(usage.SourceDurability, d) |
| + key := usage.DurabilityDayKey(box, date) |
| + raw, _, gerr := s.cfg.Releases.Get(ctx, key) |
| + if gerr != nil { |
| + if !errors.Is(gerr, os.ErrNotExist) { |
| + log.Printf("console: sweep usage: read %s: %v", key, gerr) |
| + sweep.ReadFailed = true |
| + complete = false |
| + } else if expected { |
| + complete = false // expected and silent |
| + } |
| + continue |
| + } |
| + day, perr := usage.ParseDurabilityDay(raw, box, date) |
| + if perr != nil { |
| + log.Printf("console: sweep usage: parse %s: %v", key, perr) |
| + sweep.ReadFailed = true |
| + complete = false |
| + continue |
| + } |
| + if !expected { |
| + // Published for a day the lifecycle record says the box did |
| + // not exist. The rows still fold; the record is contradicted. |
| + log.Printf("console: sweep usage: %s exists outside box %s's lifecycle interval", key, box) |
| + complete = false |
| + } else if ok, unmetered, _ := day.Scope(wFrom, wTo); !ok || len(unmetered) > 0 { |
| + complete = false |
| + } |
| + for _, snap := range day.Epochs { |
| + content, derr := snap.Decode() |
| + if derr != nil { |
| + // ParseDurabilityDay already decoded every snapshot; a |
| + // failure here means the object changed under us. |
| + sweep.ReadFailed = true |
| + complete = false |
| + continue |
| + } |
| + for _, rows := range content.Hours { |
| + for _, row := range rows { |
| + if n := usage.BillableReplicationWrites(row); n > 0 { |
| + touch(row.Account)[usage.MeterDurabilityWrites1k] += float64(n) / 1000 |
| + } |
| + } |
| + } |
| + } |
| + } |
| + return complete |
| +} |
diff --git a/internal/console/economicspass.go b/internal/console/economicspass.go| index 0aa82258d..a94f447b3 100644 |
| --- a/internal/console/economicspass.go |
| +++ b/internal/console/economicspass.go |
| @@ -549,6 +549,10 @@ func (s *Server) sweepUsage(ctx context.Context, from, to time.Time, routes []re |
| if err != nil { |
| return nil, fmt.Errorf("console: sweep usage: %w", err) |
| } |
| + durabilityBoxes, err := s.cfg.Releases.ListDirs(ctx, usage.DurabilityBoxesPrefix) |
| + if err != nil { |
| + return nil, fmt.Errorf("console: sweep usage: %w", err) |
| + } |
| sweep := &usageSweep{ |
| PerAccount: map[string]map[string]float64{}, |
| @@ -581,6 +585,7 @@ func (s *Server) sweepUsage(ctx context.Context, from, to time.Time, routes []re |
| } |
| metricsCands := sourceCandidates(usage.SourceMetrics, boxDirNames(metricsBoxesPrefix, metricsBoxes), lifecycles) |
| edgeCands := sourceCandidates(usage.SourceEdge, boxDirNames(usage.EdgeBoxesPrefix, edgeBoxes), lifecycles) |
| + durabilityCands := sourceCandidates(usage.SourceDurability, boxDirNames(usage.DurabilityBoxesPrefix, durabilityBoxes), lifecycles) |
| touch := func(acct string) map[string]float64 { |
| mm := sweep.PerAccount[acct] |
| @@ -805,6 +810,11 @@ func (s *Server) sweepUsage(ctx context.Context, from, to time.Time, routes []re |
| } |
| } |
| + // Replication writes (durabilityfold.go): folded as a lower bound |
| + // whatever the proof says; the day counts only when every expected |
| + // box's ledger attests it box-wide. |
| + durabilityOK := s.foldDurabilityDay(ctx, durabilityCands, d, date, lifecycleOK, sweep, touch) |
| + |
| // Storage is read ONCE here and cached (sweep.Storage), the same way |
| // census is, so foldStorageAndCensus folds the EXACT bytes this |
| // completeness check saw — previously a second, independent Get |
| @@ -876,6 +886,9 @@ func (s *Server) sweepUsage(ctx context.Context, from, to time.Time, routes []re |
| if edgeOK { |
| sweep.SourceDays[usage.SourceEdge]++ |
| } |
| + if durabilityOK { |
| + sweep.SourceDays[usage.SourceDurability]++ |
| + } |
| if storageOK { |
| sweep.SourceDays[usage.SourceStorage]++ |
| } |
| @@ -1315,10 +1328,23 @@ func (s *Server) statementsForMonth(ctx context.Context, month string, start, en |
| dst[meter] += v |
| } |
| } |
| + // Object-store transfer, from the month's AWS Cost and Usage Report |
| + // (storeegressfold.go). Whole-month: complete or not at all. |
| + storeEgress := reader.foldStoreEgress(ctx, start, meters) |
| + if storeEgress.Complete { |
| + sweep.SourceDays[usage.SourceStoreEgress] = daysInMonth |
| + } |
| if closing && (sweep.ReadFailed || fleetReadFailed) { |
| log.Printf("console: statements pass: %s: read failure during the closing sweep; leaving the month open and retrying next hour", month) |
| closing = false |
| } |
| + if closing && storeEgress.HoldClose { |
| + // A closed statement is never rewritten, so closing before AWS has |
| + // invoiced the month (or before an ownership hole is repaired) |
| + // would leave the month unbillable for good. |
| + log.Printf("console: statements pass: %s: %s; leaving the month open and retrying next hour", month, storeEgress.Reason) |
| + closing = false |
| + } |
| evidence, evidenceErr := inputs.Snapshot() |
| if evidenceErr == nil && closing { |
| evidenceErr = evidence.Verify(ctx, s.cfg.Releases) |
diff --git a/internal/console/meterfolds_test.go b/internal/console/meterfolds_test.go| new file mode 100644 |
| index 000000000..faad6e6cb |
| --- /dev/null |
| +++ b/internal/console/meterfolds_test.go |
| @@ -0,0 +1,368 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/awsbilling" |
| + "github.com/carlosframework/platform/internal/objstore" |
| + "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +const ( |
| + foldReportPrefix = "billing/cur/carloku-transfer/" |
| + foldAcctA = "bcd" |
| + foldAcctB = "bce" |
| + foldBucketA = "carlos-store-bcd-shop" |
| + foldBucketB = "carlos-store-bce-blog" |
| +) |
| + |
| +var foldSept = time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) |
| + |
| +const foldCURHeader = "identity_line_item_id,identity_time_interval,bill_invoice_id,bill_billing_period_start_date,bill_billing_period_end_date,bill_payer_account_id,line_item_usage_account_id,line_item_line_item_type,line_item_usage_start_date,line_item_usage_end_date,line_item_product_code,line_item_usage_type,line_item_resource_id,line_item_usage_amount,line_item_unblended_rate,line_item_unblended_cost,line_item_line_item_description,product_servicecode\n" |
| + |
| +// curLine is one hourly AmazonS3 usage line for September 2026. |
| +func curLine(id, invoice, usageType, bucket, hour, gb, rate, cost string) string { |
| + start, err := time.Parse(time.RFC3339, hour) |
| + if err != nil { |
| + panic(err) |
| + } |
| + end := start.Add(time.Hour).Format(time.RFC3339) |
| + return strings.Join([]string{id, hour + "/" + end, invoice, "2026-09-01T00:00:00Z", "2026-10-01T00:00:00Z", "271376211898", "271376211898", "Usage", hour, end, "AmazonS3", usageType, bucket, gb, rate, cost, "test line", "AmazonS3"}, ",") + "\n" |
| +} |
| + |
| +// putCURDelivery writes an overwrite-layout CUR 2.0 delivery for September |
| +// into the deployment bucket under foldReportPrefix. |
| +func putCURDelivery(t *testing.T, fx *consoleFixture, id, body string) { |
| + t.Helper() |
| + ctx := context.Background() |
| + dataKey := awsbilling.DataPrefix(foldReportPrefix, foldSept) + "part-00001.csv" |
| + put := func(key string, raw []byte) { |
| + _, etag, err := fx.rel.Get(ctx, key) |
| + if err != nil { |
| + etag = "" |
| + } |
| + if _, err := fx.rel.Put(ctx, key, raw, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + put(dataKey, []byte(foldCURHeader+body)) |
| + manifest, _ := json.Marshal(map[string]any{ |
| + "executionId": id, |
| + "billingPeriod": map[string]string{"start": "2026-09-01T00:00:00.000Z", "end": "2026-10-01T00:00:00.000Z"}, |
| + "dataFiles": []string{"s3://deployment-bucket/" + dataKey}, |
| + }) |
| + put(awsbilling.MetadataPrefix(foldReportPrefix, foldSept)+"carloku-transfer-Manifest.json", manifest) |
| +} |
| + |
| +func septemberStatement(t *testing.T, fx *consoleFixture, acct string) usage.Statement { |
| + t.Helper() |
| + raw, _, err := fx.rel.Get(context.Background(), usage.StatementKey(acct, "2026-09")) |
| + if err != nil { |
| + t.Fatalf("statement %s: %v", acct, err) |
| + } |
| + var st usage.Statement |
| + if err := json.Unmarshal(raw, &st); err != nil { |
| + t.Fatal(err) |
| + } |
| + return st |
| +} |
| + |
| +func near(a, b float64) bool { d := a - b; return d < 1e-9 && d > -1e-9 } |
| + |
| +func openOwner(t *testing.T, fx *consoleFixture, bucket, acct, app string) { |
| + t.Helper() |
| + if err := objstore.OpenBucketOwner(context.Background(), fx.rel, bucket, acct, app, time.Date(2026, 8, 10, 0, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// The invoiced, fully owned month: charged transfer reaches each owner's |
| +// statement, zero-cost and CloudFront transfer never does, and the source |
| +// reads complete for the whole month. |
| +func TestStoreEgressFoldBillsOnlyChargedOwnedTransfer(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + fx.srv.cfg.StoreEgressReportPrefix = foldReportPrefix |
| + fx.srv.cfg.Bucket = "carlos-flagship-271376211898" |
| + openOwner(t, fx, foldBucketA, foldAcctA, "shop") |
| + openOwner(t, fx, foldBucketB, foldAcctB, "blog") |
| + putCURDelivery(t, fx, "exec-1", |
| + curLine("l1", "INV-1", "EUW1-DataTransfer-Out-Bytes", foldBucketA, "2026-09-03T10:00:00Z", "2.5", "0.09", "0.225")+ |
| + curLine("l2", "INV-1", "EUW1-DataTransfer-Regional-Bytes", foldBucketA, "2026-09-03T11:00:00Z", "1", "0.01", "0.01")+ |
| + curLine("l3", "INV-1", "EUW1-USE1-AWS-Out-Bytes", foldBucketB, "2026-09-04T00:00:00Z", "0.5", "0.02", "0.01")+ |
| + curLine("l4", "INV-1", "EUW1-DataTransfer-Out-Bytes", foldBucketA, "2026-09-02T00:00:00Z", "40", "0", "0")+ |
| + curLine("l5", "INV-1", "EUW1-CloudFront-Out-Bytes", foldBucketA, "2026-09-03T10:00:00Z", "7", "0", "0")+ |
| + curLine("l6", "INV-1", "EUW1-DataTransfer-Out-Bytes", "carlos-flagship-271376211898", "2026-09-05T00:00:00Z", "3", "0.09", "0.27")) |
| + |
| + if err := fx.srv.statementsPass(ctx, time.Date(2026, 10, 5, 6, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + a, b := septemberStatement(t, fx, foldAcctA), septemberStatement(t, fx, foldAcctB) |
| + if got := a.Meters[usage.MeterStoreEgressGB]; !near(got, 3.5) { |
| + t.Errorf("account A store egress = %v GB, want 3.5 (charged lines only)", got) |
| + } |
| + if got := b.Meters[usage.MeterStoreEgressGB]; !near(got, 0.5) { |
| + t.Errorf("account B store egress = %v GB, want 0.5", got) |
| + } |
| + if got := a.SourceDays[usage.SourceStoreEgress]; got != 30 { |
| + t.Errorf("store egress source days = %d, want 30", got) |
| + } |
| + if !a.MeterComplete(usage.MeterStoreEgressGB) { |
| + t.Error("an invoiced, fully owned month must read complete") |
| + } |
| + if !a.Closed { |
| + t.Error("a complete report must not hold the month open") |
| + } |
| + // The platform's own bucket is billed to nobody. |
| + if ops := septemberStatement(t, fx, fixtureAcct); ops.Meters[usage.MeterStoreEgressGB] != 0 { |
| + t.Errorf("platform bucket transfer reached the ops statement: %v", ops.Meters[usage.MeterStoreEgressGB]) |
| + } |
| + |
| + // The closed month's evidence survives the ledger growing afterwards. |
| + ev, err := a.Evidence.Load(ctx, fx.rel, "2026-09") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := objstore.CloseBucketOwner(ctx, fx.rel, foldBucketA, foldAcctA, "shop", time.Date(2026, 10, 12, 0, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := ev.Verify(ctx, fx.rel); err != nil { |
| + t.Fatalf("an October teardown changed September's evidence: %v", err) |
| + } |
| + // A restated delivery does change it: the month was billed on other bytes. |
| + putCURDelivery(t, fx, "exec-2", curLine("l1", "INV-1", "EUW1-DataTransfer-Out-Bytes", foldBucketA, "2026-09-03T10:00:00Z", "9", "0.09", "0.81")) |
| + if err := ev.Verify(ctx, fx.rel); !errors.Is(err, usage.ErrEvidenceChanged) { |
| + t.Fatalf("a restated report must read as changed evidence, got %v", err) |
| + } |
| +} |
| + |
| +// A charged bucket nobody can be shown to own holds the whole month: the |
| +// owned transfer still folds as a lower bound, nothing reads complete, and |
| +// the month stays open so a repaired ledger can complete it. |
| +func TestStoreEgressFoldHoldsOnUnownedBucket(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + fx.srv.cfg.StoreEgressReportPrefix = foldReportPrefix |
| + openOwner(t, fx, foldBucketA, foldAcctA, "shop") |
| + putCURDelivery(t, fx, "exec-1", |
| + curLine("l1", "INV-1", "EUW1-DataTransfer-Out-Bytes", foldBucketA, "2026-09-03T10:00:00Z", "2", "0.09", "0.18")+ |
| + curLine("l3", "INV-1", "EUW1-DataTransfer-Out-Bytes", foldBucketB, "2026-09-04T00:00:00Z", "5", "0.09", "0.45")) |
| + now := time.Date(2026, 10, 5, 6, 0, 0, 0, time.UTC) |
| + if err := fx.srv.statementsPass(ctx, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + a := septemberStatement(t, fx, foldAcctA) |
| + if got := a.Meters[usage.MeterStoreEgressGB]; !near(got, 2) { |
| + t.Errorf("owned transfer = %v GB, want 2", got) |
| + } |
| + if a.SourceDays[usage.SourceStoreEgress] != 0 || a.MeterComplete(usage.MeterStoreEgressGB) { |
| + t.Errorf("an unowned charged bucket must hold the month: %+v", a.SourceDays) |
| + } |
| + if a.Closed { |
| + t.Fatal("the month closed over an ownership hole; it could never be billed") |
| + } |
| + res := fx.srv.foldStoreEgress(ctx, foldSept, map[string]map[string]float64{}) |
| + if res.Reason != storeEgressUnowned || len(res.Unowned) != 1 || res.Unowned[0] != foldBucketB { |
| + t.Fatalf("hold reason %q unowned %v", res.Reason, res.Unowned) |
| + } |
| + // The ledger is repaired; the next pass completes and closes the month. |
| + openOwner(t, fx, foldBucketB, foldAcctB, "blog") |
| + if err := fx.srv.statementsPass(ctx, now.Add(time.Hour)); err != nil { |
| + t.Fatal(err) |
| + } |
| + a = septemberStatement(t, fx, foldAcctA) |
| + if !a.Closed || !a.MeterComplete(usage.MeterStoreEgressGB) { |
| + t.Fatalf("repaired month: closed=%v days=%d", a.Closed, a.SourceDays[usage.SourceStoreEgress]) |
| + } |
| + if got := septemberStatement(t, fx, foldAcctB).Meters[usage.MeterStoreEgressGB]; !near(got, 5) { |
| + t.Errorf("account B after repair = %v GB, want 5", got) |
| + } |
| +} |
| + |
| +// Before AWS invoices the month the numbers are estimates: folded, never |
| +// complete, and the month waits rather than closing unbillable. |
| +func TestStoreEgressFoldWaitsForTheInvoice(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + fx.srv.cfg.StoreEgressReportPrefix = foldReportPrefix |
| + openOwner(t, fx, foldBucketA, foldAcctA, "shop") |
| + putCURDelivery(t, fx, "exec-1", curLine("l1", "", "EUW1-DataTransfer-Out-Bytes", foldBucketA, "2026-09-03T10:00:00Z", "2", "0.09", "0.18")) |
| + if err := fx.srv.statementsPass(ctx, time.Date(2026, 10, 5, 6, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + a := septemberStatement(t, fx, foldAcctA) |
| + if !near(a.Meters[usage.MeterStoreEgressGB], 2) || a.MeterComplete(usage.MeterStoreEgressGB) || a.Closed { |
| + t.Fatalf("uninvoiced month: meters=%v days=%d closed=%v", a.Meters, a.SourceDays[usage.SourceStoreEgress], a.Closed) |
| + } |
| + if res := fx.srv.foldStoreEgress(ctx, foldSept, map[string]map[string]float64{}); res.Reason != storeEgressNotFinal || !res.HoldClose { |
| + t.Fatalf("reason %q hold %v", res.Reason, res.HoldClose) |
| + } |
| +} |
| + |
| +// A charged transfer type the classifier does not know is a hold, never a |
| +// silent zero and never a guess. |
| +func TestStoreEgressFoldHoldsOnUnclassifiedChargedTransfer(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + fx.srv.cfg.StoreEgressReportPrefix = foldReportPrefix |
| + openOwner(t, fx, foldBucketA, foldAcctA, "shop") |
| + putCURDelivery(t, fx, "exec-1", curLine("l1", "INV-1", "EUW1-S3RTC-Out-Bytes", foldBucketA, "2026-09-03T10:00:00Z", "2", "0.04", "0.08")) |
| + meters := map[string]map[string]float64{} |
| + res := fx.srv.foldStoreEgress(context.Background(), foldSept, meters) |
| + if res.Complete || res.Reason != storeEgressUnclassified || res.Holds[awsbilling.ReasonUnknownTransfer] != 1 { |
| + t.Fatalf("result %+v", res) |
| + } |
| + if len(meters) != 0 { |
| + t.Fatalf("an unclassified line was billed: %v", meters) |
| + } |
| +} |
| + |
| +// No report configured is today's behaviour: the meter never completes and |
| +// the close is not held hostage. A configured report with no delivery |
| +// holds the close. |
| +func TestStoreEgressFoldWithoutAReport(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + now := time.Date(2026, 10, 5, 6, 0, 0, 0, time.UTC) |
| + if res := fx.srv.foldStoreEgress(ctx, foldSept, nil); res.Complete || res.HoldClose || res.Reason != storeEgressUnconfigured { |
| + t.Fatalf("unconfigured: %+v", res) |
| + } |
| + if err := fx.srv.statementsPass(ctx, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if st := septemberStatement(t, fx, fixtureAcct); !st.Closed || st.MeterComplete(usage.MeterStoreEgressGB) { |
| + t.Fatalf("unconfigured month: closed=%v complete=%v", st.Closed, st.MeterComplete(usage.MeterStoreEgressGB)) |
| + } |
| + |
| + fx2 := newConsoleFixture(t) |
| + fx2.srv.cfg.StoreEgressReportPrefix = foldReportPrefix |
| + if res := fx2.srv.foldStoreEgress(ctx, foldSept, nil); res.Complete || !res.HoldClose || res.Reason != storeEgressNoDelivery { |
| + t.Fatalf("no delivery: %+v", res) |
| + } |
| + if err := fx2.srv.statementsPass(ctx, now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if st := septemberStatement(t, fx2, fixtureAcct); st.Closed { |
| + t.Fatal("a configured report with no delivery must hold the close") |
| + } |
| +} |
| + |
| +func TestBucketOwnershipPrefixAgreesAcrossPackages(t *testing.T) { |
| + if usage.BucketOwnershipPrefix != objstore.BucketOwnershipPrefix { |
| + t.Fatalf("usage pins %q, objstore writes %q", usage.BucketOwnershipPrefix, objstore.BucketOwnershipPrefix) |
| + } |
| +} |
| + |
| +// ---- durability_writes_1k --------------------------------------------------- |
| + |
| +type foldClock struct{ t time.Time } |
| + |
| +func (c *foldClock) now() time.Time { return c.t } |
| + |
| +// publishDurabilityDay runs a real ledger on box for 2026-09-17, records |
| +// attempts through it and publishes the closed day into the fixture bucket. |
| +func publishDurabilityDay(t *testing.T, fx *consoleFixture, box string, attested bool, record func(bind func(usage.ReplicationAttribution) *usage.DurabilityClient, at time.Time)) { |
| + t.Helper() |
| + ctx := context.Background() |
| + day := time.Date(2026, 9, 17, 0, 0, 0, 0, time.UTC) |
| + clock := &foldClock{t: day.Add(-time.Hour)} |
| + l, err := usage.OpenDurabilityLedger(t.TempDir(), box, clock.now, usage.DurabilityLedgerOptions{AllClientsInstrumented: attested}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer l.Close(ctx, nil) //nolint:errcheck |
| + clock.t = day.Add(10 * time.Hour) |
| + record(func(attr usage.ReplicationAttribution) *usage.DurabilityClient { |
| + c, err := l.Bind(attr) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return c |
| + }, clock.t) |
| + clock.t = day.Add(24*time.Hour + time.Minute) |
| + if err := l.Seal(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := l.Publish(ctx, fx.rel); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func recordAttempt(t *testing.T, c *usage.DurabilityClient, op string, status int, at time.Time) { |
| + t.Helper() |
| + tok, err := c.Start(op, at) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := c.Finish(tok, usage.RequestOutcome{At: at, Status: status, Answered: true}); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func tenantWrites(t *testing.T) func(func(usage.ReplicationAttribution) *usage.DurabilityClient, time.Time) { |
| + return func(bind func(usage.ReplicationAttribution) *usage.DurabilityClient, at time.Time) { |
| + rep := bind(usage.ReplicationAttribution{Account: foldAcctA, App: "shop", Host: "shop.bcd.example", Purpose: usage.PurposeReplicate}) |
| + recordAttempt(t, rep, "PutObject", 200, at) |
| + recordAttempt(t, rep, "PutObject", 200, at) |
| + recordAttempt(t, rep, "PutObject", 500, at) // a failed retry: not billed |
| + recordAttempt(t, rep, "GetObject", 200, at) // a read: not billed |
| + lease := bind(usage.ReplicationAttribution{Account: foldAcctA, App: "shop", Host: "shop.bcd.example", Purpose: usage.PurposeLease}) |
| + recordAttempt(t, lease, "PutObject", 200, at) // housekeeping: not billed |
| + } |
| +} |
| + |
| +// An attested, closed box-day folds its billable writes and counts as a |
| +// source day; the statement carries the meter. |
| +func TestDurabilityFoldCountsAnAttestedDay(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + publishDurabilityDay(t, fx, "box1", true, tenantWrites(t)) |
| + if err := fx.srv.statementsPass(context.Background(), time.Date(2026, 9, 20, 6, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + st := septemberStatement(t, fx, foldAcctA) |
| + if got := st.Meters[usage.MeterDurabilityWrites1k]; !near(got, 0.002) { |
| + t.Errorf("durability_writes_1k = %v, want 0.002 (two answered tenant writes)", got) |
| + } |
| + if got := st.SourceDays[usage.SourceDurability]; got != 1 { |
| + t.Errorf("durability source days = %d, want 1", got) |
| + } |
| + if st.MeterComplete(usage.MeterDurabilityWrites1k) { |
| + t.Error("one attested day is not a complete month") |
| + } |
| +} |
| + |
| +// What production publishes today: the ledger is not attested. The writes |
| +// still fold as a lower bound and the day never counts. |
| +func TestDurabilityFoldNeverCountsAnUnattestedDay(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + publishDurabilityDay(t, fx, "box1", false, tenantWrites(t)) |
| + if err := fx.srv.statementsPass(context.Background(), time.Date(2026, 9, 20, 6, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + st := septemberStatement(t, fx, foldAcctA) |
| + if got := st.Meters[usage.MeterDurabilityWrites1k]; !near(got, 0.002) { |
| + t.Errorf("lower bound = %v, want 0.002", got) |
| + } |
| + if got := st.SourceDays[usage.SourceDurability]; got != 0 { |
| + t.Errorf("an unattested day counted: %d", got) |
| + } |
| +} |
| + |
| +// One attested box must never vouch for a box that published nothing. |
| +func TestDurabilityFoldNeedsEveryBox(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + publishDurabilityDay(t, fx, "box1", true, tenantWrites(t)) |
| + // box2 has a ledger directory (it published another day) and is silent |
| + // on the 17th. |
| + putStoreJSON(t, fx.rel, usage.DurabilityDayKey("box2", "2026-09-10"), usage.DurabilityDay{V: 1, Box: "box2", Date: "2026-09-10"}) |
| + if err := fx.srv.statementsPass(context.Background(), time.Date(2026, 9, 20, 6, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := septemberStatement(t, fx, foldAcctA).SourceDays[usage.SourceDurability]; got != 0 { |
| + t.Errorf("durability source days = %d, want 0: box2 was silent", got) |
| + } |
| +} |
diff --git a/internal/console/server.go b/internal/console/server.go| index 874b561ea..4f9f99f0d 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -516,6 +516,17 @@ type Config struct { |
| // operator's store-mode `carlos instances enable`. |
| AppsDomain string |
| AppsDomainAliasAccounts []string |
| + // StoreEgressReportPrefix is the prefix, in the deployment bucket, that |
| + // the AWS Cost and Usage Report (CUR 2.0 Data Export) for this |
| + // deployment's account is delivered under, ending in "/". It is the |
| + // billing authority for object-store transfer (storeegressfold.go). |
| + // Empty means no report: the store-transfer meter is never complete and |
| + // the monthly freeze holds, exactly as before the collector existed. |
| + StoreEgressReportPrefix string |
| + // StoreEgressPlatformBuckets names buckets whose charged transfer is |
| + // the platform's own cost and is billed to no tenant. The deployment |
| + // bucket (Bucket) is always one of them. |
| + StoreEgressPlatformBuckets []string |
| // RedirectHosts are request hosts this console answers with a permanent |
| // redirect to Origin (CONSOLE_REDIRECT_HOSTS, comma-separated) — retired |
| // hostnames after a canonical-origin move. Opt-in per host: hostnames |
diff --git a/internal/console/storeegressfold.go b/internal/console/storeegressfold.go| new file mode 100644 |
| index 000000000..bbaeaf89c |
| --- /dev/null |
| +++ b/internal/console/storeegressfold.go |
| @@ -0,0 +1,214 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "sort" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/awsbilling" |
| + "github.com/carlosframework/platform/internal/objstore" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +// The store_egress_gb fold (pricing spec P5; ruling: object-store transfer |
| +// counts only transfer AWS charges for, and missing evidence holds). |
| +// |
| +// The billing authority is the AWS Cost and Usage Report for the month |
| +// (internal/awsbilling), read from a prefix of the deployment bucket so |
| +// every object it rests on passes through the month's evidence store like |
| +// every other source. One charged line is one bucket-hour of transfer out |
| +// that AWS priced above zero; the bucket ownership ledger |
| +// (objstore/ownership.go) says whose it was. |
| +// |
| +// The month is complete for billing only when ALL of these hold: |
| +// |
| +// - a report prefix is configured and a delivery exists for the month; |
| +// - AWS has invoiced it (awsbilling.Delivery.Final). Until then the |
| +// numbers are estimates, folded for the modelling views and nothing |
| +// else; |
| +// - no charged line was unclassifiable (awsbilling.IsHold); |
| +// - every charged bucket-hour resolved to exactly one owner, or belongs |
| +// to a bucket this deployment declares its own. |
| +// |
| +// Anything less folds what could be attributed as a lower bound and leaves |
| +// the source at zero days, so the freeze is refused. No byte reaches a |
| +// statement without a charged line and an owner behind it. |
| + |
| +// Hold reasons, as logged and reported. Each names the first condition |
| +// that failed. |
| +const ( |
| + storeEgressUnconfigured = "store_egress_report_unconfigured" |
| + storeEgressNoDelivery = "store_egress_no_delivery" |
| + storeEgressRefused = "store_egress_delivery_refused" |
| + storeEgressNotFinal = "store_egress_not_invoiced" |
| + storeEgressUnclassified = "store_egress_unclassified_charge" |
| + storeEgressUnowned = "store_egress_unowned_bucket" |
| + storeEgressOwnerUnread = "store_egress_ownership_unreadable" |
| +) |
| + |
| +// storeEgressResult is what one month's fold found. |
| +type storeEgressResult struct { |
| + // Complete: the month may count as a whole-month source. |
| + Complete bool |
| + // Reason is the hold reason when Complete is false. |
| + Reason string |
| + // HoldClose: the report is configured but not yet provable, so a |
| + // closing month must stay open and be retried. A closed statement is |
| + // never rewritten, so closing over an incomplete report would make the |
| + // month unbillable for good. An unconfigured deployment never holds a |
| + // close: it has no report to wait for. |
| + HoldClose bool |
| + Delivery *awsbilling.Delivery |
| + // Unowned names the charged buckets no owner could be shown for. |
| + Unowned []string |
| + // Holds counts charged lines the classifier could not count, by reason. |
| + Holds map[string]int |
| + // PlatformBytes is charged transfer from the deployment's own buckets: |
| + // the platform's cost, billed to nobody. |
| + PlatformBytes int64 |
| +} |
| + |
| +// foldStoreEgress reads the month's Cost and Usage Report delivery and |
| +// folds each account's charged object-store transfer into meters as |
| +// usage.MeterStoreEgressGB. period is the first instant of a UTC month. |
| +func (s *Server) foldStoreEgress(ctx context.Context, period time.Time, meters map[string]map[string]float64) storeEgressResult { |
| + prefix := s.cfg.StoreEgressReportPrefix |
| + if prefix == "" { |
| + return storeEgressResult{Reason: storeEgressUnconfigured} |
| + } |
| + type bucketHour struct { |
| + bucket string |
| + hour time.Time |
| + } |
| + charged := map[bucketHour]int64{} |
| + res := storeEgressResult{HoldClose: true, Holds: map[string]int{}} |
| + overflow := false |
| + delivery, err := awsbilling.ReadDelivery(ctx, s.cfg.Releases, prefix, period, func(l awsbilling.Line) error { |
| + bucket, n, ok, reason := awsbilling.ChargedS3Transfer(l) |
| + if !ok { |
| + if awsbilling.IsHold(reason) { |
| + res.Holds[reason]++ |
| + } |
| + return nil |
| + } |
| + k := bucketHour{bucket, l.UsageStart.UTC()} |
| + sum := charged[k] + n |
| + if sum < charged[k] { |
| + overflow = true |
| + } |
| + charged[k] = sum |
| + return nil |
| + }) |
| + if err != nil { |
| + // Everything accumulated is discarded: a refused delivery proves |
| + // nothing (awsbilling.ReadDelivery's contract). |
| + res.Reason = storeEgressRefused |
| + if errors.Is(err, awsbilling.ErrNoDelivery) { |
| + res.Reason = storeEgressNoDelivery |
| + } |
| + log.Printf("console: store egress %s: %s: %v", awsbilling.PeriodLabel(period), res.Reason, err) |
| + return res |
| + } |
| + res.Delivery = delivery |
| + if overflow { |
| + res.Reason = storeEgressRefused |
| + log.Printf("console: store egress %s: byte total overflow; holding", awsbilling.PeriodLabel(period)) |
| + return res |
| + } |
| + |
| + platform := map[string]bool{} |
| + if s.cfg.Bucket != "" { |
| + platform[s.cfg.Bucket] = true |
| + } |
| + for _, b := range s.cfg.StoreEgressPlatformBuckets { |
| + platform[b] = true |
| + } |
| + ledgers := map[string]*objstore.BucketOwnership{} |
| + unowned := map[string]bool{} |
| + ownerUnread := false |
| + perAccount := map[string]int64{} |
| + // Deterministic order, so logs and any overflow are reproducible. |
| + keys := make([]bucketHour, 0, len(charged)) |
| + for k := range charged { |
| + keys = append(keys, k) |
| + } |
| + sort.Slice(keys, func(i, j int) bool { |
| + if keys[i].bucket != keys[j].bucket { |
| + return keys[i].bucket < keys[j].bucket |
| + } |
| + return keys[i].hour.Before(keys[j].hour) |
| + }) |
| + for _, k := range keys { |
| + n := charged[k] |
| + if platform[k.bucket] { |
| + res.PlatformBytes += n |
| + continue |
| + } |
| + ledger, loaded := ledgers[k.bucket] |
| + if !loaded { |
| + var rerr error |
| + ledger, _, rerr = objstore.ReadBucketOwnership(ctx, s.cfg.Releases, k.bucket) |
| + if rerr != nil { |
| + // An unreadable or malformed ledger is not "no owner" to be |
| + // absorbed: it is a hold a person must look at. |
| + log.Printf("console: store egress %s: ownership of %s: %v", awsbilling.PeriodLabel(period), k.bucket, rerr) |
| + ownerUnread = true |
| + ledger = nil |
| + } |
| + ledgers[k.bucket] = ledger |
| + } |
| + acct, _, ok := ledger.OwnerAt(k.hour) |
| + if !ok || !release.ValidAccount(acct) { |
| + unowned[k.bucket] = true |
| + continue |
| + } |
| + perAccount[acct] += n |
| + } |
| + for acct, n := range perAccount { |
| + mm := meters[acct] |
| + if mm == nil { |
| + mm = map[string]float64{} |
| + meters[acct] = mm |
| + } |
| + mm[usage.MeterStoreEgressGB] += float64(n) / 1e9 |
| + } |
| + for b := range unowned { |
| + res.Unowned = append(res.Unowned, b) |
| + } |
| + sort.Strings(res.Unowned) |
| + |
| + switch { |
| + case ownerUnread: |
| + res.Reason = storeEgressOwnerUnread |
| + case len(res.Unowned) > 0: |
| + res.Reason = storeEgressUnowned |
| + case len(res.Holds) > 0: |
| + res.Reason = storeEgressUnclassified |
| + case !delivery.Final: |
| + res.Reason = storeEgressNotFinal |
| + default: |
| + res.Complete, res.HoldClose = true, false |
| + return res |
| + } |
| + log.Printf("console: store egress %s: %s", awsbilling.PeriodLabel(period), res.describe()) |
| + return res |
| +} |
| + |
| +// describe is the one-line diagnostic for a held month. |
| +func (r storeEgressResult) describe() string { |
| + out := r.Reason |
| + if len(r.Unowned) > 0 { |
| + out += fmt.Sprintf(": charged transfer from %d bucket(s) with no owner on record: %v", len(r.Unowned), r.Unowned) |
| + } |
| + if len(r.Holds) > 0 { |
| + out += fmt.Sprintf(": charged lines not counted: %v", r.Holds) |
| + } |
| + if r.Delivery != nil && !r.Delivery.Final { |
| + out += ": delivery " + r.Delivery.AssemblyID + " " + r.Delivery.NotFinal |
| + } |
| + return out |
| +} |
diff --git a/internal/usage/billable_test.go b/internal/usage/billable_test.go| new file mode 100644 |
| index 000000000..0cb14cabf |
| --- /dev/null |
| +++ b/internal/usage/billable_test.go |
| @@ -0,0 +1,106 @@ |
| +package usage |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +func TestBillableReplicationWritesPolicy(t *testing.T) { |
| + row := func(account, purpose, op string, answered map[string]int64, unanswered, unknown int64) DurabilityRow { |
| + return DurabilityRow{Host: "shop.acme.example", Account: account, App: "shop", Purpose: purpose, Operation: op, Answered: answered, Unanswered: unanswered, Unknown: unknown} |
| + } |
| + cases := []struct { |
| + name string |
| + row DurabilityRow |
| + want int64 |
| + }{ |
| + {"tenant replicate put, 2xx only", row("acme", "replicate", "PutObject", map[string]int64{"200": 3, "500": 2, "403": 1}, 4, 1), 3}, |
| + {"checkpoint multipart part", row("acme", "checkpoint", "UploadPart", map[string]int64{"200": 5}, 0, 0), 5}, |
| + {"copy counts as a write", row("acme", "replicate", "CopyObject", map[string]int64{"200": 1}, 0, 0), 1}, |
| + {"reads never count", row("acme", "replicate", "GetObject", map[string]int64{"200": 9}, 0, 0), 0}, |
| + {"lists never count", row("acme", "replicate", "ListObjectsV2", map[string]int64{"200": 9}, 0, 0), 0}, |
| + {"deletes never count", row("acme", "replicate", "DeleteObject", map[string]int64{"204": 9}, 0, 0), 0}, |
| + {"lease housekeeping is the platform's", row("acme", "lease", "PutObject", map[string]int64{"200": 9}, 0, 0), 0}, |
| + {"a move is the platform's", row("acme", "move", "CopyObject", map[string]int64{"200": 9}, 0, 0), 0}, |
| + {"restore is the platform's", row("acme", "restore", "PutObject", map[string]int64{"200": 9}, 0, 0), 0}, |
| + {"the platform account owes nothing", row(release.OpsAccount, "replicate", "PutObject", map[string]int64{"200": 9}, 0, 0), 0}, |
| + {"unanswered and unknown are not billed", row("acme", "replicate", "PutObject", nil, 7, 2), 0}, |
| + } |
| + for _, c := range cases { |
| + if got := BillableReplicationWrites(c.row); got != c.want { |
| + t.Errorf("%s: got %d, want %d", c.name, got, c.want) |
| + } |
| + } |
| +} |
| + |
| +// A bucket ownership ledger grows after the month it is evidence for. What |
| +// it proved about the month must not change when an interval is appended |
| +// or closed outside the month, and must change when one is closed inside. |
| +func TestBucketOwnershipEvidenceIsPinnedByItsMonthProjection(t *testing.T) { |
| + ctx := context.Background() |
| + raw := release.NewDirStore(t.TempDir()) |
| + key := BucketOwnershipPrefix + "carlos-store-acme-shop.json" |
| + put := func(body map[string]any) { |
| + t.Helper() |
| + b, err := json.Marshal(body) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + _, etag, err := raw.Get(ctx, key) |
| + if err != nil { |
| + etag = "" |
| + } |
| + if _, err := raw.Put(ctx, key, b, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + interval := func(acct string, from time.Time, until *time.Time) map[string]any { |
| + iv := map[string]any{"account_sqid": acct, "app": "shop", "from": from} |
| + if until != nil { |
| + iv["until"] = *until |
| + } |
| + return iv |
| + } |
| + aug := time.Date(2026, 8, 10, 0, 0, 0, 0, time.UTC) |
| + put(map[string]any{"v": 1, "bucket": "carlos-store-acme-shop", "intervals": []any{interval("acme", aug, nil)}}) |
| + |
| + es, err := NewMonthEvidenceStore(raw, "2026-09") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, _, err := es.Get(ctx, key); err != nil { |
| + t.Fatal(err) |
| + } |
| + ev, err := es.Snapshot() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(ev.Objects) != 1 || ev.Objects[0].Projection != "bucket_ownership_month" || ev.Objects[0].ETag != "" { |
| + t.Fatalf("evidence %+v", ev.Objects) |
| + } |
| + if err := ev.Verify(ctx, raw); err != nil { |
| + t.Fatalf("unchanged ledger: %v", err) |
| + } |
| + // Closed in October, and a new owner from then: September is untouched. |
| + oct := time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC) |
| + put(map[string]any{"v": 1, "bucket": "carlos-store-acme-shop", "intervals": []any{interval("acme", aug, &oct), interval("beta", oct, nil)}}) |
| + if err := ev.Verify(ctx, raw); err != nil { |
| + t.Fatalf("a change after the month must not read as changed evidence: %v", err) |
| + } |
| + // Closed inside September: what the month was billed on has changed. |
| + sep := time.Date(2026, 9, 20, 0, 0, 0, 0, time.UTC) |
| + put(map[string]any{"v": 1, "bucket": "carlos-store-acme-shop", "intervals": []any{interval("acme", aug, &sep), interval("beta", oct, nil)}}) |
| + if err := ev.Verify(ctx, raw); !errors.Is(err, ErrEvidenceChanged) { |
| + t.Fatalf("a change inside the month must read as changed evidence, got %v", err) |
| + } |
| + // A record that names another bucket is refused, not projected. |
| + put(map[string]any{"v": 1, "bucket": "carlos-store-other", "intervals": []any{interval("acme", aug, nil)}}) |
| + if err := ev.Verify(ctx, raw); !errors.Is(err, ErrEvidenceChanged) { |
| + t.Fatalf("a ledger naming another bucket must be refused, got %v", err) |
| + } |
| +} |
diff --git a/internal/usage/durability.go b/internal/usage/durability.go| index d7e9d14df..2baea67a1 100644 |
| --- a/internal/usage/durability.go |
| +++ b/internal/usage/durability.go |
| @@ -213,6 +213,47 @@ type DurabilityRow struct { |
| Unknown int64 `json:"unknown,omitempty"` |
| } |
| +// BillableReplicationWrites is the pricing policy for one ledger row: how |
| +// many of its attempts count toward durability_writes_1k. The ledger keeps |
| +// operation, purpose and outcome apart so this can be decided here, once, |
| +// without recollecting. |
| +// |
| +// A write counts when all three hold: |
| +// |
| +// - the purpose is the tenant's own replication: replicate or checkpoint. |
| +// Lease housekeeping, restores, moves, the verify drill and the |
| +// platform's own buckets are the platform's cost, never the tenant's; |
| +// - the operation is one S3 prices as a write: PutObject, UploadPart, |
| +// CreateMultipartUpload, CompleteMultipartUpload, CopyObject. Reads, |
| +// lists, heads and deletes never count; |
| +// - the server answered it with a 2xx. A retry that failed, an attempt |
| +// that got no reply and an attempt whose outcome is unknown are not |
| +// billed: every ambiguity resolves in the customer's favour. |
| +// |
| +// The platform account never owes anything, whatever the row says. |
| +func BillableReplicationWrites(r DurabilityRow) int64 { |
| + if r.Account == "" || r.Account == release.OpsAccount { |
| + return 0 |
| + } |
| + switch ReplicationPurpose(r.Purpose) { |
| + case PurposeReplicate, PurposeCheckpoint: |
| + default: |
| + return 0 |
| + } |
| + switch r.Operation { |
| + case "PutObject", "UploadPart", "CreateMultipartUpload", "CompleteMultipartUpload", "CopyObject": |
| + default: |
| + return 0 |
| + } |
| + var n int64 |
| + for status, count := range r.Answered { |
| + if len(status) == 3 && status[0] == '2' && count > 0 { |
| + n += count |
| + } |
| + } |
| + return n |
| +} |
| + |
| // DurabilityTotals is the sum over every row of a snapshot. |
| type DurabilityTotals struct { |
| Attempts int64 `json:"attempts"` |
diff --git a/internal/usage/durabilityledger_test.go b/internal/usage/durabilityledger_test.go| index 58f8f5fbc..120972d03 100644 |
| --- a/internal/usage/durabilityledger_test.go |
| +++ b/internal/usage/durabilityledger_test.go |
| @@ -693,8 +693,11 @@ func TestDurabilityLifecycleSourceIsAdditive(t *testing.T) { |
| if _, err := RegisterBoxSource(ctx, s, "box1", SourceStorage, ledgerDay0); err == nil { |
| t.Fatal("storage is still not a per-box source") |
| } |
| - if got := MeterSource(MeterDurabilityWrites1k); got != "" { |
| - t.Fatalf("MeterSource(durability_writes_1k) = %q: the meter must stay unbillable until the collector is wired", got) |
| + // The console fold exists now (console/durabilityfold.go), so the meter |
| + // names its source. It is still unbillable in production: a day counts |
| + // only when its box attests box-wide, and no production open does. |
| + if got := MeterSource(MeterDurabilityWrites1k); got != SourceDurability { |
| + t.Fatalf("MeterSource(durability_writes_1k) = %q, want %q", got, SourceDurability) |
| } |
| } |
diff --git a/internal/usage/evidence.go b/internal/usage/evidence.go| index a2224886f..ae747a51c 100644 |
| --- a/internal/usage/evidence.go |
| +++ b/internal/usage/evidence.go |
| @@ -77,14 +77,20 @@ func (s *EvidenceStore) Get(ctx context.Context, key string) ([]byte, string, er |
| x.ETag = "" |
| } else if err == nil { |
| x.SHA256 = evidenceDigest(data) |
| - if s.month != "" && isLifecycleEvidenceKey(key) { |
| - projected, projectionErr := projectLifecycleMonth(key, data, s.month) |
| - if projectionErr != nil { |
| - err = projectionErr |
| - } else { |
| - x.SHA256 = evidenceDigest(projected) |
| - x.ETag = "" |
| - x.Projection = "lifecycle_month" |
| + if s.month != "" { |
| + // Records that legitimately grow after the month (a box's |
| + // lifecycle, a bucket's ownership ledger) are pinned by their |
| + // projection onto the month, so a later append outside it |
| + // does not read as changed evidence. |
| + if projection := monthProjection(key); projection != "" { |
| + projected, projectionErr := projectMonth(projection, key, data, s.month) |
| + if projectionErr != nil { |
| + err = projectionErr |
| + } else { |
| + x.SHA256 = evidenceDigest(projected) |
| + x.ETag = "" |
| + x.Projection = projection |
| + } |
| } |
| } |
| } |
| @@ -184,10 +190,10 @@ func (e InputEvidence) Verify(ctx context.Context, s release.Store) error { |
| return err |
| } |
| if x.Projection != "" { |
| - if x.Projection != "lifecycle_month" || x.Missing || x.ETag != "" { |
| + if x.Projection != monthProjection(x.Key) || x.Missing || x.ETag != "" { |
| return ErrEvidenceChanged |
| } |
| - data, err = projectLifecycleMonth(x.Key, data, e.Month) |
| + data, err = projectMonth(x.Projection, x.Key, data, e.Month) |
| if err != nil { |
| return err |
| } |
diff --git a/internal/usage/evidence_month.go b/internal/usage/evidence_month.go| index fb96b86d0..151f9b88f 100644 |
| --- a/internal/usage/evidence_month.go |
| +++ b/internal/usage/evidence_month.go |
| @@ -13,6 +13,98 @@ func isLifecycleEvidenceKey(key string) bool { |
| return strings.HasPrefix(key, LifecyclePrefix) && strings.HasSuffix(key, ".json") |
| } |
| +// Month projections. A record pinned this way is evidence only for what it |
| +// says ABOUT the month, so the month's proof survives the record growing. |
| +const ( |
| + projectionLifecycleMonth = "lifecycle_month" |
| + projectionBucketOwnershipMonth = "bucket_ownership_month" |
| +) |
| + |
| +// BucketOwnershipPrefix is where objstore keeps its bucket ownership |
| +// ledgers (objstore.BucketOwnershipPrefix; repeated here because usage |
| +// cannot import objstore, and pinned equal by a test in the console). |
| +const BucketOwnershipPrefix = "usage/buckets/" |
| + |
| +func isBucketOwnershipEvidenceKey(key string) bool { |
| + return strings.HasPrefix(key, BucketOwnershipPrefix) && strings.HasSuffix(key, ".json") |
| +} |
| + |
| +// monthProjection names the projection a key is pinned by, or "". |
| +func monthProjection(key string) string { |
| + switch { |
| + case isLifecycleEvidenceKey(key): |
| + return projectionLifecycleMonth |
| + case isBucketOwnershipEvidenceKey(key): |
| + return projectionBucketOwnershipMonth |
| + } |
| + return "" |
| +} |
| + |
| +func projectMonth(projection, key string, raw []byte, month string) ([]byte, error) { |
| + switch projection { |
| + case projectionLifecycleMonth: |
| + return projectLifecycleMonth(key, raw, month) |
| + case projectionBucketOwnershipMonth: |
| + return projectBucketOwnershipMonth(key, raw, month) |
| + } |
| + return nil, ErrEvidenceChanged |
| +} |
| + |
| +// projectBucketOwnershipMonth reduces a bucket's ownership ledger to the |
| +// intervals that touch the month, clipped to it. The ledger is append-only |
| +// (objstore/ownership.go): a later grant, or an interval closed after the |
| +// month ended, changes the object and must not change what it proved about |
| +// the month. An interval closed INSIDE the month does change the |
| +// projection, as it should. The wire shape is objstore.BucketOwnership's. |
| +func projectBucketOwnershipMonth(key string, raw []byte, month string) ([]byte, error) { |
| + start, err := time.Parse("2006-01", month) |
| + if err != nil || start.Format("2006-01") != month || !isBucketOwnershipEvidenceKey(key) { |
| + return nil, ErrEvidenceChanged |
| + } |
| + bucket := strings.TrimSuffix(strings.TrimPrefix(key, BucketOwnershipPrefix), ".json") |
| + var rec struct { |
| + V int `json:"v"` |
| + Bucket string `json:"bucket"` |
| + Intervals []struct { |
| + Account string `json:"account_sqid"` |
| + App string `json:"app"` |
| + From time.Time `json:"from"` |
| + Until *time.Time `json:"until,omitempty"` |
| + } `json:"intervals"` |
| + } |
| + if json.Unmarshal(raw, &rec) != nil || rec.V != 1 || rec.Bucket != bucket || bucket == "" { |
| + return nil, ErrEvidenceChanged |
| + } |
| + end := start.AddDate(0, 1, 0) |
| + type span struct { |
| + Account string `json:"account_sqid"` |
| + App string `json:"app"` |
| + From time.Time `json:"from"` |
| + To time.Time `json:"to"` |
| + } |
| + spans := []span{} |
| + for _, iv := range rec.Intervals { |
| + if iv.From.IsZero() || (iv.Until != nil && iv.Until.Before(iv.From)) { |
| + return nil, ErrEvidenceChanged |
| + } |
| + lo, hi := iv.From.UTC(), end |
| + if lo.Before(start) { |
| + lo = start |
| + } |
| + if iv.Until != nil && iv.Until.Before(hi) { |
| + hi = iv.Until.UTC() |
| + } |
| + if lo.Before(hi) { |
| + spans = append(spans, span{iv.Account, iv.App, lo, hi}) |
| + } |
| + } |
| + return json.Marshal(struct { |
| + Bucket string `json:"bucket"` |
| + Month string `json:"month"` |
| + Spans []span `json:"spans"` |
| + }{bucket, month, spans}) |
| +} |
| + |
| func projectLifecycleMonth(key string, raw []byte, month string) ([]byte, error) { |
| start, err := time.Parse("2006-01", month) |
| if err != nil || start.Format("2006-01") != month || !isLifecycleEvidenceKey(key) { |
| @@ -55,6 +147,8 @@ func verifyMonthDiscoveryChange(ctx context.Context, s release.Store, month stri |
| source = SourceMetrics |
| case previous.Directories && previous.Prefix == EdgeBoxesPrefix: |
| source = SourceEdge |
| + case previous.Directories && previous.Prefix == DurabilityBoxesPrefix: |
| + source = SourceDurability |
| case !previous.Directories && previous.Prefix == LifecyclePrefix: |
| default: |
| return ErrEvidenceChanged |
| @@ -108,6 +202,7 @@ func rejectEarlierBoxEvidence(ctx context.Context, s release.Store, box, source |
| {EdgeDaysPrefix(box), "2006-01-02", SourceEdge}, |
| {"metrics/boxes/" + box + "/days/", "2006-01-02", SourceMetrics}, |
| {"metrics/boxes/" + box + "/raw/", "2006-01-02-15", SourceMetrics}, |
| + {DurabilityDaysPrefix(box), "2006-01-02", SourceDurability}, |
| } { |
| if source != "" && source != scan.source { |
| continue |
diff --git a/internal/usage/passes_test.go b/internal/usage/passes_test.go| index 7ff9044ab..63fd9e073 100644 |
| --- a/internal/usage/passes_test.go |
| +++ b/internal/usage/passes_test.go |
| @@ -113,6 +113,8 @@ func TestMeterSource(t *testing.T) { |
| MeterBackhaulRequests1k: SourceEdge, |
| MeterStorageGBMonth: SourceStorage, |
| MeterAppMonths: SourceCensus, |
| + MeterDurabilityWrites1k: SourceDurability, |
| + MeterStoreEgressGB: SourceStoreEgress, |
| "not_a_meter": "", |
| } |
| for meter, want := range cases { |
diff --git a/internal/usage/usage.go b/internal/usage/usage.go| index 06c041d7f..5e5b4d563 100644 |
| --- a/internal/usage/usage.go |
| +++ b/internal/usage/usage.go |
| @@ -67,10 +67,18 @@ const ( |
| // box's replication-request ledger (durability.go, pricing spec P6). |
| // Registered on the lifecycle record like metrics and edge, so a box |
| // that has a ledger is expected to publish it every day it exists. |
| - // MeterSource does NOT name it for MeterDurabilityWrites1k on purpose: |
| - // nothing wires a replication client to the ledger yet, and a meter |
| - // whose collector is not wired must stay unbillable. |
| + // MeterDurabilityWrites1k derives from it (console/durabilityfold.go): |
| + // a day counts only when every expected box published a day its ledger |
| + // attests box-wide, which no production box can do while an |
| + // uninstrumented replicator runs beside it. |
| SourceDurability = "durability" |
| + // SourceStoreEgress is the AWS Cost and Usage Report for the month |
| + // (internal/awsbilling, console/storeegressfold.go): object-store |
| + // transfer counts only lines AWS charged for, on a delivery AWS has |
| + // invoiced, with every charged bucket-hour resolved to an owner by the |
| + // bucket ownership ledger. It is a whole-month source: its day count is |
| + // either the month's length or zero. |
| + SourceStoreEgress = "store_egress" |
| ) |
| // MeterSource names the collection source a meter derives from, or "" for a |
| @@ -106,6 +114,10 @@ func MeterSource(meter string) string { |
| return SourceStorage |
| case MeterAppMonths: |
| return SourceCensus |
| + case MeterDurabilityWrites1k: |
| + return SourceDurability |
| + case MeterStoreEgressGB: |
| + return SourceStoreEgress |
| } |
| return "" |
| } |