Review fixes: offline-drop durability, gateway handoff retry, restore-log freshness, coalesced persist

Adversarial review of the storm/deadlock PR surfaced four issues:

- Offline blackhole (must-fix): a deposit made while relays were down
  persisted its dedup key even though the drop only sat in the in-memory
  pending queue — app killed before reconnect meant the relaunch lost the
  drop but the persisted key blocked every re-deposit for 24h. The
  persisted snapshot now excludes keys still pending; they become durable
  only when flushPendingDrops actually publishes them.
- Gateway handoff: seen-event IDs were consumed before the deliverToPeer
  handoff; a failed handoff (peer walked away) permanently dropped the
  event for a single-gateway island. deliverToPeer now reports whether
  the handoff was attempted, and a failure releases the seen slot so a
  relaunch or backlog redelivery retries.
- Restore-path logs: central/peripheral-restore captures logged the init
  sentinel bgRemaining=∞. The cache is now seeded in init's main-thread
  branch and restore captures route through the sampler, which refreshes
  the cached budget before logging.
- Persist cost: the dedup record was a full JSON encode + atomic write on
  the main actor per mutation (once per event during a backlog re-fetch).
  Writes now coalesce behind a 1s window, flushed immediately on
  background/terminate; panic wipe stays immediate.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
jack
2026-07-09 18:11:49 +02:00
co-authored by Claude Fable 5
parent d905452df2
commit 07b5ac31d8
5 changed files with 163 additions and 9 deletions
+13 -3
View File
@@ -281,12 +281,18 @@ final class BLEService: NSObject {
// Set up application state tracking (iOS only) // Set up application state tracking (iOS only)
#if os(iOS) #if os(iOS)
// Check initial state on main thread // Check initial state on main thread. The background-budget cache is
// seeded here too: a background-restore launch captures Bluetooth
// status before any lifecycle notification fires, and the init-time
// sentinel would log a meaningless bgRemaining= for exactly the
// wake window that matters.
if Thread.isMainThread { if Thread.isMainThread {
isAppActive = UIApplication.shared.applicationState == .active isAppActive = UIApplication.shared.applicationState == .active
refreshCachedBackgroundTimeRemaining()
} else { } else {
DispatchQueue.main.sync { DispatchQueue.main.sync {
isAppActive = UIApplication.shared.applicationState == .active isAppActive = UIApplication.shared.applicationState == .active
refreshCachedBackgroundTimeRemaining()
} }
} }
@@ -1729,7 +1735,10 @@ extension BLEService: CBCentralManagerDelegate {
recentPeripheralCache.record(peripheral, peripheralID: identifier, at: Date()) recentPeripheralCache.record(peripheral, peripheralID: identifier, at: Date())
} }
captureBluetoothStatus(context: "central-restore") // Via the sampler (not a direct capture): it refreshes the cached
// background budget on main first, so the restore log shows the real
// wake window instead of the init sentinel.
logBluetoothStatus("central-restore")
if central.state == .poweredOn { if central.state == .poweredOn {
startScanning() startScanning()
@@ -2488,7 +2497,8 @@ extension BLEService: CBPeripheralManagerDelegate {
} }
} }
captureBluetoothStatus(context: "peripheral-restore") // Via the sampler for a fresh background budget (see central-restore).
logBluetoothStatus("peripheral-restore")
if peripheral.state == .poweredOn && !peripheral.isAdvertising { if peripheral.state == .poweredOn && !peripheral.isAdvertising {
peripheral.startAdvertising(buildAdvertisementData()) peripheral.startAdvertising(buildAdvertisementData())
@@ -9,6 +9,11 @@
import BitFoundation import BitFoundation
import BitLogger import BitLogger
import Foundation import Foundation
#if os(iOS)
import UIKit
#elseif os(macOS)
import AppKit
#endif
/// Courier delivery over the internet bridge: sealed courier envelopes are /// Courier delivery over the internet bridge: sealed courier envelopes are
/// parked on relays as kind-1401 "drops" tagged with their rotating /// parked on relays as kind-1401 "drops" tagged with their rotating
@@ -48,6 +53,10 @@ final class BridgeCourierService: ObservableObject {
/// Encoded envelope cap for a drop (16 KiB ciphertext + TLV slack). /// Encoded envelope cap for a drop (16 KiB ciphertext + TLV slack).
static let maxDropEnvelopeBytes = 20 * 1024 static let maxDropEnvelopeBytes = 20 * 1024
static let maxTrackedIDs = 512 static let maxTrackedIDs = 512
/// Coalescing window for dedup-record writes: a backlog re-fetch
/// mutates the seen set once per event, and each snapshot save is a
/// full JSON encode + atomic write on the main actor.
static let dedupPersistCoalesceSeconds: TimeInterval = 1.0
} }
static let shared = BridgeCourierService() static let shared = BridgeCourierService()
@@ -70,7 +79,9 @@ final class BridgeCourierService: ObservableObject {
/// Opens a drop addressed to us (tag verified inside). /// Opens a drop addressed to us (tag verified inside).
var openEnvelope: (@MainActor (CourierEnvelope) -> Void)? var openEnvelope: (@MainActor (CourierEnvelope) -> Void)?
/// Hands a drop to a matching local peer as a directed courier packet. /// Hands a drop to a matching local peer as a directed courier packet.
var deliverToPeer: (@MainActor (CourierEnvelope, PeerID) -> Void)? /// Returns false when the handoff could not even be attempted (peer no
/// longer reachable), so the drop event stays retryable.
var deliverToPeer: (@MainActor (CourierEnvelope, PeerID) -> Bool)?
/// Held envelopes eligible for (re)publish, honoring the cooldown. /// Held envelopes eligible for (re)publish, honoring the cooldown.
var heldEnvelopes: (@MainActor (TimeInterval) -> [CourierEnvelope])? var heldEnvelopes: (@MainActor (TimeInterval) -> [CourierEnvelope])?
/// Timer injection for tests; nil arms a real `Task`. /// Timer injection for tests; nil arms a real `Task`.
@@ -98,6 +109,8 @@ final class BridgeCourierService: ObservableObject {
private let now: () -> Date private let now: () -> Date
private let dedupStore: BridgeDropDedupStore private let dedupStore: BridgeDropDedupStore
private var dedupPersistScheduled = false
init(now: @escaping () -> Date = Date.init, dedupStore: BridgeDropDedupStore? = nil) { init(now: @escaping () -> Date = Date.init, dedupStore: BridgeDropDedupStore? = nil) {
self.now = now self.now = now
self.dedupStore = dedupStore ?? BridgeDropDedupStore(persistsToDisk: !TestEnvironment.isRunningTests) self.dedupStore = dedupStore ?? BridgeDropDedupStore(persistsToDisk: !TestEnvironment.isRunningTests)
@@ -115,11 +128,44 @@ final class BridgeCourierService: ObservableObject {
entries: snapshot.seenDropEventIDs, entries: snapshot.seenDropEventIDs,
now: date now: date
) )
// A coalesced dedup write scheduled just before a background kill
// would be lost; flush when the app backgrounds or terminates.
#if os(iOS)
let flushNotifications = [UIApplication.didEnterBackgroundNotification, UIApplication.willTerminateNotification]
#else
let flushNotifications = [NSApplication.willTerminateNotification]
#endif
for name in flushNotifications {
NotificationCenter.default.addObserver(forName: name, object: nil, queue: .main) { [weak self] _ in
MainActor.assumeIsolated { self?.flushDedupSnapshot() }
}
}
} }
/// Schedules a coalesced write of the dedup record (see
/// `Limits.dedupPersistCoalesceSeconds`); lifecycle notifications flush
/// any scheduled write before a background kill could drop it.
private func persistDedup() { private func persistDedup() {
guard !dedupPersistScheduled else { return }
dedupPersistScheduled = true
Task { @MainActor [weak self] in
try? await Task.sleep(nanoseconds: UInt64(Limits.dedupPersistCoalesceSeconds * 1_000_000_000))
guard let self else { return }
self.dedupPersistScheduled = false
self.flushDedupSnapshot()
}
}
/// Writes the dedup record now. Sender keys still sitting in the
/// in-memory `pendingDrops` queue are excluded: their drop is not durable
/// until it actually reaches a relay, and persisting the key early would
/// turn "app killed before relays connected" into a silent 24h blackhole
/// (the relaunch loses the queued drop but the persisted key blocks every
/// re-deposit). `flushPendingDrops` re-persists once they publish.
func flushDedupSnapshot() {
let pendingKeys = Set(pendingDrops.compactMap(\.dedupKey))
dedupStore.save(BridgeDropDedupStore.Snapshot( dedupStore.save(BridgeDropDedupStore.Snapshot(
publishedDropKeys: publishedDropKeys.entries, publishedDropKeys: publishedDropKeys.entries.filter { !pendingKeys.contains($0.key) },
seenDropEventIDs: seenDropEventIDs.entries seenDropEventIDs: seenDropEventIDs.entries
)) ))
} }
@@ -221,9 +267,11 @@ final class BridgeCourierService: ObservableObject {
// retry sweep can attempt a fresh deposit. // retry sweep can attempt a fresh deposit.
if let key = item.dedupKey { if let key = item.dedupKey {
publishedDropKeys.remove(key) publishedDropKeys.remove(key)
persistDedup()
} }
} }
// Flushed keys just became durable (published, so no longer excluded
// as pending) or were released above; either way the record changed.
persistDedup()
} }
// MARK: - Subscription (recipient + gateway watch) // MARK: - Subscription (recipient + gateway watch)
@@ -328,7 +376,14 @@ final class BridgeCourierService: ObservableObject {
} }
if let match = watchedPeerTags.first(where: { $0.tagsHex.contains(tagHex) }) { if let match = watchedPeerTags.first(where: { $0.tagsHex.contains(tagHex) }) {
SecureLogger.info("📦🌉 Courier drop fetched for local peer \(match.peerID.id.prefix(8))", category: .session) SecureLogger.info("📦🌉 Courier drop fetched for local peer \(match.peerID.id.prefix(8))", category: .session)
deliverToPeer?(envelope, match.peerID) if deliverToPeer?(envelope, match.peerID) != true {
// The best-effort handoff never left this device (the peer
// walked away between the relay fetch and the mesh send).
// Release the seen slot so a relaunch or backlog redelivery
// retries a single-gateway island has no other carrier.
seenDropEventIDs.remove(event.id)
persistDedup()
}
} }
} }
@@ -97,6 +97,10 @@ final class BridgeDropDedupStore {
func save(_ snapshot: Snapshot) { func save(_ snapshot: Snapshot) {
guard let fileURL else { return } guard let fileURL else { return }
guard !(snapshot.publishedDropKeys.isEmpty && snapshot.seenDropEventIDs.isEmpty) else {
try? FileManager.default.removeItem(at: fileURL)
return
}
do { do {
try FileManager.default.createDirectory( try FileManager.default.createDirectory(
at: fileURL.deletingLastPathComponent(), at: fileURL.deletingLastPathComponent(),
@@ -567,7 +567,7 @@ private extension ChatViewModelBootstrapper {
bleService?.openBridgedCourierEnvelope(envelope) bleService?.openBridgedCourierEnvelope(envelope)
} }
courier.deliverToPeer = { [weak bleService] envelope, peerID in courier.deliverToPeer = { [weak bleService] envelope, peerID in
bleService?.deliverBridgedEnvelope(envelope, to: peerID) bleService?.deliverBridgedEnvelope(envelope, to: peerID) ?? false
} }
courier.heldEnvelopes = { cooldown in courier.heldEnvelopes = { cooldown in
CourierStore.shared.envelopesForBridgePublish(cooldown: cooldown) CourierStore.shared.envelopesForBridgePublish(cooldown: cooldown)
@@ -24,6 +24,7 @@ struct BridgeCourierServiceTests {
var localPeers: [(peerID: PeerID, noiseKey: Data)] = [] var localPeers: [(peerID: PeerID, noiseKey: Data)] = []
var held: [CourierEnvelope] = [] var held: [CourierEnvelope] = []
var sealResult: CourierEnvelope? var sealResult: CourierEnvelope?
var deliverResult = true
private(set) var publishedEvents: [NostrEvent] = [] private(set) var publishedEvents: [NostrEvent] = []
private(set) var openedSubscriptions: [[String]] = [] private(set) var openedSubscriptions: [[String]] = []
@@ -49,7 +50,10 @@ struct BridgeCourierServiceTests {
return self?.sealResult return self?.sealResult
} }
service.openEnvelope = { [weak self] envelope in self?.openedEnvelopes.append(envelope) } service.openEnvelope = { [weak self] envelope in self?.openedEnvelopes.append(envelope) }
service.deliverToPeer = { [weak self] envelope, peer in self?.delivered.append((envelope, peer)) } service.deliverToPeer = { [weak self] envelope, peer in
self?.delivered.append((envelope, peer))
return self?.deliverResult ?? false
}
service.heldEnvelopes = { [weak self] cooldown in service.heldEnvelopes = { [weak self] cooldown in
self?.heldCooldowns.append(cooldown) self?.heldCooldowns.append(cooldown)
return self?.held ?? [] return self?.held ?? []
@@ -190,6 +194,9 @@ struct BridgeCourierServiceTests {
fixture.sealResult = makeEnvelope(recipientKey: recipientKey) fixture.sealResult = makeEnvelope(recipientKey: recipientKey)
#expect(fixture.service.depositDrop(content: "hello", messageID: messageID, recipientNoiseKey: recipientKey)) #expect(fixture.service.depositDrop(content: "hello", messageID: messageID, recipientNoiseKey: recipientKey))
#expect(fixture.publishedEvents.count == 1) #expect(fixture.publishedEvents.count == 1)
// Persistence is coalesced; a real launch flushes within a second or
// on backgrounding tests flush explicitly.
fixture.service.flushDedupSnapshot()
// "Relaunch": a fresh service over the same store must refuse to // "Relaunch": a fresh service over the same store must refuse to
// publish the same message ID again (before even re-sealing it). // publish the same message ID again (before even re-sealing it).
@@ -214,6 +221,7 @@ struct BridgeCourierServiceTests {
let event = try makeDropEvent(for: makeEnvelope(recipientKey: myKey)) let event = try makeDropEvent(for: makeEnvelope(recipientKey: myKey))
fixture.service.handleDropEvent(event) fixture.service.handleDropEvent(event)
#expect(fixture.openedEnvelopes.count == 1) #expect(fixture.openedEnvelopes.count == 1)
fixture.service.flushDedupSnapshot()
let relaunched = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL)) let relaunched = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL))
relaunched.myKey = myKey relaunched.myKey = myKey
@@ -222,6 +230,83 @@ struct BridgeCourierServiceTests {
#expect(relaunched.openedEnvelopes.isEmpty) #expect(relaunched.openedEnvelopes.isEmpty)
} }
@Test func offlineQueuedDropStaysRedepositableAfterRelaunch() throws {
// A deposit made while relays are down only joins the in-memory
// pending queue. Its dedup key must NOT be durable yet: if the app is
// killed before relays connect, the relaunch loses the queued drop
// a persisted key would then block every 120s re-deposit for 24h and
// the message would silently never reach a relay.
let fileURL = FileManager.default.temporaryDirectory
.appendingPathComponent("bridge-dedup-\(UUID().uuidString).json")
defer { try? FileManager.default.removeItem(at: fileURL) }
let recipientKey = Fixture.randomKey()
let messageID = UUID().uuidString
let fixture = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL))
fixture.relaysConnected = false
fixture.sealResult = makeEnvelope(recipientKey: recipientKey)
#expect(fixture.service.depositDrop(content: "later", messageID: messageID, recipientNoiseKey: recipientKey))
#expect(fixture.publishedEvents.isEmpty)
// Even a flush while the drop is still pending must exclude its key.
fixture.service.flushDedupSnapshot()
// "App killed before relays connected": pendingDrops were memory-only.
let relaunched = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL))
relaunched.sealResult = makeEnvelope(recipientKey: recipientKey)
#expect(relaunched.service.depositDrop(content: "later", messageID: messageID, recipientNoiseKey: recipientKey))
#expect(relaunched.publishedEvents.count == 1)
}
@Test func publishedPendingDropBecomesDurableAfterFlush() throws {
// Counterpart: once the queued drop actually publishes on reconnect,
// its key becomes durable and a relaunch must not republish.
let fileURL = FileManager.default.temporaryDirectory
.appendingPathComponent("bridge-dedup-\(UUID().uuidString).json")
defer { try? FileManager.default.removeItem(at: fileURL) }
let recipientKey = Fixture.randomKey()
let messageID = UUID().uuidString
let fixture = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL))
fixture.relaysConnected = false
fixture.sealResult = makeEnvelope(recipientKey: recipientKey)
#expect(fixture.service.depositDrop(content: "later", messageID: messageID, recipientNoiseKey: recipientKey))
fixture.relaysConnected = true
fixture.service.flushPendingDrops()
#expect(fixture.publishedEvents.count == 1)
fixture.service.flushDedupSnapshot()
let relaunched = Fixture(dedupStore: BridgeDropDedupStore(fileURL: fileURL))
relaunched.sealResult = makeEnvelope(recipientKey: recipientKey)
#expect(!relaunched.service.depositDrop(content: "later", messageID: messageID, recipientNoiseKey: recipientKey))
#expect(relaunched.publishedEvents.isEmpty)
}
@Test func failedGatewayHandoffReleasesSeenSlot() throws {
// A gateway's deliverToPeer handoff is best-effort: when it fails
// (the peer walked away between relay fetch and mesh send), the drop
// event must stay retryable for a single-gateway mesh island this
// gateway is the recipient's only carrier.
let fixture = Fixture()
let peerKey = Fixture.randomKey()
let peer = PeerID(str: "aabbccdd00112233")
fixture.localPeers = [(peer, peerKey)]
fixture.service.refresh()
let event = try makeDropEvent(for: makeEnvelope(recipientKey: peerKey))
fixture.deliverResult = false
fixture.service.handleDropEvent(event)
#expect(fixture.delivered.count == 1)
// Redelivery (relaunch/backlog re-fetch) retries the handoff
fixture.deliverResult = true
fixture.service.handleDropEvent(event)
#expect(fixture.delivered.count == 2)
// and a successful handoff consumes the event for good.
fixture.service.handleDropEvent(event)
#expect(fixture.delivered.count == 2)
}
@Test func distinctDropsUseDistinctThrowawayKeys() { @Test func distinctDropsUseDistinctThrowawayKeys() {
let fixture = Fixture() let fixture = Fixture()
let keyA = Fixture.randomKey() let keyA = Fixture.randomKey()