feat(orchestrator): add AWS S3 seed objects to LocalStack init

internal/config/cloud.go

@@ -77,6 +77,30 @@ type AWSConfig struct {
 	// Subscriptions wires SNS → SQS (RawMessageDelivery=true) so an
 	// SNS-target case is observable by the `sqs` receiver mode.
 	Subscriptions []AWSSubscription `yaml:"subscriptions"`
+
+	// SeedObjects pre-uploads synthetic objects into a declared bucket at
+	// init, before the subject starts — for list-mode / first-run backlog
+	// source cases where the subject must find objects already present rather
+	// than a generator streaming them in during the run.
+	SeedObjects []AWSSeedObjects `yaml:"seed_objects"`
+}
+
+// AWSSeedObjects pre-uploads Objects synthetic objects (Lines lines each,
+// each line prefixed with Marker) under Prefix in Bucket during LocalStack
+// init.
+type AWSSeedObjects struct {
+	// Bucket must be one of the declared Buckets.
+	Bucket string `yaml:"bucket"`
+	// Prefix is the key prefix for seeded objects (default "seed/").
+	Prefix string `yaml:"prefix"`
+	// Objects is the number of objects to create (> 0).
+	Objects int `yaml:"objects"`
+	// Lines is the number of lines per object (> 0).
+	Lines int `yaml:"lines"`
+	// Marker is the per-line content prefix (default "SEED"); validated
+	// against the cloud-name charset so it cannot inject into the init shell
+	// script.
+	Marker string `yaml:"marker"`
 }
 
 // AWSStream declares a Kinesis stream created at init.
@@ -420,6 +444,27 @@ func (tc *TestCase) validateAWS() error {
 			return err
 		}
 	}
+	for _, so := range tc.AWS.SeedObjects {
+		if _, ok := buckets[so.Bucket]; !ok {
+			return fmt.Errorf("case %q: seed_objects references undeclared bucket %q", tc.Name, so.Bucket)
+		}
+		if so.Prefix != "" {
+			if err := validateCloudName(tc.Name, "aws seed prefix", so.Prefix); err != nil {
+				return err
+			}
+		}
+		if so.Marker != "" {
+			if err := validateCloudName(tc.Name, "aws seed marker", so.Marker); err != nil {
+				return err
+			}
+		}
+		if so.Objects <= 0 {
+			return fmt.Errorf("case %q: seed_objects for bucket %q requires objects > 0, got %d", tc.Name, so.Bucket, so.Objects)
+		}
+		if so.Lines <= 0 {
+			return fmt.Errorf("case %q: seed_objects for bucket %q requires lines > 0, got %d", tc.Name, so.Bucket, so.Lines)
+		}
+	}
 	return nil
 }
 

internal/config/cloud_test.go

@@ -219,6 +219,43 @@ func TestAWSConfigDefaults(t *testing.T) {
 	}
 }
 
+func TestValidateAWSSeedObjects(t *testing.T) {
+	base := func(so AWSSeedObjects) *TestCase {
+		return &TestCase{
+			Name:     "seed-case",
+			Type:     "correctness",
+			Duration: "10s",
+			AWS: &AWSConfig{
+				Buckets:     []string{"bench-in"},
+				SeedObjects: []AWSSeedObjects{so},
+			},
+			Receiver:    ReceiverConfig{Mode: "tcp", Listen: ":9001"},
+			Correctness: CorrectnessConfig{},
+		}
+	}
+
+	tests := []struct {
+		name    string
+		so      AWSSeedObjects
+		wantErr bool
+	}{
+		{name: "valid", so: AWSSeedObjects{Bucket: "bench-in", Objects: 10, Lines: 100}},
+		{name: "valid with prefix and marker", so: AWSSeedObjects{Bucket: "bench-in", Prefix: "seed/", Objects: 1, Lines: 1, Marker: "SEED"}},
+		{name: "undeclared bucket", so: AWSSeedObjects{Bucket: "nope", Objects: 1, Lines: 1}, wantErr: true},
+		{name: "zero objects", so: AWSSeedObjects{Bucket: "bench-in", Objects: 0, Lines: 1}, wantErr: true},
+		{name: "zero lines", so: AWSSeedObjects{Bucket: "bench-in", Objects: 1, Lines: 0}, wantErr: true},
+		{name: "injecting marker", so: AWSSeedObjects{Bucket: "bench-in", Objects: 1, Lines: 1, Marker: "a'; rm -rf /"}, wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := base(tt.so).validateAWS()
+			if (err != nil) != tt.wantErr {
+				t.Fatalf("validateAWS() err = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
+
 func TestMinioConfigDefaults(t *testing.T) {
 	m := &MinioConfig{Buckets: []string{"bench-out"}}
 	if got, want := m.ImageOrDefault(), "minio/minio:RELEASE.2025-04-22T22-12-26Z"; got != want {

internal/orchestrator/awsinit.go

@@ -56,6 +56,25 @@ func writeAWSInit(path string, aws *config.AWSConfig) error {
 			"awslocal sns subscribe --topic-arn '%s' --protocol sqs --notification-endpoint '%s' --attributes RawMessageDelivery=true\n",
 			aws.TopicARN(s.Topic), aws.QueueARN(s.Queue))
 	}
+	for _, so := range aws.SeedObjects {
+		prefix := so.Prefix
+		if prefix == "" {
+			prefix = "seed/"
+		}
+		marker := so.Marker
+		if marker == "" {
+			marker = "SEED"
+		}
+		// Build each object body with awk (busybox ships it), then upload —
+		// so the objects exist before LocalStack reports init "completed" and
+		// the subject's depends_on gate releases. All values are int-formatted
+		// or charset-validated, so single-quoting is defense in depth.
+		fmt.Fprintf(&b, "i=0; while [ \"$i\" -lt %d ]; do\n", so.Objects)
+		fmt.Fprintf(&b, "  awk -v o=\"$i\" 'BEGIN{for(l=0;l<%d;l++) printf \"%s-OBJ%%d-LINE%%d\\n\", o, l}' > /tmp/pb-seed-obj\n", so.Lines, marker)
+		fmt.Fprintf(&b, "  awslocal s3 cp /tmp/pb-seed-obj 's3://%s/%sobj-'\"$i\"'.log'\n", so.Bucket, prefix)
+		b.WriteString("  i=$((i+1))\n")
+		b.WriteString("done\n")
+	}
 
 	b.WriteString("echo 'pipebench aws init complete'\n")
 	return os.WriteFile(path, []byte(b.String()), 0o755)

internal/orchestrator/compose_render_test.go

@@ -632,6 +632,53 @@ func TestComposeRendersAWS(t *testing.T) {
 	mustContain(t, string(script), `"QueueArn":"arn:aws:sqs:us-east-1:000000000000:bench-events"`)
 }
 
+// TestComposeRendersAWSSeedObjects verifies a `seed_objects:` entry renders a
+// pre-upload loop into the LocalStack init script (objects exist before the
+// subject starts, for list-mode backlog cases).
+func TestComposeRendersAWSSeedObjects(t *testing.T) {
+	tc := &config.TestCase{
+		Name:     "aws-seed",
+		Type:     "correctness",
+		Duration: "10s",
+		AWS: &config.AWSConfig{
+			Buckets: []string{"bench-in"},
+			SeedObjects: []config.AWSSeedObjects{
+				{Bucket: "bench-in", Prefix: "seed/", Objects: 10, Lines: 100, Marker: "SEED"},
+			},
+		},
+		Receiver:    config.ReceiverConfig{Mode: "tcp", Listen: ":9001"},
+		Correctness: config.CorrectnessConfig{MinReceived: 1000},
+	}
+	if err := tc.Validate(); err != nil {
+		t.Fatalf("validate: %v", err)
+	}
+	subj := config.Subject{Name: "vmetric", Image: "vmetric/director", Version: "2.0.3", ConfigPath: "/config.yml"}
+	tmp, err := os.MkdirTemp("", "compose-seed-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(tmp)
+	composePath := filepath.Join(tmp, "compose.yaml")
+	cfg := RunConfig{
+		TestCase: tc, Subject: subj, ConfigName: "default",
+		ConfigSrcPath: composePath, TmpDir: tmp,
+		GeneratorImage: "img-gen", ReceiverImage: "img-recv", CollectorImage: "img-coll",
+		ReceiverHostPort: 19001,
+	}
+	if err := writeCompose(composePath, cfg); err != nil {
+		t.Fatalf("writeCompose: %v", err)
+	}
+	script, err := os.ReadFile(filepath.Join(tmp, "aws-init.sh"))
+	if err != nil {
+		t.Fatalf("aws-init.sh not written: %v", err)
+	}
+	mustContain(t, string(script), "awslocal s3 mb 's3://bench-in'")
+	mustContain(t, string(script), `while [ "$i" -lt 10 ]`)
+	mustContain(t, string(script), "for(l=0;l<100;l++)")
+	mustContain(t, string(script), "SEED-OBJ%d-LINE%d")
+	mustContain(t, string(script), "'s3://bench-in/seed/obj-'\"$i\"'.log'")
+}
+
 // TestComposeRendersAzure verifies an `azure:` case renders the Azurite
 // service plus the one-shot azure-init (receiver image), gates the subject on
 // init completion, and injects the connection string where needed.

internal/runner/runner.go

@@ -189,6 +189,14 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 	if tc.Type == "kafka_inflight_crash_correctness" {
 		return r.runKafkaInflightCrash(tc, subject)
 	}
+	// Object-storage in-flight crash: same receiver-up mid-delivery flow as the
+	// kafka in-flight crash, for a poll-mode source (S3/Azure bucket). The
+	// bucket is the durable store, decoupled from the subject, so the generator
+	// keeps uploading across the SIGKILL+restart. Verifies no loss; duplicates
+	// (crash-resistance replay + cursor re-list) are reported, not failed.
+	if tc.Type == "persistence_inflight_crash_correctness" {
+		return r.runInflightCrashCorrectness(tc, subject)
+	}
 	// Kafka offset-commit restart: receiver stays UP, ALL records are
 	// delivered cleanly, then the subject is restarted gracefully. A
 	// consumer whose offset commits actually persist resumes from the
@@ -832,15 +840,20 @@ func (r *Runner) Run(tc *config.TestCase, subject config.Subject) (results.RunRe
 		// Re-deriving pass/fail from line-count loss + a strict over-delivery
 		// cap here would wrongly flip a valid allow_overdelivery verifier pass.
 		lossOK := lossPct <= tc.Correctness.ExpectedLossPct
-		// Kafka consumption is at-least-once: the consumer may re-deliver a
-		// fetch batch on its initial group join / rebalance, so allow bounded
-		// over-delivery (Correctness.MaxOverDeliveryPct, default 0 = exact).
-		// Non-kafka correctness stays strict.
+		// Over-delivery policy, in precedence order:
+		//   - allow_overdelivery: the source is at-least-once by design (e.g. the
+		//     list-poll since-cursor re-lists the tip object every idle cycle), so
+		//     any over-delivery is accepted; only loss fails. This mirrors the
+		//     carve-out the in-flight-crash / restart handlers already apply.
+		//   - kafka: consumption is at-least-once — the consumer may re-deliver a
+		//     fetch batch on its initial group join / rebalance, so allow bounded
+		//     over-delivery (Correctness.MaxOverDeliveryPct, default 0 = exact).
+		//   - otherwise: strict exact count.
 		overCap := expectedOut
 		if tc.IsKafkaType() {
 			overCap += int64(float64(expectedOut) * tc.Correctness.MaxOverDeliveryPct / 100.0)
 		}
-		overOK := recvMetrics.LinesReceived <= overCap
+		overOK := tc.Correctness.AllowOverDelivery || recvMetrics.LinesReceived <= overCap
 		recvOK := result.Passed == nil || *result.Passed
 
 		var failReasons []string
@@ -1580,12 +1593,13 @@ func (r *Runner) runPersistenceShutdownCorrectness(tc *config.TestCase, subject
 	return result, nil
 }
 
-// midDeliveryFlow parameterizes the shared kafka correctness driver
-// (runKafkaMidDeliveryAction): produce to the broker with the receiver live,
-// fire one disruptive action once the receiver has seen ~half of total_lines,
-// then drain and assert no loss (over-delivery from at-least-once recovery is
-// reported, not failed). The action is the only thing that varies between the
-// flows — an in-flight subject crash, a broker cert rotation, etc.
+// midDeliveryFlow parameterizes the shared mid-delivery correctness driver
+// (runMidDeliveryAction): produce to a durable source (Kafka broker, S3/Azure
+// bucket) with the receiver live, fire one disruptive action once the receiver
+// has seen ~half of total_lines, then drain and assert no loss (over-delivery
+// from at-least-once recovery is reported, not failed). The action is the only
+// thing that varies between the flows — an in-flight subject crash, a broker
+// cert rotation, etc.
 type midDeliveryFlow struct {
 	// verdictLabel names the flow in the PASS/FAIL line, e.g.
 	// "kafka cert rotation correctness".
@@ -1609,12 +1623,15 @@ type midDeliveryFlow struct {
 	action func(orch orchestrator.Orchestrator) error
 }
 
-// runKafkaMidDeliveryAction is the shared driver behind the kafka in-flight
-// crash and cert-rotation flows: both bring everything up with the receiver
-// live, wait until the receiver has seen half the records, fire one disruptive
-// action, then drain and apply the same no-loss / at-least-once verdict. Only
-// the action (and a little setup/labelling) differs — see midDeliveryFlow.
-func (r *Runner) runKafkaMidDeliveryAction(tc *config.TestCase, subject config.Subject, f midDeliveryFlow) (results.RunResult, error) {
+// runMidDeliveryAction is the shared driver behind the receiver-up mid-delivery
+// flows (kafka in-flight crash, kafka cert rotation, and object-storage
+// in-flight crash): all bring everything up with the receiver live, wait until
+// the receiver has seen half the records, fire one disruptive action, then
+// drain and apply the same no-loss / at-least-once verdict. Only the action
+// (and a little setup/labelling) differs — see midDeliveryFlow. The source is
+// decoupled from the subject (broker or bucket), so the generator keeps
+// producing across a subject restart.
+func (r *Runner) runMidDeliveryAction(tc *config.TestCase, subject config.Subject, f midDeliveryFlow) (results.RunResult, error) {
 	configName := r.opts.ConfigName
 	subject = r.applySubjectOverrides(subject)
 
@@ -1878,7 +1895,7 @@ func rotateAndReload(orch orchestrator.Orchestrator, service string, rotate func
 // offset-committed are re-consumed on restart. Verdict: no loss; duplicates are
 // reported, not failed.
 func (r *Runner) runKafkaInflightCrash(tc *config.TestCase, subject config.Subject) (results.RunResult, error) {
-	return r.runKafkaMidDeliveryAction(tc, subject, midDeliveryFlow{
+	return r.runMidDeliveryAction(tc, subject, midDeliveryFlow{
 		verdictLabel:  "kafka in-flight crash correctness",
 		actionLog:     "SIGKILL subject (no graceful shutdown), then restart",
 		overDelivNote: "expected for a mid-delivery crash",
@@ -1899,6 +1916,35 @@ func (r *Runner) runKafkaInflightCrash(tc *config.TestCase, subject config.Subje
 	})
 }
 
+// runInflightCrashCorrectness SIGKILLs the subject WHILE it is actively
+// delivering to a live receiver, then restarts it — the receiver-up
+// mid-delivery worst case for any source whose store is decoupled from the
+// subject (e.g. the S3/Azure bucket a poll-mode device lists). The since-cursor
+// advances at cycle end with no per-object delivery commit, so recovery must
+// still lose nothing; duplicates from the crash-resistance replay + re-list are
+// reported, not failed.
+func (r *Runner) runInflightCrashCorrectness(tc *config.TestCase, subject config.Subject) (results.RunResult, error) {
+	return r.runMidDeliveryAction(tc, subject, midDeliveryFlow{
+		verdictLabel:  "in-flight crash correctness",
+		actionLog:     "SIGKILL subject (no graceful shutdown), then restart",
+		overDelivNote: "expected for a mid-delivery crash",
+		totalLinesErr: "persistence_inflight_crash_correctness requires generator.total_lines > 0",
+		action: func(orch orchestrator.Orchestrator) error {
+			if err := orch.KillServices("subject"); err != nil {
+				return fmt.Errorf("killing subject: %w", err)
+			}
+			// Settle before the subject restarts and re-lists from its cursor.
+			if err := sleepCtx(r.ctx, 3*time.Second); err != nil {
+				return fmt.Errorf("interrupted: %w", err)
+			}
+			if err := orch.UpServices("subject"); err != nil {
+				return fmt.Errorf("restarting subject: %w", err)
+			}
+			return nil
+		},
+	})
+}
+
 // runKafkaCertRotation verifies the subject's broker-cert handling over mTLS in
 // TWO halves, so the run fails if EITHER property breaks:
 //
@@ -1919,7 +1965,7 @@ func (r *Runner) runKafkaCertRotation(tc *config.TestCase, subject config.Subjec
 	// and prepare runs before action, so the capture is well-ordered.
 	var certsDir string
 	hosts := []string{"subject", "localhost", "redpanda"}
-	return r.runKafkaMidDeliveryAction(tc, subject, midDeliveryFlow{
+	return r.runMidDeliveryAction(tc, subject, midDeliveryFlow{
 		verdictLabel:  "kafka cert rotation correctness",
 		actionLog:     "rotating broker cert to an UNTRUSTED CA (must be rejected), then back to a trusted cert",
 		overDelivNote: "expected across the broker reconnects",
@@ -7178,7 +7224,7 @@ func (r *Runner) runSyslogVaultCertRotation(tc *config.TestCase, subject config.
 	hosts := []string{"subject", "localhost"}
 	var certsDir string
 
-	return r.runKafkaMidDeliveryAction(tc, subject, midDeliveryFlow{
+	return r.runMidDeliveryAction(tc, subject, midDeliveryFlow{
 		verdictLabel:  "syslog TLS vault cert rotation correctness",
 		actionLog:     "rotating syslog server cert to UNTRUSTED CA (generator TLS must fail), then restoring trusted cert",
 		overDelivNote: "expected after the trusted cert is restored and the generator reconnects",
@@ -7222,7 +7268,7 @@ func (r *Runner) runSyslogVaultCertRotation(tc *config.TestCase, subject config.
 
 			mount := tc.Vault.MountOrDefault()
 			token := tc.Vault.TokenOrDefault()
-			// Receiver metrics port is already forwarded by runKafkaMidDeliveryAction.
+			// Receiver metrics port is already forwarded by runMidDeliveryAction.
 			metricsPort := orch.ReceiverMetricsPorts()["default"]
 
 			// ---- Phase 1: UNTRUSTED cert — generator TLS must fail ----