Add bounded HQStore backstop reclaim assertions

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,19 +85,21 @@ 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)
 
 	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)

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)

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})

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
+}

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
+}

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()

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,
+	}
+}