amadan

carlos / platform Public

File every unmetered replication fallback as billing incompleteness

When the host could not bind a durability recorder (invalid owner, bind
table full, no bind slot, bind budget exceeded, ledger refused the bind,
publisher stopped) it set bindingFailed, which nothing in production
reads, and the caller fell back to an unmetered S3 client: lease.Acquire
in the activator, restoreReplica, ReplicateWithLease in the move
checkpoint, and lease.NewS3Mirror. Only the permanently false attestation
kept those days from reading complete; the day the attestation flips,
an escaped write would have billed a lower bound as the whole.

usage.DurabilityLedger.Escaped(account, at) makes the escape evidence:

- a recording failure latched under at's date (the Incomplete path, no
  I/O, no wait on the ledger mutex), so that box-day never reads
  complete for anyone;
- from at until the next open, the account joins the run's monotonic
  exclusion set, or, when the owner cannot be named, the run stops
  attesting for anyone and a later SetAttestation cannot undo it. The
  unmetered client dies with the process, so a new open starts over.

The session change is applied by a single background applier, and in
any case by the next Seal before it closes a day and by SetAttestation
before it starts a new claim. An escape after Close latches late and
voids the clean stop through the dirty marker, as Incomplete does.

host.durabilityPublisher.recorder now files the escape on every nil it
returns except a nil publisher (no ledger opened; the open failure
already wrote the dirty marker). Every fallback site runs only on such
a nil, so the four fallbacks are covered at the one place they start;
comments at each site say so. Attestation stays false everywhere.

Tests: usage escape tests (per-account scope the day after, bystander
account stays complete, unknown owner withdraws the run, escape while
unattested carries into a later claim, new run resets) and host tests
that open an attested ledger (test only) and show each bind failure
publishes an incomplete day and withholds the scope the next day, plus
a control day that publishes complete. The host tests fail with the
Escaped call removed.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 9a3df4fdb5762b45cffb97f49b105b94a403c3f0 parent 72a5189
6 files changed, +556 −11
  • internal/activator/backend_exec.go +5 −0
  • internal/host/durability.go +35 −7
  • internal/host/durabilityescape_test.go +206 −0
  • internal/host/movespass.go +2 −0
  • internal/usage/durabilityescape_test.go +148 −0
  • internal/usage/durabilityledger.go +160 −4
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 87bc40e45..68b5657c7 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -240,6 +240,9 @@ func (b *ExecBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, er
if meters.Lease != nil && meters.Replica != nil {
held, err = acquireMeteredLease(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region, meters)
} else {
+ // Unmetered fallback. A RequestMeter that returned nil has already
+ // filed the escape with the durability ledger, so this route's
+ // replication days read incomplete rather than short.
held, err = lease.Acquire(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
}
if err != nil {
@@ -328,6 +331,8 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err
if recorder != nil {
restored, err = restoreMeteredReplica(ctx, dbPath, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region, recorder)
} else {
+ // Unmetered fallback: as in Acquire, a nil RequestMeter answer has
+ // already been filed with the ledger as an escape.
restored, err = restoreReplica(ctx, dbPath, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region)
}
if err != nil {
diff --git a/internal/host/durability.go b/internal/host/durability.go
index 2193cb252..8d3ca3657 100644
--- a/internal/host/durability.go
+++ b/internal/host/durability.go
@@ -32,6 +32,7 @@ type durabilityBinding struct {
type durabilityPublisher struct {
ledger *usage.DurabilityLedger
box string
+ now func() time.Time
ctx context.Context
cancel context.CancelFunc
sem chan struct{}
@@ -50,19 +51,49 @@ func newDurabilityPublisher(dd, box string, now func() time.Time) (*durabilityPu
return nil, errors.Join(err, usage.MarkDurabilityLedgerDirty(dir))
}
ctx, cancel := context.WithCancel(context.Background())
- return &durabilityPublisher{ledger: ledger, box: box, ctx: ctx, cancel: cancel, sem: make(chan struct{}, 1), bindings: make(map[usage.ReplicationAttribution]*durabilityBinding), bindSlots: make(chan struct{}, 4)}, nil
+ return &durabilityPublisher{ledger: ledger, box: box, now: now, ctx: ctx, cancel: cancel, sem: make(chan struct{}, 1), bindings: make(map[usage.ReplicationAttribution]*durabilityBinding), bindSlots: make(chan struct{}, 4)}, nil
}
+// recorder binds the route's replication client for purpose to the
+// ledger, or returns nil. Every caller falls back to an UNMETERED client
+// on nil (the activator's Acquire and Restore, the move checkpoint, the
+// move mirror), so a nil here is not silence: recorder files the escape
+// with the ledger (usage.DurabilityLedger.Escaped) before returning it,
+// and the box-day, and the owner's later days in this run, can no longer
+// read complete. The one nil that files nothing is a nil publisher: no
+// ledger opened, and the open's failure already left the dirty marker the
+// next open turns into a gap.
func (p *durabilityPublisher) recorder(rt registry.Route, purpose usage.ReplicationPurpose) lease.RequestRecorder {
- if p == nil || p.stopped.Load() {
+ if p == nil {
return nil
}
if rt.Account == release.OpsAccount {
purpose = usage.PurposePlatform
}
attr := usage.ReplicationAttribution{Account: rt.Account, App: rt.App, Host: rt.Host, Purpose: purpose}
+ rec := p.bind(attr)
+ if rec == nil {
+ p.escaped(attr)
+ }
+ return rec
+}
+
+// escaped records that a client for attr runs unmetered. The account is
+// named when it is a valid sqid, so only its scope is withheld after the
+// day of the escape; an owner that cannot be named withdraws the run's
+// attestation for everyone (the ledger decides which).
+func (p *durabilityPublisher) escaped(attr usage.ReplicationAttribution) {
+ p.bindingFailed.Store(true)
+ p.ledger.Escaped(attr.Account, p.now())
+}
+
+// bind is the lookup-or-bind step, bounded by durabilityBindBudget. Nil
+// means the caller has no recorder; recorder files that as an escape.
+func (p *durabilityPublisher) bind(attr usage.ReplicationAttribution) lease.RequestRecorder {
+ if p.stopped.Load() {
+ return nil
+ }
if err := attr.Validate(); err != nil {
- p.bindingFailed.Store(true)
return nil
}
p.mu.Lock()
@@ -70,14 +101,12 @@ func (p *durabilityPublisher) recorder(rt registry.Route, purpose usage.Replicat
if entry == nil {
if len(p.bindings) >= durabilityBindLimit {
p.mu.Unlock()
- p.bindingFailed.Store(true)
return nil
}
select {
case p.bindSlots <- struct{}{}:
default:
p.mu.Unlock()
- p.bindingFailed.Store(true)
return nil
}
entry = &durabilityBinding{ready: make(chan struct{})}
@@ -86,7 +115,6 @@ func (p *durabilityPublisher) recorder(rt registry.Route, purpose usage.Replicat
defer func() { <-p.bindSlots; close(entry.ready) }()
client, err := p.ledger.Bind(attr)
if err != nil {
- p.bindingFailed.Store(true)
return
}
entry.recorder = &durabilityRecorder{client}
@@ -99,7 +127,6 @@ func (p *durabilityPublisher) recorder(rt registry.Route, purpose usage.Replicat
case <-entry.ready:
return entry.recorder
case <-timer.C:
- p.bindingFailed.Store(true)
return nil
case <-p.ctx.Done():
return nil
@@ -110,6 +137,7 @@ func (p *durabilityPublisher) mirror(ctx context.Context, rt registry.Route, src
if recorder := p.recorder(rt, usage.PurposeMove); recorder != nil {
return lease.NewS3MirrorWithRequestMeter(ctx, srcRegion, dstRegion, recorder)
}
+ // Unmetered fallback: recorder already filed the escape.
return lease.NewS3Mirror(ctx, srcRegion, dstRegion)
}
diff --git a/internal/host/durabilityescape_test.go b/internal/host/durabilityescape_test.go
new file mode 100644
index 000000000..a7f7936ce
--- /dev/null
+++ b/internal/host/durabilityescape_test.go
@@ -0,0 +1,206 @@
+package host
+
+import (
+ "context"
+ "fmt"
+ "os"
+ "path/filepath"
+ "slices"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/usage"
+)
+
+// hostLedgerClock is a movable wall clock for the publisher's ledger.
+type hostLedgerClock struct {
+ mu sync.Mutex
+ t time.Time
+}
+
+func (c *hostLedgerClock) now() time.Time { c.mu.Lock(); defer c.mu.Unlock(); return c.t }
+func (c *hostLedgerClock) set(t time.Time) {
+ c.mu.Lock()
+ c.t = t
+ c.mu.Unlock()
+}
+
+// newAttestedTestPublisher opens a publisher whose ledger claims every
+// client is instrumented. Only a test makes that claim: it is what lets
+// a test tell "incomplete because of the escape" apart from "incomplete
+// because production never attests".
+func newAttestedTestPublisher(t *testing.T, day time.Time) (*durabilityPublisher, *hostLedgerClock, string) {
+ t.Helper()
+ clock := &hostLedgerClock{t: day.Add(-time.Hour)}
+ dd := t.TempDir()
+ p, err := newDurabilityPublisher(dd, "box1", clock.now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := p.ledger.SetAllClientsInstrumented(true, clock.now()); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() {
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if !p.stopped.Load() {
+ if err := p.stop(ctx, nil); err != nil {
+ t.Error(err)
+ }
+ }
+ _ = p.ledger.WaitLate(ctx)
+ })
+ return p, clock, dd
+}
+
+func publishedDurabilityDay(t *testing.T, p *durabilityPublisher, store release.Store, date string) usage.DurabilityDay {
+ t.Helper()
+ if err := p.publish(context.Background(), store); err != nil {
+ t.Fatal(err)
+ }
+ raw, _, err := store.Get(context.Background(), usage.DurabilityDayKey("box1", date))
+ if err != nil {
+ t.Fatal(err)
+ }
+ d, err := usage.ParseDurabilityDay(raw, "box1", date)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return d
+}
+
+// The control: an attested ledger with no escape publishes a complete
+// day, so the failures below are the escape's doing and nothing else.
+func TestDurabilityHostAttestedIdleDayPublishesComplete(t *testing.T) {
+ day := time.Date(2026, 9, 17, 0, 0, 0, 0, time.UTC)
+ p, clock, _ := newAttestedTestPublisher(t, day)
+ store := release.NewDirStore(t.TempDir())
+ clock.set(day.Add(24*time.Hour + time.Second))
+ d := publishedDurabilityDay(t, p, store, usage.DayOf(day))
+ if ok, reason := d.Completeness(); !ok {
+ t.Fatalf("control day incomplete: %s", reason)
+ }
+}
+
+// Each way the host can fail to bind a recorder leaves the published day
+// incomplete: the caller runs an unmetered client, and a day that reads
+// complete would bill a lower bound as the whole.
+func TestDurabilityHostFailedBindPublishesIncompleteDay(t *testing.T) {
+ day := time.Date(2026, 9, 17, 0, 0, 0, 0, time.UTC)
+ rt := registry.Route{Account: "bac", App: "app", Host: "app.example.com"}
+ attr := usage.ReplicationAttribution{Account: rt.Account, App: rt.App, Host: rt.Host, Purpose: usage.PurposeReplicate}
+ cases := map[string]func(t *testing.T, p *durabilityPublisher){
+ "unknown owner": func(t *testing.T, p *durabilityPublisher) {
+ if p.recorder(registry.Route{App: "app", Host: "app.example.com"}, usage.PurposeReplicate) != nil {
+ t.Fatal("unknown owner bound")
+ }
+ },
+ "invalid host": func(t *testing.T, p *durabilityPublisher) {
+ if p.recorder(registry.Route{Account: "bac", App: "app", Host: "not a host"}, usage.PurposeReplicate) != nil {
+ t.Fatal("invalid host bound")
+ }
+ },
+ "binding table full": func(t *testing.T, p *durabilityPublisher) {
+ p.mu.Lock()
+ for i := 0; i < durabilityBindLimit; i++ {
+ p.bindings[usage.ReplicationAttribution{Account: "bac", App: "app", Host: fmt.Sprintf("h%d.example.com", i), Purpose: usage.PurposeReplicate}] = &durabilityBinding{ready: make(chan struct{})}
+ }
+ p.mu.Unlock()
+ if p.recorder(rt, usage.PurposeReplicate) != nil {
+ t.Fatal("full binding table bound")
+ }
+ },
+ "no bind slot": func(t *testing.T, p *durabilityPublisher) {
+ for i := 0; i < cap(p.bindSlots); i++ {
+ p.bindSlots <- struct{}{}
+ }
+ defer func() {
+ for i := 0; i < cap(p.bindSlots); i++ {
+ <-p.bindSlots
+ }
+ }()
+ if p.recorder(rt, usage.PurposeReplicate) != nil {
+ t.Fatal("bound without a slot")
+ }
+ },
+ "bind budget exceeded": func(t *testing.T, p *durabilityPublisher) {
+ p.mu.Lock()
+ p.bindings[attr] = &durabilityBinding{ready: make(chan struct{})} // never ready
+ p.mu.Unlock()
+ if p.recorder(rt, usage.PurposeReplicate) != nil {
+ t.Fatal("bound past the budget")
+ }
+ },
+ "ledger refused the bind": func(t *testing.T, p *durabilityPublisher) {
+ done := &durabilityBinding{ready: make(chan struct{})}
+ close(done.ready) // the bind goroutine finished with no recorder
+ p.mu.Lock()
+ p.bindings[attr] = done
+ p.mu.Unlock()
+ if p.recorder(rt, usage.PurposeReplicate) != nil {
+ t.Fatal("refused bind returned a recorder")
+ }
+ },
+ }
+ for name, fail := range cases {
+ t.Run(name, func(t *testing.T) {
+ p, clock, _ := newAttestedTestPublisher(t, day)
+ store := release.NewDirStore(t.TempDir())
+ clock.set(day.Add(6 * time.Hour))
+ fail(t, p)
+ if !p.bindingFailed.Load() {
+ t.Fatal("failed bind not observed")
+ }
+ clock.set(day.Add(24*time.Hour + time.Second))
+ d := publishedDurabilityDay(t, p, store, usage.DayOf(day))
+ if d.BillingComplete() {
+ t.Fatal("a day on which a client ran unmetered published complete")
+ }
+ if ok, _, _ := d.Scope(day, day.Add(24*time.Hour)); ok {
+ t.Fatal("the escape day vouched for some accounts")
+ }
+ // The day after is still inside the run, so the escaped scope
+ // stays withheld.
+ next := day.Add(24 * time.Hour)
+ clock.set(next.Add(24*time.Hour + time.Second))
+ d2 := publishedDurabilityDay(t, p, store, usage.DayOf(next))
+ if d2.BillingComplete() {
+ t.Fatal("the day after an escape published complete")
+ }
+ if ok, unmetered, _ := d2.Scope(next, next.Add(24*time.Hour)); ok && !slices.Contains(unmetered, "bac") {
+ t.Fatalf("the day after vouched for the escaped account: unmetered=%v", unmetered)
+ }
+ })
+ }
+}
+
+// A recorder asked for after the publisher stopped returns nil, and the
+// caller runs unmetered: that voids the clean stop, so the next open
+// files a gap over it.
+func TestDurabilityHostRecorderAfterStopVoidsCleanStop(t *testing.T) {
+ day := time.Date(2026, 9, 17, 0, 0, 0, 0, time.UTC)
+ p, _, dd := newAttestedTestPublisher(t, day)
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if err := p.stop(ctx, nil); err != nil {
+ t.Fatal(err)
+ }
+ if p.ledger.HandoffToken() == "" {
+ t.Fatal("control: the stop was not clean")
+ }
+ if p.recorder(registry.Route{Account: "bac", App: "app", Host: "app.example.com"}, usage.PurposeReplicate) != nil {
+ t.Fatal("stopped publisher bound a client")
+ }
+ if err := p.ledger.WaitLate(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if !p.bindingFailed.Load() {
+ t.Fatal("unmetered client after stop not observed")
+ }
+ if _, err := os.Stat(filepath.Join(dd, "usage", "durability", "dirty")); err != nil {
+ t.Fatalf("no dirty marker after an unmetered client outlived the stop: %v", err)
+ }
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 90b388dec..2e56d077b 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -108,6 +108,8 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,
if meters.Lease != nil && meters.Replica != nil {
rep, err = lease.ReplicateWithRequestMeters(ctx, dbPath, hib.Bucket, activator.HibernatePrefix(hib.Prefix, rec.Host), hib.Region, meters)
} else {
+ // Unmetered fallback: the recorder that returned nil
+ // already filed the escape with the ledger.
rep, err = lease.ReplicateWithLease(ctx, dbPath, hib.Bucket, activator.HibernatePrefix(hib.Prefix, rec.Host), hib.Region)
}
if err != nil {
diff --git a/internal/usage/durabilityescape_test.go b/internal/usage/durabilityescape_test.go
new file mode 100644
index 000000000..a7c5acb3d
--- /dev/null
+++ b/internal/usage/durabilityescape_test.go
@@ -0,0 +1,148 @@
+package usage
+
+import (
+ "context"
+ "slices"
+ "strings"
+ "testing"
+ "time"
+)
+
+// An unmetered client for one account voids the whole box-day it escaped
+// on, and for every later day of the run it voids that account alone. A
+// new open starts over: the client died with the process.
+func TestDurabilityEscapedAccountIsIncompleteForTheRestOfTheRun(t *testing.T) {
+ dir := t.TempDir()
+ clock := newLedgerClock(ledgerDay0.Add(-time.Hour))
+ l := openTestLedger(t, dir, "box1", clock, true)
+ clock.set(ledgerDay0.Add(6 * time.Hour))
+ l.Escaped(tenantAttr.Account, clock.now())
+
+ day0, day1 := DayOf(ledgerDay0), DayOf(ledgerDay0.Add(24*time.Hour))
+ d1Start, d1End := ledgerDay0.Add(24*time.Hour), ledgerDay0.Add(48*time.Hour)
+
+ clock.set(ledgerDay0.Add(24*time.Hour + time.Second))
+ if err := l.Seal(); err != nil {
+ t.Fatal(err)
+ }
+ c0, err := l.Snapshot(day0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if c0.RecordingFailures == 0 {
+ t.Fatalf("the escape day carries no recording failure: %+v", c0)
+ }
+ if ok, _ := c0.CompletenessFor(tenant2Attr.Account, ledgerDay0, d1Start); ok {
+ t.Fatal("the escape day read complete for another account")
+ }
+
+ clock.set(d1End.Add(time.Second))
+ if err := l.Seal(); err != nil {
+ t.Fatal(err)
+ }
+ c1, err := l.Snapshot(day1)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if ok, reason := c1.CompletenessFor(tenant2Attr.Account, d1Start, d1End); !ok {
+ t.Fatalf("a bystander account lost its complete day: %s (coverage %+v)", reason, c1.Coverage)
+ }
+ if ok, reason := c1.CompletenessFor(tenantAttr.Account, d1Start, d1End); ok || !strings.Contains(reason, tenantAttr.Account) {
+ t.Fatalf("the escaped account read complete the day after: ok=%v reason=%q", ok, reason)
+ }
+ if ok, _ := c1.Completeness(); ok {
+ t.Fatal("box-wide strict completeness ignored the exclusion")
+ }
+ att := l.Attestation()
+ if !att.Attested || !slices.Contains(att.Unmetered, tenantAttr.Account) {
+ t.Fatalf("attestation after escape: %+v", att)
+ }
+
+ ctx := context.Background()
+ if err := l.Close(ctx, nil); err != nil {
+ t.Fatal(err)
+ }
+ token := l.HandoffToken()
+ if token == "" {
+ t.Fatal("no clean stop")
+ }
+ l2 := openTestLedgerWith(t, dir, "box1", clock, DurabilityLedgerOptions{AllClientsInstrumented: true, HandoffToken: token})
+ defer l2.Close(ctx, nil)
+ if got := l2.Attestation().Unmetered; len(got) != 0 {
+ t.Fatalf("a new run inherited the old run's exclusion set: %v", got)
+ }
+ d3Start, d3End := ledgerDay0.Add(72*time.Hour), ledgerDay0.Add(96*time.Hour)
+ clock.set(d3End.Add(time.Second))
+ if err := l2.Seal(); err != nil {
+ t.Fatal(err)
+ }
+ c3, err := l2.Snapshot(DayOf(d3Start))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if ok, reason := c3.CompletenessFor(tenantAttr.Account, d3Start, d3End); !ok {
+ t.Fatalf("the next run's day stayed incomplete for the account: %s", reason)
+ }
+}
+
+// An owner that cannot be named cannot be excluded: the run stops
+// attesting for anyone, and a later claim in the same run cannot undo it.
+func TestDurabilityEscapedUnknownOwnerWithdrawsTheRun(t *testing.T) {
+ dir := t.TempDir()
+ clock := newLedgerClock(ledgerDay0.Add(-time.Hour))
+ l := openTestLedger(t, dir, "box1", clock, true)
+ defer l.Close(context.Background(), nil)
+ clock.set(ledgerDay0.Add(6 * time.Hour))
+ l.Escaped("", clock.now())
+ clock.set(ledgerDay0.Add(7 * time.Hour))
+ if err := l.SetAllClientsInstrumented(true, clock.now()); err != nil {
+ t.Fatal(err)
+ }
+ if l.Attestation().Attested {
+ t.Fatal("a claim after an owner-unknown escape re-attested the run")
+ }
+ d1Start, d1End := ledgerDay0.Add(24*time.Hour), ledgerDay0.Add(48*time.Hour)
+ clock.set(d1End.Add(time.Second))
+ if err := l.Seal(); err != nil {
+ t.Fatal(err)
+ }
+ c1, err := l.Snapshot(DayOf(d1Start))
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, acct := range []string{"", tenantAttr.Account, tenant2Attr.Account} {
+ if ok, reason := c1.CompletenessFor(acct, d1Start, d1End); ok || !strings.Contains(reason, "attest") {
+ t.Fatalf("account %q after withdrawal: ok=%v reason=%q", acct, ok, reason)
+ }
+ }
+}
+
+// An escape while unattested is remembered: a later claim in the same run
+// carries the account in its exclusion set.
+func TestDurabilityEscapedWhileUnattestedCarriesIntoALaterClaim(t *testing.T) {
+ dir := t.TempDir()
+ clock := newLedgerClock(ledgerDay0.Add(-time.Hour))
+ l := openTestLedger(t, dir, "box1", clock, false)
+ defer l.Close(context.Background(), nil)
+ clock.set(ledgerDay0.Add(time.Hour))
+ l.Escaped(tenantAttr.Account, clock.now())
+ clock.set(ledgerDay0.Add(2 * time.Hour))
+ if err := l.SetAllClientsInstrumented(true, clock.now()); err != nil {
+ t.Fatal(err)
+ }
+ att := l.Attestation()
+ if !att.Attested || !slices.Contains(att.Unmetered, tenantAttr.Account) {
+ t.Fatalf("claim forgot the escape: %+v", att)
+ }
+ clock.set(ledgerDay0.Add(24*time.Hour + time.Second))
+ if err := l.Seal(); err != nil {
+ t.Fatal(err)
+ }
+ c, err := l.Snapshot(DayOf(ledgerDay0))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := validateDurabilityContent(&c); err != nil {
+ t.Fatalf("snapshot with an unattested exclusion did not validate: %v", err)
+ }
+}
diff --git a/internal/usage/durabilityledger.go b/internal/usage/durabilityledger.go
index 175f0ffc7..37c6f4013 100644
--- a/internal/usage/durabilityledger.go
+++ b/internal/usage/durabilityledger.go
@@ -129,6 +129,23 @@ type DurabilityLedger struct {
closed bool
stopToken string // the handoff token minted by a clean Close
latch failureLatch
+ // withdrawn: an owner-unknown client escaped the ledger in this run
+ // (Escaped), so no span of the run may be attested from then on,
+ // whatever SetAttestation asks. Guarded by mu; reset only by a new open.
+ withdrawn bool
+ escapes escapeLatch
+}
+
+// escapeLatch holds Escaped calls not yet folded into the coverage
+// sessions. Its mutex is its own and is never held across I/O, so the
+// caller (a wake path that just failed to bind a meter) never waits on
+// the database. Entries are removed only once a session change carrying
+// them has committed.
+type escapeLatch struct {
+ mu sync.Mutex
+ accounts map[string]time.Time // account -> earliest unapplied escape
+ boxFrom time.Time // earliest unapplied owner-unknown escape (zero: none)
+ applying bool // the single background applier is running
}
// failureLatch is where Incomplete lands. Its mutex is its own — never
@@ -1470,8 +1487,24 @@ func (l *DurabilityLedger) SetAttestation(a DurabilityAttestation, at time.Time)
if at.Before(l.sessionFrom) {
return fmt.Errorf("usage ledger: attestation change at %s precedes the session start %s", at.Format(time.RFC3339), l.sessionFrom.Format(time.RFC3339))
}
- unmetered := normalizeUnmetered(append(append([]string(nil), l.unmetered...), a.Unmetered...))
- if a.Attested == l.attested && joinUnmetered(unmetered) == joinUnmetered(l.unmetered) {
+ // An escape not yet folded in lands first, at its own instant, so a
+ // new claim can never start a session that forgets it.
+ if err := l.applyEscapesLocked(); err != nil {
+ return err
+ }
+ if at.Before(l.sessionFrom) {
+ at = l.sessionFrom
+ }
+ return l.setAttestationLocked(a.Attested && !l.withdrawn, a.Unmetered, at)
+}
+
+// setAttestationLocked ends the current session at `at` and starts a new
+// one with the claim attested and the run's exclusion set grown by
+// unmetered. Caller holds mu, has validated unmetered, and guarantees
+// at >= sessionFrom.
+func (l *DurabilityLedger) setAttestationLocked(attested bool, extra []string, at time.Time) error {
+ unmetered := normalizeUnmetered(append(append([]string(nil), l.unmetered...), extra...))
+ if attested == l.attested && joinUnmetered(unmetered) == joinUnmetered(l.unmetered) {
return nil
}
tx, err := l.db.Begin()
@@ -1483,7 +1516,7 @@ func (l *DurabilityLedger) SetAttestation(a DurabilityAttestation, at time.Time)
if _, err := tx.Exec(`UPDATE sessions SET to_ns = max(to_ns, ?) WHERE seq = ?`, nsOf(at), l.session); err != nil {
return err
}
- seq, err := l.insertSession(tx, at, at, a.Attested, unmetered)
+ seq, err := l.insertSession(tx, at, at, attested, unmetered)
if err != nil {
l.state = before
return err
@@ -1500,11 +1533,129 @@ func (l *DurabilityLedger) SetAttestation(a DurabilityAttestation, at time.Time)
l.state = before
return err
}
- l.session, l.sessionFrom, l.attested, l.unmetered = seq, at, a.Attested, unmetered
+ l.session, l.sessionFrom, l.attested, l.unmetered = seq, at, attested, unmetered
l.eventful = true
return nil
}
+// Escaped records that a replication client in account's scope issues,
+// or may issue, S3 requests this ledger does not see: its caller could
+// not bind a recorder (an invalid owner, the bind budget ran out, the
+// binding table is full, the ledger refused the bind) and fell back to an
+// unmetered client. That is billing incompleteness, never silence. It
+// returns at once (no database, no wait on mu) and has two effects:
+//
+// - a recording failure is latched under at's date, exactly as
+// Incomplete does, so that box-day can never read complete for
+// anyone, whatever the attestation says;
+// - from at until the next open, the run's exclusion set carries the
+// account (a valid sqid), or, when the owner is unknown and so cannot
+// be named, the run stops attesting for anyone. Both are monotonic
+// for the run, for the reason SetAttestation's exclusion set is: the
+// caller cannot know when the unmetered client stops, only that it
+// dies with this process.
+//
+// The session change is written by a background applier within
+// milliseconds, and in any case before the next Seal closes a day or
+// SetAttestation starts a new claim; until it lands, the latched failure
+// keeps the day of the escape incomplete. An Escaped after Close latches
+// as late and voids the clean stop, as Incomplete does.
+func (l *DurabilityLedger) Escaped(account string, at time.Time) {
+ at = at.UTC()
+ if at.IsZero() {
+ at = l.now().UTC()
+ }
+ l.latchFailure(at)
+ e := &l.escapes
+ e.mu.Lock()
+ if release.ValidAccount(account) {
+ if e.accounts == nil {
+ e.accounts = map[string]time.Time{}
+ }
+ if t, ok := e.accounts[account]; !ok || at.Before(t) {
+ e.accounts[account] = at
+ }
+ } else if e.boxFrom.IsZero() || at.Before(e.boxFrom) {
+ e.boxFrom = at
+ }
+ spawn := !e.applying
+ e.applying = true
+ e.mu.Unlock()
+ if spawn {
+ go l.applyEscapes()
+ }
+}
+
+// applyEscapes is the single background applier. It stops when nothing
+// is pending, when the ledger is closed (Close's own seal applied what it
+// could; the latched failures and the late marker cover the rest), or on
+// a database error (the next Seal retries, and refuses to close a day
+// until it succeeds).
+func (l *DurabilityLedger) applyEscapes() {
+ e := &l.escapes
+ for {
+ l.mu.Lock()
+ var err error
+ if !l.closed {
+ err = l.applyEscapesLocked()
+ }
+ closed := l.closed
+ l.mu.Unlock()
+ e.mu.Lock()
+ if closed || err != nil || (len(e.accounts) == 0 && e.boxFrom.IsZero()) {
+ e.applying = false
+ e.mu.Unlock()
+ if err != nil {
+ log.Printf("usage ledger: unmetered client not yet folded into coverage (next seal retries): %v", err)
+ }
+ return
+ }
+ e.mu.Unlock()
+ }
+}
+
+// applyEscapesLocked folds every pending escape into the coverage
+// sessions: one session change at the earliest pending instant (clamped
+// to the current session's start), carrying the accounts into the run's
+// exclusion set and withdrawing attestation when an owner was unknown.
+// Caller holds mu. Entries leave the latch only after the commit.
+func (l *DurabilityLedger) applyEscapesLocked() error {
+ e := &l.escapes
+ e.mu.Lock()
+ if len(e.accounts) == 0 && e.boxFrom.IsZero() {
+ e.mu.Unlock()
+ return nil
+ }
+ var accounts []string
+ from := e.boxFrom
+ box := !e.boxFrom.IsZero()
+ for a, t := range e.accounts {
+ accounts = append(accounts, a)
+ if from.IsZero() || t.Before(from) {
+ from = t
+ }
+ }
+ e.mu.Unlock()
+ if from.Before(l.sessionFrom) {
+ from = l.sessionFrom
+ }
+ if err := l.setAttestationLocked(l.attested && !box && !l.withdrawn, accounts, from); err != nil {
+ return err
+ }
+ if box {
+ l.withdrawn = true
+ }
+ e.mu.Lock()
+ for _, a := range accounts {
+ delete(e.accounts, a)
+ }
+ if box {
+ e.boxFrom = time.Time{}
+ }
+ e.mu.Unlock()
+ return nil
+}
+
// Seal advances this run's coverage to now, files any gap it has learned
// of (a dirty marker, a clock step), closes every day the clock has moved
// past whose attempts have all finished, and retries recording failures
@@ -1525,6 +1676,11 @@ func (l *DurabilityLedger) Seal() error {
// seal runs with drainMu and l.mu held, in that order.
func (l *DurabilityLedger) seal(force bool) error {
+ // An unmetered client must be in the coverage before any day is
+ // closed: a seal that cannot fold it in closes nothing.
+ if err := l.applyEscapesLocked(); err != nil {
+ return err
+ }
now := l.now().UTC()
// A marker that appeared after this open: something recorded outside.
var markerAt time.Time