blockwatcher-sinks

blockwatcher-sinks implements the Sink port once per module it ships: webhook, script, and log (crates/blockwatcher-sinks/src/registry.rs). Each sink delivers one sink event, once, and reports a classified failure: the engine owns every retry, every backoff, and every dead-letter decision, so a sink module that retries internally is a bug (lib.rs). It is a module crate: adding, removing, or changing one sink module never touches blockwatcher-core.

Delivery guarantees already documents the generic contract every sink goes through from the engine’s side: deliver_with_retry (crates/blockwatcher-core/src/pipeline/delivery.rs) is the one retry loop every sink call passes through, and exhausting it is what produces a DeadLetter. This page does not restate that loop; it covers what is specific to each sink module: the wire format it sends, its own timeout distinct from the engine’s retry backoff, and exactly what each module classifies as retryable versus permanent, since that classification is what deliver_with_retry acts on.

Key takeaways

• blockwatcher-sinks implements the Sink port once per module it ships: webhook, script, and log. Each delivers one SinkEvent, once, and reports a classified failure; the engine owns every retry, backoff, and dead-letter decision.
• A sink module that retries internally is a bug: returning from deliver after exactly one attempt is the contract every module here follows.
• This page covers what is specific to each sink module (wire format, timeout, retry classification); Delivery guarantees already documents the generic engine-side contract.

Responsibilities

• Register exactly the sink modules it ships ("log", "webhook", "script") under the same family-enumeration convention every module crate follows (registry.rs).
• Render every event the identical way, once, for every sink to reuse: canonical_body serializes an blockwatcher_types::SinkEvent to tagged JSON (type: match | retracted | digest) and is the one wire rendering every sink that touches the network or another process uses (lib.rs).
• webhook: POST the canonical body to a secret-referenced URL, with operator-configured headers and its own request timeout (webhook.rs).
• script: pipe the canonical body (or a rendered body_template) to a subprocess’s stdin under a configured timeout, and read its exit code (or lack of one) as the delivery outcome (script.rs).
• log: write one line of the canonical body (or a rendered body_template) to stdout, serialized against concurrent LogSink instances by a process-wide lock so two deliveries can never interleave mid-line (log.rs).
• Classify every failure (an HTTP status, a transport error, a process exit code, an I/O error) into the shared ErrorClass the engine’s retry loop reads, at the point closest to the fact that produced it (each module’s own logic; see Per-sink delivery semantics below).

Not this crate’s job: retrying a failed delivery, deciding when enough attempts have been spent, or writing a DeadLetter: blockwatcher-core’s sink_worker.rs/delivery.rs own every one of those, and every sink here returns from deliver after exactly one attempt, success or failure (lib.rs); defining Sink, SinkError, or ErrorClass: those are blockwatcher-ports, only used here; resolving an env: secret reference into a value: blockwatcher_types::SecretRef does that, called by webhook.rs per delivery (see webhook).

Key types and traits

Every module’s own Config struct (webhook.rs, script.rs, log.rs) is private, not pub: reachable only through its ModuleRegistry::factory, documented via its own registry_examples/*.json rather than as a public type. There is no crate-local error or classification type: SinkError/ErrorClass/Classify are all blockwatcher-ports concepts this crate only uses.

The Sink port and what “delivered vs. dead-lettered” hinges on

Sink is one method (crates/blockwatcher-ports/src/sink.rs:6-12):

#![allow(unused)]
fn main() {
#[cfg_attr(feature = "testing", mockall::automock)]
#[async_trait]
pub trait Sink: Send + Sync {
    /// Deliver one event. The ENGINE owns retry/backoff/dead-letter policy;
    /// the sink reports classified errors and does not retry internally.
    async fn deliver(&self, event: &SinkEvent) -> Result<(), SinkError>;
}
}

SinkError (blockwatcher-ports/src/error.rs) is Delivery { message, class: ErrorClass } or InvalidConfig { message }, and its Classify impl reports class for Delivery and always Permanent for InvalidConfig. So structurally, “delivered” is deliver returning Ok(()); “dead-lettered” is the engine’s deliver_with_retry exhausting its budget on a stream of Errs whose ErrorClass values it read straight off each module’s own return: every classification decision a sink module makes here is a direct input into that outcome, one module never sees the other’s retries, and none of the shipped modules ever emits ErrorClass::RetryNarrower (that class exists for RPC calls whose request can shrink, not deliveries).

Every delivery attempt, across every sink module, funnels through the same shape:

flowchart LR
    attempt["deliver(event)<br/>one attempt"] -->|"Ok(())"| delivered["delivered"]
    attempt -->|"Err(SinkError)"| classify{"ErrorClass"}
    classify -->|"Transient / RateLimited"| retry["engine retries<br/>with backoff"]
    classify -->|"Permanent"| dead["dead letter"]
    retry -->|"attempts exhausted"| dead
    retry --> attempt

Per-sink delivery semantics

webhook

Config (webhook.rs, deny_unknown_fields): url_secret: String (an env:NAME reference, resolved fresh at every delivery: no resolved copy outlives one request), headers: BTreeMap<String, String> (default empty, plain values), header_secrets: BTreeMap<String, String> (default empty, env:NAME references resolved fresh at every delivery like url_secret; a name also present in headers refuses the write), timeout_ms: u64 (default 10_000, rejected if 0 at construction since it would make every attempt time out immediately), body_template: Option<String> (a minijinja template validated at write time; absent keeps the canonical SinkEvent JSON unchanged). WebhookSink itself stores the unresolved SecretRefs, never a resolved URL or header value (webhook.rs).

deliver (webhook.rs) resolves the secret, parses it as a URL, builds body = canonical_body(event)?, and sends POST <url> with Content-Type: application/json plus every configured header, body exactly the canonical JSON with no wrapping or renaming. Redirects are disabled outright: following one would re-send the secret-addressed request to a host the operator never named, so any 3xx is a visible, permanent failure (webhook.rs, classify_status). Retry classification (classify_status, webhook.rs):

The timeout_ms config bounds one HTTP attempt via reqwest::Client::builder().timeout(...); it is set once, at construction, and is orthogonal to blockwatcher-core’s own retry backoff between attempts.

An optional config field, body_template (optional string, absent by default), is a minijinja template for the request body. It renders against a JSON value (canonical_value, lib.rs) that shares SinkEvent’s serde shape with the wire body but is serialized independently of it: canonical_body serializes the event directly, never through this serde_json::Value, because routing it through Value would re-sort top-level keys and break the byte-pinned wire contract (lib.rs). So a template and the default wire body always see identical field names: for a match, top-level type, id, monitor, network, and event (with event.kind, event.name, event.fields, event.cursor); for a retracted event, type and match_id; for a digest, type and matches, each element shaped like a match without its type. A template branches per shape with a guard — {% if type == "digest" %}…{% endif %} — whose digest branch renders only against digest events. Validation and rendering split by when each runs:

• Write time (Registry::factory, webhook.rs, calling compile_body_template("webhook", source) in lib.rs): a configured body_template is parsed and rendered against three synthetic events — a match, a retracted event, and a two-match digest (synthetic_validation_events, lib.rs) — before construction returns, so every shape a sink can be handed is exercised at the write. A syntax error or a filter that does not exist fails the write as SinkError::InvalidConfig (an API 422), message-prefixed invalid webhook body_template:. A configuration mistake surfaces at the write that introduced it, never as a dead-lettered delivery later.
• Delivery time (deliver, webhook.rs, calling render_body_template in lib.rs): a template that passed write-time validation still renders fresh per real event. A render failure there is a permanent delivery error, body_template render failed: {e}: deterministic over the same payload, so a retry cannot change the outcome.
• Undefined fields render empty. The template environment sets UndefinedBehavior::Lenient: a field the template reads that a given event lacks (a retracted event has no network, for instance) renders that lookup as empty rather than failing the render. A template is presentation over the canonical event, and a missing decoration must not dead-letter a correct match. Leniency covers plain printing and iteration only: piping an undefined field through a filter still errors (e.g. {{ event.fields.map.amount.uint | int }} fails write-time validation against the synthetic events, neither of which has that field) — | default(...) before the filter is the escape hatch.
• Absent body_template keeps the wire byte-identical. With no template configured, deliver sends canonical_body(event) unchanged.

A template replaces the canonical wire contract with an operator-authored one — the canonical body remains the default and the only shape the compatibility test pins.

A worked example: the template

{"text": "{{ type }} on {{ network }}"}

validates at write time against both synthetic events, and against the pinned match fixture (monitor m, network net) renders the POST body:

{"text": "match on net"}

type and network above are operator-known enum-shaped strings, safe to interpolate bare inside hand-written quotes. event.fields and event.name are decoded chain data, not operator-controlled: a value there can carry a quote or brace, so interpolating one into a JSON payload without escaping can produce malformed JSON or let the value inject structure into the destination payload (a forged Slack block, for instance). Pipe a chain-derived value through |tojson instead of wrapping it in hand-written quotes — tojson supplies its own quoting, so it replaces the surrounding "..." rather than nesting inside them:

{"text": {{ type | tojson }}}

renders {"text": "match"}; used on a value that actually contains a quote, the equivalent hand-written-quotes form would break the JSON.

script

Config (script.rs, deny_unknown_fields): command: String, args: Vec<String> (default empty, passed verbatim), timeout_ms: u64 (default 30_000), body_template: Option<String> (absent by default). No env field: the child inherits the process’s own environment; no working-directory field.

deliver (script.rs) spawns tokio::process::Command::new(&command).args(&args) with stdin piped, stdout discarded, stderr captured (a 4 KiB tail), and kill_on_drop(true). With no body_template configured, canonical_body(event)? is written to the child’s stdin, not argv and not an environment variable; with one configured, render_body_template’s rendered string is written instead. Either way stdin is shut down once the write completes. The whole write-plus-wait is wrapped in tokio::time::timeout(self.timeout, ...); on elapse the child is killed via the drop guard and the outcome is Transient (“script timed out after {ms}ms”). Exit-code classification is an explicit, operator-facing contract, not transport inference:

body_template is the same minijinja mechanism webhook uses (see webhook for the context shape, the digest type guard, and the lenient-undefined rules), through the same shared compile_body_template/render_body_template functions in lib.rs: write-time validation against the three synthetic events, invalid script body_template: ... on a syntax or filter error at the write, and a permanent body_template render failed: {e} at delivery if a real event’s shape ever slips past that validation. The one difference is what the rendered text becomes: a request body has to stay valid JSON for the receiving webhook, but a script’s stdin has no inherent structure to protect — an operator whose script parses stdin as JSON (or any other structured format) still owns the same |tojson-style escaping discipline webhook’s worked example shows, since decoded chain data is not operator-controlled and can carry a quote or brace.

The sink-script-monitor example configures a body_template that renders each match as a Match |- Transfer from … to … of … USDC line (and a digest as one Digest |- … line plus one |- … continuation line per bundled match), so the example script simply appends the rendered text — no jq or other post-processing needed on the receiving end.

log

Config (log.rs): body_template: Option<String> (absent by default) is its one field; deny_unknown_fields still rejects any other key as a boot-time typo rather than a silent ignore. LogSink itself carries the compiled template (Option<minijinja::Environment<'static>>), so unlike webhook/script it is no longer a unit struct — tests construct it via LogSink::default(). line (log.rs) builds the emitted line: with no template, canonical_body(event)? plus a trailing newline; with one, render_body_template’s rendered string plus the same trailing newline. deliver writes that line directly to tokio::io::stdout(), under a process-wide AsyncMutex<()> that serializes concurrent LogSink instances so two deliveries can never interleave mid-line. It is not routed through the tracing/log facade at all.

body_template here is the same mechanism documented under webhook and script, through the same shared compile_body_template/render_body_template functions in lib.rs: write-time validation against the three synthetic events, an invalid log body_template: ... message on a syntax or filter error at the write, and a permanent body_template render failed: {e} at delivery if a real event’s shape ever slips past that validation.

Delivery can fail, narrowly: an I/O error writing or flushing stdout (a broken pipe, most plausibly) maps to Transient rather than Permanent (a deliberate loss-aversion choice, since an under-classified retry costs one wasted attempt while an over-eager Permanent would dead-letter a match a momentarily-blocked consumer would otherwise have accepted), in the transient helper (log.rs).

The canonical_body wire-contract test

Two layers pin the same shape:

• A crate-internal unit test, canonical_body_match_is_tagged (lib.rs), calls canonical_body directly against a fixed SinkEvent::Match and asserts the resulting bytes, parsed back to JSON, equal a hand-written literal that includes "type": "match". A second test, canonical_body_retracted_is_tagged, pins {"type":"retracted","match_id":"…"}. A third, canonical_body_digest_is_tagged, pins a SinkEvent::Digest of two matches to a literal with "type": "digest" and a matches array of their own canonical objects, each rendered exactly as the lone-match literal renders one. Together they pin both the field shape of blockwatcher_types::Match/DecodedEvent/SinkEvent and the exact hash MatchId::derive produces for that fixed input ("14e563bd9d376e738be5e13e4054327c").
• An integration-level echo of the identical literal through a real HTTP round trip, a_delivery_posts_the_canonical_body_exactly_once (tests/sinks/webhook.rs): a mock axum server records what WebhookSink::deliver actually sent, and the test asserts the received body equals the same pinned literal, byte for byte.

Every other body-comparing test in the crate re-serializes the input and compares against that: those tests would keep passing through a shape change that these two would catch. That is the point of keeping one literal assertion rather than none: the sink modules render SinkEvent to something outside the process, which makes its serde shape an external wire contract the moment any one of them ships, and a hand-written literal is the only form of assertion that a coordinated rename (changing both the struct and every call site that serializes it) cannot silently slip past.

Neighbours

blockwatcher-sinks depends on, in production:

• blockwatcher-types: vocabulary crate (SinkEvent, SecretRef)
• blockwatcher-ports: the Sink trait this crate implements once per module, and ErrorClass/SinkError
• async-trait: required to implement the async fn-bearing Sink trait
• serde: (de)serialization derive for each module’s Config
• serde_json: the canonical wire format canonical_body produces
• tokio (io-util/io-std/process/time/macros features): stdin piping and process timeout (script), async stdout (log)
• reqwest: the webhook module’s HTTP client, this crate’s one check-dep-graph.sh family exemption
• minijinja (default-features = false, builtins + serde + json features): the webhook module’s optional body_template renderer; json gates the |tojson escaping filter separately from builtins and depends on serde_json, already in the tree, so memo-map remains the only wholly new transitive dependency — no network, TLS, or async-runtime dependency of its own

and, in [dev-dependencies] only:

• tokio (macros/rt/net/time features): async tests and the mock server’s listener
• axum: a mock HTTP receiver webhook’s tests assert against, not a client
• tempfile: filesystem fixtures for script’s tests
• blockwatcher-core: exercises a real sink through blockwatcher-core’s actual Engine/sink-worker/deliver_with_retry machinery rather than a reimplemented harness; blockwatcher-core itself may never depend on a module crate, so this dev-only edge crosses that seam from the module side instead, and is how the retry-then-dead-letter contract from Delivery guarantees gets proven end to end for webhook and script specifically
• blockwatcher-ports (fakes feature): fake ports for the fake source/decoder/matcher/storage blockwatcher-core’s test engine needs alongside a real sink
• blockwatcher-testkit: shared test scaffolding
• metrics: asserted against in the engine-delivery tests’ metric checks
• tracing: structured logging the log-capture tests read
• tracing-subscriber: captures log output for those assertions (e.g. confirming a resolved webhook URL never reaches a log line)

The following crate depends on it directly (per the dependency table):

• blockwatcher-embed: the composition façade registers every sink module into the engine’s module catalog; the blockwatcher binary reaches this crate only through embed

Reading the source

1. Start at lib.rs: the crate doc comment states the whole contract in one sentence, and canonical_body is the one function every sink module below calls before it sends anything anywhere.
2. registry.rs: sinks::get_all, and its self-verifying test, which constructs every registered module from its own documented example config and panics on an unmatched name, the mechanism that keeps this enumeration and the registry_examples/*.json files from drifting apart.
3. webhook.rs: WebhookSink, classify_status, transport_error; read the module doc comment first for why redirects are refused and for the split between plain headers and secret-referenced header_secrets.
4. script.rs: ScriptSink, its exit-code contract (EXIT_TEMPFAIL), and the timeout-then-kill path; read the module doc comment for the inherited-environment and no-working-directory decisions.
5. log.rs: LogSink, the process-wide STDOUT_LOCK, and why a broken pipe classifies Transient rather than Permanent.