Harden Nostr envelope migration compatibility

bitchatTests/Services/NostrRelayManagerTests.swift

@@ -267,6 +267,588 @@ final class NostrRelayManagerTests: XCTestCase {
         XCTAssertEqual(context.manager.debugPendingMessageQueueCount, TransportConfig.nostrPendingSendQueueCap)
     }
 
+    func test_privateEnvelopeBatchEvictsEphemeralTrafficAndRemainsAtomic() async throws {
+        let relayURL = "wss://private-batch-priority.example"
+        let context = makeContext(
+            permission: .denied,
+            userTorEnabled: true,
+            torEnforced: true,
+            torIsReady: false
+        )
+        var ephemeralIDs: [String] = []
+        for i in 0..<(TransportConfig.nostrPendingSendQueueCap - 1) {
+            let event = try makeSignedEvent(
+                content: "ephemeral-\(i)",
+                kind: .ephemeralEvent
+            )
+            ephemeralIDs.append(event.id)
+            context.manager.sendEvent(event, to: [relayURL])
+        }
+        let regular = try makeSignedEvent(content: "regular survives")
+        context.manager.sendEvent(regular, to: [relayURL])
+        XCTAssertEqual(
+            context.manager.debugPendingMessageQueueCount,
+            TransportConfig.nostrPendingSendQueueCap
+        )
+
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch([primary, legacy], to: [relayURL]))
+
+        let batches = context.manager.debugPendingMessageQueueEventIDsByBatch
+        let allIDs = Set(batches.flatMap { $0 })
+        XCTAssertEqual(
+            context.manager.debugPendingMessageQueueCount,
+            TransportConfig.nostrPendingSendQueueCap
+        )
+        XCTAssertTrue(batches.contains([primary.id, legacy.id]))
+        XCTAssertTrue(allIDs.contains(regular.id))
+        XCTAssertFalse(allIDs.contains(ephemeralIDs[0]))
+        XCTAssertFalse(allIDs.contains(ephemeralIDs[1]))
+
+        // Later low-priority traffic may evict another ephemeral event, but
+        // never one half (or both halves) of the protected private batch.
+        let lateEphemeral = try makeSignedEvent(content: "late", kind: .geohashPresence)
+        context.manager.sendEvent(lateEphemeral, to: [relayURL])
+        XCTAssertTrue(
+            context.manager.debugPendingMessageQueueEventIDsByBatch
+                .contains([primary.id, legacy.id])
+        )
+    }
+
+    func test_privateEnvelopeBatchDoesNotWriteStaleSocketUntilTorRecovers() async throws {
+        let relayURL = "wss://private-batch-tor-gate.example"
+        let context = makeContext(
+            permission: .denied,
+            userTorEnabled: true,
+            torEnforced: true,
+            torIsReady: true
+        )
+        context.manager.ensureConnections(to: [relayURL])
+        let connected = await waitUntil {
+            context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
+        }
+        XCTAssertTrue(connected)
+        let connection = try XCTUnwrap(context.sessionFactory.latestConnection(for: relayURL))
+
+        context.torWaiter.isReady = false
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch([primary, legacy], to: [relayURL]))
+
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertTrue(connection.sentStrings.isEmpty)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        XCTAssertEqual(context.torWaiter.awaitCallCount, 1)
+
+        context.torWaiter.resolve(true)
+        let flushed = await waitUntil {
+            connection.sentStrings.count == 2
+        }
+        XCTAssertTrue(flushed)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        try await emitAcceptedPair([primary, legacy], on: connection)
+        let acknowledged = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(acknowledged)
+    }
+
+    func test_privateEnvelopeBatchPrunesTerminalRelayAfterHealthyDelivery() async throws {
+        let healthyURL = "wss://private-batch-healthy.example"
+        let deadURL = "wss://private-batch-dead.example"
+        let context = makeContext(permission: .denied)
+        context.sessionFactory.pingErrorByURL[deadURL] = NSError(
+            domain: NSURLErrorDomain,
+            code: NSURLErrorCannotFindHost
+        )
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        var terminalFailureCount = 0
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(
+            [primary, legacy],
+            to: [healthyURL, deadURL],
+            terminalFailure: { terminalFailureCount += 1 }
+        ))
+
+        let healthyConnection = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: healthyURL)
+        )
+        let written = await waitUntil {
+            healthyConnection.sentStrings.count == 2
+        }
+        XCTAssertTrue(written)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        try await emitAcceptedPair([primary, legacy], on: healthyConnection)
+        let initiallyAcknowledged = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(initiallyAcknowledged)
+        XCTAssertEqual(terminalFailureCount, 0)
+
+        // The terminal target is excluded during cooldown. More than one full
+        // protected-capacity worth of subsequent logical sends must continue
+        // to drain through the healthy relay instead of wedging globally.
+        for _ in 0...100 {
+            let sentCountBefore = healthyConnection.sentStrings.count
+            XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(
+                [primary, legacy],
+                to: [healthyURL, deadURL]
+            ))
+            let written = await waitUntil {
+                healthyConnection.sentStrings.count == sentCountBefore + 2
+            }
+            XCTAssertTrue(written)
+            try await emitAcceptedPair([primary, legacy], on: healthyConnection)
+            let acknowledged = await waitUntil {
+                context.manager.debugPendingMessageQueueCount == 0
+            }
+            XCTAssertTrue(acknowledged)
+        }
+    }
+
+    func test_privateEnvelopeBatchAllTerminalTargetsReportsWholeBatchFailure() async throws {
+        let deadURL = "wss://private-batch-all-dead.example"
+        let context = makeContext(permission: .denied)
+        context.sessionFactory.pingErrorByURL[deadURL] = NSError(
+            domain: NSURLErrorDomain,
+            code: NSURLErrorCannotFindHost
+        )
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        var terminalFailureCount = 0
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(
+            [primary, legacy],
+            to: [deadURL],
+            terminalFailure: { terminalFailureCount += 1 }
+        ))
+
+        let failed = await waitUntil {
+            terminalFailureCount == 1 &&
+                context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(failed)
+    }
+
+    func test_privateEnvelopeBatchRejectsBeforeWritingWhenProtectedCapacityIsFull() async throws {
+        let stalledRelayURL = "wss://private-batch-full.example"
+        let connectedRelayURL = "wss://private-batch-connected.example"
+        let context = makeContext(
+            permission: .denied,
+            userTorEnabled: true,
+            torEnforced: true,
+            torIsReady: true,
+            torIsForeground: true
+        )
+        context.manager.ensureConnections(to: [connectedRelayURL])
+        let connected = await waitUntil {
+            context.manager.relays.first(where: { $0.url == connectedRelayURL })?.isConnected == true
+        }
+        XCTAssertTrue(connected)
+        context.torForeground.value = false
+
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        let pair = [primary, legacy]
+
+        for _ in 0..<(TransportConfig.nostrPendingSendQueueCap / pair.count) {
+            XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(pair, to: [stalledRelayURL]))
+        }
+        XCTAssertEqual(
+            context.manager.debugPendingMessageQueueCount,
+            TransportConfig.nostrPendingSendQueueCap
+        )
+
+        let rejectedPrimary = try makeSignedEvent(content: "rejected-primary", kind: .privateEnvelope)
+        let rejectedLegacy = try makeSignedEvent(content: "rejected-legacy", kind: .legacyNIP59GiftWrap)
+        XCTAssertFalse(
+            context.manager.sendPrivateEnvelopeBatch(
+                [rejectedPrimary, rejectedLegacy],
+                to: [connectedRelayURL]
+            )
+        )
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertTrue(
+            context.sessionFactory.latestConnection(for: connectedRelayURL)?.sentStrings.isEmpty == true
+        )
+        let queuedIDs = Set(
+            context.manager.debugPendingMessageQueueEventIDsByBatch.flatMap { $0 }
+        )
+        XCTAssertFalse(queuedIDs.contains(rejectedPrimary.id))
+        XCTAssertFalse(queuedIDs.contains(rejectedLegacy.id))
+    }
+
+    func test_privateEnvelopeBatchStaysQueuedAndRetriesAfterFirstWriteFails() async throws {
+        try await assertPrivateEnvelopeBatchWriteFailure(
+            failureSequence: [NSError(domain: "send", code: 1)],
+            expectedFirstConnectionWrites: 1
+        )
+    }
+
+    func test_privateEnvelopeBatchStaysQueuedAndRetriesAfterSecondWriteFails() async throws {
+        try await assertPrivateEnvelopeBatchWriteFailure(
+            failureSequence: [nil, NSError(domain: "send", code: 2)],
+            expectedFirstConnectionWrites: 2
+        )
+    }
+
+    func test_privateEnvelopeBatchRemainsPendingUntilBothRelayOKsArrive() async throws {
+        let relayURL = "wss://private-batch-two-write-commit.example"
+        let context = makeContext(permission: .denied)
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch([primary, legacy], to: [relayURL]))
+        let connection = try XCTUnwrap(context.sessionFactory.latestConnection(for: relayURL))
+        connection.deferSendCompletions = true
+
+        let firstWriteStarted = await waitUntil { connection.sentStrings.count == 1 }
+        XCTAssertTrue(firstWriteStarted)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+
+        connection.flushDeferredSendCompletions()
+        let secondWriteStarted = await waitUntil { connection.sentStrings.count == 2 }
+        XCTAssertTrue(secondWriteStarted)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+
+        connection.flushDeferredSendCompletions()
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+
+        try connection.emitOK(eventID: primary.id, success: true, reason: "accepted")
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+
+        try connection.emitOK(eventID: legacy.id, success: true, reason: "accepted")
+        let committed = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(committed)
+    }
+
+    func test_privateEnvelopeBatchRejectedHalfReportsWholeBatchFailure() async throws {
+        let relayURL = "wss://private-batch-rejected-half.example"
+        let context = makeContext(permission: .denied)
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        var terminalFailureCount = 0
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(
+            [primary, legacy],
+            to: [relayURL],
+            terminalFailure: { terminalFailureCount += 1 }
+        ))
+        let connection = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        let written = await waitUntil { connection.sentStrings.count == 2 }
+        XCTAssertTrue(written)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+
+        try connection.emitOK(eventID: primary.id, success: true, reason: "accepted")
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        XCTAssertEqual(terminalFailureCount, 0)
+
+        try connection.emitOK(eventID: legacy.id, success: false, reason: "blocked: policy")
+        let failed = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0 &&
+                terminalFailureCount == 1
+        }
+        XCTAssertTrue(failed)
+    }
+
+    func test_privateEnvelopeBatchDuplicateOKsCountAsDurableAcceptance() async throws {
+        let relayURL = "wss://private-batch-duplicate.example"
+        let context = makeContext(permission: .denied)
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        var terminalFailureCount = 0
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(
+            [primary, legacy],
+            to: [relayURL],
+            terminalFailure: { terminalFailureCount += 1 }
+        ))
+        let connection = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        let written = await waitUntil { connection.sentStrings.count == 2 }
+        XCTAssertTrue(written)
+
+        for event in [primary, legacy] {
+            try connection.emitOK(
+                eventID: event.id,
+                success: false,
+                reason: "  DuPlIcAtE: already have this event"
+            )
+            try? await Task.sleep(nanoseconds: 20_000_000)
+        }
+        let acknowledged = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(acknowledged)
+        XCTAssertEqual(terminalFailureCount, 0)
+    }
+
+    func test_privateEnvelopeBatchAcknowledgementTimeoutReportsWholeBatchFailure() async throws {
+        let relayURL = "wss://private-batch-ok-timeout.example"
+        let context = makeContext(permission: .denied)
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        var terminalFailureCount = 0
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(
+            [primary, legacy],
+            to: [relayURL],
+            terminalFailure: { terminalFailureCount += 1 }
+        ))
+        let connection = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        let timeoutScheduled = await waitUntil {
+            connection.sentStrings.count == 2 &&
+                context.scheduler.scheduled.contains {
+                    $0.delay == TransportConfig.nostrConfirmedSendAckTimeoutSeconds
+                }
+        }
+        XCTAssertTrue(timeoutScheduled)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        NostrRelayManager.registerPendingPrivateEnvelope(id: primary.id)
+        NostrRelayManager.registerPendingPrivateEnvelope(id: legacy.id)
+
+        context.scheduler.runNext()
+        let failed = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0 &&
+                terminalFailureCount == 1
+        }
+        XCTAssertTrue(failed)
+
+        try connection.emitOK(eventID: primary.id, success: true, reason: "late")
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(terminalFailureCount, 1)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 0)
+        XCTAssertFalse(NostrRelayManager.pendingPrivateEnvelopeIDs.contains(primary.id))
+        XCTAssertFalse(NostrRelayManager.pendingPrivateEnvelopeIDs.contains(legacy.id))
+    }
+
+    func test_privateEnvelopeBatchFastPrimaryRejectionStillWritesCompatibilityHalf() async throws {
+        let relayURL = "wss://private-batch-early-reject.example"
+        let context = makeContext(permission: .denied)
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        var terminalFailureCount = 0
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(
+            [primary, legacy],
+            to: [relayURL],
+            terminalFailure: { terminalFailureCount += 1 }
+        ))
+        let connection = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        connection.deferSendCompletions = true
+        let firstWriteStarted = await waitUntil { connection.sentStrings.count == 1 }
+        XCTAssertTrue(firstWriteStarted)
+        NostrRelayManager.registerPendingPrivateEnvelope(id: primary.id)
+        NostrRelayManager.registerPendingPrivateEnvelope(id: legacy.id)
+
+        try connection.emitOK(eventID: primary.id, success: false, reason: "invalid: rejected")
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        XCTAssertEqual(terminalFailureCount, 0)
+        XCTAssertEqual(connection.sentStrings.count, 1)
+
+        // Completing the primary write must attempt the legacy 1059 even
+        // though the relay has already rejected the provisional 1402.
+        connection.flushDeferredSendCompletions()
+        let compatibilityWriteStarted = await waitUntil {
+            connection.sentStrings.count == 2
+        }
+        XCTAssertTrue(compatibilityWriteStarted)
+        let compatibilityRequest = try XCTUnwrap(connection.sentStrings.last)
+        let compatibilityJSON = try XCTUnwrap(
+            JSONSerialization.jsonObject(
+                with: Data(compatibilityRequest.utf8)
+            ) as? [Any]
+        )
+        let compatibilityEvent = try XCTUnwrap(
+            compatibilityJSON.dropFirst().first as? [String: Any]
+        )
+        XCTAssertEqual(
+            compatibilityEvent["kind"] as? Int,
+            NostrProtocol.EventKind.legacyNIP59GiftWrap.rawValue
+        )
+        XCTAssertEqual(compatibilityEvent["id"] as? String, legacy.id)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        XCTAssertEqual(terminalFailureCount, 0)
+
+        connection.flushDeferredSendCompletions()
+        let failed = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0 &&
+                terminalFailureCount == 1
+        }
+        XCTAssertTrue(failed)
+        XCTAssertEqual(connection.cancelCallCount, 0)
+        XCTAssertEqual(
+            context.manager.relays.first(where: { $0.url == relayURL })?.isConnected,
+            true
+        )
+        XCTAssertEqual(terminalFailureCount, 1)
+        XCTAssertFalse(NostrRelayManager.pendingPrivateEnvelopeIDs.contains(primary.id))
+        XCTAssertFalse(NostrRelayManager.pendingPrivateEnvelopeIDs.contains(legacy.id))
+    }
+
+    func test_privateEnvelopeBatchFastRejectionThenDisconnectReplaysWholePair() async throws {
+        let relayURL = "wss://private-batch-early-reject-replay.example"
+        let context = makeContext(permission: .denied)
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+        var terminalFailureCount = 0
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch(
+            [primary, legacy],
+            to: [relayURL],
+            terminalFailure: { terminalFailureCount += 1 }
+        ))
+        let firstConnection = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        firstConnection.deferSendCompletions = true
+        let firstWriteStarted = await waitUntil {
+            firstConnection.sentStrings.count == 1
+        }
+        XCTAssertTrue(firstWriteStarted)
+
+        try firstConnection.emitOK(
+            eventID: primary.id,
+            success: false,
+            reason: "invalid: rejected"
+        )
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        XCTAssertEqual(terminalFailureCount, 0)
+
+        // Replacing the connection before the sibling write clears the
+        // attempt-scoped rejection. The durable pair must replay in full.
+        context.manager.disconnect()
+        context.manager.connect()
+        let replacementReady = await waitUntil {
+            let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
+            guard connections.count == 2, let replacement = connections.last else {
+                return false
+            }
+            return replacement.sentStrings.count == 2
+        }
+        XCTAssertTrue(replacementReady)
+        let replacement = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        try await emitAcceptedPair([primary, legacy], on: replacement)
+        let acknowledged = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(acknowledged)
+        XCTAssertEqual(terminalFailureCount, 0)
+
+        // The canceled connection's late completion cannot send its sibling
+        // or mutate the replacement attempt.
+        firstConnection.flushDeferredSendCompletions()
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(firstConnection.sentStrings.count, 1)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 0)
+        XCTAssertEqual(terminalFailureCount, 0)
+    }
+
+    func test_privateEnvelopeBatchDisconnectDuringWriteReplaysOnReplacementConnection() async throws {
+        let relayURL = "wss://private-batch-background-reconnect.example"
+        let context = makeContext(permission: .denied)
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch([primary, legacy], to: [relayURL]))
+        let firstConnection = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        firstConnection.deferSendCompletions = true
+        let firstWriteStarted = await waitUntil { firstConnection.sentStrings.count == 1 }
+        XCTAssertTrue(firstWriteStarted)
+
+        // Backgrounding cancels the socket while its first write callback is
+        // still outstanding. The pair must remain replayable in the queue.
+        context.manager.disconnect()
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        context.manager.connect()
+
+        let replacementReady = await waitUntil {
+            let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
+            guard connections.count == 2, let replacement = connections.last else {
+                return false
+            }
+            return replacement.sentStrings.count == 2
+        }
+        XCTAssertTrue(replacementReady)
+        let replacement = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+        try await emitAcceptedPair([primary, legacy], on: replacement)
+        let acknowledged = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(acknowledged)
+
+        // A late success callback from the canceled socket must neither send
+        // the legacy half on that stale socket nor disturb the completed pair.
+        firstConnection.flushDeferredSendCompletions()
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        XCTAssertEqual(firstConnection.sentStrings.count, 1)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 0)
+    }
+
+    private func assertPrivateEnvelopeBatchWriteFailure(
+        failureSequence: [Error?],
+        expectedFirstConnectionWrites: Int
+    ) async throws {
+        let relayURL = "wss://private-batch-write-failure.example"
+        let context = makeContext(permission: .denied)
+        let primary = try makeSignedEvent(content: "primary", kind: .privateEnvelope)
+        let legacy = try makeSignedEvent(content: "legacy", kind: .legacyNIP59GiftWrap)
+
+        XCTAssertTrue(context.manager.sendPrivateEnvelopeBatch([primary, legacy], to: [relayURL]))
+        let firstConnection = try XCTUnwrap(context.sessionFactory.latestConnection(for: relayURL))
+        firstConnection.sendErrorSequence = failureSequence
+
+        let stayedQueued = await waitUntil {
+            firstConnection.sentStrings.count == expectedFirstConnectionWrites &&
+            context.manager.debugPendingMessageQueueEventIDsByBatch.contains([primary.id, legacy.id])
+        }
+        XCTAssertTrue(stayedQueued)
+        XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 2)
+
+        // The failed socket is disconnected with bounded backoff. Its queue
+        // item is reused—not re-admitted—and the replacement retries both
+        // copies before the pair is finally removed.
+        XCTAssertFalse(context.scheduler.scheduled.isEmpty)
+        context.scheduler.runNext()
+        let replacementReady = await waitUntil {
+            guard let replacement = context.sessionFactory.latestConnection(for: relayURL),
+                  replacement !== firstConnection else { return false }
+            return replacement.sentStrings.count == 2
+        }
+        XCTAssertTrue(replacementReady)
+        let replacement = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)
+        )
+        try await emitAcceptedPair([primary, legacy], on: replacement)
+        let acknowledged = await waitUntil {
+            context.manager.debugPendingMessageQueueCount == 0
+        }
+        XCTAssertTrue(acknowledged)
+    }
+
     func test_sendEvent_waitsForTorReadinessBeforeSending() async throws {
         let relayURL = "wss://tor-ready.example"
         let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
@@ -548,6 +1130,52 @@ final class NostrRelayManagerTests: XCTestCase {
         XCTAssertEqual(context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count, 1)
     }
 
+    func test_subscribe_multiplePrivateEnvelopeFiltersEncodesIndependentLimits() async throws {
+        let relayURL = "wss://private-recovery-filters.example"
+        let context = makeContext(permission: .denied)
+        let filters = NostrFilter.privateEnvelopeFiltersFor(
+            pubkey: "recipient",
+            since: Date(timeIntervalSince1970: 1_234_567)
+        )
+
+        context.manager.subscribe(
+            filters: filters,
+            id: "private-recovery",
+            relayUrls: [relayURL],
+            handler: { _ in }
+        )
+
+        let sent = await waitUntil {
+            context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
+        }
+        XCTAssertTrue(sent)
+        let request = try XCTUnwrap(
+            context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.first
+        )
+        let data = try XCTUnwrap(request.data(using: .utf8))
+        let array = try XCTUnwrap(
+            try JSONSerialization.jsonObject(with: data) as? [Any]
+        )
+        XCTAssertEqual(array.count, 4)
+        XCTAssertEqual(array[0] as? String, "REQ")
+        XCTAssertEqual(array[1] as? String, "private-recovery")
+        let encodedFilters = try [array[2], array[3]].map {
+            try XCTUnwrap($0 as? [String: Any])
+        }
+        XCTAssertEqual(encodedFilters.compactMap { $0["kinds"] as? [Int] }, [
+            [NostrProtocol.EventKind.privateEnvelope.rawValue],
+            [NostrProtocol.EventKind.legacyNIP59GiftWrap.rawValue]
+        ])
+        for filter in encodedFilters {
+            XCTAssertEqual(
+                filter["limit"] as? Int,
+                TransportConfig.nostrPrivateEnvelopeFetchLimitPerKind
+            )
+            XCTAssertEqual(filter["#p"] as? [String], ["recipient"])
+            XCTAssertEqual(filter["since"] as? Int, 1_234_567)
+        }
+    }
+
     func test_subscribe_coalescesDuplicateRequestsBeforeTorReadyAndDefersEOSE() async throws {
         let relayURL = "wss://tor-subscribe-coalesce.example"
         let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
@@ -1646,12 +2274,31 @@ final class NostrRelayManagerTests: XCTestCase {
         return filter
     }
 
-    private func makeSignedEvent(content: String) throws -> NostrEvent {
+    private func emitAcceptedPair(
+        _ events: [NostrEvent],
+        on connection: MockRelayConnection
+    ) async throws {
+        for event in events {
+            try connection.emitOK(
+                eventID: event.id,
+                success: true,
+                reason: "accepted"
+            )
+            // Each inbound frame consumes the mock receive handler. Let the
+            // manager re-arm it before delivering the next relay response.
+            try? await Task.sleep(nanoseconds: 20_000_000)
+        }
+    }
+
+    private func makeSignedEvent(
+        content: String,
+        kind: NostrProtocol.EventKind = .textNote
+    ) throws -> NostrEvent {
         let identity = try NostrIdentity.generate()
         let event = NostrEvent(
             pubkey: identity.publicKeyHex,
             createdAt: Date(),
-            kind: .textNote,
+            kind: kind,
             tags: [],
             content: content
         )
@@ -1791,6 +2438,7 @@ private final class MockRelaySessionFactory: NostrRelaySessionProtocol {
 private final class MockRelayConnection: NostrRelayConnectionProtocol {
     private let pingError: Error?
     var sendError: Error?
+    var sendErrorSequence: [Error?] = []
     private var receiveHandler: ((Result<URLSessionWebSocketTask.Message, Error>) -> Void)?
     private(set) var resumeCallCount = 0
     private(set) var cancelCallCount = 0
@@ -1820,14 +2468,15 @@ private final class MockRelayConnection: NostrRelayConnectionProtocol {
     }
 
     var deferSendCompletions = false
-    private var deferredSendCompletions: [(Error?) -> Void] = []
+    private var deferredSendCompletions: [(error: Error?, completion: (Error?) -> Void)] = []
 
     func send(_ message: URLSessionWebSocketTask.Message, completionHandler: @escaping (Error?) -> Void) {
         sentMessages.append(message)
+        let error = sendErrorSequence.isEmpty ? sendError : sendErrorSequence.removeFirst()
         if deferSendCompletions {
-            deferredSendCompletions.append(completionHandler)
+            deferredSendCompletions.append((error, completionHandler))
         } else {
-            completionHandler(sendError)
+            completionHandler(error)
         }
     }
 
@@ -1835,7 +2484,7 @@ private final class MockRelayConnection: NostrRelayConnectionProtocol {
         let pending = deferredSendCompletions
         deferredSendCompletions = []
         pending.forEach {
-            $0(sendError)
+            $0.completion($0.error)
         }
     }
 

bitchatTests/Services/NostrTransportTests.swift

@@ -131,7 +131,7 @@ struct NostrTransportTests {
         #expect(undeliverable)
     }
 
-    @Test("Private message resolves short peer ID and emits decryptable packet")
+    @Test("Private message resolves short peer ID and emits both migration formats")
     @MainActor
     func sendPrivateMessageResolvesShortPeerID() async throws {
         let keychain = MockKeychain()
@@ -154,7 +154,7 @@ struct NostrTransportTests {
                 favoriteStatusForPeerID: { $0 == shortPeerID ? relationship : nil },
                 currentIdentity: { sender },
                 registerPendingPrivateEnvelope: probe.recordPendingPrivateEnvelope(id:),
-                sendEvent: probe.record(event:),
+                sendPrivateEnvelopeBatch: { events, _ in probe.record(batch: events) },
                 scheduleAfter: { delay, action in
                     probe.enqueueScheduledAction(delay: delay, action: action)
                 }
@@ -164,8 +164,12 @@ struct NostrTransportTests {
 
         transport.sendPrivateMessage("hello over nostr", to: shortPeerID, recipientNickname: "Carol", messageID: "pm-1")
 
-        let didSend = await TestHelpers.waitUntil({ probe.sentEvents.count == 1 }, timeout: 5.0)
+        let didSend = await TestHelpers.waitUntil({ probe.sentEvents.count == 2 }, timeout: 5.0)
         #expect(didSend)
+        #expect(probe.sentEvents.map(\.kind) == [
+            NostrProtocol.EventKind.privateEnvelope.rawValue,
+            NostrProtocol.EventKind.legacyNIP59GiftWrap.rawValue
+        ])
         let result = try decodeEmbeddedPayload(from: probe.sentEvents[0], recipient: recipient)
         let privateMessage = try decodePrivateMessage(from: result.payload)
 
@@ -176,9 +180,9 @@ struct NostrTransportTests {
         #expect(probe.pendingPrivateEnvelopeIDs.isEmpty)
     }
 
-    @Test("Migration window publishes primary and compatibility envelopes, then stops")
+    @Test("Coordinated migration always publishes primary and compatibility envelopes")
     @MainActor
-    func migrationWindowDualPublishesUntilDeadline() async throws {
+    func migrationAlwaysDualPublishes() async throws {
         let keychain = MockKeychain()
         let idBridge = NostrIdentityBridge(keychain: keychain)
         let sender = try NostrIdentity.generate()
@@ -198,11 +202,7 @@ struct NostrTransportTests {
             dependencies: makeDependencies(
                 favoriteStatusForNoiseKey: { $0 == noiseKey ? relationship : nil },
                 currentIdentity: { sender },
-                sendEvent: migrationProbe.record(event:),
-                now: {
-                    NostrProtocol.legacyPrivateEnvelopePublicationDeadline
-                        .addingTimeInterval(-1)
-                }
+                sendPrivateEnvelopeBatch: { events, _ in migrationProbe.record(batch: events) }
             )
         )
         migrationTransport.senderPeerID = PeerID(str: "0123456789abcdef")
@@ -228,32 +228,281 @@ struct NostrTransportTests {
             #expect(message.messageID == "migration-pm")
             #expect(message.content == "migration payload")
         }
+    }
 
-        let postMigrationProbe = NostrTransportProbe()
-        let postMigrationTransport = NostrTransport(
+    @Test("Rejected migration batch does not register half-delivery state")
+    @MainActor
+    func rejectedMigrationBatchRegistersNothing() async throws {
+        let keychain = MockKeychain()
+        let idBridge = NostrIdentityBridge(keychain: keychain)
+        let sender = try NostrIdentity.generate()
+        let recipient = try NostrIdentity.generate()
+        let probe = NostrTransportProbe()
+        var rejectedKinds: [Int] = []
+        let transport = NostrTransport(
+            keychain: keychain,
+            idBridge: idBridge,
+            dependencies: makeDependencies(
+                currentIdentity: { sender },
+                registerPendingPrivateEnvelope: probe.recordPendingPrivateEnvelope(id:),
+                sendPrivateEnvelopeBatch: { events, _ in
+                    rejectedKinds = events.map(\.kind)
+                    return false
+                }
+            )
+        )
+        transport.senderPeerID = PeerID(str: "0123456789abcdef")
+
+        transport.sendPrivateMessageGeohash(
+            content: "must stay atomic",
+            toRecipientHex: recipient.publicKeyHex,
+            from: sender,
+            messageID: "atomic-reject"
+        )
+
+        let attempted = await TestHelpers.waitUntil({ rejectedKinds.count == 2 }, timeout: 5.0)
+        #expect(attempted)
+        #expect(rejectedKinds == [
+            NostrProtocol.EventKind.privateEnvelope.rawValue,
+            NostrProtocol.EventKind.legacyNIP59GiftWrap.rawValue
+        ])
+        #expect(probe.pendingPrivateEnvelopeIDs.isEmpty)
+    }
+
+    @Test("Rejected user message emits a visible failed-delivery event")
+    @MainActor
+    func rejectedUserMessageEmitsFailureEvent() async throws {
+        let keychain = MockKeychain()
+        let idBridge = NostrIdentityBridge(keychain: keychain)
+        let sender = try NostrIdentity.generate()
+        let recipient = try NostrIdentity.generate()
+        let eventProbe = NostrTransportEventProbe()
+        let transport = NostrTransport(
+            keychain: keychain,
+            idBridge: idBridge,
+            dependencies: makeDependencies(
+                currentIdentity: { sender },
+                sendPrivateEnvelopeBatch: { _, _ in false }
+            )
+        )
+        transport.senderPeerID = PeerID(str: "0123456789abcdef")
+        transport.eventDelegate = eventProbe
+
+        transport.sendPrivateMessageGeohash(
+            content: "must fail visibly",
+            toRecipientHex: recipient.publicKeyHex,
+            from: sender,
+            messageID: "visible-reject"
+        )
+
+        let reported = await TestHelpers.waitUntil(
+            { eventProbe.failedMessageIDs == ["visible-reject"] },
+            timeout: 5.0
+        )
+        #expect(reported)
+    }
+
+    @Test("Rejected favorite notification retains and retries the exact pair")
+    @MainActor
+    func rejectedFavoriteNotificationRetriesExactPair() async throws {
+        let keychain = MockKeychain()
+        let idBridge = NostrIdentityBridge(keychain: keychain)
+        let sender = try NostrIdentity.generate()
+        let recipient = try NostrIdentity.generate()
+        let noiseKey = Data((96..<128).map(UInt8.init))
+        let fullPeerID = PeerID(hexData: noiseKey)
+        let relationship = makeRelationship(
+            peerNoisePublicKey: noiseKey,
+            peerNostrPublicKey: recipient.npub,
+            peerNickname: "Retry favorite"
+        )
+        let probe = NostrTransportProbe()
+        var attempts: [[String]] = []
+        weak var releasedTransport: NostrTransport?
+        do {
+            let transport = NostrTransport(
+                keychain: keychain,
+                idBridge: idBridge,
+                dependencies: makeDependencies(
+                    favoriteStatusForNoiseKey: { $0 == noiseKey ? relationship : nil },
+                    currentIdentity: { sender },
+                    sendPrivateEnvelopeBatch: { events, _ in
+                        attempts.append(events.map(\.id))
+                        return attempts.count > 1
+                    },
+                    scheduleAfter: { delay, action in
+                        probe.enqueueScheduledAction(delay: delay, action: action)
+                    }
+                )
+            )
+            transport.senderPeerID = PeerID(str: "0123456789abcdef")
+            releasedTransport = transport
+
+            transport.sendFavoriteNotification(to: fullPeerID, isFavorite: true)
+            let retryScheduled = await TestHelpers.waitUntil(
+                { attempts.count == 1 && probe.scheduledActionCount == 1 },
+                timeout: 5.0
+            )
+            #expect(retryScheduled)
+        }
+        #expect(releasedTransport == nil)
+        #expect(probe.runNextScheduledAction())
+
+        let retried = await TestHelpers.waitUntil({ attempts.count == 2 }, timeout: 5.0)
+        #expect(retried)
+        #expect(attempts[0] == attempts[1])
+    }
+
+    @Test("Rejected delivery acknowledgement remains queued for retry")
+    @MainActor
+    func rejectedDeliveryAckRetries() async throws {
+        let keychain = MockKeychain()
+        let idBridge = NostrIdentityBridge(keychain: keychain)
+        let sender = try NostrIdentity.generate()
+        let recipient = try NostrIdentity.generate()
+        let noiseKey = Data((128..<160).map(UInt8.init))
+        let fullPeerID = PeerID(hexData: noiseKey)
+        let relationship = makeRelationship(
+            peerNoisePublicKey: noiseKey,
+            peerNostrPublicKey: recipient.npub,
+            peerNickname: "Retry ack"
+        )
+        let probe = NostrTransportProbe()
+        var attempts: [[String]] = []
+        let transport = NostrTransport(
             keychain: keychain,
             idBridge: idBridge,
             dependencies: makeDependencies(
                 favoriteStatusForNoiseKey: { $0 == noiseKey ? relationship : nil },
                 currentIdentity: { sender },
-                sendEvent: postMigrationProbe.record(event:),
-                now: { NostrProtocol.legacyPrivateEnvelopePublicationDeadline }
+                sendPrivateEnvelopeBatch: { events, _ in
+                    attempts.append(events.map(\.id))
+                    return attempts.count > 1
+                },
+                scheduleAfter: { delay, action in
+                    probe.enqueueScheduledAction(delay: delay, action: action)
+                }
             )
         )
-        postMigrationTransport.senderPeerID = PeerID(str: "0123456789abcdef")
-        postMigrationTransport.sendPrivateMessage(
-            "post migration",
-            to: peerID,
-            recipientNickname: "Migration peer",
-            messageID: "post-migration-pm"
-        )
+        transport.senderPeerID = PeerID(str: "0123456789abcdef")
 
-        let sentPrimaryOnly = await TestHelpers.waitUntil(
-            { postMigrationProbe.sentEvents.count == 1 },
+        transport.sendDeliveryAck(for: "retry-ack", to: fullPeerID)
+        let firstAttempt = await TestHelpers.waitUntil(
+            { attempts.count == 1 && probe.scheduledActionCount >= 1 },
             timeout: 5.0
         )
-        #expect(sentPrimaryOnly)
-        #expect(postMigrationProbe.sentEvents.first?.kind == NostrProtocol.EventKind.privateEnvelope.rawValue)
+        #expect(firstAttempt)
+
+        for _ in 0..<3 where attempts.count < 2 {
+            _ = probe.runNextScheduledAction()
+            _ = await TestHelpers.waitUntil(
+                { attempts.count == 2 || probe.scheduledActionCount > 0 },
+                timeout: 1.0
+            )
+        }
+        #expect(attempts.count == 2)
+        #expect(attempts[0] == attempts[1])
+    }
+
+    @Test("Control retry queue is bounded and evicted callbacks are harmless")
+    @MainActor
+    func controlRetryQueueIsBounded() async throws {
+        let sender = try NostrIdentity.generate()
+        let recipient = try NostrIdentity.generate()
+        let events = try NostrProtocol.createPrivateEnvelopePublicationBatch(
+            content: "bounded retry fixture",
+            recipientPubkey: recipient.publicKeyHex,
+            senderIdentity: sender
+        )
+        let probe = NostrTransportProbe()
+        var retryAttempts = 0
+        let queue = NostrPrivateEnvelopeRetryQueue(
+            sendPrivateEnvelopeBatch: { _, _ in
+                retryAttempts += 1
+                return false
+            },
+            registerPendingPrivateEnvelope: { _ in },
+            scheduleAfter: { delay, action in
+                probe.enqueueScheduledAction(delay: delay, action: action)
+            }
+        )
+
+        for index in 0...TransportConfig.nostrPrivateEnvelopeRetryQueueCap {
+            queue.enqueue(
+                key: "control-\(index)",
+                events: events,
+                registerPending: false
+            )
+        }
+
+        #expect(queue.debugPendingCount == TransportConfig.nostrPrivateEnvelopeRetryQueueCap)
+        #expect(!queue.debugContains(key: "control-0"))
+        #expect(queue.debugContains(key: "control-1"))
+
+        // The oldest callback was scheduled before eviction. Running it must
+        // observe the missing key and return without touching dependencies.
+        #expect(probe.runNextScheduledAction())
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        #expect(retryAttempts == 0)
+
+        queue.removeAll()
+        #expect(queue.debugPendingCount == 0)
+        #expect(probe.runNextScheduledAction())
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        #expect(retryAttempts == 0)
+    }
+
+    @Test("Multiple transports share one globally bounded control retry owner")
+    @MainActor
+    func multipleTransportsShareControlRetryQueue() async throws {
+        let keychain = MockKeychain()
+        let idBridge = NostrIdentityBridge(keychain: keychain)
+        let sender = try NostrIdentity.generate()
+        let recipient = try NostrIdentity.generate()
+        let events = try NostrProtocol.createPrivateEnvelopePublicationBatch(
+            content: "shared retry fixture",
+            recipientPubkey: recipient.publicKeyHex,
+            senderIdentity: sender
+        )
+        let probe = NostrTransportProbe()
+        var retryAttempts = 0
+        let sharedQueue = NostrPrivateEnvelopeRetryQueue(
+            sendPrivateEnvelopeBatch: { _, _ in
+                retryAttempts += 1
+                return false
+            },
+            registerPendingPrivateEnvelope: { _ in },
+            scheduleAfter: { delay, action in
+                probe.enqueueScheduledAction(delay: delay, action: action)
+            }
+        )
+        let first = NostrTransport(
+            keychain: keychain,
+            idBridge: idBridge,
+            dependencies: makeDependencies(envelopeRetryQueue: sharedQueue)
+        )
+        let second = NostrTransport(
+            keychain: keychain,
+            idBridge: idBridge,
+            dependencies: makeDependencies(envelopeRetryQueue: sharedQueue)
+        )
+
+        first.debugEnqueueControlRetry(key: "shared", events: events)
+        second.debugEnqueueControlRetry(key: "shared", events: events)
+        #expect(first.debugControlRetryCount == 1)
+        #expect(second.debugControlRetryCount == 1)
+
+        for index in 0...TransportConfig.nostrPrivateEnvelopeRetryQueueCap {
+            let transport = index.isMultiple(of: 2) ? first : second
+            transport.debugEnqueueControlRetry(key: "global-\(index)", events: events)
+        }
+        #expect(first.debugControlRetryCount == TransportConfig.nostrPrivateEnvelopeRetryQueueCap)
+        #expect(second.debugControlRetryCount == TransportConfig.nostrPrivateEnvelopeRetryQueueCap)
+
+        // The first scheduled callback belongs to the now-evicted shared key.
+        #expect(probe.runNextScheduledAction())
+        try? await Task.sleep(nanoseconds: 20_000_000)
+        #expect(retryAttempts == 0)
     }
 
     @Test("Favorite notification embeds current npub")
@@ -279,7 +528,7 @@ struct NostrTransportTests {
                 favoriteStatusForPeerID: { _ in nil },
                 currentIdentity: { sender },
                 registerPendingPrivateEnvelope: probe.recordPendingPrivateEnvelope(id:),
-                sendEvent: probe.record(event:),
+                sendPrivateEnvelopeBatch: { events, _ in probe.record(batch: events) },
                 scheduleAfter: { delay, action in
                     probe.enqueueScheduledAction(delay: delay, action: action)
                 }
@@ -289,7 +538,7 @@ struct NostrTransportTests {
 
         transport.sendFavoriteNotification(to: fullPeerID, isFavorite: true)
 
-        let didSend = await TestHelpers.waitUntil({ probe.sentEvents.count == 1 }, timeout: 5.0)
+        let didSend = await TestHelpers.waitUntil({ probe.sentEvents.count == 2 }, timeout: 5.0)
         #expect(didSend)
         let result = try decodeEmbeddedPayload(from: probe.sentEvents[0], recipient: recipient)
         let privateMessage = try decodePrivateMessage(from: result.payload)
@@ -320,7 +569,7 @@ struct NostrTransportTests {
                 favoriteStatusForPeerID: { _ in nil },
                 currentIdentity: { sender },
                 registerPendingPrivateEnvelope: probe.recordPendingPrivateEnvelope(id:),
-                sendEvent: probe.record(event:),
+                sendPrivateEnvelopeBatch: { events, _ in probe.record(batch: events) },
                 scheduleAfter: { delay, action in
                     probe.enqueueScheduledAction(delay: delay, action: action)
                 }
@@ -330,7 +579,7 @@ struct NostrTransportTests {
 
         transport.sendDeliveryAck(for: "ack-1", to: fullPeerID)
 
-        let didSend = await TestHelpers.waitUntil({ probe.sentEvents.count == 1 }, timeout: 5.0)
+        let didSend = await TestHelpers.waitUntil({ probe.sentEvents.count == 2 }, timeout: 5.0)
         #expect(didSend)
         let result = try decodeEmbeddedPayload(from: probe.sentEvents[0], recipient: recipient)
 
@@ -353,7 +602,7 @@ struct NostrTransportTests {
             dependencies: makeDependencies(
                 currentIdentity: { sender },
                 registerPendingPrivateEnvelope: probe.recordPendingPrivateEnvelope(id:),
-                sendEvent: probe.record(event:),
+                sendPrivateEnvelopeBatch: { events, _ in probe.record(batch: events) },
                 scheduleAfter: { delay, action in
                     probe.enqueueScheduledAction(delay: delay, action: action)
                 }
@@ -368,7 +617,7 @@ struct NostrTransportTests {
             messageID: "geo-1"
         )
 
-        let didSend = await TestHelpers.waitUntil({ probe.sentEvents.count == 1 }, timeout: 5.0)
+        let didSend = await TestHelpers.waitUntil({ probe.sentEvents.count == 2 }, timeout: 5.0)
         #expect(didSend)
         let event = probe.sentEvents[0]
         let result = try decodeEmbeddedPayload(from: event, recipient: recipient)
@@ -377,7 +626,7 @@ struct NostrTransportTests {
         #expect(privateMessage.messageID == "geo-1")
         #expect(privateMessage.content == "geo hello")
         #expect(result.packet.recipientID == nil)
-        #expect(probe.pendingPrivateEnvelopeIDs == [event.id])
+        #expect(probe.pendingPrivateEnvelopeIDs == probe.sentEvents.map(\.id))
     }
 
     @Test("Read receipt queue sends in order and waits for scheduler")
@@ -403,7 +652,7 @@ struct NostrTransportTests {
                 favoriteStatusForPeerID: { _ in nil },
                 currentIdentity: { sender },
                 registerPendingPrivateEnvelope: probe.recordPendingPrivateEnvelope(id:),
-                sendEvent: probe.record(event:),
+                sendPrivateEnvelopeBatch: { events, _ in probe.record(batch: events) },
                 scheduleAfter: { delay, action in
                     probe.enqueueScheduledAction(delay: delay, action: action)
                 }
@@ -418,20 +667,20 @@ struct NostrTransportTests {
         transport.sendReadReceipt(second, to: fullPeerID)
 
         let readReceiptTimeout: TimeInterval = 5.0
-        let sentFirst = await TestHelpers.waitUntil({ probe.sentEvents.count >= 1 }, timeout: readReceiptTimeout)
-        try #require(sentFirst, "Expected first queued read receipt event")
+        let sentFirst = await TestHelpers.waitUntil({ probe.sentEvents.count == 2 }, timeout: readReceiptTimeout)
+        try #require(sentFirst, "Expected first queued read receipt pair")
         let scheduledThrottle = await TestHelpers.waitUntil({ probe.scheduledActionCount == 1 }, timeout: readReceiptTimeout)
         try #require(scheduledThrottle, "Expected queued throttle action after first read receipt")
-        let firstEvent = try #require(probe.sentEvents.first, "Expected first queued read receipt event")
+        let firstEvent = try #require(probe.sentEvents.first, "Expected first queued read receipt pair")
         let firstPayload = try decodeEmbeddedPayload(from: firstEvent, recipient: recipient).payload
         #expect(firstPayload.type == .readReceipt)
         #expect(String(data: firstPayload.data, encoding: .utf8) == "read-1")
 
         try #require(probe.runNextScheduledAction(), "Expected queued throttle action after first read receipt")
 
-        let sentSecond = await TestHelpers.waitUntil({ probe.sentEvents.count >= 2 }, timeout: readReceiptTimeout)
-        try #require(sentSecond, "Expected second read receipt after running throttle action")
-        let secondEvent = try #require(probe.sentEvents.last, "Expected second queued read receipt event")
+        let sentSecond = await TestHelpers.waitUntil({ probe.sentEvents.count == 4 }, timeout: readReceiptTimeout)
+        try #require(sentSecond, "Expected second read receipt pair after running throttle action")
+        let secondEvent = probe.sentEvents[2]
         let secondPayload = try decodeEmbeddedPayload(from: secondEvent, recipient: recipient).payload
         #expect(secondPayload.type == .readReceipt)
         #expect(String(data: secondPayload.data, encoding: .utf8) == "read-2")
@@ -500,12 +749,13 @@ struct NostrTransportTests {
         favoriteStatusForPeerID: @escaping @MainActor (PeerID) -> FavoriteRelationship? = { _ in nil },
         currentIdentity: @escaping @MainActor () throws -> NostrIdentity? = { nil },
         registerPendingPrivateEnvelope: @escaping @MainActor (String) -> Void = { _ in },
-        sendEvent: @escaping @MainActor (NostrEvent) -> Void = { _ in },
+        sendPrivateEnvelopeBatch: @escaping @MainActor (
+            [NostrEvent],
+            @escaping @MainActor () -> Void
+        ) -> Bool = { _, _ in true },
         scheduleAfter: @escaping @Sendable (TimeInterval, @escaping @Sendable () -> Void) -> Void = { _, _ in },
         relayConnectivity: @escaping @MainActor () -> AnyPublisher<Bool, Never> = { Just(false).eraseToAnyPublisher() },
-        now: @escaping @MainActor () -> Date = {
-            NostrProtocol.legacyPrivateEnvelopePublicationDeadline.addingTimeInterval(1)
-        }
+        envelopeRetryQueue: NostrPrivateEnvelopeRetryQueue? = nil
     ) -> NostrTransport.Dependencies {
         NostrTransport.Dependencies(
             notificationCenter: notificationCenter,
@@ -514,10 +764,10 @@ struct NostrTransportTests {
             favoriteStatusForPeerID: favoriteStatusForPeerID,
             currentIdentity: currentIdentity,
             registerPendingPrivateEnvelope: registerPendingPrivateEnvelope,
-            sendEvent: sendEvent,
+            sendPrivateEnvelopeBatch: sendPrivateEnvelopeBatch,
             scheduleAfter: scheduleAfter,
             relayConnectivity: relayConnectivity,
-            now: now
+            envelopeRetryQueue: envelopeRetryQueue
         )
     }
 
@@ -572,6 +822,17 @@ private enum NostrTransportTestError: Error {
     case invalidPrivateMessage
 }
 
+@MainActor
+private final class NostrTransportEventProbe: TransportEventDelegate {
+    private(set) var failedMessageIDs: [String] = []
+
+    func didReceiveTransportEvent(_ event: TransportEvent) {
+        guard case .messageDeliveryStatusUpdated(let messageID, let status) = event,
+              case .failed = status else { return }
+        failedMessageIDs.append(messageID)
+    }
+}
+
 private func base64URLDecode(_ string: String) -> Data? {
     var candidate = string
     let padding = (4 - (candidate.count % 4)) % 4
@@ -608,10 +869,11 @@ private final class NostrTransportProbe: @unchecked Sendable {
         return scheduledActionsStorage.count
     }
 
-    func record(event: NostrEvent) {
+    func record(batch: [NostrEvent]) -> Bool {
         lock.lock()
-        sentEventsStorage.append(event)
+        sentEventsStorage.append(contentsOf: batch)
         lock.unlock()
+        return true
     }
 
     func recordPendingPrivateEnvelope(id: String) {