Documentation
¶
Overview ¶
Package harness implements the distributed test harness used by cmd/tpch-harness. It orchestrates a multi-process wadjet cluster on the dev box (local mode) or drives a pre-existing cluster (golden mode), runs the TPC-H query suite plus synthetic micro-queries, captures structured measurements, and compares them against a calibrated baseline.
The package is intentionally importable (not test-only) so the harness binary can use it directly. Helpers for setting up an embedded NATS, loading the catalog, and submitting queries are extracted from the existing distributed_tpch_test.go in internal/coordinator so there is exactly one implementation.
See docs/_archive/specs/2026-04-08-distributed-test-harness-design.md for the full design.
Index ¶
- Constants
- Variables
- func AllTPCHQueries() []string
- func CompareValueSigs(base, got string, relTol float64) (ok bool, detail string)
- func LoadQuery(name string) (string, error)
- func SelectQueries(requested []string) []string
- func SweepStaleRunArtifacts(harnessRoot, dataDir string, pruneOlderThan time.Duration, logger *slog.Logger)
- type BaselineFile
- type Cluster
- func (c *Cluster) ConnectNATS() (*nats.Conn, error)
- func (c *Cluster) DebugPorts() map[string]int
- func (c *Cluster) NATSURL() string
- func (c *Cluster) PgAddr() string
- func (c *Cluster) Shutdown(ctx context.Context) error
- func (c *Cluster) Start(ctx context.Context) error
- func (c *Cluster) StartCoordinator(ctx context.Context) error
- func (c *Cluster) StartWorkers(ctx context.Context) error
- type ClusterConfig
- type Config
- type HangDetector
- type MeasurementCollector
- type Mode
- type PreflightResult
- type Projection
- type QueryBaseline
- type QueryDelta
- type QueryMeasurement
- func RunMicroGraceHashJoin(ctx context.Context, coordURL string, collector *MeasurementCollector) (QueryMeasurement, error)
- func RunMicroHashAggHighCard(ctx context.Context, coordURL string, collector *MeasurementCollector) (QueryMeasurement, error)
- func RunMicroReverseBloom(ctx context.Context, coordURL string, collector *MeasurementCollector) (QueryMeasurement, error)
- func RunSkewQuery(ctx context.Context, coordURL string, name string, ...) (QueryMeasurement, error)
- type RunResult
- type Slice
- type SliceConfig
- type ValueSigAccum
Constants ¶
const ( KB = 1 << (10 * iota) MB GB )
const ( ExitOK = 0 ExitRegression = 1 // perf regression, missing spill paths, or hang ExitSetup = 2 // setup error, cluster crash, internal harness bug ExitCorrectness = 3 // row count or checksum diverged )
Exit codes returned by cmd/tpch-harness. Higher numbers outrank lower when multiple failures occur in one run.
const ValueSigRelTol = 1e-6
ValueSigRelTol is the relative tolerance CompareValueSigs applies per column. Summation-order noise on float64 column sums is ~1e-13 relative; the corruption classes this gate exists for are ≥ percents.
Variables ¶
var SliceConfigs = map[Slice]SliceConfig{ SliceSmall: { Name: SliceSmall, LineitemFiles: 4, OrdersFiles: 1, GoMemLimit: 4 * GB, ExpectSpill: false, }, SliceLarge: { Name: SliceLarge, LineitemFiles: 12, OrdersFiles: 3, GoMemLimit: 8 * GB, ExpectSpill: true, MemoryBudget: 64 * MB, }, }
SliceConfigs maps each Slice to its configuration.
Functions ¶
func AllTPCHQueries ¶
func AllTPCHQueries() []string
AllTPCHQueries returns the names of all 22 TPC-H queries in canonical order.
func CompareValueSigs ¶
CompareValueSigs compares two signatures column-wise with relative tolerance. Returns ok=true when every shared column agrees within relTol and both signatures cover the same columns. The detail string names the first divergence.
func LoadQuery ¶
LoadQuery returns the SQL text for the given TPC-H query name (e.g. "q05"). Uses SF100 scale factor for Q11 fraction calculation.
func SelectQueries ¶
SelectQueries resolves the --queries flag to a final ordered list. An empty input means all 22 TPC-H queries plus all micros.
func SweepStaleRunArtifacts ¶
func SweepStaleRunArtifacts(harnessRoot, dataDir string, pruneOlderThan time.Duration, logger *slog.Logger)
SweepStaleRunArtifacts removes leftover transient state from prior harness runs that crashed, timed out, or otherwise didn't reach the deferred cleanup paths. Called from Run() *before* CheckPreflight so the disk-space check sees a clean slate.
Two sources of leakage we observe in practice:
/tmp/wadjet-harness/run-<unix>/ - per-run logs + spill + JetStream store. The harness removes it on success unless WADJET_HARNESS_KEEP=1 is set, but a panic / external SIGKILL leaves it. Each abandoned SF1 run dir is several GB.
<dataDir>/wadjet/queries/<query_id>/ - per-query intermediates that the coordinator's cleanupQuery now removes on completion (committed 2542260), but a coordinator killed mid-flight by harness teardown never reaches that hook. Each orphan query is ~1 GB at SF1 and ~100 GB at SF10.
Safety: the caller must invoke checkNoOrphanedWadjet first OR otherwise guarantee no concurrent wadjet process is touching these paths. With pruneOlderThan = 0 the function deletes everything; otherwise only entries with mtime older than that threshold are removed (use this to avoid sweeping a sibling harness's just-created run dir).
Types ¶
type BaselineFile ¶
type BaselineFile struct {
Version int `json:"version"`
CapturedAt string `json:"captured_at"`
CapturedOn string `json:"captured_on"`
Queries map[string]QueryBaseline `json:"queries"`
ProjectionFactors map[string]Projection `json:"projection_factors"`
}
BaselineFile is the on-disk schema for the calibration table. The version field is incremented when incompatible changes are made.
func LoadBaseline ¶
func LoadBaseline(path string) (*BaselineFile, error)
LoadBaseline reads and parses a baseline file.
func (*BaselineFile) Compare ¶
func (bf *BaselineFile) Compare(m QueryMeasurement) []QueryDelta
Compare returns one QueryDelta per metric for the given measurement. Status is "PASS" if drift is within tolerance, "REGRESS" otherwise. Row count and checksum mismatches always REGRESS regardless of tolerance.
func (*BaselineFile) Project ¶
func (bf *BaselineFile) Project(sliceKey string, local QueryMeasurement) (QueryMeasurement, error)
Project converts a local-mode measurement into projected golden-mode values using the projection factors for the given slice.
func (*BaselineFile) Save ¶
func (bf *BaselineFile) Save(path string) error
Save writes the baseline to disk as pretty-printed JSON.
type Cluster ¶
type Cluster struct {
// contains filtered or unexported fields
}
Cluster is a process supervisor for one coordinator + N workers. The coordinator owns the embedded NATS server; the harness connects to it via TCP to seed catalog data and subscribe to heartbeats.
func NewCluster ¶
func NewCluster(cfg ClusterConfig) *Cluster
NewCluster constructs a Cluster but does not start anything.
PgAddr defaults: when empty or ":15433" (the historical default), the cluster picks a free port at coordinator-start time. The default :15433 silently collided with any process — including unrelated services that happen to occupy that port (in one observed case, an external tool called `warden` on a developer machine), causing the coordinator to fail pgwire-listen and shut down NATS, leaving the harness's catalog seeding stuck on a dead embedded NATS with a misleading "kv get … context deadline exceeded" error several seconds later. Fixing the port choice at construct time made local-mode harness runs reliable.
Callers that need a specific port (e.g. interactive psql) can still pass an explicit PgAddr like ":5432".
func (*Cluster) ConnectNATS ¶
ConnectNATS returns a NATS connection to the coordinator's embedded NATS.
func (*Cluster) DebugPorts ¶
DebugPorts returns a map of role → HTTP port for pprof access. Only valid after StartCoordinator and StartWorkers have been called.
func (*Cluster) NATSURL ¶
NATSURL returns the coordinator's NATS URL. Only valid after StartCoordinator.
func (*Cluster) Start ¶
Start is a convenience that calls StartCoordinator + StartWorkers without seeding data in between.
func (*Cluster) StartCoordinator ¶
StartCoordinator spawns the coordinator process (which owns the embedded NATS server) and waits until NATS is accepting connections. Call this first, then seed data via ConnectNATS, then call StartWorkers.
type ClusterConfig ¶
type ClusterConfig struct {
WadjetBin string // path to wadjet binary
RunDir string // /tmp/wadjet-harness/run-X
NumWorkers int
GoMemLimit int64
PgAddr string // pgwire listen address for coordinator (default ":15433")
DataDir string // local data dir (FileStore) for StorageType=="file"
// MemoryBudget, when > 0, is passed as both --memory-budget and
// --shared-pool-budget to every spawned process (coordinator and
// workers). 0 leaves both flags unset, falling through to the
// engine's cgroup/physical-memory auto-detection — which floors the
// per-task budget near 2 GB (cmd/wadjet/main.go minBudgetPerTask) and
// auto-sizes the shared pool as (envelope - cache) independent of
// --memory-budget, so GOMEMLIMIT alone cannot force spill at these
// fixture sizes: "spill triggers are pool-driven" (main.go), so both
// flags need the explicit override, not just --memory-budget.
MemoryBudget int64
// StorageType selects the coordinator/worker storage backend.
// "" or "file" -> FileStore at DataDir (default)
// "s3" -> MinIO/S3 at Endpoint/Region/Bucket
StorageType string
Bucket string
Region string
Endpoint string
SSL bool
// DataPlane selects the worker↔coord transport. "" or "nats" uses
// the legacy NATS reply-subject path; "grpc" enables the new
// data-plane gRPC stream (Phase B+). Both transports route into the
// same coord-side gatherReceiver so cross-flag results are
// identical; the toggle exists to validate the new path without
// committing to it as default.
DataPlane string
// ExtraServeArgs are appended verbatim to every spawned `wadjet serve`
// command (coordinator and workers). Used to exercise deploy-gated
// flags through the local harness before an EC2 run.
ExtraServeArgs []string
// SpawnWrapper, when non-empty, is prepended to every spawned process's
// argv: wrapper[0] wrapper[1:]... <wadjet-bin> <args...>. Used to run
// the cluster under an enforcement harness — e.g. a docker memory-cap
// wrapper for edge-box simulation (hard OOM-kill semantics that
// GOMEMLIMIT alone cannot provide). The wrapper owns env forwarding:
// exec.Command env vars (GOMEMLIMIT, GODEBUG, heap-dump paths) do NOT
// cross a container boundary unless the wrapper forwards them.
SpawnWrapper []string
Logger *slog.Logger
}
ClusterConfig describes a local-mode cluster to spawn.
type Config ¶
type Config struct {
Mode Mode
Slice Slice // local only
CoordURL string // golden only
DataDir string // local only; default /tmp/sf100-sample
BaselinePath string
OutPath string
Queries []string // empty means all 22 + micros
UpdateBaseline bool
NoCompare bool
WadjetBin string // path to wadjet binary; auto-built if empty
PgAddr string // local only; override for coordinator pgwire listen addr
NumWorkers int // local only; cluster size to spawn (0 = default of 2)
// Runs repeats the whole query suite against the same live cluster
// (0/1 = once). The EC2 SF100 protocol is benchmark_runs=2 — run 1
// populates the NVMe base-table cache cold, run 2 measures the
// steady regime over it (docs/benchmarks/
// steady-slower-than-cold-2026-08-08.md); this is the local mirror.
// Baseline comparison applies to run 1 only; later runs are recorded
// with QueryMeasurement.Run set, plus a cross-run row-count/value-sig
// identity check.
Runs int
// ScaleFactor is the TPC-H data volume for local mode generation.
// 0 (zero value) defaults to 0.01 (~10 MB total, lineitem 60K rows).
// Larger values let the harness exercise per-stage round-trip and
// memory-pressure paths at meaningful scale without EC2 spend:
// 0.1 → ~150 MB total, lineitem 600K rows (~5-10 sec total benchmark)
// 1.0 → ~1.5 GB total, lineitem 6M rows (~30-60 sec)
// 10.0 → ~15 GB total — local generation will take minutes; only useful
// on machines with disk + memory to spare.
ScaleFactor float64 // 0 = SF0.01 default
// S3 source (Source=="s3" only)
Source string // "local" (default) or "s3"
Bucket string
Region string
Endpoint string
SSL bool
DataPrefix string // prefix under Bucket containing table data (e.g. "tables/")
// DataPlane selects worker↔coord transport for the spawned cluster.
// Empty / "nats" uses the legacy NATS reply-subject path. "grpc"
// enables the data-plane gRPC stream (Phase B+).
DataPlane string
// ExtraServeArgs are appended verbatim to every spawned `wadjet serve`
// command (coordinator and workers). Use to exercise deploy-gated
// flags (e.g. --bounded-dirty-writes, --mmap-relief) through the
// local harness before an EC2 run.
ExtraServeArgs []string
// SpawnWrapper is prepended to every spawned process argv (see
// ClusterConfig.SpawnWrapper). E.g. a docker memory-cap wrapper for
// edge-box simulation.
SpawnWrapper []string
}
Config is the parsed flag set passed into Run.
type HangDetector ¶
type HangDetector struct {
// contains filtered or unexported fields
}
HangDetector watches a goroutine count series and trips when the count grows monotonically for longer than the threshold duration.
func NewHangDetector ¶
func NewHangDetector(threshold time.Duration) *HangDetector
NewHangDetector creates a detector with the given threshold (e.g. 30s).
func (*HangDetector) Observe ¶
func (h *HangDetector) Observe(t time.Time, count int) bool
Observe records one (timestamp, goroutine count) sample. Returns true if a hang has been detected. Once tripped, returns true forever until Reset is called.
func (*HangDetector) Reset ¶
func (h *HangDetector) Reset()
Reset clears the trip state and starts over.
type MeasurementCollector ¶
type MeasurementCollector struct {
// contains filtered or unexported fields
}
MeasurementCollector subscribes to worker heartbeats and aggregates per-query measurement windows. One Collector instance is used for the entire run; queries demarcate measurement windows by calling StartWindow / EndWindow.
func NewCollector ¶
func NewCollector() *MeasurementCollector
NewCollector creates a fresh collector with no active window.
func (*MeasurementCollector) EndWindow ¶
func (c *MeasurementCollector) EndWindow(query string) QueryMeasurement
EndWindow finalizes the active window and returns its measurement. Returns the zero value if there's no active window or the query name doesn't match.
func (*MeasurementCollector) Observe ¶
func (c *MeasurementCollector) Observe(hb distributed.WorkerHeartbeat)
Observe feeds a heartbeat into the active window. Safe to call from the heartbeat subscriber goroutine.
func (*MeasurementCollector) RunPeakSpillBytes ¶
func (c *MeasurementCollector) RunPeakSpillBytes() int64
RunPeakSpillBytes returns the highest hb.SpillDiskUsed seen across every heartbeat this collector has observed, regardless of per-query window boundaries. See the runPeakSpill field comment for why this is the reliable signal for a run-level "did spill happen at all" assertion.
func (*MeasurementCollector) StartWindow ¶
func (c *MeasurementCollector) StartWindow(query string)
StartWindow begins a new measurement window for the given query. Any prior window is discarded.
type PreflightResult ¶
PreflightResult holds the results of all preflight checks.
func CheckPreflight ¶
func CheckPreflight(slice SliceConfig, runDir string, workerCount int) PreflightResult
CheckPreflight runs all checks for the given slice and run dir.
func (PreflightResult) Error ¶
func (r PreflightResult) Error() string
type Projection ¶
type Projection struct {
WallMsMultiplier float64 `json:"wall_ms_multiplier"`
HeapMultiplier float64 `json:"heap_multiplier"`
AllocCountMultiplier float64 `json:"alloc_count_multiplier"`
SpillMultiplier float64 `json:"spill_multiplier"`
}
Projection maps a local-mode metric to the equivalent golden-mode value via per-metric multipliers. local / multiplier = projected golden value.
type QueryBaseline ¶
type QueryBaseline struct {
WallMsP50 int64 `json:"wall_ms_p50"`
WallMsTolerancePct float64 `json:"wall_ms_tolerance_pct"`
PeakHeapMB int64 `json:"peak_heap_mb"`
PeakHeapTolerancePct float64 `json:"peak_heap_tolerance_pct"`
AllocCount int64 `json:"alloc_count"`
AllocCountTolerancePct float64 `json:"alloc_count_tolerance_pct"`
SpillBytesWritten int64 `json:"spill_bytes_written"`
SpillTolerancePct float64 `json:"spill_tolerance_pct"`
RowCount int64 `json:"row_count"`
RowChecksum string `json:"row_checksum"`
// ValueSig is the canonical per-column numeric-sum signature
// (valuesig.go), compared with ValueSigRelTol relative tolerance —
// unlike RowChecksum it is order-insensitive and float-wobble-proof,
// so it can gate VALUES (the #278 / eager-§14.3 corruption class that
// row counts cannot see). Empty = not gated.
ValueSig string `json:"value_sig,omitempty"`
}
QueryBaseline holds the golden numbers and tolerances for one query.
type QueryDelta ¶
type QueryDelta struct {
Query string `json:"query"`
Metric string `json:"metric"`
Baseline float64 `json:"baseline"`
Projected float64 `json:"projected"`
DriftPct float64 `json:"drift_pct"`
TolerancePct float64 `json:"tolerance_pct"`
Status string `json:"status"` // "PASS", "REGRESS"
// Detail carries non-scalar divergence context (e.g. which value-
// signature column diverged and by how much).
Detail string `json:"detail,omitempty"`
}
QueryDelta records a single per-metric drift between projected and baseline.
type QueryMeasurement ¶
type QueryMeasurement struct {
Query string `json:"query"`
WallMs int64 `json:"wall_ms"`
PeakHeapMB int64 `json:"peak_heap_mb"`
AllocCount int64 `json:"alloc_count"`
SpillBytes int64 `json:"spill_bytes"`
RowCount int64 `json:"row_count"`
RowChecksum string `json:"row_checksum"`
ValueSig string `json:"value_sig,omitempty"`
GoroutinePeak int `json:"goroutine_peak"`
Hung bool `json:"hung"`
HangDumpPath string `json:"hang_dump_path,omitempty"`
StartedAt time.Time `json:"started_at"`
// Run numbers the suite pass this measurement came from under
// Config.Runs > 1 (0/absent = single-pass run or run 1).
Run int `json:"run,omitempty"`
}
QueryMeasurement is the result of running one query (or micro).
func RunMicroGraceHashJoin ¶
func RunMicroGraceHashJoin(ctx context.Context, coordURL string, collector *MeasurementCollector) (QueryMeasurement, error)
RunMicroGraceHashJoin forces grace hash join partitioning by joining a memory-heavy build side (micro_build, 500K padded rows) against a smaller probe side (micro_probe, 50K rows), then asserts spill occurred.
func RunMicroHashAggHighCard ¶
func RunMicroHashAggHighCard(ctx context.Context, coordURL string, collector *MeasurementCollector) (QueryMeasurement, error)
RunMicroHashAggHighCard runs a high-cardinality GROUP BY (100K distinct keys) and asserts allocation discipline — no per-row allocation leak.
func RunMicroReverseBloom ¶
func RunMicroReverseBloom(ctx context.Context, coordURL string, collector *MeasurementCollector) (QueryMeasurement, error)
RunMicroReverseBloom forces the reverseBloomBridge into its spill path by joining a large build side (micro_lineitem, 200K rows) against a small probe side (micro_orders, 20K rows), then asserts spill occurred.
func RunSkewQuery ¶
func RunSkewQuery(ctx context.Context, coordURL string, name string, collector *MeasurementCollector) (QueryMeasurement, error)
RunSkewQuery executes one skew-suite query by name and asserts its result shape (both queries return at least one row; row values are checksummed by the shared runner for cross-arm parity).
type RunResult ¶
type RunResult struct {
Mode Mode `json:"mode"`
Slice Slice `json:"slice,omitempty"`
StartedAt time.Time `json:"started_at"`
DurationMs int64 `json:"duration_ms"`
Queries []QueryMeasurement `json:"queries"`
BaselinePath string `json:"baseline_path"`
Regressions []QueryDelta `json:"regressions"`
Hangs []string `json:"hangs"`
Passed bool `json:"passed"`
ExitCode int `json:"exit_code"`
}
RunResult is the top-level structured output written to the result JSON.
type SliceConfig ¶
type SliceConfig struct {
Name Slice
LineitemFiles int
OrdersFiles int
GoMemLimit int64 // bytes; passed to worker via GOMEMLIMIT env
ExpectSpill bool // if true and total spill bytes == 0, fail the run
// MemoryBudget, when > 0, is passed as --memory-budget to every
// spawned process (bytes; see ClusterConfig.MemoryBudget). GOMEMLIMIT
// alone does not force spill: when --memory-budget is left at its
// default (0), the engine auto-detects a per-task budget from the
// cgroup/physical-memory envelope and floors it near 2 GB
// (cmd/wadjet/main.go minBudgetPerTask) to stay viable for SF100-class
// joins — far above anything this slice's fixtures need, so it never
// spills regardless of GoMemLimit. 0 = flag unset (small slice; no
// spill expected, so let the engine auto-detect as usual).
MemoryBudget int64
}
SliceConfig describes a local-mode data slice. The harness uses these to choose how many sample files to load into the catalog and what GOMEMLIMIT each worker process is started with.
type ValueSigAccum ¶
type ValueSigAccum struct {
// contains filtered or unexported fields
}
ValueSigAccum accumulates per-column numeric sums over result rows.
func (*ValueSigAccum) AddFloat ¶
func (a *ValueSigAccum) AddFloat(col int, v float64)
AddFloat folds one numeric value into column col.
func (*ValueSigAccum) AddVals ¶
func (a *ValueSigAccum) AddVals(vals []any)
AddVals folds one row of driver-level values (database/sql scan targets or engine row slices). Numeric types and numeric-parseable strings (pgwire renders decimals as text) contribute; everything else is skipped.
func (*ValueSigAccum) Signature ¶
func (a *ValueSigAccum) Signature() string
Signature renders the accumulated sums as "c<i>:<sum %.9e>" pairs joined by ",". Columns that never contributed a numeric value are omitted. Empty string when no column was numeric.