fix: event source tracking#470
Draft
nogringo wants to merge 4 commits into
Draft
Conversation
nogringo
force-pushed
the
test/469-event-source-tracking
branch
from
ede4168 to
6172c98
Compare
📝 WalkthroughWalkthroughAdded relay-side content filtering via a configurable Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant NDK
participant EventVerifier
participant MockRelay
participant Cache
Client->>NDK: broadcast(event)
NDK->>EventVerifier: verify(event)
EventVerifier-->>NDK: verified
NDK->>MockRelay: EVENT message
MockRelay->>MockRelay: check bannedWord in content
alt Content contains banned word
MockRelay-->>NDK: OK(false, "blocked: content contains banned word")
NDK-->>Client: broadcastDone (failed)
else Content allowed
MockRelay->>Cache: store event (relay-side)
MockRelay-->>NDK: OK(true)
NDK->>Cache: load event, merge sources, save updated event
NDK-->>Client: broadcastDone (success)
end
sequenceDiagram
participant Client
participant NDK
participant RelayA
participant RelayB
participant RelayC
participant Cache
Client->>NDK: broadcast(event)
NDK->>RelayA: EVENT
NDK->>RelayB: EVENT
NDK->>RelayC: EVENT
Note over RelayA,RelayC: RelayA or RelayC may block via bannedWord
RelayA-->>NDK: OK(false) or OK(true)
RelayB-->>NDK: OK(true)
RelayC-->>NDK: OK(false) or OK(true)
NDK->>Cache: for each OK(true), load event, merge relay URL into event.sources, save
Client->>NDK: request(event_id)
Cache-->>NDK: event with merged sources
NDK-->>Client: event (sources show only relays that accepted it)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
packages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart (1)
48-52: Consider usingaddTearDownfor robust cleanup.If the assertion on line 46 fails,
ndk.destroy()and the relaystopServer()calls won't execute, potentially leaving orphan servers. UsingaddTearDownensures cleanup runs regardless of test outcome.Proposed refactor using addTearDown
await relay1.startServer(); await relay2.startServer(); await relay3.startServer(); + addTearDown(() async { + await relay1.stopServer(); + await relay2.stopServer(); + await relay3.stopServer(); + }); final ndk = Ndk(NdkConfig( eventVerifier: MockEventVerifier(), cache: MemCacheManager(), bootstrapRelays: [relay1.url, relay2.url, relay3.url], )); + addTearDown(() => ndk.destroy()); // ... rest of test ... expect(events.first.sources.length, equals(2)); - - await ndk.destroy(); - await relay1.stopServer(); - await relay2.stopServer(); - await relay3.stopServer(); });🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart` around lines 48 - 52, Replace the direct cleanup calls at the end of the test with an addTearDown registration so cleanup always runs even if assertions fail: register an async tearDown via addTearDown(() async { await ndk.destroy(); await relay1.stopServer(); await relay2.stopServer(); await relay3.stopServer(); }); — locate the cleanup calls around ndk.destroy() and relay*.stopServer() and move them into that addTearDown callback (ensuring the callback is registered before assertions run).packages/ndk/test/usecases/broadcast_sources_test.dart (1)
41-43: Consider usingaddTearDownfor robust cleanup.Same as the other test file—if assertions fail, cleanup won't execute. Consider wrapping
ndk.destroy()andrelay.stopServer()inaddTearDowncallbacks.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@packages/ndk/test/usecases/broadcast_sources_test.dart` around lines 41 - 43, The test currently calls await ndk.destroy() and await relay.stopServer() directly at the end, which can be skipped if assertions fail; instead register cleanup with addTearDown so they always run — e.g., in the test body call addTearDown(() async => await ndk.destroy()) and addTearDown(() async => await relay.stopServer()), ensuring you reference the existing ndk.destroy() and relay.stopServer() methods so teardown always executes even on failures.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/ndk/test/usecases/broadcast_sources_test.dart`:
- Line 9: The test name string has a typo: change the test declaration
test("braodcast should update source", ...) to test("broadcast should update
source", ...) so the test description reads correctly; update the string in the
test function in broadcast_sources_test.dart (the test call named test(...)
containing "braodcast should update source") to "broadcast should update
source".
- Around line 34-39: The test fails because relay OK responses for broadcasts
aren't updating the cached event's sources; modify the relay response handling
in RelayManager (the code path handling RelayBroadcastResponse in
_processDecodedMessage) to load the event from cache, append the relay URL to
its sources (if not present), and call cacheManager.saveEvent(updatedEvent) so
broadcasts update sources similar to _handleIncomingEvent; ensure you reference
the same event.id and avoid duplicating sources when saving.
In
`@packages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart`:
- Line 14: The test has a copy-paste error where MockRelay instantiation for the
variable relay3 uses the name "relay 2"; update the MockRelay call for relay3 to
use the correct name (e.g., "relay 3") so the MockRelay(name: ...) for relay3 is
distinct from relay2 and improves test clarity and debugging.
---
Nitpick comments:
In `@packages/ndk/test/usecases/broadcast_sources_test.dart`:
- Around line 41-43: The test currently calls await ndk.destroy() and await
relay.stopServer() directly at the end, which can be skipped if assertions fail;
instead register cleanup with addTearDown so they always run — e.g., in the test
body call addTearDown(() async => await ndk.destroy()) and addTearDown(() async
=> await relay.stopServer()), ensuring you reference the existing ndk.destroy()
and relay.stopServer() methods so teardown always executes even on failures.
In
`@packages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart`:
- Around line 48-52: Replace the direct cleanup calls at the end of the test
with an addTearDown registration so cleanup always runs even if assertions fail:
register an async tearDown via addTearDown(() async { await ndk.destroy(); await
relay1.stopServer(); await relay2.stopServer(); await relay3.stopServer(); }); —
locate the cleanup calls around ndk.destroy() and relay*.stopServer() and move
them into that addTearDown callback (ensuring the callback is registered before
assertions run).
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 52333132-7d7d-46f2-a5b9-c09c5b9229e2
📒 Files selected for processing (3)
packages/ndk/test/mocks/mock_relay.dartpackages/ndk/test/usecases/broadcast_sources_test.dartpackages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #470 +/- ##
==========================================
+ Coverage 73.36% 73.37% +0.01%
==========================================
Files 195 195
Lines 8969 9012 +43
==========================================
+ Hits 6580 6613 +33
- Misses 2389 2399 +10 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
1-leo
requested changes
Mar 25, 2026
frnandu
requested changes
Mar 26, 2026
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart`:
- Around line 74-91: The listener on broadcastState.networkController.stream
currently does a read-modify-write using the cached event per-relay, causing
races; instead, when response.broadcastSuccessful is true use the in-memory
broadcastState.event.sources (the monotonic union maintained by relay_manager)
as the source of truth: load the cached event via
_cacheManager.loadEvent(currentEvent.id), merge its sources with
broadcastState.event.sources (ensuring uniqueness), create an updatedEvent via
cachedEvent.copyWith(sources: mergedSources) and call
_cacheManager.saveEvent(updatedEvent); keep checks for currentEvent and
cachedEvent but stop merging only response.relayUrl so the cache reflects
broadcastState.event.sources.
In `@packages/ndk/lib/domain_layer/usecases/requests/requests.dart`:
- Line 278: The code unconditionally passes _cacheWrite.cacheManager (via the
cacheManager parameter) which allows cache mutations even when cacheWrite is
false; change the call site that currently sets cacheManager:
_cacheWrite.cacheManager to pass the cache manager only when caching is enabled
(e.g., cacheManager: _cacheWrite.cacheWrite ? _cacheWrite.cacheManager : null),
so that downstream saveEvent() in stream_response_cleaner (which calls
saveEvent() at lines 86-95) will not mutate sources for requests with
cacheWrite: false.
- Around line 377-390: The code mutates the stored Set by doing
existingSources..addAll(event.sources), so the size comparison always fails;
instead create a new Set clone when merging sources (e.g., newSources =
{...existingSources, ...event.sources}) and compare newSources.length to
existingSources.length, and only update seenEvents[event.id] and call
aggregatedController.add(event.copyWith(...)) when newSources is larger; apply
this change for both occurrences handling seenEvents, existingSources,
aggregatedController, and event.copyWith in the file.
In
`@packages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dart`:
- Around line 14-15: The async cache merge in StreamResponseCleaner currently
fires unordered updates using _cacheManager which can race with the initial
cache write in the Requests use-case; implement a per-event-ID queue (map of
eventId -> Future) inside StreamResponseCleaner to chain cache update operations
so each event's writes run sequentially, on each queued turn read the current
cache entry, merge its sources with the incoming event's sources to produce a
merged snapshot, then write that merged snapshot back via _cacheManager; ensure
the queue entry is replaced with the next Future only after the write completes
or errors so subsequent updates for the same eventId never run concurrently or
lose sources.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 658e055b-17bd-4060-b722-4c03beddfc9b
📒 Files selected for processing (8)
packages/ndk/lib/domain_layer/entities/broadcast_state.dartpackages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dartpackages/ndk/lib/domain_layer/usecases/relay_manager.dartpackages/ndk/lib/domain_layer/usecases/requests/requests.dartpackages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dartpackages/ndk/lib/presentation_layer/init.dartpackages/ndk/test/usecases/broadcast_sources_test.dartpackages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart
💤 Files with no reviewable changes (1)
- packages/ndk/lib/presentation_layer/init.dart
🚧 Files skipped from review as they are similar to previous changes (2)
- packages/ndk/test/usecases/stream_response_cleaner/event_sources_merge_test.dart
- packages/ndk/test/usecases/broadcast_sources_test.dart
Comment on lines
+74
to
+91
| // Listen for broadcast responses to update cache sources | ||
| broadcastState.networkController.stream.listen((response) async { | ||
| if (response.broadcastSuccessful) { | ||
| final currentEvent = broadcastState.event; | ||
| if (currentEvent != null) { | ||
| final cachedEvent = await _cacheManager.loadEvent(currentEvent.id); | ||
| if (cachedEvent != null) { | ||
| final updatedSources = { | ||
| ...cachedEvent.sources, | ||
| response.relayUrl | ||
| }.toList(); | ||
| final updatedEvent = | ||
| cachedEvent.copyWith(sources: updatedSources); | ||
| await _cacheManager.saveEvent(updatedEvent); | ||
| } | ||
| } | ||
| } | ||
| }); |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -type f -name "broadcast.dart" | grep -E "domain_layer|usecases"Repository: relaystr/ndk
Length of output: 121
🏁 Script executed:
find . -type f -name "relay_manager.dart" | grep -E "domain_layer|usecases"Repository: relaystr/ndk
Length of output: 115
🏁 Script executed:
cat -n packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart | sed -n '50,110p'Repository: relaystr/ndk
Length of output: 2890
🏁 Script executed:
cat -n packages/ndk/lib/domain_layer/usecases/relay_manager.dart | sed -n '490,530p'Repository: relaystr/ndk
Length of output: 2053
🏁 Script executed:
cat -n packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart | sed -n '1,20p'Repository: relaystr/ndk
Length of output: 844
🏁 Script executed:
wc -l packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dartRepository: relaystr/ndk
Length of output: 123
🏁 Script executed:
cat -n packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart | sed -n '93,100p'Repository: relaystr/ndk
Length of output: 333
🏁 Script executed:
cat -n packages/ndk/lib/domain_layer/usecases/relay_manager.dart | sed -n '500,530p'Repository: relaystr/ndk
Length of output: 1503
🏁 Script executed:
cat -n packages/ndk/lib/domain_layer/entities/broadcast_state.dart | head -30Repository: relaystr/ndk
Length of output: 1066
🏁 Script executed:
rg -A 3 "class BroadcastState" packages/ndk/lib/domain_layer/entities/broadcast_state.dartRepository: relaystr/ndk
Length of output: 219
Use the in-memory broadcast state to avoid losing relay sources to cache read-modify-write races.
Lines 75–91 listen to broadcast responses and load the cached event once per relay, merge only that relay's URL, and save. If two relays respond before the first cache write completes, both listeners read the same stale cached snapshot, and the second save loses the first relay's source. Since relay_manager.dart (lines 504–517) already maintains broadcastState.event with the monotonic union of all successful relays before emitting the response, merge the cached sources with broadcastState.event.sources instead of patching the cache one relay at a time.
Suggested fix
- broadcastState.networkController.stream.listen((response) async {
- if (response.broadcastSuccessful) {
- final currentEvent = broadcastState.event;
- if (currentEvent != null) {
- final cachedEvent = await _cacheManager.loadEvent(currentEvent.id);
- if (cachedEvent != null) {
- final updatedSources = {
- ...cachedEvent.sources,
- response.relayUrl
- }.toList();
- final updatedEvent =
- cachedEvent.copyWith(sources: updatedSources);
- await _cacheManager.saveEvent(updatedEvent);
- }
- }
- }
- });
+ broadcastState.networkController.stream.listen((response) async {
+ if (!mySaveToCache || !response.broadcastSuccessful) {
+ return;
+ }
+
+ final eventId = broadcastState.event?.id;
+ if (eventId == null) {
+ return;
+ }
+
+ final cachedEvent = await _cacheManager.loadEvent(eventId);
+ final latestEvent = broadcastState.event;
+ if (latestEvent == null) {
+ return;
+ }
+
+ final updatedSources = {
+ ...?cachedEvent?.sources,
+ ...latestEvent.sources,
+ }.toList();
+
+ await _cacheManager.saveEvent(
+ latestEvent.copyWith(sources: updatedSources),
+ );
+ });🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/ndk/lib/domain_layer/usecases/broadcast/broadcast.dart` around lines
74 - 91, The listener on broadcastState.networkController.stream currently does
a read-modify-write using the cached event per-relay, causing races; instead,
when response.broadcastSuccessful is true use the in-memory
broadcastState.event.sources (the monotonic union maintained by relay_manager)
as the source of truth: load the cached event via
_cacheManager.loadEvent(currentEvent.id), merge its sources with
broadcastState.event.sources (ensuring uniqueness), create an updatedEvent via
cachedEvent.copyWith(sources: mergedSources) and call
_cacheManager.saveEvent(updatedEvent); keep checks for currentEvent and
cachedEvent but stop merging only response.relayUrl so the cache reflects
broadcastState.event.sources.
| trackingSet: state.returnedIds, | ||
| outController: state.controller, | ||
| eventOutFilters: _eventOutFilters, | ||
| cacheManager: _cacheWrite.cacheManager, |
There was a problem hiding this comment.
Honor cacheWrite at Line 278.
packages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dart Lines 86-95 now call saveEvent(). Passing _cacheWrite.cacheManager unconditionally here means a request with cacheWrite: false can still mutate cached sources when the same event arrives again.
💡 Suggested fix
- cacheManager: _cacheWrite.cacheManager,
+ cacheManager: request.cacheWrite ? _cacheWrite.cacheManager : null,🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/ndk/lib/domain_layer/usecases/requests/requests.dart` at line 278,
The code unconditionally passes _cacheWrite.cacheManager (via the cacheManager
parameter) which allows cache mutations even when cacheWrite is false; change
the call site that currently sets cacheManager: _cacheWrite.cacheManager to pass
the cache manager only when caching is enabled (e.g., cacheManager:
_cacheWrite.cacheWrite ? _cacheWrite.cacheManager : null), so that downstream
saveEvent() in stream_response_cleaner (which calls saveEvent() at lines 86-95)
will not mutate sources for requests with cacheWrite: false.
Comment on lines
+377
to
+390
| final existingSources = seenEvents[event.id]; | ||
| if (existingSources == null) { | ||
| // First time seeing this event | ||
| seenEvents[event.id] = event.sources.toSet(); | ||
| aggregatedController.add(event); | ||
| } else { | ||
| // Merge sources if this event has new sources | ||
| if (event.sources.isNotEmpty) { | ||
| final newSources = existingSources..addAll(event.sources); | ||
| if (newSources.length > (seenEvents[event.id]?.length ?? 0)) { | ||
| seenEvents[event.id] = newSources; | ||
| aggregatedController.add(event.copyWith(sources: newSources.toList())); | ||
| } | ||
| } |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
fd -t f -i 'requests.dart' | head -5Repository: relaystr/ndk
Length of output: 191
🏁 Script executed:
wc -l packages/ndk/lib/domain_layer/usecases/requests/requests.dartRepository: relaystr/ndk
Length of output: 121
🏁 Script executed:
sed -n '370,400p' packages/ndk/lib/domain_layer/usecases/requests/requests.dart | cat -nRepository: relaystr/ndk
Length of output: 1470
🏁 Script executed:
sed -n '460,485p' packages/ndk/lib/domain_layer/usecases/requests/requests.dart | cat -nRepository: relaystr/ndk
Length of output: 1413
🏁 Script executed:
# Let me verify the exact context around these lines and understand the data flow
sed -n '375,405p' packages/ndk/lib/domain_layer/usecases/requests/requests.dart | cat -nRepository: relaystr/ndk
Length of output: 1385
🏁 Script executed:
cat > /tmp/test_dart.txt << 'EOF'
In Dart:
- The cascade operator (..) mutates the object and returns self
- existingSources is a reference to the Set in seenEvents
- existingSources..addAll(event.sources) mutates that Set and returns it
- So seenEvents[event.id] points to the same mutated Set
- Comparison happens after mutation, so both sides have same length
- This means the condition is always false when sources are added
EOF
cat /tmp/test_dart.txtRepository: relaystr/ndk
Length of output: 450
🏁 Script executed:
# Let me check the actual line numbers in the original review to map them correctly
sed -n '377,390p' packages/ndk/lib/domain_layer/usecases/requests/requests.dart | cat -nRepository: relaystr/ndk
Length of output: 810
🏁 Script executed:
sed -n '468,480p' packages/ndk/lib/domain_layer/usecases/requests/requests.dart | cat -nRepository: relaystr/ndk
Length of output: 843
🏁 Script executed:
cat << 'EOF'
Dart Set behavior with cascade operator:
Given:
final existingSources = seenEvents[event.id]; // reference to Set in map
When executing:
final newSources = existingSources..addAll(event.sources);
The cascade operator (..) on existingSources:
1. Calls addAll() which MUTATES existingSources in-place
2. Returns the mutated existingSources (same object reference)
3. Assigns it to newSources
Result: newSources and existingSources reference the SAME Set object
When checking:
if (newSources.length > (seenEvents[event.id]?.length ?? 0))
Both sides reference the SAME mutated Set, so:
- newSources.length == seenEvents[event.id].length (always)
- Condition is always FALSE
- Event is never emitted with merged sources
The suggested fix {...existingSources, ...event.sources} creates a NEW Set
via spread syntax, avoiding mutation and allowing proper comparison.
EOFRepository: relaystr/ndk
Length of output: 928
Clone the tracked set before comparing sizes.
existingSources..addAll(event.sources) mutates the same Set stored in seenEvents, causing the length comparison on line 386 to always be false since both sides reference the identical mutated object. This prevents merged events from being emitted and blocks relay discovery from expanding when a new relay appears through duplicates.
Replace with final newSources = {...existingSources, ...event.sources}; to create a new Set, and simplify the comparison to if (newSources.length > existingSources.length). This pattern appears identically at both locations (lines 377-390 and 468-480).
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@packages/ndk/lib/domain_layer/usecases/requests/requests.dart` around lines
377 - 390, The code mutates the stored Set by doing
existingSources..addAll(event.sources), so the size comparison always fails;
instead create a new Set clone when merging sources (e.g., newSources =
{...existingSources, ...event.sources}) and compare newSources.length to
existingSources.length, and only update seenEvents[event.id] and call
aggregatedController.add(event.copyWith(...)) when newSources is larger; apply
this change for both occurrences handling seenEvents, existingSources,
aggregatedController, and event.copyWith in the file.
Comment on lines
+14
to
15
| final CacheManager? _cacheManager; | ||
|
|
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
fd -t f "stream_response_cleaner.dart"Repository: relaystr/ndk
Length of output: 147
🏁 Script executed:
fd -t f "requests.dart" packages/ndk/lib/domain_layer/usecases/Repository: relaystr/ndk
Length of output: 117
🏁 Script executed:
cat -n packages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dartRepository: relaystr/ndk
Length of output: 4389
🏁 Script executed:
sed -n '260,285p' packages/ndk/lib/domain_layer/usecases/requests/requests.dartRepository: relaystr/ndk
Length of output: 756
🏁 Script executed:
# Also check the CacheManager interface to understand the contract
fd -t f "cache_manager.dart" packages/ndk/libRepository: relaystr/ndk
Length of output: 195
🏁 Script executed:
cat -n packages/ndk/lib/domain_layer/repositories/cache_manager.dartRepository: relaystr/ndk
Length of output: 8290
Queue cache writes per event ID to ensure sources never shrink.
Line 61 fires an unordered async cache update that can silently drop on cache miss (line 91) or cause earlier writes to complete last and shrink stored sources. In packages/ndk/lib/domain_layer/usecases/requests/requests.dart lines 264-279, the initial cache write and this merge run independently off the same verified stream, enabling out-of-order completion. Add a per-event-ID queue to chain updates sequentially and save the merged event snapshot (sources from both cache and current event).
Suggested fix
final List<EventFilter> _eventOutFilters;
final CacheManager? _cacheManager;
+ final Map<String, Future<void>> _pendingCacheUpdates = {};
@@
- final mergedEvent = event.copyWith(sources: newSources.toList());
- _outController.add(mergedEvent);
- // Update cache with merged sources
- _updateCacheSources(event.id, newSources);
+ final mergedEvent = event.copyWith(sources: newSources.toList());
+ _outController.add(mergedEvent);
+ _updateCacheSources(mergedEvent);
@@
- void _updateCacheSources(String eventId, Set<String> sources) {
- if (_cacheManager == null) return;
-
- // Load existing event from cache and update sources
- _cacheManager.loadEvent(eventId).then((cachedEvent) {
- if (cachedEvent != null) {
- final updatedEvent = cachedEvent.copyWith(sources: sources.toList());
- _cacheManager.saveEvent(updatedEvent);
- }
- });
+ Future<void> _updateCacheSources(Nip01Event event) {
+ if (_cacheManager == null) return Future.value();
+
+ final previous = _pendingCacheUpdates[event.id] ?? Future.value();
+ final queued = previous.then((_) async {
+ final cachedEvent = await _cacheManager!.loadEvent(event.id);
+ final mergedSources = {
+ ...?cachedEvent?.sources,
+ ...event.sources,
+ }.toList();
+
+ await _cacheManager!.saveEvent(
+ (cachedEvent ?? event).copyWith(sources: mergedSources),
+ );
+ });
+
+ _pendingCacheUpdates[event.id] = queued;
+ return queued;
+ }Also applies to: 58-61, 85-95
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In
`@packages/ndk/lib/domain_layer/usecases/stream_response_cleaner/stream_response_cleaner.dart`
around lines 14 - 15, The async cache merge in StreamResponseCleaner currently
fires unordered updates using _cacheManager which can race with the initial
cache write in the Requests use-case; implement a per-event-ID queue (map of
eventId -> Future) inside StreamResponseCleaner to chain cache update operations
so each event's writes run sequentially, on each queued turn read the current
cache entry, merge its sources with the incoming event's sources to produce a
merged snapshot, then write that merged snapshot back via _cacheManager; ensure
the queue entry is replaced with the next Future only after the write completes
or errors so subsequent updates for the same eventId never run concurrently or
lose sources.
Contributor
|
Focus on query broadcast is not as important (for now)
Idea: |
Collaborator
|
because of the race conditions concerns I'm changing this to a draft, low priority so we can re-discuss possible solutions later |
frnandu
requested changes
May 6, 2026
Collaborator
frnandu
left a comment
frnandu
left a comment
There was a problem hiding this comment.
race conditions concerns, we need to re-think solution
Contributor
|
could use |
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
3 participants
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.
Tests to demonstrate #469
Summary by CodeRabbit
New Features
Improvements
Tests