windowed, content/header-based message deduplication

[almostintuitive]

Discussion: windowed, content-based message deduplication

Hi all 👋

We'd like to add windowed, content-based message deduplication to Iggy. A partition's primary would drop any message whose identity (its message ID, or the value of a configured user header) was already appended within a time window, before the message goes into the replication pipeline. We built this design before on a Raft-based broker. Before writing any Iggy code, we'd like to know whether you want this, and how it should fit with VSR and the partition plane.

Below: a short proposal, the lessons that carry over, and questions for you.

Proposal

User-facing semantics

Topic option	Meaning
dedup_window	Duration. Unset means disabled.
dedup_identity	message_id (default) or header:<name>.

Plus a server-wide dedup_max_entries_per_partition ceiling.

• First-wins. The first message for an identity within the window is appended. Later ones within the window are dropped silently, and the request still succeeds.
• The window is measured on the prepare timestamp, so it means "arrived within W of the first arrival". Client-supplied origin_timestamp is never used (see the lessons below for why).
• No identity, no dedup. Messages with id == 0 (server-assigned UUID) are always admitted and never indexed.
• A missing header rejects the whole batch. In header mode, a message without the header fails the request. It doesn't fall back to the ID, and it isn't partially admitted. Either fallback would silently defeat the point of choosing a header.
• It's complementary to the client table. The client table answers retries of the same request. The dedup index catches the same content arriving in a different request.

Core design idea: an advisory index derived from the log

The index (identity → timestamp) is a deterministic function of the committed data in the partition log. It is not separately replicated state. Three facts make that safe:

1. Every admitted message is already in the replicated log, with the identity and timestamp the index stores. Replicating the index too would duplicate data every replica already has.
2. Only the primary reads the index, and only before an op is prepared. A replica's copy has no effect until that replica becomes primary, and by then it has applied the committed prefix.
3. Window dedup is approximate by nature: the window edge is fuzzy. If replicas disagree, the only effect is whether a duplicate near the edge gets caught. The log can never be corrupted.

What follows from that:

• Replicas fold identities at commit, like the client table folds watermarks. There are no extra ops, no metadata records and no extra log reads.
• Updates are max-wins and idempotent, so replaying any prefix twice is harmless. Snapshots and state transfer therefore only need to converge: a replica that installs the current index at an older op and replays the rest ends up in the same state. No byte-exact historical reconstruction is needed.
• Recovery is bounded. On boot, seek the timestamp index to now − window and replay only about one window of committed batches. This also addresses a problem the client table slice has today: it comes back empty after a restart.

Our first design replicated the index as its own metadata stream (key lists, mutations, checkpoints) so every replica could rebuild it byte for byte. That brought unbounded undo history, a checkpoint protocol, retention and compaction coordination, and a mutex that allowed only one write per partition per commit round-trip. We deleted all of it, about 2k lines, in favor of the log-derived design. We recommend skipping that step.

Mechanics

• Index entry: in ID mode the key is the u128 ID itself. In header mode it's a 128-bit digest from two fixed-seed xxh64 passes. The seed must be fixed (no RandomState) because the digest is persisted and compared across replicas. Every entry has the same fixed size regardless of identity size, and collisions are negligible at 10⁷ entries. Collision resistance against attackers doesn't matter, since anyone can already suppress an identity by sending it first.
• Eviction: prepare timestamps are monotonic, so entries arrive in time order. A FIFO queue next to the map (lazy deletion when an identity is re-admitted after expiry) makes both expiry and capacity eviction exact, oldest-first and O(1). The ceiling is hard. At the ceiling, evict the oldest; don't fail open. Fail-open silently turns dedup off for new identities under high cardinality.
• Primary path:
  1. Classify synchronously, in message order, before project().
  2. If nothing was dropped, forward the batch unchanged. If some messages were dropped, rebuild the batch from the survivors. If all were dropped, prepare nothing.
  3. An all-duplicate request must not be acked until the op that introduced the identity has committed. Otherwise it can ack a duplicate of a write that is later lost in a view change. Waiting for commit ≥ current op is a simple, conservative way to do this.
  4. If an op is truncated by a view change, the primary's entries for it must go too. Either use a per-request undo list with compare-and-revert (what we did), or keep a committed index plus a pending overlay that is discarded on view change.
• Config changes: bump a generation, and have each replica clear its index, whenever dedup is disabled or the identity source changes. Changing only the window doesn't clear anything.
• Observability (per partition): messages dropped, admitted unindexed, live entries evicted at capacity (a non-zero count means the effective window is narrower than configured), entries truncated on snapshot install, and current index size. "Dedup is working" and "dedup is quietly degraded" look identical from outside without these.

Lessons that carry over (?)

1. Never put client timestamps in the index. One future-dated record pushed our eviction watermark past every entry, and dedup stayed off permanently, including across restarts, because the watermark was in the snapshot. A one-sided window check (new − stored ≤ W) also turned any older timestamp into a "duplicate". Iggy's monotonic prepare timestamp avoids this whole class of bugs. Using it is most of why this proposal is short.
2. Ack on commit, not on visibility. An all-duplicate ack has to wait on the true commit point, not on anything that can advance ahead of durability.
3. Ordering fence: if requests can be classified before earlier admitted ones are appended, a request that saw a duplicate must wait for those appends, and must fail (not succeed) if the introducing write failed. If classification and op assignment happen in one synchronous step, the fence isn't needed.
4. Free the index when dedup is disabled, regardless of what traffic the partition receives afterwards.
5. When truncating an oversized snapshot (peers with different ceilings), keep the newest entries. A prefix of an insertion-ordered map keeps the oldest.
6. An unreadable snapshot should trigger a rebuild from the log, not a failed start. The state is advisory.
7. Bounded replay must start on a batch boundary, and must never move a restored apply position backwards.

Known gaps in the design

• Identity source changes: the log doesn't record which identity source was active at each op. After a restart, a key→header change inside the last window can re-index older messages under the new source. Closing this needs a persisted "generation G starts at op O" boundary.
• Raising the window doesn't retroactively restore entries that were already evicted under the old window, until the next bounded replay.

Questions for maintainers

1. Do you want content-based dedup in the server? Was the old message-ID dedup removed on purpose, and is there a reason it shouldn't come back in this form?
2. Where should classification go? Can the primary filter and rebuild a SendMessages batch between convert_request_message and project()? Is that step synchronous with op assignment, which would remove the ordering fence?
3. Is there a natural commit-time hook for folding identities on every replica, next to where the client-table slice is folded?
4. State transfer: should the index travel with the consumer-offsets/dedup artifact, or have its own section? Is a convergent (not byte-exact) payload acceptable there?
5. Reply shape: when messages are dropped, should SendMessages confirmations report anything beyond base_offset (for example a dropped count or per-ID status)? What should base_offset be when every message was dropped?
6. Defaults and placement: per-topic options plus a server-wide entry ceiling (our rough estimate is 50–60 B per entry per replica, including the FIFO). Does that fit how you split config?

We're happy to start with the self-contained index (map, FIFO eviction and digest, plus tests) as a first PR if the direction looks right.

[justinmclean]

Thanks for the detailed write-up. On the history question: the old message deduplicator was removed when server-ng became the Iggy server (#3856), with the last of it going in #3880. The client-table dedup you mention came in with #3934 and #3959.

@hubcio @numinnex could you take a look at the questions here?

[hubcio]

yes. it'll be implemented after multi-leader, and it'll work exactly like before.

[numinnex]

Hi, we've dropped that functionality as the purpose of that mechanism was dedup of retired messages rather than content based deduplication. I am actually curious what is use case for such deduplication that request_id based one does not cut it?