Publish validated durability holds with fail-closed reader compatibility
4 files changed,
+512
−24
internal/console/durability_hold_reader_test.go+83 −0internal/usage/durability.go+125 −22internal/usage/durability_hold_reader_test.go+271 −0internal/usage/durabilityledger.go+33 −2
diff --git a/internal/console/durability_hold_reader_test.go b/internal/console/durability_hold_reader_test.go| new file mode 100644 |
| index 000000000..9fc5bedc5 |
| --- /dev/null |
| +++ b/internal/console/durability_hold_reader_test.go |
| @@ -0,0 +1,83 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/usage" |
| +) |
| + |
| +func TestDurabilityFoldHoldRows(t *testing.T) { |
| + for _, kind := range []string{"owner", "box", "barrier", "owner_and_failure"} { |
| + t.Run(kind, func(t *testing.T) { |
| + fx := newConsoleFixture(t) |
| + ctx := context.Background() |
| + if err := fx.st.CreateAccount(testAccountID(), foldAcctB, "Always on", "owner-b@example.test", time.Date(2026, 8, 20, 0, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + publishDurabilityDay(t, fx, "box1", true, tenantWrites(t)) |
| + key := usage.DurabilityDayKey("box1", "2026-09-17") |
| + raw, etag, err := fx.rel.Get(ctx, key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + day, err := usage.ParseDurabilityDay(raw, "box1", "2026-09-17") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, snap := range day.Epochs { |
| + content, err := snap.Decode() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + content.LegacyIncomplete = true |
| + content.RecordingFailures++ |
| + if kind == "barrier" { |
| + content.Barred = 1 |
| + } else { |
| + hold := usage.DurabilityHold{Account: foldAcctB, App: "shop", UncertaintyID: 1, WrittenAt: time.Date(2026, 9, 17, 12, 0, 0, 0, time.UTC)} |
| + if kind == "box" { |
| + hold.Account = "" |
| + hold.App = "" |
| + } |
| + content.Holds = []usage.DurabilityHold{hold} |
| + content.Held = 1 |
| + } |
| + if kind == "owner_and_failure" { |
| + *content.ActualRecordingFailures++ |
| + content.RecordingFailures++ |
| + } |
| + contentRaw, err := json.Marshal(content) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + digest := sha256.Sum256(contentRaw) |
| + snap.Content = contentRaw |
| + snap.Hash = hex.EncodeToString(digest[:]) |
| + } |
| + raw, err = json.Marshal(day) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := fx.rel.Put(ctx, key, raw, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := fx.srv.statementsPass(ctx, time.Date(2026, 9, 20, 6, 0, 0, 0, time.UTC)); err != nil { |
| + t.Fatal(err) |
| + } |
| + a := septemberStatement(t, fx, foldAcctA).SourceDays[usage.SourceDurability] |
| + b := septemberStatement(t, fx, foldAcctB).SourceDays[usage.SourceDurability] |
| + wantA := 0 |
| + if kind == "owner" { |
| + wantA = 1 |
| + } |
| + if a != wantA || b != 0 { |
| + t.Fatalf("%s: source days A=%d B=%d, want A=%d B=0", kind, a, b, wantA) |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/usage/durability.go b/internal/usage/durability.go| index 78666846b..c19cf21ab 100644 |
| --- a/internal/usage/durability.go |
| +++ b/internal/usage/durability.go |
| @@ -7,6 +7,7 @@ import ( |
| "encoding/json" |
| "errors" |
| "fmt" |
| + "io/fs" |
| "math" |
| "regexp" |
| "sort" |
| @@ -179,9 +180,7 @@ type DurabilitySpan struct { |
| Unmetered []string `json:"unmetered,omitempty"` |
| } |
| -// durabilityContentVersion is what this binary writes. Readers accept 1 |
| -// (no exclusion sets) and 2 (this). |
| -const durabilityContentVersion = 2 |
| +const durabilityContentVersion = 3 |
| // DurabilityGap says requests may have gone unrecorded between From and To: |
| // the process started at To without a clean stop, the wall clock stepped |
| @@ -275,7 +274,7 @@ type DurabilityTotals struct { |
| // are the same — so every field must encode deterministically (sorted |
| // rows, sorted map keys, UTC instants). |
| type DurabilityContent struct { |
| - V int `json:"v"` // 1 or 2 (durabilityContentVersion) |
| + V int `json:"v"` |
| Box string `json:"box"` |
| Date string `json:"date"` |
| Epoch string `json:"epoch"` |
| @@ -289,12 +288,9 @@ type DurabilityContent struct { |
| Coverage []DurabilitySpan `json:"coverage,omitempty"` |
| Gaps []DurabilityGap `json:"gaps,omitempty"` |
| // Hours is "09" -> rows, sorted by host, account, app, purpose, operation. |
| - Hours map[string][]DurabilityRow `json:"hours,omitempty"` |
| - Totals DurabilityTotals `json:"totals"` |
| - // RecordingFailures counts requests the ledger was asked to record and |
| - // could not (a full disk, a refused operation name): each is a request |
| - // that happened and is not in the rows. |
| - RecordingFailures int64 `json:"recording_failures,omitempty"` |
| + Hours map[string][]DurabilityRow `json:"hours,omitempty"` |
| + Totals DurabilityTotals `json:"totals"` |
| + RecordingFailures int64 `json:"recording_failures,omitempty"` |
| // Late counts attempts recorded under this date after it was closed |
| // (a wall clock stepped back across midnight). |
| Late int64 `json:"late,omitempty"` |
| @@ -302,13 +298,25 @@ type DurabilityContent struct { |
| // design, Part C) admitted on or before this day and not yet settled: |
| // a request may still be in flight, so the day is not complete while |
| // any remains (the barrier, temporary and box-wide; it writes nothing |
| - // and lifts at settlement). An older reader ignores the field; the |
| - // design's box-wide marker for older readers is a later slice. |
| + // and lifts at settlement). |
| Barred int64 `json:"barred,omitempty"` |
| // Held counts the hold rows on this day (Part C's uncertainty: a |
| // fleet's authority was found changed and this day is inside the |
| - // interval it may have been changed in); permanent and box-wide here. |
| - Held int64 `json:"held,omitempty"` |
| + // interval it may have been changed in). |
| + Held int64 `json:"held,omitempty"` |
| + Holds []DurabilityHold `json:"holds,omitempty"` |
| + // ActualRecordingFailures counts requests the ledger was asked to record and |
| + // could not (a full disk, a refused operation name): each is a request |
| + // that happened and is not in the rows. |
| + ActualRecordingFailures *int64 `json:"actual_recording_failures,omitempty"` |
| + LegacyIncomplete bool `json:"legacy_incomplete,omitempty"` |
| +} |
| + |
| +type DurabilityHold struct { |
| + Account string `json:"account,omitempty"` |
| + App string `json:"app,omitempty"` |
| + UncertaintyID int64 `json:"uncertainty_id"` |
| + WrittenAt time.Time `json:"written_at"` |
| } |
| // DurabilitySnapshot is one epoch's published snapshot of a day, with the |
| @@ -339,6 +347,17 @@ func (s *DurabilitySnapshot) Decode() (DurabilityContent, error) { |
| if err := json.Unmarshal(s.Content, &c); err != nil { |
| return DurabilityContent{}, fmt.Errorf("usage: durability snapshot content: %w", err) |
| } |
| + if c.V < 3 { |
| + var fields map[string]json.RawMessage |
| + if err := json.Unmarshal(s.Content, &fields); err != nil { |
| + return DurabilityContent{}, err |
| + } |
| + for key := range fields { |
| + if strings.EqualFold(key, "holds") || strings.EqualFold(key, "actual_recording_failures") || strings.EqualFold(key, "legacy_incomplete") { |
| + return DurabilityContent{}, fmt.Errorf("usage: durability snapshot content: %w", fs.ErrInvalid) |
| + } |
| + } |
| + } |
| if err := validateDurabilityContent(&c); err != nil { |
| return DurabilityContent{}, err |
| } |
| @@ -534,7 +553,19 @@ func (d DurabilityDay) Scope(from, to time.Time) (ok bool, unmetered []string, r |
| if !ok { |
| return false, nil, reason |
| } |
| - return true, c.UnmeteredWithin(from, to), "" |
| + excluded := c.UnmeteredWithin(from, to) |
| + seen := make(map[string]bool, len(excluded)) |
| + for _, account := range excluded { |
| + seen[account] = true |
| + } |
| + for _, hold := range c.Holds { |
| + if hold.Account != "" && !seen[hold.Account] { |
| + excluded = append(excluded, hold.Account) |
| + seen[hold.Account] = true |
| + } |
| + } |
| + sort.Strings(excluded) |
| + return true, excluded, "" |
| } |
| return false, nil, "no snapshot" |
| } |
| @@ -553,7 +584,7 @@ func (c DurabilityContent) completenessAnyAttested(from, to time.Time) (bool, st |
| if to.After(end) { |
| to = end |
| } |
| - if ok, reason := c.CompletenessFor("", from, to); ok { |
| + if ok, reason := c.completenessFor("", from, to, true); ok { |
| return true, "" |
| } else if !strings.HasPrefix(reason, "coverage not attested") && !strings.HasPrefix(reason, "coverage does not span") { |
| return false, reason |
| @@ -592,14 +623,34 @@ func (c DurabilityContent) Completeness() (bool, string) { |
| // An empty account is the strict box-wide question: only spans attested |
| // with an empty exclusion set count. |
| func (c DurabilityContent) CompletenessFor(account string, from, to time.Time) (bool, string) { |
| - if c.RecordingFailures > 0 { |
| - return false, fmt.Sprintf("%d recording failure(s): requests happened that the ledger could not record", c.RecordingFailures) |
| + return c.completenessFor(account, from, to, false) |
| +} |
| + |
| +func (c DurabilityContent) completenessFor(account string, from, to time.Time, ignoreOwnerHolds bool) (bool, string) { |
| + if err := validateDurabilityHolds(&c); err != nil { |
| + return false, err.Error() |
| + } |
| + failures := c.RecordingFailures |
| + if c.V == 3 { |
| + failures = *c.ActualRecordingFailures |
| + } |
| + if failures > 0 { |
| + return false, fmt.Sprintf("%d recording failure(s): requests happened that the ledger could not record", failures) |
| } |
| if c.Barred > 0 { |
| return false, fmt.Sprintf("%d replication admission(s) unsettled: a request admitted on or before this day may still be in flight", c.Barred) |
| } |
| - if c.Held > 0 { |
| - return false, fmt.Sprintf("%d replication hold(s): a fleet's authority was found changed within an interval that covers this day", c.Held) |
| + held := c.Held |
| + if c.V == 3 { |
| + held = 0 |
| + for _, hold := range c.Holds { |
| + if hold.Account == "" || !ignoreOwnerHolds && (account == "" || hold.Account == account) { |
| + held++ |
| + } |
| + } |
| + } |
| + if held > 0 { |
| + return false, fmt.Sprintf("%d replication hold(s): a fleet's authority was found changed within an interval that covers this day", held) |
| } |
| if c.Late > 0 { |
| return false, fmt.Sprintf("%d attempt(s) recorded after the day was closed", c.Late) |
| @@ -736,8 +787,8 @@ func addInt64(a, b int64) (int64, bool) { |
| // version, non-negative counters, rows that add up, totals that match the |
| // rows, hours that are hours, spans and gaps that lie inside the day. |
| func validateDurabilityContent(c *DurabilityContent) error { |
| - if c.V != 1 && c.V != durabilityContentVersion { |
| - return fmt.Errorf("usage: durability content: schema v%d is not v1 or v%d", c.V, durabilityContentVersion) |
| + if c.V < 1 || c.V > durabilityContentVersion { |
| + return fmt.Errorf("usage: durability snapshot content: %w", fs.ErrInvalid) |
| } |
| start, end, err := DayBounds(c.Date) |
| if err != nil { |
| @@ -749,6 +800,9 @@ func validateDurabilityContent(c *DurabilityContent) error { |
| if c.Watermark < 0 || c.RecordingFailures < 0 || c.Late < 0 || c.Barred < 0 || c.Held < 0 { |
| return errors.New("usage: durability content: negative counter") |
| } |
| + if err := validateDurabilityHolds(c); err != nil { |
| + return err |
| + } |
| var sum DurabilityTotals |
| for h, rows := range c.Hours { |
| if !validHourLabel(h) { |
| @@ -829,3 +883,52 @@ func validateDurabilityContent(c *DurabilityContent) error { |
| } |
| return nil |
| } |
| + |
| +func validateDurabilityHolds(c *DurabilityContent) error { |
| + invalid := func() error { return fmt.Errorf("usage: durability snapshot content: %w", fs.ErrInvalid) } |
| + if c.V < 1 || c.V > durabilityContentVersion { |
| + return invalid() |
| + } |
| + if c.V < 3 { |
| + if c.Holds != nil || c.ActualRecordingFailures != nil || c.LegacyIncomplete { |
| + return invalid() |
| + } |
| + return nil |
| + } |
| + if c.ActualRecordingFailures == nil || *c.ActualRecordingFailures < 0 || c.Held < 0 || c.Barred < 0 || c.RecordingFailures < 0 { |
| + return invalid() |
| + } |
| + if c.Held != int64(len(c.Holds)) || c.LegacyIncomplete != (c.Held > 0 || c.Barred > 0) { |
| + return invalid() |
| + } |
| + want := *c.ActualRecordingFailures |
| + if c.LegacyIncomplete { |
| + var ok bool |
| + if want, ok = addInt64(want, 1); !ok { |
| + return errors.New("usage: durability content: counter overflow") |
| + } |
| + } |
| + if c.RecordingFailures != want { |
| + return invalid() |
| + } |
| + for i, hold := range c.Holds { |
| + if hold.UncertaintyID <= 0 || hold.WrittenAt.IsZero() || (hold.Account == "" && hold.App != "") || |
| + (hold.Account != "" && !release.ValidAccount(hold.Account)) || (hold.App != "" && !release.ValidApp(hold.App)) { |
| + return invalid() |
| + } |
| + if i > 0 && !durabilityHoldLess(c.Holds[i-1], hold) { |
| + return invalid() |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +func durabilityHoldLess(a, b DurabilityHold) bool { |
| + if a.Account != b.Account { |
| + return a.Account < b.Account |
| + } |
| + if a.App != b.App { |
| + return a.App < b.App |
| + } |
| + return a.UncertaintyID < b.UncertaintyID |
| +} |
diff --git a/internal/usage/durability_hold_reader_test.go b/internal/usage/durability_hold_reader_test.go| new file mode 100644 |
| index 000000000..9b641b504 |
| --- /dev/null |
| +++ b/internal/usage/durability_hold_reader_test.go |
| @@ -0,0 +1,271 @@ |
| +package usage |
| + |
| +import ( |
| + "encoding/json" |
| + "fmt" |
| + "math" |
| + "reflect" |
| + "strings" |
| + "testing" |
| + "time" |
| +) |
| + |
| +func holdReaderContent() DurabilityContent { |
| + zero := int64(0) |
| + return DurabilityContent{V: 3, Box: "box", Date: "2026-10-01", Epoch: "0123456789abcdef", Closed: true, |
| + Coverage: []DurabilitySpan{{From: at(1, 0, 0), To: at(2, 0, 0), Attested: true}}, ActualRecordingFailures: &zero} |
| +} |
| + |
| +func holdReaderDay(t *testing.T, c DurabilityContent) (DurabilityDay, []byte) { |
| + t.Helper() |
| + raw, err := json.Marshal(c) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + day := DurabilityDay{V: 1, Box: c.Box, Date: c.Date, Epochs: map[string]*DurabilitySnapshot{ |
| + c.Epoch: {Revision: 1, Hash: durabilityContentHash(raw), Content: raw}, |
| + }} |
| + wire, err := json.Marshal(day) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + parsed, err := ParseDurabilityDay(wire, c.Box, c.Date) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return parsed, raw |
| +} |
| + |
| +func TestHoldReaderScopeAndLegacyMarker(t *testing.T) { |
| + for _, kind := range []string{"owner", "box", "barrier", "failure_and_owner"} { |
| + t.Run(kind, func(t *testing.T) { |
| + c := holdReaderContent() |
| + c.LegacyIncomplete = true |
| + c.RecordingFailures = 1 |
| + if kind == "barrier" { |
| + c.Barred = 1 |
| + } else { |
| + h := DurabilityHold{Account: "alice", App: "app", UncertaintyID: 1, WrittenAt: at(1, 12, 0)} |
| + if kind == "box" { |
| + h.Account = "" |
| + h.App = "" |
| + } |
| + c.Holds = []DurabilityHold{h} |
| + c.Held = 1 |
| + } |
| + if kind == "failure_and_owner" { |
| + *c.ActualRecordingFailures = 1 |
| + c.RecordingFailures = 2 |
| + } |
| + day, raw := holdReaderDay(t, c) |
| + decoded, err := day.Epochs[c.Epoch].Decode() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, account := range []string{"", "alice", "bob"} { |
| + complete, _ := decoded.CompletenessFor(account, at(1, 0, 0), at(2, 0, 0)) |
| + want := kind == "owner" && account == "bob" |
| + if complete != want { |
| + t.Fatalf("%s complete=%v, want %v", account, complete, want) |
| + } |
| + } |
| + ok, excluded, why := day.Scope(at(1, 0, 0), at(2, 0, 0)) |
| + if kind == "owner" { |
| + if !ok || !reflect.DeepEqual(excluded, []string{"alice"}) { |
| + t.Fatalf("scope %v %v %s", ok, excluded, why) |
| + } |
| + } else if ok { |
| + t.Fatalf("scope accepted %s: %v", kind, excluded) |
| + } |
| + var legacy struct { |
| + V int `json:"v"` |
| + RecordingFailures int64 `json:"recording_failures"` |
| + } |
| + if err := json.Unmarshal(raw, &legacy); err != nil { |
| + t.Fatal(err) |
| + } |
| + // The pre-hold v2 reader accepts only versions 1/2 and fails every owner on any recording failure. |
| + if legacy.V == 1 || legacy.V == 2 { |
| + t.Fatal("old strict version gate accepted new content") |
| + } |
| + if legacy.RecordingFailures == 0 { |
| + t.Fatal("old reader ignoring version/unknown fields can read protected day complete") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestHoldReaderScopeUnionsCoverageAndHolds(t *testing.T) { |
| + c := holdReaderContent() |
| + c.Coverage[0].Unmetered = []string{"bob"} |
| + c.Holds = []DurabilityHold{{Account: "alice", UncertaintyID: 1, WrittenAt: at(1, 1, 0)}, {Account: "bob", UncertaintyID: 2, WrittenAt: at(1, 2, 0)}} |
| + c.Held = 2 |
| + c.LegacyIncomplete = true |
| + c.RecordingFailures = 1 |
| + day, _ := holdReaderDay(t, c) |
| + ok, excluded, reason := day.Scope(at(1, 0, 0), at(2, 0, 0)) |
| + if !ok || !reflect.DeepEqual(excluded, []string{"alice", "bob"}) { |
| + t.Fatalf("%v %v %s", ok, excluded, reason) |
| + } |
| + if ok, _ := c.CompletenessFor("carol", at(1, 0, 0), at(2, 0, 0)); !ok { |
| + t.Fatal("unaffected account refused") |
| + } |
| +} |
| + |
| +func TestHoldReaderRejectsMalformedMetadata(t *testing.T) { |
| + tests := map[string]func(*DurabilityContent){ |
| + "missing actual": func(c *DurabilityContent) { c.ActualRecordingFailures = nil }, |
| + "negative actual": func(c *DurabilityContent) { *c.ActualRecordingFailures = -1 }, |
| + "negative barrier": func(c *DurabilityContent) { c.Barred = -1 }, |
| + "missing marker": func(c *DurabilityContent) { c.LegacyIncomplete = false }, |
| + "marker without protection": func(c *DurabilityContent) { c.Holds = nil; c.Held = 0 }, |
| + "missing legacy count": func(c *DurabilityContent) { c.RecordingFailures = 0 }, |
| + "subtract real failure": func(c *DurabilityContent) { *c.ActualRecordingFailures = 1 }, |
| + "extra legacy failures": func(c *DurabilityContent) { c.RecordingFailures = 2 }, |
| + "overflow": func(c *DurabilityContent) { *c.ActualRecordingFailures = math.MaxInt64 }, |
| + "wrong row count": func(c *DurabilityContent) { c.Held = 2 }, |
| + "missing row": func(c *DurabilityContent) { c.Holds = nil }, |
| + "box with app": func(c *DurabilityContent) { c.Holds[0].Account = "" }, |
| + "bad account": func(c *DurabilityContent) { c.Holds[0].Account = "bad/account" }, |
| + "bad app": func(c *DurabilityContent) { c.Holds[0].App = "bad/app" }, |
| + "zero uncertainty": func(c *DurabilityContent) { c.Holds[0].UncertaintyID = 0 }, |
| + "missing written": func(c *DurabilityContent) { c.Holds[0].WrittenAt = time.Time{} }, |
| + "duplicate row": func(c *DurabilityContent) { c.Holds = append(c.Holds, c.Holds[0]); c.Held = 2 }, |
| + "unordered": func(c *DurabilityContent) { |
| + c.Holds = append(c.Holds, DurabilityHold{Account: "aaa", UncertaintyID: 2, WrittenAt: at(1, 0, 0)}) |
| + c.Held = 2 |
| + }, |
| + } |
| + for name, mutate := range tests { |
| + t.Run(name, func(t *testing.T) { |
| + c := holdReaderContent() |
| + c.Held = 1 |
| + c.LegacyIncomplete = true |
| + c.RecordingFailures = 1 |
| + c.Holds = []DurabilityHold{{Account: "alice", App: "app", UncertaintyID: 1, WrittenAt: at(1, 12, 0)}} |
| + mutate(&c) |
| + raw, err := json.Marshal(c) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + snap := DurabilitySnapshot{Revision: 1, Hash: durabilityContentHash(raw), Content: raw} |
| + if _, err := snap.Decode(); err == nil { |
| + t.Fatal("malformed protection metadata accepted") |
| + } |
| + if ok, _ := c.CompletenessFor("bob", at(1, 0, 0), at(2, 0, 0)); ok { |
| + t.Fatal("direct completeness hid invalid protection") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestHoldReaderOldPublishedBytesAndHashes(t *testing.T) { |
| + for _, version := range []int{1, 2} { |
| + raw := []byte(fmt.Sprintf(`{"v":%d,"box":"box","date":"2026-10-01","epoch":"0123456789abcdef","watermark":4,"closed":true,"coverage":[{"from":"2026-10-01T00:00:00Z","to":"2026-10-02T00:00:00Z","attested":true}],"totals":{"attempts":0,"answered":0,"unanswered":0,"unknown":0}}`, version)) |
| + hashes := map[int]string{1: "e5ee420999d81e02e776c26f964428114812616624d9c56afec36aeb6818acd6", 2: "8e6f33b764b3508576efdba9c7f84a421e6939641d7b8e8323b7268213b7a7fc"} |
| + snap := DurabilitySnapshot{Revision: 1, Hash: hashes[version], Content: raw} |
| + c, err := snap.Decode() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + remarshal, err := json.Marshal(c) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if string(remarshal) != string(raw) || durabilityContentHash(remarshal) != snap.Hash { |
| + t.Fatalf("v%d bytes/hash changed: %s", version, remarshal) |
| + } |
| + if ok, why := c.Completeness(); !ok { |
| + t.Fatalf("v%d %s", version, why) |
| + } |
| + for _, metadata := range []string{`"holds":[]`, `"holds":null`, `"actual_recording_failures":0`, `"actual_recording_failures":null`, `"legacy_incomplete":false`, `"legacy_incomplete":true`, `"LEGACY_INCOMPLETE":false`, `"ACTUAL_RECORDING_FAILURES":null`} { |
| + amended := []byte(strings.TrimSuffix(string(raw), "}") + "," + metadata + "}") |
| + bad := DurabilitySnapshot{Revision: 1, Hash: durabilityContentHash(amended), Content: amended} |
| + if _, err := bad.Decode(); err == nil { |
| + t.Fatalf("v%d accepted v3 metadata %s", version, metadata) |
| + } |
| + } |
| + } |
| +} |
| + |
| +func TestHoldReaderSnapshotRowsBarrierAndFailures(t *testing.T) { |
| + for _, kind := range []string{"owner", "box", "barrier", "latch", "owner_and_failure"} { |
| + t.Run(kind, func(t *testing.T) { |
| + l, _ := openLedgerAt(t, at(1, 12, 0)) |
| + switch kind { |
| + case "barrier": |
| + mustAdmit(t, l, "f", "alice", "app", at(1, 1, 0)) |
| + case "latch": |
| + l.latchDiscovery(Discovery{FleetID: "f", At: at(1, 12, 0), Proof: at(1, 0, 0)}, time.Time{}, nil) |
| + case "box": |
| + if _, err := l.Discover(Discovery{FleetID: "f", At: at(1, 12, 0), Proof: at(1, 0, 0)}); err != nil { |
| + t.Fatal(err) |
| + } |
| + default: |
| + if _, err := l.db.Exec(`INSERT INTO holds(day,account,app,uncertainty_id,written_at) VALUES (?,?,?,?,?)`, "2026-10-01", "alice", "app", 1, nsOf(at(1, 12, 0))); err != nil { |
| + t.Fatal(err) |
| + } |
| + if kind == "owner_and_failure" { |
| + l.latchFailure(at(1, 12, 0)) |
| + } |
| + } |
| + c, err := l.Snapshot("2026-10-01") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if c.V != 3 || !c.LegacyIncomplete || c.ActualRecordingFailures == nil || c.RecordingFailures != *c.ActualRecordingFailures+1 { |
| + t.Fatalf("missing compatibility marker: %+v", c) |
| + } |
| + if kind == "barrier" || kind == "latch" { |
| + if c.Barred != 1 || len(c.Holds) != 0 { |
| + t.Fatalf("barrier snapshot: %+v", c) |
| + } |
| + } else if len(c.Holds) != 1 || c.Held != 1 { |
| + t.Fatalf("missing hold rows: %+v", c) |
| + } |
| + if kind == "box" && (c.Holds[0].Account != "" || c.Holds[0].App != "") { |
| + t.Fatal("writer activated owner policy") |
| + } |
| + if kind == "owner_and_failure" && *c.ActualRecordingFailures != 1 { |
| + t.Fatal("real failure was lost") |
| + } |
| + holdReaderDay(t, c) |
| + }) |
| + } |
| +} |
| + |
| +func TestHoldReaderSnapshotCanonicalRows(t *testing.T) { |
| + l, _ := openLedgerAt(t, at(1, 12, 0)) |
| + for _, hold := range []DurabilityHold{ |
| + {Account: "bob", App: "shop", UncertaintyID: 3, WrittenAt: at(1, 3, 0)}, |
| + {Account: "alice", UncertaintyID: 2, WrittenAt: at(1, 2, 0)}, |
| + {Account: "bob", App: "shop", UncertaintyID: 1, WrittenAt: at(1, 1, 0)}, |
| + } { |
| + if _, err := l.db.Exec(`INSERT INTO holds(day,account,app,uncertainty_id,written_at) VALUES (?,?,?,?,?)`, "2026-10-01", hold.Account, hold.App, hold.UncertaintyID, nsOf(hold.WrittenAt)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + c, err := l.Snapshot("2026-10-01") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + want := []DurabilityHold{{Account: "alice", UncertaintyID: 2, WrittenAt: at(1, 2, 0)}, {Account: "bob", App: "shop", UncertaintyID: 1, WrittenAt: at(1, 1, 0)}, {Account: "bob", App: "shop", UncertaintyID: 3, WrittenAt: at(1, 3, 0)}} |
| + if !reflect.DeepEqual(c.Holds, want) { |
| + t.Fatalf("rows %+v, want %+v", c.Holds, want) |
| + } |
| + raw, err := json.Marshal(c) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + again, err := l.Snapshot("2026-10-01") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + second, err := json.Marshal(again) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if string(raw) != string(second) { |
| + t.Fatal("unchanged snapshot changed canonical content") |
| + } |
| +} |
diff --git a/internal/usage/durabilityledger.go b/internal/usage/durabilityledger.go| index 48aaaab20..ebc8a26aa 100644 |
| --- a/internal/usage/durabilityledger.go |
| +++ b/internal/usage/durabilityledger.go |
| @@ -2083,9 +2083,26 @@ func (l *DurabilityLedger) snapshotOnce(ctx context.Context, date string, before |
| } |
| // the permanent holds of the day (box-wide rows), and the latched |
| // discoveries not yet replayed whose bound reaches the day |
| - if err := tx.QueryRow(`SELECT COUNT(*) FROM holds WHERE day = ?`, date).Scan(&c.Held); err != nil { |
| + holds, err := tx.Query(`SELECT account, app, uncertainty_id, written_at FROM holds WHERE day = ? ORDER BY account, app, uncertainty_id`, date) |
| + if err != nil { |
| + return DurabilityContent{}, dayLineage{}, err |
| + } |
| + for holds.Next() { |
| + var hold DurabilityHold |
| + var written int64 |
| + if err := holds.Scan(&hold.Account, &hold.App, &hold.UncertaintyID, &written); err != nil { |
| + holds.Close() |
| + return DurabilityContent{}, dayLineage{}, err |
| + } |
| + hold.WrittenAt = time.Unix(0, written).UTC() |
| + c.Holds = append(c.Holds, hold) |
| + } |
| + if err := holds.Err(); err != nil { |
| + holds.Close() |
| return DurabilityContent{}, dayLineage{}, err |
| } |
| + holds.Close() |
| + c.Held = int64(len(c.Holds)) |
| if snapshotBeforeOverlay != nil { |
| snapshotBeforeOverlay() |
| } |
| @@ -2110,7 +2127,21 @@ func (l *DurabilityLedger) snapshotOnce(ctx context.Context, date string, before |
| if after != before { |
| return DurabilityContent{}, dayLineage{}, errSnapshotLatchChanged |
| } |
| - c.RecordingFailures += pending |
| + var ok bool |
| + if c.RecordingFailures, ok = addInt64(c.RecordingFailures, pending); !ok { |
| + return DurabilityContent{}, dayLineage{}, errors.New("usage: durability content: counter overflow") |
| + } |
| + actual := c.RecordingFailures |
| + c.ActualRecordingFailures = &actual |
| + c.LegacyIncomplete = c.Held > 0 || c.Barred > 0 |
| + if c.LegacyIncomplete { |
| + if c.RecordingFailures, ok = addInt64(c.RecordingFailures, 1); !ok { |
| + return DurabilityContent{}, dayLineage{}, errors.New("usage: durability content: counter overflow") |
| + } |
| + } |
| + if err := validateDurabilityHolds(&c); err != nil { |
| + return DurabilityContent{}, dayLineage{}, err |
| + } |
| return c, lineage, nil |
| } |