Skip to content

Commit 58f0768

Browse files
browbrow
committed
Retry notification cache writes within a bounded export budget
Signed-off-by: Tom Brow <[email protected]>
1 parent dab9fed commit 58f0768

3 files changed

Lines changed: 335 additions & 14 deletions

File tree

‎mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushPresentationCacheTests.swift‎

Lines changed: 124 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -574,6 +574,130 @@ struct BuzzPushPresentationCacheTests {
574574
)
575575
}
576576

577+
@Test("Retrying a partial batch prefix preserves newer state and original scope")
578+
func partialBatchPrefixReplayIsIdempotent() throws {
579+
let directory = try temporaryDirectory()
580+
defer { try? FileManager.default.removeItem(at: directory) }
581+
let store = BuzzPushPresentationCacheStore(
582+
containerURL: directory,
583+
now: { Date(timeIntervalSince1970: 1_700_000_100) }
584+
)
585+
let relayPubkey = try pubkey(for: relayKey)
586+
let member = try pubkey(for: profileKey)
587+
let addedMember = try pubkey(for: otherRelayKey)
588+
let profiles = try [profileKey, otherRelayKey].map { key in
589+
try signedEvent(privateKey: key, createdAt: 100, kind: 0, content: #"{"name":"Original"}"#)
590+
}
591+
let metadata = try ["prefix", "suffix"].map { channel in
592+
try signedEvent(
593+
privateKey: relayKey, createdAt: 100, kind: 39_000,
594+
tags: [["d", channel], ["name", "Original"], ["t", "stream"]]
595+
)
596+
}
597+
let memberships = try ["prefix", "suffix"].map { channel in
598+
try signedEvent(
599+
privateKey: relayKey, createdAt: 100, kind: 39_002,
600+
tags: [["d", channel], ["p", member]]
601+
)
602+
}
603+
func deliver(
604+
_ index: Int, community: String = "original", origin: String = "https://relay.example"
605+
) throws {
606+
try store.updateProfiles(
607+
communityID: community, relayOrigin: origin,
608+
updates: [BuzzPushProfileCacheUpdate(event: profiles[index])]
609+
)
610+
try store.updateChannels(
611+
communityID: community, relayOrigin: origin, relayMetadataPubkey: relayPubkey,
612+
metadataEvents: [metadata[index]], membershipEvents: [memberships[index]]
613+
)
614+
}
615+
616+
// Native storage accepts the prefix, but the caller loses its acknowledgement.
617+
try deliver(0)
618+
try deliver(0, community: "other-community")
619+
try deliver(0, origin: "https://other-relay.example")
620+
let beforeRetry = try loadSnapshot(directory)
621+
let untouchedProfiles = beforeRetry.profiles.filter {
622+
$0.communityID != "original" || $0.relayOrigin != "https://relay.example"
623+
}
624+
let untouchedChannels = beforeRetry.channels.filter {
625+
$0.communityID != "original" || $0.relayOrigin != "https://relay.example"
626+
}
627+
let newerProfile = try signedEvent(
628+
privateKey: profileKey, createdAt: 101, kind: 0, content: #"{"name":"Newest"}"#
629+
)
630+
let newerMetadata = try signedEvent(
631+
privateKey: relayKey, createdAt: 101, kind: 39_000,
632+
tags: [["d", "prefix"], ["name", "Newest"], ["t", "forum"]]
633+
)
634+
let newerMembership = try signedEvent(
635+
privateKey: relayKey, createdAt: 101, kind: 39_002,
636+
tags: [["d", "prefix"], ["p", member], ["p", addedMember]]
637+
)
638+
try store.updateProfiles(
639+
communityID: "original", relayOrigin: "https://relay.example",
640+
updates: [BuzzPushProfileCacheUpdate(event: newerProfile)]
641+
)
642+
try store.updateChannels(
643+
communityID: "original", relayOrigin: "https://relay.example",
644+
relayMetadataPubkey: relayPubkey,
645+
metadataEvents: [newerMetadata], membershipEvents: [newerMembership]
646+
)
647+
648+
// Replay exact events from the accepted prefix, then deliver the missing suffix.
649+
try deliver(0)
650+
try deliver(1)
651+
let completed = try loadSnapshot(directory)
652+
try deliver(0)
653+
try deliver(1)
654+
let replayed = try loadSnapshot(directory)
655+
#expect(replayed.profiles == completed.profiles)
656+
#expect(replayed.channels == completed.channels)
657+
#expect(replayed.profiles.count == 4)
658+
#expect(replayed.channels.count == 4)
659+
#expect(
660+
replayed.profiles.filter {
661+
$0.communityID != "original" || $0.relayOrigin != "https://relay.example"
662+
} == untouchedProfiles)
663+
#expect(
664+
replayed.channels.filter {
665+
$0.communityID != "original" || $0.relayOrigin != "https://relay.example"
666+
} == untouchedChannels)
667+
let profile = try #require(
668+
replayed.profile(
669+
communityID: "original", relayOrigin: "https://relay.example", pubkey: member
670+
))
671+
#expect(profile.eventID == newerProfile.id)
672+
#expect(profile.displayName == "Newest")
673+
let channel = try #require(
674+
replayed.channel(
675+
communityID: "original", relayOrigin: "https://relay.example", channelID: "prefix"
676+
))
677+
#expect(channel.eventID == newerMetadata.id)
678+
#expect(channel.displayName == "Newest")
679+
#expect(channel.channelType == "forum")
680+
#expect(channel.membershipEventID == newerMembership.id)
681+
#expect(channel.memberCount == 2)
682+
#expect(
683+
channel.memberDigests
684+
== [member, addedMember].map {
685+
BuzzPushPresentationIdentity.channelMember(
686+
communityID: "original", channelID: "prefix", pubkey: $0
687+
)
688+
}.sorted())
689+
#expect(
690+
replayed.profile(
691+
communityID: "original", relayOrigin: "https://relay.example", pubkey: addedMember
692+
)?.eventID == profiles[1].id)
693+
let suffix = try #require(
694+
replayed.channel(
695+
communityID: "original", relayOrigin: "https://relay.example", channelID: "suffix"
696+
))
697+
#expect(suffix.eventID == metadata[1].id)
698+
#expect(suffix.membershipEventID == memberships[1].id)
699+
}
700+
577701
@Test("Channel authority rotation clears membership signed by the old authority")
578702
func channelAuthorityRotationClearsMembership() throws {
579703
let directory = try temporaryDirectory()

‎mobile/lib/shared/push/push_presentation_cache.dart‎

Lines changed: 41 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,8 @@ bool isVerifiedPushPresentationEvent(NostrEvent event) {
4848

4949
/// Exports raw verified kind-0 events. Native code verifies them again before storage.
5050
/// Fails with [StateError] when eight exports are already outstanding.
51-
/// Native write failures propagate and stop any remaining chunks.
51+
/// Native persistence failures retry with bounded backoff, then propagate and
52+
/// stop any remaining chunks. Other native errors propagate immediately.
5253
Future<void> cacheBuzzPushProfileEvents(
5354
String communityID,
5455
Iterable<NostrEvent> events,
@@ -65,19 +66,21 @@ Future<void> cacheBuzzPushProfileEvents(
6566
debugLabel: 'buzz-push-profile-cache',
6667
);
6768
if (verified.isEmpty) return;
69+
final retryBudget = _NativeWriteRetryBudget();
6870
for (final chunk in _boundedChunks(verified, _maximumProfilesPerWrite)) {
69-
await _invokeSnapshot({
71+
await _invokeVerifiedChunk({
7072
'section': 'profiles',
7173
'communityId': communityID,
7274
'events': [for (final event in chunk) event.toJson()],
73-
});
75+
}, retryBudget);
7476
}
7577
});
7678
}
7779

7880
/// Exports verified channel metadata and membership for native authority checks.
7981
/// Fails with [StateError] when eight exports are already outstanding.
80-
/// Native write failures propagate and stop any remaining chunks.
82+
/// Native persistence failures retry with bounded backoff, then propagate and
83+
/// stop any remaining chunks. Other native errors propagate immediately.
8184
Future<void> cacheBuzzPushChannelEvents(
8285
String? communityID,
8386
Iterable<NostrEvent> metadataEvents,
@@ -107,10 +110,11 @@ Future<void> cacheBuzzPushChannelEvents(
107110
for (final event in verified.membership) event.getTagValue('d')!: event,
108111
};
109112
final channelIDs = {...metadata.keys, ...membership.keys}.toList();
113+
final retryBudget = _NativeWriteRetryBudget();
110114
// Keep each channel's metadata and membership in the same native write.
111115
// All chunks retain this export's FIFO slot until handoff is complete.
112116
for (final ids in _boundedChunks(channelIDs, _maximumChannelsPerWrite)) {
113-
await _invokeSnapshot({
117+
await _invokeVerifiedChunk({
114118
'section': 'channels',
115119
'communityId': communityID,
116120
'metadataEvents': [
@@ -121,7 +125,7 @@ Future<void> cacheBuzzPushChannelEvents(
121125
for (final id in ids)
122126
if (membership[id] case final event?) event.toJson(),
123127
],
124-
});
128+
}, retryBudget);
125129
}
126130
});
127131
}
@@ -264,6 +268,37 @@ Future<void> cacheBuzzPushAvatarFromLoadedBytes(
264268
}
265269
}
266270

271+
// One bounded backoff budget for the entire export, not one per chunk.
272+
class _NativeWriteRetryBudget {
273+
static const _delays = [250, 500, 1000, 2000, 4000];
274+
int _used = 0;
275+
276+
Duration? takeDelay() =>
277+
_used == _delays.length ? null : Duration(milliseconds: _delays[_used++]);
278+
}
279+
280+
// Retry only native persistence failures, retaining the verified payload and
281+
// the surrounding export's FIFO slot. Never repeat relay reads or verification.
282+
Future<void> _invokeVerifiedChunk(
283+
Map<String, Object> arguments,
284+
_NativeWriteRetryBudget budget,
285+
) async {
286+
final retryableCode = arguments['section'] == 'profiles'
287+
? 'profile_cache_failed'
288+
: 'channel_cache_failed';
289+
while (true) {
290+
try {
291+
await _invokeSnapshot(arguments);
292+
return;
293+
} on PlatformException catch (error) {
294+
if (error.code != retryableCode) rethrow;
295+
final delay = budget.takeDelay();
296+
if (delay == null) rethrow;
297+
await Future<void>.delayed(delay);
298+
}
299+
}
300+
}
301+
267302
Future<void> _invokeSnapshot(
268303
Map<String, Object> arguments, {
269304
bool bestEffort = false,

0 commit comments

Comments
 (0)