test(longhaul): add retention pruner to bound workload collection disk usage

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
 

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", "retention pruner started")
+	} 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),

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
 }
 

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

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.

test/longhaul/workload/pruner.go

@@ -0,0 +1,158 @@
+// 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.
+//
+// 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
+	backend         pruneBackend
+	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
+// 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) {
+	// 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)
+	}
+}
+
+// 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.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
+}