fix(sink): verify durable storage before dropping replays - #4902
fix(sink): verify durable storage before dropping replays#4902PatrickLiu022 wants to merge 4 commits into
Conversation
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: Cursor <cursoragent@cursor.com>
📝 WalkthroughWalkthroughFlush processing now plans deduplication and durable-storage actions before persistence. It batches uniqueness and durability checks, persists only required messages, stores offsets, and updates deduplication state afterward. ChangesSink deduplication flow
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Flush
participant Deduplicator
participant ClickHouseStorage
participant SinkStorage
participant KafkaOffsets
Flush->>Deduplicator: Check batch uniqueness
Flush->>ClickHouseStorage: Check durable events
Flush->>SinkStorage: Persist planned messages
Flush->>KafkaOffsets: Store offsets
Flush->>Deduplicator: Set planned dedupe items
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| for _, message := range messages { | ||
| item, ok := dedupeItemFromMessage(message) | ||
| if !ok { | ||
| continue |
There was a problem hiding this comment.
With a deduplicator configured, filtering DROP messages out of the flush plan prevents them from reaching the existing LogDroppedEvents branch in persistToStorage, so configured warnings disappear while their Kafka offsets still advance.
Prompt To Fix With AI
This is a comment left during a code review.
Path: openmeter/sink/sink.go
Line: 407-410
Comment:
**Dropped-event warnings are suppressed**
With a deduplicator configured, filtering `DROP` messages out of the flush plan prevents them from reaching the existing `LogDroppedEvents` branch in `persistToStorage`, so configured warnings disappear while their Kafka offsets still advance.
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.
| func dedupeItemsFromMessages(messages []sinkmodels.SinkMessage) []dedupe.Item { | ||
| items := make([]dedupe.Item, 0, len(messages)) | ||
| for _, message := range messages { | ||
| item, ok := dedupeItemFromMessage(message) | ||
| if ok { | ||
| items = append(items, item) | ||
| } | ||
| } | ||
|
|
||
| return items | ||
| } |
There was a problem hiding this comment.
dedupeItemsFromMessages has no callers and duplicates the iteration already performed by planFlushMessages, adding misleading maintenance surface without encapsulating any reused behavior.
| func dedupeItemsFromMessages(messages []sinkmodels.SinkMessage) []dedupe.Item { | |
| items := make([]dedupe.Item, 0, len(messages)) | |
| for _, message := range messages { | |
| item, ok := dedupeItemFromMessage(message) | |
| if ok { | |
| items = append(items, item) | |
| } | |
| } | |
| return items | |
| } |
Context Used: AGENTS.md (source)
Prompt To Fix With AI
This is a comment left during a code review.
Path: openmeter/sink/sink.go
Line: 481-491
Comment:
**Remove the unused dedupe helper**
`dedupeItemsFromMessages` has no callers and duplicates the iteration already performed by `planFlushMessages`, adding misleading maintenance surface without encapsulating any reused behavior.
```suggestion
```
**Context Used:** AGENTS.md ([source](https://github.com/openmeterio/openmeter/blob/main/AGENTS.md))
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (8)
openmeter/sink/storage.go (1)
106-123: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy liftQuery fan-out scales with distinct (namespace, source) pairs.
HasEventsissues at least one ClickHouse query per(namespace, source)group per flush, and this now runs on every flush cycle. A mixed-tenant buffer with many sources produces many sequential round trips before any insert happens, which adds latency directly to the flush critical path.If the flush batch commonly spans many groups, consider a single query with a namespace filter plus an ID
Inlist, or bounded parallel queries across groups. This is a follow-up idea, not a blocker.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openmeter/sink/storage.go` around lines 106 - 123, Reduce query fan-out in HasEvents by avoiding one sequential query per groupedItems (namespace, source) pair; consolidate existence checks into a single namespace/ID-filtered query or bounded parallel queries while preserving durableItems results and error propagation.Source: Path instructions
openmeter/sink/sink_dedupe_test.go (3)
156-234: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd short intent comments to the replay-transition tests.
The four state transitions encode non-obvious crash semantics. For example, unique + durable means the process crashed after the ClickHouse write and before the Redis key was set. The test names carry the outcome but not the reason.
A one-line
given/when/thenper test makes the replay intent readable without cross-referencingplanFlushMessages. Example for lines 186-203:// given a replay whose event is already durable but whose dedupe key was lost in a crash // when the flush is planned // then the sink restores the dedupe key and skips a duplicate insertBased on coding guidelines: "Begin non-trivial service or lifecycle subtests with concise
given,when, andthenintent comments."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openmeter/sink/sink_dedupe_test.go` around lines 156 - 234, Add concise three-line given/when/then intent comments to the four replay-transition tests: TestPlanFlushMessages_IgnoresDropMissingIdentity, TestPlanFlushMessages_UniqueDurableSetsDedupeWithoutInsert, TestPlanFlushMessages_ExistingNotDurableReinsertsWithoutSet, and TestPlanFlushMessages_RejectsUnclassifiedDedupeResult. Describe the relevant crash/replay state, the planFlushMessages action, and the expected outcome without changing test behavior.Source: Coding guidelines
284-303: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReuse
newTestSinkhere.This builds the same
Sinkinline.dedupeSetnever touchesStorage, sonewTestSink(deduplicator, nil)covers it.♻️ Suggested tweak
deduplicator := newCountingDeduplicator(nil) - logger := slog.New(slog.NewTextHandler(io.Discard, nil)) - s := &Sink{ - config: SinkConfig{ - Logger: logger, - Tracer: noop.NewTracerProvider().Tracer("test"), - Deduplicator: deduplicator, - }, - } + s := newTestSink(deduplicator, nil)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openmeter/sink/sink_dedupe_test.go` around lines 284 - 303, Update TestDedupeSet_UsesSingleBatchCall to construct the sink with newTestSink(deduplicator, nil) instead of duplicating the inline Sink and logger configuration. Preserve the existing dedupeSet call and assertions.
106-141: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd an error hook to
countingStoragefor the durability-check failure path.
planFlushMessageswraps aHasEventsfailure into "failed to verify durable state for flush messages" and aborts the flush. That path is new in this PR and no test reaches it.countingStoragehas no way to return an error, so the fail-closed guarantee is unverified.💚 Suggested addition
type countingStorage struct { durable dedupe.ItemSet + err error hasEventsCalls int hasEventsSizes []int hasEventsItems [][]dedupe.Item } @@ s.hasEventsItems = append(s.hasEventsItems, slices.Clone(items)) + if s.err != nil { + return nil, s.err + } + result := dedupe.ItemSet{}func TestPlanFlushMessages_FailsWhenDurabilityCheckFails(t *testing.T) { // given a storage that cannot answer the durability question storage := newCountingStorage(nil) storage.err = errors.New("clickhouse unavailable") s := newTestSink(newCountingDeduplicator(nil), storage) // when planning a flush _, err := s.planFlushMessages(t.Context(), []sinkmodels.SinkMessage{ testSinkMessage("tenant-a", "evt-1", "gateway"), }) // then planning fails closed instead of inserting blindly require.ErrorContains(t, err, "failed to verify durable state") }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openmeter/sink/sink_dedupe_test.go` around lines 106 - 141, Extend countingStorage with an err field and make HasEvents return that error when configured, while preserving its existing call tracking. Add TestPlanFlushMessages_FailsWhenDurabilityCheckFails to configure the error, invoke planFlushMessages with a test message, and assert the returned error contains “failed to verify durable state”.Source: Path instructions
openmeter/sink/storage_test.go (2)
68-102: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNice grouping test. Two behaviors from this PR are still uncovered.
The grouping and filter assertions look solid. The new
HasEventscontract also introduces two behaviors that no test currently reaches:
- Validation: an item with an empty
Namespace,Source, orIDmust return the "namespace, source and id are required" error.- Error propagation: when
ListEventsV2fails,HasEventsmust return the wrapped "failed to check event durability" error. Theerrfield onhasEventStreamingStubalready exists for exactly this.A chunking case with more than
existsQueryBatchSizeIDs would also confirm that the 200-ID batching produces multiple calls, since the current test only asserts the upper bound.💚 Sketch of the missing cases
func TestClickHouseStorageHasEvents_RejectsIncompleteItems(t *testing.T) { stub := newHasEventStreamingStub(t) storage := &ClickHouseStorage{config: ClickHouseStorageConfig{Streaming: stub}} _, err := storage.HasEvents(t.Context(), []dedupe.Item{{Namespace: "tenant-a", ID: "evt-1"}}) require.ErrorContains(t, err, "namespace, source and id are required") require.Empty(t, stub.calls) } func TestClickHouseStorageHasEvents_PropagatesQueryError(t *testing.T) { stub := newHasEventStreamingStub(t) stub.err = errors.New("clickhouse down") storage := &ClickHouseStorage{config: ClickHouseStorageConfig{Streaming: stub}} _, err := storage.HasEvents(t.Context(), []dedupe.Item{ {Namespace: "tenant-a", Source: "gateway", ID: "evt-1"}, }) require.ErrorContains(t, err, "failed to check event durability") }Error propagation matters most here:
planFlushMessagesturns that error into a flush failure, so it is a real replay-safety path.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openmeter/sink/storage_test.go` around lines 68 - 102, Extend the HasEvents tests with coverage for incomplete dedupe.Item values, asserting the required-fields error and no streaming calls; add a ListEventsV2 failure case using hasEventStreamingStub.err and assert the wrapped “failed to check event durability” error. Also add a batch-size case with more than existsQueryBatchSize IDs and verify multiple calls are made.Source: Path instructions
31-66: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe stub ignores
LimitandCursor, and mutates the caller's slice.Two small things here:
ListEventsV2never honorsparams.Limitorparams.Cursor, so the pagination branch inqueryExistingItemsByNamespaceAndSourceis never exercised. That is also why the loop-safety concern flagged inopenmeter/sink/storage.gois invisible to this suite.- Line 45:
lo.FromPtr(params.ID.In)returns the caller's slice, andsort.Stringssorts it in place. TodayHasEventsalready sorts, so nothing breaks. A stub that mutates production input can still hide a future regression.♻️ Suggested tweak
events := []streaming.RawEvent{} - ids := lo.FromPtr(params.ID.In) - sort.Strings(ids) + ids := slices.Clone(lo.FromPtr(params.ID.In)) + slices.Sort(ids)The
errfield on the struct is also never set by any test right now.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openmeter/sink/storage_test.go` around lines 31 - 66, Update hasEventStreamingStub.ListEventsV2 to copy params.ID.In before sorting, then implement params.Cursor and params.Limit pagination semantics so the stub exercises queryExistingItemsByNamespaceAndSource pagination and loop-safety paths. Preserve existing filtering and error behavior, and ensure the stub’s err field can be configured by tests rather than remaining unused.openmeter/sink/sink.go (2)
438-442: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueReuse
dedupeItemsinstead of recomputing the item.
validMessagesanddedupeItemsare built together in the same loop, so they stay index-aligned. Callingmessage.GetDedupeItem()again creates a second source for the same value. The map lookups againstdedupeResultsanddurableItemsonly work while both paths agree.♻️ Suggested tweak
- for _, message := range validMessages { - item := message.GetDedupeItem() + for i, message := range validMessages { + item := dedupeItems[i] _, isUnique := dedupeResults.UniqueItems[item]🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openmeter/sink/sink.go` around lines 438 - 442, Update the loop over validMessages to reuse the corresponding precomputed entry from dedupeItems instead of calling message.GetDedupeItem() again. Preserve the existing UniqueItems, AlreadyProcessedItems, and durableItems lookups using that aligned dedupe item.
481-491: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the unused
dedupeItemsFromMessageshelper. The repository contains only its declaration;planFlushMessagesperforms this conversion inline.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@openmeter/sink/sink.go` around lines 481 - 491, Remove the unused dedupeItemsFromMessages helper and its declaration, leaving the inline conversion in planFlushMessages unchanged.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@openmeter/sink/sink_dedupe_test.go`:
- Around line 156-234: Add concise three-line given/when/then intent comments to
the four replay-transition tests:
TestPlanFlushMessages_IgnoresDropMissingIdentity,
TestPlanFlushMessages_UniqueDurableSetsDedupeWithoutInsert,
TestPlanFlushMessages_ExistingNotDurableReinsertsWithoutSet, and
TestPlanFlushMessages_RejectsUnclassifiedDedupeResult. Describe the relevant
crash/replay state, the planFlushMessages action, and the expected outcome
without changing test behavior.
- Around line 284-303: Update TestDedupeSet_UsesSingleBatchCall to construct the
sink with newTestSink(deduplicator, nil) instead of duplicating the inline Sink
and logger configuration. Preserve the existing dedupeSet call and assertions.
- Around line 106-141: Extend countingStorage with an err field and make
HasEvents return that error when configured, while preserving its existing call
tracking. Add TestPlanFlushMessages_FailsWhenDurabilityCheckFails to configure
the error, invoke planFlushMessages with a test message, and assert the returned
error contains “failed to verify durable state”.
In `@openmeter/sink/sink.go`:
- Around line 438-442: Update the loop over validMessages to reuse the
corresponding precomputed entry from dedupeItems instead of calling
message.GetDedupeItem() again. Preserve the existing UniqueItems,
AlreadyProcessedItems, and durableItems lookups using that aligned dedupe item.
- Around line 481-491: Remove the unused dedupeItemsFromMessages helper and its
declaration, leaving the inline conversion in planFlushMessages unchanged.
In `@openmeter/sink/storage_test.go`:
- Around line 68-102: Extend the HasEvents tests with coverage for incomplete
dedupe.Item values, asserting the required-fields error and no streaming calls;
add a ListEventsV2 failure case using hasEventStreamingStub.err and assert the
wrapped “failed to check event durability” error. Also add a batch-size case
with more than existsQueryBatchSize IDs and verify multiple calls are made.
- Around line 31-66: Update hasEventStreamingStub.ListEventsV2 to copy
params.ID.In before sorting, then implement params.Cursor and params.Limit
pagination semantics so the stub exercises
queryExistingItemsByNamespaceAndSource pagination and loop-safety paths.
Preserve existing filtering and error behavior, and ensure the stub’s err field
can be configured by tests rather than remaining unused.
In `@openmeter/sink/storage.go`:
- Around line 106-123: Reduce query fan-out in HasEvents by avoiding one
sequential query per groupedItems (namespace, source) pair; consolidate
existence checks into a single namespace/ID-filtered query or bounded parallel
queries while preserving durableItems results and error propagation.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 10bef959-7a9e-4844-a2a0-c5a11f6db3c3
📒 Files selected for processing (4)
openmeter/sink/sink.goopenmeter/sink/sink_dedupe_test.goopenmeter/sink/storage.goopenmeter/sink/storage_test.go
2 participants
Overview
Sink-side dedupe can classify a Kafka replay as already processed even when the corresponding event is absent from durable storage. This creates a post-accept loss window: the API succeeds and Kafka retains the event, but the replacement sink worker drops replay solely because Redis contains the stable ID.
We observed this during an EKS node rollout: 77 seconds of audited usage were absent from ClickHouse despite successful API responses and healthy producer metrics.
This change moves replay classification to flush planning and combines two batched checks:
CheckUniqueBatchdetermines whether each ID is new or already known.(namespace, source, id)is durable.The resulting state machine is:
Invalid/dropped events stay outside durability checks, and incomplete dedupe classifications fail closed instead of silently discarding an event.
Notes for reviewer
The implementation preserves batch behavior: one Redis classification call, one Redis Set pipeline, and ClickHouse existence queries grouped by namespace/source and chunked to 200 IDs. It removes the unsafe parse-time duplicate drop.
The tests deterministically cover both crash boundaries, dropped/missing identities, state restoration, incomplete classifications, and batch call counts.
Verification:
Companion repair tooling: https://github.com/voltagepark/takao/pull/2850
Summary by CodeRabbit
Bug Fixes
Performance
Tests
Greptile Summary
This PR moves replay classification from parse time into flush planning, combining Redis dedupe classification with ClickHouse durability checks to close the post-accept event-loss window.
Confidence Score: 4/5
The PR appears safe to merge, with non-blocking cleanup needed for suppressed dropped-event warnings and an unused helper.
The durability-aware replay paths are coherently implemented and tested, while the accepted issues affect operational diagnostics and maintainability rather than durable event correctness.
Files Needing Attention: openmeter/sink/sink.go
Important Files Changed
Flowchart
%%{init: {'theme': 'neutral'}}%% flowchart TD A[Buffered valid event] --> B[Redis CheckUniqueBatch] B --> C[ClickHouse HasEvents] C --> D{Dedupe and durability state} D -->|Unique, not durable| E[Insert event] D -->|Unique, durable| F[Skip insert] D -->|Existing, not durable| G[Reinsert event] D -->|Existing, durable| H[Skip insert] E --> I[Store Kafka offset] F --> I G --> I H --> I I --> J{Unique in Redis?} J -->|Yes| K[Set or restore dedupe key] J -->|No| L[Keep existing dedupe key]Prompt To Fix All With AI
Reviews (1): Last reviewed commit: "fix(sink): reject incomplete dedupe clas..." | Re-trigger Greptile
Context used: