feat(router): deduplicate near-simultaneous Kafka events before dispatch - #1
Open
mwisner wants to merge 1 commit into
Open
Conversation
The live-shows Kafka topics re-publish the same message many times (an audit log for one show showed 91% of 414 records were the identical contentless Show reference, with a run of 70 at the same timestamp). Cosmo has no native event dedup, so every received record runs the full pipeline — BeforeEventsDispatch, subscription filtering, per-subscriber fan-out and _entities resolution — and is dispatched to every subscriber individually. Collapse near-simultaneous duplicates at the earliest point, the Kafka poll loop, before Update(), so the whole downstream pipeline is skipped for dropped copies. A per-poller (per-subscription) time window guards correctness: same- instant bursts collapse while identical payloads re-emitted seconds apart — which for a contentless entity reference mean "re-resolve again" — fall outside the window and are delivered untouched. Disabled by default and tunable via env (toggle in Helm, no rebuild): KAFKA_DEDUP_ENABLED master switch (default false) KAFKA_DEDUP_WINDOW_MS suppression window; 0 = same-timestamp only (default 50) KAFKA_DEDUP_KEY content | value | exact (default content) KAFKA_DEDUP_MAX_KEYS per-poller identity cap, bounds memory (default 4096) router.streams.received.messages still counts every received record; a new router.streams.deduplicated.messages counter records drops so the collapse rate is observable (deduplicated / received). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Router-nonroot image scan passed✅ No security vulnerabilities found in image: |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
Collapse near-simultaneous duplicate Kafka events at the earliest point — the poll loop in
topicPoller, beforeUpdate()— so the whole downstream pipeline (BeforeEventsDispatch, subscription filtering, per-subscriber fan-out,_entitiesresolution) is skipped for dropped copies.Stacked on wundergraph#3133 (
mwisner/fix/kafka-consumer-client-leak). Base will be retargeted tomainonce wundergraph#3133 merges. This PR's diff is only the dedup change.Why
The live-shows topics re-publish the same message many times. An audit log for a single show: 414 records, 37 distinct values; 91% were the identical contentless
Showreference; a run of 70 identical records at the same timestamp. Cosmo has no native event dedup — every received record is processed and dispatched to every subscriber individually.Correctness guardrail
A per-poller (per-subscription) time window separates the two duplicate classes:
Each
Subscribeowns its ownkgo.ClientandtopicPollergoroutine, so the window is single-goroutine (no locking) and can never collapse across subscriptions.Levers (env; default OFF — no behavior change until enabled)
KAFKA_DEDUP_ENABLEDfalseKAFKA_DEDUP_WINDOW_MS500= same-timestamp onlyKAFKA_DEDUP_KEYcontentcontent(partition+key+value) |value|exact(+timestamp, strictest)KAFKA_DEDUP_MAX_KEYS4096Observability
router.streams.received.messagesstill counts every received record; newrouter.streams.deduplicated.messagescounts drops → collapse rate = deduplicated / received.Tests
dedup_window_test.go: same-timestamp burst + 70-run collapse; ~2s re-emits survive; distinct payloads/partitions survive;WINDOW_MS=0same-timestamp-only; each key mode;MAX_KEYSeviction; nil/disabled; mixed-fetch delivery count; env parsing + fallbacks.go vet+ fullpkg/pubsub/...andpkg/metric/...suites pass.Do not merge before wundergraph#3133.