diff --git a/CHANGELOG.md b/CHANGELOG.md index 2cdcb54a8..9ca8dc658 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Cold launch to a usable window, 470ms down to 260ms. - Plugin signature checks run after the first window rather than on the launch thread, at 13ms each. - One gate in front of every path that loads a plugin's executable, enabling one included. +- Stale `cloudflared` and `cloud-sql-proxy` cleanup waits for the process to exit before a connection reuses its port. ### Fixed diff --git a/TablePro/Core/CloudSQL/CloudSQLProxyManager.swift b/TablePro/Core/CloudSQL/CloudSQLProxyManager.swift index 2ed509ee8..46b96d18e 100644 --- a/TablePro/Core/CloudSQL/CloudSQLProxyManager.swift +++ b/TablePro/Core/CloudSQL/CloudSQLProxyManager.swift @@ -28,6 +28,8 @@ actor CloudSQLProxyManager: TunnelManaging { private let runnerFactory: () -> any SupervisedProcessRunner private let binaryManager: CloudSQLProxyBinaryManager private let systemBinaryLookup: (String) -> String? + private var staleSweep: Task? + private let reaperTimings: StaleProcessReaper.Timings private static let runnerRegistry = OSAllocatedUnfairLock(initialState: [UUID: any SupervisedProcessRunner]()) @@ -36,11 +38,13 @@ actor CloudSQLProxyManager: TunnelManaging { init( runnerFactory: @escaping () -> any SupervisedProcessRunner = { ProcessSupervisedRunner() }, binaryManager: CloudSQLProxyBinaryManager = .shared, - systemBinaryLookup: @escaping (String) -> String? = { CLIExecutableFinder.findExecutable($0) } + systemBinaryLookup: @escaping (String) -> String? = { CLIExecutableFinder.findExecutable($0) }, + reaperTimings: StaleProcessReaper.Timings = .production ) { self.runnerFactory = runnerFactory self.binaryManager = binaryManager self.systemBinaryLookup = systemBinaryLookup + self.reaperTimings = reaperTimings } func createTunnel( @@ -50,6 +54,10 @@ actor CloudSQLProxyManager: TunnelManaging { ) async throws -> Int { guard config.isValid else { throw CloudSQLProxyError.invalidInstanceConnectionName } + /// A `cloud-sql-proxy` a crashed session left behind still owns the port this is about to + /// ask for, and a configuration with a fixed `localPort` gets exactly one attempt at it. + await sweepStalePidsIfNeeded() + if tunnels[connectionId] != nil { try await closeTunnel(connectionId: connectionId) } @@ -149,16 +157,42 @@ actor CloudSQLProxyManager: TunnelManaging { tunnels[connectionId]?.localPort } - func sweepStalePidsIfNeeded() { + /// Runs at most once per process. `createTunnel` awaits it, so it is both the launch-time + /// cleanup and the barrier a connection needs before it can trust the port is free. + func sweepStalePidsIfNeeded() async { + if let staleSweep { + await staleSweep.value + return + } + let task = Task { await self.performStaleSweep() } + staleSweep = task + await task.value + } + + private func performStaleSweep() async { Self.purgeCredentialsFiles() - defer { AppStorageEnvironment.shared.defaults.removeObject(forKey: Self.stalePidsDefaultsKey) } - guard let data = AppStorageEnvironment.shared.defaults.data(forKey: Self.stalePidsDefaultsKey), + let defaults = AppStorageEnvironment.shared.defaults + guard let data = defaults.data(forKey: Self.stalePidsDefaultsKey), let records = try? JSONDecoder().decode([CloudSQLProxyPidRecord].self, from: data) else { + defaults.removeObject(forKey: Self.stalePidsDefaultsKey) return } - for record in records where Self.isLiveCloudSQLProxy(record) { - kill(record.pid, SIGTERM) - Self.logger.notice("Reaped stale cloud-sql-proxy pid \(record.pid)") + + let survivors = await StaleProcessReaper.reap( + records.map { CloudSQLProxyPidRecord.reaperTarget($0) }, + timings: reaperTimings + ) + + /// A process that outlived even `SIGKILL` keeps its record, so the next launch tries again + /// rather than forgetting a port is still held. + guard !survivors.isEmpty else { + defaults.removeObject(forKey: Self.stalePidsDefaultsKey) + return + } + let surviving = Set(survivors.map(\.pid)) + let kept = records.filter { surviving.contains($0.pid) } + if let data = try? JSONEncoder().encode(kept) { + defaults.set(data, forKey: Self.stalePidsDefaultsKey) } } @@ -318,17 +352,6 @@ actor CloudSQLProxyManager: TunnelManaging { } } - private static func isLiveCloudSQLProxy(_ record: CloudSQLProxyPidRecord) -> Bool { - guard record.pid > 0 else { return false } - let pathBufferSize = 4 * Int(PATH_MAX) - var buffer = [CChar](repeating: 0, count: pathBufferSize) - let length = proc_pidpath(record.pid, &buffer, UInt32(pathBufferSize)) - guard length > 0 else { return false } - let path = String(cString: buffer) - if !record.binaryPath.isEmpty, path == record.binaryPath { return true } - return (path as NSString).lastPathComponent == "cloud-sql-proxy" - } - private static func isPortInUse(_ stderrTail: String) -> Bool { stderrTail.lowercased().contains("address already in use") } @@ -353,6 +376,14 @@ actor CloudSQLProxyManager: TunnelManaging { struct CloudSQLProxyPidRecord: Codable, Sendable, Equatable { let pid: Int32 let binaryPath: String + + static func reaperTarget(_ record: CloudSQLProxyPidRecord) -> StaleProcessReaper.Target { + StaleProcessReaper.Target( + pid: record.pid, + binaryPath: record.binaryPath, + executableName: "cloud-sql-proxy" + ) + } } // MARK: - Startup monitor diff --git a/TablePro/Core/Cloudflare/CloudflareTunnelManager.swift b/TablePro/Core/Cloudflare/CloudflareTunnelManager.swift index 6bd617ab5..43b2ff97f 100644 --- a/TablePro/Core/Cloudflare/CloudflareTunnelManager.swift +++ b/TablePro/Core/Cloudflare/CloudflareTunnelManager.swift @@ -26,6 +26,8 @@ actor CloudflareTunnelManager: TunnelManaging { private var tunnels: [UUID: TunnelState] = [:] private var pidRecords: [UUID: CloudflaredPidRecord] = [:] private let runnerFactory: () -> any SupervisedProcessRunner + private var staleSweep: Task? + private let reaperTimings: StaleProcessReaper.Timings /// Static registry for synchronous termination during app shutdown. private static let runnerRegistry = OSAllocatedUnfairLock(initialState: [UUID: any SupervisedProcessRunner]()) @@ -33,8 +35,12 @@ actor CloudflareTunnelManager: TunnelManaging { /// Prevents App Nap from throttling the supervised process while tunnels are active. private var appNapActivity: NSObjectProtocol? - init(runnerFactory: @escaping () -> any SupervisedProcessRunner = { ProcessSupervisedRunner() }) { + init( + runnerFactory: @escaping () -> any SupervisedProcessRunner = { ProcessSupervisedRunner() }, + reaperTimings: StaleProcessReaper.Timings = .production + ) { self.runnerFactory = runnerFactory + self.reaperTimings = reaperTimings } /// Create a Cloudflare Access TCP tunnel for a database connection. @@ -45,6 +51,10 @@ actor CloudflareTunnelManager: TunnelManaging { tokenId: String? = nil, tokenSecret: String? = nil ) async throws -> Int { + /// A `cloudflared` a crashed session left behind still owns the port this is about to ask + /// for, and a configuration with a fixed `localPort` gets exactly one attempt at it. + await sweepStalePidsIfNeeded() + if tunnels[connectionId] != nil { try await closeTunnel(connectionId: connectionId) } @@ -135,15 +145,41 @@ actor CloudflareTunnelManager: TunnelManaging { /// Reap cloudflared processes left running by a previous session that crashed /// or was force-quit. Verifies each recorded PID still points at cloudflared /// before signalling it, so a recycled PID is never killed. - func sweepStalePidsIfNeeded() { - defer { AppStorageEnvironment.shared.defaults.removeObject(forKey: Self.stalePidsDefaultsKey) } - guard let data = AppStorageEnvironment.shared.defaults.data(forKey: Self.stalePidsDefaultsKey), + /// Runs at most once per process. `createTunnel` awaits it, so it is both the launch-time + /// cleanup and the barrier a connection needs before it can trust the port is free. + func sweepStalePidsIfNeeded() async { + if let staleSweep { + await staleSweep.value + return + } + let task = Task { await self.performStaleSweep() } + staleSweep = task + await task.value + } + + private func performStaleSweep() async { + let defaults = AppStorageEnvironment.shared.defaults + guard let data = defaults.data(forKey: Self.stalePidsDefaultsKey), let records = try? JSONDecoder().decode([CloudflaredPidRecord].self, from: data) else { + defaults.removeObject(forKey: Self.stalePidsDefaultsKey) + return + } + + let survivors = await StaleProcessReaper.reap( + records.map { CloudflaredPidRecord.reaperTarget($0) }, + timings: reaperTimings + ) + + /// A process that outlived even `SIGKILL` keeps its record, so the next launch tries again + /// rather than forgetting a port is still held. + guard !survivors.isEmpty else { + defaults.removeObject(forKey: Self.stalePidsDefaultsKey) return } - for record in records where Self.isLiveCloudflared(record) { - kill(record.pid, SIGTERM) - Self.logger.notice("Reaped stale cloudflared pid \(record.pid)") + let surviving = Set(survivors.map(\.pid)) + let kept = records.filter { surviving.contains($0.pid) } + if let data = try? JSONEncoder().encode(kept) { + defaults.set(data, forKey: Self.stalePidsDefaultsKey) } } @@ -265,16 +301,6 @@ actor CloudflareTunnelManager: TunnelManaging { } } - private static func isLiveCloudflared(_ record: CloudflaredPidRecord) -> Bool { - guard record.pid > 0 else { return false } - let pathBufferSize = 4 * Int(PATH_MAX) - var buffer = [CChar](repeating: 0, count: pathBufferSize) - let length = proc_pidpath(record.pid, &buffer, UInt32(pathBufferSize)) - guard length > 0 else { return false } - let path = String(cString: buffer) - if !record.binaryPath.isEmpty, path == record.binaryPath { return true } - return (path as NSString).lastPathComponent == "cloudflared" - } private static func isPortInUse(_ stderrTail: String) -> Bool { stderrTail.lowercased().contains("address already in use") @@ -300,6 +326,14 @@ actor CloudflareTunnelManager: TunnelManaging { struct CloudflaredPidRecord: Codable, Sendable, Equatable { let pid: Int32 let binaryPath: String + + static func reaperTarget(_ record: CloudflaredPidRecord) -> StaleProcessReaper.Target { + StaleProcessReaper.Target( + pid: record.pid, + binaryPath: record.binaryPath, + executableName: "cloudflared" + ) + } } // MARK: - Startup monitor diff --git a/TablePro/Core/Process/StaleProcessReaper.swift b/TablePro/Core/Process/StaleProcessReaper.swift new file mode 100644 index 000000000..1dcccee2a --- /dev/null +++ b/TablePro/Core/Process/StaleProcessReaper.swift @@ -0,0 +1,123 @@ +// +// StaleProcessReaper.swift +// TablePro +// + +import Darwin +import Foundation +import os + +/// Terminates helper processes a previous session left behind, and does not return until they are +/// actually gone. +/// +/// A crash or force-quit leaves `cloudflared` or `cloud-sql-proxy` running, still holding the local +/// port it was told to listen on. Signalling it and returning is not enough: a connection with a +/// fixed local port gets one attempt at that port, so a replacement started while the orphan is +/// still exiting fails with an address already in use. The reaper therefore signals, polls until the +/// process is gone, escalates once, and reports whatever survived so the caller can keep its record +/// and try again next launch. +/// +/// These processes are not this process's children, so `waitpid` is unavailable and polling +/// `proc_pidpath` is the only way to see them exit. +internal enum StaleProcessReaper { + internal struct Target: Sendable, Equatable { + internal let pid: pid_t + internal let binaryPath: String + internal let executableName: String + + internal init(pid: pid_t, binaryPath: String, executableName: String) { + self.pid = pid + self.binaryPath = binaryPath + self.executableName = executableName + } + } + + internal struct Timings: Sendable { + internal let grace: Duration + internal let forcedGrace: Duration + internal let poll: Duration + + internal init(grace: Duration, forcedGrace: Duration, poll: Duration) { + self.grace = grace + self.forcedGrace = forcedGrace + self.poll = poll + } + + /// `cloudflared` and `cloud-sql-proxy` both close their listener and exit on `SIGTERM` well + /// inside the grace period. It is long enough that the escalation is the rare path, and the + /// wait is only ever paid by a connection that would otherwise have failed outright. + internal static let production = Timings( + grace: .seconds(2), + forcedGrace: .milliseconds(500), + poll: .milliseconds(50) + ) + } + + nonisolated private static let logger = Logger(subsystem: "com.TablePro", category: "StaleProcessReaper") + + /// A pid the OS recycled after the crash belongs to something else, so the executable behind it + /// is checked before every signal rather than once at the start. + internal static func isLive(_ target: Target) -> Bool { + guard target.pid > 0 else { return false } + let bufferSize = 4 * Int(PATH_MAX) + var buffer = [CChar](repeating: 0, count: bufferSize) + let length = proc_pidpath(target.pid, &buffer, UInt32(bufferSize)) + guard length > 0 else { return false } + let path = String(cString: buffer) + if !target.binaryPath.isEmpty, path == target.binaryPath { return true } + return (path as NSString).lastPathComponent == target.executableName + } + + /// Returns the targets still running when the reaper gave up, which is empty in every ordinary + /// case. + internal static func reap( + _ targets: [Target], + timings: Timings = .production, + isLive: @Sendable (Target) -> Bool = StaleProcessReaper.isLive, + signal send: @Sendable (pid_t, Int32) -> Void = { kill($0, $1) } + ) async -> [Target] { + var live = targets.filter(isLive) + guard !live.isEmpty else { return [] } + + for target in live { + send(target.pid, SIGTERM) + logger.notice("Signalled stale \(target.executableName, privacy: .public) pid \(target.pid)") + } + + live = await waitForExit(of: live, within: timings.grace, poll: timings.poll, isLive: isLive) + guard !live.isEmpty else { return [] } + + for target in live { + send(target.pid, SIGKILL) + logger.warning( + "Stale \(target.executableName, privacy: .public) pid \(target.pid) ignored SIGTERM, forcing" + ) + } + + live = await waitForExit(of: live, within: timings.forcedGrace, poll: timings.poll, isLive: isLive) + for target in live { + logger.error( + "Stale \(target.executableName, privacy: .public) pid \(target.pid) survived, keeping its record" + ) + } + return live + } + + private static func waitForExit( + of targets: [Target], + within budget: Duration, + poll: Duration, + isLive: @Sendable (Target) -> Bool + ) async -> [Target] { + var remaining = targets.filter(isLive) + guard !remaining.isEmpty else { return [] } + + let deadline = ContinuousClock.now.advanced(by: budget) + while ContinuousClock.now < deadline { + try? await Task.sleep(for: poll) + remaining = remaining.filter(isLive) + if remaining.isEmpty { return [] } + } + return remaining + } +} diff --git a/TableProTests/Core/Process/StaleProcessReaperTests.swift b/TableProTests/Core/Process/StaleProcessReaperTests.swift new file mode 100644 index 000000000..51b3d7298 --- /dev/null +++ b/TableProTests/Core/Process/StaleProcessReaperTests.swift @@ -0,0 +1,171 @@ +// +// StaleProcessReaperTests.swift +// TableProTests +// + +import Darwin +import Foundation +import os +import Testing + +@testable import TablePro + +/// A set of pretend processes. `dies(on:)` says which signal each one answers, so a test can model +/// a well-behaved helper, one that ignores `SIGTERM`, and one that ignores everything. +private final class FakeProcessTable: @unchecked Sendable { + private struct State { + var alive: Set + var diesOn: [pid_t: Int32] + var signals: [(pid: pid_t, signal: Int32)] + } + + private let state: OSAllocatedUnfairLock + + init(alive: [pid_t], diesOn: [pid_t: Int32]) { + state = OSAllocatedUnfairLock(initialState: State(alive: Set(alive), diesOn: diesOn, signals: [])) + } + + var signals: [(pid: pid_t, signal: Int32)] { state.withLock { $0.signals } } + var aliveCount: Int { state.withLock { $0.alive.count } } + + func isLive(_ target: StaleProcessReaper.Target) -> Bool { + state.withLock { $0.alive.contains(target.pid) } + } + + func send(_ pid: pid_t, _ signal: Int32) { + state.withLock { + $0.signals.append((pid, signal)) + if $0.diesOn[pid] == signal { $0.alive.remove(pid) } + } + } +} + +@Suite("Stale process reaper") +struct StaleProcessReaperTests { + private static let fast = StaleProcessReaper.Timings( + grace: .milliseconds(60), + forcedGrace: .milliseconds(60), + poll: .milliseconds(5) + ) + + private func target(_ pid: pid_t) -> StaleProcessReaper.Target { + StaleProcessReaper.Target(pid: pid, binaryPath: "/opt/fake/cloudflared", executableName: "cloudflared") + } + + @Test("A process that is already gone is never signalled") + func deadProcessIsNotSignalled() async { + let table = FakeProcessTable(alive: [], diesOn: [:]) + + let survivors = await StaleProcessReaper.reap( + [target(4242)], + timings: Self.fast, + isLive: table.isLive, + signal: table.send + ) + + #expect(survivors.isEmpty) + #expect(table.signals.isEmpty) + } + + /// The pid the OS handed to something else after the crash. Signalling it would kill a stranger. + @Test("A recycled pid whose executable no longer matches is never signalled") + func recycledPidIsNotSignalled() async { + let table = FakeProcessTable(alive: [], diesOn: [:]) + + _ = await StaleProcessReaper.reap( + [target(99)], + timings: Self.fast, + isLive: table.isLive, + signal: table.send + ) + + #expect(table.signals.isEmpty) + } + + @Test("A process that answers SIGTERM is never forced") + func politeProcessGetsOnlySigterm() async { + let table = FakeProcessTable(alive: [10], diesOn: [10: SIGTERM]) + + let survivors = await StaleProcessReaper.reap( + [target(10)], + timings: Self.fast, + isLive: table.isLive, + signal: table.send + ) + + #expect(survivors.isEmpty) + #expect(table.signals.map(\.signal) == [SIGTERM]) + #expect(table.aliveCount == 0) + } + + /// The case the barrier exists for: the port is not free until this one is actually gone. + @Test("A process that ignores SIGTERM is escalated to SIGKILL") + func stubbornProcessIsForced() async { + let table = FakeProcessTable(alive: [11], diesOn: [11: SIGKILL]) + + let survivors = await StaleProcessReaper.reap( + [target(11)], + timings: Self.fast, + isLive: table.isLive, + signal: table.send + ) + + #expect(survivors.isEmpty) + #expect(table.signals.map(\.signal) == [SIGTERM, SIGKILL]) + #expect(table.aliveCount == 0) + } + + @Test("A process that survives everything is reported so its record is kept") + func unkillableProcessIsReported() async { + let table = FakeProcessTable(alive: [12], diesOn: [:]) + + let survivors = await StaleProcessReaper.reap( + [target(12)], + timings: Self.fast, + isLive: table.isLive, + signal: table.send + ) + + #expect(survivors.map(\.pid) == [12]) + #expect(table.signals.map(\.signal) == [SIGTERM, SIGKILL]) + } + + @Test("Every stale process is signalled, not only the first") + func allTargetsAreSignalled() async { + let table = FakeProcessTable(alive: [20, 21, 22], diesOn: [20: SIGTERM, 21: SIGTERM, 22: SIGTERM]) + + let survivors = await StaleProcessReaper.reap( + [target(20), target(21), target(22)], + timings: Self.fast, + isLive: table.isLive, + signal: table.send + ) + + #expect(survivors.isEmpty) + #expect(Set(table.signals.map(\.pid)) == [20, 21, 22]) + #expect(table.aliveCount == 0) + } + + @Test("An empty set of targets does no work") + func noTargetsIsANoop() async { + let table = FakeProcessTable(alive: [1], diesOn: [:]) + + let survivors = await StaleProcessReaper.reap( + [], + timings: Self.fast, + isLive: table.isLive, + signal: table.send + ) + + #expect(survivors.isEmpty) + #expect(table.signals.isEmpty) + } + + /// `isLive` reads the executable behind the pid, so this is the guard against killing whatever + /// inherited a recycled pid. Nothing is running under this one in the test host. + @Test("The real liveness check rejects a pid that is not running") + func realLivenessCheckRejectsDeadPid() { + #expect(!StaleProcessReaper.isLive(target(-1))) + #expect(!StaleProcessReaper.isLive(target(0))) + } +}