From d8db5e4286d243c4829eb2414f399e9ba0346a3b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 10:15:41 +0000 Subject: [PATCH 1/3] fix(swift-ios): a slow computer no longer delays other computers' home rows The passive aggregate refresh awaited every environment's shell load before publishing or scheduling the next cycle, so one slow but healthy computer held back a peer's rename by seconds and stretched its 5/10s cadence. Each passive environment now loads in its own task and reports back to the existing refresh loop, which publishes that result immediately. A load still running from an earlier tick is skipped rather than restarted, and the next tick is scheduled from the cycle's first successful answer. Failure backoff, credential rejection handling, generation guards and cancellation are kept. Co-Authored-By: Claude Opus 5.5 --- apps/swift-ios/App/NativeFeatureClient.swift | 227 +++++++++++------- .../NativeMultiEnvironmentTests.swift | 54 +++++ 2 files changed, 191 insertions(+), 90 deletions(-) diff --git a/apps/swift-ios/App/NativeFeatureClient.swift b/apps/swift-ios/App/NativeFeatureClient.swift index 5793e527de36..9e24a41235f3 100644 --- a/apps/swift-ios/App/NativeFeatureClient.swift +++ b/apps/swift-ios/App/NativeFeatureClient.swift @@ -4998,6 +4998,8 @@ 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. + /// Each environment's load reports back on its own, so a slow computer + /// cannot hold back a peer's rows or its next refresh. private func startAggregateRefresh(_ activeClient: T3Client) { aggregateRefreshTask?.cancel() let generation = environmentGeneration @@ -5009,15 +5011,37 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, let loadEnvironments = aggregateEnvironmentLoader aggregateRefreshID = refreshID aggregateRefreshTask = Task { [weak self] in - var nextInterval = fastInterval + let (events, eventSink) = AsyncStream.makeStream(of: AggregateRefreshEvent.self) + var tickID = 0 + var tickTask: Task? + var tickInterval = fastInterval + var loadTasks: [String: Task] = [:] + var cycleEnvironmentIDs: Set = [] + var cycleNeedsFastRefresh = false + var environments: [Environment] = [] var failureBackoffs: [String: Duration] = [:] - while !Task.isCancelled { - let elapsedInterval = nextInterval - do { - try await sleep(nextInterval) - } catch { - return + defer { + tickTask?.cancel() + loadTasks.values.forEach { $0.cancel() } + eventSink.finish() + } + func scheduleTick(after interval: Duration) { + tickTask?.cancel() + tickID &+= 1 + tickInterval = interval + let id = tickID + tickTask = Task { + do { + try await sleep(interval) + } catch { + return + } + eventSink.yield(.tick(id: id, elapsed: interval)) } + } + + scheduleTick(after: fastInterval) + for await event in events { guard let self, self.aggregateRefreshID == refreshID, self.isCurrentSession( @@ -5027,91 +5051,109 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, let activeEnvironment = self.activeEnvironment else { return } - let environments: [Environment] - let loadGenerations = self.environmentLoadGenerations - do { - environments = try await loadEnvironments(self.runtime) - } 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 - } - 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 - continue - } - let loads = await self.loadEnvironmentShells(refreshableEnvironments, loadGenerations: loadGenerations) - 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 + switch event { + case let .tick(id, elapsedInterval): + guard id == tickID else { continue } + tickTask = nil + let loadGenerations = self.environmentLoadGenerations + do { + environments = try await loadEnvironments(self.runtime) + } 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. + scheduleTick(after: failureInterval) + continue } - } - self.reconcileEnvironmentLoads(loads, savedEnvironments: environments) - let currentConnection = self.latestSnapshot?.connection - ?? FeatureConnection( - state: .disconnected, - environmentName: activeEnvironment.label, - endpoint: activeEnvironment.httpBaseURL.absoluteString + 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 { + scheduleTick(after: idleInterval) + continue + } + 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) + } + // A load still running from an earlier tick keeps its + // place; it is neither restarted nor waited for. + let refreshableEnvironments = passiveEnvironments.filter { + failureBackoffs[$0.id, default: .zero] <= .zero && loadTasks[$0.id] == nil + } + guard !refreshableEnvironments.isEmpty else { + scheduleTick(after: fastInterval) + continue + } + cycleEnvironmentIDs = Set(refreshableEnvironments.map(\.id)) + cycleNeedsFastRefresh = false + for environment in refreshableEnvironments { + loadTasks[environment.id] = Task { [weak self] in + let loads = await self?.loadEnvironmentShells( + [environment], loadGenerations: loadGenerations + ) ?? [] + eventSink.yield(.loaded(environmentID: environment.id, loads: loads)) + } + } + case let .loaded(environmentID, loads): + loadTasks[environmentID] = nil + 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 + } + } + 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 ) - 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 + self.publish(snapshot) + + let needsFastRefresh = shellsChanged || hasActiveWork + cycleNeedsFastRefresh = cycleNeedsFastRefresh || needsFastRefresh + cycleEnvironmentIDs.remove(environmentID) + if tickTask == nil { + // The first answer sets the cadence; a peer that is + // still loading reports back whenever it finishes. + let succeeded = loads.contains { $0.shell != nil } + if succeeded || cycleEnvironmentIDs.isEmpty { + scheduleTick(after: cycleNeedsFastRefresh ? fastInterval : idleInterval) + } + } else if needsFastRefresh, tickInterval > fastInterval { + scheduleTick(after: fastInterval) + } } } } @@ -8931,6 +8973,11 @@ private struct CachedAttachmentURL { let expiresAt: Date } +private enum AggregateRefreshEvent: Sendable { + case tick(id: Int, elapsed: Duration) + case loaded(environmentID: String, loads: [EnvironmentShellLoad]) +} + private struct EnvironmentShellLoad: Sendable { let environment: Environment let client: T3Client diff --git a/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift b/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift index 6016eb151b9f..b2fc035015b0 100644 --- a/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift @@ -1728,6 +1728,60 @@ struct NativePassiveThreadRefreshTests { await fixture.client.disconnect() } + @Test( + "A slow passive environment does not delay a peer's rows or cadence", + .timeLimit(.minutes(1)) + ) + func slowPassiveEnvironmentDoesNotDelayPeerRowsOrCadence() async throws { + let refreshSleep = ControllableAggregateRefreshSleep() + let fixture = try await NativeMultiEnvironmentTests.makeFixture( + includeThirdEnvironment: true, + aggregateRefreshSleep: { + try await refreshSleep.sleep(for: $0) + } + ) + defer { try? FileManager.default.removeItem(at: fixture.directory) } + let initial = try await fixture.client.initialSnapshot() + let thread = try #require( + initial.threads.first(where: { $0.environmentID == "two" }) + ) + let updatedTitle = "Peer work published while another computer is slow" + let eventProbe = ThreadTitleEventProbe( + events: fixture.client.events(), + threadID: thread.id, + title: updatedTitle + ) + eventProbe.start() + + _ = await refreshSleep.waitUntilRequested(count: 1) + await fixture.transport.holdNextShellRead(host: "three.example") + await fixture.transport.setShell( + multiEnvironmentShell( + projectID: "project-two", + threadID: "thread-two", + title: updatedTitle, + providerID: "claudeAgent", + modelID: "claude-opus-4-1" + ), + host: "two.example" + ) + await refreshSleep.resume() + + await fixture.transport.waitForHeldShellRead() + await eventProbe.waitUntilObserved() + let changedCadence = await refreshSleep.waitUntilRequested(count: 2) + #expect(changedCadence == .seconds(5)) + + // The held read keeps its place instead of being started again. + let heldShellReads = await fixture.transport.shellReadCount(host: "three.example") + await refreshSleep.resume() + _ = await refreshSleep.waitUntilRequested(count: 3) + #expect(await fixture.transport.shellReadCount(host: "three.example") == heldShellReads) + + await fixture.transport.releaseShellRead() + await fixture.client.disconnect() + } + @Test("A failed passive environment backs off without slowing an active peer") func failedPassiveEnvironmentBacksOffWithoutSlowingActivePeer() async throws { let refreshSleep = ControllableAggregateRefreshSleep() From ad20c0a347ee5c606d702a179ad40f0ed1ceba12 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 10:25:07 +0000 Subject: [PATCH 2/3] fix(swift-ios): a late peer answer no longer postpones the next refresh A changed result arriving while the idle refresh was pending restarted a fast refresh from that moment, which could land after the deadline already scheduled. The loop now keeps the pending deadline and only replaces it with an earlier one. Tests: the controllable sleep releases only the cancelled sleep's own continuation, and the backoff test waits for the offline read to report back before counting reads. Co-Authored-By: Claude Opus 5.5 --- apps/swift-ios/App/NativeFeatureClient.swift | 14 ++- .../NativeMultiEnvironmentTests.swift | 101 +++++++++++++++++- 2 files changed, 107 insertions(+), 8 deletions(-) diff --git a/apps/swift-ios/App/NativeFeatureClient.swift b/apps/swift-ios/App/NativeFeatureClient.swift index 9e24a41235f3..71780baa01aa 100644 --- a/apps/swift-ios/App/NativeFeatureClient.swift +++ b/apps/swift-ios/App/NativeFeatureClient.swift @@ -51,6 +51,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, private let aggregateIdleRefreshInterval: Duration private let aggregateFailureRefreshInterval: Duration private let aggregateRefreshSleep: @Sendable (Duration) async throws -> Void + private let aggregateRefreshNow: @Sendable () -> ContinuousClock.Instant private let environmentShellTimeoutInterval: TimeInterval private let threadSnapshotTimeoutInterval: TimeInterval private let catchUpDelay: @Sendable () async throws -> Void @@ -165,6 +166,9 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, aggregateRefreshSleep: @escaping @Sendable (Duration) async throws -> Void = { try await Task.sleep(for: $0) }, + aggregateRefreshNow: @escaping @Sendable () -> ContinuousClock.Instant = { + ContinuousClock.now + }, environmentShellTimeoutInterval: TimeInterval = 6, threadSnapshotTimeoutInterval: TimeInterval = 8, catchUpDelay: @escaping @Sendable () async throws -> Void = { @@ -214,6 +218,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, self.aggregateIdleRefreshInterval = aggregateIdleRefreshInterval self.aggregateFailureRefreshInterval = aggregateFailureRefreshInterval self.aggregateRefreshSleep = aggregateRefreshSleep + self.aggregateRefreshNow = aggregateRefreshNow self.environmentShellTimeoutInterval = environmentShellTimeoutInterval self.threadSnapshotTimeoutInterval = threadSnapshotTimeoutInterval self.catchUpDelay = catchUpDelay @@ -5008,13 +5013,14 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, let idleInterval = aggregateIdleRefreshInterval let failureInterval = aggregateFailureRefreshInterval let sleep = aggregateRefreshSleep + let now = aggregateRefreshNow let loadEnvironments = aggregateEnvironmentLoader aggregateRefreshID = refreshID aggregateRefreshTask = Task { [weak self] in let (events, eventSink) = AsyncStream.makeStream(of: AggregateRefreshEvent.self) var tickID = 0 var tickTask: Task? - var tickInterval = fastInterval + var tickDeadline = now() + fastInterval var loadTasks: [String: Task] = [:] var cycleEnvironmentIDs: Set = [] var cycleNeedsFastRefresh = false @@ -5028,7 +5034,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, func scheduleTick(after interval: Duration) { tickTask?.cancel() tickID &+= 1 - tickInterval = interval + tickDeadline = now() + interval let id = tickID tickTask = Task { do { @@ -5151,7 +5157,9 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, if succeeded || cycleEnvironmentIDs.isEmpty { scheduleTick(after: cycleNeedsFastRefresh ? fastInterval : idleInterval) } - } else if needsFastRefresh, tickInterval > fastInterval { + } else if needsFastRefresh, now() + fastInterval < tickDeadline { + // Only ever pull the pending refresh earlier, never + // push it past the deadline it already has. scheduleTick(after: fastInterval) } } diff --git a/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift b/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift index b2fc035015b0..72610a1b3ef6 100644 --- a/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift @@ -1448,6 +1448,9 @@ final class NativeMultiEnvironmentTests: XCTestCase { aggregateRefreshSleep: @escaping @Sendable (Duration) async throws -> Void = { try await Task.sleep(for: $0) }, + aggregateRefreshNow: @escaping @Sendable () -> ContinuousClock.Instant = { + ContinuousClock.now + }, aggregateEnvironmentLoader: @escaping @Sendable (EnvironmentRuntime) async throws -> [Environment] = { try await $0.environments() } @@ -1565,6 +1568,7 @@ final class NativeMultiEnvironmentTests: XCTestCase { aggregateIdleRefreshInterval: aggregateIdleRefreshInterval, aggregateFailureRefreshInterval: aggregateFailureRefreshInterval, aggregateRefreshSleep: aggregateRefreshSleep, + aggregateRefreshNow: aggregateRefreshNow, aggregateEnvironmentLoader: aggregateEnvironmentLoader ) ) @@ -1782,6 +1786,63 @@ struct NativePassiveThreadRefreshTests { await fixture.client.disconnect() } + @Test( + "A late changed answer never postpones the scheduled refresh", + .timeLimit(.minutes(1)) + ) + func lateChangedAnswerDoesNotPostponeScheduledRefresh() async throws { + let refreshSleep = ControllableAggregateRefreshSleep() + let refreshClock = ManualAggregateRefreshClock() + let fixture = try await NativeMultiEnvironmentTests.makeFixture( + includeThirdEnvironment: true, + aggregateRefreshSleep: { + try await refreshSleep.sleep(for: $0) + }, + aggregateRefreshNow: { + refreshClock.now() + } + ) + defer { try? FileManager.default.removeItem(at: fixture.directory) } + let initial = try await fixture.client.initialSnapshot() + let thread = try #require( + initial.threads.first(where: { $0.environmentID == "three" }) + ) + let updatedTitle = "Work from a slow computer" + let eventProbe = ThreadTitleEventProbe( + events: fixture.client.events(), + threadID: thread.id, + title: updatedTitle + ) + eventProbe.start() + + _ = await refreshSleep.waitUntilRequested(count: 1) + await fixture.transport.setShell( + multiEnvironmentShell( + projectID: "project-three", + threadID: "thread-three", + title: updatedTitle, + providerID: "codex", + modelID: "gpt-5.6-sol" + ), + host: "three.example" + ) + await fixture.transport.holdNextShellRead(host: "three.example") + await refreshSleep.resume() + await fixture.transport.waitForHeldShellRead() + // The unchanged peer answers first and schedules the idle refresh at 10s. + let idleCadence = await refreshSleep.waitUntilRequested(count: 2) + #expect(idleCadence == .seconds(10)) + + // A fast refresh from 5.8s would land at 10.8s, after the one already due. + refreshClock.advance(by: .milliseconds(5_800)) + await fixture.transport.releaseShellRead() + await eventProbe.waitUntilObserved() + await refreshSleep.resume() + let nextCadence = await refreshSleep.waitUntilRequested(count: 3) + #expect(nextCadence == .seconds(10)) + await fixture.client.disconnect() + } + @Test("A failed passive environment backs off without slowing an active peer") func failedPassiveEnvironmentBacksOffWithoutSlowingActivePeer() async throws { let refreshSleep = ControllableAggregateRefreshSleep() @@ -1809,9 +1870,18 @@ struct NativePassiveThreadRefreshTests { ) await fixture.transport.setReachable(false, host: "three.example") + var events = fixture.client.events().makeAsyncIterator() let firstCadence = await refreshSleep.waitUntilRequested(count: 1) #expect(firstCadence == .seconds(5)) await refreshSleep.resume() + // The peer's answer alone can schedule the next refresh, so wait + // until the offline read has reported back before counting reads. + while let event = await events.next() { + if case let .snapshot(snapshot) = event, + snapshot.environments.first(where: { $0.id == "three" })?.connectionState == .disconnected { + break + } + } let secondCadence = await refreshSleep.waitUntilRequested(count: 2) #expect(secondCadence == .seconds(5)) // Discovery is cached for 60 seconds. The offline read fails at the shell endpoint. @@ -2083,10 +2153,11 @@ private actor ControllableAggregateRefreshSleep { count: Int, continuation: CheckedContinuation )] = [] - private var sleepContinuation: CheckedContinuation? + private var sleepContinuations: [Int: CheckedContinuation] = [:] func sleep(for cadence: Duration) async throws { requestedCadences.append(cadence) + let id = requestedCadences.count let satisfied = requestWaiters.filter { requestedCadences.count >= $0.count } requestWaiters.removeAll { requestedCadences.count >= $0.count } satisfied.forEach { @@ -2097,11 +2168,12 @@ private actor ControllableAggregateRefreshSleep { if Task.isCancelled { continuation.resume() } else { - sleepContinuation = continuation + sleepContinuations[id] = continuation } } } onCancel: { - Task { await self.resume() } + // A cancelled sleep releases only itself, never the sleep that replaced it. + Task { await self.resume(id: id) } } try Task.checkCancellation() } @@ -2113,9 +2185,28 @@ private actor ControllableAggregateRefreshSleep { } } + /// Fires the newest pending sleep, the one the refresh loop is waiting on. func resume() { - sleepContinuation?.resume() - sleepContinuation = nil + guard let id = sleepContinuations.keys.max() else { return } + resume(id: id) + } + + private func resume(id: Int) { + sleepContinuations.removeValue(forKey: id)?.resume() + } +} + +private final class ManualAggregateRefreshClock: @unchecked Sendable { + private let lock = NSLock() + private let start = ContinuousClock.now + private var offset = Duration.zero + + func now() -> ContinuousClock.Instant { + lock.withLock { start + offset } + } + + func advance(by duration: Duration) { + lock.withLock { offset += duration } } } From 6b0ce9b0bc70939c7cd1a81a009b0af70077fed5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 22 Oct 2026 10:37:08 +0000 Subject: [PATCH 3/3] fix(swift-ios): an offline computer waits the full backoff from when it failed The passive refresh subtracted each whole tick from a computer's failure backoff, even when the failure arrived partway through that tick or the tick had been pulled earlier. A read that failed 6s into a 10s wait was retried 14s later instead of 20s. Each computer now keeps a retry deadline measured from when its failure was answered, compared against the injected clock at each tick. Co-Authored-By: Claude Opus 5.5 --- apps/swift-ios/App/NativeFeatureClient.swift | 24 ++-- .../NativeMultiEnvironmentTests.swift | 107 +++++++++++++++++- 2 files changed, 118 insertions(+), 13 deletions(-) diff --git a/apps/swift-ios/App/NativeFeatureClient.swift b/apps/swift-ios/App/NativeFeatureClient.swift index 71780baa01aa..9ddc44712cf6 100644 --- a/apps/swift-ios/App/NativeFeatureClient.swift +++ b/apps/swift-ios/App/NativeFeatureClient.swift @@ -5025,7 +5025,9 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, var cycleEnvironmentIDs: Set = [] var cycleNeedsFastRefresh = false var environments: [Environment] = [] - var failureBackoffs: [String: Duration] = [:] + // Measured from when each failure was answered, not from ticks, + // because a tick can be pulled earlier or land mid-load. + var retryDeadlines: [String: ContinuousClock.Instant] = [:] defer { tickTask?.cancel() loadTasks.values.forEach { $0.cancel() } @@ -5042,7 +5044,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, } catch { return } - eventSink.yield(.tick(id: id, elapsed: interval)) + eventSink.yield(.tick(id: id)) } } @@ -5058,7 +5060,7 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, return } switch event { - case let .tick(id, elapsedInterval): + case let .tick(id): guard id == tickID else { continue } tickTask = nil let loadGenerations = self.environmentLoadGenerations @@ -5089,14 +5091,12 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, continue } 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) - } + retryDeadlines = retryDeadlines.filter { passiveIDs.contains($0.key) } + let tickInstant = now() // A load still running from an earlier tick keeps its // place; it is neither restarted nor waited for. let refreshableEnvironments = passiveEnvironments.filter { - failureBackoffs[$0.id, default: .zero] <= .zero && loadTasks[$0.id] == nil + (retryDeadlines[$0.id] ?? tickInstant) <= tickInstant && loadTasks[$0.id] == nil } guard !refreshableEnvironments.isEmpty else { scheduleTick(after: fastInterval) @@ -5125,11 +5125,11 @@ final class NativeFeatureClient: FeatureClient, FeatureDeviceManaging, 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) + retryDeadlines[load.environment.id] = now() + .seconds(24 * 60 * 60) } else if load.shell == nil { - failureBackoffs[load.environment.id] = failureInterval + retryDeadlines[load.environment.id] = now() + failureInterval } else { - failureBackoffs[load.environment.id] = nil + retryDeadlines[load.environment.id] = nil } } self.reconcileEnvironmentLoads(loads, savedEnvironments: environments) @@ -8982,7 +8982,7 @@ private struct CachedAttachmentURL { } private enum AggregateRefreshEvent: Sendable { - case tick(id: Int, elapsed: Duration) + case tick(id: Int) case loaded(environmentID: String, loads: [EnvironmentShellLoad]) } diff --git a/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift b/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift index 72610a1b3ef6..e05167828cab 100644 --- a/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift +++ b/apps/swift-ios/Tests/FeatureTests/NativeMultiEnvironmentTests.swift @@ -1843,13 +1843,85 @@ struct NativePassiveThreadRefreshTests { await fixture.client.disconnect() } + @Test( + "A failure answered mid-wait backs off from when it failed", + .timeLimit(.minutes(1)) + ) + func failureAnsweredMidWaitBacksOffFromWhenItFailed() async throws { + let refreshSleep = ControllableAggregateRefreshSleep() + let refreshClock = ManualAggregateRefreshClock() + let fixture = try await NativeMultiEnvironmentTests.makeFixture( + includeThirdEnvironment: true, + aggregateRefreshSleep: { + try await refreshSleep.sleep(for: $0) + }, + aggregateRefreshNow: { + refreshClock.now() + } + ) + defer { try? FileManager.default.removeItem(at: fixture.directory) } + _ = try await fixture.client.initialSnapshot() + let initialShellReads = await fixture.transport.shellReadCount(host: "three.example") + var events = fixture.client.events().makeAsyncIterator() + + _ = await refreshSleep.waitUntilRequested(count: 1) + await fixture.transport.holdNextShellRead(host: "three.example") + refreshClock.advance(by: .seconds(5)) + await refreshSleep.resume() + await fixture.transport.waitForHeldShellRead() + // The unchanged peer answers first and schedules the idle refresh at 15s. + let idleCadence = await refreshSleep.waitUntilRequested(count: 2) + #expect(idleCadence == .seconds(10)) + + // The slow computer fails at 11s, so it is not due again until 31s. + refreshClock.advance(by: .seconds(6)) + await fixture.transport.setReachable(false, host: "three.example") + await fixture.transport.failHeldShellRead(host: "three.example") + while let event = await events.next() { + if case let .snapshot(snapshot) = event, + snapshot.environments.first(where: { $0.id == "three" })?.connectionState == .disconnected { + break + } + } + + // With every read failing, a tick is only scheduled once each + // environment it read has answered, so read counts are settled. + await fixture.transport.setReachable(false, host: "two.example") + refreshClock.advance(by: .seconds(4)) + await refreshSleep.resume() + let afterPeerFailure = await refreshSleep.waitUntilRequested(count: 3) + #expect(afterPeerFailure == .seconds(10)) + + // Two ten-second waits have passed, but only fourteen seconds since the failure. + refreshClock.advance(by: .seconds(10)) + await refreshSleep.resume() + let nothingDue = await refreshSleep.waitUntilRequested(count: 4) + #expect(nothingDue == .seconds(5)) + #expect(await fixture.transport.shellReadCount(host: "three.example") == initialShellReads + 1) + + refreshClock.advance(by: .seconds(5)) + await refreshSleep.resume() + _ = await refreshSleep.waitUntilRequested(count: 5) + #expect(await fixture.transport.shellReadCount(host: "three.example") == initialShellReads + 1) + + refreshClock.advance(by: .seconds(5)) + await refreshSleep.resume() + _ = await refreshSleep.waitUntilRequested(count: 6) + #expect(await fixture.transport.shellReadCount(host: "three.example") == initialShellReads + 2) + await fixture.client.disconnect() + } + @Test("A failed passive environment backs off without slowing an active peer") func failedPassiveEnvironmentBacksOffWithoutSlowingActivePeer() async throws { let refreshSleep = ControllableAggregateRefreshSleep() + let refreshClock = ManualAggregateRefreshClock() let fixture = try await NativeMultiEnvironmentTests.makeFixture( includeThirdEnvironment: true, aggregateRefreshSleep: { try await refreshSleep.sleep(for: $0) + }, + aggregateRefreshNow: { + refreshClock.now() } ) defer { try? FileManager.default.removeItem(at: fixture.directory) } @@ -1873,6 +1945,7 @@ struct NativePassiveThreadRefreshTests { var events = fixture.client.events().makeAsyncIterator() let firstCadence = await refreshSleep.waitUntilRequested(count: 1) #expect(firstCadence == .seconds(5)) + refreshClock.advance(by: firstCadence) await refreshSleep.resume() // The peer's answer alone can schedule the next refresh, so wait // until the offline read has reported back before counting reads. @@ -1890,6 +1963,7 @@ struct NativePassiveThreadRefreshTests { #expect(await fixture.transport.shellReadCount(host: "two.example") == initialPeerReads + 1) for requestCount in 2...4 { + refreshClock.advance(by: .seconds(5)) await refreshSleep.resume() let cadence = await refreshSleep.waitUntilRequested(count: requestCount + 1) #expect(cadence == .seconds(5)) @@ -1898,7 +1972,10 @@ struct NativePassiveThreadRefreshTests { #expect(await fixture.transport.shellReadCount(host: "two.example") == initialPeerReads + requestCount) } + // Twenty seconds after the failure, the offline computer is read again. + refreshClock.advance(by: .seconds(5)) await refreshSleep.resume() + await fixture.transport.waitUntilFailedShellReads(host: "three.example", count: 2) _ = await refreshSleep.waitUntilRequested(count: 6) #expect(await fixture.transport.failedShellReadCount(host: "three.example") == 2) #expect(await fixture.transport.shellReadCount(host: "three.example") == initialShellReads + 2) @@ -2228,6 +2305,11 @@ private actor MultiEnvironmentHTTPTransport: HTTPTransport { private var shellReadsEnabledHosts: Set private var shellReadCounts: [String: Int] = [:] private var failedShellReadCounts: [String: Int] = [:] + private var failedShellReadWaiters: [( + host: String, + count: Int, + continuation: CheckedContinuation + )] = [] private var dispatched: [MultiEnvironmentDispatchRecord] = [] private var hostsDroppingNextCreateReply = Set() private var diffRequests: [(host: String, input: JSONValue)] = [] @@ -2285,6 +2367,22 @@ private actor MultiEnvironmentHTTPTransport: HTTPTransport { failedShellReadCounts[host, default: 0] } + func waitUntilFailedShellReads(host: String, count: Int) async { + if failedShellReadCounts[host, default: 0] >= count { return } + await withCheckedContinuation { failedShellReadWaiters.append((host, count, $0)) } + } + + private func recordFailedShellRead(host: String) { + failedShellReadCounts[host, default: 0] += 1 + let satisfied = failedShellReadWaiters.filter { + $0.host == host && failedShellReadCounts[host, default: 0] >= $0.count + } + failedShellReadWaiters.removeAll { + $0.host == host && failedShellReadCounts[host, default: 0] >= $0.count + } + satisfied.forEach { $0.continuation.resume() } + } + func dropNextCreateReply(host: String) { hostsDroppingNextCreateReply.insert(host) } @@ -2320,6 +2418,13 @@ private actor MultiEnvironmentHTTPTransport: HTTPTransport { held.continuation.resume(returning: held.response) } + func failHeldShellRead(host: String) { + guard let held = heldShellRead else { return } + heldShellRead = nil + recordFailedShellRead(host: host) + held.continuation.resume(throwing: URLError(.cannotConnectToHost)) + } + func data(for request: URLRequest) async throws -> (Data, HTTPURLResponse) { let host = request.url?.host ?? "" let path = request.url?.path ?? "" @@ -2328,7 +2433,7 @@ private actor MultiEnvironmentHTTPTransport: HTTPTransport { } guard reachableHosts.contains(host) else { if path == "/api/orchestration/shell" { - failedShellReadCounts[host, default: 0] += 1 + recordFailedShellRead(host: host) } throw URLError(.cannotConnectToHost) }