Crate weir_sink_sdk 

Sink trait and error contract for building sinks for the weir daemon.

A weir sink is a downstream commit target. The daemon’s drain reads batches of records out of sealed write-ahead-buffer segments and hands them to a sink to commit to a database, HTTP endpoint, object store, etc. Implement Sink (and SinkError for your error type); the drain retries transient failures with backoff and dead-letters permanent ones.

A sink whose commit can never fail picks type Error = std::convert::Infallible — no error type to hand-roll, no is_transient to write:

use weir_sink_sdk::{CommitResult, Payload, Sink, SinkBatch, SinkHealth};

struct StdoutSink;

impl Sink for StdoutSink {
    type Error = std::convert::Infallible; // this commit never returns Err
    async fn commit(&self, batch: SinkBatch) -> Result<CommitResult, Self::Error> {
        let batch = batch.into_records();
        for r in &batch {
            println!("{} bytes", r.len());
        }
        Ok(CommitResult::new(batch, Vec::new()))
    }
    async fn health(&self) -> SinkHealth {
        SinkHealth::Healthy
    }
}

A sink that can fail returns a SinkError that classifies the failure as transient (retried) or permanent (dead-lettered). Reach for the ready-made BasicSinkError before writing your own error type.

Idempotency contract

The drain guarantees at-least-once delivery per segment, not per record. If the daemon crashes after a partial commit but before recording the segment as confirmed, commit is called again with the full segment — including records already committed. The same whole-segment replay also occurs when a segment is stranded (its transient-retry budget is exhausted while the sink is unavailable) and later auto-resumed once the sink recovers: the resume reprocesses from the start, not from the last durably-committed sub-batch. Implementations must handle duplicates gracefully — upsert, INSERT IGNORE, an Idempotency-Key, whatever the downstream offers. This is the explicit durability trade-off, not a protocol weakness.

Which key to dedup on

Two handles arrive on every SinkBatch, and they are not interchangeable:

• SinkBatch::record_ids — one RecordId per record, derived from its WAB coordinate. Distinct for two records carrying identical bytes, and unchanged when sink_max_batch_size changes between two delivery attempts. Use this when a duplicate is expensive.
• SinkBatch::dedup_token — one DedupToken for the whole batch, derived from its bytes. Simpler, but it names the batch, so it moves when the drain re-splits a segment at different boundaries. Fine for an archive; not for a billing row.

Do not dedup on a hash of the payload. It is not a weaker version of the two above, it is a different thing that loses data: a producer emitting repetitive records — a heartbeat, a fixed-shape metering event — hands the downstream the same key for genuinely distinct events, and a correct idempotent endpoint keeps the first and discards the rest. weir acked those records and delivered them; the key threw them away.

Running your sink in the daemon

This crate lets you implement and test a sink against a stable trait, independent of the daemon’s internals. Running it is a separate matter: the released weir-server binary wires only the built-in sinks selected by the sink_type config. There is no dynamic plugin or registration path yet — to run a custom sink today you build a weir-server with your sink compiled into the sink-selection path (effectively a small fork). A first-class entry-point for downstream sinks is a candidate for a future minor release; because it is purely additive it would be a SemVer-compatible change.

Structs

BasicSinkError

A ready-made SinkError for sinks that don’t need a bespoke error type — a message plus a transient/permanent classification. Construct it with BasicSinkError::transient (the drain retries the segment) or BasicSinkError::permanent (the records are dead-lettered).

CommitResult

The result of a successful Sink::commit.

DedupToken

A content-derived, batch-scoped idempotency handle for one Sink::commit call.

Payload

Opaque record payload bytes (re-exported from weir-core). A newtype over ref-counted bytes::Bytes that derefs to [u8], so clones through the drain are O(1). Sinks normally receive Payloads from the drain; to build one yourself (e.g. in a unit test) use Payload::copy_from_slice(bytes), Payload::from(&b"..."[..]), or Payload::from(vec_of_u8). Opaque payload bytes — a newtype over ref-counted bytes::Bytes so clones through the drain / sink path are O(1) instead of heap copies.

RecordId

A per-record identity: unique even across records carrying identical bytes.

SinkBatch

One batch of records on its way to Sink::commit, with the DedupToken the drain derived from it.

Enums

SinkHealth

Coarse health signal from Sink::health.

Traits

Sink

A downstream commit target for weir records.

SinkError

An error returned by Sink::commit.