feat(harness): S3 seed/verdict knobs for windowed polls

containers/receiver/main.go

@@ -27,11 +27,12 @@ type config struct {
 	Timeout     time.Duration
 
 	// Correctness validation (all optional, off by default)
-	ValidateDedup     bool
-	ValidateContent   bool   // O(1) per line, no heap map — safe for high-volume tests
-	ExpectedLines     int64  // 0 = don't check
-	RequiredSubstring string // empty = don't check; protocol-agnostic decode check
-	ValidateJSON      bool   // every emitted line must parse as JSON
+	ValidateDedup      bool
+	ValidateContent    bool   // O(1) per line, no heap map — safe for high-volume tests
+	ExpectedLines      int64  // 0 = don't check
+	RequiredSubstring  string // empty = don't check; protocol-agnostic decode check
+	ForbiddenSubstring string // empty = don't check; the verdict fails if any line contains it
+	ValidateJSON       bool   // every emitted line must parse as JSON
 
 	// RecordArrivalTimes, when true, has the receiver capture the
 	// wall-clock arrival nanosecond of every record into an in-memory
@@ -207,6 +208,12 @@ type validator struct {
 	missingSubstr     atomic.Int64
 	missingSubstrSamp []string
 
+	// Forbidden-substring check: no line may contain the configured token
+	// (e.g. the marker of objects a lookback / end_date / prefix window must
+	// exclude). Counts + up to 10 sample violators.
+	forbiddenHit     atomic.Int64
+	forbiddenHitSamp []string
+
 	// JSON-validity check: every emitted line must parse as a JSON value.
 	// Enabled by ValidateJSON. Without this, a JSON test could be passed
 	// by a subject that emits the right line count of garbage. Counts +
@@ -288,6 +295,23 @@ func (v *validator) recordLine(line []byte, cfg config) {
 		}
 	}
 
+	// Negative content check: a line carrying the forbidden token proves an
+	// exclusion (lookback / end_date / prefix pruning) was NOT enforced.
+	if cfg.ForbiddenSubstring != "" {
+		if bytes.Contains(line, []byte(cfg.ForbiddenSubstring)) {
+			v.forbiddenHit.Add(1)
+			v.mu.Lock()
+			if len(v.forbiddenHitSamp) < 10 {
+				s := string(line)
+				if len(s) > 120 {
+					s = s[:120] + "…"
+				}
+				v.forbiddenHitSamp = append(v.forbiddenHitSamp, s)
+			}
+			v.mu.Unlock()
+		}
+	}
+
 	// JSON-validity check: every line must parse as JSON. json.Valid is
 	// streaming-friendly (no allocation for the parse tree) so it's cheap
 	// enough to run per-line at multi-million lines/s.
@@ -484,6 +508,18 @@ func (v *validator) validate(cfg config, totalLines int64) (bool, []string) {
 		}
 	}
 
+	// Forbidden-substring — no emitted line may carry the configured token.
+	if cfg.ForbiddenSubstring != "" {
+		if m := v.forbiddenHit.Load(); m > 0 {
+			msg := fmt.Sprintf("forbidden substring %q present in %d line(s)", cfg.ForbiddenSubstring, m)
+			if len(v.forbiddenHitSamp) > 0 {
+				msg += "; samples: " + strings.Join(v.forbiddenHitSamp, " | ")
+			}
+			errors = append(errors, msg)
+			passed = false
+		}
+	}
+
 	return passed, errors
 }
 
@@ -547,7 +583,7 @@ func main() {
 		otlpShard := cnt.newShard()
 		onLine := func(line []byte) {
 			otlpShard.recordLine(int64(len(line)) + 1)
-			if cfg.ValidateDedup || cfg.ValidateContent || cfg.RequiredSubstring != "" {
+			if cfg.ValidateDedup || cfg.ValidateContent || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != "" {
 				val.recordLine(line, cfg)
 			}
 		}
@@ -671,7 +707,7 @@ func handleConn(conn net.Conn, shard *connStats, val *validator, cfg config) {
 	// time, inflating the receive window when the sender idles before closing.
 	scanner := bufio.NewScanner(&stampingReader{r: conn, shard: shard})
 	scanner.Buffer(make([]byte, 1024*1024), 1024*1024)
-	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
+	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != ""
 	// Pass the scanner's internal slice directly. shard.recordLine + validator
 	// only read len() and copy via `string(line)` / hash when they need to
 	// keep bytes — they never retain the slice past return. Skipping the
@@ -695,7 +731,7 @@ func handleConn(conn net.Conn, shard *connStats, val *validator, cfg config) {
 func receiveFile(cfg config, cnt *counters, val *validator) error {
 	shard := cnt.newShard()
 	defer shard.finish()
-	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
+	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != ""
 
 	deadline := time.Now().Add(cfg.Timeout + 5*time.Minute) // generous for file tests
 	f, err := os.Open(cfg.Listen)
@@ -742,7 +778,7 @@ func receiveHTTP(cfg config, cnt *counters, val *validator) error {
 	// per-shard ceiling so contention here doesn't matter, and a single
 	// shard keeps the per-request bookkeeping trivial.
 	httpShard := cnt.newShard()
-	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
+	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != ""
 	mux := http.NewServeMux()
 
 	// Generic POST handler — counts every line in the body
@@ -831,7 +867,7 @@ func receiveHTTP(cfg config, cnt *counters, val *validator) error {
 // POST. It handles both plain line-per-request and ES /_bulk NDJSON format.
 func startHTTPDataEndpoint(port string, cnt *counters, val *validator, cfg config) error {
 	httpShard := cnt.newShard()
-	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != ""
+	needsValidation := cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != ""
 	lineCallback := func(line []byte) {
 		httpShard.recordLine(int64(len(line)) + 1)
 		if needsValidation {
@@ -942,7 +978,7 @@ func serveMetrics(port string, cnt *counters, val *validator, cfg config) {
 			"last_received_ns":  lastNs,
 		}
 		// Include correctness data if validation is enabled
-		if cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.ExpectedLines > 0 || cfg.RequiredSubstring != "" {
+		if cfg.ValidateDedup || cfg.ValidateContent || cfg.ValidateJSON || cfg.ExpectedLines > 0 || cfg.RequiredSubstring != "" || cfg.ForbiddenSubstring != "" {
 			passed, errors := val.validate(cfg, totalLines)
 			resp["passed"] = passed
 			if len(errors) > 0 {
@@ -1020,6 +1056,7 @@ func loadConfig() config {
 		ValidateContent:    getEnvBool("RECEIVER_VALIDATE_CONTENT", false),
 		ExpectedLines:      int64(getEnvInt("RECEIVER_EXPECTED_LINES", 0)),
 		RequiredSubstring:  getEnv("RECEIVER_REQUIRED_SUBSTRING", ""),
+		ForbiddenSubstring: getEnv("RECEIVER_FORBIDDEN_SUBSTRING", ""),
 		ValidateJSON:       getEnvBool("RECEIVER_VALIDATE_JSON", false),
 		RecordArrivalTimes: getEnvBool("RECEIVER_RECORD_ARRIVAL_TIMES", false),
 	}

internal/config/case.go

@@ -1204,6 +1204,9 @@ func (tc *TestCase) Validate() error {
 	if tc.Correctness.MaxOverDeliveryPct < 0 {
 		return fmt.Errorf("case %q: max_overdelivery_pct must be non-negative, got %.2f", tc.Name, tc.Correctness.MaxOverDeliveryPct)
 	}
+	if tc.Correctness.MaxReceived < 0 {
+		return fmt.Errorf("case %q: max_received must be non-negative (0 = disabled), got %d", tc.Name, tc.Correctness.MaxReceived)
+	}
 	if err := tc.validateVault(); err != nil {
 		return err
 	}
@@ -2641,6 +2644,11 @@ type CorrectnessConfig struct {
 	// successful decode is proven by the presence of a value the generator
 	// embedded in every record. Empty = check disabled.
 	RequiredSubstring string `yaml:"required_substring"`
+	// ForbiddenSubstring is the inverse of RequiredSubstring: the verdict FAILS
+	// if ANY received line contains it. It proves an exclusion (a lookback or
+	// end_date cutoff, prefix pruning) is enforced, not just that inclusion
+	// works. Generic — any correctness case may use it. Empty = disabled.
+	ForbiddenSubstring string `yaml:"forbidden_substring"`
 	// ValidateJSON, when true, requires every emitted line to parse as a
 	// JSON object. Without this, a subject can pass a JSON-shape test by
 	// truncating to a matching line count or by re-emitting binary garbage —
@@ -2664,6 +2672,12 @@ type CorrectnessConfig struct {
 	// "LinesReceived >= MinReceived" (default 1) AND the receiver didn't flag a
 	// content failure. Ignored when the case has a generator.
 	MinReceived int64 `yaml:"min_received"`
+	// MaxReceived, when > 0, fails the verdict if LinesReceived exceeds it —
+	// independent of expect_failure/loss/over-delivery logic. Use it to prove
+	// an upper bound (e.g. an end_date cutoff) where the exact count is
+	// timing-sensitive and only a ceiling is safe to assert. Generic — any
+	// correctness case may use it. 0 = disabled.
+	MaxReceived int64 `yaml:"max_received"`
 
 	// DrainSeconds extends how long the harness waits for backlog to
 	// arrive after the generator(s) stop. The case still finishes early if

internal/config/cloud.go

@@ -101,8 +101,46 @@ type AWSSeedObjects struct {
 	// against the cloud-name charset so it cannot inject into the init shell
 	// script.
 	Marker string `yaml:"marker"`
+	// DelaySeconds sleeps that many seconds inside the LocalStack init script
+	// immediately before this group's upload loop, giving consecutive seed
+	// groups real wall-clock separation in their S3 LastModified (LocalStack
+	// cannot backdate objects, so back-to-back groups otherwise share one
+	// second). 0 = no delay (default, existing behaviour).
+	DelaySeconds int `yaml:"delay_seconds"`
+	// Extension is the key suffix of every seeded object (default ".log").
+	// Bodies are always plain text, so a compressed-looking extension such
+	// as ".gz" deliberately produces a poison object whose decode fails —
+	// the fixture for retry-cap / skip-on-error cases. Must match
+	// SeedExtensionPattern.
+	Extension string `yaml:"extension"`
 }
 
+// SeedExtensionPattern bounds seed_objects[].extension: a leading dot then
+// lowercase alphanumerics/dots, so it can neither inject into the init shell
+// script nor produce a key the subject's extension dispatch cannot classify.
+var SeedExtensionPattern = regexp.MustCompile(`^\.[a-z0-9.]{1,15}$`)
+
+// TotalSeedDelaySeconds sums seed_objects[].delay_seconds — the wall-clock
+// the LocalStack init hook sleeps before it can report "completed", which the
+// compose healthcheck budget must cover.
+func (a *AWSConfig) TotalSeedDelaySeconds() int {
+	if a == nil {
+		return 0
+	}
+	total := 0
+	for _, so := range a.SeedObjects {
+		if so.DelaySeconds > 0 {
+			total += so.DelaySeconds
+		}
+	}
+	return total
+}
+
+// SeedTodayToken is the one prefix placeholder the harness substitutes itself
+// (UTC YYYY/MM/DD at render time) so a case can seed "today's" partition. It is
+// stripped before the charset check and never reaches the shell unexpanded.
+const SeedTodayToken = "$TODAY"
+
 // AWSStream declares a Kinesis stream created at init.
 type AWSStream struct {
 	Name string `yaml:"name"`
@@ -449,10 +487,15 @@ func (tc *TestCase) validateAWS() error {
 			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 {
+			// Validate the token as the date shape it expands to, so a prefix that
+			// is only "$TODAY" is accepted like "logs/$TODAY/".
+			if err := validateCloudName(tc.Name, "aws seed prefix", strings.ReplaceAll(so.Prefix, SeedTodayToken, "2006/01/02")); err != nil {
 				return err
 			}
 		}
+		if so.DelaySeconds < 0 {
+			return fmt.Errorf("case %q: seed_objects for bucket %q delay_seconds must be non-negative, got %d", tc.Name, so.Bucket, so.DelaySeconds)
+		}
 		if so.Marker != "" {
 			if err := validateCloudName(tc.Name, "aws seed marker", so.Marker); err != nil {
 				return err
@@ -464,6 +507,9 @@ func (tc *TestCase) validateAWS() error {
 		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)
 		}
+		if so.Extension != "" && !SeedExtensionPattern.MatchString(so.Extension) {
+			return fmt.Errorf("case %q: seed_objects for bucket %q extension %q must match %s", tc.Name, so.Bucket, so.Extension, SeedExtensionPattern)
+		}
 	}
 	return nil
 }

internal/config/cloud_test.go

@@ -33,6 +33,56 @@ func TestValidateCloud(t *testing.T) {
 				Generator: GeneratorConfig{Mode: "s3", Target: "http://localstack:4566"},
 			},
 		},
+		{
+			name: "seed group with negative delay",
+			tc: TestCase{
+				Name: "c",
+				AWS: &AWSConfig{
+					Buckets:     []string{"bench-in"},
+					SeedObjects: []AWSSeedObjects{{Bucket: "bench-in", Objects: 1, Lines: 1, DelaySeconds: -1}},
+				},
+			},
+			wantErr: "delay_seconds must be non-negative",
+		},
+		{
+			name: "seed prefix that is only the $TODAY token",
+			tc: TestCase{
+				Name: "x",
+				AWS: &AWSConfig{
+					Buckets:     []string{"bench-in"},
+					SeedObjects: []AWSSeedObjects{{Bucket: "bench-in", Prefix: "$TODAY", Objects: 1, Lines: 1}},
+				},
+			},
+		},
+		{
+			name: "seed prefix with $TODAY token and delay",
+			tc: TestCase{
+				Name: "c",
+				AWS: &AWSConfig{
+					Buckets:     []string{"bench-in"},
+					SeedObjects: []AWSSeedObjects{{Bucket: "bench-in", Prefix: "logs/$TODAY/", Objects: 1, Lines: 1, DelaySeconds: 45}},
+				},
+			},
+		},
+		{
+			name: "seed prefix with other shell token rejected",
+			tc: TestCase{
+				Name: "c",
+				AWS: &AWSConfig{
+					Buckets:     []string{"bench-in"},
+					SeedObjects: []AWSSeedObjects{{Bucket: "bench-in", Prefix: "logs/$HOME/", Objects: 1, Lines: 1}},
+				},
+			},
+			wantErr: "aws seed prefix",
+		},
+		{
+			name: "negative max_received rejected",
+			tc: TestCase{
+				Name:        "c",
+				Correctness: CorrectnessConfig{MaxReceived: -1},
+			},
+			wantErr: "max_received must be non-negative",
+		},
 		{
 			name: "azure block with container",
 			tc: TestCase{

internal/orchestrator/awsinit.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"time"
 
 	"github.com/VirtualMetric/PipeBench/internal/config"
 )
@@ -17,7 +18,10 @@ import (
 // Every name interpolated here passed config.Validate()'s charset check
 // ([a-zA-Z0-9._/-]), so single-quoting is belt-and-braces, not the only
 // injection defense.
-func writeAWSInit(path string, aws *config.AWSConfig) error {
+//
+// now is the harness render instant; it resolves the $TODAY seed-prefix token
+// (UTC YYYY/MM/DD) so it shares T0 with configTemplateContext.T0Plus.
+func writeAWSInit(path string, aws *config.AWSConfig, now time.Time) error {
 	var b strings.Builder
 	b.WriteString("#!/bin/sh\n")
 	b.WriteString("# Auto-generated by harness — creates bench resources inside LocalStack.\n")
@@ -61,17 +65,29 @@ func writeAWSInit(path string, aws *config.AWSConfig) error {
 		if prefix == "" {
 			prefix = "seed/"
 		}
+		// $TODAY is substituted here, in Go — the upload command below is
+		// single-quoted, so the shell would never expand it.
+		prefix = strings.ReplaceAll(prefix, config.SeedTodayToken, now.UTC().Format("2006/01/02"))
 		marker := so.Marker
 		if marker == "" {
 			marker = "SEED"
 		}
+		ext := so.Extension
+		if ext == "" {
+			ext = ".log"
+		}
+		// A delayed group sleeps before uploading so its LastModified is
+		// separated from the previous group by real wall-clock time.
+		if so.DelaySeconds > 0 {
+			fmt.Fprintf(&b, "sleep %d\n", so.DelaySeconds)
+		}
 		// 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)
+		fmt.Fprintf(&b, "  awslocal s3 cp /tmp/pb-seed-obj 's3://%s/%sobj-'\"$i\"'%s'\n", so.Bucket, prefix, ext)
 		b.WriteString("  i=$((i+1))\n")
 		b.WriteString("done\n")
 	}