mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-24 23:45:18 +00:00
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:
@@ -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()
|
||||||
|
|||||||
Reference in New Issue
Block a user