diff --git a/GraphcodeKit/Sources/Domain/LoopGraph.swift b/GraphcodeKit/Sources/Domain/LoopGraph.swift index bb6e55e5..ca4b9258 100644 --- a/GraphcodeKit/Sources/Domain/LoopGraph.swift +++ b/GraphcodeKit/Sources/Domain/LoopGraph.swift @@ -39,6 +39,11 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable { /// Set only on the copy a daemon sends (`wireSnapshot()`); `nil` on the graph the /// daemon owns and persists. public var mailroomDigest: MailroomDigest? + /// Where this snapshot sits in the daemon's sequence of frames for the graph — see + /// `DaemonEvent.nodesChanged`. Wire-only like `mailroomDigest`: stamped on the copy a + /// daemon sends, `nil` on the graph it owns and persists, and on a snapshot from a + /// daemon that predates deltas. + public var revision: Int? public var project: ProjectRef { get { scope.projectRef } @@ -55,13 +60,26 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable { /// This graph as a `.graphChanged` frame carries it: the posts stripped and their /// digest stamped in their place. Everything else is the graph exactly as it is. - public func wireSnapshot() -> LoopGraph { + public func wireSnapshot(revision: Int? = nil) -> LoopGraph { var copy = self copy.mailroomDigest = MailroomDigest(of: mailroom) copy.mailroom = [] + copy.revision = revision return copy } + /// This graph with `nodes` replaced by id — how a client applies a + /// `DaemonEvent.nodesChanged` to the snapshot it holds. A node the graph does not + /// have is ignored: a delta says how a loop changed, never that one appeared. + public func applying(nodesChanged nodes: [LoopNode], revision: Int) -> LoopGraph { + var merged = self + for node in nodes where merged.nodes[id: node.id] != nil { + merged.nodes[id: node.id] = node + } + merged.revision = revision + return merged + } + public init( id: UUID = UUID(), scope: LoopGraphScope, @@ -278,7 +296,7 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable { // MARK: - Coding private enum CodingKeys: String, CodingKey { - case id, nodes, edges, mailroom, mailroomDigest + case id, nodes, edges, mailroom, mailroomDigest, revision /// Persisted as a `ProjectRef` rather than as the scope enum. Every graph on disk /// predates `LoopGraphScope`, and the ref round-trips both cases losslessly (the /// global graph's reserved path decodes straight back to `.global`), so there was @@ -297,6 +315,7 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable { try container.decodeIfPresent([MailroomPost].self, forKey: .mailroom) ?? decoder.legacyMailroomValue([MailroomPost].self, "artifactory") ?? [] mailroomDigest = try container.decodeIfPresent(MailroomDigest.self, forKey: .mailroomDigest) + revision = try container.decodeIfPresent(Int.self, forKey: .revision) } public func encode(to encoder: Encoder) throws { @@ -309,5 +328,6 @@ public struct LoopGraph: Identifiable, Codable, Equatable, Sendable { // what it was — the same reason `hasActiveDependents` never reaches disk. if !mailroom.isEmpty { try container.encode(mailroom, forKey: .mailroom) } if let mailroomDigest { try container.encode(mailroomDigest, forKey: .mailroomDigest) } + if let revision { try container.encode(revision, forKey: .revision) } } } diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index a7844c36..dc38be82 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -32,6 +32,12 @@ import MailroomKit public actor GraphStore { public private(set) var graph: LoopGraph private var connections: [UUID: Int32] = [:] + /// What each connection announced it can read (`DaemonCommand.announce`) — what + /// decides whether a presence tick reaches it as a delta or as the whole snapshot. + private var connectionCapabilities: [UUID: Set] = [:] + /// One counter for every frame this store sends about its graph — snapshots and + /// presence deltas alike — so a client can order them (`DaemonEvent.nodesChanged`). + private var revision = 0 private let onGraphChanged: (@Sendable (LoopGraph) -> Void)? private let onEnsureSession: (@Sendable (LoopNode, String?) -> Void)? private let onTerminateSession: (@Sendable (LoopNode, String?) -> Void)? @@ -501,18 +507,29 @@ public actor GraphStore { // MARK: - Connections - public func addConnection(id: UUID, fileDescriptor: Int32) { + public func addConnection( + id: UUID, fileDescriptor: Int32, capabilities: Set = [] + ) { + connectionCapabilities[id] = capabilities // Joining a project registers the connection here too, so ensure its outbound half // 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) connections[id] = fileDescriptor - send(.graphChanged(graph.wireSnapshot()), to: id) + send(.graphChanged(graph.wireSnapshot(revision: revision)), to: id) } public func removeConnection(_ id: UUID) { connections.removeValue(forKey: id) + connectionCapabilities.removeValue(forKey: id) + } + + /// An announcement that arrived after the connection joined — see + /// `ProjectRegistry`'s handling of `.announce`. + public func setCapabilities(_ capabilities: Set, for id: UUID) { + guard connections[id] != nil else { return } + connectionCapabilities[id] = capabilities } // MARK: - Commands @@ -1193,6 +1210,7 @@ public actor GraphStore { // loop is working and the line under it says what at. Reading the second only when // someone presses refresh left every card describing the tool call its session made // whenever that happened to be. + let before = graph.nodes var changed = await refreshPresence() if await refreshActivity() { changed = true } if await refreshSummary() { changed = true } @@ -1203,7 +1221,12 @@ public actor GraphStore { // what was waiting on exactly that. await drainPendingFollowUps() guard changed else { return } - notifyClients() + // Only the loops the tick touched, never the whole graph: everything above edits + // fields on top-level nodes, so the diff is exact, and a busy graph's fifteen-second + // pulse becomes a kilobyte per loop that moved instead of the whole snapshot. + let moved = Array(graph.nodes.filter { before[id: $0.id] != $0 }) + guard !moved.isEmpty else { return } + notifyClients(nodesChanged: moved) } // MARK: - Renaming @@ -3259,20 +3282,46 @@ public actor GraphStore { // Encoded once, not once per connection: the frame is the same bytes for every // client, and encoding a full graph is the expensive half of a broadcast — with C // clients attached it was C encodes of the same snapshot on every change, presence - // tick included (issue #288's CPU amplifier). And not at all with no client: the - // daemon runs clientless most of the day, and `send` used to find no descriptor - // before it encoded anything — hoisting the encode must not turn zero into one. - guard !connections.isEmpty else { return } - guard let frame = Self.encode(.graphChanged(graph.wireSnapshot())) else { return } + // tick included (issue #288's CPU amplifier). + revision += 1 + guard let frame = Self.encode(.graphChanged(graph.wireSnapshot(revision: revision))) + else { return } for id in connections.keys { deliver(frame, to: id) } } + /// The presence poll's broadcast — see `DaemonEvent.nodesChanged`. Not superseded: + /// a newer delta does not carry what an older one said, so both go out; the revision + /// is what lets a client drop one a later snapshot has overtaken. + /// + /// Only a connection that announced `ClientCapability.nodesChanged` gets the delta. + /// Every other connection gets the whole snapshot for the same tick, stamped with the + /// same revision — what every client got before deltas existed, so an older app + /// never meets a frame it cannot read. Both frames are encoded at most once. + private func notifyClients(nodesChanged nodes: [LoopNode]) { + revision += 1 + let delta = Self.encode( + .nodesChanged(projectPath: graph.project.path, revision: revision, nodes: nodes)) + var snapshot: EncodedEvent? + for (id, capabilities) in connectionCapabilities where connections[id] != nil { + if capabilities.contains(ClientCapability.nodesChanged.rawValue) { + if let delta { deliver(delta, to: id) } + } else { + if snapshot == nil { + snapshot = Self.encode(.graphChanged(graph.wireSnapshot(revision: revision))) + } + if let snapshot { deliver(snapshot, to: id) } + } + } + } + /// An event as the bytes and the superseding key it goes out with — everything about /// a frame that does not depend on which connection receives it. private struct EncodedEvent { let data: Data + /// `DaemonEvent.requiredCapability` — what `deliver` checks a connection announced. + let requiredCapability: ClientCapability? /// A snapshot still waiting to go out is replaced by a newer one rather than queued /// behind it — the event carries the whole graph, so the older one has nothing /// left to say. Keyed per graph, never on the event name alone: one connection joins @@ -3289,7 +3338,8 @@ public actor GraphStore { if case .graphChanged(let changed) = event { return "graphChanged:\(changed.id)" } return nil }() - return EncodedEvent(data: data, supersedingKey: supersedingKey) + return EncodedEvent( + data: data, requiredCapability: event.requiredCapability, supersedingKey: supersedingKey) } /// One event to one connection — the unicast shape (`addConnection`'s joining @@ -3301,6 +3351,15 @@ public actor GraphStore { private func deliver(_ frame: EncodedEvent, to connectionID: UUID) { guard let fileDescriptor = connections[connectionID] else { return } + // 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 + // (`notifyClients(nodesChanged:)` sends the snapshot). + if let required = frame.requiredCapability, + connectionCapabilities[connectionID]?.contains(required.rawValue) != true + { + return + } // 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 // inline blocked the actor for as long as the slowest client took to read diff --git a/GraphcodeKit/Sources/IPC/DaemonProtocol.swift b/GraphcodeKit/Sources/IPC/DaemonProtocol.swift index 99a004e8..31c220cb 100644 --- a/GraphcodeKit/Sources/IPC/DaemonProtocol.swift +++ b/GraphcodeKit/Sources/IPC/DaemonProtocol.swift @@ -46,6 +46,40 @@ public enum DaemonCommand: Codable, Sendable, Equatable { /// wants (issue #288). Requires the project to be resident in the daemon — opened by /// some connection — the same rule as `.graphCommand`. case mailbox(projectPath: String, query: MailboxQuery) + /// What this client can read beyond the events every client has always been sent — + /// the first frame the app puts on every socket, dial and redial alike. A daemon + /// sends a client only what it announced (`ClientCapability`): a client that never + /// announces — an older app — keeps getting the whole snapshot on every presence + /// tick, exactly as before, rather than a frame it cannot decode. Which matters + /// because the app's reader used to take an undecodable frame for a dead socket and + /// redial, every fifteen seconds, for ever. Unknown names are ignored, so a newer + /// client against this daemon is simply treated as what it is: a client of the + /// capabilities this daemon knows. Never answered. + case announce(capabilities: [String]) +} + +/// The names a client announces (`DaemonCommand.announce`) — strings on the wire so a +/// name this build does not know decodes rather than fails. +public enum ClientCapability: String, Sendable { + /// Reads `DaemonEvent.nodesChanged` and folds it into the snapshot it holds. + case nodesChanged +} + +extension DaemonEvent { + /// What a connection must have announced to be sent this event, or `nil` for the + /// events every client has always been sent. **Exhaustive on purpose**: adding a + /// case to `DaemonEvent` does not compile until its author has decided here whether + /// a client that predates it may receive it — and the one delivery path + /// (`GraphStore.deliver`) enforces the answer, so the daemon's default is that a + /// connection which never announced anything gets no new event type. That default is + /// the safety mechanism for clients already in the field, which are exactly the ones + /// that never announce; the handshake is how a newer client opts in. + public var requiredCapability: ClientCapability? { + switch self { + case .recentProjectsListed, .graphChanged, .errorOccurred, .mailbox: return nil + case .nodesChanged: return .nodesChanged + } + } } /// Mutations against exactly one project's graph — this is what `GraphStore.handle` @@ -218,4 +252,18 @@ public enum DaemonEvent: Codable, Sendable, Equatable { /// `projectPath` is the canonical spelling the daemon routed the query to, which is /// the id an app keys its projects by. case mailbox(projectPath: String, mailbox: Mailbox) + /// The presence poll's broadcast: only the loops whose reading, activity, summary or + /// board changed on this tick, as whole `LoopNode` values, instead of the whole graph + /// every fifteen seconds (issue #288's background load — on a busy graph something + /// changes almost every tick, and the tick shipped 50 KB to say which pill moved). + /// A client merges them into the snapshot it holds by id. Sent only to a connection + /// that announced `ClientCapability.nodesChanged`; every other connection gets the + /// whole snapshot for the same tick, as it always did. + /// + /// `revision` orders it against snapshots: the daemon stamps every frame for a graph + /// from one counter (`LoopGraph.revision`), and a client applies a delta only when it + /// is newer than the graph it holds. That is what makes it safe for an undelivered + /// snapshot to be superseded by a later one while a delta queued behind it still + /// arrives — the delta is older than what the client has, and is dropped. + case nodesChanged(projectPath: String, revision: Int, nodes: [LoopNode]) } diff --git a/GraphcodeKit/Sources/ProjectRegistry.swift b/GraphcodeKit/Sources/ProjectRegistry.swift index fd0c2ae8..42417f32 100644 --- a/GraphcodeKit/Sources/ProjectRegistry.swift +++ b/GraphcodeKit/Sources/ProjectRegistry.swift @@ -114,6 +114,11 @@ public actor ProjectRegistry { // MARK: - Connections + /// What each connection announced it can read — see `DaemonCommand.announce`. Kept + /// here, per connection, and handed to every store the connection joins, before or + /// after the announcement arrives. + private var connectionCapabilities: [UUID: Set] = [:] + public func addConnection(id: UUID, fileDescriptor: Int32) { // Registering the connection is what opens its outbound half — here rather than in // the daemon's accept loop because this is the one place every caller goes through, @@ -132,6 +137,7 @@ public actor ProjectRegistry { } connectionFileDescriptors.removeValue(forKey: id) connectionProjectPaths.removeValue(forKey: id) + connectionCapabilities.removeValue(forKey: id) sidebarConnections.remove(id) if connectionFileDescriptors.isEmpty { stopPresencePolling() } } @@ -357,6 +363,16 @@ public actor ProjectRegistry { send(.errorOccurred(reason), to: fileDescriptor) } + case .announce(let capabilities): + // Reaches every store this connection has already joined too: the app's launch + // sends its joins and its announcement together, and which lands first must not + // decide what the client is sent. + let announced = Set(capabilities) + connectionCapabilities[connectionID] = announced + for path in connectionProjectPaths[connectionID] ?? [] { + await stores[path]?.setCapabilities(announced, for: connectionID) + } + case .mailbox(let path, let query): // Routed and gated exactly as a `.graphCommand`: the room belongs to the project // the open landed on, and a query against a project this connection never @@ -385,7 +401,9 @@ public actor ProjectRegistry { private func open(_ canonicalPath: String, for connectionID: UUID, fileDescriptor: Int32) async { let store = await store(forProjectPath: canonicalPath) connectionProjectPaths[connectionID, default: []].insert(canonicalPath) - await store.addConnection(id: connectionID, fileDescriptor: fileDescriptor) + await store.addConnection( + id: connectionID, fileDescriptor: fileDescriptor, + capabilities: connectionCapabilities[connectionID] ?? []) // The global graph is always resident and isn't a folder anyone opened, so it stays // out of both the recents list and the restore-on-launch set — the app asks for it // by name every launch instead. @@ -415,7 +433,8 @@ public actor ProjectRegistry { for id in sidebarConnections where id != opener { guard let fileDescriptor = connectionFileDescriptors[id] else { continue } connectionProjectPaths[id, default: []].insert(path) - await store.addConnection(id: id, fileDescriptor: fileDescriptor) + await store.addConnection( + id: id, fileDescriptor: fileDescriptor, capabilities: connectionCapabilities[id] ?? []) } } diff --git a/graphcode-cli/Sources/main.swift b/graphcode-cli/Sources/main.swift index 91db2785..ac62cd74 100644 --- a/graphcode-cli/Sources/main.swift +++ b/graphcode-cli/Sources/main.swift @@ -101,7 +101,7 @@ func openProject(_ projectPath: String) throws -> LoopGraph? { let opened = try client.waitForEvent { switch $0 { case .graphChanged, .errorOccurred: return true - case .recentProjectsListed, .mailbox: return false + case .recentProjectsListed, .mailbox, .nodesChanged: return false } } if case .errorOccurred(let message) = opened { fail(message) } diff --git a/graphcode/Sources/Clients/OrchestratorClient.swift b/graphcode/Sources/Clients/OrchestratorClient.swift index 75acfdbb..73a09943 100644 --- a/graphcode/Sources/Clients/OrchestratorClient.swift +++ b/graphcode/Sources/Clients/OrchestratorClient.swift @@ -17,6 +17,14 @@ struct OrchestratorClient: Sendable { var send: @Sendable (_ command: DaemonCommand) async throws -> Void } +extension OrchestratorClient { + /// What the app shows when the daemon sent something it cannot read — surfaced + /// through the same `.errorOccurred` path a daemon refusal takes, once per connection. + static let unreadableFrameMessage = + "graphcoded sent an event this app cannot read — the daemon is newer than the app. " + + "The app keeps its connection but may fall behind until it is updated." +} + enum OrchestratorClientError: Error, Equatable { case connectFailed(errno: Int32) } @@ -85,9 +93,24 @@ private actor DaemonConnection { let fileDescriptor = try await ensureConnected() connectedDescriptor = fileDescriptor if await isReconnect() { try await rejoinProjects() } + var saidUnreadable = false while true { let data = try await readFrameAsync(from: fileDescriptor) - let event = try JSONDecoder().decode(DaemonEvent.self, from: data) + // A frame that read fine but did not decode is a daemon newer than this + // app, not a dead socket — the mirror of how `graphcoded` treats a command + // it does not know. Redialling here made an older app tear its connection + // down and rejoin every fifteen seconds for as long as the daemon kept + // sending something new. + // Skipped, and said once per connection through the same path a daemon + // refusal takes: a quietly skipped frame is a board going stale with + // nothing on screen to say why. + guard let event = try? JSONDecoder().decode(DaemonEvent.self, from: data) else { + if !saidUnreadable { + saidUnreadable = true + continuation.yield(.errorOccurred(OrchestratorClient.unreadableFrameMessage)) + } + continue + } continuation.yield(event) } } catch { @@ -173,11 +196,22 @@ private actor DaemonConnection { generation += 1 } + /// Connects, and announces on the new socket before anyone can use it: the + /// announcement is written inside the one connect attempt every caller awaits + /// (`ensureConnected`), so it is the first frame on every socket by construction — + /// `send` and `events()` both resume only after it has gone out. Sending it from the + /// stream task instead let a caller's first command race ahead of it, and a + /// connection that announced second was a connection that announced nothing for the + /// commands the daemon handled in between. private func connectWithBackoff() async throws -> Int32 { var lastError: any Error = OrchestratorClientError.connectFailed(errno: 0) for attempt in 0..<10 { do { - return try await connectAsync() + let fileDescriptor = try await connectAsync() + let announce = try JSONEncoder().encode( + DaemonCommand.announce(capabilities: [ClientCapability.nodesChanged.rawValue])) + try await writeFrameAsync(announce, to: fileDescriptor) + return fileDescriptor } catch { lastError = error try? await Task.sleep(for: .milliseconds(200 * (attempt + 1))) diff --git a/graphcode/Sources/Features/App/AppFeature.swift b/graphcode/Sources/Features/App/AppFeature.swift index 69f52d90..48561fb2 100644 --- a/graphcode/Sources/Features/App/AppFeature.swift +++ b/graphcode/Sources/Features/App/AppFeature.swift @@ -403,6 +403,9 @@ struct AppFeature { state.openLoop?.graph.mailroom = mailbox.posts } return .send(.projects(.element(id: path, action: .daemonEvent(event)))) + + case .nodesChanged(let path, let revision, let nodes): + return foldDelta(state, path: path, revision: revision, nodes: nodes) } case .projectHeaderTapped(let path): @@ -843,6 +846,21 @@ extension AppFeature { /// surface is deliberately not on the kill list: the loop's session belongs to the /// node, and whichever side deletes the node ends it (`GraphStore` on the daemon's /// path, `QuickChats` for a chat). + /// The presence tick, as the loops it moved: merged into the snapshot this app holds + /// and then handled as the snapshot it amounts to, so every reader of `.graphChanged` + /// — the activity log, the open workspace, the project — sees one shape. Dropped when + /// older than what is held: a snapshot that overtook it in the daemon's queue already + /// carries these loops as they are now. + func foldDelta( + _ state: State, path: String, revision: Int, nodes: [LoopNode] + ) -> Effect { + guard let held = state.projects[id: path]?.graph, revision > (held.revision ?? -1) else { + return .none + } + return .send( + .daemonEvent(.graphChanged(held.applying(nodesChanged: nodes, revision: revision)))) + } + /// A broadcast graph with the room's posts filled back in from the copy this app /// holds for the project — see `LoopGraph.mailroom`. A snapshot from a daemon that /// still ships posts keeps its own. diff --git a/graphcode/Sources/Features/Project/ProjectFeature.swift b/graphcode/Sources/Features/Project/ProjectFeature.swift index 401da885..0dae5ede 100644 --- a/graphcode/Sources/Features/Project/ProjectFeature.swift +++ b/graphcode/Sources/Features/Project/ProjectFeature.swift @@ -376,8 +376,10 @@ struct ProjectFeature { state.graph.mailroomDigest = mailbox.digest case .errorOccurred(let message): state.connectionError = message - case .recentProjectsListed: - break // Not this feature's concern — AppFeature routes this to `welcome`. + case .recentProjectsListed, .nodesChanged: + // Not this feature's concern: AppFeature routes the listing to `welcome` + // and folds a delta into the snapshot it holds before routing it here. + break } return .none diff --git a/graphcode/Tests/OrchestratorClientTests.swift b/graphcode/Tests/OrchestratorClientTests.swift index 3960bea6..0c976b10 100644 --- a/graphcode/Tests/OrchestratorClientTests.swift +++ b/graphcode/Tests/OrchestratorClientTests.swift @@ -27,6 +27,9 @@ struct OrchestratorClientTests { async let received = firstEvent(of: events) try await client.send(.listRecentProjects) + // First on the socket, before anything the app asks: what it can read. + let announce = try #require(await daemon.nextCommand()) + #expect(announce == .announce(capabilities: [ClientCapability.nodesChanged.rawValue])) let command = try #require(await daemon.nextCommand()) #expect(command == .listRecentProjects) @@ -70,13 +73,17 @@ struct OrchestratorClientTests { let events = client.connect() async let received = firstEvent(of: events) try await client.send(.listRecentProjects) + _ = try #require(await daemon.nextCommand()) // the announcement let opening = try #require(await daemon.nextCommand()) #expect(opening == .listRecentProjects) // The daemon hangs up, the way a restart does. daemon.closeConnection(at: 0) - // The replacement socket announces itself instead of waiting to be spoken to. + // The replacement socket announces itself instead of waiting to be spoken to — + // what it can read first, then what it wants back. + let reannounce = try #require(await daemon.nextCommand(onConnection: 1)) + #expect(reannounce == .announce(capabilities: [ClientCapability.nodesChanged.rawValue])) let rejoin = try #require(await daemon.nextCommand(onConnection: 1)) #expect(rejoin == .restoreOpenProjects) let joinGlobal = try #require(await daemon.nextCommand(onConnection: 1)) @@ -87,6 +94,38 @@ struct OrchestratorClientTests { #expect(await received == .errorOccurred("after reconnect")) } + /// A frame this app cannot decode — a daemon newer than it — is skipped, not taken + /// for a dead socket: the stream carries on over the same connection and the next + /// readable event arrives. Redialling here is what made an older app rejoin every + /// fifteen seconds against a daemon that had learned a new event. + @Test + func anUnknownEventIsSkippedNotTakenForADeadSocket() async throws { + let daemon = try StubDaemon() + defer { daemon.stop() } + let client = OrchestratorClient.live(socketPath: daemon.socketPath) + let events = client.connect() + async let received = firstEvent(of: events) + try await client.send(.listRecentProjects) + _ = try #require(await daemon.nextCommand()) + + try daemon.replyRaw(Data(#"{"somethingNewer":{"_0":42}}"#.utf8)) + try daemon.replyRaw(Data(#"{"somethingNewer":{"_0":43}}"#.utf8)) + let project = ProjectRef(path: "/tmp/stub-project", name: "stub-project") + try daemon.reply(.recentProjectsListed([project])) + + // Said once — loudly, through the path a daemon refusal takes — then the stream + // carries on over the same socket, and the second unreadable frame is silent. + let first = await received + #expect(first == .errorOccurred(OrchestratorClient.unreadableFrameMessage)) + var next: DaemonEvent? + for await event in events { + next = event + break + } + #expect(next == .recentProjectsListed([project])) + #expect(daemon.acceptedConnectionCount == 1) + } + private func firstEvent(of events: AsyncStream) async -> DaemonEvent? { for await event in events { return event } return nil @@ -181,6 +220,12 @@ private final class StubDaemon: @unchecked Sendable { } /// Writes an event back the way `graphcoded` does — on the accepted connection. + /// Bytes as given — for a frame this build's `DaemonEvent` cannot decode. + func replyRaw(_ data: Data, onConnection index: Int = 0) throws { + guard let descriptor = waitForConnection(at: index) else { throw StubError.noConnection } + try FramedMessageIO.writeFrame(data, to: descriptor) + } + func reply(_ event: DaemonEvent, onConnection index: Int = 0) throws { guard let descriptor = waitForConnection(at: index) else { throw StubError.noConnection } try FramedMessageIO.writeFrame(try JSONEncoder().encode(event), to: descriptor) diff --git a/graphcode/Tests/PresenceDeltaTests.swift b/graphcode/Tests/PresenceDeltaTests.swift new file mode 100644 index 00000000..1f015a72 --- /dev/null +++ b/graphcode/Tests/PresenceDeltaTests.swift @@ -0,0 +1,235 @@ +import ComposableArchitecture +import Foundation +import GraphcodeKit +import MailroomKit +import Testing + +@testable import graphcode + +#if canImport(Darwin) + import Darwin +#endif + +/// The presence tick's broadcast is the loops it moved, not the whole graph +/// (`DaemonEvent.nodesChanged`, issue #288's background load) — and a revision on every +/// frame is what lets a client hold snapshots and deltas in one sequence. +@Suite +struct PresenceDeltaTests { + private static let project = ProjectRef(path: "/tmp/project-a", name: "project-a") + + private func node(_ title: String) -> LoopNode { + LoopNode(title: title, loopType: .goalBased, goal: GoalSpec(summary: "done"), state: .running) + } + + private func nextEvent(from descriptor: Int32) async throws -> DaemonEvent { + try await withCheckedThrowingContinuation { continuation in + DispatchQueue.global().async { + do { + let data = try FramedMessageIO.readFrame(from: descriptor) + continuation.resume(returning: try JSONDecoder().decode(DaemonEvent.self, from: data)) + } catch { + continuation.resume(throwing: error) + } + } + } + } + + private func nothingPending(on descriptor: Int32) -> Bool { + var probe = [UInt8](repeating: 0, count: 1) + let peeked = recv(descriptor, &probe, 1, MSG_PEEK | MSG_DONTWAIT) + return peeked < 0 && (errno == EAGAIN || errno == EWOULDBLOCK) + } + + private actor Readings { + private var answers: [String: Presence] = [:] + func set(_ title: String, _ presence: Presence) { answers[title] = presence } + func read(_ node: LoopNode) -> PresenceReading { + PresenceReading(presence: answers[node.title] ?? .idle, confidence: .reported) + } + } + + @Test + func aTickShipsOnlyTheLoopsItMovedAndNothingWhenNoneDid() async throws { + let readings = Readings() + var graph = LoopGraph(project: Self.project) + graph.nodes.append(node("Still")) + graph.nodes.append(node("Moving")) + let store = GraphStore( + graph: graph, onEnsureSession: { _, _ in }, + onReadPresence: { node, _ in await readings.read(node) }) + var pair: [Int32] = [0, 0] + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &pair) == 0) + defer { + OutboundChannels.close(pair[0]) + close(pair[1]) + } + // Announced, the way the app's socket is: a connection that never announces gets + // the whole snapshot on every tick instead (the test below). + await store.addConnection( + id: UUID(), fileDescriptor: pair[0], + capabilities: [ClientCapability.nodesChanged.rawValue]) + guard case .graphChanged(let joined) = try await nextEvent(from: pair[1]) else { + Issue.record("expected the joining snapshot") + return + } + let joinedRevision = try #require(joined.revision) + + // First tick: both loops get their first reading, so both move. + await store.pollPresence() + guard case .nodesChanged(let path, let first, let both) = try await nextEvent(from: pair[1]) + else { + Issue.record("expected a delta") + return + } + #expect(path == Self.project.path) + #expect(first > joinedRevision) + #expect(Set(both.map(\.title)) == ["Still", "Moving"]) + + // Second tick: one reading changes, one frame, one loop in it. + await readings.set("Moving", .busy) + await store.pollPresence() + guard case .nodesChanged(_, let second, let moved) = try await nextEvent(from: pair[1]) + else { + Issue.record("expected a delta for the loop that moved") + return + } + #expect(second > first) + #expect(moved.map(\.title) == ["Moving"]) + #expect(moved[0].presence?.presence == .busy) + + // Third tick: nothing changed, nothing sent. + await store.pollPresence() + #expect(nothingPending(on: pair[1])) + + // A command still broadcasts a whole snapshot, stamped later in the same sequence. + await store.handle(.renameNode(graph.nodes[0].id, title: "Renamed")) + guard case .graphChanged(let renamed) = try await nextEvent(from: pair[1]) else { + Issue.record("expected a snapshot for the rename") + return + } + #expect(try #require(renamed.revision) > second) + #expect(renamed.nodes[0].title == "Renamed") + #expect(renamed.nodes[1].presence?.presence == .busy) + } + + /// Only a connection that announced the capability gets the delta; every other gets + /// the whole snapshot for the same tick — so an app that predates deltas is never + /// sent a frame it cannot read. An announcement that arrives after the join counts. + @Test + func aTickReachesEachConnectionInTheShapeItAnnounced() async throws { + let readings = Readings() + var graph = LoopGraph(project: Self.project) + graph.nodes.append(node("Moving")) + let store = GraphStore( + graph: graph, onEnsureSession: { _, _ in }, + onReadPresence: { node, _ in await readings.read(node) }) + var modern: [Int32] = [0, 0] + var legacy: [Int32] = [0, 0] + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &modern) == 0) + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &legacy) == 0) + defer { + OutboundChannels.close(modern[0]) + OutboundChannels.close(legacy[0]) + close(modern[1]) + close(legacy[1]) + } + let modernID = UUID() + await store.addConnection(id: modernID, fileDescriptor: modern[0]) + await store.addConnection(id: UUID(), fileDescriptor: legacy[0]) + _ = try await nextEvent(from: modern[1]) + _ = try await nextEvent(from: legacy[1]) + // The app's announcement lands after its join; the registry forwards it. + await store.setCapabilities([ClientCapability.nodesChanged.rawValue], for: modernID) + + await readings.set("Moving", .busy) + await store.pollPresence() + + guard case .nodesChanged(_, let revision, let moved) = try await nextEvent(from: modern[1]) + else { + Issue.record("the announcing connection should get the delta") + return + } + #expect(moved.map(\.title) == ["Moving"]) + guard case .graphChanged(let snapshot) = try await nextEvent(from: legacy[1]) else { + Issue.record("a connection that announced nothing should get the snapshot") + return + } + #expect(snapshot.revision == revision) + #expect(snapshot.nodes[0].presence?.presence == .busy) + #expect(snapshot.mailroom.isEmpty) + } + + /// The invariant lives in the type, not in memory: every event either predates + /// capabilities (sent to everyone, as always) or names the capability that gates it. + /// A new case does not compile until its author has decided which, and `deliver` + /// enforces the answer for every call site. + @Test + func everyEventSaysWhetherAnOlderClientMayReceiveIt() { + let legacy: [DaemonEvent] = [ + .recentProjectsListed([]), .graphChanged(LoopGraph(project: Self.project)), + .errorOccurred("x"), + .mailbox( + projectPath: "/p", + mailbox: Mailbox(posts: [], bodiesTrimmed: false, digest: MailroomDigest(of: []))), + ] + #expect(legacy.allSatisfy { $0.requiredCapability == nil }) + #expect( + DaemonEvent.nodesChanged(projectPath: "/p", revision: 1, nodes: []).requiredCapability + == .nodesChanged) + } + + @Test + func aDeltaAppliesByIdAndNeverInventsALoop() { + var graph = LoopGraph(project: Self.project) + let kept = node("Kept") + graph.nodes.append(kept) + var moved = kept + moved.presence = PresenceReading(presence: .busy, confidence: .reported) + let stranger = node("Stranger") + + let applied = graph.applying(nodesChanged: [moved, stranger], revision: 7) + #expect(applied.nodes.map(\.title) == ["Kept"]) + #expect(applied.nodes[0].presence?.presence == .busy) + #expect(applied.revision == 7) + } + + /// The app folds a delta into the snapshot it holds and handles the result as a + /// snapshot; a delta older than what it holds — one a superseding snapshot overtook + /// in the daemon's queue — is dropped. + @Test + @MainActor + func theAppFoldsADeltaIntoItsSnapshotAndDropsAStaleOne() async { + let loop = node("Loop") + var held = LoopGraph(project: Self.project, nodes: [loop]) + held.revision = 5 + var state = AppFeature.State() + state.projects.append(ProjectFeature.State(graph: held)) + state.openLoop = LoopWorkspaceFeature.State( + node: loop, layout: .defaultLayout(forNode: loop.id), projectPath: Self.project.path, + projectName: Self.project.name) + let store = TestStore(initialState: state) { + AppFeature() + } withDependencies: { + $0.orchestratorClient.send = { _ in } + } + store.exhaustivity = .off + + var busy = loop + busy.presence = PresenceReading(presence: .busy, confidence: .reported) + await store.send( + .daemonEvent(.nodesChanged(projectPath: Self.project.path, revision: 6, nodes: [busy]))) + // The fold re-dispatches a snapshot, which the app hands its project one hop down. + await store.receive(\.daemonEvent) + await store.receive(\.projects) + #expect(store.state.projects[id: Self.project.path]?.graph.revision == 6) + #expect(store.state.openLoop?.node.presence?.presence == .busy) + + var idle = loop + idle.presence = PresenceReading(presence: .idle, confidence: .reported) + await store.send( + .daemonEvent(.nodesChanged(projectPath: Self.project.path, revision: 4, nodes: [idle]))) + await store.finish() + #expect(store.state.projects[id: Self.project.path]?.graph.revision == 6) + #expect(store.state.openLoop?.node.presence?.presence == .busy) + } +}