diff --git a/router/pkg/metric/noop_stream_metrics.go b/router/pkg/metric/noop_stream_metrics.go index c312cc247..1c352ffe2 100644 --- a/router/pkg/metric/noop_stream_metrics.go +++ b/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 } diff --git a/router/pkg/metric/oltp_stream_metric_store.go b/router/pkg/metric/oltp_stream_metric_store.go index 8d30c1536..386d21fa7 100644 --- a/router/pkg/metric/oltp_stream_metric_store.go +++ b/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...) +} diff --git a/router/pkg/metric/prom_stream_metric_store.go b/router/pkg/metric/prom_stream_metric_store.go index 30309f244..4b021f706 100644 --- a/router/pkg/metric/prom_stream_metric_store.go +++ b/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) } diff --git a/router/pkg/metric/stream_measurements.go b/router/pkg/metric/stream_measurements.go index a5e1dadfb..1345a8546 100644 --- a/router/pkg/metric/stream_measurements.go +++ b/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 } diff --git a/router/pkg/metric/stream_metric_store.go b/router/pkg/metric/stream_metric_store.go index 361f49388..62df8bb26 100644 --- a/router/pkg/metric/stream_metric_store.go +++ b/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) + } +} diff --git a/router/pkg/pubsub/kafka/adapter.go b/router/pkg/pubsub/kafka/adapter.go index 3ee51437f..cd4252c36 100644 --- a/router/pkg/pubsub/kafka/adapter.go +++ b/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,11 +102,8 @@ 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, @@ -109,6 +111,27 @@ func (p *ProviderAdapter) topicPoller(ctx context.Context, client *kgo.Client, u 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 } diff --git a/router/pkg/pubsub/kafka/dedup_window.go b/router/pkg/pubsub/kafka/dedup_window.go new file mode 100644 index 000000000..6f3bdec68 --- /dev/null +++ b/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) + } +} diff --git a/router/pkg/pubsub/kafka/dedup_window_test.go b/router/pkg/pubsub/kafka/dedup_window_test.go new file mode 100644 index 000000000..b8277f064 --- /dev/null +++ b/router/pkg/pubsub/kafka/dedup_window_test.go @@ -0,0 +1,166 @@ +package kafka + +import ( + "fmt" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/twmb/franz-go/pkg/kgo" +) + +const showRef = `{"__typename":"Show","id":"show_01kz6fvagtef5abnxbcb9036jj"}` + +func rec(key, val string, partition int32, tsMs int64) *kgo.Record { + return &kgo.Record{ + Key: []byte(key), + Value: []byte(val), + Partition: partition, + Offset: 0, + Timestamp: time.UnixMilli(tsMs), + Topic: "live.graphql.show-events", + } +} + +func newWindow(t *testing.T, windowMs int64, mode dedupKeyMode, maxKeys int) *dedupWindow { + t.Helper() + return newDedupWindow(dedupConfig{enabled: true, windowMs: windowMs, keyMode: mode, maxKeys: maxKeys}) +} + +// countDropped runs records through the window and returns how many were dropped as duplicates. +func countDropped(w *dedupWindow, records []*kgo.Record) int { + dropped := 0 + for _, r := range records { + if w.isDuplicate(r) { + dropped++ + } + } + return dropped +} + +func TestDedup_SameTimestampBurstCollapses(t *testing.T) { + // The audit log's worst case: a run of 70 byte-identical Show refs at the same timestamp. + // First is delivered; the other 69 collapse. + w := newWindow(t, 50, dedupKeyContent, 4096) + records := make([]*kgo.Record, 70) + for i := range records { + records[i] = rec("show_x", showRef, 31, 1000) + } + require.Equal(t, 69, countDropped(w, records)) +} + +func TestDedup_ReEmitsSecondsApartSurvive(t *testing.T) { + // Identical contentless Show refs re-emitted ~2s apart are separate "re-resolve" signals and + // must all be delivered under a small window. + w := newWindow(t, 50, dedupKeyContent, 4096) + records := []*kgo.Record{ + rec("show_x", showRef, 31, 1000), + rec("show_x", showRef, 31, 3000), + rec("show_x", showRef, 31, 5000), + rec("show_x", showRef, 31, 7000), + } + require.Equal(t, 0, countDropped(w, records)) +} + +func TestDedup_DistinctPayloadsNeverCollapse(t *testing.T) { + w := newWindow(t, 50, dedupKeyContent, 4096) + records := []*kgo.Record{ + rec("lot_1", `{"__typename":"LotItem","id":"lot_1"}`, 31, 1000), + rec("lot_2", `{"__typename":"LotItem","id":"lot_2"}`, 31, 1000), + rec("lot_3", `{"__typename":"LotItem","id":"lot_3"}`, 31, 1000), + } + require.Equal(t, 0, countDropped(w, records)) +} + +func TestDedup_WindowZeroOnlyCollapsesSameTimestamp(t *testing.T) { + w := newWindow(t, 0, dedupKeyContent, 4096) + require.False(t, w.isDuplicate(rec("show_x", showRef, 31, 1000))) + require.True(t, w.isDuplicate(rec("show_x", showRef, 31, 1000)), "same ms should collapse") + require.False(t, w.isDuplicate(rec("show_x", showRef, 31, 1001)), "1ms later must survive when window=0") +} + +func TestDedup_ExactModeIgnoresWindowAcrossTimestamps(t *testing.T) { + // exact keys include the timestamp, so only byte-identical same-timestamp records collapse, + // regardless of a non-zero window. + w := newWindow(t, 50, dedupKeyExact, 4096) + require.False(t, w.isDuplicate(rec("show_x", showRef, 31, 1000))) + require.True(t, w.isDuplicate(rec("show_x", showRef, 31, 1000)), "same ts collapses") + require.False(t, w.isDuplicate(rec("show_x", showRef, 31, 1010)), "diff ts within window still survives in exact mode") +} + +func TestDedup_ValueModeIgnoresKey(t *testing.T) { + w := newWindow(t, 50, dedupKeyValue, 4096) + require.False(t, w.isDuplicate(rec("key_a", showRef, 31, 1000))) + require.True(t, w.isDuplicate(rec("key_b", showRef, 31, 1000)), "same value, different key collapses in value mode") +} + +func TestDedup_DifferentPartitionsDoNotCollapse(t *testing.T) { + w := newWindow(t, 50, dedupKeyContent, 4096) + require.False(t, w.isDuplicate(rec("show_x", showRef, 31, 1000))) + require.False(t, w.isDuplicate(rec("show_x", showRef, 7, 1000)), "same content on a different partition must survive") +} + +func TestDedup_NilWindowNeverDeduplicates(t *testing.T) { + require.Nil(t, newDedupWindow(dedupConfig{enabled: false})) + var w *dedupWindow + require.False(t, w.isDuplicate(rec("show_x", showRef, 31, 1000))) +} + +func TestDedup_MaxKeysBoundsMemory(t *testing.T) { + // Many distinct keys all at the same timestamp force the clear() fallback; the map must stay + // within the cap. + w := newWindow(t, 50, dedupKeyContent, 8) + for i := 0; i < 1000; i++ { + w.isDuplicate(rec(fmt.Sprintf("k%d", i), showRef, 0, 1000)) + require.LessOrEqual(t, len(w.lastSeen), 8) + } +} + +func TestDedup_MixedFetchDeliveryCount(t *testing.T) { + // A representative fetch: a same-timestamp Show burst interleaved with distinct LotItems and a + // later re-emit. Delivered = unique-at-instant + all re-emits + all lot items. + w := newWindow(t, 50, dedupKeyContent, 4096) + records := []*kgo.Record{ + rec("show_x", showRef, 31, 1000), // delivered (first) + rec("show_x", showRef, 31, 1000), // dropped + rec("show_x", showRef, 31, 1000), // dropped + rec("lot_1", `{"__typename":"LotItem","id":"lot_1"}`, 31, 1000), // delivered + rec("lot_2", `{"__typename":"LotItem","id":"lot_2"}`, 31, 1000), // delivered + rec("show_x", showRef, 31, 3000), // delivered (re-emit, 2s later) + } + require.Equal(t, 2, countDropped(w, records)) +} + +func TestDedupConfigFromEnv_Defaults(t *testing.T) { + // No env set → safe defaults. + t.Setenv("KAFKA_DEDUP_ENABLED", "") + cfg := dedupConfigFromEnv() + require.False(t, cfg.enabled) + require.Equal(t, int64(50), cfg.windowMs) + require.Equal(t, dedupKeyContent, cfg.keyMode) + require.Equal(t, 4096, cfg.maxKeys) +} + +func TestDedupConfigFromEnv_Overrides(t *testing.T) { + t.Setenv("KAFKA_DEDUP_ENABLED", "true") + t.Setenv("KAFKA_DEDUP_WINDOW_MS", "0") + t.Setenv("KAFKA_DEDUP_KEY", "exact") + t.Setenv("KAFKA_DEDUP_MAX_KEYS", "1024") + cfg := dedupConfigFromEnv() + require.True(t, cfg.enabled) + require.Equal(t, int64(0), cfg.windowMs) + require.Equal(t, dedupKeyExact, cfg.keyMode) + require.Equal(t, 1024, cfg.maxKeys) +} + +func TestDedupConfigFromEnv_InvalidValuesFallBack(t *testing.T) { + t.Setenv("KAFKA_DEDUP_ENABLED", "notabool") + t.Setenv("KAFKA_DEDUP_WINDOW_MS", "-5") + t.Setenv("KAFKA_DEDUP_KEY", "bogus") + t.Setenv("KAFKA_DEDUP_MAX_KEYS", "0") + cfg := dedupConfigFromEnv() + require.False(t, cfg.enabled) + require.Equal(t, int64(50), cfg.windowMs) + require.Equal(t, dedupKeyContent, cfg.keyMode) + require.Equal(t, 4096, cfg.maxKeys) +}