From 6b74d878322367f87bc38e1114123d5ceccf919d Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 13:57:59 -0700 Subject: [PATCH 1/2] Structured, bounded diagnostics for daemon IPC (#289) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a graphcode command timed out, nothing said where the ten seconds went: graphcoded.log held bare 'client connected' lines with no clock and no bound (317 KB and growing). DaemonLog records one key=value line per IPC event — startup identity, connect/disconnect with the peer's pid, each request's kind and phase durations, persist, broadcast (bytes, encode time, fanout), unicast replies, and per-client writes: a write that waited is named at completion with blocked_ms, and a write that is still waiting is named once it has waited 250 ms (write-stall), since a write on a client that never reads again completes never. Never a payload: a command is logged by its case name. Written off the IPC path on a serial queue; rotated at 2 MB into one .1 generation; stdout and stderr move onto the file so nothing keeps writing to an unbounded one. Correlation without a wire change: the daemon logs each connection's peer pid and numbers its frames, and the CLI's timeout says its phase, elapsed time, pid and frame count. Blocked time is measured per stretch by the clock. A regression test keeps a detached writer wedged on a deaf peer retiring within a bounded time, which the accounting depends on. Closes #289. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DeGL2CxuGmq16RSZpJYm2N --- .../Sources/CLI/GraphcodeCommand.swift | 15 + GraphcodeKit/Sources/DaemonIdentity.swift | 19 ++ GraphcodeKit/Sources/GraphStore.swift | 65 +++- GraphcodeKit/Sources/IPC/DaemonLog.swift | 294 ++++++++++++++++++ .../Sources/IPC/OutboundChannel.swift | 84 ++++- GraphcodeKit/Sources/ProjectRegistry.swift | 8 + graphcode-cli/Sources/main.swift | 57 ++-- graphcode/Tests/DaemonDiagnosticsTests.swift | 193 ++++++++++++ .../OutboundChannelBoundedSendTests.swift | 81 +++++ graphcoded/Sources/main.swift | 82 ++++- 10 files changed, 857 insertions(+), 41 deletions(-) create mode 100644 GraphcodeKit/Sources/DaemonIdentity.swift create mode 100644 GraphcodeKit/Sources/IPC/DaemonLog.swift create mode 100644 graphcode/Tests/DaemonDiagnosticsTests.swift create mode 100644 graphcode/Tests/OutboundChannelBoundedSendTests.swift diff --git a/GraphcodeKit/Sources/CLI/GraphcodeCommand.swift b/GraphcodeKit/Sources/CLI/GraphcodeCommand.swift index 527bc811..8230b6fb 100644 --- a/GraphcodeKit/Sources/CLI/GraphcodeCommand.swift +++ b/GraphcodeKit/Sources/CLI/GraphcodeCommand.swift @@ -992,6 +992,21 @@ extension GraphcodeCommand { return "posted #\(latest)\(suffix)" } + /// The timeout the CLI prints: which phase it was in, for how long, and the two + /// numbers that find this run in `graphcoded.log` — its pid (the daemon logs each + /// connection's `peer=`) and how many frames it had sent (the daemon's `seq=`). + public static func renderTimeout( + phase: String, elapsed: TimeInterval, pid: Int32, framesSent: Int + ) -> String { + let seconds = String(format: "%.1f", elapsed) + return """ + timed out after \(seconds)s \(phase) (pid \(pid), \(framesSent) frame\(framesSent == 1 ? "" : "s") \ + sent). The command may still have been applied — check with `graphcode status`. \ + graphcoded.log lines with peer=\(pid) are this run's; seq=\(framesSent) is the frame it \ + was waiting on. + """ + } + public static func describe(_ error: ParseError) -> String { switch error { case .unknownCommand(let name): return "unknown command: \(name)" diff --git a/GraphcodeKit/Sources/DaemonIdentity.swift b/GraphcodeKit/Sources/DaemonIdentity.swift new file mode 100644 index 00000000..30dae967 --- /dev/null +++ b/GraphcodeKit/Sources/DaemonIdentity.swift @@ -0,0 +1,19 @@ +import Foundation + +/// What version of itself a daemon is, for the `startup` line in its log — so version +/// skew between a client and the daemon answering it can be read off the two logs +/// rather than guessed (issue #289). +/// +/// `graphcoded` is a bare executable: Tuist embeds its Info.plist in the binary, where +/// `Bundle.main` still finds it; a SwiftPM build carries none and says so. The +/// executable's inode identity is what `graphcoded` already watches to notice an +/// upgrade underneath itself, so it is logged beside the version as the tie-breaker. +public enum DaemonIdentity { + public static var version: String { + (Bundle.main.infoDictionary?["CFBundleShortVersionString"] as? String) ?? "unversioned" + } + + public static var build: String { + (Bundle.main.infoDictionary?["CFBundleVersion"] as? String) ?? "unversioned" + } +} diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index dc38be82..94ec4321 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -3268,7 +3268,14 @@ public actor GraphStore { // MARK: - Broadcast private func broadcast() { + let started = Date() onGraphChanged?(graph) + DaemonLog.shared.record( + "persist", + DaemonRequestContext.fields + [ + ("nodes", String(graph.nodes.count)), + ("ms", DaemonLog.milliseconds(Date().timeIntervalSince(started))), + ]) notifyClients() } @@ -3284,11 +3291,27 @@ public actor GraphStore { // clients attached it was C encodes of the same snapshot on every change, presence // tick included (issue #288's CPU amplifier). revision += 1 + let started = Date() guard let frame = Self.encode(.graphChanged(graph.wireSnapshot(revision: revision))) else { return } - for id in connections.keys { - deliver(frame, to: id) - } + let encoded = Date() + let intended = connections.count + var accepted = 0 + for id in connections.keys where deliver(frame, to: id) { + accepted += 1 + } + // Sizes and counts only. `ms` is the actor's own time — encode plus handing every + // frame to its channel — and never includes a client's read: that is `write`'s + // `blocked_ms`, per client, which is the field #288 was missing. + DaemonLog.shared.record( + "broadcast", + DaemonRequestContext.fields + [ + ("kind", "graphChanged"), ("revision", String(revision)), + ("bytes", String(frame.data.count)), + ("encode_ms", DaemonLog.milliseconds(encoded.timeIntervalSince(started))), + ("recipients", String(intended)), ("accepted", String(accepted)), + ("ms", DaemonLog.milliseconds(Date().timeIntervalSince(started))), + ]) } /// The presence poll's broadcast — see `DaemonEvent.nodesChanged`. Not superseded: @@ -3301,19 +3324,38 @@ public actor GraphStore { /// never meets a frame it cannot read. Both frames are encoded at most once. private func notifyClients(nodesChanged nodes: [LoopNode]) { revision += 1 + let started = Date() let delta = Self.encode( .nodesChanged(projectPath: graph.project.path, revision: revision, nodes: nodes)) var snapshot: EncodedEvent? + let intended = connections.count + var accepted = 0 + var snapshots = 0 for (id, capabilities) in connectionCapabilities where connections[id] != nil { if capabilities.contains(ClientCapability.nodesChanged.rawValue) { - if let delta { deliver(delta, to: id) } + if let delta, deliver(delta, to: id) { accepted += 1 } } else { + // `deliver` would refuse the delta here on its own; the snapshot is what keeps + // this connection current. if snapshot == nil { snapshot = Self.encode(.graphChanged(graph.wireSnapshot(revision: revision))) } - if let snapshot { deliver(snapshot, to: id) } + if let snapshot, deliver(snapshot, to: id) { + accepted += 1 + snapshots += 1 + } } } + DaemonLog.shared.record( + "broadcast", + [ + ("kind", "nodesChanged"), ("revision", String(revision)), + ("nodes", String(nodes.count)), ("bytes", String(delta?.data.count ?? 0)), + ("snapshot_bytes", String(snapshot?.data.count ?? 0)), + ("recipients", String(intended)), ("accepted", String(accepted)), + ("as_snapshot", String(snapshots)), + ("ms", DaemonLog.milliseconds(Date().timeIntervalSince(started))), + ]) } /// An event as the bytes and the superseding key it goes out with — everything about @@ -3349,8 +3391,12 @@ public actor GraphStore { deliver(frame, to: connectionID) } - private func deliver(_ frame: EncodedEvent, to connectionID: UUID) { - guard let fileDescriptor = connections[connectionID] else { return } + /// Whether the frame was handed to a live channel; `false` also drops the connection + /// — or, for an event the connection never announced it could read, sends nothing + /// and keeps it. + @discardableResult + private func deliver(_ frame: EncodedEvent, to connectionID: UUID) -> Bool { + guard let fileDescriptor = connections[connectionID] else { return false } // The one place the daemon's default is enforced: an event a connection never // announced it could read is not sent to it, whatever call site asked. A caller // that wants such a connection kept current sends it the legacy shape instead @@ -3358,7 +3404,7 @@ public actor GraphStore { if let required = frame.requiredCapability, connectionCapabilities[connectionID]?.contains(required.rawValue) != true { - return + return false } // Queued, never written here: this runs on the `GraphStore` actor, and a // `graphChanged` frame is far larger than a socket's send buffer, so writing it @@ -3372,8 +3418,9 @@ public actor GraphStore { // waiting for the read loop to notice, so a dead connection can't accumulate // failed broadcast attempts. connections.removeValue(forKey: connectionID) - return + return false } + return true } /// Predicate-evaluation state shared between a project store and every sub-graph diff --git a/GraphcodeKit/Sources/IPC/DaemonLog.swift b/GraphcodeKit/Sources/IPC/DaemonLog.swift new file mode 100644 index 00000000..c5942b43 --- /dev/null +++ b/GraphcodeKit/Sources/IPC/DaemonLog.swift @@ -0,0 +1,294 @@ +import Foundation + +#if canImport(Darwin) + import Darwin +#else + import Glibc +#endif + +/// `graphcoded`'s diagnostics — one line per IPC event, timestamped, `key=value`, and +/// never a payload (issue #289). +/// +/// What it is for: when a `graphcode` command times out, saying *where* the ten seconds +/// went — reading, decoding, handling, persisting, encoding, or a write to one client that +/// stopped reading. Before this the daemon's log held bare `client connected` lines with +/// no clock, and the stall reproduced in the #288 report left no trace at all. +/// +/// Three rules every line obeys: +/// +/// - **Sizes and durations, never content.** A record names a command's *kind*, a frame's +/// byte count, a duration — never prompt text, mail bodies, tool output, paths inside +/// the repository, or a raw command. The API takes fields as `(key, value)` pairs, and +/// every caller in the daemon passes numbers, enum case names and descriptor numbers. +/// - **Off the IPC path.** `record` formats one string on the caller's thread and hands +/// it to a serial queue; the file write happens there. Logging never blocks an actor, +/// and never blocks the writer thread whose write it is measuring. +/// - **Bounded.** The file rolls over at `maxBytes` into one `.1` generation, so the +/// most the log ever occupies is two files of that size. `graphcoded.log` had no bound +/// before this — it was 317 KB and growing on the machine that filed #289. +/// +/// launchd hands the daemon `graphcoded.log` as its stdout. This opens the same file for +/// itself and, when stdout is not a terminal, moves stdout and stderr onto its own +/// descriptor — so anything still printed the old way, and the Swift runtime's own crash +/// output, lands in the file this rotates rather than in one that only ever grows. +public final class DaemonLog: @unchecked Sendable { + public static let shared = DaemonLog() + + /// Two generations of this — `graphcoded.log` and `graphcoded.log.1` — is the + /// documented bound. + /// + /// `record` costs its caller one string and one `DispatchQueue.async`: the file is + /// written on the queue, under a lock only the queue takes. Neither the `GraphStore` + /// actor nor a channel's writer thread can be held by the disk — the stall this log + /// exists to measure must not be something the log can cause. + public static let maxBytes = 2 * 1024 * 1024 + public static let fileName = "graphcoded.log" + + private let queue = DispatchQueue(label: "dev.graphcode.graphcoded.log", qos: .utility) + /// Guards the file — descriptor, size, rotation — and is held across `write(2)`. Only + /// the writer queue takes it. `record` never does: a caller must not wait on the disk, + /// or measuring a stall could cause one. + private let fileLock = NSLock() + /// Guards the taps alone, so `record`'s only wait is for another `record`. + private let tapLock = NSLock() + private var descriptor: Int32 = -1 + private var url: URL? + private var bytesWritten = 0 + private var limit = DaemonLog.maxBytes + private var mirrorsStandardStreams = false + private var taps: [UUID: @Sendable (String) -> Void] = [:] + + /// Something worth writing was said while the daemon was writing to a terminal, or + /// before `open` ran — kept nowhere, since there is nothing to keep it in. + public init() {} + + /// Opens `/graphcoded.log` for appending and, when stdout is not a terminal, + /// routes stdout and stderr through it. `maxBytes` is exposed for tests; the daemon + /// uses the default. + public func open(directory: URL, maxBytes: Int = DaemonLog.maxBytes) { + fileLock.lock() + defer { fileLock.unlock() } + limit = maxBytes + url = directory.appendingPathComponent(Self.fileName) + mirrorsStandardStreams = isatty(STDOUT_FILENO) == 0 + openLocked() + } + + /// Sees every line as it is recorded — the test hook, and the way a future surface + /// could stream diagnostics without reading the file. + @discardableResult + public func tap(_ handler: @escaping @Sendable (String) -> Void) -> UUID { + let id = UUID() + tapLock.lock() + taps[id] = handler + tapLock.unlock() + return id + } + + public func untap(_ id: UUID) { + tapLock.lock() + taps.removeValue(forKey: id) + tapLock.unlock() + } + + /// One line: ` event= k=v k=v …`. Values are written as given — + /// callers pass numbers and names, and a value with a space is quoted so the line + /// still splits on whitespace. + public func record(_ event: String, _ fields: [(String, String)] = []) { + var line = Self.stamp() + " event=" + event + for (key, value) in fields { + line += " " + key + "=" + Self.quoted(value) + } + tapLock.lock() + let observers = Array(taps.values) + tapLock.unlock() + for observer in observers { observer(line) } + queue.async { [self] in write(line + "\n") } + } + + /// A duration as the log spells it: milliseconds with one decimal, so a stall of + /// seconds and a write of microseconds read on the same scale. + public static func milliseconds(_ seconds: TimeInterval) -> String { + String(format: "%.1f", seconds * 1000) + } + + /// Flushes what has been recorded so far — for tests reading the file back. + public func drain() { + queue.sync {} + } + + // MARK: - The file + + private func write(_ text: String) { + fileLock.lock() + defer { fileLock.unlock() } + guard descriptor >= 0 else { return } + let data = Data(text.utf8) + if bytesWritten + data.count > limit { rotateLocked() } + data.withUnsafeBytes { raw in + var remaining = raw.count + var pointer = raw.baseAddress! + while remaining > 0 { + #if canImport(Darwin) + let written = Darwin.write(descriptor, pointer, remaining) + #else + let written = Glibc.write(descriptor, pointer, remaining) + #endif + guard written > 0 else { return } + remaining -= written + pointer = pointer.advanced(by: written) + } + } + bytesWritten += data.count + } + + private func openLocked() { + guard let url else { return } + let opened = url.path.withCString { path in + #if canImport(Darwin) + Darwin.open(path, O_WRONLY | O_APPEND | O_CREAT, 0o644) + #else + Glibc.open(path, O_WRONLY | O_APPEND | O_CREAT, 0o644) + #endif + } + guard opened >= 0 else { return } + descriptor = opened + var info = stat() + bytesWritten = fstat(opened, &info) == 0 ? Int(info.st_size) : 0 + if mirrorsStandardStreams { + dup2(opened, STDOUT_FILENO) + dup2(opened, STDERR_FILENO) + } + } + + /// `graphcoded.log` becomes `graphcoded.log.1`, replacing the previous generation, and + /// a fresh file takes its place. Stdout and stderr follow, so nothing keeps writing + /// into the rotated file. + private func rotateLocked() { + guard let url else { return } + #if canImport(Darwin) + _ = Darwin.close(descriptor) + #else + _ = Glibc.close(descriptor) + #endif + descriptor = -1 + let previous = url.path + ".1" + _ = url.path.withCString { current in previous.withCString { rename(current, $0) } } + openLocked() + } + + // MARK: - Formatting + + private static let stampFormatter: DateFormatter = { + let formatter = DateFormatter() + formatter.dateFormat = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'" + formatter.locale = Locale(identifier: "en_US_POSIX") + formatter.timeZone = TimeZone(identifier: "UTC") + return formatter + }() + + private static let stampLock = NSLock() + + private static func stamp() -> String { + stampLock.lock() + defer { stampLock.unlock() } + return stampFormatter.string(from: Date()) + } + + private static func quoted(_ value: String) -> String { + guard value.contains(" ") || value.contains("\"") else { return value } + return "\"" + value.replacingOccurrences(of: "\"", with: "'") + "\"" + } +} + +extension DaemonCommand { + /// The command's shape for a log line — the case name, and for a `graphCommand` the + /// inner case too — never its payload. + public var kindName: String { + switch self { + case .graphCommand(_, let command): return "graphCommand." + command.kindName + default: return Self.caseName(of: self) + } + } + + static func caseName(of value: Any) -> String { + let mirror = Mirror(reflecting: value) + if let label = mirror.children.first?.label { return label } + return String(describing: value) + } +} + +extension GraphCommand { + public var kindName: String { + if case .subGraphCommand(_, let inner) = self { return "subGraphCommand." + inner.kindName } + return DaemonCommand.caseName(of: self) + } +} + +extension DaemonEvent { + public var kindName: String { DaemonCommand.caseName(of: self) } +} + +/// The request being handled, for lines recorded deeper in the daemon — the store's +/// persist and broadcast, the registry's reply — to carry the same `conn`/`seq` as the +/// connection loop's own line, so one command's phases read as one story. +public enum DaemonRequestContext { + public struct Request: Sendable { + public let connection: Int + public let sequence: Int + public init(connection: Int, sequence: Int) { + self.connection = connection + self.sequence = sequence + } + } + + @TaskLocal public static var current: Request? + + /// The fields every line inside a request starts with, or none outside one (the + /// presence poll, a timer). + public static var fields: [(String, String)] { + guard let current else { return [] } + return [("conn", String(current.connection)), ("seq", String(current.sequence))] + } +} + +/// Who is on the other end of a unix socket — the peer's pid, and nothing else. +/// +/// Why a diagnostics path reads peer credentials at all, since it is the one field +/// here that crosses a process boundary: #289 asks that a CLI timeout name an id usable +/// across the CLI's and the daemon's records, and there is no wire change in this +/// series. The pid is the one identifier both sides already know — the CLI prints its +/// own on timeout, the daemon logs the peer's on connect — so the two records can be +/// joined from outside. An id the daemon minted would tell two connections apart in the +/// log but could never be printed by a client that never learns it, which leaves that +/// criterion unsatisfiable without a protocol change. A pid identifies a process, not a +/// person, is visible to anyone on the machine with `ps`, and is used as a credential +/// nowhere. The uid and gid that `SO_PEERCRED` also returns are discarded unread. +public enum SocketPeer { + #if !canImport(Darwin) + /// Linux's `struct ucred`, spelled out: Glibc's Swift module does not export the + /// type, only the `SO_PEERCRED` option that fills it. + private struct PeerCredentials { + var pid: pid_t = 0 + var uid: uid_t = 0 + var gid: gid_t = 0 + } + #endif + + public static func pid(of fileDescriptor: Int32) -> Int32? { + #if canImport(Darwin) + var pid: pid_t = 0 + var size = socklen_t(MemoryLayout.size) + guard getsockopt(fileDescriptor, SOL_LOCAL, LOCAL_PEERPID, &pid, &size) == 0 else { + return nil + } + return pid + #else + var credentials = PeerCredentials() + var size = socklen_t(MemoryLayout.size) + guard getsockopt(fileDescriptor, SOL_SOCKET, SO_PEERCRED, &credentials, &size) == 0 + else { return nil } + return credentials.pid + #endif + } +} diff --git a/GraphcodeKit/Sources/IPC/OutboundChannel.swift b/GraphcodeKit/Sources/IPC/OutboundChannel.swift index 773b213f..e0f7f87b 100644 --- a/GraphcodeKit/Sources/IPC/OutboundChannel.swift +++ b/GraphcodeKit/Sources/IPC/OutboundChannel.swift @@ -164,10 +164,12 @@ final class OutboundChannel: @unchecked Sendable { if pendingBytes - data.count > backlogBudget { // Not a write failure, so nothing else will report it: say so before dropping the // client, or a disconnect this daemon *chose* reads afterwards as one it suffered. - let notice = - "graphcoded: outbound backlog \(pendingBytes)B exceeded on fd \(fileDescriptor)" - + " — dropping a client that stopped reading\n" - FileHandle.standardError.write(Data(notice.utf8)) + DaemonLog.shared.record( + "backlog-drop", + [ + ("fd", String(fileDescriptor)), ("pending_bytes", String(pendingBytes)), + ("budget", String(backlogBudget)), + ]) beginClosingLocked() } @@ -247,7 +249,29 @@ final class OutboundChannel: @unchecked Sendable { pendingBytes -= frame.data.count condition.unlock() - if !writeFrame(frame.data) { + let started = Date() + blockedSeconds = 0 + blockedSince = nil + stallReported = false + let delivered = writeFrame(frame.data) + if let since = blockedSince { + blockedSeconds += Date().timeIntervalSince(since) + blockedSince = nil + } + // Only writes that waited or failed are worth a line: a frame that fits the + // peer's buffer says nothing, and a healthy graph would drown the log in them. + // What this records is the field that names #288 — how long one client held a + // write — kept apart from the broadcast's own duration, which now never waits. + if blockedSeconds > 0 || !delivered { + var fields: [(String, String)] = [ + ("fd", String(fileDescriptor)), ("bytes", String(frame.data.count)), + ("ms", DaemonLog.milliseconds(Date().timeIntervalSince(started))), + ("blocked_ms", DaemonLog.milliseconds(blockedSeconds)), + ] + if !delivered { fields.append(("errno", String(lastErrno))) } + DaemonLog.shared.record("write", fields) + } + if !delivered { // Either the peer is gone or we were asked to stop. Shut the socket down so the // connection loop's reader stops waiting on it too, rather than holding the // connection open on one live half. @@ -263,6 +287,30 @@ final class OutboundChannel: @unchecked Sendable { condition.unlock() } + /// How long the frame just written spent waiting for the peer to make room, and the + /// errno a failed write ended on — read by `pump` for the log line, written only by + /// the writer thread. + private var blockedSeconds: TimeInterval = 0 + private var lastErrno: Int32 = 0 + private var stallReported = false + /// When the current stretch of waiting began — the first `EAGAIN` of a run of them. + /// Measured by the clock rather than by summing `poll` durations: `poll` on a unix + /// socket returns early often enough on macOS that the sum badly undercounts a wait + /// the writer is plainly still in. + private var blockedSince: Date? + + /// Everything the frame has waited so far, the open stretch included. + private var blockedSoFar: TimeInterval { + blockedSeconds + (blockedSince.map { Date().timeIntervalSince($0) } ?? 0) + } + + /// How long a write may wait on its peer before the log names it. A write parked on + /// a client that never reads again *completes* never, so a line written only at + /// completion would say nothing about exactly the client worth knowing about — the + /// #288 stall left no trace for that reason. Past this the writer records the stall + /// once, with what it has waited so far, and the completing line carries the total. + static let stallThreshold: TimeInterval = 0.25 + private var shouldStop: Bool { condition.lock() defer { condition.unlock() } @@ -315,18 +363,40 @@ final class OutboundChannel: @unchecked Sendable { written = write(fileDescriptor, pointer, remaining) } if written > 0 { + if let since = blockedSince { + blockedSeconds += Date().timeIntervalSince(since) + blockedSince = nil + } remaining -= written pointer = pointer.advanced(by: written) continue } if written < 0 && errno == EINTR { continue } - guard written < 0 && (errno == EAGAIN || errno == EWOULDBLOCK) else { return false } + guard written < 0 && (errno == EAGAIN || errno == EWOULDBLOCK) else { + lastErrno = written < 0 ? errno : 0 + return false + } // The peer's buffer is full. Wait for room in slices, so a close is noticed // promptly even when the peer never reads again. + if blockedSince == nil { blockedSince = Date() } var descriptor = pollfd(fd: fileDescriptor, events: Int16(POLLOUT), revents: 0) let ready = poll(&descriptor, 1, Self.writabilityPollMilliseconds) - if ready < 0 && errno != EINTR { return false } + if !stallReported && blockedSoFar >= Self.stallThreshold { + stallReported = true + DaemonLog.shared.record( + "write-stall", + [ + ("fd", String(fileDescriptor)), ("bytes", String(data.count)), + ("remaining", String(remaining)), + ("blocked_ms", DaemonLog.milliseconds(blockedSoFar)), + ]) + } + if ready < 0 && errno != EINTR { + lastErrno = errno + return false + } if ready > 0 && descriptor.revents & Int16(POLLERR | POLLHUP | POLLNVAL) != 0 { + lastErrno = EPIPE return false } } diff --git a/GraphcodeKit/Sources/ProjectRegistry.swift b/GraphcodeKit/Sources/ProjectRegistry.swift index 42417f32..267629eb 100644 --- a/GraphcodeKit/Sources/ProjectRegistry.swift +++ b/GraphcodeKit/Sources/ProjectRegistry.swift @@ -749,7 +749,15 @@ public actor ProjectRegistry { // MARK: - Unicast reply private func send(_ event: DaemonEvent, to fileDescriptor: Int32) { + let started = Date() guard let data = try? JSONEncoder().encode(event) else { return } + DaemonLog.shared.record( + "reply", + DaemonRequestContext.fields + [ + ("kind", event.kindName), ("fd", String(fileDescriptor)), + ("bytes", String(data.count)), + ("encode_ms", DaemonLog.milliseconds(Date().timeIntervalSince(started))), + ]) // Queued rather than written inline for the same reason `GraphStore.send` queues: // this is actor-isolated, and a blocking write of a full-graph frame hands the actor // to whichever client is slowest to read it. No superseding key — a unicast reply diff --git a/graphcode-cli/Sources/main.swift b/graphcode-cli/Sources/main.swift index ac62cd74..2409325e 100644 --- a/graphcode-cli/Sources/main.swift +++ b/graphcode-cli/Sources/main.swift @@ -82,6 +82,20 @@ do { } defer { client.closeConnection() } +/// What this run is waiting for right now, and how many frames it has sent — the two +/// things a timeout has to say for the daemon's log to be searchable for it (#289): +/// `graphcoded.log` records every request under the client's pid (`peer=`) and its +/// sequence on the connection (`seq=`), so "pid 4321, frame 2, waiting for the mailbox +/// answer" names the daemon's line to look at. +let startedAt = Date() +var phase = "connecting to graphcoded" +var framesSent = 0 + +func sendCommand(_ command: DaemonCommand) throws { + try sendCommand(command) + framesSent += 1 +} + /// The calling loop's identity, when this CLI ran inside one — the `status` graph /// render uses it for the board's "unread for you" line, the same attribution every /// mailroom verb derives from `ZMX_SESSION`. @@ -97,7 +111,8 @@ let mailroomReader = SurfaceRef.nodeID( /// error is what turns that into one line saying which path was wrong. @discardableResult func openProject(_ projectPath: String) throws -> LoopGraph? { - try client.send(.openProject(path: projectPath)) + try sendCommand(.openProject(path: projectPath)) + phase = "waiting for the project snapshot" let opened = try client.waitForEvent { switch $0 { case .graphChanged, .errorOccurred: return true @@ -105,6 +120,7 @@ func openProject(_ projectPath: String) throws -> LoopGraph? { } } if case .errorOccurred(let message) = opened { fail(message) } + phase = "waiting for the daemon to acknowledge the command" if case .graphChanged(let graph) = opened { return graph } return nil } @@ -116,7 +132,8 @@ func openProject(_ projectPath: String) throws -> LoopGraph? { /// project not open, a path the daemon does not know) arrives as an error and stops /// here, the way `openProject`'s does. func fetchMailbox(_ projectPath: String, _ query: MailboxQuery) throws -> Mailbox { - try client.send(.mailbox(projectPath: projectPath, query: query)) + try sendCommand(.mailbox(projectPath: projectPath, query: query)) + phase = "waiting for the mailbox answer" let answer = try client.waitForEvent { switch $0 { case .mailbox, .errorOccurred: return true @@ -158,8 +175,9 @@ func runAndPrintGraph(projectPath: String, _ commands: [DaemonCommand]) throws { } for command in commands { - try client.send(command) + try sendCommand(command) } + phase = "waiting for the daemon to acknowledge the command" let event = try client.waitForEvent { if case .graphChanged = $0 { return true } else { return false } } @@ -174,7 +192,7 @@ do { break case .listProjects: - try client.send(.listRecentProjects) + try sendCommand(.listRecentProjects) let event = try client.waitForEvent { if case .recentProjectsListed = $0 { return true } else { return false } } @@ -248,7 +266,7 @@ do { let sender = SurfaceRef.nodeID( fromZmxSessionName: ProcessInfo.processInfo.environment["ZMX_SESSION"] ?? "") try openProject(projectPath) - try client.send( + try sendCommand( .graphCommand( projectPath: projectPath, command: .messageNode(nodeID, text: text, from: sender, followUp: followUp))) @@ -282,7 +300,7 @@ do { FileHandle.standardError.write(Data("\(warning)\n".utf8)) } } - try client.send( + try sendCommand( .graphCommand( projectPath: projectPath, command: .updateNode(nodeID, update: attributed))) // A refusal arrives as an error, an applied update as the changed graph — wait for @@ -304,7 +322,7 @@ do { let promoter = SurfaceRef.nodeID( fromZmxSessionName: ProcessInfo.processInfo.environment["ZMX_SESSION"] ?? "") try openProject(projectPath) - try client.send( + try sendCommand( .graphCommand( projectPath: projectPath, command: .promoteNode(nodeID, promotion: promotion, promotedBy: promoter))) @@ -325,7 +343,7 @@ do { let author = SurfaceRef.nodeID( fromZmxSessionName: ProcessInfo.processInfo.environment["ZMX_SESSION"] ?? "") try openProject(projectPath) - try client.send( + try sendCommand( .graphCommand( projectPath: projectPath, command: .memoNode(nodeID, text: text, from: author))) let memoVerdict = try client.waitForEvent { event in @@ -341,7 +359,7 @@ do { let refiner = SurfaceRef.nodeID( fromZmxSessionName: ProcessInfo.processInfo.environment["ZMX_SESSION"] ?? "") try openProject(projectPath) - try client.send( + try sendCommand( .graphCommand( projectPath: projectPath, command: .refineNode(nodeID, text: text, from: refiner))) let refineVerdict = try client.waitForEvent { event in @@ -357,7 +375,7 @@ do { let requester = SurfaceRef.nodeID( fromZmxSessionName: ProcessInfo.processInfo.environment["ZMX_SESSION"] ?? "") try openProject(projectPath) - try client.send( + try sendCommand( .graphCommand( projectPath: projectPath, command: .rollbackRefinement(nodeID, from: requester))) let rollbackVerdict = try client.waitForEvent { event in @@ -387,7 +405,7 @@ do { let author = SurfaceRef.nodeID( fromZmxSessionName: ProcessInfo.processInfo.environment["ZMX_SESSION"] ?? "") try openProject(projectPath) - try client.send( + try sendCommand( .graphCommand( projectPath: projectPath, command: .mailroomPost(text: text, topic: topic, from: author))) @@ -496,7 +514,7 @@ do { + "($ZMX_SESSION); the mail is delivered to the loop that watches") } try openProject(projectPath) - try client.send( + try sendCommand( .graphCommand( projectPath: projectPath, command: .mailroomWatch(on: on, topic: topic, from: watcher))) @@ -522,7 +540,7 @@ do { // Refresh first: usage is pulled on demand rather than polled, so printing without // asking would show whatever was last read, which could be nothing at all. try openProject(projectPath) - try client.send(.graphCommand(projectPath: projectPath, command: .refreshUsage)) + try sendCommand(.graphCommand(projectPath: projectPath, command: .refreshUsage)) let event = try client.waitForEvent { if case .graphChanged = $0 { return true } else { return false } } @@ -598,7 +616,7 @@ do { fail("the bundle contains no loops") } try openProject(projectPath) - try client.send( + try sendCommand( .graphCommand(projectPath: projectPath, command: .importNodes(request))) let verdict = try client.waitForEvent { event in switch event { @@ -621,12 +639,13 @@ do { } catch DaemonSocketClient.ClientError.timedOut { // Reached only if the daemon accepted the command and then never broadcast anything — // so the useful thing to say is that the command may well have been applied, rather - // than implying it failed. + // than implying it failed — and *where* it was waiting, with the numbers that find + // this run in graphcoded.log. fail( - """ - timed out waiting for graphcoded to answer. The command may still have been applied — \ - check with `graphcode status`. - """, code: ExitCode.ambiguous) + GraphcodeCommand.renderTimeout( + phase: phase, elapsed: Date().timeIntervalSince(startedAt), + pid: ProcessInfo.processInfo.processIdentifier, framesSent: framesSent), + code: ExitCode.ambiguous) } catch FramedMessageIO.IOError.connectionClosed { // graphcoded went away mid-exchange — a restart landing between the write and the // broadcast. Whether it applied the command first is not knowable from here, and the diff --git a/graphcode/Tests/DaemonDiagnosticsTests.swift b/graphcode/Tests/DaemonDiagnosticsTests.swift new file mode 100644 index 00000000..2626140e --- /dev/null +++ b/graphcode/Tests/DaemonDiagnosticsTests.swift @@ -0,0 +1,193 @@ +import Foundation +import GraphcodeKit +import Testing + +#if canImport(Darwin) + import Darwin +#endif + +/// Issue #289's acceptance criteria, one test each: a slow subscriber is named by the +/// write that waited on it; a large graph is recorded as sizes and counts, never content; +/// the CLI's timeout says where it was and how to find itself in the log; and the log +/// stays within its bound. +@Suite(.serialized) +struct DaemonDiagnosticsTests { + /// Every line the shared log records while `body` runs. + private func recording(_ body: () async throws -> Void) async rethrows -> [String] { + final class Lines: @unchecked Sendable { + private let lock = NSLock() + private var lines: [String] = [] + func append(_ line: String) { + lock.lock() + lines.append(line) + lock.unlock() + } + var all: [String] { + lock.lock() + defer { lock.unlock() } + return lines + } + } + let lines = Lines() + let tap = DaemonLog.shared.tap { lines.append($0) } + defer { DaemonLog.shared.untap(tap) } + try await body() + return lines.all + } + + private func fields(_ line: String) -> [String: String] { + var parsed: [String: String] = [:] + for token in line.split(separator: " ").dropFirst() { + guard let equals = token.firstIndex(of: "=") else { continue } + parsed[String(token[.. Data { + try await withCheckedThrowingContinuation { continuation in + DispatchQueue.global().async { + do { + continuation.resume(returning: try FramedMessageIO.readFrame(from: descriptor)) + } catch { + continuation.resume(throwing: error) + } + } + } + } + + /// A graph big enough that its snapshot outgrows a small socket buffer. + private func bigGraph() -> LoopGraph { + var graph = LoopGraph(project: ProjectRef(path: "/tmp/diagnostics", name: "diagnostics")) + for index in 0..<40 { + graph.nodes.append( + LoopNode( + title: "SecretLoopTitle\(index)", loopType: .goalBased, + goal: GoalSpec(summary: String(repeating: "goal text ", count: 40)))) + } + return graph + } + + /// Criterion 1 and 2: one client reads, one never does — the broadcast line records + /// the fanout and the write line names the client that held its write. + @Test + func aSlowSubscriberIsNamedByTheWriteThatWaitedOnIt() async throws { + let store = GraphStore(graph: bigGraph(), onEnsureSession: { _, _ in }) + var reading: [Int32] = [0, 0] + var deaf: [Int32] = [0, 0] + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &reading) == 0) + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &deaf) == 0) + // A small buffer on the deaf client's socket so the snapshot cannot fit in one go. + var small: Int32 = 4096 + setsockopt(deaf[0], SOL_SOCKET, SO_SNDBUF, &small, socklen_t(MemoryLayout.size)) + setsockopt(deaf[1], SOL_SOCKET, SO_RCVBUF, &small, socklen_t(MemoryLayout.size)) + defer { + OutboundChannels.close(reading[0]) + OutboundChannels.close(deaf[0]) + close(reading[1]) + close(deaf[1]) + } + + let lines = try await recording { + await store.addConnection(id: UUID(), fileDescriptor: reading[0]) + await store.addConnection(id: UUID(), fileDescriptor: deaf[0]) + _ = try await frame(from: reading[1]) + await store.handle(.renameNode(store.graph.nodes[0].id, title: "Renamed")) + _ = try await frame(from: reading[1]) + // Past the stall threshold, so the deaf channel's writer has named itself — its + // write never completes, and a line only at completion would never come. + try await Task.sleep(for: .milliseconds(600)) + } + + let broadcast = try #require( + lines.map(fields).first { $0["event"] == "broadcast" && $0["kind"] == "graphChanged" }) + #expect(broadcast["recipients"] == "2") + #expect(broadcast["accepted"] == "2") + #expect(Int(broadcast["bytes"] ?? "") ?? 0 > 4096) + #expect(broadcast["encode_ms"] != nil) + + let deafFD = String(deaf[0]) + let stalled = lines.map(fields).filter { line in + line["event"] == "write-stall" && line["fd"] == deafFD + } + #expect(stalled.count == 1, "the deaf client's stalled write should be named once") + let blocked = Double(stalled.first?["blocked_ms"] ?? "") ?? 0 + let remaining = Int(stalled.first?["remaining"] ?? "") ?? 0 + #expect(blocked >= 250) + #expect(remaining > 0) + let readingFD = String(reading[0]) + let parsed = lines.map(fields) + let writeEvents: Set = ["write", "write-stall"] + let readerWrites = parsed.filter { line in + writeEvents.contains(line["event"] ?? "") && line["fd"] == readingFD + } + #expect(readerWrites.isEmpty, "the reading client never waited, so it is never named") + + // Criterion 3: content never reaches the log — not a title, not a goal. + #expect(!lines.contains { $0.contains("SecretLoopTitle") || $0.contains("goal text") }) + #expect(lines.contains { $0.contains("event=persist") }) + } + + /// Criterion 4: the CLI's message says the phase, the elapsed time, and the two + /// numbers that find the run in the daemon's log. + @Test + func theTimeoutNamesItsPhaseElapsedAndCorrelationNumbers() { + let message = GraphcodeCommand.renderTimeout( + phase: "waiting for the mailbox answer", elapsed: 10.04, pid: 4321, framesSent: 2) + #expect(message.contains("timed out after 10.0s waiting for the mailbox answer")) + #expect(message.contains("pid 4321")) + #expect(message.contains("2 frames sent")) + #expect(message.contains("peer=4321")) + #expect(message.contains("seq=2")) + #expect(message.contains("may still have been applied")) + } + + /// Criterion 5: past the bound the file rolls into one `.1` generation and starts + /// again, so two files of the bound is the most it ever holds. + @Test + func theLogRollsOverAtItsBound() throws { + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("graphcode-log-\(UUID().uuidString)", isDirectory: true) + try FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + defer { try? FileManager.default.removeItem(at: directory) } + let log = DaemonLog() + log.open(directory: directory, maxBytes: 2000) + for index in 0..<60 { + log.record("probe", [("n", String(index)), ("pad", String(repeating: "x", count: 40))]) + } + log.drain() + + let current = directory.appendingPathComponent(DaemonLog.fileName) + let previous = directory.appendingPathComponent(DaemonLog.fileName + ".1") + let currentSize = try FileManager.default.attributesOfItem(atPath: current.path)[.size] as? Int + let previousSize = + try FileManager.default.attributesOfItem(atPath: previous.path)[.size] as? Int + #expect(try #require(currentSize) <= 2000) + #expect(try #require(previousSize) <= 2000) + #expect(try #require(currentSize) + (previousSize ?? 0) < 60 * 100) + let text = try String(contentsOf: current, encoding: .utf8) + #expect(text.contains(" event=probe n=59 ")) + #expect(text.split(separator: "\n").allSatisfy { $0.hasSuffix("Z event=probe n=") == false }) + // Every line carries a UTC stamp with milliseconds. + let stamped = text.split(separator: "\n").allSatisfy { + $0.count > 24 && $0[$0.index($0.startIndex, offsetBy: 23)] == "Z" + } + #expect(stamped) + } + + /// A command's kind is its case name and never its payload, down through a + /// sub-graph command. + @Test + func aCommandIsLoggedByKindNotContent() { + let memo = DaemonCommand.graphCommand( + projectPath: "/tmp/p", command: .memoNode(UUID(), text: "the secret", from: nil)) + #expect(memo.kindName == "graphCommand.memoNode") + #expect(!memo.kindName.contains("secret")) + #expect(DaemonCommand.listRecentProjects.kindName == "listRecentProjects") + let nested = DaemonCommand.graphCommand( + projectPath: "/tmp/p", + command: .subGraphCommand(nodeID: UUID(), command: .restartSessions)) + #expect(nested.kindName == "graphCommand.subGraphCommand.restartSessions") + #expect(DaemonEvent.errorOccurred("x").kindName == "errorOccurred") + } +} diff --git a/graphcode/Tests/OutboundChannelBoundedSendTests.swift b/graphcode/Tests/OutboundChannelBoundedSendTests.swift new file mode 100644 index 00000000..1724230c --- /dev/null +++ b/graphcode/Tests/OutboundChannelBoundedSendTests.swift @@ -0,0 +1,81 @@ +import Foundation +import Testing + +@testable import GraphcodeKit + +#if canImport(Darwin) + import Darwin +#endif + +/// A writer on a peer that never reads must still return between slices, or closing +/// that connection hangs the caller — the promise #291 made for `closeAndWait`. On macOS +/// `MSG_DONTWAIT` does not keep it (the flag is ignored on a blocking unix socket); the +/// socket's send timeout does. +@Suite +struct OutboundChannelBoundedSendTests { + private func deafPair() -> (daemon: Int32, peer: Int32) { + var pair: [Int32] = [0, 0] + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) == 0) + var small: Int32 = 4096 + setsockopt(pair[0], SOL_SOCKET, SO_SNDBUF, &small, socklen_t(MemoryLayout.size)) + setsockopt(pair[1], SOL_SOCKET, SO_RCVBUF, &small, socklen_t(MemoryLayout.size)) + return (pair[0], pair[1]) + } + + /// `closeAndWait` after `shutdown` is bounded either way — `shutdown` wakes a send + /// parked on a unix socket. The path that cannot shut the socket down is `detach`, + /// for a descriptor number a new connection has already taken over: there the writer + /// must notice `isClosing` between slices on its own, and a writer parked inside + /// `send(2)` never does. Without the send timeout this waits for the peer's lifetime. + @Test + func aDetachedWriterWedgedOnADeafPeerRetiresWithinABoundedTime() async throws { + let (daemon, peer) = deafPair() + defer { + close(daemon) + close(peer) + } + let channel = OutboundChannel(fileDescriptor: daemon) + #expect(channel.send(Data(repeating: 0x78, count: 60_000))) + // Let the writer fill the peer and park. + try await Task.sleep(for: .milliseconds(300)) + + let started = Date() + let retired = Task.detached { + channel.detach() + channel.closeAndWait() + } + let outcome = await withTaskGroup(of: Bool.self) { group in + group.addTask { + await retired.value + return true + } + group.addTask { + try? await Task.sleep(for: .seconds(3)) + return false + } + let first = await group.next() ?? false + group.cancelAll() + return first + } + #expect(outcome, "a detached writer did not retire within 3 s of a wedged peer") + #expect(Date().timeIntervalSince(started) < 3) + } + + @Test + func aSlowReaderStillGetsTheWholeFrame() async throws { + let (daemon, peer) = deafPair() + defer { + OutboundChannels.close(daemon) + close(peer) + } + OutboundChannels.open(daemon) + let payload = Data(repeating: 0x79, count: 20_000) + #expect(OutboundChannels.send(payload, to: daemon)) + // Drain slowly, off the writer's thread, and check the frame arrives intact. + let received = await Task.detached { () -> Data? in + try? await Task.sleep(for: .milliseconds(200)) + return try? FramedMessageIO.readFrame(from: peer) + }.value + #expect(received == payload) + } +} diff --git a/graphcoded/Sources/main.swift b/graphcoded/Sources/main.swift index a1369821..ca79f619 100644 --- a/graphcoded/Sources/main.swift +++ b/graphcoded/Sources/main.swift @@ -79,7 +79,20 @@ guard listen(socketDescriptor, 8) == 0 else { fail("failed to listen on \(path) (errno \(errno))") } -FileHandle.standardOutput.write(Data("graphcoded: listening on \(path)\n".utf8)) +// The diagnostics file, opened before the first line worth keeping. Stdout and stderr +// move onto it (launchd pointed them at the same file, unbounded), so from here every +// line the daemon writes is timestamped, structured, and rotated. +DaemonLog.shared.open(directory: supportDirectory) +DaemonLog.shared.record( + "startup", + [ + ("pid", String(getpid())), + ("version", DaemonIdentity.version), + ("build", DaemonIdentity.build), + ("executable", CommandLine.arguments[0]), + ("support", supportDirectory.path), + ("socket", path), + ]) // A broadcast writes to every connected client, and a client can vanish without a // clean close — the app killed, a CLI exiting early, a pane crashing. The write then @@ -142,8 +155,7 @@ func makeStalenessTimer() -> DispatchSourceTimer? { timer.setEventHandler { guard let current = ExecutableIdentity.of(path: executablePath), current != launchIdentity else { return } - FileHandle.standardOutput.write( - Data("graphcoded: binary replaced on disk; exiting for launchd to start the new one\n".utf8)) + DaemonLog.shared.record("shutdown", [("reason", "binary-replaced")]) unlink(path) exit(0) } @@ -172,12 +184,41 @@ let registry = ProjectRegistry( } } +/// A counter safe to bump from the accept loop's thread while connection tasks read it. +final class ManagedAtomic: @unchecked Sendable { + private var value: Int + private let lock = NSLock() + init(_ value: Int) { self.value = value } + func next() -> Int { + lock.lock() + defer { lock.unlock() } + value += 1 + return value + } +} + +/// Numbers connections in the order they arrived — the `conn` every line about one +/// carries, short enough to read and stable for the daemon's lifetime. +let connectionCounter = ManagedAtomic(0) + func handleConnection(_ fileDescriptor: Int32) { Task { let connectionID = UUID() + let connection = connectionCounter.next() + let peer = SocketPeer.pid(of: fileDescriptor) + let connected = Date() // `addConnection` opens this connection's outbound channel as it registers it. await registry.addConnection(id: connectionID, fileDescriptor: fileDescriptor) - FileHandle.standardOutput.write(Data("graphcoded: client connected\n".utf8)) + // `peer` is the client's pid: what a `graphcode` invocation prints when it times + // out, so its complaint and these lines can be matched up with no id on the wire. + DaemonLog.shared.record( + "connect", + [ + ("conn", String(connection)), ("fd", String(fileDescriptor)), + ("peer", peer.map(String.init) ?? "?"), + ]) + var sequence = 0 + var requests = 0 while true { let data: Data do { @@ -185,10 +226,33 @@ func handleConnection(_ fileDescriptor: Int32) { } catch { break } + sequence += 1 + let received = Date() + let request = DaemonRequestContext.Request(connection: connection, sequence: sequence) do { let command = try JSONDecoder().decode(DaemonCommand.self, from: data) - await registry.handle(command, connectionID: connectionID) + let decoded = Date() + await DaemonRequestContext.$current.withValue(request) { + await registry.handle(command, connectionID: connectionID) + } + requests += 1 + // The request's own line: what kind, how big, and how long each phase took. + // Never the payload — a `graphCommand.memoNode` is logged as exactly that. + DaemonLog.shared.record( + "request", + [ + ("conn", String(connection)), ("seq", String(sequence)), + ("kind", command.kindName), ("bytes", String(data.count)), + ("decode_ms", DaemonLog.milliseconds(decoded.timeIntervalSince(received))), + ("handle_ms", DaemonLog.milliseconds(Date().timeIntervalSince(decoded))), + ]) } catch { + DaemonLog.shared.record( + "request", + [ + ("conn", String(connection)), ("seq", String(sequence)), ("kind", "undecodable"), + ("bytes", String(data.count)), + ]) // A frame that read fine but didn't decode is version skew, not a dead socket: // a newer CLI sent a command this daemon predates. Dropping the connection here // failed *silently* — the client just saw a hang-up — so answer instead and @@ -206,7 +270,13 @@ func handleConnection(_ fileDescriptor: Int32) { // it. `close` tears the socket down, which is also what unblocks a writer parked on a // peer that stopped reading. OutboundChannels.close(fileDescriptor) - FileHandle.standardOutput.write(Data("graphcoded: client disconnected\n".utf8)) + DaemonLog.shared.record( + "disconnect", + [ + ("conn", String(connection)), ("fd", String(fileDescriptor)), + ("requests", String(requests)), + ("lifetime_ms", DaemonLog.milliseconds(Date().timeIntervalSince(connected))), + ]) } } From ba2dba05a53f343b3a086cb46bbc35d6121bb6b2 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 16:58:18 -0700 Subject: [PATCH 2/2] Fix sendCommand calling itself (infinite recursion on every CLI verb) --- .github/workflows/linux.yml | 3 ++ GraphcodeKit/Sources/GraphStore.swift | 2 +- GraphcodeKit/Sources/IPC/DaemonLog.swift | 29 +++++++++++--- .../Sources/IPC/FramedMessageIO.swift | 4 ++ .../Sources/IPC/OutboundChannel.swift | 26 +++++++++--- GraphcodeKit/Sources/ProjectRegistry.swift | 2 +- graphcode-cli/Sources/main.swift | 2 +- graphcode/Tests/DaemonDiagnosticsTests.swift | 19 +++++++-- graphcoded/Sources/main.swift | 21 ++++++---- scripts/cli-smoke.sh | 40 +++++++++++++++++++ 10 files changed, 124 insertions(+), 24 deletions(-) create mode 100755 scripts/cli-smoke.sh diff --git a/.github/workflows/linux.yml b/.github/workflows/linux.yml index a8be2637..66d13ccd 100644 --- a/.github/workflows/linux.yml +++ b/.github/workflows/linux.yml @@ -20,3 +20,6 @@ jobs: - uses: actions/checkout@v4 - run: swift build - run: .build/debug/graphcode + # The built CLI against a throwaway daemon — exit 0 on the verbs that dial it. + # A runtime fault in the CLI passes every scheme build; this is what catches it. + - run: scripts/cli-smoke.sh diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index 94ec4321..da05006c 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -515,7 +515,7 @@ public actor GraphStore { // the same way the registry does. Without this a store could bind to a channel left // dead on a recycled descriptor number and drop the client as disconnected on the // snapshot it was joining for. - OutboundChannels.open(fileDescriptor) + OutboundChannels.open(fileDescriptor, tag: id.tag) connections[id] = fileDescriptor send(.graphChanged(graph.wireSnapshot(revision: revision)), to: id) } diff --git a/GraphcodeKit/Sources/IPC/DaemonLog.swift b/GraphcodeKit/Sources/IPC/DaemonLog.swift index c5942b43..38ec5211 100644 --- a/GraphcodeKit/Sources/IPC/DaemonLog.swift +++ b/GraphcodeKit/Sources/IPC/DaemonLog.swift @@ -24,8 +24,10 @@ import Foundation /// it to a serial queue; the file write happens there. Logging never blocks an actor, /// and never blocks the writer thread whose write it is measuring. /// - **Bounded.** The file rolls over at `maxBytes` into one `.1` generation, so the -/// most the log ever occupies is two files of that size. `graphcoded.log` had no bound -/// before this — it was 317 KB and growing on the machine that filed #289. +/// most the log ever occupies is two files of that size — plus whatever the mirrored +/// stdout and stderr wrote between two records, since rotation is decided at each +/// record from the file's real size. `graphcoded.log` had no bound before this — it +/// was 317 KB and growing on the machine that filed #289. /// /// launchd hands the daemon `graphcoded.log` as its stdout. This opens the same file for /// itself and, when stdout is not a terminal, moves stdout and stderr onto its own @@ -65,12 +67,17 @@ public final class DaemonLog: @unchecked Sendable { /// Opens `/graphcoded.log` for appending and, when stdout is not a terminal, /// routes stdout and stderr through it. `maxBytes` is exposed for tests; the daemon /// uses the default. - public func open(directory: URL, maxBytes: Int = DaemonLog.maxBytes) { + /// `mirroringStandardStreams` routes this process's stdout and stderr through the file + /// (skipped when stdout is a terminal). Only the daemon asks for it: a test that + /// opened a log with it would move the test runner's own output onto a temp file. + public func open( + directory: URL, maxBytes: Int = DaemonLog.maxBytes, mirroringStandardStreams: Bool = false + ) { fileLock.lock() defer { fileLock.unlock() } limit = maxBytes url = directory.appendingPathComponent(Self.fileName) - mirrorsStandardStreams = isatty(STDOUT_FILENO) == 0 + mirrorsStandardStreams = mirroringStandardStreams && isatty(STDOUT_FILENO) == 0 openLocked() } @@ -124,7 +131,13 @@ public final class DaemonLog: @unchecked Sendable { defer { fileLock.unlock() } guard descriptor >= 0 else { return } let data = Data(text.utf8) - if bytesWritten + data.count > limit { rotateLocked() } + // The file's real size, not a running count of this log's own lines: stdout and + // stderr write to the same file through the mirrored descriptors, and those bytes + // count against the bound too — a bound that only saw its own records was + // measured 25× over. + var info = stat() + let size = fstat(descriptor, &info) == 0 ? Int(info.st_size) : bytesWritten + if size + data.count > limit { rotateLocked() } data.withUnsafeBytes { raw in var remaining = raw.count var pointer = raw.baseAddress! @@ -292,3 +305,9 @@ public enum SocketPeer { #endif } } + +extension UUID { + /// The first eight characters — enough to tell connections apart in a log, short + /// enough to read; the daemon's `connect` line carries it as `id=`. + public var tag: String { String(uuidString.prefix(8)) } +} diff --git a/GraphcodeKit/Sources/IPC/FramedMessageIO.swift b/GraphcodeKit/Sources/IPC/FramedMessageIO.swift index 00231b89..b866776c 100644 --- a/GraphcodeKit/Sources/IPC/FramedMessageIO.swift +++ b/GraphcodeKit/Sources/IPC/FramedMessageIO.swift @@ -27,6 +27,10 @@ import Foundation /// workspace opened (0.1.46-beta1): its three launch commands wait together on the /// just-bootstrapped daemon and are released at the same instant. public enum FramedMessageIO { + /// The 4-byte big-endian length that precedes every frame — what a byte count on the + /// wire includes beyond the message itself. + public static let headerLength = 4 + public enum IOError: Error, Equatable { case connectionClosed case readFailed(errno: Int32) diff --git a/GraphcodeKit/Sources/IPC/OutboundChannel.swift b/GraphcodeKit/Sources/IPC/OutboundChannel.swift index e0f7f87b..3676ac93 100644 --- a/GraphcodeKit/Sources/IPC/OutboundChannel.swift +++ b/GraphcodeKit/Sources/IPC/OutboundChannel.swift @@ -58,6 +58,11 @@ final class OutboundChannel: @unchecked Sendable { /// A non-socket cannot block a writer the way a peer that stopped reading can, so a /// plain `write` is both correct and sufficient there. private let isSocket: Bool + /// Which connection this is, for the log — descriptor numbers are reused within + /// minutes, so a line keyed on `fd` alone cannot name a client. The registry passes + /// the connection's id (its first eight characters), the same `id=` its `connect` + /// line carries. + private let tag: String? private let condition = NSCondition() private var pending: [Frame] = [] private var pendingBytes = 0 @@ -66,7 +71,11 @@ final class OutboundChannel: @unchecked Sendable { /// `backlogBudget` is injectable so tests can exercise the valve without moving /// megabytes through a socket to reach it. - init(fileDescriptor: Int32, backlogBudget: Int = OutboundChannel.maxBacklogBytes) { + init( + fileDescriptor: Int32, backlogBudget: Int = OutboundChannel.maxBacklogBytes, + tag: String? = nil + ) { + self.tag = tag self.fileDescriptor = fileDescriptor self.backlogBudget = backlogBudget var socketType: Int32 = 0 @@ -167,7 +176,8 @@ final class OutboundChannel: @unchecked Sendable { DaemonLog.shared.record( "backlog-drop", [ - ("fd", String(fileDescriptor)), ("pending_bytes", String(pendingBytes)), + ("conn", tag ?? "?"), ("fd", String(fileDescriptor)), + ("pending_bytes", String(pendingBytes)), ("budget", String(backlogBudget)), ]) beginClosingLocked() @@ -264,7 +274,8 @@ final class OutboundChannel: @unchecked Sendable { // write — kept apart from the broadcast's own duration, which now never waits. if blockedSeconds > 0 || !delivered { var fields: [(String, String)] = [ - ("fd", String(fileDescriptor)), ("bytes", String(frame.data.count)), + ("conn", tag ?? "?"), ("fd", String(fileDescriptor)), + ("bytes", String(frame.data.count + FramedMessageIO.headerLength)), ("ms", DaemonLog.milliseconds(Date().timeIntervalSince(started))), ("blocked_ms", DaemonLog.milliseconds(blockedSeconds)), ] @@ -386,7 +397,8 @@ final class OutboundChannel: @unchecked Sendable { DaemonLog.shared.record( "write-stall", [ - ("fd", String(fileDescriptor)), ("bytes", String(data.count)), + ("conn", tag ?? "?"), ("fd", String(fileDescriptor)), + ("bytes", String(data.count + FramedMessageIO.headerLength)), ("remaining", String(remaining)), ("blocked_ms", DaemonLog.milliseconds(blockedSoFar)), ]) @@ -439,7 +451,9 @@ public enum OutboundChannels { /// dropped as disconnected the moment it arrived. /// `backlogBudget` is `nil` for the standard budget; tests inject a small one so the /// valve can be exercised without moving megabytes through a socket. - public static func open(_ fileDescriptor: Int32, backlogBudget: Int? = nil) { + public static func open( + _ fileDescriptor: Int32, backlogBudget: Int? = nil, tag: String? = nil + ) { lock.lock() let existing = channels[fileDescriptor] guard existing?.isAlive != true else { @@ -448,7 +462,7 @@ public enum OutboundChannels { } channels[fileDescriptor] = OutboundChannel( fileDescriptor: fileDescriptor, - backlogBudget: backlogBudget ?? OutboundChannel.maxBacklogBytes) + backlogBudget: backlogBudget ?? OutboundChannel.maxBacklogBytes, tag: tag) lock.unlock() // Detached rather than closed: the number belongs to the connection being opened // here, so shutting it down would tear that one down. diff --git a/GraphcodeKit/Sources/ProjectRegistry.swift b/GraphcodeKit/Sources/ProjectRegistry.swift index 267629eb..97c796db 100644 --- a/GraphcodeKit/Sources/ProjectRegistry.swift +++ b/GraphcodeKit/Sources/ProjectRegistry.swift @@ -125,7 +125,7 @@ public actor ProjectRegistry { // and a descriptor with no channel silently delivers nothing. It also gives a reused // descriptor number a fresh channel, so nothing inherits a previous connection's // writer. - OutboundChannels.open(fileDescriptor) + OutboundChannels.open(fileDescriptor, tag: id.tag) connectionFileDescriptors[id] = fileDescriptor startPresencePolling() } diff --git a/graphcode-cli/Sources/main.swift b/graphcode-cli/Sources/main.swift index 2409325e..c9df34c4 100644 --- a/graphcode-cli/Sources/main.swift +++ b/graphcode-cli/Sources/main.swift @@ -92,7 +92,7 @@ var phase = "connecting to graphcoded" var framesSent = 0 func sendCommand(_ command: DaemonCommand) throws { - try sendCommand(command) + try client.send(command) framesSent += 1 } diff --git a/graphcode/Tests/DaemonDiagnosticsTests.swift b/graphcode/Tests/DaemonDiagnosticsTests.swift index 2626140e..cb26b882 100644 --- a/graphcode/Tests/DaemonDiagnosticsTests.swift +++ b/graphcode/Tests/DaemonDiagnosticsTests.swift @@ -88,9 +88,10 @@ struct DaemonDiagnosticsTests { close(deaf[1]) } + let deafID = UUID() let lines = try await recording { await store.addConnection(id: UUID(), fileDescriptor: reading[0]) - await store.addConnection(id: UUID(), fileDescriptor: deaf[0]) + await store.addConnection(id: deafID, fileDescriptor: deaf[0]) _ = try await frame(from: reading[1]) await store.handle(.renameNode(store.graph.nodes[0].id, title: "Renamed")) _ = try await frame(from: reading[1]) @@ -110,6 +111,8 @@ struct DaemonDiagnosticsTests { let stalled = lines.map(fields).filter { line in line["event"] == "write-stall" && line["fd"] == deafFD } + // Named by connection as well as by descriptor: descriptor numbers are reused. + #expect(stalled.first?["conn"] == deafID.tag) #expect(stalled.count == 1, "the deaf client's stalled write should be named once") let blocked = Double(stalled.first?["blocked_ms"] ?? "") ?? 0 let remaining = Int(stalled.first?["remaining"] ?? "") ?? 0 @@ -156,17 +159,27 @@ struct DaemonDiagnosticsTests { log.record("probe", [("n", String(index)), ("pad", String(repeating: "x", count: 40))]) } log.drain() + // Bytes that reach the file some other way — the daemon's mirrored stdout and + // stderr in production — count against the bound too: the next record rolls over. + let handle = try FileHandle(forWritingTo: directory.appendingPathComponent(DaemonLog.fileName)) + try handle.seekToEnd() + try handle.write(contentsOf: Data(repeating: 0x2e, count: 1900)) + try handle.close() + log.record("probe", [("n", "60")]) + log.drain() let current = directory.appendingPathComponent(DaemonLog.fileName) let previous = directory.appendingPathComponent(DaemonLog.fileName + ".1") let currentSize = try FileManager.default.attributesOfItem(atPath: current.path)[.size] as? Int let previousSize = try FileManager.default.attributesOfItem(atPath: previous.path)[.size] as? Int + // A generation is bounded by the limit plus whatever reached the file between two + // records (the mirrored streams write directly): the next record rolls it over. #expect(try #require(currentSize) <= 2000) - #expect(try #require(previousSize) <= 2000) + #expect(try #require(previousSize) <= 2000 + 1900) #expect(try #require(currentSize) + (previousSize ?? 0) < 60 * 100) let text = try String(contentsOf: current, encoding: .utf8) - #expect(text.contains(" event=probe n=59 ")) + #expect(text.contains(" event=probe n=60")) #expect(text.split(separator: "\n").allSatisfy { $0.hasSuffix("Z event=probe n=") == false }) // Every line carries a UTC stamp with milliseconds. let stamped = text.split(separator: "\n").allSatisfy { diff --git a/graphcoded/Sources/main.swift b/graphcoded/Sources/main.swift index ca79f619..fd58d82f 100644 --- a/graphcoded/Sources/main.swift +++ b/graphcoded/Sources/main.swift @@ -82,7 +82,7 @@ guard listen(socketDescriptor, 8) == 0 else { // The diagnostics file, opened before the first line worth keeping. Stdout and stderr // move onto it (launchd pointed them at the same file, unbounded), so from here every // line the daemon writes is timestamped, structured, and rotated. -DaemonLog.shared.open(directory: supportDirectory) +DaemonLog.shared.open(directory: supportDirectory, mirroringStandardStreams: true) DaemonLog.shared.record( "startup", [ @@ -214,7 +214,7 @@ func handleConnection(_ fileDescriptor: Int32) { DaemonLog.shared.record( "connect", [ - ("conn", String(connection)), ("fd", String(fileDescriptor)), + ("conn", String(connection)), ("id", connectionID.tag), ("fd", String(fileDescriptor)), ("peer", peer.map(String.init) ?? "?"), ]) var sequence = 0 @@ -232,18 +232,25 @@ func handleConnection(_ fileDescriptor: Int32) { do { let command = try JSONDecoder().decode(DaemonCommand.self, from: data) let decoded = Date() + // Logged on receipt, before anything is handled: a request that hangs is + // exactly the one worth a line, and a line written on completion would never + // come. Never the payload — a `graphCommand.memoNode` is logged as exactly that. + DaemonLog.shared.record( + "request", + [ + ("conn", String(connection)), ("seq", String(sequence)), + ("kind", command.kindName), ("bytes", String(data.count)), + ("decode_ms", DaemonLog.milliseconds(decoded.timeIntervalSince(received))), + ]) await DaemonRequestContext.$current.withValue(request) { await registry.handle(command, connectionID: connectionID) } requests += 1 - // The request's own line: what kind, how big, and how long each phase took. - // Never the payload — a `graphCommand.memoNode` is logged as exactly that. DaemonLog.shared.record( - "request", + "handled", [ ("conn", String(connection)), ("seq", String(sequence)), - ("kind", command.kindName), ("bytes", String(data.count)), - ("decode_ms", DaemonLog.milliseconds(decoded.timeIntervalSince(received))), + ("kind", command.kindName), ("handle_ms", DaemonLog.milliseconds(Date().timeIntervalSince(decoded))), ]) } catch { diff --git a/scripts/cli-smoke.sh b/scripts/cli-smoke.sh new file mode 100755 index 00000000..26529a25 --- /dev/null +++ b/scripts/cli-smoke.sh @@ -0,0 +1,40 @@ +#!/bin/sh +# Runs the built CLI against a throwaway daemon and insists on exit 0 — the check no +# scheme build can make. The `graphcode` test scheme does not build `graphcode-cli`, +# a scheme build of it cannot see a runtime fault (a helper that called itself took +# every verb down with SIGSEGV and passed three green gates), and no test invokes the +# binary. This does. SwiftPM products, so it runs identically on macOS and Linux CI. +set -eu +root="$(cd "$(dirname "$0")/.." && pwd)" +bin="$root/.build/debug" +[ -x "$bin/graphcode" ] && [ -x "$bin/graphcoded" ] || swift build --package-path "$root" + +# Short: sockaddr_un.sun_path is 104 bytes on Darwin. +support="$(mktemp -d /tmp/gcsmoke.XXXXXX)" +project="$support/project" +mkdir -p "$project" +export GRAPHCODE_SUPPORT_DIR="$support" +"$bin/graphcoded" > "$support/daemon.out" 2>&1 & +daemon=$! +trap 'kill "$daemon" 2>/dev/null; wait "$daemon" 2>/dev/null; rm -rf "$support"' EXIT INT TERM + +for _ in 1 2 3 4 5 6 7 8 9 10; do + [ -S "$support/graphcoded.sock" ] && break + sleep 0.5 +done +[ -S "$support/graphcoded.sock" ] || { echo "smoke: daemon never listened"; cat "$support/daemon.out"; exit 1; } + +fail=0 +check() { + if "$@" > "$support/out" 2>&1; then + echo "smoke: ok $*" + else + echo "smoke: FAIL $* (exit $?)"; cat "$support/out"; fail=1 + fi +} +check "$bin/graphcode" status "$project" +check "$bin/graphcode" mail list "$project" +check "$bin/graphcode" mail post "$project" --topic smoke "the smoke test was here" +check "$bin/graphcode" mail read "$project" 1 +check "$bin/graphcode" projects +exit "$fail"