github/openclaw-openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-07-07 19:48:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(macos): lock and unify PortGuardian tunnel record persistence so concurrent app instances cannot lose orphan records (#100601)

Browse Source 
* refactor(macos): single locked write path and per-instance ownership for PortGuardian tunnel records

* chore(i18n): sync native app string inventory for PortGuardian refactor

* chore(i18n): sync native app string inventory for PortGuardian refactor
This commit is contained in:
Peter Steinberger
2026-07-06 07:43:45 +01:00
committed by GitHub
parent ead92d7078
commit 1256aa0a47
5 changed files with 160 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@@ -21,6 +21,7 @@ Docs: https://docs.openclaw.ai
### Fixes
- **Gateway provider auth startup:** avoid default provider-auth warm sweeps while preserving on-demand auth checks and auth-failure rewarming, preventing startup discovery from starving gateway RPCs and channel timers. (#100667, #86752, #86512) Thanks @vincentkoc.
- **Small-context compaction:** cap the effective reserve against the known model context window so small local models do not enter compaction from the first token. (#100621) Thanks @vincentkoc.
- **Control UI terminal reset:** isolate each web terminal's Ghostty WASM runtime so closing a tab and reopening starts with an empty screen instead of reusing freed terminal cells.
- **Plugin install diagnostics:** suppress the misleading hook-pack fallback after plugin install failures only when the hook manifest is absent, while preserving actionable malformed hook-pack errors. (#100554) Thanks @vincentkoc.

6
apps/.i18n/native-source.json
View File

@@ -18515,7 +18515,7 @@
},
{
"kind": "conditional-branch",
"line": 172,
"line": 176,
"path": "apps/macos/Sources/OpenClaw/RemotePortTunnel.swift",
"source": "ssh tunnel exited before listening",
"surface": "apple",
@@ -18523,7 +18523,7 @@
},
{
"kind": "conditional-branch",
"line": 188,
"line": 192,
"path": "apps/macos/Sources/OpenClaw/RemotePortTunnel.swift",
"source": "ssh tunnel did not open local port \\(localPort)",
"surface": "apple",
@@ -18531,7 +18531,7 @@
},
{
"kind": "conditional-branch",
"line": 188,
"line": 192,
"path": "apps/macos/Sources/OpenClaw/RemotePortTunnel.swift",
"source": "ssh tunnel failed: \\(stderr)",
"surface": "apple",

241
apps/macos/Sources/OpenClaw/PortGuardian.swift
View File

@@ -21,7 +21,10 @@ actor PortGuardian {
let executablePath: String?
}
private var records: [Record] = []
/// Tunnels spawned by THIS process. Disk (`port-guard.json`) holds the union
/// across app instances; entries there that we do not own are reap candidates,
/// never adopted — adopting them would resurrect records a sibling removed.
private var ownRecords: [Int32: Record] = [:]
private let logger = Logger(subsystem: "ai.openclaw", category: "portguard")
#if DEBUG
private var testingDescriptors: [Int: Descriptor] = [:]
@@ -35,8 +38,8 @@ actor PortGuardian {
self.appSupportDir.appendingPathComponent("port-guard.json", isDirectory: false)
}
init() {
self.records = Self.loadRecords(from: Self.recordPath)
private nonisolated static var recordLockPath: URL {
self.appSupportDir.appendingPathComponent("port-guard.lock", isDirectory: false)
}
func sweep(mode: AppState.ConnectionMode) async {
@@ -48,60 +51,56 @@ actor PortGuardian {
self.logger.info("port sweep skipped (mode=unconfigured)")
return
}
let ports = [GatewayEnvironment.gatewayPort()]
for port in ports {
let listeners = await self.listeners(on: port)
guard !listeners.isEmpty else { continue }
for listener in listeners {
if Self.isExpected(listener, port: port, mode: mode) {
let message = """
port \(port) already served by expected \(listener.command)
(pid \(listener.pid)) — keeping
"""
self.logger.info("\(message, privacy: .public)")
continue
}
if mode == .remote {
let message = """
port \(port) held by \(listener.command)
(pid \(listener.pid)) in remote mode — not killing
"""
self.logger.warning(message)
continue
}
let killed = await self.kill(listener.pid)
if killed {
let message = """
port \(port) was held by \(listener.command)
(pid \(listener.pid)); terminated
"""
self.logger.error("\(message, privacy: .public)")
} else {
self.logger.error("failed to terminate pid \(listener.pid) on port \(port, privacy: .public)")
}
let port = GatewayEnvironment.gatewayPort()
for listener in await self.listeners(on: port) {
if Self.isExpected(listener, port: port, mode: mode) {
let message = """
port \(port) already served by expected \(listener.command)
(pid \(listener.pid)) — keeping
"""
self.logger.info("\(message, privacy: .public)")
continue
}
if mode == .remote {
let message = """
port \(port) held by \(listener.command)
(pid \(listener.pid)) in remote mode — not killing
"""
self.logger.warning(message)
continue
}
if await Self.terminateProcess(listener.pid) {
let message = """
port \(port) was held by \(listener.command)
(pid \(listener.pid)); terminated
"""
self.logger.error("\(message, privacy: .public)")
} else {
self.logger.error("failed to terminate pid \(listener.pid) on port \(port, privacy: .public)")
}
}
self.logger.info("port sweep done")
}
func record(port: Int, pid: Int32, command: String, mode: AppState.ConnectionMode) async {
try? FileManager().createDirectory(at: Self.appSupportDir, withIntermediateDirectories: true)
self.records.removeAll { $0.pid == pid }
self.records.append(
Record(
port: port,
pid: pid,
command: command,
mode: mode.rawValue,
timestamp: Date().timeIntervalSince1970))
self.save()
func record(port: Int, pid: Int32, command: String, mode: AppState.ConnectionMode) {
let record = Record(
port: port,
pid: pid,
command: command,
mode: mode.rawValue,
timestamp: Date().timeIntervalSince1970)
self.ownRecords[pid] = record
Self.persistRecords { $0[pid] = record }
}
func removeRecord(pid: Int32) {
let before = self.records.count
self.records.removeAll { $0.pid == pid }
if self.records.count != before {
self.save()
guard let owned = self.ownRecords.removeValue(forKey: pid) else { return }
// Content-guarded like reap: a sibling instance may have re-recorded a
// recycled pid for its own new tunnel; that fresh record must survive.
Self.persistRecords { merged in
if merged[pid] == owned {
merged.removeValue(forKey: pid)
}
}
}
@@ -127,32 +126,34 @@ actor PortGuardian {
/// leaves the tunnel reparented to launchd, holding the remote connection and
/// squatting the preferred local port so new tunnels drift to ephemeral ports.
func reapOrphanedTunnels() async {
let disk = Self.loadRecords(from: Self.recordPath)
let plan = Self.planTunnelReap(
memory: self.records,
disk: disk,
own: Array(self.ownRecords.values),
disk: Self.loadRecords(from: Self.recordPath),
processInfo: Self.tunnelProcessInfo(pid:))
var kept = plan.keep
var removals = plan.drop
for record in plan.reap {
if await self.reapTunnelProcess(record.pid) {
if await Self.terminateProcess(record.pid) {
removals.append(record)
let message = """
reaped orphaned ssh tunnel (pid \(record.pid), local port \(record.port))
"""
self.logger.error("\(message, privacy: .public)")
} else {
// Keep the record so the next sweep retries the kill.
kept.append(record)
// Leave the record in place so the next sweep retries the kill.
self.logger.error("failed to reap orphaned tunnel pid \(record.pid, privacy: .public)")
}
}
// Persist whenever the kept records differ from either source by content, not
// just pid set: a reaped/dropped record may exist only in the file (crashed
// sibling), and a pid-colliding stale disk record must be overwritten too.
let byTime: (Record, Record) -> Bool = { ($0.timestamp, $0.pid) < ($1.timestamp, $1.pid) }
kept.sort(by: byTime)
if kept != self.records.sorted(by: byTime) || kept != disk.sorted(by: byTime) {
self.records = kept
self.save()
guard !removals.isEmpty else { return }
for record in removals where self.ownRecords[record.pid] == record {
self.ownRecords.removeValue(forKey: record.pid)
}
// Remove only records whose content still matches what was classified: a
// sibling instance may have re-recorded the same pid for a new tunnel in the
// meantime, and that fresh record must survive.
Self.persistRecords { merged in
for record in removals where merged[record.pid] == record {
merged.removeValue(forKey: record.pid)
}
}
}
@@ -194,40 +195,47 @@ actor PortGuardian {
}
static func planTunnelReap(
memory: [Record],
own: [Record],
disk: [Record],
processInfo: (Int32) -> TunnelProcessInfo?) -> (reap: [Record], keep: [Record])
processInfo: (Int32) -> TunnelProcessInfo?) -> (reap: [Record], keep: [Record], drop: [Record])
{
// All app instances share port-guard.json, so disk can hold records from a
// crashed sibling this instance never saw. In-memory wins on pid collisions.
// crashed sibling this instance never saw. Own records win pid collisions —
// they are fresher than anything a dead sibling left behind.
var merged: [Int32: Record] = [:]
for record in disk {
merged[record.pid] = record
}
for record in memory {
for record in own {
merged[record.pid] = record
}
let ordered = merged.values.sorted { ($0.timestamp, $0.pid) < ($1.timestamp, $1.pid) }
var reap: [Record] = []
var keep: [Record] = []
var drop: [Record] = []
for record in ordered {
switch self.classifyTunnelRecord(record, process: processInfo(record.pid)) {
case .keep: keep.append(record)
case .drop: continue
case .drop: drop.append(record)
case .reap: reap.append(record)
}
}
return (reap, keep)
return (reap, keep, drop)
}
/// Only a confirmed exit may drop the record (later sweeps retry otherwise), and
/// callers pick a local port right after reaping — the port is only free once ssh
/// has actually exited, not when the signal was delivered. TERM first, then KILL.
private func reapTunnelProcess(_ pid: Int32) async -> Bool {
_ = await ShellExecutor.run(command: ["kill", "-TERM", "\(pid)"], cwd: nil, env: nil, timeout: 2)
if await Self.waitForProcessExit(pid: pid) { return true }
_ = await ShellExecutor.run(command: ["kill", "-KILL", "\(pid)"], cwd: nil, env: nil, timeout: 2)
return await Self.waitForProcessExit(pid: pid)
/// TERM first, then KILL, returning only once the process is confirmed gone:
/// sweep and reap callers rebind the freed port immediately, and reap must not
/// forget a still-running tunnel (the record is its only retry path).
private static func terminateProcess(_ pid: Int32) async -> Bool {
#if canImport(Darwin)
guard pid > 0 else { return false }
_ = Darwin.kill(pid, SIGTERM)
if await self.waitForProcessExit(pid: pid) { return true }
_ = Darwin.kill(pid, SIGKILL)
return await self.waitForProcessExit(pid: pid)
#else
return false
#endif
}
private static func waitForProcessExit(pid: Int32, timeout: TimeInterval = 1.0) async -> Bool {
@@ -325,23 +333,17 @@ actor PortGuardian {
if mode == .unconfigured {
return []
}
let ports = [GatewayEnvironment.gatewayPort()]
var reports: [PortReport] = []
for port in ports {
let listeners = await self.listeners(on: port)
let tunnelHealthy = await self.probeGatewayHealthIfNeeded(
port: port,
mode: mode,
listeners: listeners)
reports.append(Self.buildReport(
port: port,
listeners: listeners,
mode: mode,
tunnelHealthy: tunnelHealthy))
}
return reports
let port = GatewayEnvironment.gatewayPort()
let listeners = await self.listeners(on: port)
let tunnelHealthy = await self.probeGatewayHealthIfNeeded(
port: port,
mode: mode,
listeners: listeners)
return [Self.buildReport(
port: port,
listeners: listeners,
mode: mode,
tunnelHealthy: tunnelHealthy)]
}
func probeGatewayHealth(port: Int, timeout: TimeInterval = 2.0) async -> Bool {
@@ -518,13 +520,6 @@ actor PortGuardian {
#endif
}
private func kill(_ pid: Int32) async -> Bool {
let term = await ShellExecutor.run(command: ["kill", "-TERM", "\(pid)"], cwd: nil, env: nil, timeout: 2)
if term.ok { return true }
let sigkill = await ShellExecutor.run(command: ["kill", "-KILL", "\(pid)"], cwd: nil, env: nil, timeout: 2)
return sigkill.ok
}
private static func isExpected(_ listener: Listener, port: Int, mode: AppState.ConnectionMode) -> Bool {
let cmd = listener.command.lowercased()
let full = listener.fullCommand.lowercased()
@@ -565,9 +560,41 @@ actor PortGuardian {
return decoded
}
private func save() {
guard let data = try? JSONEncoder().encode(self.records) else { return }
try? data.write(to: Self.recordPath, options: [.atomic])
/// The single write path for port-guard.json: read-merge-write under a file
/// lock so concurrent app instances (prod + dev) never clobber each other's
/// records — a lost record means an unreapable orphan after a crash.
private static func persistRecords(_ mutate: (inout [Int32: Record]) -> Void) {
try? FileManager().createDirectory(at: self.appSupportDir, withIntermediateDirectories: true)
self.withRecordFileLock {
let disk = self.loadRecords(from: self.recordPath)
var merged: [Int32: Record] = [:]
for record in disk {
merged[record.pid] = record
}
mutate(&merged)
let ordered = merged.values.sorted { ($0.timestamp, $0.pid) < ($1.timestamp, $1.pid) }
guard ordered != disk else { return }
guard let data = try? JSONEncoder().encode(ordered) else { return }
try? data.write(to: self.recordPath, options: [.atomic])
}
}
/// flock on a sidecar (atomic writes swap the json inode, so locking the json
/// itself would not serialize anything). The kernel releases it if we die.
private static func withRecordFileLock(_ body: () -> Void) {
#if canImport(Darwin)
let fd = open(self.recordLockPath.path, O_CREAT | O_RDWR, 0o644)
guard fd >= 0 else {
body()
return
}
defer { close(fd) }
_ = flock(fd, LOCK_EX)
defer { _ = flock(fd, LOCK_UN) }
body()
#else
body()
#endif
}
}
@@ -580,11 +607,7 @@ extension PortGuardian {
self.testingDescriptors.removeValue(forKey: port)
}
}
}
#endif
#if DEBUG
extension PortGuardian {
static func _testTunnelProcessInfo(pid: Int32) -> TunnelProcessInfo? {
self.tunnelProcessInfo(pid: pid)
}

36
apps/macos/Sources/OpenClaw/RemotePortTunnel.swift
View File

@@ -49,18 +49,23 @@ final class RemotePortTunnel: @unchecked Sendable {
}
deinit {
Self.cleanupStderr(self.stderrHandle)
let pid = self.process.processIdentifier
self.process.terminate()
Task { await PortGuardian.shared.removeRecord(pid: pid) }
self.teardown(waitUntilExit: false)
}
func terminate() {
self.teardown(waitUntilExit: true)
}
/// deinit cannot block on waitUntilExit; explicit terminate() must, so callers
/// can rebind the tunnel's local port immediately after it returns.
private func teardown(waitUntilExit: Bool) {
Self.cleanupStderr(self.stderrHandle)
let pid = self.process.processIdentifier
if self.process.isRunning {
self.process.terminate()
self.process.waitUntilExit()
if waitUntilExit {
self.process.waitUntilExit()
}
}
Task { await PortGuardian.shared.removeRecord(pid: pid) }
}
@@ -147,14 +152,13 @@ final class RemotePortTunnel: @unchecked Sendable {
stderrHandle: stderrHandle,
stderrCapture: stderrCapture)
// Track tunnel so we can clean up stale listeners on restart.
Task {
await PortGuardian.shared.record(
port: Int(localPort),
pid: process.processIdentifier,
command: process.executableURL?.path ?? "ssh",
mode: CommandResolver.connectionSettings().mode)
}
// Record before returning: a crash inside this window would otherwise leave
// an untracked (unreapable) orphan.
await PortGuardian.shared.record(
port: Int(localPort),
pid: process.processIdentifier,
command: process.executableURL?.path ?? "ssh",
mode: CommandResolver.connectionSettings().mode)
return RemotePortTunnel(process: process, localPort: localPort, stderrHandle: stderrHandle)
}
@@ -367,10 +371,6 @@ final class RemotePortTunnel: @unchecked Sendable {
try? handle.close()
}
private static func drainStderr(_ handle: FileHandle) -> String {
self.drainStderr(handle, captured: "")
}
private static func drainStderr(_ handle: FileHandle, captured: String) -> String {
handle.readabilityHandler = nil
defer { try? handle.close() }
@@ -410,7 +410,7 @@ final class RemotePortTunnel: @unchecked Sendable {
}
static func _testDrainStderr(_ handle: FileHandle) -> String {
self.drainStderr(handle)
self.drainStderr(handle, captured: "")
}
#endif

9
apps/macos/Tests/OpenClawIPCTests/LowCoverageHelperTests.swift
View File

@@ -260,13 +260,13 @@ struct LowCoverageHelperTests {
// pid 10: our live tunnel (parent alive). Disk-only records from a crashed
// sibling instance: pid 20 orphaned, pid 30 already gone.
let memory = [record(pid: 10, port: 18790, timestamp: 300)]
let own = [record(pid: 10, port: 18790, timestamp: 300)]
let disk = [
record(pid: 20, port: 18789, timestamp: 100),
record(pid: 30, port: 18791, timestamp: 200),
record(pid: 10, port: 1, timestamp: 1), // superseded by the in-memory record
record(pid: 10, port: 1, timestamp: 1), // superseded by the own record
]
let plan = PortGuardian.planTunnelReap(memory: memory, disk: disk, processInfo: { pid in
let plan = PortGuardian.planTunnelReap(own: own, disk: disk, processInfo: { pid in
switch pid {
case 10: .init(parentPid: 987, startedAt: 299, fullCommand: tunnel(port: 18790))
case 20: .init(parentPid: 1, startedAt: 99, fullCommand: tunnel(port: 18789))
@@ -276,6 +276,9 @@ struct LowCoverageHelperTests {
#expect(plan.reap.map(\.pid) == [20])
#expect(plan.keep.map(\.pid) == [10])
#expect(plan.keep.first?.port == 18790)
// The dead pid 30 is reported as a drop; the shadowed pid-10 disk record is
// superseded, not dropped, so a reap cycle cannot delete the fresh record.
#expect(plan.drop.map(\.pid) == [30])
}
@Test func `port guardian classifies a real orphaned tunnel process for reaping`() async throws {