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)
         }
     }