mirror of
https://github.com/permissionlesstech/bitchat.git
synced 2026-07-25 02:25:20 +00:00
Reconnect hygiene: paced acks, persistent gift-wrap dedup, self-fragment sync fix (#1398)
* Reconnect hygiene: paced acks, persistent gift-wrap dedup, self-fragment sync fix
Remaining findings from the July 7 locked-phone test sessions, all rooted
in reconnect/relaunch behavior:
- All Nostr acks (READ and DELIVERED, direct and geohash) now flow through
the paced queue that previously only throttled direct READ receipts.
Reconnect redelivery produced 8 DELIVERED acks in under a second, which
damus rejects ("noting too much").
- Processed gift-wrap event IDs persist across launches
(NostrProcessedEventStore, wired through MessageDeduplicationService,
debounced writes, wiped on the existing clear/panic paths). NIP-59
randomizes gift-wrap timestamps, so the 24h-lookback DM subscriptions
redeliver the same events every launch; without a cross-launch record
each relaunch reprocessed old PMs and acks — the re-ack bursts and the
"delivered ack for unknown mid" warnings (now debug: a stale ack is
expected occasionally and not actionable).
- Own fragments handed back by sync replay (the deliberate RSR ttl=0
restore path) now re-enter the gossip sync store before the self-drop.
The fragment store is not archived, so after a relaunch our sync filter
did not cover our own fragments and peers re-offered them every 30s
round indefinitely; recording them stops the redelivery after one round
while keeping assembly skipped.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
* Address Codex review on #1398: shared ack pacer, transient clears keep disk record
- Geohash acks are sent through short-lived NostrTransport instances
(makeGeohashNostrTransport creates one per ack), so the per-instance ack
queue never paced a burst. Acks now flow through a pacer shared across
instances: Dependencies.live wires the process-wide sharedAckPacer, and
the default Dependencies init builds an isolated pacer from the same
injected scheduleAfter so tests keep stepping the throttle manually.
- clearNostrCaches() runs on every geohash channel switch, so it no longer
wipes the persisted gift-wrap record (that stays on the clearAll/panic
path). Persistence is now append-merge instead of snapshot-overwrite —
serialized on the store's IO queue — so a transient in-memory clear
between debounced flushes can't shrink the on-disk record either.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---------
Co-authored-by: jack <jackjackbits@users.noreply.github.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
jack
Claude Fable 5
parent
963d3a089f
commit
81c45e9649
@@ -33,13 +33,21 @@ final class BLEFragmentHandler {
|
|||||||
|
|
||||||
func handle(_ packet: BitchatPacket, from peerID: PeerID) {
|
func handle(_ packet: BitchatPacket, from peerID: PeerID) {
|
||||||
let env = environment
|
let env = environment
|
||||||
// Don't process our own fragments
|
guard let header = BLEFragmentHeader(packet: packet) else { return }
|
||||||
|
|
||||||
|
// Sync replay legitimately hands us our own fragments back (the RSR
|
||||||
|
// ttl=0 restore path): after a relaunch the fragment store starts
|
||||||
|
// empty, so our sync filter doesn't cover them and peers re-offer
|
||||||
|
// them. Record them as seen — the next round's filter then covers
|
||||||
|
// them and the redelivery stops — but skip assembly: we authored
|
||||||
|
// the original, there is nothing to reassemble.
|
||||||
if peerID == env.localPeerID() {
|
if peerID == env.localPeerID() {
|
||||||
|
if header.isBroadcastFragment {
|
||||||
|
env.trackPacketSeen(packet)
|
||||||
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
guard let header = BLEFragmentHeader(packet: packet) else { return }
|
|
||||||
|
|
||||||
if header.isBroadcastFragment {
|
if header.isBroadcastFragment {
|
||||||
env.trackPacketSeen(packet)
|
env.trackPacketSeen(packet)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -172,17 +172,37 @@ final class MessageDeduplicationService {
|
|||||||
/// Cache for Nostr ACK deduplication (messageId:ackType:senderPubkey format)
|
/// Cache for Nostr ACK deduplication (messageId:ackType:senderPubkey format)
|
||||||
private let nostrAckCache: LRUDeduplicationCache<Bool>
|
private let nostrAckCache: LRUDeduplicationCache<Bool>
|
||||||
|
|
||||||
|
/// Optional cross-launch persistence for the Nostr event cache. NIP-59
|
||||||
|
/// randomizes gift-wrap timestamps, so DM subscriptions look back 24h and
|
||||||
|
/// relays redeliver the same events on every launch; without this record
|
||||||
|
/// each relaunch reprocesses old PMs and acks. Nil (tests, macOS callers
|
||||||
|
/// that don't opt in) keeps the cache purely in-memory.
|
||||||
|
private let nostrEventStore: NostrProcessedEventStore?
|
||||||
|
private let nostrEventCapacity: Int
|
||||||
|
private var persistScheduled = false
|
||||||
|
private var pendingPersistIDs: [String] = []
|
||||||
|
|
||||||
/// Creates a new deduplication service with specified capacities.
|
/// Creates a new deduplication service with specified capacities.
|
||||||
/// - Parameters:
|
/// - Parameters:
|
||||||
/// - contentCapacity: Max entries for content cache
|
/// - contentCapacity: Max entries for content cache
|
||||||
/// - nostrEventCapacity: Max entries for Nostr event cache
|
/// - nostrEventCapacity: Max entries for Nostr event cache
|
||||||
|
/// - nostrEventStore: Optional disk store preloading and persisting
|
||||||
|
/// processed Nostr event IDs across launches
|
||||||
init(
|
init(
|
||||||
contentCapacity: Int = TransportConfig.contentLRUCap,
|
contentCapacity: Int = TransportConfig.contentLRUCap,
|
||||||
nostrEventCapacity: Int = TransportConfig.uiProcessedNostrEventsCap
|
nostrEventCapacity: Int = TransportConfig.uiProcessedNostrEventsCap,
|
||||||
|
nostrEventStore: NostrProcessedEventStore? = nil
|
||||||
) {
|
) {
|
||||||
self.contentCache = LRUDeduplicationCache(capacity: contentCapacity)
|
self.contentCache = LRUDeduplicationCache(capacity: contentCapacity)
|
||||||
self.nostrEventCache = LRUDeduplicationCache(capacity: nostrEventCapacity)
|
self.nostrEventCache = LRUDeduplicationCache(capacity: nostrEventCapacity)
|
||||||
self.nostrAckCache = LRUDeduplicationCache(capacity: nostrEventCapacity)
|
self.nostrAckCache = LRUDeduplicationCache(capacity: nostrEventCapacity)
|
||||||
|
self.nostrEventStore = nostrEventStore
|
||||||
|
self.nostrEventCapacity = nostrEventCapacity
|
||||||
|
if let nostrEventStore {
|
||||||
|
for eventID in nostrEventStore.load() {
|
||||||
|
nostrEventCache.record(eventID, value: true)
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// MARK: - Content Deduplication
|
// MARK: - Content Deduplication
|
||||||
@@ -239,6 +259,26 @@ final class MessageDeduplicationService {
|
|||||||
/// - Parameter eventId: The event ID
|
/// - Parameter eventId: The event ID
|
||||||
func recordNostrEvent(_ eventId: String) {
|
func recordNostrEvent(_ eventId: String) {
|
||||||
nostrEventCache.record(eventId, value: true)
|
nostrEventCache.record(eventId, value: true)
|
||||||
|
if nostrEventStore != nil {
|
||||||
|
pendingPersistIDs.append(eventId)
|
||||||
|
schedulePersistIfNeeded()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Debounced persistence: bursts of inbound events (reconnect redelivery)
|
||||||
|
/// collapse into one append. Append-merge rather than snapshot, so a
|
||||||
|
/// transient in-memory clear between flushes can't shrink the disk record.
|
||||||
|
private func schedulePersistIfNeeded() {
|
||||||
|
guard let nostrEventStore, !persistScheduled else { return }
|
||||||
|
persistScheduled = true
|
||||||
|
Task { @MainActor [weak self] in
|
||||||
|
try? await Task.sleep(nanoseconds: 2_000_000_000)
|
||||||
|
guard let self else { return }
|
||||||
|
self.persistScheduled = false
|
||||||
|
let newIDs = self.pendingPersistIDs
|
||||||
|
self.pendingPersistIDs.removeAll()
|
||||||
|
nostrEventStore.append(newIDs, cap: self.nostrEventCapacity)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// MARK: - Nostr ACK Deduplication
|
// MARK: - Nostr ACK Deduplication
|
||||||
@@ -263,14 +303,20 @@ final class MessageDeduplicationService {
|
|||||||
|
|
||||||
// MARK: - Clear
|
// MARK: - Clear
|
||||||
|
|
||||||
/// Clears all caches
|
/// Clears all caches. This is the wipe/panic path: the persisted
|
||||||
|
/// gift-wrap record goes with everything else.
|
||||||
func clearAll() {
|
func clearAll() {
|
||||||
contentCache.clear()
|
contentCache.clear()
|
||||||
nostrEventCache.clear()
|
nostrEventCache.clear()
|
||||||
nostrAckCache.clear()
|
nostrAckCache.clear()
|
||||||
|
pendingPersistIDs.removeAll()
|
||||||
|
nostrEventStore?.wipe()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Clears only the Nostr caches (events and ACKs)
|
/// Clears only the in-memory Nostr caches (events and ACKs). Runs on
|
||||||
|
/// every geohash channel switch, so the disk record deliberately
|
||||||
|
/// survives — wiping it here would forfeit cross-launch gift-wrap dedup
|
||||||
|
/// each time the user changes channels (flagged by Codex on #1398).
|
||||||
func clearNostrCaches() {
|
func clearNostrCaches() {
|
||||||
nostrEventCache.clear()
|
nostrEventCache.clear()
|
||||||
nostrAckCache.clear()
|
nostrAckCache.clear()
|
||||||
|
|||||||
@@ -0,0 +1,106 @@
|
|||||||
|
//
|
||||||
|
// NostrProcessedEventStore.swift
|
||||||
|
// bitchat
|
||||||
|
//
|
||||||
|
// This is free and unencumbered software released into the public domain.
|
||||||
|
// For more information, see <https://unlicense.org>
|
||||||
|
//
|
||||||
|
|
||||||
|
import BitLogger
|
||||||
|
import Foundation
|
||||||
|
|
||||||
|
/// Disk persistence for processed gift-wrap event IDs. NIP-59 randomizes
|
||||||
|
/// gift-wrap timestamps, so DM subscriptions must look back generously (24h)
|
||||||
|
/// and relays redeliver the same events on every launch — without a
|
||||||
|
/// cross-launch record, each relaunch reprocesses old PMs and acks
|
||||||
|
/// (re-sent DELIVERED bursts, "delivered ack for unknown mid" noise).
|
||||||
|
///
|
||||||
|
/// Contents are event IDs already visible to every relay, so
|
||||||
|
/// until-first-unlock file protection is the right at-rest posture — the
|
||||||
|
/// file must also load during a locked-background restoration relaunch.
|
||||||
|
/// Wiped on panic via the dedup service's clear paths.
|
||||||
|
final class NostrProcessedEventStore {
|
||||||
|
private let fileURL: URL?
|
||||||
|
// All file access is serialized here: appends are read-modify-write, and
|
||||||
|
// overlapping debounced flushes would otherwise race and drop IDs.
|
||||||
|
private let ioQueue = DispatchQueue(label: "chat.bitchat.nostr-processed-events", qos: .utility)
|
||||||
|
|
||||||
|
init(fileURL: URL? = nil) {
|
||||||
|
self.fileURL = fileURL ?? Self.defaultFileURL()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Processed event IDs, oldest first (insertion order).
|
||||||
|
func load() -> [String] {
|
||||||
|
ioQueue.sync { loadLocked() }
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Merge new IDs onto the persisted record, oldest-first, trimming from
|
||||||
|
/// the front past `cap`. Append-merge (not snapshot-overwrite) so the
|
||||||
|
/// in-memory cache being cleared transiently (channel switches) can
|
||||||
|
/// never shrink the on-disk record.
|
||||||
|
func append(_ newIDs: [String], cap: Int) {
|
||||||
|
guard !newIDs.isEmpty else { return }
|
||||||
|
ioQueue.async { [self] in
|
||||||
|
var merged = loadLocked()
|
||||||
|
var known = Set(merged)
|
||||||
|
for id in newIDs where !known.contains(id) {
|
||||||
|
merged.append(id)
|
||||||
|
known.insert(id)
|
||||||
|
}
|
||||||
|
if merged.count > cap {
|
||||||
|
merged.removeFirst(merged.count - cap)
|
||||||
|
}
|
||||||
|
saveLocked(merged)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func wipe() {
|
||||||
|
ioQueue.async { [self] in
|
||||||
|
guard let fileURL else { return }
|
||||||
|
try? FileManager.default.removeItem(at: fileURL)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private func loadLocked() -> [String] {
|
||||||
|
guard let fileURL,
|
||||||
|
let data = try? Data(contentsOf: fileURL),
|
||||||
|
let ids = try? JSONDecoder().decode([String].self, from: data) else {
|
||||||
|
return []
|
||||||
|
}
|
||||||
|
return ids
|
||||||
|
}
|
||||||
|
|
||||||
|
private func saveLocked(_ eventIDs: [String]) {
|
||||||
|
guard let fileURL else { return }
|
||||||
|
guard !eventIDs.isEmpty else {
|
||||||
|
try? FileManager.default.removeItem(at: fileURL)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
do {
|
||||||
|
try FileManager.default.createDirectory(
|
||||||
|
at: fileURL.deletingLastPathComponent(),
|
||||||
|
withIntermediateDirectories: true
|
||||||
|
)
|
||||||
|
let data = try JSONEncoder().encode(eventIDs)
|
||||||
|
var options: Data.WritingOptions = [.atomic]
|
||||||
|
#if os(iOS)
|
||||||
|
options.insert(.completeFileProtectionUntilFirstUserAuthentication)
|
||||||
|
#endif
|
||||||
|
try data.write(to: fileURL, options: options)
|
||||||
|
} catch {
|
||||||
|
SecureLogger.error("Failed to persist processed Nostr events: \(error)", category: .session)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static func defaultFileURL() -> URL? {
|
||||||
|
guard let base = try? FileManager.default.url(
|
||||||
|
for: .applicationSupportDirectory,
|
||||||
|
in: .userDomainMask,
|
||||||
|
appropriateFor: nil,
|
||||||
|
create: true
|
||||||
|
) else { return nil }
|
||||||
|
return base
|
||||||
|
.appendingPathComponent("nostr", isDirectory: true)
|
||||||
|
.appendingPathComponent("processed-events.json")
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -19,6 +19,36 @@ final class NostrTransport: Transport, @unchecked Sendable {
|
|||||||
/// doesn't count: DM sends target the default relay set and would
|
/// doesn't count: DM sends target the default relay set and would
|
||||||
/// still queue.
|
/// still queue.
|
||||||
let relayConnectivity: @MainActor () -> AnyPublisher<Bool, Never>
|
let relayConnectivity: @MainActor () -> AnyPublisher<Bool, Never>
|
||||||
|
/// Paces outbound acks. Defaults to an isolated pacer so tests don't
|
||||||
|
/// serialize behind each other; `live` passes the process-wide one.
|
||||||
|
let ackPacer: AckPacer
|
||||||
|
|
||||||
|
init(
|
||||||
|
notificationCenter: NotificationCenter,
|
||||||
|
loadFavorites: @escaping @MainActor () -> [Data: FavoritesPersistenceService.FavoriteRelationship],
|
||||||
|
favoriteStatusForNoiseKey: @escaping @MainActor (Data) -> FavoritesPersistenceService.FavoriteRelationship?,
|
||||||
|
favoriteStatusForPeerID: @escaping @MainActor (PeerID) -> FavoritesPersistenceService.FavoriteRelationship?,
|
||||||
|
currentIdentity: @escaping @MainActor () throws -> NostrIdentity?,
|
||||||
|
registerPendingGiftWrap: @escaping @MainActor (String) -> Void,
|
||||||
|
sendEvent: @escaping @MainActor (NostrEvent) -> Void,
|
||||||
|
scheduleAfter: @escaping @Sendable (TimeInterval, @escaping @Sendable () -> Void) -> Void,
|
||||||
|
relayConnectivity: @escaping @MainActor () -> AnyPublisher<Bool, Never>,
|
||||||
|
ackPacer: AckPacer? = nil
|
||||||
|
) {
|
||||||
|
self.notificationCenter = notificationCenter
|
||||||
|
self.loadFavorites = loadFavorites
|
||||||
|
self.favoriteStatusForNoiseKey = favoriteStatusForNoiseKey
|
||||||
|
self.favoriteStatusForPeerID = favoriteStatusForPeerID
|
||||||
|
self.currentIdentity = currentIdentity
|
||||||
|
self.registerPendingGiftWrap = registerPendingGiftWrap
|
||||||
|
self.sendEvent = sendEvent
|
||||||
|
self.scheduleAfter = scheduleAfter
|
||||||
|
self.relayConnectivity = relayConnectivity
|
||||||
|
// Default pacer drives its throttle through the same injected
|
||||||
|
// scheduler, so tests that step scheduleAfter manually keep
|
||||||
|
// control of the ack cadence.
|
||||||
|
self.ackPacer = ackPacer ?? AckPacer(scheduleAfter: scheduleAfter)
|
||||||
|
}
|
||||||
|
|
||||||
@MainActor
|
@MainActor
|
||||||
static func live(idBridge: NostrIdentityBridge) -> Dependencies {
|
static func live(idBridge: NostrIdentityBridge) -> Dependencies {
|
||||||
@@ -33,7 +63,8 @@ final class NostrTransport: Transport, @unchecked Sendable {
|
|||||||
scheduleAfter: { delay, action in
|
scheduleAfter: { delay, action in
|
||||||
DispatchQueue.main.asyncAfter(deadline: .now() + delay, execute: action)
|
DispatchQueue.main.asyncAfter(deadline: .now() + delay, execute: action)
|
||||||
},
|
},
|
||||||
relayConnectivity: { NostrRelayManager.shared.$isDMRelayConnected.eraseToAnyPublisher() }
|
relayConnectivity: { NostrRelayManager.shared.$isDMRelayConnected.eraseToAnyPublisher() },
|
||||||
|
ackPacer: NostrTransport.sharedAckPacer
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -41,14 +72,61 @@ final class NostrTransport: Transport, @unchecked Sendable {
|
|||||||
// Provide BLE short peer ID for BitChat embedding
|
// Provide BLE short peer ID for BitChat embedding
|
||||||
var senderPeerID = PeerID(str: "")
|
var senderPeerID = PeerID(str: "")
|
||||||
|
|
||||||
// Throttle READ receipts to avoid relay rate limits
|
// Throttle outbound acks — READ receipts and DELIVERED acks, direct and
|
||||||
private struct QueuedRead {
|
// geohash — to avoid relay rate limits. Reconnect redelivery produces a
|
||||||
let receipt: ReadReceipt
|
// burst of acks at once: 8 DELIVERED in under a second tripped damus's
|
||||||
let peerID: PeerID
|
// "noting too much" during July 2026 device testing.
|
||||||
|
private enum QueuedAck {
|
||||||
|
case readDirect(ReadReceipt, PeerID)
|
||||||
|
case deliveredDirect(messageID: String, peerID: PeerID)
|
||||||
|
case deliveredGeohash(messageID: String, recipientHex: String, identity: NostrIdentity)
|
||||||
|
case readGeohash(messageID: String, recipientHex: String, identity: NostrIdentity)
|
||||||
}
|
}
|
||||||
private var readQueue: [QueuedRead] = []
|
|
||||||
private var isSendingReadAcks = false
|
/// Ack pacing shared across transport instances. Geohash acks are sent
|
||||||
private let readAckInterval: TimeInterval = TransportConfig.nostrReadAckInterval
|
/// through short-lived transports created per ack
|
||||||
|
/// (`makeGeohashNostrTransport()`), so a per-instance queue would only
|
||||||
|
/// ever hold one item and never pace a burst (flagged by Codex on
|
||||||
|
/// #1398). Production wires `sharedAckPacer` via `Dependencies.live`;
|
||||||
|
/// tests get an isolated instance per `Dependencies` by default.
|
||||||
|
final class AckPacer {
|
||||||
|
typealias Scheduler = @Sendable (TimeInterval, @escaping @Sendable () -> Void) -> Void
|
||||||
|
|
||||||
|
private let queue = DispatchQueue(label: "chat.bitchat.nostr-ack-pacer")
|
||||||
|
private var pending: [() -> Void] = []
|
||||||
|
private var isSending = false
|
||||||
|
private let interval: TimeInterval = TransportConfig.nostrReadAckInterval
|
||||||
|
private let scheduleAfter: Scheduler
|
||||||
|
|
||||||
|
init(scheduleAfter: @escaping Scheduler = { delay, action in
|
||||||
|
DispatchQueue.global(qos: .utility).asyncAfter(deadline: .now() + delay, execute: action)
|
||||||
|
}) {
|
||||||
|
self.scheduleAfter = scheduleAfter
|
||||||
|
}
|
||||||
|
|
||||||
|
func enqueue(_ send: @escaping () -> Void) {
|
||||||
|
queue.async {
|
||||||
|
self.pending.append(send)
|
||||||
|
self.processNext()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Must be called on `queue`.
|
||||||
|
private func processNext() {
|
||||||
|
guard !isSending, !pending.isEmpty else { return }
|
||||||
|
isSending = true
|
||||||
|
let send = pending.removeFirst()
|
||||||
|
send()
|
||||||
|
scheduleAfter(interval) { [weak self] in
|
||||||
|
guard let self else { return }
|
||||||
|
self.queue.async {
|
||||||
|
self.isSending = false
|
||||||
|
self.processNext()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
static let sharedAckPacer = AckPacer()
|
||||||
private let keychain: KeychainManagerProtocol
|
private let keychain: KeychainManagerProtocol
|
||||||
private let idBridge: NostrIdentityBridge
|
private let idBridge: NostrIdentityBridge
|
||||||
private let dependencies: Dependencies
|
private let dependencies: Dependencies
|
||||||
@@ -187,12 +265,14 @@ final class NostrTransport: Transport, @unchecked Sendable {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func sendReadReceipt(_ receipt: ReadReceipt, to peerID: PeerID) {
|
func sendReadReceipt(_ receipt: ReadReceipt, to peerID: PeerID) {
|
||||||
// Enqueue and process with throttling to avoid relay rate limits
|
enqueueAck(.readDirect(receipt, peerID))
|
||||||
// Use barrier to synchronize access to readQueue
|
|
||||||
queue.async(flags: .barrier) {
|
|
||||||
self.readQueue.append(QueuedRead(receipt: receipt, peerID: peerID))
|
|
||||||
self.processReadQueueIfNeeded()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Enqueue an ack for paced sending. Captures self strongly on purpose:
|
||||||
|
/// geohash acks ride throwaway transport instances that must stay alive
|
||||||
|
/// until their ack leaves the queue.
|
||||||
|
private func enqueueAck(_ ack: QueuedAck) {
|
||||||
|
dependencies.ackPacer.enqueue { self.sendAckItem(ack) }
|
||||||
}
|
}
|
||||||
|
|
||||||
func sendFavoriteNotification(to peerID: PeerID, isFavorite: Bool) {
|
func sendFavoriteNotification(to peerID: PeerID, isFavorite: Bool) {
|
||||||
@@ -212,17 +292,7 @@ final class NostrTransport: Transport, @unchecked Sendable {
|
|||||||
|
|
||||||
func sendBroadcastAnnounce() { /* no-op for Nostr */ }
|
func sendBroadcastAnnounce() { /* no-op for Nostr */ }
|
||||||
func sendDeliveryAck(for messageID: String, to peerID: PeerID) {
|
func sendDeliveryAck(for messageID: String, to peerID: PeerID) {
|
||||||
Task { @MainActor in
|
enqueueAck(.deliveredDirect(messageID: messageID, peerID: peerID))
|
||||||
guard let recipientNpub = resolveRecipientNpub(for: peerID),
|
|
||||||
let recipientHex = npubToHex(recipientNpub),
|
|
||||||
let senderIdentity = try? dependencies.currentIdentity() else { return }
|
|
||||||
SecureLogger.debug("NostrTransport: preparing DELIVERED ack id=\(messageID.prefix(8))…", category: .session)
|
|
||||||
guard let ack = NostrEmbeddedBitChat.encodeAckForNostr(type: .delivered, messageID: messageID, recipientPeerID: peerID, senderPeerID: senderPeerID) else {
|
|
||||||
SecureLogger.error("NostrTransport: failed to embed DELIVERED ack", category: .session)
|
|
||||||
return
|
|
||||||
}
|
|
||||||
sendWrappedMessage(content: ack, recipientHex: recipientHex, senderIdentity: senderIdentity)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -232,19 +302,11 @@ extension NostrTransport {
|
|||||||
|
|
||||||
// MARK: Geohash ACK helpers
|
// MARK: Geohash ACK helpers
|
||||||
func sendDeliveryAckGeohash(for messageID: String, toRecipientHex recipientHex: String, from identity: NostrIdentity) {
|
func sendDeliveryAckGeohash(for messageID: String, toRecipientHex recipientHex: String, from identity: NostrIdentity) {
|
||||||
Task { @MainActor in
|
enqueueAck(.deliveredGeohash(messageID: messageID, recipientHex: recipientHex, identity: identity))
|
||||||
SecureLogger.debug("GeoDM: send DELIVERED mid=\(messageID.prefix(8))…", category: .session)
|
|
||||||
guard let embedded = NostrEmbeddedBitChat.encodeAckForNostrNoRecipient(type: .delivered, messageID: messageID, senderPeerID: senderPeerID) else { return }
|
|
||||||
sendWrappedMessage(content: embedded, recipientHex: recipientHex, senderIdentity: identity, registerPending: true)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func sendReadReceiptGeohash(_ messageID: String, toRecipientHex recipientHex: String, from identity: NostrIdentity) {
|
func sendReadReceiptGeohash(_ messageID: String, toRecipientHex recipientHex: String, from identity: NostrIdentity) {
|
||||||
Task { @MainActor in
|
enqueueAck(.readGeohash(messageID: messageID, recipientHex: recipientHex, identity: identity))
|
||||||
SecureLogger.debug("GeoDM: send READ mid=\(messageID.prefix(8))…", category: .session)
|
|
||||||
guard let embedded = NostrEmbeddedBitChat.encodeAckForNostrNoRecipient(type: .readReceipt, messageID: messageID, senderPeerID: senderPeerID) else { return }
|
|
||||||
sendWrappedMessage(content: embedded, recipientHex: recipientHex, senderIdentity: identity, registerPending: true)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// MARK: Geohash DMs (per-geohash identity)
|
// MARK: Geohash DMs (per-geohash identity)
|
||||||
@@ -290,36 +352,42 @@ extension NostrTransport {
|
|||||||
dependencies.sendEvent(event)
|
dependencies.sendEvent(event)
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Must be called within a barrier on `queue`
|
|
||||||
private func processReadQueueIfNeeded() {
|
|
||||||
guard !isSendingReadAcks else { return }
|
|
||||||
guard !readQueue.isEmpty else { return }
|
|
||||||
isSendingReadAcks = true
|
|
||||||
let item = readQueue.removeFirst()
|
|
||||||
sendReadAckItem(item)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Sends a single read ack item (called after extraction from queue within barrier)
|
/// Sends a single ack item (invoked by the pacer, one per interval)
|
||||||
private func sendReadAckItem(_ item: QueuedRead) {
|
private func sendAckItem(_ item: QueuedAck) {
|
||||||
Task { @MainActor in
|
Task { @MainActor in
|
||||||
defer { scheduleNextReadAck() }
|
switch item {
|
||||||
guard let recipientNpub = resolveRecipientNpub(for: item.peerID),
|
case .readDirect(let receipt, let peerID):
|
||||||
|
guard let recipientNpub = resolveRecipientNpub(for: peerID),
|
||||||
let recipientHex = npubToHex(recipientNpub),
|
let recipientHex = npubToHex(recipientNpub),
|
||||||
let senderIdentity = try? dependencies.currentIdentity() else { return }
|
let senderIdentity = try? dependencies.currentIdentity() else { return }
|
||||||
SecureLogger.debug("NostrTransport: preparing READ ack id=\(item.receipt.originalMessageID.prefix(8))…", category: .session)
|
SecureLogger.debug("NostrTransport: preparing READ ack id=\(receipt.originalMessageID.prefix(8))…", category: .session)
|
||||||
guard let ack = NostrEmbeddedBitChat.encodeAckForNostr(type: .readReceipt, messageID: item.receipt.originalMessageID, recipientPeerID: item.peerID, senderPeerID: senderPeerID) else {
|
guard let ack = NostrEmbeddedBitChat.encodeAckForNostr(type: .readReceipt, messageID: receipt.originalMessageID, recipientPeerID: peerID, senderPeerID: senderPeerID) else {
|
||||||
SecureLogger.error("NostrTransport: failed to embed READ ack", category: .session)
|
SecureLogger.error("NostrTransport: failed to embed READ ack", category: .session)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
sendWrappedMessage(content: ack, recipientHex: recipientHex, senderIdentity: senderIdentity)
|
sendWrappedMessage(content: ack, recipientHex: recipientHex, senderIdentity: senderIdentity)
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
private func scheduleNextReadAck() {
|
case .deliveredDirect(let messageID, let peerID):
|
||||||
dependencies.scheduleAfter(readAckInterval) { [weak self] in
|
guard let recipientNpub = resolveRecipientNpub(for: peerID),
|
||||||
self?.queue.async(flags: .barrier) { [weak self] in
|
let recipientHex = npubToHex(recipientNpub),
|
||||||
self?.isSendingReadAcks = false
|
let senderIdentity = try? dependencies.currentIdentity() else { return }
|
||||||
self?.processReadQueueIfNeeded()
|
SecureLogger.debug("NostrTransport: preparing DELIVERED ack id=\(messageID.prefix(8))…", category: .session)
|
||||||
|
guard let ack = NostrEmbeddedBitChat.encodeAckForNostr(type: .delivered, messageID: messageID, recipientPeerID: peerID, senderPeerID: senderPeerID) else {
|
||||||
|
SecureLogger.error("NostrTransport: failed to embed DELIVERED ack", category: .session)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
sendWrappedMessage(content: ack, recipientHex: recipientHex, senderIdentity: senderIdentity)
|
||||||
|
|
||||||
|
case .deliveredGeohash(let messageID, let recipientHex, let identity):
|
||||||
|
SecureLogger.debug("GeoDM: send DELIVERED mid=\(messageID.prefix(8))…", category: .session)
|
||||||
|
guard let embedded = NostrEmbeddedBitChat.encodeAckForNostrNoRecipient(type: .delivered, messageID: messageID, senderPeerID: senderPeerID) else { return }
|
||||||
|
sendWrappedMessage(content: embedded, recipientHex: recipientHex, senderIdentity: identity, registerPending: true)
|
||||||
|
|
||||||
|
case .readGeohash(let messageID, let recipientHex, let identity):
|
||||||
|
SecureLogger.debug("GeoDM: send READ mid=\(messageID.prefix(8))…", category: .session)
|
||||||
|
guard let embedded = NostrEmbeddedBitChat.encodeAckForNostrNoRecipient(type: .readReceipt, messageID: messageID, senderPeerID: senderPeerID) else { return }
|
||||||
|
sendWrappedMessage(content: embedded, recipientHex: recipientHex, senderIdentity: identity, registerPending: true)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -514,7 +514,10 @@ final class ChatPrivateConversationCoordinator {
|
|||||||
category: .session
|
category: .session
|
||||||
)
|
)
|
||||||
} else {
|
} else {
|
||||||
SecureLogger.warning("GeoDM: delivered ack for unknown mid=\(messageID.prefix(8))… conv=\(convKey)", category: .session)
|
// A stale ack for a message this device no longer tracks (dropped
|
||||||
|
// outbox entry, cleared chat, or a peer re-acking after losing our
|
||||||
|
// receipt) — expected occasionally, not actionable.
|
||||||
|
SecureLogger.debug("GeoDM: delivered ack for unknown mid=\(messageID.prefix(8))… conv=\(convKey)", category: .session)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -41,7 +41,11 @@ struct ChatViewModelServiceBundle {
|
|||||||
self.privateChatManager = privateChatManager
|
self.privateChatManager = privateChatManager
|
||||||
self.unifiedPeerService = unifiedPeerService
|
self.unifiedPeerService = unifiedPeerService
|
||||||
self.autocompleteService = AutocompleteService()
|
self.autocompleteService = AutocompleteService()
|
||||||
self.deduplicationService = MessageDeduplicationService()
|
// Persist processed gift-wrap event IDs: NIP-59 randomizes their
|
||||||
|
// timestamps, so the 24h-lookback DM subscriptions redeliver the same
|
||||||
|
// events on every launch and only a cross-launch record stops the
|
||||||
|
// reprocessing (re-sent DELIVERED bursts, phantom-ack noise).
|
||||||
|
self.deduplicationService = MessageDeduplicationService(nostrEventStore: NostrProcessedEventStore())
|
||||||
self.publicMessagePipeline = PublicMessagePipeline()
|
self.publicMessagePipeline = PublicMessagePipeline()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -40,14 +40,17 @@ struct BLEFragmentHandlerTests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
func ownFragmentIsIgnored() {
|
func ownFragmentIsTrackedForSyncButNotAssembled() {
|
||||||
let recorder = Recorder()
|
let recorder = Recorder()
|
||||||
let handler = makeHandler(recorder: recorder)
|
let handler = makeHandler(recorder: recorder)
|
||||||
let packet = makeFragmentPacket(sender: localPeerID, index: 0, total: 2)
|
let packet = makeFragmentPacket(sender: localPeerID, index: 0, total: 2)
|
||||||
|
|
||||||
handler.handle(packet, from: localPeerID)
|
handler.handle(packet, from: localPeerID)
|
||||||
|
|
||||||
#expect(recorder.trackedPackets.isEmpty)
|
// Sync replay hands own fragments back after a relaunch; they must
|
||||||
|
// re-enter the sync store (so the next round's filter covers them
|
||||||
|
// and redelivery stops) without being reassembled.
|
||||||
|
#expect(recorder.trackedPackets.count == 1)
|
||||||
#expect(recorder.appendedHeaders.isEmpty)
|
#expect(recorder.appendedHeaders.isEmpty)
|
||||||
#expect(recorder.reinjectedPackets.isEmpty)
|
#expect(recorder.reinjectedPackets.isEmpty)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user