Skip to main content

audit input

Generates synthetic records stamped with a verifiable identity, for a matching audit output to check.

# Common config fields, showing default values
input:
  label: ""
  audit:
    stream_id: "" # No default (required)
    traffic: {} # No default (required)
    payload:
      sizes: [] # No default (optional)
      seed: 0
      entropy: "random"
    watermark_interval: "5s"

Produces traffic whose every record carries the identity needed to prove what happened to it: a per-priority-class sequence number, a timestamp, and a checksum binding the stamp to a deterministic payload. A matching audit output recomputes all of it and reports what it found -- records verified, lost, duplicated, corrupted, delivered out of order, or delivered against the priority contract.

The pair is only meaningful together. Whatever sits between the two -- a buffer, a processor chain, a whole pipeline -- is what is being audited.

Pairing​

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

Truth boundaries​

Alongside the data the source emits periodic in-band watermark records saying how far its durable truth extends: which sequences the pipeline has acknowledged, and which it reserved but cannot prove it emitted after a restart. That boundary is what lets the output tell a lost record from one that is merely still in flight, so a pipeline that drops these control records reports no loss at all.

State​

Sequence counters are persisted under the edge data directory before the sequences they cover are stamped, so a restart resumes at an unused sequence and declares the gap it left rather than reusing an identity. This input 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 output that verifies these records: it keys the persisted sequence counters, appears in every stamp, and is what pairs the two halves -- and the output may be in this pipeline or in a separate one. Two audit inputs MUST NOT share this value: they would open the same counters and hand out the same record identities.

Type: string

traffic​

One generator per priority class. Keys are high, normal, low or none; none stamps no priority metadata, for single-lane pipelines. Each class carries its own dense sequence and its own shape, and is verified independently.

Type: map of object

traffic.<name>.rate​

Records per unit time, as N (per second) or N/<duration> -- 5/s, 30/m, 1/500ms. Must be greater than zero.

Type: string

traffic.<name>.burst​

Periodic extra load on top of the steady rate.

Type: object

traffic.<name>.burst.every​

How often a burst fires, before jitter.

Type: string

traffic.<name>.burst.size​

How many extra records the burst injects at once.

Type: int

traffic.<name>.burst.jitter​

Fraction of every the interval is randomly varied by, so bursts do not land in lockstep across lanes.

Type: float
Default: 0

traffic.<name>.idle​

Deliberate silences, which exercise empty-poll and age-based paths downstream.

Type: object

traffic.<name>.idle.every​

How often an idle window starts.

Type: string

traffic.<name>.idle.for​

How long the lane stays silent.

Type: string

payload​

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

Type: object

payload.sizes​

Weighted mix of payload sizes. Large entries exercise byte caps and codec framing; omitting the list uses a single 512B size.

Type: array of object

payload.sizes[].size​

Payload size, e.g. 512B, 64KiB.

Type: string
Default: "512B"

payload.sizes[].weight​

Relative likelihood of this size being chosen.

Type: int
Default: 1

payload.seed​

Seed for the deterministic payload. MUST match the output'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 output's.

Type: string
Default: "random"

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

watermark_interval​

How often the truth boundary is emitted. Shorter means loss is finalized sooner and restarts are announced sooner; longer means fewer control records on the wire. Cannot be 0.

Type: string
Default: "5s"

priority_field​

Metadata key the priority class is written to. MUST match the priority_field of a priority_queue buffer in the same pipeline, or every record is silently classified as normal and the cross-lane promises go unmeasured.

Type: string
Default: "priority"

max_batch​

Largest number of records emitted in one batch. Small batches keep the blast radius of a single rejected batch small.

Type: int
Default: 32

max_records​

Stop after this many records, emitting a final watermark first. 0 runs forever.

Type: int
Default: 0

failure_injection​

Faults this source injects on purpose, each a fraction of records. corrupt_rate and omit_rate are selected by a pure function of payload.seed and the record's own identity, so a run is reproducible and an external check can recompute exactly which records were affected; they also never touch sequence 0 of a lane, which is the baseline the verifier adopts from and where a fault could never be accounted for. invalid_priority_rate is different on both counts: it is drawn before a sequence exists, from the generator's own stream, so it is neither identity-keyed nor externally recomputable, and it has no sequence-0 exemption (harmlessly: a mis-stamped record is still trusted and still adopts).

Type: object

failure_injection.invalid_priority_rate​

Fraction of records whose priority metadata is set to an unrecognised value. The stamp keeps the truthful class, so the record still verifies in its own lane while the buffer must route it as normal and count it unclassified. Lanes that stamp no priority are unaffected. Those records are delivered against their stamped lane, which is the contract being tested and is reported as records arriving out of order; they are kept in batches of their own so the records beside them are not delivered out of lane too. Unlike the two below, this one is decided before a sequence exists -- it is the only source fault that changes which records may share a batch.

Type: float
Default: 0

failure_injection.corrupt_rate​

Fraction of records whose payload is damaged after stamping, so the line still names its record but cannot vouch for it. Exercises the verifier's corrupt path end to end: the damaged record occupies its sequence rather than aging into loss, which is the misclassification this subsystem exists to prevent. NOTE: applied only to records omit_rate did not already withhold, so the effective fraction is corrupt_rate * (1 - omit_rate).

Type: float
Default: 0

failure_injection.omit_rate​

Fraction of records whose sequence is spent but never emitted, producing REAL, PERMANENT loss that the verifier is supposed to find. This is the only setting here that manufactures the subsystem's most serious verdict, and unlike an injected sink fault the loss does not refute, because the record genuinely does not exist. A CRASH pardons a bounded window of it: the next boot declares the tail it cannot vouch for in doubt, and withheld sequences inside that window are written off rather than finalized as loss. A clean stop pardons nothing. Do not combine this with deliberate restarts. Incompatible with alerting on unrefuted loss. Intended for proving that detection works, not for production pipelines.

Type: float
Default: 0

tail​

Settings for mode: tail, read only in that mode. Instead of emitting stamped records straight into the pipeline, the source appends them to a local file and fsyncs, and an embedded real tail input -- the same implementation the registry constructs, with its full ack-gated checkpoint machinery -- reads that file back. What it emits is what this input emits.

The writer half never traverses the pipeline, so in this mode any gap the verifier finds is definitively a read-side event: the tail, a buffer, or delivery. The two halves coordinate only through the filesystem, which is exactly the tail input's contract with the world.

The truth boundary changes meaning with the mode. Here committed means appended and fsynced to the file, so it is the WRITER that commits; in direct mode it means acknowledged by the pipeline. The remaining fields are the tail input's own, passed through verbatim -- the ones an embedded reader over a file this component wrote itself can meaningfully vary. Where it reads from, how it groups lines and how it decodes them are NOT exposed: each has exactly one workable value here, and a knob whose only legal setting is its default is one an operator can only get wrong.

Type: object

tail.enabled​

Whether the tail loop runs. Deliberately has no default: writing this block without it is a configuration error rather than a pipeline that quietly stays in direct mode. Set false to keep the settings while running direct.

Type: bool

tail.file​

The file the two halves meet at.

Type: object

tail.file.path​

Absolute path of the file the writer appends to. Empty puts it at source.log under this stream's state directory, which is where it belongs unless the point is to put the file on a different volume from the state.

Type: string
Default: ""

tail.file.rotate​

Rotation, which is always on and triggers on age OR size: it bounds disk on a device that has little, and it keeps rotation handling -- the part of file tailing most likely to lose data -- continuously exercised rather than tested once.

Type: object

tail.file.rotate.every​

How often the writer rotates the source file.

Type: string
Default: "1h"

tail.file.rotate.max_bytes​

Rotate once the current file reaches this size, whichever trigger comes first. This is the trigger that actually bounds disk: a period alone does not, because an hour of traffic at any real rate with a large payload arm is hundreds of megabytes, and a node restarting more often than the period never reaches it at all. Size is carried across restarts; the period cannot be.

Type: string
Default: "32MiB"

tail.file.rotate.keep​

How many rotated files are retained. A segment pruned before the reader reaches it is data genuinely deleted unread, and the audit will report it as loss -- correctly, and indistinguishably from a real read-side defect. What must exceed the worst read lag is therefore the RETAINED WINDOW, and that window is keep times whichever trigger fires first: every, or max_bytes divided by how fast this source writes. At any real rate it is the size trigger, so the window is keep x max_bytes of stream -- minutes, not hours. Size the pair against the restart and backlog this pipeline can actually suffer.

Type: int
Default: 4

tail.poll_interval​

How often to check for new data and new/rotated files.

Type: string
Default: "200ms"

tail.max_log_size​

Maximum size of a single log entry. Entries exceeding this size are split.

Type: string
Default: "1MiB"

tail.checkpoint_id​

Explicit checkpoint identity. If set, this value alone determines which checkpoint state is used. Change it to start fresh (e.g. 'v1' → 'v2'). If not set, the checkpoint is derived automatically from the paths.

Type: string
Default: ""