audit output

Verifies a stamped stream produced by the audit input and reports what it found.

# Config fields, showing default values
output:
  label: ""
  audit:
    stream_id: "" # No default (required)
    payload:
      seed: 0
      entropy: "random"
    measurement:
      loss_grace: "30s"
      fault_correlation_window: "60s"
      dedup_horizon: 100000
    failure_injection:
      seed: 0
      interval: "10m"
      steady_fraction: 0.3
      faults: [] # No default (optional)

Checks every record against the identity stamped on it by a matching audit input, and publishes what it finds: records verified, lost, duplicated, corrupted, delivered out of order, or delivered against the priority contract.

The pair is only meaningful together. An audit output reads records an audit input produced, so whatever sits between the two -- a buffer, a processor chain, a whole pipeline -- is what is being audited.

This component classifies and never judges. It emits no pass/fail verdict and holds no thresholds; alerting policy belongs in the metrics backend, over the series it publishes.

Pairing

The stream_id, payload.seed and payload.entropy values MUST match the input's. The payload check recomputes each record's bytes from its identity, so a mismatch fails every record rather than degrading gracefully.

State

Verdict state is persisted under the edge data directory before any batch is acked, so a restart resumes with its frontier intact and reports the redelivery it caused as correlated duplicates rather than as loss. This output therefore requires a pipeline runner and refuses to run without one.

Examples

Buffer integrity

Three priority classes through a durable buffer, verified end to end. This is the shortest pipeline that exercises the pair: run it on a local edge and the output reports every record it received, and every one it did not.

input:
  audit:
    stream_id: local-audit-v1
    traffic:
      high:
        rate: 5/s
      normal:
        rate: 1/s
      low:
        rate: 1/s
        idle: { every: 30m, for: 2m }

buffer:
  priority_queue:
    max_bytes: 256MiB

output:
  audit:
    stream_id: local-audit-v1

Fields

stream_id

Identity of the audited stream. MUST match the stream_id of the audit input that produced these records: it keys the persisted verdict state, and records stamped for any other stream are reported as foreign rather than verified against this state. A matched pair MUST share this value -- that is what pairs them, and the input may be in this pipeline or in a separate one. Two audit outputs MUST NOT: they would overwrite each other's verdicts.

Type: string

payload

Payload parameters, which must match the audit input's payload block.

Type: object

payload.seed

Seed for the deterministic payload check. MUST match the input's. 0 derives the seed from stream_id, which both halves do identically, so leaving it unset on both is correct.

Type: int
Default: 0

payload.entropy

Shape of the deterministic payload. MUST match the input's.

Type: string
Default: "random"

Option	Summary
random	Payloads are filled from the deterministic pseudo-random stream, so a substituted body is detected.
zeros	Payloads are all zero bytes, the compressible arm. Every identity shares one payload, so the content check degenerates to a length check.

measurement

Measurement definitions. These are the only knobs that must live in the component, because the classification cannot be computed correctly without per-record state that is gone by the time the numbers reach a metrics backend. Duplicate rates, displacement bounds and latency targets are alert expressions over the published series, not settings here.

Type: object

measurement.loss_grace

How long a gap below the source's committed watermark must persist before it is finalized as loss. Too short reports records that were merely slow; too long delays a real alert.

Type: string
Default: "30s"

measurement.fault_correlation_window

How long after a disturbance -- a source restart, or this component reloading its state -- duplicates are labelled correlated=true. An uncorrelated duplicate is the one that should alert. Keep this at or above loss_grace: a loss line names the injected fault that opened its hole only while the fault is still inside this window, so a much larger loss_grace yields loss lines with no fault named.

Type: string
Default: "60s"

measurement.dedup_horizon

How many sequences per lane are remembered below the frontier. Beyond it a duplicate is still reported, but without the state to say anything more precise about it.

Type: int
Default: 100000

failure_injection

Deliberate misbehaviour, so the components between the audit pair can be checked against what their documentation promises about failure. The output refuses or delays batches on a seeded schedule and records every window it did so in.

Listing a fault is what arms it. Omit this block entirely and the component is inert: nothing is scheduled, no extra work is done per batch, and nothing extra is written to the checkpoint.

That record is what makes the audit readable under load. A duplicate arriving inside a fault window is labelled correlated=true -- the pipeline being at-least-once, as its contract permits -- rather than correlated=false, which is the one that should alert. Records the injector itself refused are separately exempted from the out-of-order verdict, for as long as the lane has not accounted for them: they were displaced by this component, and reporting that would be the injector accusing the pipeline of its own work.

Nothing else is hidden. A correlated duplicate is still counted and still published; only the label changes. Loss classification is untouched by every fault here -- a reject window can DELAY a loss verdict by stalling the watermark, which audit.watermark.age reports, but nothing here can excuse one.

This belongs on a dedicated test or canary pipeline, not on one whose numbers are read as a production baseline.

Type: object

failure_injection.seed

Seed for the schedule, so a run can be reproduced. 0 derives one from the clock. NOTE: this is not the same 0 as payload.seed, which derives from stream_id -- the schedule has no stream-derived form, because two pipelines auditing the same stream should not misbehave in lockstep. Reproducibility is per process: the seed fixes which faults fire and in what order, not the wall-clock instant the first cycle starts.

Type: int
Default: 0

failure_injection.interval

Nominal length of one schedule cycle, jittered +/-50%; each cycle fires at most one fault. The mean gap between firings is this value PLUS the mean fault duration, because a cycle does not end before its fault does. Between 1s and 24h.

Type: string
Default: "10m"

failure_injection.steady_fraction

The leading fraction of every cycle kept fault-free. This is the only guaranteed quiet interval -- the one in which gaps close, a spool drains and the loss grace runs undisturbed -- and it is also where a REAL defect shows itself: no duplicate is correlated inside it, because there is no recent fault to attribute one to. Records the injector refused earlier stay exempt from the ordering verdict wherever they eventually arrive, which is deliberate.

Type: float
Default: 0.3

failure_injection.faults

The faults to draw from, and the switch that arms injection. Listing at least one is required if the block is present at all -- a block that configures injection and lists nothing would lint clean and inject nothing. To turn injection off, remove the block.

Type: array of object

failure_injection.faults[].type

Which fault this entry injects.

Type: string

Option	Summary
latency	Answer slowly. This does not itself reorder anything -- the audit output writes one batch at a time -- but it maximises the leased-but-unacked set, which is the condition a buffer's cross-lane gate exists to survive.
reject_all	Refuse every batch for the window: a sink outage. A buffered pipeline spools; an unbuffered one simply retries.
reject_seq	Refuse any batch carrying one of a seeded subset of sequences, which drives a buffer's park and redrive path. duration is how long the poison keeps refusing before it lifts, standing in for an operator fixing whatever was rejecting the record. Only meaningful with a buffer that parks: set max_delivery_attempts above 0 on the priority_queue. Without one there is nothing to park the batch, so it is retried for the whole duration and the stream stalls for that long.

failure_injection.faults[].weight

Relative likelihood of this entry being the one drawn when a cycle fires. Entries do not have their own cadence: there is one schedule, and interval governs it.

Type: int
Default: 1

failure_injection.faults[].duration

How long a firing of this fault lasts, drawn uniformly. REQUIRED for every fault type.

Type: object

failure_injection.faults[].duration.min

Shortest window this fault runs for. Between 100ms and 1h.

Type: string

failure_injection.faults[].duration.max

Longest window this fault runs for. Setting only min gives a fixed window.

Type: string

failure_injection.faults[].delay

How slowly to answer. latency only; setting it on another fault type is rejected rather than ignored.

Type: object

failure_injection.faults[].delay.base

Central sleep applied before each batch.

Type: string

failure_injection.faults[].delay.jitter

Spread around base, drawn per batch and clamped at zero. At most 1m. Drawn per batch rather than per window because a constant delay shifts every completion equally and changes nothing about their order.

Type: string

failure_injection.faults[].probability

Fraction of sequences to poison. reject_seq only, and required for it. Whether a given sequence is poisoned is a function of the seed and the record's identity, so a batch redriven hours later in another process is poisoned identically.

Type: float