permissionlesstech/bitchat
1
0
Fork 0
mirror of https://github.com/permissionlesstech/bitchat.git synced 2026-07-27 10:45:20 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
0152196ac2648ef3f6a1bbab4f24f1a88e11b3b2
bitchat/bitchatTests/Services/NostrRelayManagerTests.swift
T
jackGitHubjackClaude Opus 5
 0152196ac2 Deflake CI, and make the flake class unrepeatable (#1491) 
* Deflake two iOS-sim CI tests with unbounded timing assumptions

Both failed on main-adjacent CI runs during this session's PRs. Neither
was a product bug; both asserted things about real time that a loaded
runner is under no obligation to honour.

**NetworkReachabilityGateTests: a wall-clock upper bound.**
test_monitor_duplicateUpdatesDoNotPostponeOfflineCommit slept 500ms for
real, then asserted total elapsed time was under 1.4s to prove a duplicate
mid-window had not restarted the 1.0s debounce. One CI run took 3.75s. No
wall-clock bound can separate "deadline preserved" from "runner is slow",
because Task.sleep and the asyncAfter flush are both real time and neither
is bounded above.

The deadline property was already covered deterministically one level
down: test_debounce_duplicateObservationsPreservePendingDeadline drives
ReachabilityDebounce with injected timestamps and checks pendingRemaining
directly. So the monitor test now asserts only what needs a real monitor —
that a duplicate still yields exactly one committed false through the
debounce — with an injected clock for the arithmetic and a generous
liveness budget. Renamed to say what it actually checks. No coverage lost,
and it runs in 0.14s instead of ~3.8s because the real sleep is gone.

**NoiseEncryptionServiceTests: injected timeouts that also arm during
setup.** #1483 diagnosed and fixed exactly this in the quarantine-restore
test, but two sibling tests kept the shape. Their injected
ordinaryResponderHandshakeTimeout (0.04 and 0.06) also arms during the
establishSessions setup handshake, where bob is the responder — so a
preempted runner fires it mid-setup, tears down the half-open responder,
and message 3 gets answered as a fresh initiation. The CI failure named
setup's own `#expect(finalMessage == nil)` seeing a 96-byte message 2,
which is precisely the signature #1483 recorded.

Raised both to 1.0s, matching #1483's remedy: the scenarios still need the
responder timeout to fire, and it still does, just with room for setup to
complete first. Thin 1-second waitUntil budgets in the file now use the
shared TestConstants.longTimeout; every one of them backs a positive
assertion, so waitUntil still returns the moment the condition holds and
nothing gets slower in the passing case.

No assertion weakened in either file. Verified 3x sequentially and 3x with
all 18 cores saturated.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* Raise two more starvation-prone test deadlines

Both surfaced on the CI runs for this PR and #1487, both in tests neither
PR touches, both the same shape as the two already fixed here: a deadline
sized for the work rather than for a runner executing many suites at once.

VoiceNotePlaybackControllerTests waited 5s for a @MainActor Task that
playback schedules for the session acquire and its failure path. When that
Task is not scheduled in time the helper reports *two* failures — the wait,
and the `!isPlaying` the un-run failure path has not reset yet — which
reads like a playback bug rather than a starved scheduler. That is exactly
what CI showed.

GeoRelayDirectoryTests waited 10s for work the directory runs in
Task.detached(priority: .utility); utility priority competes with every
other suite. The retry-scheduling case timed out at exactly 10.06s with the
retry never scheduled, reading like a missing retry rather than a starved
background task.

Raised to 30s each with the reasoning recorded at the helper. Both helpers
return as soon as their condition holds, so nothing slows down when tests
pass — only the genuine-failure case takes longer to report.

Verified 3x with all cores saturated: both suites clean.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* Make the flake class unrepeatable, not just fixed

Raising four deadlines fixed the four tests that happened to fire. The
class was still there: fourteen separate waitUntil helpers, most defaulting
to 1.0s, plus wait call sites with literal budgets. The fifth instance
would have landed the same way, on someone else's unrelated PR.

The rule, now written down in TestConstants.settleTimeout: **a wait
deadline is not a latency budget.** It exists so a genuine hang eventually
fails the suite, so size it for the worst-case scheduler, never for how
long the operation should take. Waits return as soon as their condition
holds, so a generous deadline is free in the passing case and only extends
genuine failures.

- Every wait helper now defaults to TestConstants.settleTimeout (30s), and
  the literal wait call sites below the floor were converted too — sixteen
  sites across ten files.
- TestTimingHygieneTests enforces it by scanning the test sources: wait
  defaults and wait call sites must be at least minimumSettleTimeout, and
  no test may assert an upper bound on elapsed wall-clock time (the
  assertion that started this, which cannot separate correct behaviour from
  a slow machine).
- Both rules waive per line with "test-timing-ok: <reason>", accepted on
  the line or in the comment block above it so the reason has room to be a
  sentence. One legitimate use so far: a NEGATIVE wait in NoiseCoverageTests
  asserting a promotion has *not* completed within 50ms, where a long
  deadline would only make the suite slow while still passing.

Injected production timeouts are deliberately not matched — the Noise
handshake timeouts are the behaviour under test, and short values are
correct there.

Verified the guard actually fails: a canary file with both banned shapes
was flagged with file and line, and the suite went green again once it was
removed. Full suite passes with all 18 cores saturated.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* Close the guard's blind spot: named short timeouts

A fifth flake landed on CI while the first guard was in review —
GossipSyncBoardTests timing out at 1.03s — and the guard did not catch it,
because the deadline was `TestConstants.shortTimeout` rather than a
literal. A literals-only scan cannot see a short value behind a symbol,
which is the more common way it is written: 81 wait sites used
shortTimeout (1s) or defaultTimeout (5s), every one of them below the
floor.

Rather than guess which of those were safe to raise, all 81 were converted
and the suite timed. Runtime went 15s -> 72s, which located the genuine
negative waits precisely: ten tests that assert something does *not*
happen and therefore always run their deadline out. Measurement instead of
a heuristic, since a mis-guess in either direction is invisible — too
short reintroduces the flake, too long silently costs a minute a run.

Those 28 sites now use `TestConstants.negativeWaitWindow`, a named
constant whose doc explains the inverted reasoning: for a negative wait,
starvation can only make the assertion *more* likely to hold, so short is
correct, and the name states the polarity instead of leaving a bare
literal that reads like the mistake. Suite is back to 15.3s.

The guard now also rejects `timeout: TestConstants.shortTimeout` and
`defaultTimeout`, while accepting `negativeWaitWindow` by name.

Re-verified with a canary carrying every banned shape — literal default,
both named constants, and an elapsed-time upper bound. All four were
flagged with file and line; the suite went green again on removal.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* Delete shortTimeout now that nothing may use it

The hygiene guard bans `TestConstants.shortTimeout` at every wait site
and the last users were converted, so Periphery correctly flagged the
constant itself as dead and failed CI. Remove it from both TestConstants
copies; the banned-name entry stays so the symbol cannot quietly return.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: jack <jackjackbits@users.noreply.github.com>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
2026-07-26 23:45:20 +01:00

2153 lines
89 KiB
Swift
Raw Blame History

import Combine
import XCTest
@testable import bitchat
@MainActor
final class NostrRelayManagerTests: XCTestCase {
private let expectedDefaultRelayCount = 4
func test_connect_directMode_connectsExistingDefaultRelaysWhenActivationBecomesAllowed() async {
let context = makeContext(permission: .authorized, activationAllowed: false)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.activationAllowed.value = true
context.manager.connect()
let connected = await waitUntil {
context.sessionFactory.requestedURLs.count == self.expectedDefaultRelayCount &&
context.manager.relays.allSatisfy(\.isConnected)
}
XCTAssertTrue(connected)
}
func test_permissionPublisher_addsAndRemovesDefaultRelays() async {
let context = makeContext(permission: .denied, favorites: [])
XCTAssertEqual(context.manager.getRelayStatuses().count, 0)
context.permissionSubject.send(.authorized)
let defaultRelaysConnected = await waitUntil {
context.manager.getRelayStatuses().count == self.expectedDefaultRelayCount &&
context.manager.relays.allSatisfy(\.isConnected)
}
XCTAssertTrue(defaultRelaysConnected)
context.permissionSubject.send(.denied)
let defaultRelaysRemoved = await waitUntil {
context.manager.getRelayStatuses().isEmpty
}
XCTAssertTrue(defaultRelaysRemoved)
XCTAssertEqual(context.sessionFactory.allConnections.count, expectedDefaultRelayCount)
XCTAssertTrue(context.sessionFactory.allConnections.allSatisfy { $0.cancelCallCount >= 1 })
}
/// A relay removed while its connection is queued behind Tor bootstrap
/// must stay removed: draining the pending set used to resurrect it,
/// because `dropRelays` never touched `pendingTorConnectionURLs` and a
/// custom relay is in neither the default set nor the allow-list filter.
func test_relayRemovedWhileWaitingForTor_staysRemovedWhenTorBecomesReady() async {
let customURL = "wss://custom-removed.example"
let center = NotificationCenter()
let customRelays = MutableRelayList(urls: [customURL])
let context = makeContext(
permission: .authorized,
userTorEnabled: true,
torEnforced: true,
torIsReady: false,
notificationCenter: center,
customRelays: customRelays
)
// Defaults plus the custom relay all queue while Tor bootstraps.
context.manager.connect()
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
XCTAssertEqual(context.torWaiter.awaitCallCount, 1)
// The relay is removed by hand before Tor is ready.
customRelays.urls = []
center.post(name: NostrRelaySettings.didChangeNotification, object: nil)
// The settings sink hops through the main queue; let it land.
try? await Task.sleep(nanoseconds: 20_000_000)
context.torWaiter.resolve(true)
let defaultsConnected = await waitUntil {
context.sessionFactory.requestedURLs.count == self.expectedDefaultRelayCount
}
XCTAssertTrue(defaultsConnected)
XCTAssertFalse(
context.sessionFactory.requestedURLs.contains(customURL),
"a relay removed while Tor was bootstrapping must not reconnect when the pending queue drains"
)
}
func test_connect_waitsForTorReadinessBeforeCreatingSessions() async {
let context = makeContext(permission: .authorized, userTorEnabled: true, torEnforced: true, torIsReady: false)
context.manager.connect()
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.torWaiter.resolve(true)
let connectedAfterTorReady = await waitUntil {
context.sessionFactory.requestedURLs.count == self.expectedDefaultRelayCount &&
context.manager.relays.allSatisfy(\.isConnected)
}
XCTAssertTrue(connectedAfterTorReady)
}
func test_connect_coalescesRepeatedCallsWhileWaitingForTor() async {
let context = makeContext(permission: .authorized, userTorEnabled: true, torEnforced: true, torIsReady: false)
context.manager.connect()
context.manager.connect()
context.manager.connect()
XCTAssertEqual(context.torWaiter.awaitCallCount, 1)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.torWaiter.resolve(true)
let connectedAfterTorReady = await waitUntil {
context.sessionFactory.requestedURLs.count == self.expectedDefaultRelayCount &&
context.manager.relays.allSatisfy(\.isConnected)
}
XCTAssertTrue(connectedAfterTorReady)
}
func test_connect_whenTorReadinessFailsDoesNotCreateSessions() async {
let context = makeContext(permission: .authorized, userTorEnabled: true, torEnforced: true, torIsReady: false)
context.manager.connect()
context.torWaiter.resolve(false)
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
XCTAssertFalse(context.manager.isConnected)
}
func test_connect_retriesTorWaitAndConnectsWhenTorBecomesReady() async {
let context = makeContext(permission: .authorized, userTorEnabled: true, torEnforced: true, torIsReady: false)
context.manager.connect()
XCTAssertEqual(context.torWaiter.awaitCallCount, 1)
context.torWaiter.resolve(false)
// A failed wait re-queues the same targets and waits again instead of dropping them.
XCTAssertEqual(context.torWaiter.awaitCallCount, 2)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.torWaiter.resolve(true)
let connected = await waitUntil {
context.sessionFactory.requestedURLs.count == self.expectedDefaultRelayCount &&
context.manager.relays.allSatisfy(\.isConnected)
}
XCTAssertTrue(connected)
}
func test_subscribe_unblocksDeferredEOSEWhenTorWaitAttemptsExhausted() async {
let relayURL = "wss://tor-eose-unblock.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "tor-sub-unblock",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
for _ in 0..<TransportConfig.nostrTorReadyMaxWaitAttempts {
context.torWaiter.resolve(false)
}
// Fail-closed (no sessions), but the EOSE caller is unblocked.
XCTAssertEqual(eoseCount, 1)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
}
func test_subscribe_parkedEOSEFiresAfterFallbackTimeoutWhenTorNeverBecomesReady() async {
let relayURL = "wss://tor-eose-parked-fallback.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "tor-parked-fallback",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
// Parking the callback schedules the normal EOSE fallback, not just
// the Tor retry-exhaustion unblock (~minutes later).
XCTAssertEqual(context.scheduler.scheduled.first?.delay, TransportConfig.nostrSubscriptionEOSEFallbackSeconds)
XCTAssertEqual(eoseCount, 0)
context.scheduler.runNext()
let unblocked = await waitUntil { eoseCount == 1 }
XCTAssertTrue(unblocked)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
// A later retry-exhaustion unblock must not fire the callback again.
for _ in 0..<TransportConfig.nostrTorReadyMaxWaitAttempts {
context.torWaiter.resolve(false)
}
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(eoseCount, 1)
}
func test_subscribe_parkedEOSEFallbackIsNoOpWhenTorRecoversFirst() async throws {
let relayURL = "wss://tor-eose-parked-recover.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "tor-parked-recover",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
// Tor recovers before the fallback fires; the callback is promoted to
// a real EOSE tracker when the subscription flushes.
context.torWaiter.resolve(true)
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscribed)
// The parked fallback (scheduled first) is now a no-op.
context.scheduler.runNext()
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(eoseCount, 0)
// The real EOSE still completes the initial load exactly once...
try context.sessionFactory.latestConnection(for: relayURL)?.emitEOSE(subscriptionID: "tor-parked-recover")
let eoseCompleted = await waitUntil { eoseCount == 1 }
XCTAssertTrue(eoseCompleted)
// ...and the promoted tracker's own fallback does not double-fire.
context.scheduler.runNext()
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(eoseCount, 1)
}
func test_subscribe_parkedEOSEFallbackIsNoOpAfterRetryExhaustionUnblock() async {
let relayURL = "wss://tor-eose-parked-exhausted.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "tor-parked-exhausted",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
// Retry exhaustion unblocks the parked callback first.
for _ in 0..<TransportConfig.nostrTorReadyMaxWaitAttempts {
context.torWaiter.resolve(false)
}
XCTAssertEqual(eoseCount, 1)
// The parked fallback finds nothing to do.
context.scheduler.runNext()
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(eoseCount, 1)
}
func test_sendEvent_survivesFailedTorWaitAndSendsWhenTorRecovers() async throws {
let relayURL = "wss://tor-send-retry.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
let event = try makeSignedEvent(content: "queued through tor stall")
context.manager.sendEvent(event, to: [relayURL])
XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 1)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.torWaiter.resolve(false)
XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 1)
context.torWaiter.resolve(true)
let sent = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1 &&
context.manager.debugPendingMessageQueueCount == 0
}
XCTAssertTrue(sent)
}
func test_sendEvent_pendingQueueDropsOldestBeyondCap() async throws {
let relayURL = "wss://tor-send-cap.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
let identity = try NostrIdentity.generate()
for i in 0...(TransportConfig.nostrPendingSendQueueCap + 4) {
let event = NostrEvent(
pubkey: identity.publicKeyHex,
createdAt: Date(),
kind: .textNote,
tags: [],
content: "cap-\(i)"
)
context.manager.sendEvent(try event.sign(with: identity.schnorrSigningKey()), to: [relayURL])
}
XCTAssertEqual(context.manager.debugPendingMessageQueueCount, TransportConfig.nostrPendingSendQueueCap)
}
func test_sendEvent_waitsForTorReadinessBeforeSending() async throws {
let relayURL = "wss://tor-ready.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
let event = try makeSignedEvent(content: "deferred")
context.manager.sendEvent(event, to: [relayURL])
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.torWaiter.resolve(true)
let sentAfterTorReady = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1 &&
context.manager.relays.first(where: { $0.url == relayURL })?.messagesSent == 1
}
XCTAssertTrue(sentAfterTorReady)
}
func test_sendEvent_queuesUntilRelayIsMarkedConnected() async throws {
let relayURL = "wss://connect-before-send.example"
let context = makeContext(permission: .denied)
let event = try makeSignedEvent(content: "wait for connected")
context.manager.sendEvent(event, to: [relayURL])
XCTAssertEqual(context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count, 0)
XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 1)
let sentAfterConnected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1 &&
context.manager.debugPendingMessageQueueCount == 0 &&
context.manager.relays.first(where: { $0.url == relayURL })?.messagesSent == 1
}
XCTAssertTrue(sentAfterConnected)
}
func test_sendEvent_queuesWhileBackgroundedAndFlushesWhenForegrounded() async throws {
let relayURL = "wss://queue-flush.example"
let context = makeContext(
permission: .denied,
userTorEnabled: true,
torEnforced: true,
torIsReady: true,
torIsForeground: false
)
let event = try makeSignedEvent(content: "queued")
context.manager.sendEvent(event, to: [relayURL])
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.torForeground.value = true
context.manager.ensureConnections(to: [relayURL])
let flushed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1 &&
context.manager.relays.first(where: { $0.url == relayURL })?.messagesSent == 1
}
XCTAssertTrue(flushed)
}
func test_sendEvent_sendFailureDoesNotIncrementMessageCount() async throws {
let relayURL = "wss://send-failure.example"
let context = makeContext(permission: .denied)
context.sessionFactory.sendErrorByURL[relayURL] = NSError(domain: "send", code: 1)
let event = try makeSignedEvent(content: "send failure")
context.manager.sendEvent(event, to: [relayURL])
let attempted = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(attempted)
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(context.manager.relays.first(where: { $0.url == relayURL })?.messagesSent, 0)
}
func test_sendEventImmediately_allRejectsFailThenOKRetrySucceeds() async throws {
let relays = [
"wss://confirmed-reject-one.example",
"wss://confirmed-reject-two.example"
]
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: relays)
let connected = await waitUntil {
relays.allSatisfy { relay in
context.manager.relays.first(where: { $0.url == relay })?.isConnected == true
}
}
XCTAssertTrue(connected)
let event = try makeSignedEvent(content: "confirmed reject then retry")
var results: [Bool] = []
var completionsWereOnMain: [Bool] = []
context.manager.sendEventImmediately(event, to: relays) {
results.append($0)
completionsWereOnMain.append(Thread.isMainThread)
}
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertTrue(results.isEmpty, "socket writes alone must not confirm durability")
for relay in relays {
try context.sessionFactory.latestConnection(for: relay)?.emitOK(
eventID: event.id,
success: false,
reason: "rejected"
)
}
let failedCompleted = await waitUntil { results.count == 1 }
XCTAssertTrue(failedCompleted)
XCTAssertEqual(results, [false])
XCTAssertEqual(completionsWereOnMain, [true])
XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 0)
// The explicit rejection leaves the same event retryable.
try? await Task.sleep(nanoseconds: 20_000_000)
context.manager.sendEventImmediately(event, to: relays) {
results.append($0)
completionsWereOnMain.append(Thread.isMainThread)
}
try context.sessionFactory.latestConnection(for: relays[0])?.emitOK(
eventID: event.id,
success: true,
reason: "accepted"
)
let successfulCompleted = await waitUntil { results.count == 2 }
XCTAssertTrue(successfulCompleted)
XCTAssertEqual(results, [false, true])
XCTAssertEqual(completionsWereOnMain, [true, true])
XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 0)
}
func test_sendEventImmediately_mixedRelayOKUsesAcceptedResultOnce() async throws {
let relays = ["wss://confirmed-mixed-one.example", "wss://confirmed-mixed-two.example"]
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: relays)
let connected = await waitUntil {
relays.allSatisfy { relay in
context.manager.relays.first(where: { $0.url == relay })?.isConnected == true
}
}
XCTAssertTrue(connected)
let event = try makeSignedEvent(content: "mixed confirmation")
var results: [Bool] = []
context.manager.sendEventImmediately(event, to: relays) { results.append($0) }
try context.sessionFactory.latestConnection(for: relays[0])?.emitOK(
eventID: event.id,
success: false,
reason: "policy"
)
try context.sessionFactory.latestConnection(for: relays[1])?.emitOK(
eventID: event.id,
success: true,
reason: "accepted"
)
let completed = await waitUntil { results.count == 1 }
XCTAssertTrue(completed)
XCTAssertEqual(results, [true])
}
func test_sendEventImmediately_timeoutFailsAndIgnoresLateWriteAndOK() async throws {
let relay = "wss://confirmed-timeout.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relay])
let connected = await waitUntil {
context.manager.relays.first(where: { $0.url == relay })?.isConnected == true
}
XCTAssertTrue(connected)
let connection = try XCTUnwrap(context.sessionFactory.latestConnection(for: relay))
connection.deferSendCompletions = true
let event = try makeSignedEvent(content: "confirmation timeout")
var results: [Bool] = []
context.manager.sendEventImmediately(event, to: [relay]) { results.append($0) }
XCTAssertTrue(results.isEmpty)
XCTAssertEqual(
context.scheduler.scheduled.first?.delay,
TransportConfig.nostrConfirmedSendAckTimeoutSeconds
)
context.scheduler.runNext()
let timedOut = await waitUntil { results.count == 1 }
XCTAssertTrue(timedOut)
XCTAssertEqual(results, [false])
connection.flushDeferredSendCompletions()
try connection.emitOK(eventID: event.id, success: true, reason: "late")
try? await Task.sleep(nanoseconds: 30_000_000)
XCTAssertEqual(results, [false])
}
func test_sendEvent_queueIsPrunedWhenDefaultRelaysAreRevoked() async throws {
let context = makeContext(
permission: .authorized,
userTorEnabled: true,
torEnforced: true,
torIsReady: true,
torIsForeground: false
)
let event = try makeSignedEvent(content: "queued default")
context.manager.sendEvent(event)
let queued = await waitUntil {
context.manager.debugPendingMessageQueueCount == 1
}
XCTAssertTrue(queued)
context.permissionSubject.send(.denied)
let cleared = await waitUntil {
context.manager.debugPendingMessageQueueCount == 0 &&
context.manager.relays.isEmpty
}
XCTAssertTrue(cleared)
}
func test_connect_doesNothingWhenActivationIsDisallowed() {
let context = makeContext(permission: .authorized, activationAllowed: false)
context.manager.connect()
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
XCTAssertFalse(context.manager.isConnected)
}
func test_ensureConnections_deduplicatesRelayURLs() async {
let relayOne = "wss://relay-one.example"
let relayTwo = "wss://relay-two.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relayOne, "wss://relay-one.example:443/", "WSS://RELAY-TWO.EXAMPLE:443"])
let connected = await waitUntil {
Set(context.manager.getRelayStatuses().map(\.url)) == Set([relayOne, relayTwo]) &&
context.manager.relays.allSatisfy(\.isConnected)
}
XCTAssertTrue(connected)
XCTAssertEqual(context.sessionFactory.requestedURLs, [relayOne, relayTwo])
}
func test_ensureConnections_coalescesTargetsWhileWaitingForTor() async {
let relayOne = "wss://tor-one.example"
let relayTwo = "wss://tor-two.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
context.manager.ensureConnections(to: [relayOne])
context.manager.ensureConnections(to: [relayTwo, relayOne])
XCTAssertEqual(context.torWaiter.awaitCallCount, 1)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.torWaiter.resolve(true)
let connected = await waitUntil {
Set(context.sessionFactory.requestedURLs) == Set([relayOne, relayTwo]) &&
context.manager.relays.allSatisfy(\.isConnected)
}
XCTAssertTrue(connected)
}
func test_subscribe_coalescesRapidDuplicateRequests() async {
let relayURL = "wss://subscribe.example"
let context = makeContext(permission: .denied)
let filter = makeFilter()
context.manager.subscribe(filter: filter, id: "sub", relayUrls: [relayURL], handler: { _ in })
let firstSent = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(firstSent)
context.clock.now = context.clock.now.addingTimeInterval(0.5)
context.manager.subscribe(filter: filter, id: "sub", relayUrls: [relayURL], handler: { _ in })
XCTAssertEqual(context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count, 1)
}
func test_subscribe_coalescesDuplicateRequestsBeforeTorReadyAndDefersEOSE() async throws {
let relayURL = "wss://tor-subscribe-coalesce.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "tor-sub",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
context.manager.subscribe(
filter: makeFilter(),
id: "tor-sub",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
XCTAssertEqual(context.torWaiter.awaitCallCount, 1)
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 1)
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(eoseCount, 0)
context.torWaiter.resolve(true)
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscribed)
try context.sessionFactory.latestConnection(for: relayURL)?.emitEOSE(subscriptionID: "tor-sub")
let eoseCompleted = await waitUntil { eoseCount == 1 }
XCTAssertTrue(eoseCompleted)
}
func test_subscribe_sameActiveRequestDoesNotRequeue() async {
let relayURL = "wss://active-subscribe.example"
let context = makeContext(permission: .denied)
let filter = makeFilter()
context.manager.subscribe(filter: filter, id: "sub", relayUrls: [relayURL], handler: { _ in })
let firstSent = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(firstSent)
context.clock.now = context.clock.now.addingTimeInterval(2.0)
context.manager.subscribe(filter: filter, id: "sub", relayUrls: [relayURL], handler: { _ in })
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count, 1)
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 0)
}
func test_subscribe_waitsForTorReadinessAndPreservesEOSECallback() async throws {
let relayURL = "wss://tor-subscribe.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "tor-eose",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.torWaiter.resolve(true)
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscribed)
try context.sessionFactory.latestConnection(for: relayURL)?.emitEOSE(subscriptionID: "tor-eose")
let eoseCompleted = await waitUntil { eoseCount == 1 }
XCTAssertTrue(eoseCompleted)
}
func test_subscribe_withoutAllowedRelays_callsEOSEImmediatelyAndDoesNotFlushLater() async {
let context = makeContext(permission: .denied)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "blocked-defaults",
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
XCTAssertEqual(eoseCount, 1)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.permissionSubject.send(.authorized)
let connected = await waitUntil {
context.sessionFactory.allConnections.count == self.expectedDefaultRelayCount &&
context.manager.relays.allSatisfy(\.isConnected)
}
XCTAssertTrue(connected)
XCTAssertTrue(context.sessionFactory.allConnections.allSatisfy { $0.sentStrings.isEmpty })
}
func test_permissionRevocation_clearsQueuedDefaultSubscriptions() async {
let context = makeContext(
permission: .authorized,
userTorEnabled: true,
torEnforced: true,
torIsReady: true,
torIsForeground: false
)
let defaultRelay = "wss://relay.damus.io"
context.manager.subscribe(filter: makeFilter(), id: "queued-default", handler: { _ in })
let queued = await waitUntil {
context.manager.debugPendingSubscriptionCount(for: defaultRelay) == 1
}
XCTAssertTrue(queued)
context.permissionSubject.send(.denied)
let cleared = await waitUntil {
context.manager.debugPendingSubscriptionCount(for: defaultRelay) == 0 &&
context.manager.relays.isEmpty
}
XCTAssertTrue(cleared)
}
func test_unsubscribe_allowsResubscribeWithSameID() async {
let relayURL = "wss://subscribe.example"
let context = makeContext(permission: .denied)
let filter = makeFilter()
context.manager.subscribe(filter: filter, id: "sub", relayUrls: [relayURL], handler: { _ in })
let initialSubscribeSent = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(initialSubscribeSent)
context.manager.unsubscribe(id: "sub")
let closeSent = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 2
}
XCTAssertTrue(closeSent)
context.clock.now = context.clock.now.addingTimeInterval(0.2)
context.manager.subscribe(filter: filter, id: "sub", relayUrls: [relayURL], handler: { _ in })
let resubscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 3
}
XCTAssertTrue(resubscribed)
}
func test_receiveEvent_deliversHandlerAndTracksReceivedCount() async throws {
let relayURL = "wss://events.example"
let context = makeContext(permission: .denied)
let filter = makeFilter()
let event = try makeSignedEvent(content: "hello")
var receivedEvent: NostrEvent?
context.manager.subscribe(filter: filter, id: "events", relayUrls: [relayURL]) { event in
receivedEvent = event
}
let subscriptionSent = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscriptionSent)
try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(subscriptionID: "events", event: event)
let delivered = await waitUntil {
receivedEvent?.id == event.id &&
context.manager.relays.first(where: { $0.url == relayURL })?.messagesReceived == 1
}
XCTAssertTrue(delivered)
XCTAssertEqual(receivedEvent?.id, event.id)
}
func test_receiveEvent_deduplicatesSameSubscriptionEventAcrossRelays() async throws {
let firstRelayURL = "wss://events-one.example"
let secondRelayURL = "wss://events-two.example"
let context = makeContext(permission: .denied)
let event = try makeSignedEvent(content: "duplicate")
var receivedIDs: [String] = []
context.manager.subscribe(
filter: makeFilter(),
id: "events",
relayUrls: [firstRelayURL, secondRelayURL]
) { event in
receivedIDs.append(event.id)
}
let subscriptionsSent = await waitUntil {
context.sessionFactory.latestConnection(for: firstRelayURL)?.sentStrings.count == 1 &&
context.sessionFactory.latestConnection(for: secondRelayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscriptionsSent)
try context.sessionFactory.latestConnection(for: firstRelayURL)?.emitEventMessage(subscriptionID: "events", event: event)
try context.sessionFactory.latestConnection(for: secondRelayURL)?.emitEventMessage(subscriptionID: "events", event: event)
// Wait on the DELIVERY-side state: handler dispatch and the duplicate
// drop both land on the second main hop, after off-main verification.
let settled = await waitUntil {
context.manager.relays.first(where: { $0.url == firstRelayURL })?.messagesReceived == 1 &&
context.manager.relays.first(where: { $0.url == secondRelayURL })?.messagesReceived == 1 &&
receivedIDs == [event.id] &&
context.manager.debugDuplicateInboundEventDropCount == 1
}
XCTAssertTrue(settled)
XCTAssertEqual(receivedIDs, [event.id])
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, 1)
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount(forSubscriptionID: "events"), 1)
}
func test_receiveEvent_duplicateFanInDeliversOnceAndCountsDrops() async throws {
let relayURLs = (0..<8).map { "wss://fan-in-\($0).example" }
let context = makeContext(permission: .denied)
let event = try makeSignedEvent(content: "fan-in")
var receivedIDs: [String] = []
context.manager.subscribe(
filter: makeFilter(),
id: "presence",
relayUrls: relayURLs
) { event in
receivedIDs.append(event.id)
}
let subscriptionsSent = await waitUntil {
relayURLs.allSatisfy { relayURL in
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
}
XCTAssertTrue(subscriptionsSent)
for relayURL in relayURLs {
try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(
subscriptionID: "presence",
event: event
)
}
// Wait on the DELIVERY-side state: the winner's handler dispatch and
// the losers' duplicate drops both land on the second main hop, after
// off-main verification — messagesReceived (first hop) settles sooner.
let settled = await waitUntil {
relayURLs.allSatisfy { relayURL in
context.manager.relays.first(where: { $0.url == relayURL })?.messagesReceived == 1
} &&
receivedIDs == [event.id] &&
context.manager.debugDuplicateInboundEventDropCount == relayURLs.count - 1
}
XCTAssertTrue(settled)
XCTAssertEqual(receivedIDs, [event.id])
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, relayURLs.count - 1)
XCTAssertEqual(
context.manager.debugDuplicateInboundEventDropCount(forSubscriptionID: "presence"),
relayURLs.count - 1
)
}
func test_receiveEvent_invalidSignatureDoesNotPoisonDuplicateCache() async throws {
let firstRelayURL = "wss://invalid-first-one.example"
let secondRelayURL = "wss://invalid-first-two.example"
let context = makeContext(permission: .denied)
let event = try makeSignedEvent(content: "valid-after-invalid")
let invalidEvent = invalidSignatureCopy(of: event)
var receivedIDs: [String] = []
context.manager.subscribe(
filter: makeFilter(),
id: "events",
relayUrls: [firstRelayURL, secondRelayURL]
) { event in
receivedIDs.append(event.id)
}
let subscriptionsSent = await waitUntil {
context.sessionFactory.latestConnection(for: firstRelayURL)?.sentStrings.count == 1 &&
context.sessionFactory.latestConnection(for: secondRelayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscriptionsSent)
try context.sessionFactory.latestConnection(for: firstRelayURL)?.emitEventMessage(subscriptionID: "events", event: invalidEvent)
try context.sessionFactory.latestConnection(for: secondRelayURL)?.emitEventMessage(subscriptionID: "events", event: event)
// Wait on the DELIVERY-side state (second main hop, after off-main
// verification), not just messagesReceived (first main hop) — the
// handler only fires after verify + a second hop.
let genuineDelivered = await waitUntil {
context.manager.relays.first(where: { $0.url == firstRelayURL })?.messagesReceived == 1 &&
context.manager.relays.first(where: { $0.url == secondRelayURL })?.messagesReceived == 1 &&
receivedIDs == [event.id]
}
XCTAssertTrue(genuineDelivered)
XCTAssertEqual(receivedIDs, [event.id])
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, 0)
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount(forSubscriptionID: "events"), 0)
}
/// The relay boundary is the single signature-verification point for the
/// whole inbound path (downstream pipelines no longer re-verify), so a
/// tampered gift wrap (kind 1059, the DM/mailbox path) must be dropped
/// here — and must not poison the dedup cache against the genuine copy.
func test_receiveGiftWrap_tamperedSignatureIsDroppedAndDoesNotPoisonDedup() async throws {
let firstRelayURL = "wss://giftwrap-one.example"
let secondRelayURL = "wss://giftwrap-two.example"
let context = makeContext(permission: .denied)
let sender = try NostrIdentity.generate()
let recipient = try NostrIdentity.generate()
let giftWrap = try NostrProtocol.createPrivateMessage(
content: "psst",
recipientPubkey: recipient.publicKeyHex,
senderIdentity: sender
)
let tampered = invalidSignatureCopy(of: giftWrap)
var receivedIDs: [String] = []
context.manager.subscribe(
filter: makeFilter(),
id: "gift-wraps",
relayUrls: [firstRelayURL, secondRelayURL]
) { event in
receivedIDs.append(event.id)
}
let subscriptionsSent = await waitUntil {
context.sessionFactory.latestConnection(for: firstRelayURL)?.sentStrings.count == 1 &&
context.sessionFactory.latestConnection(for: secondRelayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscriptionsSent)
try context.sessionFactory.latestConnection(for: firstRelayURL)?.emitEventMessage(subscriptionID: "gift-wraps", event: tampered)
try context.sessionFactory.latestConnection(for: secondRelayURL)?.emitEventMessage(subscriptionID: "gift-wraps", event: giftWrap)
// Wait on the DELIVERY-side state (second main hop, after off-main
// verification), not just messagesReceived (first main hop) — the
// handler only fires after verify + a second hop.
let genuineDelivered = await waitUntil {
context.manager.relays.first(where: { $0.url == firstRelayURL })?.messagesReceived == 1 &&
context.manager.relays.first(where: { $0.url == secondRelayURL })?.messagesReceived == 1 &&
receivedIDs == [giftWrap.id]
}
XCTAssertTrue(genuineDelivered)
XCTAssertEqual(receivedIDs, [giftWrap.id])
XCTAssertEqual(context.manager.debugDuplicateInboundEventDropCount, 0)
}
/// Signature verification runs off-main in a per-relay serial consumer;
/// several frames buffered on one socket must still be delivered to the
/// handler in that relay's arrival order.
func test_receiveEvent_deliversBackToBackEventsInArrivalOrder() async throws {
let relayURL = "wss://ordered.example"
let context = makeContext(permission: .denied)
let events = try (0..<12).map { try makeSignedEvent(content: "ordered-\($0)") }
var receivedIDs: [String] = []
context.manager.subscribe(filter: makeFilter(), id: "ordered", relayUrls: [relayURL]) { event in
receivedIDs.append(event.id)
}
let subscriptionSent = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscriptionSent)
for event in events {
try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(subscriptionID: "ordered", event: event)
}
let allDelivered = await waitUntil(timeout: TestConstants.settleTimeout) {
receivedIDs.count == events.count
}
XCTAssertTrue(allDelivered)
XCTAssertEqual(receivedIDs, events.map(\.id))
}
/// Each relay owns its own off-main verify pipeline, so a large backlog of
/// EVENT frames on one relay must NOT head-of-line-block a frame that
/// arrives on a different relay. Under the previous single global consumer,
/// relay B's single event (emitted after relay A's whole burst) could only
/// be delivered once every frame in A's backlog had been Schnorr-verified;
/// with per-relay pipelines B verifies concurrently and lands before A's
/// backlog drains.
func test_receiveEvent_busyRelayDoesNotBlockOtherRelayDelivery() async throws {
let busyRelayURL = "wss://busy-relay.example"
let quietRelayURL = "wss://quiet-relay.example"
let context = makeContext(permission: .denied)
// Distinct subscriptions per relay so dedup never coalesces A vs. B.
let busyEvents = try (0..<200).map { try makeSignedEvent(content: "busy-\($0)") }
let quietEvent = try makeSignedEvent(content: "quiet")
var busyDeliveredCount = 0
var quietDeliveredAfterBusyCount = -1 // busy-count observed when B lands
context.manager.subscribe(filter: makeFilter(), id: "busy", relayUrls: [busyRelayURL]) { _ in
busyDeliveredCount += 1
}
context.manager.subscribe(filter: makeFilter(), id: "quiet", relayUrls: [quietRelayURL]) { _ in
if quietDeliveredAfterBusyCount < 0 {
quietDeliveredAfterBusyCount = busyDeliveredCount
}
}
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: busyRelayURL)?.sentStrings.count == 1 &&
context.sessionFactory.latestConnection(for: quietRelayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscribed)
// Flood relay A first, then emit a single frame on relay B.
for event in busyEvents {
try context.sessionFactory.latestConnection(for: busyRelayURL)?.emitEventMessage(subscriptionID: "busy", event: event)
}
try context.sessionFactory.latestConnection(for: quietRelayURL)?.emitEventMessage(subscriptionID: "quiet", event: quietEvent)
let quietDelivered = await waitUntil(timeout: TestConstants.settleTimeout) { quietDeliveredAfterBusyCount >= 0 }
XCTAssertTrue(quietDelivered, "relay B's event was never delivered")
// The signal: B did not have to wait for A's entire backlog. If the two
// pipelines were globally serialized, B could only land after all 200 of
// A's frames, so busyDeliveredCount would be 200 when B arrived.
XCTAssertLessThan(
quietDeliveredAfterBusyCount,
busyEvents.count,
"relay B was head-of-line blocked behind relay A's backlog"
)
// Both relays still drain fully and in order.
let allDelivered = await waitUntil(timeout: TestConstants.settleTimeout) {
busyDeliveredCount == busyEvents.count
}
XCTAssertTrue(allDelivered)
}
func test_receiveEvent_withoutHandlerStillTracksReceivedCount() async throws {
let relayURL = "wss://missing-handler.example"
let context = makeContext(permission: .denied)
let event = try makeSignedEvent(content: "unhandled")
context.manager.ensureConnections(to: [relayURL])
let connected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL) != nil &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(connected)
try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(subscriptionID: "missing", event: event)
let counted = await waitUntil {
context.manager.relays.first(where: { $0.url == relayURL })?.messagesReceived == 1
}
XCTAssertTrue(counted)
}
func test_noticeAndMalformedMessages_keepReceiveLoopAliveForLaterEvents() async throws {
let relayURL = "wss://parser.example"
let context = makeContext(permission: .denied)
var receivedIDs: [String] = []
let firstEvent = try makeSignedEvent(content: "after notice")
let secondEvent = try makeSignedEvent(content: "after malformed")
context.manager.subscribe(filter: makeFilter(), id: "parser", relayUrls: [relayURL]) { event in
receivedIDs.append(event.id)
}
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscribed)
try context.sessionFactory.latestConnection(for: relayURL)?.emitNotice(message: "ignored")
try? await Task.sleep(nanoseconds: 20_000_000)
try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(subscriptionID: "parser", event: firstEvent)
let firstDelivered = await waitUntil {
receivedIDs == [firstEvent.id]
}
XCTAssertTrue(firstDelivered)
try context.sessionFactory.latestConnection(for: relayURL)?.emitRawString("not-json")
try? await Task.sleep(nanoseconds: 20_000_000)
try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(subscriptionID: "parser", event: secondEvent)
let secondDelivered = await waitUntil {
receivedIDs == [firstEvent.id, secondEvent.id]
}
XCTAssertTrue(secondDelivered)
}
func test_okMessages_clearPendingGiftWrapIDs() async throws {
let relayURL = "wss://ok.example"
let context = makeContext(permission: .denied)
let successID = "gift-wrap-success"
let failureID = "gift-wrap-failure"
context.manager.ensureConnections(to: [relayURL])
let connected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL) != nil &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(connected)
NostrRelayManager.registerPendingGiftWrap(id: successID)
try context.sessionFactory.latestConnection(for: relayURL)?.emitOK(eventID: successID, success: true, reason: "ok")
let successCleared = await waitUntil {
!NostrRelayManager.pendingGiftWrapIDs.contains(successID)
}
XCTAssertTrue(successCleared)
NostrRelayManager.registerPendingGiftWrap(id: failureID)
try context.sessionFactory.latestConnection(for: relayURL)?.emitOK(eventID: failureID, success: false, reason: "rejected")
let failureCleared = await waitUntil {
!NostrRelayManager.pendingGiftWrapIDs.contains(failureID)
}
XCTAssertTrue(failureCleared)
}
func test_eoseCallback_waitsForAllTargetedRelays() async throws {
let relayOne = "wss://one.example"
let relayTwo = "wss://two.example"
let context = makeContext(permission: .denied)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "eose",
relayUrls: [relayOne, relayTwo],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
let bothConnected = await waitUntil {
context.sessionFactory.latestConnection(for: relayOne)?.sentStrings.count == 1 &&
context.sessionFactory.latestConnection(for: relayTwo)?.sentStrings.count == 1
}
XCTAssertTrue(bothConnected)
try context.sessionFactory.latestConnection(for: relayOne)?.emitEOSE(subscriptionID: "eose")
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(eoseCount, 0)
try context.sessionFactory.latestConnection(for: relayTwo)?.emitEOSE(subscriptionID: "eose")
let eoseCompleted = await waitUntil { eoseCount == 1 }
XCTAssertTrue(eoseCompleted)
}
func test_eoseTimeout_invokesCallbackOnceAndIgnoresLateEOSE() async throws {
let relayURL = "wss://timeout.example"
let context = makeContext(permission: .denied)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "timeout",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.count == 1
}
XCTAssertTrue(subscribed)
// The fallback is scheduled but has not fired yet.
XCTAssertEqual(context.scheduler.scheduled.first?.delay, TransportConfig.nostrSubscriptionEOSEFallbackSeconds)
XCTAssertEqual(eoseCount, 0)
context.scheduler.runNext()
let timedOut = await waitUntil { eoseCount == 1 }
XCTAssertTrue(timedOut)
try context.sessionFactory.latestConnection(for: relayURL)?.emitEOSE(subscriptionID: "timeout")
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(eoseCount, 1)
}
func test_eose_completesWhenRelayDisconnectsBeforeEOSE() async throws {
let relayOne = "wss://eose-drop-one.example"
let relayTwo = "wss://eose-drop-two.example"
let context = makeContext(permission: .denied)
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "eose-drop",
relayUrls: [relayOne, relayTwo],
handler: { _ in },
onEOSE: { eoseCount += 1 }
)
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayOne)?.sentStrings.count == 1 &&
context.sessionFactory.latestConnection(for: relayTwo)?.sentStrings.count == 1
}
XCTAssertTrue(subscribed)
try context.sessionFactory.latestConnection(for: relayOne)?.emitEOSE(subscriptionID: "eose-drop")
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(eoseCount, 0)
context.sessionFactory.latestConnection(for: relayTwo)?.fail(
error: NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
)
let completed = await waitUntil { eoseCount == 1 }
XCTAssertTrue(completed)
}
func test_reconnect_replaysActiveSubscriptionsAndDeliversEvents() async throws {
let relayURL = "wss://replay.example"
let context = makeContext(permission: .denied)
var received: [NostrEvent] = []
context.manager.subscribe(
filter: makeFilter(),
id: "replay-sub",
relayUrls: [relayURL],
handler: { received.append($0) }
)
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.contains { $0.contains("replay-sub") } == true
}
XCTAssertTrue(subscribed)
// Drop the socket; the relay forgets the subscription with it.
context.sessionFactory.latestConnection(for: relayURL)?.fail(
error: NSError(domain: NSURLErrorDomain, code: NSURLErrorNetworkConnectionLost)
)
let retryScheduled = await waitUntil { !context.scheduler.scheduled.isEmpty }
XCTAssertTrue(retryScheduled)
context.scheduler.runNext()
let replayed = await waitUntil {
let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
return connections.count == 2 &&
connections.last?.sentStrings.contains { $0.contains("replay-sub") } == true
}
XCTAssertTrue(replayed)
let event = try makeSignedEvent(content: "after reconnect")
try context.sessionFactory.latestConnection(for: relayURL)?.emitEventMessage(subscriptionID: "replay-sub", event: event)
let delivered = await waitUntil { received.count == 1 }
XCTAssertTrue(delivered)
}
func test_disconnectThenConnect_restoresSubscriptions() async {
let relayURL = "wss://restore.example"
let context = makeContext(permission: .denied)
context.manager.subscribe(filter: makeFilter(), id: "restore-sub", relayUrls: [relayURL], handler: { _ in })
let subscribed = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.contains { $0.contains("restore-sub") } == true
}
XCTAssertTrue(subscribed)
// Background → foreground: connections reset, subscriptions must survive.
context.manager.disconnect()
context.manager.connect()
let resubscribed = await waitUntil {
let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
return connections.count == 2 &&
connections.last?.sentStrings.contains { $0.contains("restore-sub") } == true
}
XCTAssertTrue(resubscribed)
}
func test_subscriptionSendFailure_retriesOnReconnect() async {
let relayURL = "wss://flaky-send.example"
let context = makeContext(permission: .denied)
context.sessionFactory.sendErrorByURL[relayURL] = NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
context.manager.subscribe(filter: makeFilter(), id: "flaky-sub", relayUrls: [relayURL], handler: { _ in })
let attempted = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL)?.sentStrings.isEmpty == false
}
XCTAssertTrue(attempted)
// The REQ send failed; the subscription must survive for the next connection.
context.sessionFactory.sendErrorByURL[relayURL] = nil
context.sessionFactory.latestConnection(for: relayURL)?.fail(
error: NSError(domain: NSURLErrorDomain, code: NSURLErrorNetworkConnectionLost)
)
let retryScheduled = await waitUntil { !context.scheduler.scheduled.isEmpty }
XCTAssertTrue(retryScheduled)
context.scheduler.runNext()
let resubscribed = await waitUntil {
let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
return connections.count == 2 &&
connections.last?.sentStrings.contains { $0.contains("flaky-sub") } == true
}
XCTAssertTrue(resubscribed)
}
func test_staleSendCompletionFromDeadSocket_doesNotBlockReplayOnNextConnection() async {
let relayURL = "wss://stale-completion.example"
let context = makeContext(permission: .denied)
context.manager.subscribe(filter: makeFilter(), id: "stale-sub", relayUrls: [relayURL], handler: { _ in })
// The connection exists synchronously; its REQ flush lands on a later
// main-queue tick, so deferring completions here is race-free.
let connectionA = context.sessionFactory.latestConnection(for: relayURL)
XCTAssertNotNil(connectionA)
connectionA?.deferSendCompletions = true
let reqSent = await waitUntil {
connectionA?.sentStrings.contains { $0.contains("stale-sub") } == true
}
XCTAssertTrue(reqSent)
// Socket dies while the REQ's send completion is still in flight.
connectionA?.fail(error: NSError(domain: NSURLErrorDomain, code: NSURLErrorNetworkConnectionLost))
let disconnected = await waitUntil {
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == false
}
XCTAssertTrue(disconnected)
// The stale success completion must not mark the subscription active.
connectionA?.flushDeferredSendCompletions()
try? await Task.sleep(nanoseconds: 20_000_000)
context.scheduler.runNext()
let replayed = await waitUntil {
let connections = context.sessionFactory.connectionsByURL[relayURL] ?? []
return connections.count == 2 &&
connections.last?.sentStrings.contains { $0.contains("stale-sub") } == true
}
XCTAssertTrue(replayed)
}
func test_permanentFailure_decaysAfterCooldownAndRetries() async {
let relayURL = "wss://cooldown.example"
let context = makeContext(permission: .denied)
context.sessionFactory.pingErrorByURL[relayURL] = NSError(
domain: NSURLErrorDomain,
code: NSURLErrorCannotFindHost,
userInfo: [NSLocalizedDescriptionKey: "DNS failure"]
)
context.manager.ensureConnections(to: [relayURL])
let failed = await waitUntil {
context.manager.relays.first(where: { $0.url == relayURL })?.reconnectAttempts == TransportConfig.nostrRelayMaxReconnectAttempts
}
XCTAssertTrue(failed)
// Within the cooldown the relay is skipped.
let countBefore = context.sessionFactory.requestedURLs.count
context.manager.ensureConnections(to: [relayURL])
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(context.sessionFactory.requestedURLs.count, countBefore)
// After the cooldown it gets another chance and recovers.
context.sessionFactory.pingErrorByURL[relayURL] = nil
context.clock.now = context.clock.now.addingTimeInterval(TransportConfig.nostrRelayFailureCooldownSeconds + 1)
context.manager.ensureConnections(to: [relayURL])
let retried = await waitUntil {
context.sessionFactory.requestedURLs.count == countBefore + 1 &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(retried)
}
func test_receiveFailure_schedulesReconnectWithBackoff() async {
let relayURL = "wss://retry.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relayURL])
let firstConnected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL) != nil
}
XCTAssertTrue(firstConnected)
let firstConnection = context.sessionFactory.latestConnection(for: relayURL)
firstConnection?.fail(error: NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut))
let retryScheduled = await waitUntil {
context.scheduler.scheduled.count == 1 &&
context.manager.relays.first(where: { $0.url == relayURL })?.reconnectAttempts == 1
}
XCTAssertTrue(retryScheduled)
XCTAssertEqual(context.scheduler.scheduled.first?.delay, TransportConfig.nostrRelayInitialBackoffSeconds)
let initialRequestCount = context.sessionFactory.requestedURLs.count
context.scheduler.runNext()
let retried = await waitUntil {
context.sessionFactory.requestedURLs.count == initialRequestCount + 1
}
XCTAssertTrue(retried)
}
func test_receiveFailure_whenActivationBecomesDisallowedDoesNotScheduleReconnect() async {
let relayURL = "wss://no-retry.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relayURL])
let connected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL) != nil &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(connected)
context.activationAllowed.value = false
context.sessionFactory.latestConnection(for: relayURL)?.fail(
error: NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
)
let disconnected = await waitUntil {
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == false
}
XCTAssertTrue(disconnected)
XCTAssertTrue(context.scheduler.scheduled.isEmpty)
XCTAssertEqual(context.sessionFactory.requestedURLs.count, 1)
}
func test_disconnect_invalidatesScheduledReconnectGeneration() async {
let relayURL = "wss://disconnect.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relayURL])
let firstConnected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL) != nil
}
XCTAssertTrue(firstConnected)
context.sessionFactory.latestConnection(for: relayURL)?.fail(
error: NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
)
let retryScheduled = await waitUntil { context.scheduler.scheduled.count == 1 }
XCTAssertTrue(retryScheduled)
let requestCountBeforeDisconnect = context.sessionFactory.requestedURLs.count
context.manager.disconnect()
context.scheduler.runNext()
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(context.sessionFactory.requestedURLs.count, requestCountBeforeDisconnect)
}
func test_retryConnection_cancelsActiveConnectionBeforeReconnecting() async {
let relayURL = "wss://retry-now.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relayURL])
let connected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL) != nil &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(connected)
guard let firstConnection = context.sessionFactory.latestConnection(for: relayURL) else {
XCTFail("Expected initial connection")
return
}
let initialRequestCount = context.sessionFactory.requestedURLs.count
context.manager.retryConnection(to: relayURL)
let reconnected = await waitUntil {
guard let latest = context.sessionFactory.latestConnection(for: relayURL) else { return false }
return context.sessionFactory.requestedURLs.count == initialRequestCount + 1 &&
latest !== firstConnection
}
XCTAssertTrue(reconnected)
XCTAssertEqual(firstConnection.cancelCallCount, 1)
}
func test_staleSocketFailureCannotFailConfirmedSendOnReplacementConnection() async throws {
let relayURL = "wss://retry-stale-confirmation.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relayURL])
let initiallyConnected = await waitUntil {
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(initiallyConnected)
let oldConnection = try XCTUnwrap(context.sessionFactory.latestConnection(for: relayURL))
context.manager.retryConnection(to: relayURL)
let replaced = await waitUntil {
guard let current = context.sessionFactory.latestConnection(for: relayURL) else { return false }
return current !== oldConnection &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(replaced)
let currentConnection = try XCTUnwrap(context.sessionFactory.latestConnection(for: relayURL))
let event = try makeSignedEvent(content: "replacement confirmation")
var results: [Bool] = []
context.manager.sendEventImmediately(event, to: [relayURL]) { results.append($0) }
oldConnection.fail(error: NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut))
try? await Task.sleep(nanoseconds: 30_000_000)
XCTAssertTrue(results.isEmpty)
XCTAssertTrue(context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true)
try currentConnection.emitOK(eventID: event.id, success: true, reason: "accepted")
let confirmed = await waitUntil { results == [true] }
XCTAssertTrue(confirmed)
}
func test_retryConnection_whenTorReadinessFailsDoesNotReconnect() async {
let relayURL = "wss://retry-tor.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: true)
context.manager.ensureConnections(to: [relayURL])
let connected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL) != nil &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(connected)
guard let firstConnection = context.sessionFactory.latestConnection(for: relayURL) else {
XCTFail("Expected initial connection")
return
}
let initialRequestCount = context.sessionFactory.requestedURLs.count
context.torWaiter.isReady = false
context.manager.retryConnection(to: relayURL)
XCTAssertEqual(firstConnection.cancelCallCount, 1)
XCTAssertEqual(context.sessionFactory.requestedURLs.count, initialRequestCount)
context.torWaiter.resolve(false)
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertEqual(context.sessionFactory.requestedURLs.count, initialRequestCount)
}
func test_resetAllConnections_clearsRelayStateAndReconnects() async {
let relayURL = "wss://reset.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relayURL])
let connected = await waitUntil {
context.sessionFactory.latestConnection(for: relayURL) != nil &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(connected)
context.sessionFactory.latestConnection(for: relayURL)?.fail(
error: NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
)
let failed = await waitUntil {
context.manager.relays.first(where: { $0.url == relayURL })?.reconnectAttempts == 1 &&
context.manager.relays.first(where: { $0.url == relayURL })?.lastError != nil
}
XCTAssertTrue(failed)
let requestCountBeforeReset = context.sessionFactory.requestedURLs.count
context.manager.resetAllConnections()
let reset = await waitUntil {
context.sessionFactory.requestedURLs.count == requestCountBeforeReset + 1 &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true &&
context.manager.relays.first(where: { $0.url == relayURL })?.reconnectAttempts == 0 &&
context.manager.relays.first(where: { $0.url == relayURL })?.nextReconnectTime == nil &&
context.manager.relays.first(where: { $0.url == relayURL })?.lastError == nil
}
XCTAssertTrue(reset)
}
func test_debugFlushMessageQueue_flushesAllConnectedRelays() async throws {
let relayOne = "wss://flush-one.example"
let relayTwo = "wss://flush-two.example"
let context = makeContext(
permission: .denied,
userTorEnabled: true,
torEnforced: true,
torIsReady: true,
torIsForeground: false
)
let event = try makeSignedEvent(content: "flush-all")
context.manager.sendEvent(event, to: [relayOne, relayTwo])
let queued = await waitUntil {
context.manager.debugPendingMessageQueueCount == 1
}
XCTAssertTrue(queued)
context.torForeground.value = true
context.manager.ensureConnections(to: [relayOne, relayTwo])
context.manager.debugFlushMessageQueue()
let flushed = await waitUntil {
context.manager.debugPendingMessageQueueCount == 0 &&
context.sessionFactory.latestConnection(for: relayOne)?.sentStrings.count == 1 &&
context.sessionFactory.latestConnection(for: relayTwo)?.sentStrings.count == 1
}
XCTAssertTrue(flushed)
}
func test_dnsPingFailure_marksRelayPermanentCallsEOSEImmediatelyAndManualRetryReconnects() async {
let relayURL = "wss://dns-failure.example"
let context = makeContext(permission: .denied)
context.sessionFactory.pingErrorByURL[relayURL] = NSError(
domain: NSURLErrorDomain,
code: NSURLErrorCannotFindHost,
userInfo: [NSLocalizedDescriptionKey: "DNS failure"]
)
context.manager.subscribe(filter: makeFilter(), id: "dns-sub", relayUrls: [relayURL], handler: { _ in })
let permanentlyFailed = await waitUntil {
context.manager.relays.first(where: { $0.url == relayURL })?.reconnectAttempts == TransportConfig.nostrRelayMaxReconnectAttempts &&
context.scheduler.scheduled.isEmpty
}
XCTAssertTrue(permanentlyFailed)
var immediateEOSE = 0
context.manager.subscribe(
filter: makeFilter(),
id: "dns-eose",
relayUrls: [relayURL],
handler: { _ in },
onEOSE: { immediateEOSE += 1 }
)
XCTAssertEqual(immediateEOSE, 1)
context.sessionFactory.pingErrorByURL[relayURL] = nil
let requestCountBeforeRetry = context.sessionFactory.requestedURLs.count
context.manager.retryConnection(to: relayURL)
let reconnected = await waitUntil {
context.sessionFactory.requestedURLs.count == requestCountBeforeRetry + 1 &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true &&
context.manager.relays.first(where: { $0.url == relayURL })?.reconnectAttempts == 0
}
XCTAssertTrue(reconnected)
}
func test_pendingSubscriptions_perRelayCapEvictsOldestByInsertionOrder() async {
let relayURL = "wss://pending-cap.example"
// Tor stalled: nothing flushes, so every REQ stays pending.
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
let cap = TransportConfig.nostrPendingSubscriptionsPerRelayCap
for i in 0..<(cap + 3) {
context.manager.subscribe(filter: makeFilter(), id: "cap-sub-\(i)", relayUrls: [relayURL], handler: { _ in })
}
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), cap)
let pendingIDs = context.manager.debugPendingSubscriptionIDs(for: relayURL)
// The three oldest entries were evicted; the newest survive.
for i in 0..<3 {
XCTAssertFalse(pendingIDs.contains("cap-sub-\(i)"), "expected cap-sub-\(i) to be evicted")
}
for i in 3..<(cap + 3) {
XCTAssertTrue(pendingIDs.contains("cap-sub-\(i)"), "expected cap-sub-\(i) to be retained")
}
}
func test_pendingSubscriptions_staleEntriesSweptOnConnectAttempt() async {
let relayURL = "wss://pending-sweep.example"
// Tor stalled: the REQ stays pending and no socket ever opens.
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
context.manager.subscribe(filter: makeFilter(), id: "stale-pending-sub", relayUrls: [relayURL], handler: { _ in })
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 1)
// Just under the TTL the entry survives a connect attempt.
context.clock.now = context.clock.now.addingTimeInterval(TransportConfig.nostrPendingSubscriptionTTLSeconds - 1)
context.manager.ensureConnections(to: [relayURL])
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 1)
// Past the TTL the next connect attempt sweeps it.
context.clock.now = context.clock.now.addingTimeInterval(2)
context.manager.ensureConnections(to: [relayURL])
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 0)
}
func test_resetForPanicWipe_dropsSessionRelayStateWithoutFiringCallbacks() async throws {
let relayURL = "wss://panic-reset.example"
let context = makeContext(permission: .denied, userTorEnabled: true, torEnforced: true, torIsReady: false)
let event = try makeSignedEvent(content: "queued before panic")
var handledEvents = 0
var eoseCount = 0
context.manager.subscribe(
filter: makeFilter(),
id: "panic-sub",
relayUrls: [relayURL],
handler: { _ in handledEvents += 1 },
onEOSE: { eoseCount += 1 }
)
context.manager.sendEvent(event, to: [relayURL])
XCTAssertEqual(context.manager.debugMessageHandlerCount, 1)
XCTAssertEqual(context.manager.debugSubscriptionRequestCount, 1)
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 1)
XCTAssertEqual(context.manager.debugPendingEOSECallbackCount, 1)
XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 1)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
context.manager.resetForPanicWipe()
XCTAssertEqual(context.manager.debugMessageHandlerCount, 0)
XCTAssertEqual(context.manager.debugSubscriptionRequestCount, 0)
XCTAssertEqual(context.manager.debugPendingSubscriptionCount(for: relayURL), 0)
XCTAssertEqual(context.manager.debugPendingEOSECallbackCount, 0)
XCTAssertEqual(context.manager.debugPendingMessageQueueCount, 0)
XCTAssertEqual(handledEvents, 0)
XCTAssertEqual(eoseCount, 0)
// Stale Tor wait and fallback callbacks from the pre-wipe generation
// must not resurrect connections or settle callbacks after reset.
context.torWaiter.resolve(true)
context.scheduler.runNext()
try? await Task.sleep(nanoseconds: 20_000_000)
XCTAssertTrue(context.sessionFactory.requestedURLs.isEmpty)
XCTAssertEqual(eoseCount, 0)
}
func test_resetForPanicWipe_marksConnectedRelaysDisconnected() async {
let relayURL = "wss://panic-connected.example"
let context = makeContext(permission: .denied)
context.manager.ensureConnections(to: [relayURL])
let connected = await waitUntil {
context.manager.isConnected &&
context.manager.relays.first(where: { $0.url == relayURL })?.isConnected == true
}
XCTAssertTrue(connected)
context.manager.resetForPanicWipe()
XCTAssertFalse(context.manager.isConnected)
XCTAssertEqual(context.manager.relays.first(where: { $0.url == relayURL })?.isConnected, false)
XCTAssertEqual(context.manager.relays.first(where: { $0.url == relayURL })?.reconnectAttempts, 0)
XCTAssertNil(context.manager.relays.first(where: { $0.url == relayURL })?.lastError)
}
func test_reconnectBackoff_appliesJitterWithinConfiguredBounds() async {
let relayURL = "wss://jitter-bounds.example"
// Pin the jitter source to the extremes and the midpoint of [0, 1).
let jitter = JitterSequence([0.0, 1.0.nextDown, 0.25])
let context = makeContext(permission: .denied, jitterUnit: { jitter.next() })
// Persistent ping failure: every connect attempt fails and schedules
// the next reconnect with an increasing attempt count.
context.sessionFactory.pingErrorByURL[relayURL] = NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
context.manager.ensureConnections(to: [relayURL])
var delays: [TimeInterval] = []
for attempt in 1...3 {
let scheduled = await waitUntil { context.scheduler.scheduled.count == 1 }
XCTAssertTrue(scheduled, "reconnect for attempt \(attempt) was not scheduled")
delays.append(context.scheduler.scheduled[0].delay)
context.scheduler.runNext()
}
// Bases: 1s, 2s, 4s. Jitter factors: 0.8, ~1.2, 0.9.
XCTAssertEqual(delays[0], 0.8 * TransportConfig.nostrRelayInitialBackoffSeconds, accuracy: 1e-9)
XCTAssertEqual(delays[1], 1.2 * TransportConfig.nostrRelayInitialBackoffSeconds * TransportConfig.nostrRelayBackoffMultiplier, accuracy: 1e-6)
XCTAssertEqual(delays[2], 0.9 * TransportConfig.nostrRelayInitialBackoffSeconds * pow(TransportConfig.nostrRelayBackoffMultiplier, 2), accuracy: 1e-9)
}
func test_reconnectBackoff_realRandomJitterStaysInBoundsAndVaries() async {
let relayURL = "wss://jitter-random.example"
let context = makeContext(permission: .denied, jitterUnit: { Double.random(in: 0..<1) })
context.sessionFactory.pingErrorByURL[relayURL] = NSError(domain: NSURLErrorDomain, code: NSURLErrorTimedOut)
context.manager.ensureConnections(to: [relayURL])
var factors: [Double] = []
for attempt in 1...5 {
let scheduled = await waitUntil { context.scheduler.scheduled.count == 1 }
XCTAssertTrue(scheduled, "reconnect for attempt \(attempt) was not scheduled")
let base = min(
TransportConfig.nostrRelayInitialBackoffSeconds * pow(TransportConfig.nostrRelayBackoffMultiplier, Double(attempt - 1)),
TransportConfig.nostrRelayMaxBackoffSeconds
)
let factor = context.scheduler.scheduled[0].delay / base
XCTAssertGreaterThanOrEqual(factor, 1.0 - TransportConfig.nostrRelayBackoffJitterRatio)
XCTAssertLessThan(factor, 1.0 + TransportConfig.nostrRelayBackoffJitterRatio)
factors.append(factor)
context.scheduler.runNext()
}
// A real RNG must not produce a constant delay across attempts
// (5 identical uniform doubles is probability ~0).
XCTAssertGreaterThan(Set(factors).count, 1)
}
// MARK: - Hand-added relays
/// Adding a relay has to take effect without a restart: the whole point is
/// recovering reachability when the built-in hostnames are blocked.
@MainActor
func testAddedRelayJoinsTheTargetSetOnSettingsChange() async {
let center = NotificationCenter()
let custom = MutableRelayList(urls: [])
let context = makeContext(
permission: .authorized,
notificationCenter: center,
customRelays: custom
)
XCTAssertFalse(context.manager.relays.contains { $0.url == "wss://added.example.com" })
custom.urls = ["wss://added.example.com"]
center.post(name: NostrRelaySettings.didChangeNotification, object: nil)
let joined = await waitUntil {
context.manager.relays.contains { $0.url == "wss://added.example.com" }
}
XCTAssertTrue(joined)
}
/// Removing a relay must actually close it. The teardown path iterates the
/// current target list, and a removed relay is no longer in it, so without
/// an explicit reconcile against the previous set its socket and queued
/// sends would linger.
@MainActor
func testRemovedRelayLeavesTheTargetSet() async {
let center = NotificationCenter()
let custom = MutableRelayList(urls: ["wss://added.example.com"])
let context = makeContext(
permission: .authorized,
notificationCenter: center,
customRelays: custom
)
XCTAssertTrue(context.manager.relays.contains { $0.url == "wss://added.example.com" })
custom.urls = []
center.post(name: NostrRelaySettings.didChangeNotification, object: nil)
let dropped = await waitUntil {
!context.manager.relays.contains { $0.url == "wss://added.example.com" }
}
XCTAssertTrue(dropped)
// The built-in relays are untouched by a custom-relay removal.
XCTAssertTrue(context.manager.relays.contains { $0.url == "wss://nos.lol" })
}
private func makeContext(
permission: LocationChannelManager.PermissionState,
favorites: Set<Data> = [],
activationAllowed: Bool = true,
userTorEnabled: Bool = false,
torEnforced: Bool = false,
torIsReady: Bool = true,
torIsForeground: Bool = true,
notificationCenter: NotificationCenter = NotificationCenter(),
customRelays: MutableRelayList = MutableRelayList(urls: []),
jitterUnit: @escaping () -> Double = { 0.5 } // 0.5 -> jitter factor 1.0 (no jitter)
) -> RelayManagerTestContext {
let permissionSubject = CurrentValueSubject<LocationChannelManager.PermissionState, Never>(permission)
let favoritesSubject = CurrentValueSubject<Set<Data>, Never>(favorites)
let sessionFactory = MockRelaySessionFactory()
let scheduler = MockRelayScheduler()
let clock = MutableClock(now: Date(timeIntervalSince1970: 1_700_000_000))
let torWaiter = MockTorWaiter(isReady: torIsReady)
let torForeground = MutableBool(value: torIsForeground)
let activationFlag = MutableBool(value: activationAllowed)
let manager = NostrRelayManager(
dependencies: NostrRelayManagerDependencies(
activationAllowed: { activationFlag.value },
userTorEnabled: { userTorEnabled },
hasMutualFavorites: { !favoritesSubject.value.isEmpty },
hasLocationPermission: { permissionSubject.value == .authorized },
mutualFavoritesPublisher: favoritesSubject.eraseToAnyPublisher(),
locationPermissionPublisher: permissionSubject.eraseToAnyPublisher(),
torEnforced: { torEnforced },
torIsReady: { torWaiter.isReady },
torIsForeground: { torForeground.value },
awaitTorReady: torWaiter.await(completion:),
makeSession: { sessionFactory },
scheduleAfter: { delay, action in
scheduler.schedule(delay: delay, action: action)
},
now: { clock.now },
jitterUnit: jitterUnit,
notificationCenter: notificationCenter,
customRelays: { customRelays.urls }
)
)
return RelayManagerTestContext(
manager: manager,
permissionSubject: permissionSubject,
sessionFactory: sessionFactory,
scheduler: scheduler,
clock: clock,
activationAllowed: activationFlag,
torWaiter: torWaiter,
torForeground: torForeground
)
}
private func makeFilter() -> NostrFilter {
var filter = NostrFilter()
filter.kinds = [NostrProtocol.EventKind.textNote.rawValue]
filter.limit = 10
return filter
}
private func makeSignedEvent(content: String) throws -> NostrEvent {
let identity = try NostrIdentity.generate()
let event = NostrEvent(
pubkey: identity.publicKeyHex,
createdAt: Date(),
kind: .textNote,
tags: [],
content: content
)
return try event.sign(with: identity.schnorrSigningKey())
}
private func invalidSignatureCopy(of event: NostrEvent) -> NostrEvent {
var invalid = event
invalid.sig = String(repeating: "0", count: 128)
return invalid
}
private func waitUntil(
timeout: TimeInterval = TestConstants.settleTimeout,
condition: @escaping @MainActor () -> Bool
) async -> Bool {
let deadline = Date().addingTimeInterval(timeout)
while Date() < deadline {
if condition() {
return true
}
try? await Task.sleep(nanoseconds: 10_000_000)
}
return condition()
}
}
@MainActor
private struct RelayManagerTestContext {
let manager: NostrRelayManager
let permissionSubject: CurrentValueSubject<LocationChannelManager.PermissionState, Never>
let sessionFactory: MockRelaySessionFactory
let scheduler: MockRelayScheduler
let clock: MutableClock
let activationAllowed: MutableBool
let torWaiter: MockTorWaiter
let torForeground: MutableBool
}
private final class MutableClock {
var now: Date
init(now: Date) {
self.now = now
}
}
/// Stand-in for the persisted hand-added relay list, so tests can change it
/// without writing to shared preferences.
private final class MutableRelayList {
var urls: [String]
init(urls: [String]) {
self.urls = urls
}
}
/// Deterministic jitter source: returns the queued values in order, then a
/// neutral 0.5 (jitter factor 1.0) once exhausted.
private final class JitterSequence {
private var values: [Double]
init(_ values: [Double]) {
self.values = values
}
func next() -> Double {
values.isEmpty ? 0.5 : values.removeFirst()
}
}
private final class MutableBool {
var value: Bool
init(value: Bool) {
self.value = value
}
}
private final class MockTorWaiter {
private var completions: [(Bool) -> Void] = []
private(set) var awaitCallCount = 0
var isReady: Bool
init(isReady: Bool) {
self.isReady = isReady
}
func await(completion: @escaping (Bool) -> Void) {
awaitCallCount += 1
completions.append(completion)
}
func resolve(_ ready: Bool) {
isReady = ready
let pending = completions
completions.removeAll()
pending.forEach { $0(ready) }
}
}
private final class MockRelayScheduler: @unchecked Sendable {
struct ScheduledAction {
let delay: TimeInterval
let action: @Sendable () -> Void
}
private(set) var scheduled: [ScheduledAction] = []
func schedule(delay: TimeInterval, action: @escaping @Sendable () -> Void) {
scheduled.append(ScheduledAction(delay: delay, action: action))
}
func runNext() {
guard !scheduled.isEmpty else { return }
let next = scheduled.removeFirst()
next.action()
}
}
private final class MockRelaySessionFactory: NostrRelaySessionProtocol {
private(set) var requestedURLs: [String] = []
private(set) var connectionsByURL: [String: [MockRelayConnection]] = [:]
var pingErrorByURL: [String: Error?] = [:]
var sendErrorByURL: [String: Error?] = [:]
var allConnections: [MockRelayConnection] {
connectionsByURL.values.flatMap { $0 }
}
func webSocketTask(with url: URL) -> NostrRelayConnectionProtocol {
requestedURLs.append(url.absoluteString)
let connection = MockRelayConnection(
url: url.absoluteString,
pingError: pingErrorByURL[url.absoluteString] ?? nil,
sendError: sendErrorByURL[url.absoluteString] ?? nil
)
connectionsByURL[url.absoluteString, default: []].append(connection)
return connection
}
func latestConnection(for url: String) -> MockRelayConnection? {
connectionsByURL[url]?.last
}
}
private final class MockRelayConnection: NostrRelayConnectionProtocol {
private let pingError: Error?
var sendError: Error?
private var receiveHandler: ((Result<URLSessionWebSocketTask.Message, Error>) -> Void)?
private(set) var resumeCallCount = 0
private(set) var cancelCallCount = 0
private(set) var sentMessages: [URLSessionWebSocketTask.Message] = []
var sentStrings: [String] {
sentMessages.compactMap {
switch $0 {
case .string(let string): string
case .data(let data): String(data: data, encoding: .utf8)
@unknown default: nil
}
}
}
init(url _: String, pingError: Error? = nil, sendError: Error? = nil) {
self.pingError = pingError
self.sendError = sendError
}
func resume() {
resumeCallCount += 1
}
func cancel(with closeCode: URLSessionWebSocketTask.CloseCode, reason: Data?) {
cancelCallCount += 1
}
var deferSendCompletions = false
private var deferredSendCompletions: [(Error?) -> Void] = []
func send(_ message: URLSessionWebSocketTask.Message, completionHandler: @escaping (Error?) -> Void) {
sentMessages.append(message)
if deferSendCompletions {
deferredSendCompletions.append(completionHandler)
} else {
completionHandler(sendError)
}
}
func flushDeferredSendCompletions() {
let pending = deferredSendCompletions
deferredSendCompletions = []
pending.forEach {
$0(sendError)
}
}
func receive(completionHandler: @escaping (Result<URLSessionWebSocketTask.Message, Error>) -> Void) {
if !pendingResults.isEmpty {
completionHandler(pendingResults.removeFirst())
} else {
receiveHandler = completionHandler
}
}
func sendPing(pongReceiveHandler: @escaping (Error?) -> Void) {
pongReceiveHandler(pingError)
}
func fail(error: Error) {
let handler = receiveHandler
receiveHandler = nil
handler?(.failure(error))
}
func emitEventMessage(subscriptionID: String, event: NostrEvent) throws {
let eventData = try JSONEncoder().encode(event)
let eventJSONObject = try JSONSerialization.jsonObject(with: eventData) as! [String: Any]
let payload: [Any] = ["EVENT", subscriptionID, eventJSONObject]
try emit(jsonObject: payload)
}
func emitEOSE(subscriptionID: String) throws {
try emit(jsonObject: ["EOSE", subscriptionID])
}
func emitOK(eventID: String, success: Bool, reason: String) throws {
try emit(jsonObject: ["OK", eventID, success, reason])
}
func emitNotice(message: String) throws {
try emit(jsonObject: ["NOTICE", message])
}
func emitRawString(_ string: String) throws {
deliver(.success(.string(string)))
}
private func emit(jsonObject: Any) throws {
let data = try JSONSerialization.data(withJSONObject: jsonObject)
deliver(.success(.data(data)))
}
// Frames emitted before the manager re-arms `receive` are queued so
// back-to-back emissions model a socket with several buffered frames.
private var pendingResults: [Result<URLSessionWebSocketTask.Message, Error>] = []
private func deliver(_ result: Result<URLSessionWebSocketTask.Message, Error>) {
if let handler = receiveHandler {
receiveHandler = nil
handler(result)
} else {
pendingResults.append(result)
}
}
}
Reference in New Issue View Git Blame Copy Permalink