Delivery guarantees

What mq-bridge promises about duplicates and loss, what it needs from you, and which source/sink combinations give you which guarantee.

Short version. mq-bridge is at-least-once. A message is acked only after the output chain reports success, so nothing is lost on a crash — but a crash between the write and the ack replays the message. Combine at-least-once delivery with an idempotent write at the sink and you get effectively-once: exactly-once effects — the record lands once no matter how many times it is delivered. Nothing here is exactly-once delivery. Everything below is about how to arrange that.

Enabling effectively-once

There is no global exactly_once switch. The guarantee follows from ordinary endpoint configuration: provide a replay-stable identity when the source does not already have one, then configure an idempotent sink write. At startup, mq-bridge inspects the route and logs the inferred guarantee as at-most-once, at-least-once or effectively-once; it does not silently change how the sink writes data. The same value is available in code as Route::delivery_guarantee(), a DeliveryGuarantee.

To make the guarantee a requirement rather than a log line, set required_delivery on the route. A route whose configuration cannot meet it fails at startup instead of running with less:

orders_to_postgres:
  required_delivery: effectively_once   # at_most_once | at_least_once | effectively_once
  input:  { kafka: { topic: "orders", url: "localhost:9092" } }
  output:
    sqlx:
      url: "postgres://localhost/shop"
      insert_query: "INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO NOTHING"

at-most-once is inferred for inputs that forget a message once handed over: zeromq, core NATS (no_jetstream: true), MQTT with qos: 0, HTTP with fire_and_forget: true, a websocket listener, and a grpc dynamic client (descriptor set or reflection). A crash in the route loses what it held.

This table is advice on how to configure an idempotent sink — it is not a list of switches the engine flips. A plain INSERT is not idempotent; you make it so by writing the conflict clause yourself.

The right-hand column only describes the informational log line mq-bridge emits when a route starts. The inference is deliberately conservative: it recognises the two forms it can read off the configuration with certainty, and says at-least-once for everything else. A route in the third or fourth row is still effectively-once in practice — the sink absorbs the replay either way. Do not read at-least-once in the log as “this route creates duplicates”.

For a source without stable identity, derive one once and consume it at the sink:

input:
  middlewares:
    - id: "${payload:order_id}"       # writes metadata mqb.id
  file: { path: "orders.jsonl" }

output:
  sqlx:
    url: "sqlite://orders.db"
    table: orders
    insert_query: >
      INSERT INTO orders (id, body)
      VALUES (${metadata:mqb.id}, ${payload:body})
      ON CONFLICT (id) DO NOTHING

The target column must actually have a PRIMARY KEY or UNIQUE constraint. DO NOTHING gives insert-once semantics; an appropriate DO UPDATE clause gives convergent upsert semantics.

What effectively-once actually requires

It is four separate properties, and they fail independently:

1. Deterministic identity — a key for the record that is the same on every replay. Without this, nothing downstream can recognise a duplicate.
2. An idempotent (or transactional) write — the sink must absorb a repeat of the same key.
3. A durable source position tied to the write — so recovery resumes at the right place.
4. Fencing — a restarted or duplicated instance must not race the old one.

mq-bridge gives you 1 and 2 across most of the matrix, 3 for two sources, and does not implement 4. See What is not provided.

Sources: what identity you get

message_id is a u128. When the source can derive it from the record it carries, it is stable across replay and usable as a deduplication key. When it cannot, it defaults to a fresh fast_uuid_v7 value per read — which identifies this delivery, not that record, and is therefore useless for dedup across a restart.

Two consequences worth internalising:

• A file or object_store source cannot deduplicate on message_id. Re-reading the same file produces entirely new ids. Derive a business key instead — see Giving a source an identity.
• mq-bridge writes the message id on the sink side for Kafka, NATS, Redis Streams, MQTT 5 and AMQP (as the native message_id property), so a mq-bridge → broker → mq-bridge hop preserves identity end to end even when the broker itself has no id concept.

Giving a source an identity

For the sources in the bottom rows, the id middleware derives one from the message itself and stores it in the mqb.id metadata key:

input:
  middlewares:
    - id: "${payload:order_id}"
  file: { path: "orders.jsonl" }

Re-reading the same record now yields the same mqb.id, even though each read mints a fresh message_id. Three properties make it the right carrier:

• It is a string, so it holds the business key in its original form. message_id is a u128 and could only hold a hash of it — which a sink cannot then use as a readable _id.
• It propagates. mqb.id sits deliberately outside the mqb.src.* namespace, so publishers do not strip it: an identity describes the record, not the hop it arrived on. Kafka already forwards every non-mqb.src.* metadata entry as a header, so it survives that hop today.
• It is opt-in. No id entry means no wrapper and no cost.

Once set, read it anywhere metadata is available — ${metadata:mqb.id} in a deduplication key, a Kafka partition_key, or a handler.

Watch the ordering. Consumer middlewares wrap in reverse, so the entry closest to the end of the list touches an incoming message first. Anything reading mqb.id must be listed before the id that produces it:

input:
  middlewares:
    - deduplication: { store: "sled:///var/lib/mqb/dedup", ttl_seconds: 3600, key: "${metadata:mqb.id}" }
    - id: "${payload:order_id}"
  file: { path: "orders.jsonl" }

Reversed, mqb.id is still unset when deduplication reads it, and dedup silently falls back to message_id.

Identity is not a version. mqb.id answers “which record is this”, not “which revision”. Do not point a CDC route’s dedup key at it: every update to a row shares one business key, so they would collapse into one. That is why postgres_cdc builds its message_id from schema.table + key + **lsn** — dedup on a change stream needs identity and version, and should keep defaulting to message_id. Use mqb.id for sink keying and correlation.

id is input-only; on an output it is a startup error rather than a silent no-op. The key is set only when every selector in the template resolves — a partial render such as "acme-" would hand one identity to every message missing the field, so it is dropped instead.

Sinks: how a duplicate write is absorbed

Picking a combination

Any source → a database sink. The easy case. You do not need a replayable source position at all — the sink’s unique constraint is already shared across every writer and is the authority. Give it a deterministic key (id_field, or a ON CONFLICT column) and you are done.

Any source → files or object storage. A filesystem has no unique constraint, so this route needs a replayable source position and therefore works only from kafka or postgres_cdc. See Files & object storage.

A source → a broker sink (Kafka, NATS, MQTT, …). The sink cannot deduplicate beyond what the table above lists. Either filter before it with the deduplication middleware, or accept at-least-once and make the downstream consumer idempotent.

The matrix

What a route achieves, by what its input offers and what its output can absorb:

A handler, or any side effect it performs, sits outside this table: see Handlers. A custom endpoint (or plugin) is placed in it by its own declaration — see Declaring a delivery guarantee; one that declares nothing counts as an acknowledging source and a non-idempotent sink.

A route with a handler. Sink-side idempotency happens after the handler runs. If that matters, see Handlers.

Deduplication & idempotent writes

For ETL, at-least-once delivery plus an idempotent write gives you effectively-once: a replayed or retried record must not create a duplicate row. The most robust place to enforce this is the sink database’s own unique constraint — it is already shared across every writer, so no extra state store is needed. Both database sinks lean on this instead of an application-side cache.

MongoDB — id_field

Point id_field at a top-level payload field and its value becomes the document _id. Alternatively, use the template form id_field: "${metadata:mqb.id}" to consume an identity produced by the input id middleware. Re-inserting the same business key then hits the unique _id index and is treated as an idempotent success (the duplicate is skipped, not errored). Only _id works this way: a duplicate on another unique index is a refused document and fails as non-retryable, so it reaches a dlq if one is configured.

orders_to_mongo:
  input:  { kafka:   { topic: "orders", url: "localhost:9092" } }
  output:
    mongodb:
      url: "mongodb://localhost:27017"
      database: "shop"
      collection: "orders"
      format: json
      id_field: "order_id"   # payload {"order_id": "A-1", ...} → _id = "A-1"

The equivalent configuration using the shared identity carrier is:

input:
  middlewares:
    - id: "${payload:order_id}"
output:
  mongodb:
    # ... connection and collection ...
    id_field: "${metadata:mqb.id}"

A plain id_field preserves the payload value’s BSON type; the template form renders a string and accepts only replay-stable payload or metadata tokens.

The field’s JSON type is preserved (a number stays a BSON integer). The payload must be JSON and contain the field, otherwise the message is dead-lettered rather than written with a random _id — silently minting one would defeat deduplication. Use id_field on sink collections only: a business-key _id is not compatible with the delete competing-consumer mode, which requires a UUID _id.

SQL (sqlx) — ON CONFLICT / ON DUPLICATE KEY

The insert_query is user-supplied, so you write the dialect’s upsert directly. This requires a pre-existing UNIQUE/PRIMARY KEY on the key column, and is incompatible with bulk_copy (COPY cannot express ON CONFLICT) — so you trade peak throughput for deduplication.

# PostgreSQL — insert if absent (drop duplicates):
insert_query: "INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO NOTHING"

# PostgreSQL — upsert (last write wins):
insert_query: "INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO UPDATE SET body = EXCLUDED.body"

# MySQL / MariaDB:
insert_query: "INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON DUPLICATE KEY UPDATE body = VALUES(body)"

# SQLite:
insert_query: "INSERT INTO orders (id, body) VALUES (${payload:id}, ${payload:body}) ON CONFLICT (id) DO NOTHING"

A plain INSERT without a conflict clause instead fails the row as a non-retryable error: a configured dlq captures it, and without one it is logged and dropped. Add the conflict clause when replays are expected. ${payload:field} binds a typed value from the JSON payload; ${metadata:key} binds a metadata string.

ClickHouse — ReplacingMergeTree

ClickHouse has no unique constraints; dedup is a table-engine property. Create the target as ReplacingMergeTree(version) keyed by your business key via ORDER BY, using a monotonic column as the version (e.g. an ingest timestamp, or postgres.lsn from a CDC source). ClickHouse collapses rows with the same sort key at merge time, keeping the highest version; read with FINAL (or argMax) to see the deduplicated result:

CREATE TABLE orders (id UInt64, body String, version UInt64)
ENGINE = ReplacingMergeTree(version) ORDER BY id;
-- mq-bridge just inserts rows; duplicates for the same id collapse on merge.
SELECT * FROM orders FINAL;

ReplacingMergeTree deduplicates by business key at merge time. Separately, ClickHouse also dedupes identical re-inserted blocks natively: a retried send_batch that resends the same block is dropped server-side (default one-hour window) — insert_deduplication_token lets you make that explicit, but mq-bridge does not set one, so rely on ReplacingMergeTree for logical dedup and treat block-level dedup only as retry-safety.

Keying a sink on the source position

A source with a replay position already identifies every record: Kafka by topic, partition and offset, Postgres CDC by slot, LSN and ordinal. Keying the sink row on that position needs no business key and no extra state — it is the Kafka Connect JDBC sink’s pk.mode=kafka:

kafka_to_postgres:
  input: { kafka: { topic: "events", url: "localhost:9092" } }
  output:
    sqlx:
      url: "postgres://localhost/shop"
      insert_query: >
        INSERT INTO events (topic, part, off, body)
        VALUES (${metadata:mqb.src.kafka_topic}, ${metadata:mqb.src.kafka_partition},
                ${metadata:mqb.src.kafka_offset}, ${payload:body})
        ON CONFLICT (topic, part, off) DO NOTHING

A SQL insert_query or Mongo id_field that reads mqb.src.kafka_*, postgres_*, mongodb_*, file_* or sqlx_* turns source_metadata on for the input, as positional file names do. Over an input without a replay position those keys are absent, so such a route is not reported as effectively-once. Prefer a business key when records have one: a position identifies this copy of the log, so re-publishing the same data to a new topic writes it again.

Postgres CDC — deterministic id + postgres.key

The postgres_cdc source resumes from the slot’s durable confirmed_flush_lsn. In-band standby feedback is asynchronous and is not flushed when the stream stops, so the last acks are made durable by the consumer’s Drop, which stops the stream and advances the slot synchronously. Re-delivery is therefore avoided only on a restart that actually runs that teardown — a host process that exits without dropping the route (or one on a current-thread Tokio runtime, where the blocking advance is skipped) replays everything since the last asynchronous feedback tick. Set checkpoint_store to a file:// path for a second, per-ack durable position that survives teardown regardless. Treat the source as at-least-once and make the sink idempotent.

Each insert and update carries the row (so the primary key is in the payload; a delete carries the key columns, and an update that changes the key names the previous one in postgres.old_key), postgres.lsn (a monotonic version), postgres.operation/schema/table, and — when the table has a primary key / replica identity — postgres.key (the key value; a truncate event names no row and has none). The event’s message_id is a stable hash of schema.table + key + operation + lsn + position in the transaction, so a replayed change deduplicates through the deduplication middleware, and Mongo id_field or a SQL ON CONFLICT on the key column make the sink write idempotent. Use postgres.lsn as the version to drop stale replays (... DO UPDATE ... WHERE excluded.lsn > orders.lsn).

Several changes to one primary key within a single transaction share that transaction’s commit LSN but differ in operation or position, so each gets its own message_id and all are delivered.

The deduplication middleware

The middleware is a complement, not a replacement: it filters duplicates before the sink, and it is consumer-only (configuring it on an output is a startup error). Prefer the sink constraint for multi-writer ETL; reach for the middleware when the sink has no constraint to lean on, or when you need to keep duplicates away from a handler.

Three things decide whether it actually works:

• key. The default keys on message_id, which many sources mint fresh per read (see the source table) — the route then looks configured and dedupes nothing; for the sources that always do this, startup logs a warning. Set key to a business key, either directly ("${payload:order_id}") or via "${metadata:mqb.id}" when an id middleware already derived one. An unresolvable template also falls back to message_id with only a warning, so a typo fails silently. Content hashing is deliberately not offered as a fallback: two legitimately equal payloads would collapse.
• Ids from producers you do not trust. A string message_id that is not a UUID or a number is folded to 128 bits with FNV-1a, which is fast but not collision-resistant: a producer who chooses its ids can craft one that collides with another message’s, and deduplication on message_id then drops that message. Set MQB_ID_HASH=sha256 (or call set_string_id_hash) for such inputs. Switching changes every hashed id, so start with an empty dedup store. Numeric strings are read as numbers, so "123" and "0123" are one id.
• store. sled is single-instance. Point it at a shared MongoDB or SQL deployment for anything scaled out. The shared collection or table defaults to mqb_dedup_<route name>, so replicas must run under the same route name — or name the collection/table explicitly in the URL (not possible for SQLite, which has no path slot for it).
• ttl_seconds. A key is remembered for this long after it was processed. It must exceed the longest time in which the same record can come back: broker redelivery and retry limits, a producer’s retry horizon, and any replay an operator may trigger from a DLQ. A copy arriving after that is processed again. Longer costs only storage.

input:
  middlewares:
    - deduplication:
        store: "mongodb://localhost:27017/shop"
        ttl_seconds: 86400
        key: "${payload:order_id}"
  kafka: { topic: "orders", url: "localhost:9092" }

middlewares sits beside the endpoint type, not nested inside it. Requires the dedup feature.

How it orders its steps. A key is claimed when its message is read, marked processed after the sink accepted the write — before the source is acked — and released when the write failed. A copy that arrives while another is still in flight waits for it rather than being acked on its strength: if the first write then fails, the copy is processed; if it succeeds, the copy is dropped. A claim is renewed while its delivery is in flight, so a slow sink does not lose it; a claim whose holder died lapses after five seconds. What each crash window does:

The second row is why the middleware stays at-least-once: its marker is not committed in the sink’s transaction. Only an idempotent sink closes that window.

Replaying a reply. On a request/reply route (HTTP, gRPC, NATS or memory requests) a duplicate is normally acked with nothing to say, so a caller retrying a request gets an empty answer. replay_response: true stores each reply next to its marker and answers a duplicate with it — the idempotency-key pattern of payment APIs. The reply is kept for the whole TTL, in the store, so size the store for it and consider what the replies contain. Off by default.

MongoDB — branch on insert vs. duplicate (report_outcome)

Sometimes you need to act on whether a record was newly inserted or already existed — enrich only fresh rows, or reply with the existing entry for duplicates. Set report_outcome: true and the Mongo publisher returns the message tagged with metadata mongodb.outcome = inserted (fresh write) or existed (dup-key on the unique _id). Wrap it in a request endpoint to forward that tagged message into a switch that routes on mongodb.outcome:

orders_upsert_branch:
  input: { kafka: { topic: "orders", url: "localhost:9092" } }
  output:
    request:                       # calls `to`, forwards its response to `forward_to`
      to:
        mongodb:
          url: "mongodb://localhost:27017"
          database: "shop"
          collection: "orders"
          format: json
          id_field: "order_id"     # deterministic _id → insert-if-absent
          report_outcome: true     # → mongodb.outcome = inserted | existed
      forward_to:
        switch:
          metadata_key: "mongodb.outcome"
          cases:
            inserted: { ref: "enrich_new_order" }   # e.g. build entry X, reply
            existed:  { ref: "handle_duplicate" }   # e.g. reply with parts of X
          default:   { file: { path: "orders-unrouted.jsonl" } }

report_outcome is sink-only and pairs with id_field; without a deterministic _id there is no duplicate to detect. Do not also set request_reply: true — that switches the publisher to the reply-polling path, which never reports an outcome and times out with nothing answering. Left unwrapped by request, the tagged message is returned as the route’s response as usual.

Keep the default arm: when the Mongo send errors, request forwards the original message, which carries no mongodb.outcome, and a switch with no default drops unmatched messages.

The outcome is only truthful on the first attempt. If the process dies after the insert committed but before the branch’s downstream send committed, the replay hits a duplicate key, Mongo answers existed, and a genuinely-new record takes the duplicate branch. Write the existed branch as “may or may not have been handled — check and repair”, not “definitely already done”.

Read by key, then branch (lookup)

To branch on whether a record already exists without writing it, read it first. The lookup middleware asks a read-by-key endpoint (MongoDB find, SQLx lookup_query, HTTP) and sets lookup.found, which a switch routes on:

orders_enrich_branch:
  input: { nats: { url: "nats://localhost:4222", subject: "orders" } }
  output:
    middlewares:
      - lookup:
          from:
            sqlx:
              url: "postgres://app@localhost/crm"
              table: "customers"
              lookup_query: "SELECT id, name, tier FROM customers WHERE id IN (${payload:customer_id}::int)"
          into: customer
    switch:
      metadata_key: "lookup.found"
      cases:
        "true": { ref: "known_customer_orders" }
      default: { ref: "new_customer_orders" }

A read is not a reservation: two messages for the same new key can both see false. When exactly one of them must win, insert with report_outcome (above) instead.

Files & object storage — name_by

Check first whether you need any of this. If your records already carry a business key — an id field in the payload, or mqb.id from the id middleware — then a key-addressed sink (Mongo id_field, SQL ON CONFLICT) already gives you effectively-once, and the order objects happen to land in does not matter. Positional naming below is for the case where the sink itself has to recognise a replay, or where a downstream reader depends on replay order. Turning it on when you do not need it only adds restrictions.

A filesystem has no unique constraint, so the file and object_store sinks get replay safety from the name they write under — which works only as far as the source repeats its positions: a re-read that numbers the same records differently produces different names, and different names are not a replay to anything downstream. name_by picks the scheme:

Under source_position the sink groups each batch into runs of consecutive source positions and writes one immutable part per run, named for the range it covers:

kafka_to_s3:
  input:
    kafka: { topic: "orders", url: "localhost:9092", source_metadata: true }
  output:
    object_store:
      url: "s3://my-bucket/orders"
      name_by: source_position   # → part-orders-<partition>-<start>-<end>.jsonl (zero-padded)

auto applies to object_store only. Either scheme leaves that sink writing whole immutable objects — write_time one per flushed batch, source_position one per contiguous run within it — so only the name and the grouping change, and deriving them is safe. The file sink is different: source_position turns path from a file into a directory of part files, which is a change of structure rather than of name, so a file sink stays on write_time until you ask for parts explicitly.

All of this belongs to the source_position path. Before its first write such a sink lists the prefix once and parses the ranges out of the part names already there, so a replay skips them without re-encoding or re-uploading. Filtering is per record, not per batch, so batch boundaries are free to differ across restarts — that is what makes it work without a checkpoint protocol, and it is why the listing cannot wait for a name to collide: a replay on different boundaries names different objects and collides with nothing. A repeat that does land on the same name is caught anyway, since the PUT uses PutMode::Create and treats AlreadyExists as success. A write_time sink never lists — its names are unique per write, so there is nothing to recognise; on the source_position path a fresh prefix pays one empty listing per publisher. A listing that fails is logged rather than failing the batch, leaving same-name-only deduplication until it succeeds: a transient failure is retried on the next batch, a denied ListObjects is not retried. Local files stage to a temp name, fsync, then rename (atomic within a filesystem); object stores have no atomic rename, so a single PUT under the final name is the commit.

This needs a replayable source position. Today that means:

A route whose output resolves to source_position turns source_metadata on for its input automatically; set it explicitly only when you want the mqb.src.* keys for something else. An input from the table above is what auto looks for; anything else keeps the sink on write_time. An explicit name_by: source_position over an input that has no position is rejected when the route starts, because there is then no way to honour what was asked for. NATS and AMQP also accept source_metadata and emit provenance keys, but a subject or routing key is not a replayable offset, so they cannot drive positional naming.

Under write_time, an object_store sink names objects in write order, and key order is the only order a bucket has. At concurrency: 1 that is source order; above it the name is minted inside the worker pool, so it is arrival order, and replaying a change stream through the bucket can reorder updates to the same key. For an input that carries a replay position, auto already avoids this. For one that does not — NATS, MQTT, HTTP, gRPC, ZeroMQ, IBM MQ, Redis Streams — there is no positional name to fall back to, and concurrency: 1 is the only remedy. A concurrent route into a write_time object-store sink logs a warning at startup either way: over an input with a position it points at name_by: source_position, over one without it at concurrency: 1.

Deprecated: idempotency: true|false is the old spelling of name_by: source_position|write_time. It is still read, but an explicit name_by wins over it.

Numbers inside a part name are zero-padded so that ASCII sort equals numeric sort — a bucket is listed lexicographically and that listing order is the replay order. A bucket written by a version before this padding contains unpadded names; the two forms do not sort correctly against each other, so do not mix them in one prefix.

MongoDB capture_all reads the collection, then streams changes. Both phases are numbered into the key so the snapshot always sorts ahead of the changes; a change can never replay before the document it modifies. The snapshot does not resume across runs — it re-scans from the start — but the numbering is deterministic, so a restart reproduces the same names and the covered-range recovery skips them. The exception is inserts or deletes landing in the collection between the two runs: those shift the numbering, and some documents are then written twice.

The sqlx cursor source uses the cursor_column value itself as the position — the same shape Kafka Connect’s S3 sink gives a Kafka offset. That only works for a unique, non-negative integer column: a text cursor pages fine but has no contiguous numeric order, and a repeated value would resolve two rows to one position and drop one of them. The reader rejects both at read time rather than naming records wrongly. Consecutive ids coalesce into one part file; gaps split it.

The file source numbers records by their index in the file, not their byte offset (byte offsets are not consecutive, so every record would become its own part).

That index repeats across runs only in consume mode, which always reads from byte 0 — and repeating is the point: a re-read produces the same names, so the sink recognises them as already covered and writes nothing. subscribe starts at the current end of the file and group_subscribe resumes at a stored byte offset, so their index restarts at 0 over records an earlier run already numbered. Those two modes therefore carry a run epoch in the object name.

The epoch keeps names distinct and keeps runs in order (a later run reads later records, so its objects sort after the earlier run’s). What it does not give you is deduplication across a restart: records re-read after a crash are written again under new names. That is ordinary at-least-once, and it is the honest guarantee for a source with no durable per-record position — these modes are allowed, not rejected, because that guarantee is fine for plenty of pipelines.

A recreated source. Covered ranges are keyed on the source position alone. A source that is dropped and created again under the same name starts at low positions the sink already holds, and those records are skipped as covered. This applies to a Kafka topic or a SQL table that was recreated, and to a file that was replaced in consume mode. Write to a new path or prefix, or remove the old objects, when you recreate a source.

For the file sink, name_by: source_position changes what path means: it is the directory that receives the part files, not the file that is appended to. The sink creates it on startup, so pointing it at an existing regular file fails there with “Failed to create part-file sink directory”.

compression and encryption work as usual.

Durability of the file sink. By default the sink flushes each batch to the operating system and does not call fsync, so a power loss can lose batches that were already acknowledged. fsync: batch syncs the file before each batch is acknowledged; this costs one disk sync per batch. fsync: periodic syncs in the background every fsync_interval_ms (default 1000) and bounds the loss to that interval at almost no cost in throughput.

delete: true needs a single writer. Acknowledged records are removed by rewriting the file and renaming the copy into place. Lines that another process appends during that rewrite are lost. Use it only when the file is complete, or written by an mq-bridge file sink in the same process. A final record without its delimiter is held back until it has been unchanged for 100 ms, so a line that is still being written is not delivered in part. In a CSV file the header line stays in the file.

Current limits, all of which the sink rejects or logs rather than silently mishandling:

• No csv (each part would need its own header row).
• date_partition is ignored — parts are written flat under the prefix, since the name already carries the range. Logged at startup.
• One part file per contiguous run per batch. There is no size-based rolling yet, so a large backfill produces many small files (~1000 per 1M rows at batch_size: 1024). For postgres_cdc this is per-transaction, so a high-commit-rate stream produces one small file per commit. Rolling would require buffering across batches the route has already acked, which is not safe today.

Handlers

A handler is not a stage between the consumer and the publisher — it is a publisher wrapper. The order on the output side is:

handler  →  output middlewares  →  sink

So every sink-side idempotency mechanism on this page runs after the handler has already executed. For a pure ETL route that is fine. For a handler with side effects — enriching, calling an API, emitting an event — it means duplicates reach your code.

To keep duplicates away from a handler, put deduplication on the route’s input. It filters them out of the batch before the route ever calls the publisher chain, and the duplicates are acked straight back at the source. Set key to a business key, as above.

This is still at-least-once, deliberately. The middleware claims a key on receive and only marks it processed once the write succeeded; a claim whose holder crashed lapses after five seconds, so the redelivery is processed rather than lost. A crash after the handler ran but before the marker was written will re-run the handler.

To make a handler effectively-once, its own effect has to be idempotent:

• If the handler writes to a store with a unique key, you already have it — the write is absorbed on replay. The handler may run twice; it cannot land twice.
• If the handler makes a non-idempotent external call (charge a card, send mail), mq-bridge has no primitive for this today. You need an idempotency-key cache around the effect itself.

What is not provided

• Exactly-once across systems. Not achievable without a transaction spanning the source’s offset commit and the sink’s write. Nothing here claims it.
• Kafka transactional EOS. enable.idempotence is on, but transactional.id and send_offsets_to_transaction are not used, so Kafka→Kafka is not exactly-once. This is a choice: transactions only cover a Kafka sink fed by a Kafka source, need read_committed consumers downstream, and cost a round trip per batch. An idempotent sink or consumer covers every combination instead.
• Effects inside handlers. An HTTP call, an email or any write a handler makes on its own is not covered by any of the above; see Handlers.
• Fencing on checkpoint stores. The file/s3/mongodb/postgres checkpoint stores have no lease, owner id or epoch. Kafka consumer groups and Postgres replication slots fence structurally; the checkpoint stores do not, so a zombie instance after a partial partition can re-emit.
• A transactional outbox. When source position and sink write live in the same database, they could be committed together — mq-bridge has no way to express that today.

See also

• REFERENCE.md — every middleware and structural endpoint, with fields and defaults.
• ARCHITECTURE.md — internal design and extension points.