From c065c975b2ae1c058226271e71ef035a1bdea92f Mon Sep 17 00:00:00 2001 From: Eren Aslan <16862833+erenaslandev@users.noreply.github.com> Date: Wed, 12 Aug 2026 19:54:47 +0300 Subject: [PATCH 1/4] feat(harness): add subject entrypoint overrides --- cmd/harness/main.go | 13 ++++++++++--- internal/config/case.go | 11 +++++++++++ internal/config/subject.go | 8 ++++++++ 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/cmd/harness/main.go b/cmd/harness/main.go index 2c065cc..a1828b9 100644 --- a/cmd/harness/main.go +++ b/cmd/harness/main.go @@ -744,11 +744,15 @@ func versionCmd() *cobra.Command { } } -// applyCasePin overlays a case's subject_image/subject_version pin onto a -// registry-resolved Subject. The global --image/--version CLI flags are applied -// later in runner.applySubjectOverrides, so an explicit CLI flag still wins: +// applyCasePin overlays a case's subject_image/subject_version/subject_entrypoint +// pin onto a registry-resolved Subject. The global --image/--version CLI flags are +// applied later in runner.applySubjectOverrides, so an explicit CLI flag still +// wins: // // CLI --image/--version > case subject_image/subject_version > registry default +// +// subject_entrypoint has no CLI counterpart: it is a property of what the case +// tests (see TestCase.SubjectEntrypoint), not of which build is under test. func applyCasePin(s config.Subject, tc *config.TestCase) config.Subject { if tc.SubjectImage != "" { s = s.WithImage(tc.SubjectImage) @@ -756,6 +760,9 @@ func applyCasePin(s config.Subject, tc *config.TestCase) config.Subject { if tc.SubjectVersion != "" { s = s.WithVersion(tc.SubjectVersion) } + if len(tc.SubjectEntrypoint) > 0 { + s = s.WithEntrypoint(tc.SubjectEntrypoint) + } return s } diff --git a/internal/config/case.go b/internal/config/case.go index dcfab7e..9cc39b2 100644 --- a/internal/config/case.go +++ b/internal/config/case.go @@ -135,6 +135,17 @@ type TestCase struct { SubjectImage string `yaml:"subject_image"` SubjectVersion string `yaml:"subject_version"` + // SubjectEntrypoint replaces the registry's entrypoint for this case, with + // the registry Command still appended as arguments. Same precedence and the + // same non-strict-decode tolerance as SubjectImage. + // + // Needed where the case is about the subject's own process lifecycle: the + // director self-update cases stop and restart the service they are updating, + // which ends the container when the service is PID 1. Pointing the entrypoint + // at a wrapper that runs the service as a CHILD keeps the container alive + // across that restart, so "the service came back" is observable at all. + SubjectEntrypoint []string `yaml:"subject_entrypoint"` + Subjects []string `yaml:"subjects"` Configurations map[string]Configuration `yaml:"configurations"` Correctness CorrectnessConfig `yaml:"correctness"` diff --git a/internal/config/subject.go b/internal/config/subject.go index 6c85b19..e23fd9c 100644 --- a/internal/config/subject.go +++ b/internal/config/subject.go @@ -336,3 +336,11 @@ func (s Subject) WithImage(img string) Subject { s.Image = img return s } + +// WithEntrypoint returns a copy of the Subject with the entrypoint overridden. +// Command is left alone, so the registry's arguments still reach the new +// entrypoint — a wrapper script gets them as "$@". +func (s Subject) WithEntrypoint(entrypoint []string) Subject { + s.Entrypoint = entrypoint + return s +} From b1a793c70926a7bba065847a61faf95299542430 Mon Sep 17 00:00:00 2001 From: Eren Aslan <16862833+erenaslandev@users.noreply.github.com> Date: Thu, 13 Aug 2026 18:16:04 +0300 Subject: [PATCH 2/4] fix(runner): enforce expect_max ceiling for endpoint sources Enforce an upper bound on records delivered by endpoint sources when expect_max is specified. This detects over-delivery (e.g., from duplicated routes or failed filters) that wouldn't be caught by expect_min alone. A quiet period is used to ensure the count stabilizes before the ceiling is checked, and validation is added to ensure the ceiling is not lower than the floor. --- internal/config/case.go | 15 ++++++++++++++- internal/runner/runner.go | 32 +++++++++++++++++++++++++++++++- 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/internal/config/case.go b/internal/config/case.go index 9cc39b2..0763176 100644 --- a/internal/config/case.go +++ b/internal/config/case.go @@ -2074,7 +2074,13 @@ type EndpointSourceConfig struct { // source must NOT deliver. The driver waits the settle window then asserts the // receiver count stays <= ExpectMax. Proves auth/validation is load-bearing. Reject bool `yaml:"reject"` - // ExpectMax is the ceiling for a reject case (default 0 — nothing delivered). + // ExpectMax is the ceiling. For a reject case it is the whole assertion + // (default 0 — nothing delivered). On a POSITIVE case it turns the count into a + // window: the driver waits past the floor and then fails on over-delivery, + // which is what a case needs when its records ARE its assertion (e.g. the + // director_update_* family, where one record stands for one proven outcome) — + // a floor alone cannot notice a subject that forwarded more than the case + // meant. ExpectMax int `yaml:"expect_max"` } @@ -2108,6 +2114,13 @@ func (tc *TestCase) validateEndpointSource() error { if tc.EndpointSource.Reject && tc.EndpointSource.ExpectMax < 0 { return fmt.Errorf("case %q: endpoint_source.expect_max must be >= 0", tc.Name) } + // A ceiling below the floor can never be satisfied, so refuse it at load time + // rather than at the end of a ten-minute run. + if !tc.EndpointSource.Reject && tc.EndpointSource.ExpectMax > 0 && + tc.EndpointSource.ExpectMax < tc.EndpointSource.ExpectMin { + return fmt.Errorf("case %q: endpoint_source.expect_max (%d) must be >= expect_min (%d)", + tc.Name, tc.EndpointSource.ExpectMax, tc.EndpointSource.ExpectMin) + } want := tc.EndpointSource.SenderContainerOrDefault()[len("bench-"):] found := false for _, e := range tc.Endpoints { diff --git a/internal/runner/runner.go b/internal/runner/runner.go index 4be2247..6e6af2a 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -6498,11 +6498,18 @@ func (r *Runner) runRedisSourceCorrectness(tc *config.TestCase, subject config.S return r.saveAuxResult(tc, subject, configName, "redis source", startTime, finalCount, passed, errs, pubContainer, subjectContainer) } +// endpointSourceCeilingQuietPeriod is how long a case that states an +// endpoint_source.expect_max keeps watching after its floor is met, before the +// ceiling is judged. Short on purpose: excess records arrive in a burst, so this +// only has to outlast one, and a longer wait would tax every ceiling case. +const endpointSourceCeilingQuietPeriod = 30 * time.Second + // runEndpointSourceCorrectness drives a director source the bench generator can't // feed (snmptrap, tftp, smtp, …) via a generic CLI-sender endpoint and counts at // the receiver. The sender is an `endpoints:` container in the case; the driver // just waits for the receiver to reach endpoint_source.expect_min. Tolerant of the -// best-effort loss of UDP senders (expect_min, not exact). +// best-effort loss of UDP senders (expect_min, not exact) unless the case also +// states an expect_max, which makes the count a window — see the ceiling below. func (r *Runner) runEndpointSourceCorrectness(tc *config.TestCase, subject config.Subject) (results.RunResult, error) { configName := r.opts.ConfigName subject = r.applySubjectOverrides(subject) @@ -6598,6 +6605,29 @@ func (r *Runner) runEndpointSourceCorrectness(tc *config.TestCase, subject confi fmt.Printf(" source delivered %d records (≥ %d) ✓\n", got, expectMin) } + // Ceiling, when the case states one. A floor alone cannot fail a case whose + // records ARE its assertion: a subject that forwards more than it was asked to + // (a filter that silently stopped matching, a duplicated route, an evidence line + // the case did not intend) clears any expect_min. Cases that encode content in + // the count set expect_max == expect_min and become exact. + // + // The extra wait is deliberately short rather than the full settle window: a + // runaway forwarder delivers its excess within a second or two, and charging + // every ceiling case the whole window would add minutes per case for nothing. + if reached && es.ExpectMax > 0 { + if err := sleepCtx(r.ctx, endpointSourceCeilingQuietPeriod); err != nil { + return results.RunResult{}, fmt.Errorf("interrupted: %w", err) + } + if stable := r.ccfWaitStable(metricsPort, time.Now().Add(10*time.Second)); stable > finalCount { + finalCount = stable + } + if finalCount > int64(es.ExpectMax) { + errs = append(errs, fmt.Sprintf("over-delivery: %d records from the source (expected ≤ %d) — more arrived than the case asserts", finalCount, es.ExpectMax)) + } else { + fmt.Printf(" source stayed within the ceiling: %d ≤ %d ✓\n", finalCount, es.ExpectMax) + } + } + passed := len(errs) == 0 return r.saveAuxResult(tc, subject, configName, "endpoint source", startTime, finalCount, passed, errs, senderContainer, subjectContainer) } From c454bda74c5228e0c5b3c400696968c88cd6b94d Mon Sep 17 00:00:00 2001 From: Eren Aslan <16862833+erenaslandev@users.noreply.github.com> Date: Thu, 13 Aug 2026 21:15:36 +0300 Subject: [PATCH 3/4] fix(runner): improve endpoint source ceiling logic Update validation to check for non-zero values instead of positive ones to prevent negative values from silently bypassing assertions. The runner now bounds the quiet period by the run deadline, verifies that the receiver metrics endpoint is responsive, and provides descriptive errors for over-delivery or reachability issues. --- internal/config/case.go | 9 ++++++--- internal/runner/runner.go | 33 ++++++++++++++++++++++++++++----- 2 files changed, 34 insertions(+), 8 deletions(-) diff --git a/internal/config/case.go b/internal/config/case.go index 0763176..6dc1e08 100644 --- a/internal/config/case.go +++ b/internal/config/case.go @@ -2115,10 +2115,13 @@ func (tc *TestCase) validateEndpointSource() error { return fmt.Errorf("case %q: endpoint_source.expect_max must be >= 0", tc.Name) } // A ceiling below the floor can never be satisfied, so refuse it at load time - // rather than at the end of a ten-minute run. - if !tc.EndpointSource.Reject && tc.EndpointSource.ExpectMax > 0 && + // rather than at the end of a ten-minute run. Every NONZERO value is checked, not + // just positive ones: zero is the documented "unset", and the runner enables the + // ceiling on the same `> 0` test, so `expect_max: -1` would otherwise pass + // validation AND silently disable the assertion it looks like it is making. + if !tc.EndpointSource.Reject && tc.EndpointSource.ExpectMax != 0 && tc.EndpointSource.ExpectMax < tc.EndpointSource.ExpectMin { - return fmt.Errorf("case %q: endpoint_source.expect_max (%d) must be >= expect_min (%d)", + return fmt.Errorf("case %q: endpoint_source.expect_max (%d) must be >= expect_min (%d), or 0 for no ceiling", tc.Name, tc.EndpointSource.ExpectMax, tc.EndpointSource.ExpectMin) } want := tc.EndpointSource.SenderContainerOrDefault()[len("bench-"):] diff --git a/internal/runner/runner.go b/internal/runner/runner.go index 6e6af2a..034d0f7 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -6615,15 +6615,38 @@ func (r *Runner) runEndpointSourceCorrectness(tc *config.TestCase, subject confi // runaway forwarder delivers its excess within a second or two, and charging // every ceiling case the whole window would add minutes per case for nothing. if reached && es.ExpectMax > 0 { - if err := sleepCtx(r.ctx, endpointSourceCeilingQuietPeriod); err != nil { - return results.RunResult{}, fmt.Errorf("interrupted: %w", err) + // Bounded by the run budget: the floor above is capped at runDeadline, and a + // case that reached it late must not push the run past --timeout on the way + // out. No budget left is a timeout, not a pass. + quietUntil := time.Now().Add(endpointSourceCeilingQuietPeriod) + if quietUntil.After(runDeadline) { + quietUntil = runDeadline + } + if remaining := time.Until(quietUntil); remaining > 0 { + if err := sleepCtx(r.ctx, remaining); err != nil { + return results.RunResult{}, fmt.Errorf("interrupted: %w", err) + } } - if stable := r.ccfWaitStable(metricsPort, time.Now().Add(10*time.Second)); stable > finalCount { + + // The ceiling needs a FRESH count, and it must fail when it cannot get one: + // ccfWaitStable reports 0 or the last value it saw when the metrics endpoint + // stops answering, and treating that as "within the ceiling" would pass a case + // whose delivery nobody measured. + if stable := r.ccfWaitStable(metricsPort, quietUntil.Add(10*time.Second)); stable > finalCount { finalCount = stable } - if finalCount > int64(es.ExpectMax) { + + rm, err := r.queryReceiverMetrics(metricsPort, 5*time.Second) + switch { + case err != nil: + errs = append(errs, fmt.Sprintf("ceiling not verified: the receiver metrics endpoint stopped answering (%v) — a ≤ %d assertion cannot be made on an unmeasured window", err, es.ExpectMax)) + case rm.LinesReceived > int64(es.ExpectMax): + finalCount = rm.LinesReceived errs = append(errs, fmt.Sprintf("over-delivery: %d records from the source (expected ≤ %d) — more arrived than the case asserts", finalCount, es.ExpectMax)) - } else { + default: + if rm.LinesReceived > finalCount { + finalCount = rm.LinesReceived + } fmt.Printf(" source stayed within the ceiling: %d ≤ %d ✓\n", finalCount, es.ExpectMax) } } From 8a776dc2ef987aa11feded6ffe2f5f14f1f1f990 Mon Sep 17 00:00:00 2001 From: Eren Aslan <16862833+erenaslandev@users.noreply.github.com> Date: Fri, 14 Aug 2026 12:14:31 +0300 Subject: [PATCH 4/4] fix(runner): clamp metric collection to run budget --- internal/runner/runner.go | 52 ++++++++++++++++++++++++++++++++++----- 1 file changed, 46 insertions(+), 6 deletions(-) diff --git a/internal/runner/runner.go b/internal/runner/runner.go index 034d0f7..11a7af2 100644 --- a/internal/runner/runner.go +++ b/internal/runner/runner.go @@ -6504,6 +6504,31 @@ func (r *Runner) runRedisSourceCorrectness(tc *config.TestCase, subject config.S // only has to outlast one, and a longer wait would tax every ceiling case. const endpointSourceCeilingQuietPeriod = 30 * time.Second +// earliest returns whichever instant comes first — used to keep a step's own +// deadline inside the run budget. +func earliest(a, b time.Time) time.Time { + if a.Before(b) { + return a + } + + return b +} + +// budgetedTimeout clamps a step's timeout to the budget left before deadline, never +// returning a non-positive value: a zero timeout would read as "no deadline" to the +// callers here, which is the opposite of what a spent budget means. +func budgetedTimeout(deadline time.Time, want time.Duration) time.Duration { + if remaining := time.Until(deadline); remaining < want { + if remaining < time.Second { + return time.Second + } + + return remaining + } + + return want +} + // runEndpointSourceCorrectness drives a director source the bench generator can't // feed (snmptrap, tftp, smtp, …) via a generic CLI-sender endpoint and counts at // the receiver. The sender is an `endpoints:` container in the case; the driver @@ -6622,21 +6647,36 @@ func (r *Runner) runEndpointSourceCorrectness(tc *config.TestCase, subject confi if quietUntil.After(runDeadline) { quietUntil = runDeadline } - if remaining := time.Until(quietUntil); remaining > 0 { - if err := sleepCtx(r.ctx, remaining); err != nil { - return results.RunResult{}, fmt.Errorf("interrupted: %w", err) - } + + remaining := time.Until(quietUntil) + if remaining <= 0 { + // No budget means the window was never watched, and an unwatched window is + // not a pass — the same rule the metrics-outage branch below applies. A case + // that reaches its floor in the last seconds of --timeout fails as + // unverified, which is the honest outcome for a case that asked for a ceiling. + errs = append(errs, fmt.Sprintf("ceiling not verified: the run budget (%s) ran out before the ≤ %d window could be observed", r.opts.Timeout, es.ExpectMax)) + passed := len(errs) == 0 + + return r.saveAuxResult(tc, subject, configName, "endpoint source", startTime, finalCount, passed, errs, senderContainer, subjectContainer) + } + + if err := sleepCtx(r.ctx, remaining); err != nil { + return results.RunResult{}, fmt.Errorf("interrupted: %w", err) } // The ceiling needs a FRESH count, and it must fail when it cannot get one: // ccfWaitStable reports 0 or the last value it saw when the metrics endpoint // stops answering, and treating that as "within the ceiling" would pass a case // whose delivery nobody measured. - if stable := r.ccfWaitStable(metricsPort, quietUntil.Add(10*time.Second)); stable > finalCount { + // + // Both this settle and the authoritative sample below stay inside runDeadline: + // the whole point of capping the quiet period is lost if the two steps after it + // can push the run past --timeout on their own. + if stable := r.ccfWaitStable(metricsPort, earliest(time.Now().Add(10*time.Second), runDeadline)); stable > finalCount { finalCount = stable } - rm, err := r.queryReceiverMetrics(metricsPort, 5*time.Second) + rm, err := r.queryReceiverMetrics(metricsPort, budgetedTimeout(runDeadline, 5*time.Second)) switch { case err != nil: errs = append(errs, fmt.Sprintf("ceiling not verified: the receiver metrics endpoint stopped answering (%v) — a ≤ %d assertion cannot be made on an unmeasured window", err, es.ExpectMax))