From 41956a02ece0d23eba485654f15e1243365e70aa Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Tue, 14 Jul 2026 14:43:53 -0400 Subject: [PATCH 1/8] test(longhaul): add retention pruner to bound workload collection disk usage An unbounded long-haul write test always eventually fills a fixed PVC (the westus2 cluster exhausted its 10Gi volume after ~24.9M writes, crash-looping Postgres). Add a count-based retention pruner that keeps only the most recent LONGHAUL_RETAIN_PER_WRITER documents per writer (default 2,000,000; 0 disables). The pruner deletes strictly below the verifier's confirmed floor, so it can never remove a document the durability oracle has not already accounted for. The verifier now publishes a per-writer confirmed floor via atomics and, on startup, seeds each writer's expected seq from the collection's current minimum seq so a post-restart scan does not misread the pruned prefix as a giant gap (false data loss). - config: LONGHAUL_RETAIN_PER_WRITER env + validation - workload: Pruner (index-backed range DeleteMany), Verifier.ConfirmedFloor + DB-min startup seed, DocsPruned metric - main: wire pruner (gated on RetainPerWriter > 0) - report/deploy: surface Docs Pruned; default retention in deployment ConfigMap - tests: pruner (incl. safety invariant), verifier floor, config load/validate Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: badfdbf1-0fe9-43da-9921-854304654217 Signed-off-by: Wenting Wu --- test/longhaul/cmd/longhaul/main.go | 13 ++- test/longhaul/config/config.go | 26 +++++ test/longhaul/config/config_test.go | 29 +++++ test/longhaul/deploy/deployment.yaml | 6 + test/longhaul/report/report.go | 1 + test/longhaul/workload/metrics.go | 7 ++ test/longhaul/workload/pruner.go | 137 ++++++++++++++++++++++ test/longhaul/workload/pruner_test.go | 144 ++++++++++++++++++++++++ test/longhaul/workload/verifier.go | 75 ++++++++++++ test/longhaul/workload/verifier_test.go | 20 ++++ 10 files changed, 457 insertions(+), 1 deletion(-) create mode 100644 test/longhaul/workload/pruner.go create mode 100644 test/longhaul/workload/pruner_test.go diff --git a/test/longhaul/cmd/longhaul/main.go b/test/longhaul/cmd/longhaul/main.go index c7f375bd..3fc3ab24 100644 --- a/test/longhaul/cmd/longhaul/main.go +++ b/test/longhaul/cmd/longhaul/main.go @@ -132,9 +132,20 @@ func run(cfg config.Config) int { // Start verifier. A single verifier is sufficient — see StartVerifier // godoc. Writers are passed so the verifier can detect tail loss by // comparing each writer's acked tip against what's in the DB. - workload.StartVerifier(ctx, db, writers, metrics, j) + verifier := workload.StartVerifier(ctx, db, writers, metrics, j) j.Info("main", "verifier started") + // Start retention pruner (bounds the workload collection so an unbounded + // write test cannot eventually exhaust the PVC). It prunes only documents + // below the verifier's confirmed floor, so it never affects the durability + // verdict. Disabled when RetainPerWriter == 0. + if cfg.RetainPerWriter > 0 { + workload.StartPruner(ctx, db.Collection(workload.CollectionName), writers, verifier, cfg.RetainPerWriter, metrics, j) + j.Info("main", fmt.Sprintf("retention pruner started (retain %d docs/writer)", cfg.RetainPerWriter)) + } else { + j.Info("main", "retention pruning disabled (LONGHAUL_RETAIN_PER_WRITER=0)") + } + // Configure operations. ops := []operations.Operation{ operations.NewScaleUp(clusterClient, healthMon, cfg.MaxInstances, cfg.RecoveryTimeout), diff --git a/test/longhaul/config/config.go b/test/longhaul/config/config.go index 5665bcd7..90ef1711 100644 --- a/test/longhaul/config/config.go +++ b/test/longhaul/config/config.go @@ -34,8 +34,17 @@ const ( // Operational toggles. EnvResetData = "LONGHAUL_RESET_DATA" + + // Data retention. Bounds the workload collection so an unbounded write + // test does not eventually exhaust the PVC. + EnvRetainPerWriter = "LONGHAUL_RETAIN_PER_WRITER" ) +// DefaultRetainPerWriter is the default number of most-recent documents kept +// per writer when pruning is enabled. At ~10 writes/sec/writer this retains +// roughly 55 hours of history per writer while bounding steady-state disk use. +const DefaultRetainPerWriter = 2_000_000 + // Config holds all configuration for a long haul test run. type Config struct { // MaxDuration is the maximum test duration. Zero means run until failure. @@ -77,6 +86,11 @@ type Config struct { // Default false so that pod restarts preserve durability history; opt in // for fresh local/dev iterations. ResetData bool + + // RetainPerWriter is the number of most-recent documents kept per writer. + // Older, already-verified documents are pruned to bound disk usage. Zero + // disables pruning (unbounded growth — the pre-retention behavior). + RetainPerWriter int64 } // DefaultConfig returns a Config with safe defaults for local development. @@ -93,6 +107,7 @@ func DefaultConfig() Config { MinInstances: 1, MaxInstances: 3, ReportInterval: 1 * time.Hour, + RetainPerWriter: DefaultRetainPerWriter, } } @@ -181,6 +196,14 @@ func LoadFromEnv() (Config, error) { cfg.ResetData = v == "true" || v == "1" || v == "yes" } + if v := os.Getenv(EnvRetainPerWriter); v != "" { + n, err := strconv.ParseInt(v, 10, 64) + if err != nil { + return cfg, fmt.Errorf("invalid %s=%q: %w", EnvRetainPerWriter, v, err) + } + cfg.RetainPerWriter = n + } + return cfg, nil } @@ -213,6 +236,9 @@ func (c *Config) Validate() error { if c.MaxInstances < c.MinInstances { return fmt.Errorf("max instances (%d) must be >= min instances (%d)", c.MaxInstances, c.MinInstances) } + if c.RetainPerWriter < 0 { + return fmt.Errorf("retain per writer must not be negative, got %d", c.RetainPerWriter) + } return nil } diff --git a/test/longhaul/config/config_test.go b/test/longhaul/config/config_test.go index 506803f5..dcf920a2 100644 --- a/test/longhaul/config/config_test.go +++ b/test/longhaul/config/config_test.go @@ -23,6 +23,7 @@ var _ = Describe("Config", func() { Expect(cfg.SteadyStateWait).To(Equal(60 * time.Second)) Expect(cfg.MinInstances).To(Equal(1)) Expect(cfg.MaxInstances).To(Equal(3)) + Expect(cfg.RetainPerWriter).To(Equal(int64(DefaultRetainPerWriter))) }) }) @@ -35,6 +36,7 @@ var _ = Describe("Config", func() { EnvDocumentDBURI, EnvNumWriters, EnvOpCooldown, EnvRecoveryTimeout, EnvSteadyStateWait, EnvMinInstances, EnvMaxInstances, EnvReportInterval, + EnvResetData, EnvRetainPerWriter, } { GinkgoT().Setenv(k, "") } @@ -103,6 +105,26 @@ var _ = Describe("Config", func() { Expect(err).NotTo(HaveOccurred()) Expect(cfg.DocumentDBURI).To(Equal("mongodb://localhost:27017")) }) + + It("parses RetainPerWriter from env", func() { + GinkgoT().Setenv(EnvRetainPerWriter, "500000") + cfg, err := LoadFromEnv() + Expect(err).NotTo(HaveOccurred()) + Expect(cfg.RetainPerWriter).To(Equal(int64(500_000))) + }) + + It("parses RetainPerWriter=0 to disable pruning", func() { + GinkgoT().Setenv(EnvRetainPerWriter, "0") + cfg, err := LoadFromEnv() + Expect(err).NotTo(HaveOccurred()) + Expect(cfg.RetainPerWriter).To(BeZero()) + }) + + It("returns error for invalid RetainPerWriter", func() { + GinkgoT().Setenv(EnvRetainPerWriter, "not-a-number") + _, err := LoadFromEnv() + Expect(err).To(MatchError(ContainSubstring(EnvRetainPerWriter))) + }) }) Describe("Validate", func() { @@ -160,6 +182,13 @@ var _ = Describe("Config", func() { cfg.MaxInstances = 4 Expect(cfg.Validate()).To(MatchError(ContainSubstring("must not exceed 3"))) }) + + It("fails when RetainPerWriter is negative", func() { + cfg := DefaultConfig() + cfg.ClusterName = "test" + cfg.RetainPerWriter = -1 + Expect(cfg.Validate()).To(MatchError(ContainSubstring("retain per writer"))) + }) }) Describe("IsEnabled", func() { diff --git a/test/longhaul/deploy/deployment.yaml b/test/longhaul/deploy/deployment.yaml index 49d5c09f..835e577f 100644 --- a/test/longhaul/deploy/deployment.yaml +++ b/test/longhaul/deploy/deployment.yaml @@ -59,6 +59,12 @@ data: # DocumentDB workload collection and longhaul-report ConfigMap checkpoints. # Override to "true" only for local/dev iterations. LONGHAUL_RESET_DATA: "false" + # Retention: keep only the most recent N documents per writer so the endurance + # write load does not grow the PVC without bound (a fixed disk always fills + # eventually otherwise). Pruning removes only already-verified history, so it + # never affects the durability verdict. At ~10 writes/sec/writer, 2,000,000 + # docs is roughly 55h of per-writer history. Set to "0" to disable pruning. + LONGHAUL_RETAIN_PER_WRITER: "2000000" # Replica range for scale operations. # Bounds for spec.instancesPerNode (CRD allows 1-3). Default Min=1, Max=3 # exercises full scale-up/down cycle. Set Min=Max to disable scale ops. diff --git a/test/longhaul/report/report.go b/test/longhaul/report/report.go index e7316a7f..d8c984d8 100644 --- a/test/longhaul/report/report.go +++ b/test/longhaul/report/report.go @@ -88,6 +88,7 @@ func GenerateMarkdown(s Summary) string { fmt.Fprintf(&b, "| Verify Passes | %d |\n", s.Metrics.VerifyPasses) fmt.Fprintf(&b, "| Gaps Detected | %d |\n", s.Metrics.GapsDetected) fmt.Fprintf(&b, "| Checksum Errors | %d |\n", s.Metrics.ChecksumErrors) + fmt.Fprintf(&b, "| Docs Pruned | %d |\n", s.Metrics.DocsPruned) b.WriteString("\n") // Disruption Windows diff --git a/test/longhaul/workload/metrics.go b/test/longhaul/workload/metrics.go index 64c1a416..9d05e9ac 100644 --- a/test/longhaul/workload/metrics.go +++ b/test/longhaul/workload/metrics.go @@ -43,6 +43,11 @@ type Metrics struct { // recomputed value. Non-zero => FAIL with reason "data loss". ChecksumErrors atomic.Int64 + // DocsPruned counts documents deleted by the retention pruner. Purely + // observational — pruning removes only already-verified history and never + // affects the durability verdict. + DocsPruned atomic.Int64 + // StartTime is when this Metrics was constructed; resets on pod restart. StartTime time.Time } @@ -63,6 +68,7 @@ type MetricsSnapshot struct { VerifyPasses int64 GapsDetected int64 ChecksumErrors int64 + DocsPruned int64 Elapsed time.Duration } @@ -75,6 +81,7 @@ func (m *Metrics) Snapshot() MetricsSnapshot { VerifyPasses: m.VerifyPasses.Load(), GapsDetected: m.VerifyGapsDetected.Load(), ChecksumErrors: m.ChecksumErrors.Load(), + DocsPruned: m.DocsPruned.Load(), Elapsed: time.Since(m.StartTime), } } diff --git a/test/longhaul/workload/pruner.go b/test/longhaul/workload/pruner.go new file mode 100644 index 00000000..41fa9bb4 --- /dev/null +++ b/test/longhaul/workload/pruner.go @@ -0,0 +1,137 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +package workload + +import ( + "context" + "fmt" + "time" + + "github.com/documentdb/documentdb-operator/test/longhaul/journal" + + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" +) + +const ( + // pruneInterval is how often the pruner trims old documents. A long-haul + // run writes ~10 docs/sec/writer, so a few thousand rows accumulate per + // writer between cycles — a small, index-backed DeleteMany each time. + pruneInterval = 5 * time.Minute +) + +// floorProvider reports the highest fully-verified seq per writer. *Verifier +// satisfies this; the pruner only ever deletes strictly below the floor, so it +// can never remove a document the verifier has not already accounted for. +type floorProvider interface { + ConfirmedFloor(writerID string) int64 +} + +// pruneBackend abstracts the delete so Pruner can be unit-tested without a +// live collection. Production uses docdbPruneBackend. +type pruneBackend interface { + // deleteThrough removes all documents for writerID with seq <= throughSeq + // and returns the number deleted. + deleteThrough(ctx context.Context, writerID string, throughSeq int64) (int64, error) +} + +// docdbPruneBackend adapts *mongo.Collection to pruneBackend. The delete filter +// rides the existing unique (writer_id, seq) index, so it is a bounded range +// delete rather than a collection scan. +type docdbPruneBackend struct { + coll *mongo.Collection +} + +func (m docdbPruneBackend) deleteThrough(ctx context.Context, writerID string, throughSeq int64) (int64, error) { + res, err := m.coll.DeleteMany(ctx, bson.D{ + {Key: "writer_id", Value: writerID}, + {Key: "seq", Value: bson.D{{Key: "$lte", Value: throughSeq}}}, + }) + if err != nil { + return 0, err + } + return res.DeletedCount, nil +} + +// Pruner bounds the workload collection by keeping only the most recent +// retainPerWriter documents per writer. It deletes strictly below the +// verifier's confirmed floor, so pruning never affects the durability verdict: +// every removed document was already scanned, and the verifier's startup +// DB-min seed keeps a post-restart scan from misreading the pruned prefix as a +// gap. +type Pruner struct { + writers []*Writer + floor floorProvider + backend pruneBackend + retainPerWriter int64 + metrics *Metrics + journal *journal.Journal +} + +// NewPruner constructs a Pruner. retainPerWriter must be > 0; callers gate on +// that (0 disables pruning entirely) before constructing. +func NewPruner(coll *mongo.Collection, writers []*Writer, floor floorProvider, retainPerWriter int64, metrics *Metrics, j *journal.Journal) *Pruner { + return &Pruner{ + writers: writers, + floor: floor, + backend: docdbPruneBackend{coll: coll}, + retainPerWriter: retainPerWriter, + metrics: metrics, + journal: j, + } +} + +// Run starts the prune loop. It blocks until the context is cancelled. +func (p *Pruner) Run(ctx context.Context) { + p.journal.Info("pruner", fmt.Sprintf("pruner started (retain %d docs/writer)", p.retainPerWriter)) + defer p.journal.Info("pruner", "pruner stopped") + + ticker := time.NewTicker(pruneInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + p.pruneAll(ctx) + } + } +} + +func (p *Pruner) pruneAll(ctx context.Context) { + for _, w := range p.writers { + p.pruneWriter(ctx, w.id) + } +} + +// pruneWriter deletes the writer's documents older than the retention window. +// The delete boundary is (confirmed floor - retainPerWriter): everything at or +// below it is both fully verified and outside the retained tail, so removing it +// is safe and bounds the writer's footprint to ~retainPerWriter documents. +func (p *Pruner) pruneWriter(ctx context.Context, writerID string) { + floor := p.floor.ConfirmedFloor(writerID) + throughSeq := floor - p.retainPerWriter + if throughSeq < 1 { + // Not enough verified history yet to prune anything. + return + } + + deleted, err := p.backend.deleteThrough(ctx, writerID, throughSeq) + if err != nil { + p.journal.Warn("pruner", fmt.Sprintf("prune failed for writer %s: %v", writerID, err)) + return + } + if deleted > 0 { + p.metrics.DocsPruned.Add(deleted) + p.journal.Info("pruner", fmt.Sprintf("pruned %d docs for writer %s (seq <= %d)", deleted, writerID, throughSeq)) + } +} + +// StartPruner launches a single pruner goroutine and returns it. +func StartPruner(ctx context.Context, coll *mongo.Collection, writers []*Writer, floor floorProvider, retainPerWriter int64, metrics *Metrics, j *journal.Journal) *Pruner { + p := NewPruner(coll, writers, floor, retainPerWriter, metrics, j) + go p.Run(ctx) + return p +} diff --git a/test/longhaul/workload/pruner_test.go b/test/longhaul/workload/pruner_test.go new file mode 100644 index 00000000..4f3470a6 --- /dev/null +++ b/test/longhaul/workload/pruner_test.go @@ -0,0 +1,144 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT License. + +package workload + +import ( + "context" + "errors" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + "github.com/documentdb/documentdb-operator/test/longhaul/journal" +) + +// fakeFloor is a controllable floorProvider. +type fakeFloor map[string]int64 + +func (f fakeFloor) ConfirmedFloor(writerID string) int64 { return f[writerID] } + +// fakePruneBackend records deleteThrough calls and returns scripted results. +type fakePruneBackend struct { + // deletedThrough[writerID] is the highest throughSeq requested. + calls []pruneCall + // perWriterDeleted maps writerID -> count returned by deleteThrough. + perWriterDeleted map[string]int64 + err error +} + +type pruneCall struct { + writerID string + throughSeq int64 +} + +func (f *fakePruneBackend) deleteThrough(_ context.Context, writerID string, throughSeq int64) (int64, error) { + f.calls = append(f.calls, pruneCall{writerID: writerID, throughSeq: throughSeq}) + if f.err != nil { + return 0, f.err + } + if f.perWriterDeleted != nil { + return f.perWriterDeleted[writerID], nil + } + return 0, nil +} + +func newTestPruner(b pruneBackend, floor floorProvider, retain int64, m *Metrics) *Pruner { + return &Pruner{ + writers: []*Writer{{id: "w000"}, {id: "w001"}}, + floor: floor, + backend: b, + retainPerWriter: retain, + metrics: m, + journal: journal.New(), + } +} + +var _ = Describe("Pruner.pruneWriter", func() { + It("deletes through (floor - retainPerWriter) when there is enough history", func() { + b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 500}} + m := NewMetrics() + p := newTestPruner(b, fakeFloor{"w000": 10_000}, 2_000, m) + + p.pruneWriter(context.Background(), "w000") + + Expect(b.calls).To(HaveLen(1)) + Expect(b.calls[0]).To(Equal(pruneCall{writerID: "w000", throughSeq: 8_000})) + Expect(m.DocsPruned.Load()).To(Equal(int64(500))) + }) + + It("does not delete when the floor is below the retention window", func() { + b := &fakePruneBackend{} + m := NewMetrics() + // floor 1500, retain 2000 => throughSeq = -500 < 1 => no delete. + p := newTestPruner(b, fakeFloor{"w000": 1_500}, 2_000, m) + + p.pruneWriter(context.Background(), "w000") + + Expect(b.calls).To(BeEmpty()) + Expect(m.DocsPruned.Load()).To(BeZero()) + }) + + It("does not delete when the floor is exactly at the retention window", func() { + b := &fakePruneBackend{} + m := NewMetrics() + // floor 2000, retain 2000 => throughSeq = 0 < 1 => no delete. + p := newTestPruner(b, fakeFloor{"w000": 2_000}, 2_000, m) + + p.pruneWriter(context.Background(), "w000") + + Expect(b.calls).To(BeEmpty()) + }) + + It("prunes the first document once the floor advances one past the window", func() { + b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 1}} + m := NewMetrics() + // floor 2001, retain 2000 => throughSeq = 1 => delete seq <= 1. + p := newTestPruner(b, fakeFloor{"w000": 2_001}, 2_000, m) + + p.pruneWriter(context.Background(), "w000") + + Expect(b.calls).To(HaveLen(1)) + Expect(b.calls[0].throughSeq).To(Equal(int64(1))) + }) + + It("does not count deletions or advance metrics when the backend errors", func() { + b := &fakePruneBackend{err: errors.New("delete failed")} + m := NewMetrics() + p := newTestPruner(b, fakeFloor{"w000": 10_000}, 2_000, m) + + p.pruneWriter(context.Background(), "w000") + + Expect(b.calls).To(HaveLen(1)) + Expect(m.DocsPruned.Load()).To(BeZero()) + }) + + It("never deletes at or above the confirmed floor (safety invariant)", func() { + b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 3}} + m := NewMetrics() + floor := int64(5_000) + p := newTestPruner(b, fakeFloor{"w000": floor}, 1_000, m) + + p.pruneWriter(context.Background(), "w000") + + Expect(b.calls).To(HaveLen(1)) + Expect(b.calls[0].throughSeq).To(BeNumerically("<", floor), + "pruner must only delete strictly below the verifier's confirmed floor") + }) +}) + +var _ = Describe("Pruner.pruneAll", func() { + It("prunes every writer using its own floor", func() { + b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 10, "w001": 20}} + m := NewMetrics() + p := newTestPruner(b, fakeFloor{"w000": 5_000, "w001": 9_000}, 1_000, m) + + p.pruneAll(context.Background()) + + Expect(b.calls).To(ConsistOf( + pruneCall{writerID: "w000", throughSeq: 4_000}, + pruneCall{writerID: "w001", throughSeq: 8_000}, + )) + Expect(m.DocsPruned.Load()).To(Equal(int64(30))) + }) +}) diff --git a/test/longhaul/workload/verifier.go b/test/longhaul/workload/verifier.go index 10addf7d..177ede75 100644 --- a/test/longhaul/workload/verifier.go +++ b/test/longhaul/workload/verifier.go @@ -5,7 +5,9 @@ package workload import ( "context" + "errors" "fmt" + "sync/atomic" "time" "github.com/documentdb/documentdb-operator/test/longhaul/journal" @@ -41,6 +43,17 @@ type Verifier struct { // a late-arriving fill at a missing seq is not re-checked. // Only mutated from the verifier goroutine, so no lock is needed. nextSeq map[string]int64 + + // seeded[writerID] records whether the verifier has performed the one-time + // startup seed of expectedSeq from the collection's current minimum seq. + // Only touched from the verifier goroutine. + seeded map[string]bool + + // floor[writerID] is the highest fully-verified seq for that writer + // (nextSeq-1), published as an atomic so the retention pruner can read it + // concurrently without racing the verifier goroutine. Everything at or + // below the floor has been accounted for and is safe to prune. + floor map[string]*atomic.Int64 } // NewVerifier creates a verifier. writers is the set of writers whose tips @@ -49,15 +62,33 @@ type Verifier struct { func NewVerifier(db *mongo.Database, writers []*Writer, metrics *Metrics, j *journal.Journal) *Verifier { coll := db.Collection(CollectionName, options.Collection(). SetReadConcern(readconcern.Majority())) + floor := make(map[string]*atomic.Int64, len(writers)) + for _, w := range writers { + floor[w.id] = &atomic.Int64{} + } return &Verifier{ metrics: metrics, journal: j, collection: coll, writers: writers, nextSeq: make(map[string]int64), + seeded: make(map[string]bool), + floor: floor, } } +// ConfirmedFloor returns the highest seq the verifier has fully accounted for +// for writerID (0 if none yet). Every seq at or below this value has been +// scanned, so a pruner may safely delete documents below it without the +// verifier ever re-reading them (steady state) or misreading them as gaps +// (restart, thanks to the startup DB-min seed). Safe for concurrent reads. +func (v *Verifier) ConfirmedFloor(writerID string) int64 { + if a, ok := v.floor[writerID]; ok { + return a.Load() + } + return 0 +} + // Run starts the verifier loop. It blocks until the context is cancelled. func (v *Verifier) Run(ctx context.Context) { v.journal.Info("verifier", "verifier started") @@ -205,8 +236,27 @@ func (v *Verifier) verifyWriter(ctx context.Context, w *Writer) { if expectedSeq == 0 { expectedSeq = 1 } + + // One-time startup seed: retention pruning deletes already-verified + // documents below the confirmed floor, so after a pod restart (which + // resets the in-memory nextSeq to 1) a naive scan from seq 1 would see the + // pruned prefix as one enormous gap — a false data-loss verdict. Advance + // expectedSeq to the collection's current minimum seq so the verifier only + // audits documents that still exist. Real gaps above the surviving floor + // are still detected. Done once per writer; steady-state cycles never + // re-read below nextSeq. + if !v.seeded[writerID] { + v.seeded[writerID] = true + if minSeq, err := v.minSeq(ctx, writerID); err != nil { + v.journal.Warn("verifier", fmt.Sprintf("min-seq seed failed for writer %s: %v", writerID, err)) + } else if minSeq > expectedSeq { + expectedSeq = minSeq + } + } + if maxSeq < expectedSeq { // Nothing new committed since last cycle. + v.publishFloor(writerID, expectedSeq-1) return } @@ -223,6 +273,31 @@ func (v *Verifier) verifyWriter(ctx context.Context, w *Writer) { } v.nextSeq[writerID] = r.newExpectedSeq + v.publishFloor(writerID, r.newExpectedSeq-1) +} + +// publishFloor records the highest fully-verified seq for writerID so the +// retention pruner can read it concurrently. No-op for writers not registered +// at construction (e.g. unit tests that pass nil writers). +func (v *Verifier) publishFloor(writerID string, floor int64) { + if a, ok := v.floor[writerID]; ok { + a.Store(floor) + } +} + +// minSeq returns the lowest seq currently stored for writerID, or 0 if the +// writer has no documents. +func (v *Verifier) minSeq(ctx context.Context, writerID string) (int64, error) { + opts := options.FindOne().SetSort(bson.D{{Key: "seq", Value: 1}}) + var doc WriteDocument + err := v.collection.FindOne(ctx, bson.M{"writer_id": writerID}, opts).Decode(&doc) + if err != nil { + if errors.Is(err, mongo.ErrNoDocuments) { + return 0, nil + } + return 0, err + } + return doc.Seq, nil } // scanWriter streams all docs for writerID with seq in [expectedSeq, maxSeq] diff --git a/test/longhaul/workload/verifier_test.go b/test/longhaul/workload/verifier_test.go index 799d4e0c..97ac9063 100644 --- a/test/longhaul/workload/verifier_test.go +++ b/test/longhaul/workload/verifier_test.go @@ -4,6 +4,8 @@ package workload import ( + "sync/atomic" + . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -45,6 +47,24 @@ var _ = Describe("Verifier", func() { // what the verifier actually does. Document the boundary. Skip("verifyAll requires a *mongo.Database; covered by long-haul integration runs") }) + + Describe("ConfirmedFloor", func() { + It("returns 0 for a writer registered but not yet verified", func() { + v := &Verifier{floor: map[string]*atomic.Int64{"w000": {}}} + Expect(v.ConfirmedFloor("w000")).To(BeZero()) + }) + + It("reflects the published floor and is isolated per writer", func() { + v := &Verifier{floor: map[string]*atomic.Int64{ + "w000": {}, "w001": {}, + }} + v.publishFloor("w000", 4_200) + Expect(v.ConfirmedFloor("w000")).To(Equal(int64(4_200))) + Expect(v.ConfirmedFloor("w001")).To(BeZero()) + Expect(v.ConfirmedFloor("unknown")).To(BeZero(), + "unregistered writers read as 0 so the pruner never deletes for them") + }) + }) }) // makeDoc constructs a valid WriteDocument whose checksum matches; tests that From 14550edaa158d4e73abd37a8f2843c202452d066 Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Wed, 15 Jul 2026 12:55:33 -0400 Subject: [PATCH 2/8] test(longhaul): freeze retention pruning on detected data loss The pruner keeps only the most recent documents per writer, which means a corrupt (checksum-mismatch) document would eventually be deleted ~55h after detection, destroying the evidence needed to debug a real durability failure. Freeze pruning permanently the first time the verifier reports any data loss (gap or checksum mismatch). The run is already FAIL at that point, so bounding disk no longer matters; instead the collection is left intact so the corrupt document and its neighbours stay on disk for post-mortem. HasDataLoss() is monotonic within a run, so once frozen the pruner stays frozen, and the "pruning frozen" warning is journaled exactly once. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: badfdbf1-0fe9-43da-9921-854304654217 Signed-off-by: Wenting Wu --- test/longhaul/workload/pruner.go | 22 ++++++++++++++++ test/longhaul/workload/pruner_test.go | 37 +++++++++++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/test/longhaul/workload/pruner.go b/test/longhaul/workload/pruner.go index 41fa9bb4..421acb8d 100644 --- a/test/longhaul/workload/pruner.go +++ b/test/longhaul/workload/pruner.go @@ -60,6 +60,10 @@ func (m docdbPruneBackend) deleteThrough(ctx context.Context, writerID string, t // every removed document was already scanned, and the verifier's startup // DB-min seed keeps a post-restart scan from misreading the pruned prefix as a // gap. +// +// Pruning freezes permanently once the verifier reports any data loss (gap or +// checksum mismatch): the run is already FAIL, so the collection is left intact +// to preserve the corrupt document and its neighbours for post-mortem debugging. type Pruner struct { writers []*Writer floor floorProvider @@ -67,6 +71,11 @@ type Pruner struct { retainPerWriter int64 metrics *Metrics journal *journal.Journal + + // frozen is set once a data-loss failure freezes pruning, so the + // "pruning frozen" warning is logged exactly once. Only touched from the + // pruner goroutine. + frozen bool } // NewPruner constructs a Pruner. retainPerWriter must be > 0; callers gate on @@ -101,6 +110,19 @@ func (p *Pruner) Run(ctx context.Context) { } func (p *Pruner) pruneAll(ctx context.Context) { + // Freeze pruning the moment the durability oracle detects a failure. Once a + // gap or checksum mismatch is seen the run is already FAIL, so bounding disk + // no longer matters — instead we preserve the collection exactly as it was + // at the failure so the corrupt document (and its neighbours) stay on disk + // for post-mortem debugging. HasDataLoss() is monotonic (the counters never + // reset within a run), so once frozen the pruner stays frozen. + if p.metrics.Snapshot().HasDataLoss() { + if !p.frozen { + p.frozen = true + p.journal.Warn("pruner", "data loss detected; pruning frozen to preserve evidence for debugging") + } + return + } for _, w := range p.writers { p.pruneWriter(ctx, w.id) } diff --git a/test/longhaul/workload/pruner_test.go b/test/longhaul/workload/pruner_test.go index 4f3470a6..1603b8ed 100644 --- a/test/longhaul/workload/pruner_test.go +++ b/test/longhaul/workload/pruner_test.go @@ -141,4 +141,41 @@ var _ = Describe("Pruner.pruneAll", func() { )) Expect(m.DocsPruned.Load()).To(Equal(int64(30))) }) + + It("freezes pruning when a checksum error has been detected (preserve evidence)", func() { + b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 10, "w001": 20}} + m := NewMetrics() + p := newTestPruner(b, fakeFloor{"w000": 5_000, "w001": 9_000}, 1_000, m) + m.ChecksumErrors.Add(1) + + p.pruneAll(context.Background()) + + Expect(b.calls).To(BeEmpty(), "no deletes must be issued once data loss is detected") + Expect(m.DocsPruned.Load()).To(BeZero()) + Expect(p.frozen).To(BeTrue()) + }) + + It("freezes pruning when a gap has been detected", func() { + b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 10}} + m := NewMetrics() + p := newTestPruner(b, fakeFloor{"w000": 5_000}, 1_000, m) + m.VerifyGapsDetected.Add(1) + + p.pruneAll(context.Background()) + + Expect(b.calls).To(BeEmpty()) + }) + + It("stays frozen on later cycles even though the failure counters are unchanged", func() { + b := &fakePruneBackend{perWriterDeleted: map[string]int64{"w000": 10}} + m := NewMetrics() + p := newTestPruner(b, fakeFloor{"w000": 5_000}, 1_000, m) + m.ChecksumErrors.Add(1) + + p.pruneAll(context.Background()) + p.pruneAll(context.Background()) + + Expect(b.calls).To(BeEmpty()) + Expect(p.frozen).To(BeTrue()) + }) }) From 155def9cb14708d0c7b4df77b290133a34668737 Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Wed, 15 Jul 2026 13:03:30 -0400 Subject: [PATCH 3/8] test(longhaul): drop Docs Pruned row from report Docs pruned is operational, not part of the durability verdict, and is already redundant with the pruner's per-prune and freeze journal events that appear in the report's event log. Remove the report row to keep the report focused on pass/fail. The DocsPruned counter (incremented by the pruner and journaled) stays as the source of truth. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: badfdbf1-0fe9-43da-9921-854304654217 Signed-off-by: Wenting Wu --- test/longhaul/report/report.go | 1 - 1 file changed, 1 deletion(-) diff --git a/test/longhaul/report/report.go b/test/longhaul/report/report.go index d8c984d8..e7316a7f 100644 --- a/test/longhaul/report/report.go +++ b/test/longhaul/report/report.go @@ -88,7 +88,6 @@ func GenerateMarkdown(s Summary) string { fmt.Fprintf(&b, "| Verify Passes | %d |\n", s.Metrics.VerifyPasses) fmt.Fprintf(&b, "| Gaps Detected | %d |\n", s.Metrics.GapsDetected) fmt.Fprintf(&b, "| Checksum Errors | %d |\n", s.Metrics.ChecksumErrors) - fmt.Fprintf(&b, "| Docs Pruned | %d |\n", s.Metrics.DocsPruned) b.WriteString("\n") // Disruption Windows From a41793f329a2307bb00c395fed63057bbc6fdac2 Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Wed, 15 Jul 2026 13:04:49 -0400 Subject: [PATCH 4/8] test(longhaul): drop unused DocsPruned snapshot field MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With the Docs Pruned report row gone, MetricsSnapshot.DocsPruned had no reader. Remove the field and its copy in Snapshot(). The Metrics.DocsPruned atomic counter stays — the pruner increments it and journals each prune. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: badfdbf1-0fe9-43da-9921-854304654217 Signed-off-by: Wenting Wu --- test/longhaul/workload/metrics.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/test/longhaul/workload/metrics.go b/test/longhaul/workload/metrics.go index 9d05e9ac..cbbb6c27 100644 --- a/test/longhaul/workload/metrics.go +++ b/test/longhaul/workload/metrics.go @@ -68,7 +68,6 @@ type MetricsSnapshot struct { VerifyPasses int64 GapsDetected int64 ChecksumErrors int64 - DocsPruned int64 Elapsed time.Duration } @@ -81,7 +80,6 @@ func (m *Metrics) Snapshot() MetricsSnapshot { VerifyPasses: m.VerifyPasses.Load(), GapsDetected: m.VerifyGapsDetected.Load(), ChecksumErrors: m.ChecksumErrors.Load(), - DocsPruned: m.DocsPruned.Load(), Elapsed: time.Since(m.StartTime), } } From 77a4ea3a88d67126a7095f6b627c87c0f2cddba2 Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Wed, 15 Jul 2026 13:07:25 -0400 Subject: [PATCH 5/8] refactor: drop unused DocsPruned counter from longhaul metrics The Metrics.DocsPruned atomic counter had no production reader after the report row and snapshot field were removed; the pruner already journals each prune (and freeze) event, which is the real observable signal. Remove the counter and rewire pruner tests to assert on delete behavior (backend calls) instead of the counter. Pruner.metrics is retained for the freeze-on-failure check (Snapshot().HasDataLoss()). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: badfdbf1-0fe9-43da-9921-854304654217 Signed-off-by: Wenting Wu --- test/longhaul/workload/metrics.go | 5 ----- test/longhaul/workload/pruner.go | 1 - test/longhaul/workload/pruner_test.go | 7 +------ 3 files changed, 1 insertion(+), 12 deletions(-) diff --git a/test/longhaul/workload/metrics.go b/test/longhaul/workload/metrics.go index cbbb6c27..64c1a416 100644 --- a/test/longhaul/workload/metrics.go +++ b/test/longhaul/workload/metrics.go @@ -43,11 +43,6 @@ type Metrics struct { // recomputed value. Non-zero => FAIL with reason "data loss". ChecksumErrors atomic.Int64 - // DocsPruned counts documents deleted by the retention pruner. Purely - // observational — pruning removes only already-verified history and never - // affects the durability verdict. - DocsPruned atomic.Int64 - // StartTime is when this Metrics was constructed; resets on pod restart. StartTime time.Time } diff --git a/test/longhaul/workload/pruner.go b/test/longhaul/workload/pruner.go index 421acb8d..71fee7d0 100644 --- a/test/longhaul/workload/pruner.go +++ b/test/longhaul/workload/pruner.go @@ -146,7 +146,6 @@ func (p *Pruner) pruneWriter(ctx context.Context, writerID string) { return } if deleted > 0 { - p.metrics.DocsPruned.Add(deleted) p.journal.Info("pruner", fmt.Sprintf("pruned %d docs for writer %s (seq <= %d)", deleted, writerID, throughSeq)) } } diff --git a/test/longhaul/workload/pruner_test.go b/test/longhaul/workload/pruner_test.go index 1603b8ed..9eb9bc68 100644 --- a/test/longhaul/workload/pruner_test.go +++ b/test/longhaul/workload/pruner_test.go @@ -64,7 +64,6 @@ var _ = Describe("Pruner.pruneWriter", func() { Expect(b.calls).To(HaveLen(1)) Expect(b.calls[0]).To(Equal(pruneCall{writerID: "w000", throughSeq: 8_000})) - Expect(m.DocsPruned.Load()).To(Equal(int64(500))) }) It("does not delete when the floor is below the retention window", func() { @@ -76,7 +75,6 @@ var _ = Describe("Pruner.pruneWriter", func() { p.pruneWriter(context.Background(), "w000") Expect(b.calls).To(BeEmpty()) - Expect(m.DocsPruned.Load()).To(BeZero()) }) It("does not delete when the floor is exactly at the retention window", func() { @@ -102,7 +100,7 @@ var _ = Describe("Pruner.pruneWriter", func() { Expect(b.calls[0].throughSeq).To(Equal(int64(1))) }) - It("does not count deletions or advance metrics when the backend errors", func() { + It("swallows backend errors without issuing further deletes", func() { b := &fakePruneBackend{err: errors.New("delete failed")} m := NewMetrics() p := newTestPruner(b, fakeFloor{"w000": 10_000}, 2_000, m) @@ -110,7 +108,6 @@ var _ = Describe("Pruner.pruneWriter", func() { p.pruneWriter(context.Background(), "w000") Expect(b.calls).To(HaveLen(1)) - Expect(m.DocsPruned.Load()).To(BeZero()) }) It("never deletes at or above the confirmed floor (safety invariant)", func() { @@ -139,7 +136,6 @@ var _ = Describe("Pruner.pruneAll", func() { pruneCall{writerID: "w000", throughSeq: 4_000}, pruneCall{writerID: "w001", throughSeq: 8_000}, )) - Expect(m.DocsPruned.Load()).To(Equal(int64(30))) }) It("freezes pruning when a checksum error has been detected (preserve evidence)", func() { @@ -151,7 +147,6 @@ var _ = Describe("Pruner.pruneAll", func() { p.pruneAll(context.Background()) Expect(b.calls).To(BeEmpty(), "no deletes must be issued once data loss is detected") - Expect(m.DocsPruned.Load()).To(BeZero()) Expect(p.frozen).To(BeTrue()) }) From 132766b405ce5d9a16e7260328406b277582162f Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Wed, 15 Jul 2026 13:28:06 -0400 Subject: [PATCH 6/8] docs: document LONGHAUL_RETAIN_PER_WRITER in longhaul README Add the retention-pruner knob to the env-var reference table so the new config option is discoverable alongside the others. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: badfdbf1-0fe9-43da-9921-854304654217 Signed-off-by: Wenting Wu --- test/longhaul/README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/test/longhaul/README.md b/test/longhaul/README.md index d87f153f..2333b562 100644 --- a/test/longhaul/README.md +++ b/test/longhaul/README.md @@ -133,6 +133,7 @@ All configuration is via environment variables. | `LONGHAUL_MAX_INSTANCES` | No | `3` | Maximum `spec.instancesPerNode` for scale-up operations (CRD upper bound: 3). | | `LONGHAUL_REPORT_INTERVAL` | No | `1h` | How often to write checkpoint reports to ConfigMap. | | `LONGHAUL_RESET_DATA` | No | `false` | If `true`, drop the workload collection on startup. Off by default so a Deployment pod restart preserves durability history. | +| `LONGHAUL_RETAIN_PER_WRITER` | No | `2000000` | Retention window: most-recent verified documents kept per writer before the pruner deletes older ones, bounding disk usage. `0` disables pruning (unbounded growth). | ## CI Safety From 6bbd118f79f00db8a6d8b4ea414c7e0a156dd757 Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Wed, 15 Jul 2026 13:38:10 -0400 Subject: [PATCH 7/8] chore: trim duplicate retain detail from pruner start log main.go and pruner.go both logged the retain count on startup. Keep the detailed line in the pruner's own Run (paired with its "pruner stopped"), and make the main-scoped line bare to match the sibling verifier log. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: badfdbf1-0fe9-43da-9921-854304654217 Signed-off-by: Wenting Wu --- test/longhaul/cmd/longhaul/main.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/longhaul/cmd/longhaul/main.go b/test/longhaul/cmd/longhaul/main.go index 3fc3ab24..54401385 100644 --- a/test/longhaul/cmd/longhaul/main.go +++ b/test/longhaul/cmd/longhaul/main.go @@ -141,7 +141,7 @@ func run(cfg config.Config) int { // verdict. Disabled when RetainPerWriter == 0. if cfg.RetainPerWriter > 0 { workload.StartPruner(ctx, db.Collection(workload.CollectionName), writers, verifier, cfg.RetainPerWriter, metrics, j) - j.Info("main", fmt.Sprintf("retention pruner started (retain %d docs/writer)", cfg.RetainPerWriter)) + j.Info("main", "retention pruner started") } else { j.Info("main", "retention pruning disabled (LONGHAUL_RETAIN_PER_WRITER=0)") } From 11a91a9dfe23fb2b2c62a611e23976ac74222fe3 Mon Sep 17 00:00:00 2001 From: Wenting Wu Date: Tue, 21 Jul 2026 16:01:37 -0400 Subject: [PATCH 8/8] fix(longhaul): retry min-seq seed on transient error, add seed tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The one-time startup min-seq seed marked the writer seeded *before* the minSeq query ran, so a transient lookup error (network blip, majority-read timeout) permanently skipped the seed. The next scan then ran from seq 1 against a pruned collection, turning the surviving prefix into one giant false gap -> HasDataLoss() flips to FAIL for the rest of the run (which also freezes the pruner). This is exactly the restart-after-pruning path this feature introduces. Extract the seed decision into a pure seedExpectedSeq helper that only reports success when the lookup succeeds, and mark the writer seeded only on success so a transient error retries next cycle. An empty collection (minSeq returns (0, nil)) still counts as success. Add unit tests for the seed: advances past the pruned prefix on restart, never rewinds when the minimum is at/below the resume point, treats an empty collection as success, and — the regression guard — does NOT mark seeded on a transient error and retries successfully next cycle. Signed-off-by: Wenting Wu Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: badfdbf1-0fe9-43da-9921-854304654217 --- test/longhaul/workload/verifier.go | 37 +++++++++++-- test/longhaul/workload/verifier_test.go | 72 ++++++++++++++++++++++++- 2 files changed, 102 insertions(+), 7 deletions(-) diff --git a/test/longhaul/workload/verifier.go b/test/longhaul/workload/verifier.go index 177ede75..9f19dc09 100644 --- a/test/longhaul/workload/verifier.go +++ b/test/longhaul/workload/verifier.go @@ -246,11 +246,19 @@ func (v *Verifier) verifyWriter(ctx context.Context, w *Writer) { // are still detected. Done once per writer; steady-state cycles never // re-read below nextSeq. if !v.seeded[writerID] { - v.seeded[writerID] = true - if minSeq, err := v.minSeq(ctx, writerID); err != nil { - v.journal.Warn("verifier", fmt.Sprintf("min-seq seed failed for writer %s: %v", writerID, err)) - } else if minSeq > expectedSeq { - expectedSeq = minSeq + // Only mark seeded on success: a transient minSeq error (network blip, + // majority-read timeout) must retry next cycle, otherwise the seed is + // skipped forever and the very next scan runs from seq 1 against a + // pruned collection — turning the surviving prefix into one giant false + // gap. minSeq returning (0, nil) for an empty collection is a success. + newExpected, ok, err := seedExpectedSeq(expectedSeq, func() (int64, error) { + return v.minSeq(ctx, writerID) + }) + if err != nil { + v.journal.Warn("verifier", fmt.Sprintf("min-seq seed failed for writer %s (will retry): %v", writerID, err)) + } else { + expectedSeq = newExpected + v.seeded[writerID] = ok } } @@ -285,6 +293,25 @@ func (v *Verifier) publishFloor(writerID string, floor int64) { } } +// seedExpectedSeq performs the one-time startup min-seq seed decision for a +// single writer. lookup resolves the collection's current minimum seq for that +// writer (v.minSeq in production). It returns the possibly-advanced expectedSeq +// and whether the seed succeeded; the caller must only mark the writer seeded +// when ok is true, so a transient lookup error retries on the next cycle rather +// than being skipped forever (which would scan a pruned collection from seq 1 +// and register a false gap). An empty collection — lookup returning (0, nil) — +// counts as success and leaves expectedSeq unchanged. +func seedExpectedSeq(expectedSeq int64, lookup func() (int64, error)) (int64, bool, error) { + minSeq, err := lookup() + if err != nil { + return expectedSeq, false, err + } + if minSeq > expectedSeq { + expectedSeq = minSeq + } + return expectedSeq, true, nil +} + // minSeq returns the lowest seq currently stored for writerID, or 0 if the // writer has no documents. func (v *Verifier) minSeq(ctx context.Context, writerID string) (int64, error) { diff --git a/test/longhaul/workload/verifier_test.go b/test/longhaul/workload/verifier_test.go index 97ac9063..431fcfd0 100644 --- a/test/longhaul/workload/verifier_test.go +++ b/test/longhaul/workload/verifier_test.go @@ -4,6 +4,7 @@ package workload import ( + "errors" "sync/atomic" . "github.com/onsi/ginkgo/v2" @@ -65,10 +66,77 @@ var _ = Describe("Verifier", func() { "unregistered writers read as 0 so the pruner never deletes for them") }) }) + + Describe("startup min-seq seed (seedExpectedSeq)", func() { + // The seed is the restart-after-pruning safety net: retention pruning + // deletes already-verified docs below the floor, so a restarted pod + // (in-memory nextSeq reset to 1) must advance expectedSeq to the + // collection's surviving minimum, or the pruned prefix reads as one + // giant false gap. These tests lock the two invariants that matter: + // (1) a successful seed advances past the pruned prefix, and + // (2) a transient lookup error does NOT mark the writer seeded, so it + // retries next cycle instead of being skipped forever. + + It("advances expectedSeq to the surviving minimum after pruning (restart happy path)", func() { + // Fresh verifier => expectedSeq starts at 1. The collection was + // pruned so its lowest surviving seq is 5_000. + got, ok, err := seedExpectedSeq(1, func() (int64, error) { return 5_000, nil }) + Expect(err).NotTo(HaveOccurred()) + Expect(ok).To(BeTrue()) + Expect(got).To(Equal(int64(5_000)), + "scan must start at the surviving minimum, not seq 1, so the pruned prefix is not a false gap") + }) + + It("leaves expectedSeq unchanged when the minimum is at or below it", func() { + // Un-pruned / already-ahead resume point: min-seq must not rewind us. + got, ok, err := seedExpectedSeq(200, func() (int64, error) { return 1, nil }) + Expect(err).NotTo(HaveOccurred()) + Expect(ok).To(BeTrue()) + Expect(got).To(Equal(int64(200))) + }) + + It("treats an empty collection (0, nil) as a successful seed", func() { + got, ok, err := seedExpectedSeq(1, func() (int64, error) { return 0, nil }) + Expect(err).NotTo(HaveOccurred()) + Expect(ok).To(BeTrue(), "an empty collection is a valid state, not a retry-worthy error") + Expect(got).To(Equal(int64(1))) + }) + + It("does NOT mark seeded on a transient error, and retries successfully next cycle", func() { + // This is the regression the Major review comment called out: a + // transient minSeq error used to mark the writer seeded, permanently + // skipping the seed and producing a false data-loss FAIL. + calls := 0 + lookup := func() (int64, error) { + calls++ + if calls == 1 { + return 0, errors.New("majority-read timeout") + } + return 5_000, nil + } + + // Simulate the call-site contract: only mark seeded when ok is true. + expectedSeq := int64(1) + seeded := false + + got, ok, err := seedExpectedSeq(expectedSeq, lookup) + Expect(err).To(HaveOccurred()) + Expect(ok).To(BeFalse()) + Expect(got).To(Equal(int64(1)), "a failed seed must not move expectedSeq") + seeded = seeded || ok + Expect(seeded).To(BeFalse(), "writer must remain un-seeded so it retries next cycle") + + // Next cycle: the seed is retried (because still not seeded) and now + // succeeds, advancing past the pruned prefix. + got, ok, err = seedExpectedSeq(got, lookup) + Expect(err).NotTo(HaveOccurred()) + Expect(ok).To(BeTrue()) + Expect(got).To(Equal(int64(5_000))) + Expect(calls).To(Equal(2), "the transient error must trigger exactly one retry") + }) + }) }) -// makeDoc constructs a valid WriteDocument whose checksum matches; tests that -// want a checksum mismatch override Checksum directly. func makeDoc(writerID string, seq int64) WriteDocument { payload := "p" return WriteDocument{