diff --git a/PATCHES.md b/PATCHES.md index 43b6009..792babb 100644 --- a/PATCHES.md +++ b/PATCHES.md @@ -258,6 +258,55 @@ rebuild whenever a guest patch changes. Built locally, not in CI: the host frame image is rebuilt and repointed (mind the `vminit.ext4.reference` cache sidecar). Marked `[Nucleic vendored patch]`. +16. **Loss-free stdio teardown (`IOPair.swift`, `ManagedProcess.swift`, `VsockProxy.swift`, + `StandardIO.swift`, `TerminalIO.swift`) — the truncated-final-output class.** Four related + defects dropped the tail of a stream at teardown, plus two fd leaks: + - `IOPair`'s relay handler closed on a bare EPOLLHUP even with a backpressure flush in + flight, dropping the parked remainder (the CLI's final result line) when the + destination was momentarily full at process exit. It now returns instead; the + destination's EPOLLOUT flushes the backlog, the pump then reads EOF and closes + loss-free. + - `ManagedProcess.setExit` force-closed ALL stdio on SIGCHLD, racing the poller thread's + final EPOLLIN drain (`drain()` silently ignores short writes). It now closes only + stdin; stdout/stderr self-close on EOF (loss-free per the previous fix), with an 8s + delayed full close as a backstop for pipe-inheriting grandchildren. `ManagedProcess.IO` + is now `Sendable` for that delayed capture. + - `VsockProxy` full-hangup teardown dropped bytes already read off the dead peer that + were parked toward the SURVIVING peer; it now makes one best-effort relay pass toward + the survivor before cleanup. + - `VsockProxy` leaked the `relayTo` fd (closeOnDeinit: false) on a failed backend + `connect()`; `StandardIO.start`/`TerminalIO.start`/`attach` leaked live pairs/sockets + (retained forever by the supervisor's handler map) on partial setup failure — all + paths now close what they created before rethrowing. + Marked `[Nucleic vendored patch]`. + +17. **Epoll registration integrity (`Epoll.swift` in ContainerizationOS, `ProcessSupervisor.swift`, + `IOCloser.swift`, `TerminalIO.swift`).** + - **Registration generations:** epoll events now carry the registration's generation in + `epoll_data` (high 32 bits), and the supervisor dispatches only when it matches the + live entry — a stale event queued for a closed registration of a RECYCLED fd number + can no longer fire the new registration's handler (it could tear down a brand-new + healthy connection mid-batch). + - **Non-clobbering `registerFd`:** registering an already-registered fd now fails fast + (EEXIST) instead of overwriting the existing handler and then, on the epoll EEXIST, + deleting the map entry — which left the fd armed with NO handler (a silently dead + relay). + - **`TerminalIO` shared-fd collision:** the stdin relay's write destination is now a + `dup(2)` of the terminal fd (`DupIOCloser`), so its EPOLLOUT backpressure registration + can't collide with the stdout relay's read registration on the same number — one bulk + paste used to permanently kill the terminal's stdout relay. + - `Epoll.add` now ORs `O_NONBLOCK` into existing flags instead of replacing the flag set. + Marked `[Nucleic vendored patch]`. + +18. **Host-side stdin relay off the Swift cooperative pool (`LinuxProcess.swift`).** The + stdin fd is blocking (only the read fds get `O_NONBLOCK`), and `startStdinRelay`'s + `FileHandle.write` parked a width-limited cooperative-pool thread — non-cancellably — + whenever the guest stopped reading with the vsock buffer full (large prompt lines). A + few wedged sessions starved the whole Swift concurrency runtime (every decode loop and + watchdog: an app-wide stall). Writes are now offloaded to a per-process GCD queue via a + checked continuation; a wedged write costs one expendable GCD thread, and process + deletion closing the fd still unwedges it. Marked `[Nucleic vendored patch]`. + ## Re-vendoring a newer upstream commit 1. `git clone` upstream (or copy `.build/checkouts/containerization` after bumping the URL pin @@ -279,9 +328,17 @@ rebuild whenever a guest patch changes. Built locally, not in CI: the host frame `KeychainQuery` reads: `withoutInteractiveUI` + the `errSecInteractionNotAllowed` handling + the `save` duplicate retry), and patch #14 (the non-blocking/non-spinning guest I/O plane: `IOPair` backpressure, the `OSFile.splice` EAGAIN return, and the `VsockProxy` pre-registration - non-blocking fds — all in `vminitd/`), and patch #15 (the per-direction + non-blocking fds — all in `vminitd/`), patch #15 (the per-direction `OSFile.RelayDirection`/`OSFile.relay` rewrite + the two-direction `VsockProxy.handleConn`, - which supersede the upstream `SpliceFile`/`splice` shapes entirely — in `vminitd/`). After + which supersede the upstream `SpliceFile`/`splice` shapes entirely — in `vminitd/`), + patch #16 (loss-free stdio teardown: the `IOPair` HUP-with-pending return, the + `setExit` stdin-only close + grace pass, the `VsockProxy` hangup flush + connect-leak + close, and the `StandardIO`/`TerminalIO` partial-failure cleanup — in `vminitd/`), + patch #17 (epoll registration generations in `ContainerizationOS/Linux/Epoll.swift` + + the generation-checked, non-clobbering `ProcessSupervisor` handler table + `DupIOCloser` + and the `TerminalIO` dup destination — spans `Sources/` AND `vminitd/`), and patch #18 + (the `startStdinRelay` GCD write offload in `Sources/Containerization/LinuxProcess.swift` + — host side). After re-applying any `vminitd/` patch, rebuild + publish the custom init image with `make vminit-image` + `make vminit-image-push`, and bump `ContainerEngine.vminitReference`. 5. Update the commit hash above and in the root `Package.swift` comment. diff --git a/Sources/Containerization/LinuxProcess.swift b/Sources/Containerization/LinuxProcess.swift index bbc15f9..f2fa2fb 100644 --- a/Sources/Containerization/LinuxProcess.swift +++ b/Sources/Containerization/LinuxProcess.swift @@ -270,11 +270,33 @@ extension LinuxProcess { func startStdinRelay(handle: FileHandle) { guard let stdin = self.ioSetup.stdin else { return } + // [Nucleic vendored patch] The stdin fd is BLOCKING (only the read fds get + // O_NONBLOCK), and `FileHandle.write` parks its thread for as long as the guest + // isn't reading once the vsock buffer fills — with a large prompt line that can be + // forever, and the old direct call parked a *width-limited Swift cooperative-pool + // thread*, non-cancellably. A few wedged sessions starved the entire concurrency + // runtime: every exec's decode loop, first-output watchdogs, all of it — an + // app-wide stall. Offload each write to a per-process GCD queue so the pool thread + // suspends instead; a wedged write now costs one expendable GCD thread, and + // process deletion closing the fd still unwedges it. + let writeQueue = DispatchQueue(label: "com.nucleic.stdin-relay") + // The handle is used from one queue at a time (writes are serialized on writeQueue; + // the close paths run after the relay ends or via _closeStdin's own lock). + nonisolated(unsafe) let handle = handle self.state.withLock { $0.stdinRelay = Task { for await data in stdin.reader.stream() { do { - try handle.write(contentsOf: data) + try await withCheckedThrowingContinuation { (c: CheckedContinuation) in + writeQueue.async { + do { + try handle.write(contentsOf: data) + c.resume() + } catch { + c.resume(throwing: error) + } + } + } } catch { self.logger?.error("failed to write to stdin: \(error)") break diff --git a/Sources/ContainerizationOS/Linux/Epoll.swift b/Sources/ContainerizationOS/Linux/Epoll.swift index 3566e17..0a9e3fc 100644 --- a/Sources/ContainerizationOS/Linux/Epoll.swift +++ b/Sources/ContainerizationOS/Linux/Epoll.swift @@ -73,6 +73,13 @@ public final class Epoll: Sendable { /// An event returned by `wait()`. public struct Event: Sendable { public let fd: Int32 + /// [Nucleic vendored patch] The `generation` value passed to `add` for this fd's + /// current registration, echoed back through `epoll_data`. Lets a dispatcher detect + /// a stale event: one queued for a PREVIOUS registration of a recycled fd number + /// (the old fd was closed and a new one with the same number registered between + /// `epoll_wait` returning and the event being dispatched). Without it, a stale + /// EPOLLHUP could tear down a brand-new healthy connection that inherited the number. + public let generation: UInt32 public let mask: Mask } @@ -115,9 +122,13 @@ public final class Epoll: Sendable { close(eventFD) } - /// Register a file descriptor for edge-triggered monitoring. - public func add(_ fd: Int32, mask: Mask) throws { - guard fcntl(fd, F_SETFL, O_NONBLOCK) == 0 else { + /// Register a file descriptor for edge-triggered monitoring. `generation` is echoed + /// back in every `Event` for this registration (see `Event.generation`). + public func add(_ fd: Int32, mask: Mask, generation: UInt32 = 0) throws { + // [Nucleic vendored patch] OR O_NONBLOCK into the existing flags — a bare F_SETFL + // REPLACED the whole flag set, silently clearing e.g. O_APPEND on anything registered. + let flags = fcntl(fd, F_GETFL) + guard flags != -1, fcntl(fd, F_SETFL, flags | O_NONBLOCK) == 0 else { throw POSIXError.fromErrno() } @@ -125,7 +136,7 @@ public final class Epoll: Sendable { var event = epoll_event() event.events = events - event.data.fd = fd + event.data.u64 = UInt64(generation) << 32 | UInt64(UInt32(bitPattern: fd)) try withUnsafeMutablePointer(to: &event) { ptr in if epoll_ctl(self.epollFD, EPOLL_CTL_ADD, fd, ptr) == -1 { @@ -167,11 +178,19 @@ public final class Epoll: Sendable { var result: [Event] = [] result.reserveCapacity(Int(n)) for i in 0..> 32), + mask: Mask(rawValue: events[i].events))) } return result } diff --git a/vminitd/Sources/VminitdCore/IOCloser.swift b/vminitd/Sources/VminitdCore/IOCloser.swift index eb27ff2..e3fa9e9 100644 --- a/vminitd/Sources/VminitdCore/IOCloser.swift +++ b/vminitd/Sources/VminitdCore/IOCloser.swift @@ -16,6 +16,9 @@ #if os(Linux) +import ContainerizationOS +import Foundation + protocol IOCloser: Sendable { var fileDescriptor: Int32 { get } @@ -34,4 +37,25 @@ struct UnownedIOCloser: IOCloser { func close() throws {} } +/// [Nucleic vendored patch] Owns a `dup(2)` of another descriptor. Epoll keys registrations +/// on the fd NUMBER, so two IOPairs sharing one underlying file (TerminalIO: the stdout +/// relay reads the terminal fd, the stdin relay writes it) collide the moment the stdin +/// side needs an EPOLLOUT backpressure registration on the number the stdout side already +/// registered — which now fails fast (EEXIST) and killed the stdin relay. A dup shares the +/// open file description but has its own number, so each relay registers independently; the +/// description stays valid until every descriptor over it is closed. +struct DupIOCloser: IOCloser { + let fileDescriptor: Int32 + + init(duplicating fd: Int32) throws { + let duplicate = dup(fd) + guard duplicate != -1 else { throw POSIXError.fromErrno() } + self.fileDescriptor = duplicate + } + + func close() throws { + guard Foundation.close(fileDescriptor) == 0 else { throw POSIXError.fromErrno() } + } +} + #endif diff --git a/vminitd/Sources/VminitdCore/IOPair.swift b/vminitd/Sources/VminitdCore/IOPair.swift index 372554d..6c1ec79 100644 --- a/vminitd/Sources/VminitdCore/IOPair.swift +++ b/vminitd/Sources/VminitdCore/IOPair.swift @@ -180,7 +180,16 @@ final class IOPair: Sendable { if mask.isHangup && !mask.readyToRead { self.logger?.debug("received EPOLLHUP with no EPOLLIN") - if !ignoreHup { + // [Nucleic vendored patch] Never close on a bare HUP while a backpressure + // flush is in flight: the writer closing right after its final burst was + // stashed in `pending` (destination momentarily full) used to hit this + // close — whose single best-effort flush pass EAGAINed — and DROP the tail + // of the stream (the CLI's final result line). With pending outstanding, + // just return: the destination's EPOLLOUT edge flushes the backlog, the + // pump then reads the drained closed-writer pipe, observes EOF, and closes + // loss-free. If the destination dies instead, its own error/EPOLLERR wakes + // the write handler, whose failed flush closes the pair — no orphan. + if !ignoreHup && io.pending.isEmpty && !io.writeFdRegistered { io.close(logger: self.logger) } return diff --git a/vminitd/Sources/VminitdCore/ManagedProcess.swift b/vminitd/Sources/VminitdCore/ManagedProcess.swift index 3e91308..0a6eda3 100644 --- a/vminitd/Sources/VminitdCore/ManagedProcess.swift +++ b/vminitd/Sources/VminitdCore/ManagedProcess.swift @@ -27,7 +27,9 @@ import Synchronization final class ManagedProcess: ContainerProcess, Sendable { // swiftlint: disable type_name - protocol IO { + // [Nucleic vendored patch] Sendable so the exit path's delayed grace-close can capture + // the IO existential; both conformers (StandardIO, TerminalIO) already are. + protocol IO: Sendable { func attach(pid: Int32, fd: Int32) throws func start(process: inout Command) throws func resize(size: Terminal.Size) throws @@ -332,10 +334,30 @@ extension ManagedProcess { let exitStatus = ContainerExitStatus(exitCode: status, exitedAt: Date.now) state.exitStatus = exitStatus + // [Nucleic vendored patch] Close only stdin here. Force-closing ALL stdio at + // exit raced the relay: SIGCHLD is serviced before the poller thread drains the + // pipe's final EPOLLIN edge, and the old `io.close()` drain silently dropped + // whatever the destination wouldn't take (`drain` ignores short writes) — the + // CLI's final output line, truncated at process exit. The stdout/stderr write + // ends inside vminitd were already closed by `closeAfterExec`, so the child's + // exit delivers EOF/HUP to each relay, which self-closes loss-free (the relay + // never closes with a backpressure flush in flight). The delayed full close is + // a backstop for a grandchild that inherited the pipes and never exits, or a + // host that never drains — it must never fire on a healthy teardown path. do { - try state.io.close() + try state.io.closeStdin() } catch { - self.log.error("failed to close I/O for process: \(error)") + self.log.error("failed to close stdin for exited process: \(error)") + } + let io = state.io + let log = self.log + DispatchQueue.global(qos: .utility).asyncAfter(deadline: .now() + 8) { + // Idempotent: a relay already self-closed on EOF makes this a no-op. + do { + try io.close() + } catch { + log.error("failed to close I/O for exited process (grace pass): \(error)") + } } for waiter in state.waiters { diff --git a/vminitd/Sources/VminitdCore/ProcessSupervisor.swift b/vminitd/Sources/VminitdCore/ProcessSupervisor.swift index 82e767b..05a6d5d 100644 --- a/vminitd/Sources/VminitdCore/ProcessSupervisor.swift +++ b/vminitd/Sources/VminitdCore/ProcessSupervisor.swift @@ -23,7 +23,18 @@ import Synchronization final class ProcessSupervisor: Sendable { private let poller: Epoll - private let handlers = Mutex<[Int32: @Sendable (Epoll.Mask) -> Void]>([:]) + + /// [Nucleic vendored patch] Handler table keyed by fd, each entry stamped with the + /// registration generation echoed back through epoll (`Epoll.Event.generation`). + /// Dispatch compares the event's generation against the live entry's, so an event + /// queued for a CLOSED registration of a recycled fd number can never fire the new + /// registration's handler (a stale EPOLLHUP used to be able to tear down a brand-new + /// healthy connection that inherited the number mid-batch). + private struct HandlerTable { + var nextGeneration: UInt32 = 1 + var entries: [Int32: (generation: UInt32, handler: @Sendable (Epoll.Mask) -> Void)] = [:] + } + private let handlers = Mutex(HandlerTable()) private let queue: DispatchQueue // `DispatchSourceSignal` is thread-safe. @@ -58,8 +69,13 @@ final class ProcessSupervisor: Sendable { return } for event in events { - let handler = self.handlers.withLock { $0[event.fd] } - handler?(event.mask) + // [Nucleic vendored patch] Dispatch only when the event belongs to the + // CURRENT registration of this fd number — an earlier handler in this + // batch may have closed the fd and something else re-registered the + // recycled number already (see HandlerTable). + let entry = self.handlers.withLock { $0.entries[event.fd] } + guard let entry, entry.generation == event.generation else { continue } + entry.handler(event.mask) } } } @@ -70,23 +86,36 @@ final class ProcessSupervisor: Sendable { /// /// The handler is stored before the fd is added to epoll, ensuring no /// events are missed. + /// + /// [Nucleic vendored patch] Refuses (EEXIST) an fd that is already registered rather + /// than clobbering its handler: the old overwrite-then-fail-EEXIST path destroyed the + /// existing registration's handler AND removed the map entry, leaving the fd armed in + /// epoll with no handler — a silently dead relay (the TerminalIO shared-fd case). func registerFd( _ fd: Int32, mask: Epoll.Mask = [.input, .output], handler: @escaping @Sendable (Epoll.Mask) -> Void ) throws { - self.handlers.withLock { $0[fd] = handler } + let generation: UInt32 = try self.handlers.withLock { table in + guard table.entries[fd] == nil else { throw POSIXError(.EEXIST) } + let generation = table.nextGeneration + // 0 is reserved for the poller's internal eventFD registration. + table.nextGeneration = table.nextGeneration &+ 1 + if table.nextGeneration == 0 { table.nextGeneration = 1 } + table.entries[fd] = (generation, handler) + return generation + } do { - try self.poller.add(fd, mask: mask) + try self.poller.add(fd, mask: mask, generation: generation) } catch { - self.handlers.withLock { _ = $0.removeValue(forKey: fd) } + self.handlers.withLock { _ = $0.entries.removeValue(forKey: fd) } throw error } } /// Remove a file descriptor from epoll monitoring and discard its handler. func unregisterFd(_ fd: Int32) throws { - self.handlers.withLock { _ = $0.removeValue(forKey: fd) } + self.handlers.withLock { _ = $0.entries.removeValue(forKey: fd) } try self.poller.delete(fd) } diff --git a/vminitd/Sources/VminitdCore/StandardIO.swift b/vminitd/Sources/VminitdCore/StandardIO.swift index 49d32c4..45345ac 100644 --- a/vminitd/Sources/VminitdCore/StandardIO.swift +++ b/vminitd/Sources/VminitdCore/StandardIO.swift @@ -50,74 +50,82 @@ final class StandardIO: ManagedProcess.IO & Sendable { func attach(pid: Int32, fd: Int32) throws {} func start(process: inout Command) throws { + // [Nucleic vendored patch] All-or-nothing: a failure partway (a vsock connect or a + // relay registration throwing) used to discard the IO object with earlier pairs + // LIVE — the supervisor's handler map retains a registered IOPair forever, so a + // dead exec left a relay pumping host stdin into a pipe no child would ever read, + // plus its socket, pipe fds, and epoll slot. Dial each socket safely, and on any + // failure close every pair created so far before rethrowing. try self.state.withLock { - if let stdinPort = self.hostStdio.stdin { - let inPipe = Pipe() - process.stdin = inPipe.fileHandleForReading - $0.stdinPipe = inPipe - - let type = VsockType( - port: stdinPort, - cid: VsockType.hostCID - ) - let stdinSocket = try Socket(type: type, closeOnDeinit: false) - try stdinSocket.connect() - - let pair = IOPair( - readFrom: stdinSocket, - writeTo: inPipe.fileHandleForWriting, - reason: "StandardIO stdin", - logger: log - ) - $0.stdin = pair - - try pair.relay() + func dialHost(port: UInt32) throws -> Socket { + let type = VsockType(port: port, cid: VsockType.hostCID) + let socket = try Socket(type: type, closeOnDeinit: false) + do { + try socket.connect() + } catch { + try? socket.close() + throw error + } + return socket } + do { + if let stdinPort = self.hostStdio.stdin { + let inPipe = Pipe() + process.stdin = inPipe.fileHandleForReading + $0.stdinPipe = inPipe - if let stdoutPort = self.hostStdio.stdout { - let outPipe = Pipe() - process.stdout = outPipe.fileHandleForWriting - $0.stdoutPipe = outPipe + let pair = IOPair( + readFrom: try dialHost(port: stdinPort), + writeTo: inPipe.fileHandleForWriting, + reason: "StandardIO stdin", + logger: log + ) + $0.stdin = pair - let type = VsockType( - port: stdoutPort, - cid: VsockType.hostCID - ) - let stdoutSocket = try Socket(type: type, closeOnDeinit: false) - try stdoutSocket.connect() + try pair.relay() + } - let pair = IOPair( - readFrom: outPipe.fileHandleForReading, - writeTo: stdoutSocket, - reason: "StandardIO stdout", - logger: log - ) - $0.stdout = pair + if let stdoutPort = self.hostStdio.stdout { + let outPipe = Pipe() + process.stdout = outPipe.fileHandleForWriting + $0.stdoutPipe = outPipe - try pair.relay() - } + let pair = IOPair( + readFrom: outPipe.fileHandleForReading, + writeTo: try dialHost(port: stdoutPort), + reason: "StandardIO stdout", + logger: log + ) + $0.stdout = pair - if let stderrPort = self.hostStdio.stderr { - let errPipe = Pipe() - process.stderr = errPipe.fileHandleForWriting - $0.stderrPipe = errPipe + try pair.relay() + } - let type = VsockType( - port: stderrPort, - cid: VsockType.hostCID - ) - let stderrSocket = try Socket(type: type, closeOnDeinit: false) - try stderrSocket.connect() + if let stderrPort = self.hostStdio.stderr { + let errPipe = Pipe() + process.stderr = errPipe.fileHandleForWriting + $0.stderrPipe = errPipe - let pair = IOPair( - readFrom: errPipe.fileHandleForReading, - writeTo: stderrSocket, - reason: "StandardIO stderr", - logger: log - ) - $0.stderr = pair + let pair = IOPair( + readFrom: errPipe.fileHandleForReading, + writeTo: try dialHost(port: stderrPort), + reason: "StandardIO stderr", + logger: log + ) + $0.stderr = pair - try pair.relay() + try pair.relay() + } + } catch { + // IOPair.close is idempotent and closes both of a pair's fds; pairs whose + // relay registration never happened are closed the same way. + $0.stdin?.close() + $0.stdin = nil + $0.stdout?.close() + $0.stdout = nil + $0.stderr?.close() + $0.stderr = nil + throw error } } } diff --git a/vminitd/Sources/VminitdCore/TerminalIO.swift b/vminitd/Sources/VminitdCore/TerminalIO.swift index 976e970..b6764af 100644 --- a/vminitd/Sources/VminitdCore/TerminalIO.swift +++ b/vminitd/Sources/VminitdCore/TerminalIO.swift @@ -59,24 +59,45 @@ final class TerminalIO: ManagedProcess.IO & Sendable { process.stdout = nil process.stderr = nil - if let stdinPort = self.hostStdio.stdin { - let type = VsockType( - port: stdinPort, - cid: VsockType.hostCID - ) - let stdinSocket = try Socket(type: type, closeOnDeinit: false) - try stdinSocket.connect() - $0.stdinSocket = stdinSocket - } + // [Nucleic vendored patch] Close whatever connected on a partial failure — + // these sockets are closeOnDeinit: false, so a discarded IO object would leak + // the earlier fd in PID-1. + do { + if let stdinPort = self.hostStdio.stdin { + let type = VsockType( + port: stdinPort, + cid: VsockType.hostCID + ) + let stdinSocket = try Socket(type: type, closeOnDeinit: false) + do { + try stdinSocket.connect() + } catch { + try? stdinSocket.close() + throw error + } + $0.stdinSocket = stdinSocket + } - if let stdoutPort = self.hostStdio.stdout { - let type = VsockType( - port: stdoutPort, - cid: VsockType.hostCID - ) - let stdoutSocket = try Socket(type: type, closeOnDeinit: false) - try stdoutSocket.connect() - $0.stdoutSocket = stdoutSocket + if let stdoutPort = self.hostStdio.stdout { + let type = VsockType( + port: stdoutPort, + cid: VsockType.hostCID + ) + let stdoutSocket = try Socket(type: type, closeOnDeinit: false) + do { + try stdoutSocket.connect() + } catch { + try? stdoutSocket.close() + throw error + } + $0.stdoutSocket = stdoutSocket + } + } catch { + if let stdinSocket = $0.stdinSocket { + try? stdinSocket.close() + $0.stdinSocket = nil + } + throw error } } } @@ -98,13 +119,23 @@ final class TerminalIO: ManagedProcess.IO & Sendable { $0.parent = term if let stdinSocket = $0.stdinSocket { + // [Nucleic vendored patch] The stdin relay's destination is a dup of the + // terminal fd, NOT the terminal fd itself: both relays sharing one number + // meant the stdin side's EPOLLOUT backpressure registration collided with + // the stdout side's read registration (see DupIOCloser) — one bulk paste + // permanently killed the terminal's stdout relay. let pair = IOPair( readFrom: stdinSocket, - writeTo: UnownedIOCloser(term), + writeTo: try DupIOCloser(duplicating: term.fileDescriptor), reason: "TerminalIO stdin", logger: log ) - try pair.relay(ignoreHup: true) + do { + try pair.relay(ignoreHup: true) + } catch { + pair.close() + throw error + } $0.stdin = pair } @@ -115,7 +146,17 @@ final class TerminalIO: ManagedProcess.IO & Sendable { reason: "TerminalIO stdout", logger: log ) - try pair.relay(ignoreHup: true) + do { + try pair.relay(ignoreHup: true) + } catch { + // [Nucleic vendored patch] The stdin pair (and its relay) is already + // live; a discarded IO object would leave it registered and pumping + // forever (the supervisor's handler map retains it). + pair.close() + $0.stdin?.close() + $0.stdin = nil + throw error + } $0.stdout = pair } } @@ -123,11 +164,11 @@ final class TerminalIO: ManagedProcess.IO & Sendable { func close() throws { self.state.withLock { - // stdout must close before stdin because both IOPairs share the - // Terminal fd. stdout registered that fd with epoll (as its read - // source) and needs to unregister it while the fd is still valid. - // stdin closes the Terminal as its write destination, which would - // invalidate the fd before stdout can unregister. + // stdout closes first: it registered the Terminal fd with epoll (as its read + // source) and unregisters it while the fd is still valid. The stdin pair's + // write destination is its own dup of the terminal (see attach), so its + // close-time flush stays valid regardless of ordering — the shared open file + // description outlives the stdout side's close until the dup closes too. if let stdout = $0.stdout { stdout.close() $0.stdout = nil diff --git a/vminitd/Sources/VminitdCore/VsockProxy.swift b/vminitd/Sources/VminitdCore/VsockProxy.swift index b8bbfb7..4c37833 100644 --- a/vminitd/Sources/VminitdCore/VsockProxy.swift +++ b/vminitd/Sources/VminitdCore/VsockProxy.swift @@ -242,7 +242,16 @@ extension VsockProxy { ) } - try relayTo.connect() + // [Nucleic vendored patch] A failed backend connect must close the socket it + // was dialing from: `closeOnDeinit` is false, so throwing out of here (the + // caller closes only `conn`) leaked one PID-1 fd per attempt while the + // backend was down — sustained control-plane churn walked toward EMFILE. + do { + try relayTo.connect() + } catch { + try? relayTo.close() + throw error + } // [Nucleic vendored patch] BOTH fds must be non-blocking BEFORE either is // registered. `Epoll.add` sets O_NONBLOCK only at registration time, and the first @@ -365,6 +374,15 @@ extension VsockProxy { } if mask.isHangup { + // Full hangup of the client. Before tearing down, make one best-effort + // pass toward the SURVIVING peer: bytes already read off the client may + // be parked in toServer's pipe (its earlier flush EAGAINed), and the + // server is still healthy — dropping them loses a delivered-to-us + // control message. One pass only (no spin risk: a still-full server + // just returns and cleanup proceeds). + if !toServerDone { + _ = apply(Self.relayStep(&toServer, description: "client:hangup:toServer", log: self.log)) + } toServerDone = true toClientDone = true } else if mask.isRemoteHangup && !toServerDone { @@ -404,6 +422,11 @@ extension VsockProxy { } if mask.isHangup { + // Mirror of the client handler: flush bytes parked toward the + // surviving client before teardown. + if !toClientDone { + _ = apply(Self.relayStep(&toClient, description: "server:hangup:toClient", log: self.log)) + } toServerDone = true toClientDone = true } else if mask.isRemoteHangup && !toClientDone {