feat(router): deduplicate near-simultaneous Kafka events before dispatch

router/pkg/metric/noop_stream_metrics.go

@@ -6,8 +6,9 @@ import (
 
 type NoopStreamMetricStore struct{}
 
-func (n *NoopStreamMetricStore) Produce(ctx context.Context, event StreamsEvent) {}
-func (n *NoopStreamMetricStore) Consume(ctx context.Context, event StreamsEvent) {}
+func (n *NoopStreamMetricStore) Produce(ctx context.Context, event StreamsEvent)      {}
+func (n *NoopStreamMetricStore) Consume(ctx context.Context, event StreamsEvent)      {}
+func (n *NoopStreamMetricStore) Deduplicated(ctx context.Context, event StreamsEvent) {}
 
 func (n *NoopStreamMetricStore) Flush(ctx context.Context) error    { return nil }
 func (n *NoopStreamMetricStore) Shutdown(ctx context.Context) error { return nil }

router/pkg/metric/oltp_stream_metric_store.go

@@ -46,3 +46,7 @@ func (o *otlpStreamEventMetrics) Produce(ctx context.Context, opts ...otelmetric
 func (o *otlpStreamEventMetrics) Consume(ctx context.Context, opts ...otelmetric.AddOption) {
 	o.instruments.consumedMessages.Add(ctx, 1, opts...)
 }
+
+func (o *otlpStreamEventMetrics) Deduplicated(ctx context.Context, opts ...otelmetric.AddOption) {
+	o.instruments.deduplicatedMessages.Add(ctx, 1, opts...)
+}

router/pkg/metric/prom_stream_metric_store.go

@@ -47,6 +47,10 @@ func (p *promStreamEventMetrics) Consume(ctx context.Context, opts ...otelmetric
 	p.instruments.consumedMessages.Add(ctx, 1, opts...)
 }
 
+func (p *promStreamEventMetrics) Deduplicated(ctx context.Context, opts ...otelmetric.AddOption) {
+	p.instruments.deduplicatedMessages.Add(ctx, 1, opts...)
+}
+
 func (p *promStreamEventMetrics) Flush(ctx context.Context) error {
 	return p.meterProvider.ForceFlush(ctx)
 }

router/pkg/metric/stream_measurements.go

@@ -7,8 +7,9 @@ import (
 )
 
 const (
-	messagingSentMessages     = "router.streams.sent.messages"
-	messagingConsumedMessages = "router.streams.received.messages"
+	messagingSentMessages         = "router.streams.sent.messages"
+	messagingConsumedMessages     = "router.streams.received.messages"
+	messagingDeduplicatedMessages = "router.streams.deduplicated.messages"
 )
 
 var (
@@ -18,11 +19,15 @@ var (
 	messagingConsumedMessagesOptions = []otelmetric.Int64CounterOption{
 		otelmetric.WithDescription("Number of stream consumed messages"),
 	}
+	messagingDeduplicatedMessagesOptions = []otelmetric.Int64CounterOption{
+		otelmetric.WithDescription("Number of stream messages dropped as duplicates before dispatch"),
+	}
 )
 
 type eventInstruments struct {
-	producedMessages otelmetric.Int64Counter
-	consumedMessages otelmetric.Int64Counter
+	producedMessages     otelmetric.Int64Counter
+	consumedMessages     otelmetric.Int64Counter
+	deduplicatedMessages otelmetric.Int64Counter
 }
 
 func newStreamEventInstruments(meter otelmetric.Meter) (*eventInstruments, error) {
@@ -42,8 +47,17 @@ func newStreamEventInstruments(meter otelmetric.Meter) (*eventInstruments, error
 		return nil, fmt.Errorf("failed to create received messages counter: %w", err)
 	}
 
+	deduplicatedCounter, err := meter.Int64Counter(
+		messagingDeduplicatedMessages,
+		messagingDeduplicatedMessagesOptions...,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create deduplicated messages counter: %w", err)
+	}
+
 	return &eventInstruments{
-		producedMessages: producedCounter,
-		consumedMessages: consumedCounter,
+		producedMessages:     producedCounter,
+		consumedMessages:     consumedCounter,
+		deduplicatedMessages: deduplicatedCounter,
 	}, nil
 }

router/pkg/metric/stream_metric_store.go

@@ -33,11 +33,15 @@ type StreamsEvent struct {
 type StreamMetricProvider interface {
 	Produce(ctx context.Context, opts ...otelmetric.AddOption)
 	Consume(ctx context.Context, opts ...otelmetric.AddOption)
+	// Deduplicated records an event that was dropped as a duplicate before dispatch.
+	Deduplicated(ctx context.Context, opts ...otelmetric.AddOption)
 }
 
 type StreamMetricStore interface {
 	Produce(ctx context.Context, event StreamsEvent)
 	Consume(ctx context.Context, event StreamsEvent)
+	// Deduplicated records an event that was dropped as a duplicate before dispatch.
+	Deduplicated(ctx context.Context, event StreamsEvent)
 }
 
 // StreamMetrics is the store for Event (Kafka/Redis/NATS) metrics.
@@ -121,3 +125,25 @@ func (e *StreamMetrics) Consume(ctx context.Context, event StreamsEvent) {
 		provider.Consume(ctx, opt)
 	}
 }
+
+func (e *StreamMetrics) Deduplicated(ctx context.Context, event StreamsEvent) {
+	attrs := []attribute.KeyValue{
+		otel.WgStreamOperationName.String(event.StreamOperationName),
+		otel.WgProviderType.String(string(event.ProviderType)),
+	}
+	if event.ErrorType != "" {
+		attrs = append(attrs, otel.WgErrorType.String(event.ErrorType))
+	}
+	if event.ProviderId != "" {
+		attrs = append(attrs, otel.WgProviderId.String(event.ProviderId))
+	}
+	if event.DestinationName != "" {
+		attrs = append(attrs, otel.WgDestinationName.String(event.DestinationName))
+	}
+
+	opt := e.withAttrs(attrs...)
+
+	for _, provider := range e.providers {
+		provider.Deduplicated(ctx, opt)
+	}
+}

router/pkg/pubsub/kafka/adapter.go

@@ -45,6 +45,9 @@ type ProviderAdapter struct {
 	// connectivity (kgo connects lazily otherwise) so an unreachable broker surfaces a
 	// distinct "could not connect" error, consistent with the NATS and Redis adapters.
 	skipUnavailable bool
+	// dedup holds the tunable event-deduplication levers (KAFKA_DEDUP_*). Disabled by default; a
+	// per-subscription window is built from it in topicPoller.
+	dedup dedupConfig
 }
 
 type PollerOpts struct {
@@ -53,6 +56,8 @@ type PollerOpts struct {
 
 // topicPoller polls the Kafka topic for new records and calls the updateTriggers function.
 func (p *ProviderAdapter) topicPoller(ctx context.Context, client *kgo.Client, updater datasource.SubscriptionEventUpdater, pollerOpts PollerOpts) error {
+	// One window per poller (per subscription); nil when dedup is disabled.
+	dedup := newDedupWindow(p.dedup)
 	for {
 		select {
 		case <-ctx.Done(): // Close the poller if the context was canceled (subscription ended, or router shutdown/hot reload)
@@ -97,18 +102,36 @@ func (p *ProviderAdapter) topicPoller(ctx context.Context, client *kgo.Client, u
 
 				p.logger.Debug("subscription update", zap.String("topic", r.Topic), zap.ByteString("data", r.Value))
 
-				headers := make(map[string][]byte)
-				for _, header := range r.Headers {
-					headers[header.Key] = header.Value
-				}
-
+				// Count every record received from the broker before any dedup, so
+				// router.streams.received.messages reflects true inbound volume.
 				p.streamMetricStore.Consume(ctx, metric.StreamsEvent{
 					ProviderId:          pollerOpts.providerId,
 					StreamOperationName: kafkaReceive,
 					ProviderType:        metric.ProviderTypeKafka,
 					DestinationName:     r.Topic,
 				})
 
+				// Drop near-simultaneous duplicates before the expensive downstream pipeline.
+				if dedup.isDuplicate(r) {
+					p.streamMetricStore.Deduplicated(ctx, metric.StreamsEvent{
+						ProviderId:          pollerOpts.providerId,
+						StreamOperationName: kafkaReceive,
+						ProviderType:        metric.ProviderTypeKafka,
+						DestinationName:     r.Topic,
+					})
+					p.logger.Debug("dropped duplicate event",
+						zap.String("topic", r.Topic),
+						zap.Int32("partition", r.Partition),
+						zap.Int64("offset", r.Offset),
+					)
+					continue
+				}
+
+				headers := make(map[string][]byte)
+				for _, header := range r.Headers {
+					headers[header.Key] = header.Value
+				}
+
 				updater.Update([]datasource.StreamEvent{
 					&Event{
 						evt: &MutableEvent{
@@ -363,6 +386,7 @@ func NewProviderAdapter(ctx context.Context, logger *zap.Logger, opts []kgo.Opt,
 		cancel:            cancel,
 		streamMetricStore: store,
 		skipUnavailable:   providerOpts.SkipUnavailableProviders,
+		dedup:             dedupConfigFromEnv(),
 	}, nil
 }
 

router/pkg/pubsub/kafka/dedup_window.go

@@ -0,0 +1,164 @@
+package kafka
+
+import (
+	"encoding/binary"
+	"hash/fnv"
+	"os"
+	"strconv"
+
+	"github.com/twmb/franz-go/pkg/kgo"
+)
+
+// Duplicate-event deduplication for the Kafka subscription path.
+//
+// The live-shows topics re-publish the same message many times, and Cosmo processes and dispatches
+// every received record individually (no native event dedup). This collapses near-simultaneous
+// identical records at the earliest point — the poll loop, before Update() — so the whole
+// downstream pipeline (BeforeEventsDispatch, subscription filtering, per-subscriber fan-out and
+// _entities resolution) is skipped for the dropped copies.
+//
+// The guardrail is a very small time window: 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.
+
+// dedupKeyMode selects which parts of a record form its identity for deduplication.
+type dedupKeyMode int
+
+const (
+	// dedupKeyContent keys on (partition, key, value) — the default. Collapses byte-identical
+	// payloads for the same Kafka key that arrive within the window.
+	dedupKeyContent dedupKeyMode = iota
+	// dedupKeyValue keys on the value only.
+	dedupKeyValue
+	// dedupKeyExact keys on (partition, key, value, timestamp). Only collapses records that are
+	// byte-identical AND share the same producer timestamp, regardless of window — the strictest,
+	// safest mode.
+	dedupKeyExact
+)
+
+func parseDedupKeyMode(s string) dedupKeyMode {
+	switch s {
+	case "value":
+		return dedupKeyValue
+	case "exact":
+		return dedupKeyExact
+	default:
+		return dedupKeyContent
+	}
+}
+
+// dedupConfig holds the tunable levers for Kafka event deduplication. All are sourced from the
+// environment (see dedupConfigFromEnv) so they can be toggled per-deployment without a rebuild.
+type dedupConfig struct {
+	enabled  bool
+	windowMs int64 // suppression window in milliseconds; 0 = collapse only same-timestamp records
+	keyMode  dedupKeyMode
+	maxKeys  int // per-poller cap on tracked identities (bounds memory)
+}
+
+// dedupConfigFromEnv reads the KAFKA_DEDUP_* levers. Defaults: disabled, 50ms window, content key,
+// 4096 keys. Invalid values fall back to the default for that lever.
+func dedupConfigFromEnv() dedupConfig {
+	cfg := dedupConfig{
+		enabled:  false,
+		windowMs: 50,
+		keyMode:  dedupKeyContent,
+		maxKeys:  4096,
+	}
+	if v, ok := os.LookupEnv("KAFKA_DEDUP_ENABLED"); ok {
+		if b, err := strconv.ParseBool(v); err == nil {
+			cfg.enabled = b
+		}
+	}
+	if v, ok := os.LookupEnv("KAFKA_DEDUP_WINDOW_MS"); ok {
+		if n, err := strconv.ParseInt(v, 10, 64); err == nil && n >= 0 {
+			cfg.windowMs = n
+		}
+	}
+	if v, ok := os.LookupEnv("KAFKA_DEDUP_KEY"); ok {
+		cfg.keyMode = parseDedupKeyMode(v)
+	}
+	if v, ok := os.LookupEnv("KAFKA_DEDUP_MAX_KEYS"); ok {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			cfg.maxKeys = n
+		}
+	}
+	return cfg
+}
+
+// dedupWindow tracks recently-delivered record identities to drop duplicates. It is NOT safe for
+// concurrent use: each topicPoller (one goroutine per subscription) owns its own window, so no
+// locking is needed and one subscription can never affect another's dedup state.
+type dedupWindow struct {
+	cfg      dedupConfig
+	lastSeen map[uint64]int64 // identity hash -> last delivered record timestamp (unix millis)
+}
+
+// newDedupWindow returns a window for the given config, or nil when dedup is disabled. A nil
+// window is safe to use — isDuplicate always returns false.
+func newDedupWindow(cfg dedupConfig) *dedupWindow {
+	if !cfg.enabled {
+		return nil
+	}
+	return &dedupWindow{
+		cfg:      cfg,
+		lastSeen: make(map[uint64]int64),
+	}
+}
+
+// isDuplicate reports whether r duplicates a record delivered within the window and should be
+// dropped. When it returns false it records r as delivered. A nil window never deduplicates.
+func (w *dedupWindow) isDuplicate(r *kgo.Record) bool {
+	if w == nil {
+		return false
+	}
+	h := w.identity(r)
+	tsMs := r.Timestamp.UnixMilli()
+	if last, ok := w.lastSeen[h]; ok && tsMs >= last && tsMs-last <= w.cfg.windowMs {
+		return true
+	}
+	if len(w.lastSeen) >= w.cfg.maxKeys {
+		w.evict(tsMs)
+	}
+	w.lastSeen[h] = tsMs
+	return false
+}
+
+// identity hashes the parts of r that define a duplicate under the configured key mode. Fields are
+// length-prefixed so "ab"+"c" and "a"+"bc" never collide.
+func (w *dedupWindow) identity(r *kgo.Record) uint64 {
+	h := fnv.New64a()
+	var buf [8]byte
+	writeField := func(b []byte) {
+		binary.LittleEndian.PutUint64(buf[:], uint64(len(b)))
+		_, _ = h.Write(buf[:])
+		_, _ = h.Write(b)
+	}
+
+	if w.cfg.keyMode != dedupKeyValue {
+		// Partition guards against identical payloads on different partitions collapsing together.
+		binary.LittleEndian.PutUint32(buf[:4], uint32(r.Partition))
+		_, _ = h.Write(buf[:4])
+		writeField(r.Key)
+	}
+	writeField(r.Value)
+	if w.cfg.keyMode == dedupKeyExact {
+		binary.LittleEndian.PutUint64(buf[:], uint64(r.Timestamp.UnixNano()))
+		_, _ = h.Write(buf[:])
+	}
+	return h.Sum64()
+}
+
+// evict bounds memory when the window hits its cap: it drops entries older than the window first,
+// and clears the window entirely if that is not enough. Clearing can at worst miss a future dedup;
+// it can never cause a wrong drop.
+func (w *dedupWindow) evict(nowMs int64) {
+	for h, ts := range w.lastSeen {
+		if nowMs-ts > w.cfg.windowMs {
+			delete(w.lastSeen, h)
+		}
+	}
+	if len(w.lastSeen) >= w.cfg.maxKeys {
+		clear(w.lastSeen)
+	}
+}