diff --git a/internal/beads/hqstore_ttl.go b/internal/beads/hqstore_ttl.go index 2251f8ce0b3..3c1d62e6044 100644 --- a/internal/beads/hqstore_ttl.go +++ b/internal/beads/hqstore_ttl.go @@ -13,15 +13,18 @@ var hqStoreTTLLogf = log.Printf type hqTTLSweepResult struct { Expired int Retention int - Backstop hqBackstopReclaim + Backstop HQStoreBackstopReclaim } -type hqBackstopReclaim struct { +// HQStoreBackstopReclaim records how many stale closed records a backstop +// sweep reclaimed from each HQStore tier. +type HQStoreBackstopReclaim struct { Main int Ephemeral int } -func (r hqBackstopReclaim) total() int { +// Total returns the aggregate reclaim count across all HQStore tiers. +func (r HQStoreBackstopReclaim) Total() int { return r.Main + r.Ephemeral } @@ -82,11 +85,13 @@ func (s *HQStore) DrainRetentionQueue() int { // than the supplied tier cutoff. It scans main-tier and ephemeral records with // their respective cutoffs and returns the total number deleted. func (s *HQStore) PurgeBackstop(mainCutoff, ephemeralCutoff time.Duration) (int, error) { - reclaim, err := s.purgeBackstop(mainCutoff, ephemeralCutoff) - return reclaim.total(), err + reclaim, err := s.PurgeBackstopDetailed(mainCutoff, ephemeralCutoff) + return reclaim.Total(), err } -func (s *HQStore) purgeBackstop(mainCutoff, ephemeralCutoff time.Duration) (hqBackstopReclaim, error) { +// PurgeBackstopDetailed removes orphaned closed records and returns per-tier +// reclaim counts for diagnostics and benchmark assertions. +func (s *HQStore) PurgeBackstopDetailed(mainCutoff, ephemeralCutoff time.Duration) (HQStoreBackstopReclaim, error) { now := time.Now() mainBefore := now.Add(-mainCutoff) ephemeralBefore := now.Add(-ephemeralCutoff) @@ -94,7 +99,7 @@ func (s *HQStore) purgeBackstop(mainCutoff, ephemeralCutoff time.Duration) (hqBa s.mu.Lock() defer s.mu.Unlock() if err := s.ensureOpenLocked(); err != nil { - return hqBackstopReclaim{}, err + return HQStoreBackstopReclaim{}, err } type candidate struct { @@ -114,7 +119,7 @@ func (s *HQStore) purgeBackstop(mainCutoff, ephemeralCutoff time.Duration) (hqBa } } - var reclaim hqBackstopReclaim + var reclaim HQStoreBackstopReclaim for _, candidate := range candidates { s.deleteLocked(candidate.id) if candidate.ephemeral { @@ -138,7 +143,7 @@ func (s *HQStore) runTTLSweep() hqTTLSweepResult { result.Retention = s.DrainRetentionQueue() - backstop, err := s.purgeBackstop(s.mainTierBackstopTTL, s.ephemeralBackstopTTL) + backstop, err := s.PurgeBackstopDetailed(s.mainTierBackstopTTL, s.ephemeralBackstopTTL) if err != nil { hqStoreTTLLogf("hqstore ttl sweep: purge backstop: %v", err) return result @@ -148,8 +153,8 @@ func (s *HQStore) runTTLSweep() hqTTLSweepResult { return result } -func (s *HQStore) reportBackstopReclaim(reclaim hqBackstopReclaim) { - total := reclaim.total() +func (s *HQStore) reportBackstopReclaim(reclaim HQStoreBackstopReclaim) { + total := reclaim.Total() if total == 0 { return } @@ -168,14 +173,14 @@ func (s *HQStore) reportBackstopReclaim(reclaim hqBackstopReclaim) { total, reclaim.Main, reclaim.Ephemeral) } -func (s *HQStore) emitBackstopLeakDetected(reclaim hqBackstopReclaim) { +func (s *HQStore) emitBackstopLeakDetected(reclaim HQStoreBackstopReclaim) { if s.leakRecorder == nil { return } payload := events.StoreBackstopLeakDetectedPayload{ MainCount: reclaim.Main, EphemeralCount: reclaim.Ephemeral, - TotalCount: reclaim.total(), + TotalCount: reclaim.Total(), Threshold: s.leakThreshold, } raw, err := json.Marshal(payload) diff --git a/internal/beads/hqstore_ttl_internal_test.go b/internal/beads/hqstore_ttl_internal_test.go index ec4b032293a..82ee89eb3f4 100644 --- a/internal/beads/hqstore_ttl_internal_test.go +++ b/internal/beads/hqstore_ttl_internal_test.go @@ -63,8 +63,8 @@ func TestHQStoreBackstopLeakDetectorEmitsTypedEventAboveThreshold(t *testing.T) result := store.runTTLSweep() - if result.Backstop.total() != 3 { - t.Fatalf("backstop total = %d, want 3", result.Backstop.total()) + if result.Backstop.Total() != 3 { + t.Fatalf("backstop total = %d, want 3", result.Backstop.Total()) } if len(rec.Events) != 1 { t.Fatalf("events recorded = %d, want 1: %+v", len(rec.Events), rec.Events) @@ -103,8 +103,8 @@ func TestHQStoreBackstopLeakDetectorSilentAtOrBelowThreshold(t *testing.T) { result := store.runTTLSweep() - if result.Backstop.total() != 3 { - t.Fatalf("backstop total = %d, want 3", result.Backstop.total()) + if result.Backstop.Total() != 3 { + t.Fatalf("backstop total = %d, want 3", result.Backstop.Total()) } if len(rec.Events) != 0 { t.Fatalf("events recorded = %d, want 0: %+v", len(rec.Events), rec.Events) diff --git a/internal/benchmarks/coordstore/adapters/hqstore/adapter.go b/internal/benchmarks/coordstore/adapters/hqstore/adapter.go index 9d5a24645d3..0c0299ccedf 100644 --- a/internal/benchmarks/coordstore/adapters/hqstore/adapter.go +++ b/internal/benchmarks/coordstore/adapters/hqstore/adapter.go @@ -183,6 +183,22 @@ func (a *Adapter) PurgeExpired(context.Context) (int, error) { return a.store.PurgeExpired() } +// PurgeBackstop removes stale closed records and reports per-tier reclaim +// counts for the benchmark backstop assertion. +func (a *Adapter) PurgeBackstop(_ context.Context, mainCutoff, ephemeralCutoff time.Duration) (coordstore.BackstopReclaim, error) { + if a.store == nil { + return coordstore.BackstopReclaim{}, fmt.Errorf("hqstore PurgeBackstop: adapter is not open") + } + reclaim, err := a.store.PurgeBackstopDetailed(mainCutoff, ephemeralCutoff) + if err != nil { + return coordstore.BackstopReclaim{}, err + } + return coordstore.BackstopReclaim{ + Main: reclaim.Main, + Ephemeral: reclaim.Ephemeral, + }, nil +} + // PrimeScan loads open records through the HQStore list path. func (a *Adapter) PrimeScan(context.Context) (int, error) { items, err := a.store.List(beads.ListQuery{AllowScan: true}) diff --git a/internal/benchmarks/coordstore/backstop.go b/internal/benchmarks/coordstore/backstop.go new file mode 100644 index 00000000000..b15b69a478a --- /dev/null +++ b/internal/benchmarks/coordstore/backstop.go @@ -0,0 +1,64 @@ +package coordstore + +import ( + "context" + "fmt" + "time" +) + +// BackstopReclaim records how many stale closed records a single backstop +// sweep reclaimed from each physical tier. +type BackstopReclaim struct { + Main int + Ephemeral int +} + +// Total returns the aggregate reclaim count across all tiers. +func (r BackstopReclaim) Total() int { + return r.Main + r.Ephemeral +} + +// BackstopPolicy configures a single benchmark backstop sweep assertion. +type BackstopPolicy struct { + LeakThreshold int + MainCutoff time.Duration + EphemeralCutoff time.Duration +} + +// Enabled reports whether the policy has enough data to run the assertion. +func (p BackstopPolicy) Enabled() bool { + return p.LeakThreshold > 0 && p.MainCutoff > 0 && p.EphemeralCutoff > 0 +} + +// BackstopReclaimer is implemented by adapters that can run their closed-record +// backstop sweep and report per-tier reclaim counts. +type BackstopReclaimer interface { + PurgeBackstop(ctx context.Context, mainCutoff, ephemeralCutoff time.Duration) (BackstopReclaim, error) +} + +// BackstopCheck captures one post-warm-up backstop assertion result. +type BackstopCheck struct { + Policy BackstopPolicy + Reclaim BackstopReclaim +} + +// Passed reports whether a single backstop sweep stayed within threshold. +func (c BackstopCheck) Passed() bool { + return c.Reclaim.Total() <= c.Policy.LeakThreshold +} + +// FailureReason formats the threshold failure with per-tier counts. +func (c BackstopCheck) FailureReason() string { + return fmt.Sprintf("backstop reclaim total=%d > threshold=%d per tick (main=%d ephemeral=%d)", + c.Reclaim.Total(), c.Policy.LeakThreshold, c.Reclaim.Main, c.Reclaim.Ephemeral) +} + +// CheckBackstopReclaim runs one adapter backstop sweep and returns its bounded +// reclaim assertion result. +func CheckBackstopReclaim(ctx context.Context, reclaimer BackstopReclaimer, policy BackstopPolicy) (BackstopCheck, error) { + reclaim, err := reclaimer.PurgeBackstop(ctx, policy.MainCutoff, policy.EphemeralCutoff) + if err != nil { + return BackstopCheck{}, err + } + return BackstopCheck{Policy: policy, Reclaim: reclaim}, nil +} diff --git a/internal/benchmarks/coordstore/backstop_test.go b/internal/benchmarks/coordstore/backstop_test.go new file mode 100644 index 00000000000..62b9c5c4211 --- /dev/null +++ b/internal/benchmarks/coordstore/backstop_test.go @@ -0,0 +1,76 @@ +package coordstore_test + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "github.com/gastownhall/gascity/internal/benchmarks/coordstore" +) + +func TestCheckBackstopReclaimPassesAtThreshold(t *testing.T) { + reclaimer := &fakeBackstopReclaimer{ + reclaim: coordstore.BackstopReclaim{Main: 3, Ephemeral: 2}, + } + policy := coordstore.BackstopPolicy{ + LeakThreshold: 5, + MainCutoff: 24 * time.Hour, + EphemeralCutoff: 10 * time.Minute, + } + + check, err := coordstore.CheckBackstopReclaim(context.Background(), reclaimer, policy) + if err != nil { + t.Fatalf("CheckBackstopReclaim: %v", err) + } + if !check.Passed() { + t.Fatalf("Passed() = false, want true: %s", check.FailureReason()) + } + if reclaimer.mainCutoff != policy.MainCutoff || reclaimer.ephemeralCutoff != policy.EphemeralCutoff { + t.Fatalf("cutoffs = main:%s ephemeral:%s, want main:%s ephemeral:%s", + reclaimer.mainCutoff, reclaimer.ephemeralCutoff, policy.MainCutoff, policy.EphemeralCutoff) + } +} + +func TestBackstopCheckFailureReasonIncludesTierCounts(t *testing.T) { + check := coordstore.BackstopCheck{ + Policy: coordstore.BackstopPolicy{LeakThreshold: 5}, + Reclaim: coordstore.BackstopReclaim{Main: 6, Ephemeral: 2}, + } + + if check.Passed() { + t.Fatalf("Passed() = true, want false") + } + reason := check.FailureReason() + for _, want := range []string{"total=8", "threshold=5", "main=6", "ephemeral=2"} { + if !strings.Contains(reason, want) { + t.Fatalf("FailureReason() = %q, want substring %q", reason, want) + } + } +} + +func TestCheckBackstopReclaimReturnsReclaimerErrors(t *testing.T) { + wantErr := errors.New("sweep unavailable") + _, err := coordstore.CheckBackstopReclaim(context.Background(), &fakeBackstopReclaimer{err: wantErr}, coordstore.BackstopPolicy{ + LeakThreshold: 1, + MainCutoff: time.Hour, + EphemeralCutoff: time.Hour, + }) + if !errors.Is(err, wantErr) { + t.Fatalf("CheckBackstopReclaim error = %v, want %v", err, wantErr) + } +} + +type fakeBackstopReclaimer struct { + reclaim coordstore.BackstopReclaim + err error + mainCutoff time.Duration + ephemeralCutoff time.Duration +} + +func (f *fakeBackstopReclaimer) PurgeBackstop(_ context.Context, mainCutoff, ephemeralCutoff time.Duration) (coordstore.BackstopReclaim, error) { + f.mainCutoff = mainCutoff + f.ephemeralCutoff = ephemeralCutoff + return f.reclaim, f.err +} diff --git a/internal/benchmarks/coordstore/suite_test.go b/internal/benchmarks/coordstore/suite_test.go index 58d0ad48f97..4eb75e978ac 100644 --- a/internal/benchmarks/coordstore/suite_test.go +++ b/internal/benchmarks/coordstore/suite_test.go @@ -162,6 +162,7 @@ func runSuite(t *testing.T, wl coordstore.WorkloadConfig, enforceTargets bool) { // Phase 5: Print scorecard. sc.PrintTable(testWriter{t}) + checkBackstopReclaim(ctx, t, adapter, wl) // Report target outcomes; only gate the test when enforceTargets is set. for _, r := range sc.Results { @@ -186,6 +187,28 @@ func runSuite(t *testing.T, wl coordstore.WorkloadConfig, enforceTargets bool) { } } +func checkBackstopReclaim(ctx context.Context, t *testing.T, adapter coordstore.StoreAdapter, wl coordstore.WorkloadConfig) { + t.Helper() + reclaimer, ok := adapter.(coordstore.BackstopReclaimer) + if !ok { + return + } + policy := wl.BackstopPolicy() + if !policy.Enabled() { + return + } + check, err := coordstore.CheckBackstopReclaim(ctx, reclaimer, policy) + if err != nil { + t.Errorf(" FAIL backstop reclaim: %v", err) + return + } + t.Logf(" BackstopReclaim: total=%d threshold=%d main=%d ephemeral=%d", + check.Reclaim.Total(), policy.LeakThreshold, check.Reclaim.Main, check.Reclaim.Ephemeral) + if !check.Passed() { + t.Errorf(" FAIL backstop reclaim: %s", check.FailureReason()) + } +} + // printComparison prints a side-by-side pass/fail matrix for all backends. func printComparison(t *testing.T, scorecards []coordstore.Scorecard) { t.Helper() diff --git a/internal/benchmarks/coordstore/workload.go b/internal/benchmarks/coordstore/workload.go index abaa35b067d..93d48db1279 100644 --- a/internal/benchmarks/coordstore/workload.go +++ b/internal/benchmarks/coordstore/workload.go @@ -75,6 +75,18 @@ type WorkloadConfig struct { // Live HQ: used for inbox-replay / archive; ~0.02/s. RecentScanRate float64 + // --- Backstop reclaim assertion --- + + // BackstopLeakThreshold is the maximum acceptable reclaim count for one + // post-warm-up backstop sweep. Zero disables the assertion. + BackstopLeakThreshold int + // BackstopMainCutoff is the closed main-tier age eligible for backstop + // reclaim during the post-warm-up assertion. + BackstopMainCutoff time.Duration + // BackstopEphemeralCutoff is the closed ephemeral-tier age eligible for + // backstop reclaim during the post-warm-up assertion. + BackstopEphemeralCutoff time.Duration + // --- Run parameters --- // Duration is how long the workload driver runs. @@ -121,6 +133,10 @@ var RealWorldWorkload = WorkloadConfig{ DepOpRate: 0.1, // dep add/remove (S4 W5 approx.) RecentScanRate: 0.02, // inbox-replay (S3 R11 proxy) + BackstopLeakThreshold: 5, + BackstopMainCutoff: 24 * time.Hour, + BackstopEphemeralCutoff: 10 * time.Minute, + Duration: 30 * time.Second, Concurrency: 20, } @@ -147,6 +163,10 @@ var StressWorkload = WorkloadConfig{ DepOpRate: 0.5, RecentScanRate: 0.1, + BackstopLeakThreshold: 5, + BackstopMainCutoff: 24 * time.Hour, + BackstopEphemeralCutoff: 10 * time.Minute, + Duration: 15 * time.Second, Concurrency: 50, } @@ -172,6 +192,19 @@ var SmokeWorkload = WorkloadConfig{ DepOpRate: 0.1, RecentScanRate: 0.1, + BackstopLeakThreshold: 5, + BackstopMainCutoff: 24 * time.Hour, + BackstopEphemeralCutoff: 10 * time.Minute, + Duration: 5 * time.Second, Concurrency: 5, } + +// BackstopPolicy returns the post-warm-up backstop assertion policy for wl. +func (wl WorkloadConfig) BackstopPolicy() BackstopPolicy { + return BackstopPolicy{ + LeakThreshold: wl.BackstopLeakThreshold, + MainCutoff: wl.BackstopMainCutoff, + EphemeralCutoff: wl.BackstopEphemeralCutoff, + } +}