From 5a420637125adaf0ce88ced66d2574013ea26c00 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 10:51:43 -0700 Subject: [PATCH] Encode a broadcast once, not once per connection (#288) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GraphStore.notifyClients looped over connections calling send, and send ran JSONEncoder().encode(event) — so a graph change cost one full encode of the snapshot per connection, presence tick included. The encode is hoisted out of the loop: notifyClients encodes the snapshot once into the bytes and the per-graph superseding key #291 introduced, and hands that to every connection through deliver, which keeps dropping a connection whose channel is gone. send keeps its shape for the unicast callers. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01DeGL2CxuGmq16RSZpJYm2N --- GraphcodeKit/Sources/GraphStore.swift | 67 ++++++++++++------ graphcode/Tests/BroadcastEncodingTests.swift | 71 ++++++++++++++++++++ 2 files changed, 119 insertions(+), 19 deletions(-) create mode 100644 graphcode/Tests/BroadcastEncodingTests.swift diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index fbd75e33..8917c392 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -2663,8 +2663,10 @@ public actor GraphStore { } private func announceError(_ message: String) { - for id in connections.keys { - send(.errorOccurred(message), to: id) + if let frame = Self.encode(.errorOccurred(message)) { + for id in connections.keys { + deliver(frame, to: id) + } } onAnnounceError?(message) } @@ -3213,32 +3215,59 @@ public actor GraphStore { /// other caller wants `broadcast` — a graph change that isn't saved is a graph change /// lost at the next daemon restart. private func notifyClients() { - let snapshot = graph.wireSnapshot() + // 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 } for id in connections.keys { - send(.graphChanged(snapshot), to: id) + deliver(frame, 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 + /// 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 + /// as many projects as it likes, and every project's store writes to that one + /// socket. A shared key made the newest snapshot supersede a *different* project's + /// undelivered one, so a client that had just joined two projects silently never + /// received the first — its loops simply never appeared. + let supersedingKey: String? + } + + private static func encode(_ event: DaemonEvent) -> EncodedEvent? { + guard let data = try? JSONEncoder().encode(event) else { return nil } + let supersedingKey: String? = { + if case .graphChanged(let changed) = event { return "graphChanged:\(changed.id)" } + return nil + }() + return EncodedEvent(data: data, supersedingKey: supersedingKey) + } + + /// One event to one connection — the unicast shape (`addConnection`'s joining + /// snapshot, a refusal). A broadcast goes through `notifyClients`, which encodes once. private func send(_ event: DaemonEvent, to connectionID: UUID) { + guard let frame = Self.encode(event) else { return } + deliver(frame, to: connectionID) + } + + private func deliver(_ frame: EncodedEvent, to connectionID: UUID) { guard let fileDescriptor = connections[connectionID] else { return } - guard let data = try? JSONEncoder().encode(event) else { 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 - // (issue #288). 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 as many - // projects as it likes, and every project's store writes to that one socket. A shared - // key made the newest snapshot supersede a *different* project's undelivered one, so - // a client that had just joined two projects silently never received the first — its - // loops simply never appeared. - let supersedingKey: String? = { - if case .graphChanged(let changed) = event { return "graphChanged:\(changed.id)" } - return nil - }() - guard OutboundChannels.send(data, to: fileDescriptor, supersedingKey: supersedingKey) else { + // (issue #288). + guard + OutboundChannels.send( + frame.data, to: fileDescriptor, supersedingKey: frame.supersedingKey) + else { // No live channel — the client already disconnected. Drop it here rather than // waiting for the read loop to notice, so a dead connection can't accumulate // failed broadcast attempts. diff --git a/graphcode/Tests/BroadcastEncodingTests.swift b/graphcode/Tests/BroadcastEncodingTests.swift new file mode 100644 index 00000000..5629a19f --- /dev/null +++ b/graphcode/Tests/BroadcastEncodingTests.swift @@ -0,0 +1,71 @@ +import Foundation +import GraphcodeKit +import Testing + +#if canImport(Darwin) + import Darwin +#endif + +/// A broadcast is one encode handed to every connection, not one encode per +/// connection (issue #288's CPU amplifier). Encoding is not observable from outside, +/// so this pins what the refactor must keep: every client gets the same bytes, a +/// client that is gone is dropped on the spot, and the rest still hear the change. +@Suite +struct BroadcastEncodingTests { + private func frame(from descriptor: Int32) async throws -> Data { + try await withCheckedThrowingContinuation { continuation in + DispatchQueue.global().async { + do { + continuation.resume(returning: try FramedMessageIO.readFrame(from: descriptor)) + } catch { + continuation.resume(throwing: error) + } + } + } + } + + @Test + func everyConnectionReceivesTheSameBytesAndADeadOneIsDropped() async throws { + let store = GraphStore(onEnsureSession: { _, _ in }, onDeliverMessage: { _, _, _ in true }) + var first: [Int32] = [0, 0] + var second: [Int32] = [0, 0] + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &first) == 0) + #expect(socketpair(AF_UNIX, SOCK_STREAM, 0, &second) == 0) + // The daemon ends belong to their channels (`OutboundChannels.open` in + // `addConnection`), so they close through the registry; `second[0]` goes mid-test. + defer { + OutboundChannels.close(first[0]) + close(first[1]) + close(second[1]) + } + await store.addConnection(id: UUID(), fileDescriptor: first[0]) + await store.addConnection(id: UUID(), fileDescriptor: second[0]) + _ = try await frame(from: first[1]) + _ = try await frame(from: second[1]) + + await store.handle( + .createNode(NodeDraft(title: "Loop", loopType: .turnBased, firstInstruction: "Work"))) + let toFirst = try await frame(from: first[1]) + let toSecond = try await frame(from: second[1]) + #expect(toFirst == toSecond) + guard case .graphChanged(let graph) = try JSONDecoder().decode(DaemonEvent.self, from: toFirst) + else { + Issue.record("expected the broadcast") + return + } + #expect(graph.nodes.map(\.title) == ["Loop"]) + + // A connection whose channel is gone is dropped by the broadcast that finds it so, + // and the live one still hears the change. + OutboundChannels.close(second[0]) + await store.handle(.renameNode(graph.nodes[0].id, title: "Renamed")) + guard + case .graphChanged(let renamed) = try JSONDecoder().decode( + DaemonEvent.self, from: try await frame(from: first[1])) + else { + Issue.record("expected the second broadcast") + return + } + #expect(renamed.nodes.map(\.title) == ["Renamed"]) + } +}