fix(connections): wait for a stale tunnel process to exit before reusing its local port

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
 

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<Void, Never>?
+    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

TablePro/Core/Cloudflare/CloudflareTunnelManager.swift

@@ -26,15 +26,21 @@ actor CloudflareTunnelManager: TunnelManaging {
     private var tunnels: [UUID: TunnelState] = [:]
     private var pidRecords: [UUID: CloudflaredPidRecord] = [:]
     private let runnerFactory: () -> any SupervisedProcessRunner
+    private var staleSweep: Task<Void, Never>?
+    private let reaperTimings: StaleProcessReaper.Timings
 
     /// Static registry for synchronous termination during app shutdown.
     private static let runnerRegistry = OSAllocatedUnfairLock(initialState: [UUID: any SupervisedProcessRunner]())
 
     /// 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

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