diff --git a/apps/swift-ios/App/NativeFeatureClient.swift b/apps/swift-ios/App/NativeFeatureClient.swift index a011101359bc..588574fb6689 100644 --- a/apps/swift-ios/App/NativeFeatureClient.swift +++ b/apps/swift-ios/App/NativeFeatureClient.swift @@ -53,6 +53,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, private let aggregateIdleRefreshInterval: Duration private let aggregateFailureRefreshInterval: Duration private let aggregateRefreshSleep: @Sendable (Duration) async throws -> Void + private let aggregatePeerRefreshSleep: @Sendable (String, Duration) async throws -> Void private let environmentShellTimeoutInterval: TimeInterval private let threadSnapshotTimeoutInterval: TimeInterval private let catchUpDelay: @Sendable () async throws -> Void @@ -120,6 +121,9 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, private var configurationTask: Task? private var aggregateRefreshTask: Task? private var aggregateRefreshID: UUID? + private(set) var aggregateRefreshWorkers: [ + String: (environment: Environment, task: Task) + ] = [:] private var shellPublishTask: Task? private var archivedRefreshTask: Task? private var detailRefreshTask: Task? @@ -161,6 +165,9 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, aggregateRefreshSleep: @escaping @Sendable (Duration) async throws -> Void = { try await Task.sleep(for: $0) }, + aggregatePeerRefreshSleep: @escaping @Sendable (String, Duration) async throws -> Void = { _, interval in + try await Task.sleep(for: interval) + }, environmentShellTimeoutInterval: TimeInterval = 6, threadSnapshotTimeoutInterval: TimeInterval = 8, catchUpDelay: @escaping @Sendable () async throws -> Void = { @@ -200,6 +207,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, self.aggregateRefreshInterval = aggregateRefreshInterval self.aggregateIdleRefreshInterval = aggregateIdleRefreshInterval self.aggregateFailureRefreshInterval = aggregateFailureRefreshInterval + self.aggregatePeerRefreshSleep = aggregatePeerRefreshSleep self.aggregateRefreshSleep = aggregateRefreshSleep self.environmentShellTimeoutInterval = environmentShellTimeoutInterval self.threadSnapshotTimeoutInterval = threadSnapshotTimeoutInterval @@ -216,6 +224,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, fallbackPollingTask?.cancel() configurationTask?.cancel() aggregateRefreshTask?.cancel() + aggregateRefreshWorkers.values.forEach { $0.task.cancel() } shellPublishTask?.cancel() archivedRefreshTask?.cancel() detailRefreshTask?.cancel() @@ -263,6 +272,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, connectionDetail: environmentConnectionDetails[environment.id] ) latestSnapshot = snapshot + startAggregateRefresh(activeClient) return snapshot } @@ -341,6 +351,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, pairedClient = try await runtime.pair(url: endpoint, clientLabel: "T3 Code Swift") } await adoptEnvironment(pairedClient.environment, client: pairedClient) + startAggregateRefresh(pairedClient) startPolling(pairedClient) } @@ -417,6 +428,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, ) publish(snapshot) } + startAggregateRefresh(managedClient) startPolling(managedClient) } @@ -470,6 +482,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, func setEnvironmentEnabled(id: String, enabled: Bool) async throws { try await runtime.setEnabled(id: id, enabled: enabled) if !enabled { + aggregateRefreshWorkers.removeValue(forKey: id)?.task.cancel() environmentConnectionStates[id] = .disconnected environmentConnectionDetails[id] = nil environmentClients[id] = nil @@ -490,6 +503,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, } try await runtime.remove(id: id) saveGitHubRoutingGrants(gitHubRoutingGrants.filter { $0.environmentID != id }) + aggregateRefreshWorkers.removeValue(forKey: id)?.task.cancel() if removesActiveEnvironment { await clearActiveEnvironment(disconnectClient: false) } @@ -971,18 +985,17 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, _ environment: Environment, client newClient: T3Client ) async { + cancelAggregateRefresh() if activeEnvironment?.id == environment.id, client === newClient { activeEnvironment = environment environmentClients[environment.id] = newClient latestShell = shellsByEnvironmentID[environment.id] - startAggregateRefresh(newClient) return } let previousClient = client pollingTask?.cancel() fallbackPollingTask?.cancel() configurationTask?.cancel() - aggregateRefreshTask?.cancel() archivedRefreshTask?.cancel() pollingTask = nil fallbackPollingTask = nil @@ -998,7 +1011,6 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, if let previousClient, previousClient !== newClient { await previousClient.disconnect() } - startAggregateRefresh(newClient) } private func clearActiveEnvironment(disconnectClient: Bool = true) async { @@ -1006,7 +1018,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, pollingTask?.cancel() fallbackPollingTask?.cancel() configurationTask?.cancel() - aggregateRefreshTask?.cancel() + cancelAggregateRefresh() archivedRefreshTask?.cancel() pollingTask = nil fallbackPollingTask = nil @@ -4000,122 +4012,210 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, /// Non-active environments do not hold WebSocket subscriptions. A quiet /// HTTP refresh keeps their home rows and reachability useful without /// multiplying live streams or creating a high-frequency battery cost. - private func startAggregateRefresh(_ activeClient: T3Client) { + private func cancelAggregateRefresh() { aggregateRefreshTask?.cancel() - let generation = environmentGeneration - let refreshID = UUID() - let fastInterval = aggregateRefreshInterval - let idleInterval = aggregateIdleRefreshInterval - let failureInterval = aggregateFailureRefreshInterval - let sleep = aggregateRefreshSleep - let loadEnvironments = aggregateEnvironmentLoader - aggregateRefreshID = refreshID - aggregateRefreshTask = Task { [weak self] in - var nextInterval = fastInterval - var failureBackoffs: [String: Duration] = [:] - while !Task.isCancelled { - let elapsedInterval = nextInterval + aggregateRefreshWorkers.values.forEach { $0.task.cancel() } + aggregateRefreshWorkers.removeAll() + aggregateRefreshID = nil + } + + private func startAggregateRefresh(_ activeClient: T3Client) { + cancelAggregateRefresh() + let context = AggregateRefreshContext(owner: self, activeClient: activeClient) + aggregateRefreshID = context.refreshID + aggregateRefreshTask = Task { + defer { context.cancelWorkers() } + while context.isCurrent { + var interval = context.fastInterval do { - try await sleep(nextInterval) - } catch { + let environments = try await context.loadEnvironments(context.runtime) + guard context.isCurrent else { return } + interval = context.reconcileWorkers(environments) + } catch is CancellationError where Task.isCancelled { return + } catch { + interval = context.failureInterval } - guard let self, - self.aggregateRefreshID == refreshID, - self.isCurrentSession( - client: activeClient, - generation: generation - ), - let activeEnvironment = self.activeEnvironment else { - return + do { try await context.sleep(interval) } catch { return } + } + } + } + + /// A worker retains runtime inputs, never its UI owner across a suspension. + /// This lets deinit cancel the exact outstanding shell/catalogue tasks. + @MainActor + private final class AggregateRefreshContext { + weak var owner: NativeFeatureClient? + let runtime: EnvironmentRuntime + let activeClient: T3Client + let generation: Int + let refreshID = UUID() + let fastInterval: Duration + let idleInterval: Duration + let failureInterval: Duration + let shellTimeout: TimeInterval + let sleep: @Sendable (Duration) async throws -> Void + let peerSleep: @Sendable (String, Duration) async throws -> Void + let loadEnvironments: @Sendable (EnvironmentRuntime) async throws -> [Environment] + + init(owner: NativeFeatureClient, activeClient: T3Client) { + self.owner = owner + runtime = owner.runtime + self.activeClient = activeClient + generation = owner.environmentGeneration + fastInterval = owner.aggregateRefreshInterval + idleInterval = owner.aggregateIdleRefreshInterval + failureInterval = owner.aggregateFailureRefreshInterval + shellTimeout = owner.environmentShellTimeoutInterval + sleep = owner.aggregateRefreshSleep + peerSleep = owner.aggregatePeerRefreshSleep + loadEnvironments = owner.aggregateEnvironmentLoader + } + + var isCurrent: Bool { + !Task.isCancelled && owner?.aggregateRefreshID == refreshID + && owner?.isCurrentSession(client: activeClient, generation: generation) == true + } + + func cancelWorkers() { + guard let owner, owner.aggregateRefreshID == refreshID else { return } + owner.aggregateRefreshWorkers.values.forEach { $0.task.cancel() } + owner.aggregateRefreshWorkers.removeAll() + } + + func reconcileWorkers(_ environments: [Environment]) -> Duration { + guard let owner, isCurrent else { return idleInterval } + let passive = environments.filter { $0.isEnabled && $0.id != owner.activeEnvironment?.id } + let current = Dictionary(uniqueKeysWithValues: passive.map { ($0.id, $0) }) + for (id, worker) in owner.aggregateRefreshWorkers where current[id] != worker.environment { + worker.task.cancel() + owner.aggregateRefreshWorkers[id] = nil + } + // Prune once per topology check, not once per shell result. + owner.reconcileEnvironmentLoads([], savedEnvironments: environments) + if owner.latestSnapshot?.environments != environments.map({ + owner.mapEnvironment($0, activeID: owner.activeEnvironment?.id) + }) { + owner.publishAggregateSnapshot(environments) + } + for environment in passive where owner.aggregateRefreshWorkers[environment.id] == nil { + let task = Task { + await withTaskGroup(of: Void.self) { group in + group.addTask { await self.refreshShell(environment) } + group.addTask { await self.refreshCatalogue(environment) } + } } - let environments: [Environment] + owner.aggregateRefreshWorkers[environment.id] = (environment, task) + } + return passive.isEmpty ? idleInterval : fastInterval + } + + /// Revalidate membership after suspension, including endpoint edits. + /// Persistence errors retry; removed or disabled peers end their work. + func currentEnvironments(for environment: Environment) async throws -> [Environment]? { + guard isCurrent else { return nil } + let environments = try await runtime.environments() + guard isCurrent, environments.contains(environment), environment.isEnabled, + owner?.activeEnvironment?.id != environment.id else { return nil } + return environments + } + + func refreshShell(_ environment: Environment) async { + var interval = fastInterval + while isCurrent { do { - environments = try await loadEnvironments(self.runtime) + try await peerSleep(environment.id, interval) + guard try await currentEnvironments(for: environment) != nil else { return } + let client = await runtime.client(for: environment) + guard isCurrent else { return } + let shell: OrchestrationShellSnapshot? + do { + shell = try await client.shellSnapshot(timeoutInterval: shellTimeout) + } catch { + guard let environments = try await currentEnvironments(for: environment) else { return } + owner?.applyEnvironmentLoad(EnvironmentShellLoad( + environment: environment, client: client, shell: nil, config: nil, + credentialRejected: error.isRejectedAuthorization, + failureDetail: error.localizedDescription + )) + owner?.publishAggregateSnapshot(environments) + interval = error.isRejectedAuthorization ? .seconds(24 * 60 * 60) : failureInterval + continue + } + guard let environments = try await currentEnvironments(for: environment) else { return } + interval = applyShell(shell, client: client, environment: environment, saved: environments) } catch is CancellationError where Task.isCancelled { return } catch { - // Persistence can be briefly unavailable while another - // actor atomically replaces the environment document. - // Back off while keeping the loop alive for recovery. - nextInterval = failureInterval - continue - } - guard !Task.isCancelled, - self.aggregateRefreshID == refreshID, - self.isCurrentSession( - client: activeClient, - generation: generation - ) else { - return - } - let passiveEnvironments = environments.filter { - $0.isEnabled && $0.id != activeEnvironment.id - } - guard !passiveEnvironments.isEmpty else { - nextInterval = idleInterval - continue + interval = failureInterval } - let passiveIDs = Set(passiveEnvironments.map(\.id)) - failureBackoffs = failureBackoffs.reduce(into: [:]) { result, entry in - guard passiveIDs.contains(entry.key) else { return } - result[entry.key] = max(.zero, entry.value - elapsedInterval) - } - let refreshableEnvironments = passiveEnvironments.filter { - failureBackoffs[$0.id, default: .zero] <= .zero - } - guard !refreshableEnvironments.isEmpty else { - nextInterval = fastInterval + } + } + + func applyShell( + _ shell: OrchestrationShellSnapshot?, client: T3Client, + environment: Environment, saved: [Environment] + ) -> Duration { + guard let owner, isCurrent else { return failureInterval } + let previous = owner.shellsByEnvironmentID[environment.id] + let changed = shell.map { $0 != previous } ?? false + let membershipChanged = shell.map { + $0.projects.map(\.id) != previous?.projects.map(\.id) + || $0.threads.map(\.id) != previous?.threads.map(\.id) + } ?? false + owner.applyEnvironmentLoad(EnvironmentShellLoad( + environment: environment, client: client, shell: shell, config: nil + )) + if membershipChanged { owner.rebuildEntityIndexes(saved) } + owner.publishAggregateSnapshot(saved) + guard let shell else { return failureInterval } + return changed || NativeFeatureClient.shellNeedsFrequentAggregateRefresh(shell) + ? fastInterval : idleInterval + } + + func refreshCatalogue(_ environment: Environment) async { + while isCurrent && owner?.serverConfigsByEnvironmentID[environment.id] == nil { + // Each probe mints a WebSocket ticket, and a rejected pairing only + // surfaces through the shell refresh. Wait for a re-pair instead. + if owner?.environmentConnectionStates[environment.id] == .needsPairing { + do { try await peerSleep(environment.id, failureInterval) } catch { return } continue } - let loads = await self.loadEnvironmentShells(refreshableEnvironments) - guard !Task.isCancelled, - self.aggregateRefreshID == refreshID, - self.isCurrentSession( - client: activeClient, - generation: generation - ) else { - return - } - let shellsChanged = loads.contains { load in - guard let shell = load.shell else { return false } - return shell != self.shellsByEnvironmentID[load.environment.id] - } - let hasActiveWork = loads.contains { load in - load.shell.map(Self.shellNeedsFrequentAggregateRefresh) == true - } - for load in loads { - if load.credentialRejected { - // Only re-pairing changes this. Re-pairing replaces - // the runtime client and restarts this loop. - failureBackoffs[load.environment.id] = .seconds(24 * 60 * 60) - } else if load.shell == nil { - failureBackoffs[load.environment.id] = failureInterval - } else { - failureBackoffs[load.environment.id] = nil + do { + guard try await currentEnvironments(for: environment) != nil else { return } + // Never disconnect the shared client if this peer becomes selected. + let probe = await runtime.ephemeralClient(for: environment) + let config = try? await probe.serverConfig() + await probe.disconnect() + guard let environments = try await currentEnvironments(for: environment) else { return } + if let config { + applyCatalogue(config, environment: environment, saved: environments) + return } + } catch is CancellationError where Task.isCancelled { + return + } catch { + // Retry a transient environment-document read just like a failed probe. } - self.reconcileEnvironmentLoads(loads, savedEnvironments: environments) - let currentConnection = self.latestSnapshot?.connection - ?? FeatureConnection( - state: .disconnected, - environmentName: activeEnvironment.label, - endpoint: activeEnvironment.httpBaseURL.absoluteString - ) - let snapshot = self.makeSnapshot( - environments: environments, - activeEnvironment: activeEnvironment, - connectionState: currentConnection.state, - connectionDetail: currentConnection.detail - ) - self.publish(snapshot) - if shellsChanged || hasActiveWork { - nextInterval = fastInterval - } else { - nextInterval = idleInterval - } + do { try await Task.sleep(for: failureInterval) } catch { return } } } + + func applyCatalogue(_ config: ServerConfigSnapshot, environment: Environment, saved: [Environment]) { + guard let owner, isCurrent else { return } + owner.setServerConfig(config, environmentID: environment.id) + owner.publishAggregateSnapshot(saved) + } + } + + private func publishAggregateSnapshot(_ environments: [Environment]) { + guard let activeEnvironment else { return } + let connection = latestSnapshot?.connection + publish(makeSnapshot( + environments: environments, activeEnvironment: activeEnvironment, + connectionState: connection?.state ?? .disconnected, + connectionDetail: connection?.detail + )) } nonisolated private static func shellNeedsFrequentAggregateRefresh( @@ -4870,31 +4970,33 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, savedIDs.contains($0.key) } - for load in loads { - environmentClients[load.environment.id] = load.client - if let config = load.config { - setServerConfig(config, environmentID: load.environment.id) - if load.environment.id == activeEnvironment?.id { - latestServerConfig = config - } + for load in loads { applyEnvironmentLoad(load) } + rebuildEntityIndexes(savedEnvironments) + } + + private func applyEnvironmentLoad(_ load: EnvironmentShellLoad) { + environmentClients[load.environment.id] = load.client + if let config = load.config { + setServerConfig(config, environmentID: load.environment.id) + if load.environment.id == activeEnvironment?.id { + latestServerConfig = config } - if let shell = load.shell { - if shell.snapshotSequence - >= (shellsByEnvironmentID[load.environment.id]?.snapshotSequence ?? .min) { - shellsByEnvironmentID[load.environment.id] = shell - } - environmentConnectionStates[load.environment.id] = .connected - environmentConnectionDetails[load.environment.id] = nil - } else if load.credentialRejected && load.environment.kind != .managedDPoP { - environmentConnectionStates[load.environment.id] = .needsPairing - environmentConnectionDetails[load.environment.id] = Self.needsPairingDetail - } else { - environmentConnectionStates[load.environment.id] = .disconnected - environmentConnectionDetails[load.environment.id] = - load.failureDetail ?? "That server is currently unreachable." + } + if let shell = load.shell { + if shell.snapshotSequence + >= (shellsByEnvironmentID[load.environment.id]?.snapshotSequence ?? .min) { + shellsByEnvironmentID[load.environment.id] = shell } + environmentConnectionStates[load.environment.id] = .connected + environmentConnectionDetails[load.environment.id] = nil + } else if load.credentialRejected && load.environment.kind != .managedDPoP { + environmentConnectionStates[load.environment.id] = .needsPairing + environmentConnectionDetails[load.environment.id] = Self.needsPairingDetail + } else { + environmentConnectionStates[load.environment.id] = .disconnected + environmentConnectionDetails[load.environment.id] = + load.failureDetail ?? "That server is currently unreachable." } - rebuildEntityIndexes(savedEnvironments) } static let needsPairingDetail = "This computer no longer accepts the saved pairing. Pair again." diff --git a/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift b/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift index cc188815bb14..5e698114cdd9 100644 --- a/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift @@ -1179,6 +1179,9 @@ final class NativeMultiEnvironmentTests: XCTestCase { aggregateRefreshSleep: @escaping @Sendable (Duration) async throws -> Void = { try await Task.sleep(for: $0) }, + aggregatePeerRefreshSleep: @escaping @Sendable (String, Duration) async throws -> Void = { _, interval in + try await Task.sleep(for: interval) + }, aggregateEnvironmentLoader: @escaping @Sendable (EnvironmentRuntime) async throws -> [Environment] = { try await $0.environments() } @@ -1285,6 +1288,7 @@ final class NativeMultiEnvironmentTests: XCTestCase { aggregateIdleRefreshInterval: aggregateIdleRefreshInterval, aggregateFailureRefreshInterval: aggregateFailureRefreshInterval, aggregateRefreshSleep: aggregateRefreshSleep, + aggregatePeerRefreshSleep: aggregatePeerRefreshSleep, aggregateEnvironmentLoader: aggregateEnvironmentLoader ) ) @@ -1391,8 +1395,8 @@ struct NativePassiveThreadRefreshTests { func passiveThreadEventsArriveWithinFiveSecondsAndStayFastAfterChanges() async throws { let refreshSleep = ControllableAggregateRefreshSleep() let fixture = try await NativeMultiEnvironmentTests.makeFixture( - aggregateRefreshSleep: { - try await refreshSleep.sleep(for: $0) + aggregatePeerRefreshSleep: { _, interval in + try await refreshSleep.sleep(for: interval) } ) defer { try? FileManager.default.removeItem(at: fixture.directory) } @@ -1433,8 +1437,8 @@ struct NativePassiveThreadRefreshTests { func passiveRefreshUsesTenSecondsWhenWorkIsUnchanged() async throws { let refreshSleep = ControllableAggregateRefreshSleep() let fixture = try await NativeMultiEnvironmentTests.makeFixture( - aggregateRefreshSleep: { - try await refreshSleep.sleep(for: $0) + aggregatePeerRefreshSleep: { _, interval in + try await refreshSleep.sleep(for: interval) } ) defer { try? FileManager.default.removeItem(at: fixture.directory) } @@ -1448,52 +1452,184 @@ struct NativePassiveThreadRefreshTests { await fixture.client.disconnect() } - @Test("A failed passive environment backs off without slowing an active peer") + @Test("A failed passive environment backs off without slowing an active peer", .timeLimit(.minutes(1))) func failedPassiveEnvironmentBacksOffWithoutSlowingActivePeer() async throws { - let refreshSleep = ControllableAggregateRefreshSleep() + let healthyClock = ControllableAggregateRefreshSleep() + let failedClock = ControllableAggregateRefreshSleep() let fixture = try await NativeMultiEnvironmentTests.makeFixture( includeThirdEnvironment: true, - aggregateRefreshSleep: { - try await refreshSleep.sleep(for: $0) + aggregatePeerRefreshSleep: { id, interval in + try await (id == "two" ? healthyClock : failedClock).sleep(for: interval) } ) defer { try? FileManager.default.removeItem(at: fixture.directory) } _ = try await fixture.client.initialSnapshot() - await fixture.transport.setShell( - multiEnvironmentShell( - projectID: "project-two", - threadID: "thread-two", - title: "Remote work", - providerID: "claudeAgent", - modelID: "claude-opus-4-1", - backgroundLiveness: .working - ), - host: "two.example" - ) await fixture.transport.setReachable(false, host: "three.example") + _ = await healthyClock.waitUntilRequested(count: 1) + _ = await failedClock.waitUntilRequested(count: 1) + await failedClock.resume() + let failureCadence = await failedClock.waitUntilRequested(count: 2) + #expect(failureCadence == .seconds(20)) + for count in 2...4 { + await healthyClock.resume() + _ = await healthyClock.waitUntilRequested(count: count) + let failedReads = await fixture.transport.shellReadCount(host: "three.example") + #expect(failedReads == 2) + } + await failedClock.resume() + _ = await failedClock.waitUntilRequested(count: 3) + let retriedReads = await fixture.transport.shellReadCount(host: "three.example") + #expect(retriedReads == 3) + await fixture.client.disconnect() + } - let firstCadence = await refreshSleep.waitUntilRequested(count: 1) - #expect(firstCadence == .seconds(5)) - await refreshSleep.resume() - let secondCadence = await refreshSleep.waitUntilRequested(count: 2) - #expect(secondCadence == .seconds(5)) - let initialFailedReadCount = await fixture.transport.shellReadCount(host: "three.example") - #expect(initialFailedReadCount == 2) - - for requestCount in 2...4 { - await refreshSleep.resume() - let cadence = await refreshSleep.waitUntilRequested(count: requestCount + 1) - #expect(cadence == .seconds(5)) - let failedReadCount = await fixture.transport.shellReadCount(host: "three.example") - #expect(failedReadCount == 2) + @Test("A peer that rejects its pairing is not probed for its catalogue", .timeLimit(.minutes(1))) + func rejectedPeerSkipsCatalogueProbes() async throws { + let catalogueClock = ControllableAggregateRefreshSleep() + let failureInterval = NativeFeatureClient.defaultAggregateFailureRefreshInterval + let fixture = try await NativeMultiEnvironmentTests.makeFixture( + includeThirdEnvironment: true, + aggregatePeerRefreshSleep: { id, interval in + // Only the catalogue loop waits the failure interval for this peer. + guard id == "three", interval == failureInterval else { return try await Task.sleep(for: .seconds(3_600)) } + try await catalogueClock.sleep(for: interval) + } + ) + defer { try? FileManager.default.removeItem(at: fixture.directory) } + await fixture.transport.rejectCredential(host: "three.example") + _ = try await fixture.client.initialSnapshot() + for count in 1...3 { + _ = await catalogueClock.waitUntilRequested(count: count) + #expect(await fixture.transport.ticketCount(host: "three.example") == 0) + await catalogueClock.resume() } + await fixture.client.disconnect() + } - await refreshSleep.resume() - _ = await refreshSleep.waitUntilRequested(count: 6) - let retriedReadCount = await fixture.transport.shellReadCount(host: "three.example") - #expect(retriedReadCount == 3) + @Test("Healthy rows publish twice while a peer shell and optional catalogue remain held", .timeLimit(.minutes(1))) + func healthyRowsPublishTwiceWhilePeerAndCatalogueAreHeld() async throws { + let healthyClock = ControllableAggregateRefreshSleep() + let slowClock = ControllableAggregateRefreshSleep() + let connector = GatedPassiveCatalogueConnector() + let topologyGate = PassiveRequestGate() + let fixture = try await NativeMultiEnvironmentTests.makeFixture( + includeThirdEnvironment: true, + webSocketConnector: connector, + aggregatePeerRefreshSleep: { id, interval in + try await (id == "two" ? healthyClock : slowClock).sleep(for: interval) + }, + aggregateEnvironmentLoader: { runtime in + await topologyGate.enter() + return try await runtime.environments() + } + ) + defer { try? FileManager.default.removeItem(at: fixture.directory) } + let initial = try await fixture.client.initialSnapshot() + let healthy = try #require(initial.threads.first { $0.environmentID == "two" }) + let heldShell = PassiveRequestGate() + await fixture.transport.holdNextShell(host: "three.example", gate: heldShell) + await connector.holdPassiveConnections() + await topologyGate.release() + await connector.gate.waitUntilEntered() + _ = await healthyClock.waitUntilRequested(count: 1) + _ = await slowClock.waitUntilRequested(count: 1) + await slowClock.resume() + await heldShell.waitUntilEntered() + for update in 1...2 { + let title = "Healthy update \(update)" + let probe = ThreadTitleEventProbe(events: fixture.client.events(), threadID: healthy.id, title: title) + probe.start() + await fixture.transport.setShell(multiEnvironmentShell( + projectID: "project-two", threadID: "thread-two", title: title, + snapshotSequence: update + 1 + ), host: "two.example") + await healthyClock.resume() + await probe.waitUntilObserved() + #expect(probe.didObserveTitle()) + _ = await healthyClock.waitUntilRequested(count: update + 1) + #expect(await heldShell.isHeld) + #expect(await connector.gate.isHeld) + } await fixture.client.disconnect() + await heldShell.release() + await connector.release() } + @Test("A cancelled refresh generation cannot overwrite a restarted peer", .timeLimit(.minutes(1))) + func cancelledGenerationCannotOverwriteRestartedPeer() async throws { + let clock = ControllableAggregateRefreshSleep() + let fixture = try await NativeMultiEnvironmentTests.makeFixture( + aggregatePeerRefreshSleep: { _, interval in try await clock.sleep(for: interval) } + ) + defer { try? FileManager.default.removeItem(at: fixture.directory) } + _ = try await fixture.client.initialSnapshot() + _ = await clock.waitUntilRequested(count: 1) + let oldWorker = try #require(fixture.client.aggregateRefreshWorkers["two"]?.task) + let gate = PassiveRequestGate() + await fixture.transport.setShell(multiEnvironmentShell( + projectID: "project-two", threadID: "thread-two", title: "Stale generation", + snapshotSequence: 999 + ), host: "two.example") + await fixture.transport.holdNextShell(host: "two.example", gate: gate) + await clock.resume() + await gate.waitUntilEntered() + await fixture.transport.setShell(multiEnvironmentShell( + projectID: "project-two", threadID: "thread-two", title: "Current generation", + snapshotSequence: 2 + ), host: "two.example") + _ = try await fixture.client.initialSnapshot() + #expect(oldWorker.isCancelled) + await gate.release() + await oldWorker.value + // A stale 999 would survive this read of sequence 2 if it reached the cache. + let snapshot = try await fixture.client.backgroundSnapshot() + #expect(snapshot.threads.first { $0.environmentID == "two" }?.title == "Current generation") + await fixture.client.disconnect() + } + + @Test("Removal cancels a held peer and its late response never returns a row", .timeLimit(.minutes(1))) + func removedPeerCannotPublishLateResponse() async throws { + let removedClock = ControllableAggregateRefreshSleep() + let healthyClock = ControllableAggregateRefreshSleep() + let fixture = try await NativeMultiEnvironmentTests.makeFixture( + includeThirdEnvironment: true, + aggregatePeerRefreshSleep: { id, interval in + try await (id == "two" ? removedClock : healthyClock).sleep(for: interval) + } + ) + defer { try? FileManager.default.removeItem(at: fixture.directory) } + let initial = try await fixture.client.initialSnapshot() + let healthy = try #require(initial.threads.first { $0.environmentID == "three" }) + let probe = ThreadTitleEventProbe( + events: fixture.client.events(), threadID: healthy.id, title: "Removal verified" + ) + probe.start() + _ = await removedClock.waitUntilRequested(count: 1) + _ = await healthyClock.waitUntilRequested(count: 1) + let oldWorker = try #require(fixture.client.aggregateRefreshWorkers["two"]?.task) + let gate = PassiveRequestGate() + await fixture.transport.setShell(multiEnvironmentShell( + projectID: "project-two", threadID: "thread-two", title: "Ghost removed row", + snapshotSequence: 999 + ), host: "two.example") + await fixture.transport.holdNextShell(host: "two.example", gate: gate) + await removedClock.resume() + await gate.waitUntilEntered() + try await fixture.client.removeEnvironment(id: "two") + #expect(oldWorker.isCancelled) + #expect(fixture.client.aggregateRefreshWorkers["two"] == nil) + await gate.release() + await oldWorker.value + await fixture.transport.setShell(multiEnvironmentShell( + projectID: "project-three", threadID: "thread-three", title: "Removal verified", + snapshotSequence: 2 + ), host: "three.example") + await healthyClock.resume() + await probe.waitUntilObserved() + #expect(!probe.sawThreadTitle("Ghost removed row")) + #expect(!probe.lastEnvironmentIDs.contains("two")) + await fixture.client.disconnect() + } + } private actor FailOnceAggregateEnvironmentLoader { @@ -1678,6 +1814,8 @@ private final class ThreadTitleEventProbe { private let threadID: String private let title: String private var observed = false + private var observedTitles = Set() + private(set) var lastEnvironmentIDs = Set() private var observedWaiters: [CheckedContinuation] = [] private var task: Task? @@ -1693,8 +1831,11 @@ private final class ThreadTitleEventProbe { for await event in events { switch event { case let .thread(thread): + observedTitles.insert(thread.title) observed = thread.id == threadID && thread.title == title case let .snapshot(snapshot): + observedTitles.formUnion(snapshot.threads.map(\.title)) + lastEnvironmentIDs = Set(snapshot.environments.map(\.id)) observed = snapshot.threads.contains { $0.id == self.threadID && $0.title == self.title } @@ -1710,6 +1851,8 @@ private final class ThreadTitleEventProbe { } } + func sawThreadTitle(_ title: String) -> Bool { observedTitles.contains(title) } + func didObserveTitle() -> Bool { observed } @@ -1784,8 +1927,14 @@ private actor MultiEnvironmentHTTPTransport: HTTPTransport { private var dispatched: [MultiEnvironmentDispatchRecord] = [] private var hostsDroppingNextCreateReply = Set() private var diffRequests: [(host: String, input: JSONValue)] = [] + private var rejectedHosts = Set() + private var ticketCounts: [String: Int] = [:] + + func rejectCredential(host: String) { rejectedHosts.insert(host) } + func ticketCount(host: String) -> Int { ticketCounts[host, default: 0] } func pullRequestDiffRequests() -> [(host: String, input: JSONValue)] { diffRequests } + private var nextShellGates: [String: PassiveRequestGate] = [:] init(shells: [String: OrchestrationShellSnapshot]) { self.shells = shells @@ -1837,11 +1986,26 @@ private actor MultiEnvironmentHTTPTransport: HTTPTransport { hostsDroppingNextCreateReply.insert(host) } - func data(for request: URLRequest) throws -> (Data, HTTPURLResponse) { + func holdNextShell(host: String, gate: PassiveRequestGate) { + nextShellGates[host] = gate + } + + func data(for request: URLRequest) async throws -> (Data, HTTPURLResponse) { let host = request.url?.host ?? "" let path = request.url?.path ?? "" + if path == "/api/auth/websocket-ticket" { ticketCounts[host, default: 0] += 1 } + if rejectedHosts.contains(host) { + return ( + Data(#"{"error":"Unauthorized"}"#.utf8), + HTTPURLResponse(url: request.url!, statusCode: 401, httpVersion: nil, headerFields: nil)! + ) + } if path == "/api/orchestration/shell" { shellReadCounts[host, default: 0] += 1 + if let gate = nextShellGates.removeValue(forKey: host), let data = shellData[host] { + await gate.enter() + return (data, multiEnvironmentResponse(request)) + } } guard reachableHosts.contains(host) else { throw URLError(.cannotConnectToHost) @@ -2333,3 +2497,71 @@ private func multiEnvironmentResponse(_ request: URLRequest) -> HTTPURLResponse headerFields: ["Content-Type": "application/json"] )! } + +/// Deliberately ignores cancellation to model a transport completing an old read. +private actor PassiveRequestGate { + private var entered = false + private var released = false + private var entryWaiters: [CheckedContinuation] = [] + private var releaseWaiters: [CheckedContinuation] = [] + var isHeld: Bool { entered && !released } + + func enter() async { + entered = true + entryWaiters.forEach { $0.resume() } + entryWaiters.removeAll() + guard !released else { return } + await withCheckedContinuation { releaseWaiters.append($0) } + } + + func waitUntilEntered() async { + guard !entered else { return } + await withCheckedContinuation { entryWaiters.append($0) } + } + + func release() { + released = true + releaseWaiters.forEach { $0.resume() } + releaseWaiters.removeAll() + } +} + +private actor GatedPassiveCatalogueConnector: WebSocketConnecting { + private var shouldHold = false + let gate = PassiveRequestGate() + + func holdPassiveConnections() { shouldHold = true } + + func connect(to url: URL) async throws -> any WebSocketConnection { + if shouldHold && url.host == "two.example" { + return GatedPassiveCatalogueConnection(gate: gate) + } + throw URLError(.cannotConnectToHost) + } + + func release() async { await gate.release() } +} + +private actor GatedPassiveCatalogueConnection: WebSocketConnection { + let gate: PassiveRequestGate + private var receiver: CheckedContinuation? + private var closed = false + + init(gate: PassiveRequestGate) { self.gate = gate } + + func send(_ data: Data) async throws { + let request = try JSONDecoder.t3.decode(JSONValue.self, from: data) + if request["tag"]?.stringValue != nil { await gate.enter() } + } + + func receive() async throws -> Data { + if closed { throw CancellationError() } + return try await withCheckedThrowingContinuation { receiver = $0 } + } + + func close() { + closed = true + receiver?.resume(throwing: CancellationError()) + receiver = nil + } +}