diff --git a/GraphcodeKit/Sources/GraphWriter.swift b/GraphcodeKit/Sources/GraphWriter.swift new file mode 100644 index 00000000..f89d41f0 --- /dev/null +++ b/GraphcodeKit/Sources/GraphWriter.swift @@ -0,0 +1,70 @@ +import Foundation + +/// Persists graphs off the actor that changes them — the disk-side twin of +/// `OutboundChannel` (issue #307). +/// +/// `GraphStore.broadcast()` used to call `persistence.saveGraph` synchronously, so every +/// mutation held the `GraphStore` actor across a full serialise-and-write — the shape +/// #291 removed from the socket path, one layer over: a memo measured at 0.03–2.13 s +/// against a 0.003 s socket round trip, with the variance coming from the filesystem. +/// +/// A save is handed here and the actor returns. One serial queue writes; consecutive +/// saves of the same project collapse to the newest snapshot (the graph is a value and +/// the file is a whole, so nothing older has anything left to say), which turns a burst +/// of memos into one write. `flush` waits for everything queued — what the daemon calls +/// on its way out, and what a test calls before reading the file back. +public final class GraphWriter: @unchecked Sendable { + private let persistence: ProjectPersistence + private let queue = DispatchQueue(label: "dev.graphcode.graphcoded.persist", qos: .utility) + private let lock = NSLock() + private var pending: [String: LoopGraph] = [:] + private var scheduled = false + + public init(persistence: ProjectPersistence) { + self.persistence = persistence + } + + /// Queues the newest snapshot of a project and returns at once. + public func save(_ graph: LoopGraph) { + lock.lock() + pending[graph.project.path] = graph + let drainNeeded = !scheduled + scheduled = true + lock.unlock() + guard drainNeeded else { return } + queue.async { [self] in drain() } + } + + /// The newest snapshot of a project — the one still queued, if there is one, else + /// the file. Every reader of the persisted graph goes through here rather than + /// through the file: a save that has left the actor and not yet reached the disk is + /// otherwise invisible, and a delete of a *closed* project (no live store) that read + /// the file to find the loops whose sessions it must end would end fewer than exist + /// and leave the rest running. + public func load(path: String) -> LoopGraph? { + lock.lock() + let queued = pending[path] + lock.unlock() + if let queued { return queued } + return persistence.loadGraph(path: path) + } + + /// Returns once everything queued so far is on disk. + public func flush() { + queue.sync { drain() } + } + + private func drain() { + while true { + lock.lock() + guard let (_, graph) = pending.first else { + scheduled = false + lock.unlock() + return + } + pending.removeValue(forKey: graph.project.path) + lock.unlock() + persistence.saveGraph(graph) + } + } +} diff --git a/GraphcodeKit/Sources/ProjectPersistence.swift b/GraphcodeKit/Sources/ProjectPersistence.swift index ef395400..a8765d5d 100644 --- a/GraphcodeKit/Sources/ProjectPersistence.swift +++ b/GraphcodeKit/Sources/ProjectPersistence.swift @@ -1,4 +1,5 @@ import Foundation +import MailroomKit /// Reads/writes the on-disk state Phase 4 adds: one JSON file per project's `LoopGraph` /// plus small recents and open-projects indexes, all under `~/.graphcode` (see @@ -31,12 +32,52 @@ public struct ProjectPersistence: Sendable { graph.nodes[index].presence = nil graph.nodes[index].activity = nil } + // The room's own file wins over one still inline in the graph file — a graph saved + // before the split carries its posts inline, and decodes exactly as it always did. + if let room = try? Data(contentsOf: mailroomURL(forProjectPath: path)), + let posts = try? JSONDecoder().decode([MailroomPost].self, from: room) + { + graph.mailroom = posts + } return graph } + /// Two files: the graph without its room, rewritten on every change, and the room on + /// its own, rewritten only when the room changed. The room was 84% of the graph file + /// (271 KB of 323 KB on the graph that filed #307) and changes only when a post lands, + /// while the graph changes on every memo, state tick and cursor move — the same + /// argument #293 made for the wire, applied to the file. public func saveGraph(_ graph: LoopGraph) { - guard let data = try? JSONEncoder().encode(graph) else { return } + var slim = graph + slim.mailroom = [] + guard let data = try? JSONEncoder().encode(slim) else { return } try? data.write(to: fileURL(forProjectPath: graph.project.path), options: .atomic) + let digest = MailroomDigest(of: graph.mailroom) + guard Self.roomDigests.changed(to: digest, for: graph.project.path) else { return } + let roomURL = mailroomURL(forProjectPath: graph.project.path) + if graph.mailroom.isEmpty { + try? FileManager.default.removeItem(at: roomURL) + } else if let room = try? JSONEncoder().encode(graph.mailroom) { + try? room.write(to: roomURL, options: .atomic) + } + } + + /// What the room last written for each project looked like, so an unchanged room is + /// not rewritten. Process-wide because this type is a value: every copy writes the + /// same files. A miss (first save after launch) writes once and is then remembered. + private static let roomDigests = RoomDigests() + + private final class RoomDigests: @unchecked Sendable { + private let lock = NSLock() + private var digests: [String: MailroomDigest] = [:] + + func changed(to digest: MailroomDigest, for path: String) -> Bool { + lock.lock() + defer { lock.unlock() } + guard digests[path] != digest else { return false } + digests[path] = digest + return true + } } /// Throws away a project's loops for good — the "Delete Loops…" half of the sidebar's @@ -45,6 +86,7 @@ public struct ProjectPersistence: Sendable { /// written to, deleted from, or otherwise modified. public func deleteGraph(path: String) { try? FileManager.default.removeItem(at: fileURL(forProjectPath: path)) + try? FileManager.default.removeItem(at: mailroomURL(forProjectPath: path)) } /// Filenames are the canonical path with `/` replaced by `_` — simple, deterministic, @@ -55,6 +97,12 @@ public struct ProjectPersistence: Sendable { return projectsDirectory.appendingPathComponent("\(safeName).json") } + /// The room beside its graph: `.mailroom.json`. + private func mailroomURL(forProjectPath path: String) -> URL { + let safeName = path.replacingOccurrences(of: "/", with: "_") + return projectsDirectory.appendingPathComponent("\(safeName).mailroom.json") + } + // MARK: - Recent projects public func loadRecentProjects() -> [ProjectRef] { diff --git a/GraphcodeKit/Sources/ProjectRegistry.swift b/GraphcodeKit/Sources/ProjectRegistry.swift index 97c796db..6ee1a2b1 100644 --- a/GraphcodeKit/Sources/ProjectRegistry.swift +++ b/GraphcodeKit/Sources/ProjectRegistry.swift @@ -24,6 +24,12 @@ import Foundation /// `.deleteProjectGraph` additionally discards its saved loops. public actor ProjectRegistry { private let persistence: ProjectPersistence + /// Writes graphs off the store's actor — see `GraphWriter`. `nonisolated` so the + /// daemon can flush it from a signal handler without an actor hop. + private nonisolated let writer: GraphWriter + /// For tests that read the file straight after a command: every save is flushed + /// before the store's turn ends, so the disk is exactly what the store holds. + private let persistsSynchronously: Bool private var stores: [String: GraphStore] = [:] private var connectionFileDescriptors: [UUID: Int32] = [:] private var connectionProjectPaths: [UUID: Set] = [:] @@ -79,9 +85,12 @@ public actor ProjectRegistry { sessionAlive: (@Sendable (LoopNode, String?) async -> Bool)? = CLISessionBackend.sessionAlive, composeBoard: (@Sendable (LoopNode, LoopSummary, String?, String?) async -> SummaryBoard?)? = CLISessionBackend.composeBoard, - reapCondemnedSessions: Bool = false + reapCondemnedSessions: Bool = false, + persistsSynchronously: Bool = false ) { persistence = ProjectPersistence(baseDirectory: persistenceDirectory) + writer = GraphWriter(persistence: persistence) + self.persistsSynchronously = persistsSynchronously self.ensureSession = ensureSession self.terminateSession = terminateSession self.restartSession = restartSession @@ -112,6 +121,12 @@ public actor ProjectRegistry { } } + /// Waits for every queued save — the daemon's last act on its way out, so a change + /// applied a moment before `SIGTERM` is on disk when launchd restarts it. + public nonisolated func flushPersistence() { + writer.flush() + } + // MARK: - Connections /// What each connection announced it can read — see `DaemonCommand.announce`. Kept @@ -337,7 +352,7 @@ public actor ProjectRegistry { // `store(forProjectPath:)` would run its load-time `ensureUnattendedSessions`, // *starting* sessions on the way to killing them. Memory goes with each loop, the // same as single-node deletion. - let graph = await stores[canonicalPath]?.graph ?? persistence.loadGraph(path: canonicalPath) + let graph = await stores[canonicalPath]?.graph ?? writer.load(path: canonicalPath) for node in graph?.nodesAtAnyDepth ?? [] { terminateSession?(node, canonicalPath) NodeMemory.remove(projectPath: canonicalPath, nodeID: node.id) @@ -468,7 +483,7 @@ public actor ProjectRegistry { guard path != canonical, stored.prefix(while: { $0 != path }).contains(where: { Self.canonicalize($0) == canonical }) else { return true } - let graph = persistence.loadGraph(path: path) + let graph = writer.load(path: path) let isEmpty = (graph?.nodesAtAnyDepth.isEmpty ?? true) && (graph?.mailroom.isEmpty ?? true) if isEmpty { persistence.forgetProject(path: path) } return !isEmpty @@ -604,7 +619,7 @@ public actor ProjectRegistry { private func store(forProjectPath path: String) async -> GraphStore { if let existing = stores[path] { return existing } let scope = LoopGraphScope(projectPath: path, name: Self.displayName(for: path)) - let graph = persistence.loadGraph(path: path) ?? LoopGraph(scope: scope) + let graph = writer.load(path: path) ?? LoopGraph(scope: scope) let persistence = self.persistence // A cross-graph spawn arrives here as a plain request; hopping through an unstructured // `Task` is what lets this actor re-enter itself to reach a *different* store without @@ -614,8 +629,11 @@ public actor ProjectRegistry { } let newStore = GraphStore( graph: graph, - onGraphChanged: { [weak self] updatedGraph in - persistence.saveGraph(updatedGraph) + onGraphChanged: { [weak self, writer, persistsSynchronously] updatedGraph in + // Handed to the writer and done: this closure runs on the store's actor, and a + // write of the whole graph held it for as long as the disk took (#307). + writer.save(updatedGraph) + if persistsSynchronously { writer.flush() } // Every state change is a chance for the last running loop to have stopped, or // the first to have started — see `refreshAwakeAssertion`. Task { await self?.refreshAwakeAssertion() } diff --git a/graphcode/Tests/DuplicateProjectPathTests.swift b/graphcode/Tests/DuplicateProjectPathTests.swift index 9e937cf8..8d9ef1a2 100644 --- a/graphcode/Tests/DuplicateProjectPathTests.swift +++ b/graphcode/Tests/DuplicateProjectPathTests.swift @@ -22,7 +22,8 @@ struct DuplicateProjectPathTests { let directory = FileManager.default.temporaryDirectory .appendingPathComponent("graphcode-tests-\(UUID().uuidString)", isDirectory: true) return ( - ProjectRegistry(persistenceDirectory: directory), ProjectPersistence(baseDirectory: directory) + ProjectRegistry(persistenceDirectory: directory, persistsSynchronously: true), + ProjectPersistence(baseDirectory: directory) ) } diff --git a/graphcode/Tests/ProjectPersistenceTests.swift b/graphcode/Tests/ProjectPersistenceTests.swift index 7b6b86a9..56a308da 100644 --- a/graphcode/Tests/ProjectPersistenceTests.swift +++ b/graphcode/Tests/ProjectPersistenceTests.swift @@ -1,5 +1,6 @@ import Foundation import GraphcodeKit +import MailroomKit import Testing /// Phase 4 (docs/07-roadmap.md#phase-4--projects): each project's graph is now @@ -108,3 +109,121 @@ struct ProjectPersistenceTests { #expect(recents.map(\.path) == [newer.path, older.path]) } } + +/// Issue #307: the room lives in its own file beside the graph, rewritten only when it +/// changed, and the graph file — rewritten on every change — no longer carries a post. +@Suite +struct MailroomPersistenceTests { + private func makeDirectory() -> URL { + let directory = FileManager.default.temporaryDirectory + .appendingPathComponent("graphcode-tests-\(UUID().uuidString)", isDirectory: true) + try? FileManager.default.createDirectory(at: directory, withIntermediateDirectories: true) + return directory + } + + private func post(_ id: Int, _ body: String) -> MailroomPost { + MailroomPost( + id: id, at: Date(timeIntervalSince1970: TimeInterval(id)), authorID: nil, + author: "a human", topic: nil, body: body) + } + + @Test + func theRoomIsSavedBesideTheGraphAndNeverInsideIt() throws { + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let persistence = ProjectPersistence(baseDirectory: directory) + let path = "/tmp/room-\(UUID().uuidString.prefix(6))" + var graph = LoopGraph(project: ProjectRef(path: path, name: "room")) + graph.nodes.append(LoopNode(title: "Loop", loopType: .turnBased, firstInstruction: "Work")) + graph.mailroom = [post(1, "SECRET-NONCE-A"), post(2, "SECRET-NONCE-B")] + + persistence.saveGraph(graph) + let name = path.replacingOccurrences(of: "/", with: "_") + let graphFile = directory.appendingPathComponent("projects/\(name).json") + let roomFile = directory.appendingPathComponent("projects/\(name).mailroom.json") + let graphText = try String(contentsOf: graphFile, encoding: .utf8) + #expect(!graphText.contains("SECRET-NONCE")) + #expect(!graphText.contains("\"mailroom\"")) + #expect(try String(contentsOf: roomFile, encoding: .utf8).contains("SECRET-NONCE-B")) + + let loaded = try #require(persistence.loadGraph(path: path)) + #expect(loaded.mailroom == graph.mailroom) + #expect(loaded.nodes.map(\.title) == ["Loop"]) + + // A change that leaves the room alone rewrites the graph file only. + let roomStamp = + try FileManager.default.attributesOfItem(atPath: roomFile.path)[ + .modificationDate] as? Date + graph.nodes[0].title = "Renamed" + Thread.sleep(forTimeInterval: 0.02) + persistence.saveGraph(graph) + let roomStampAfter = + try FileManager.default.attributesOfItem(atPath: roomFile.path)[ + .modificationDate] as? Date + #expect(roomStamp == roomStampAfter) + #expect(try #require(persistence.loadGraph(path: path)).nodes[0].title == "Renamed") + + persistence.deleteGraph(path: path) + #expect(!FileManager.default.fileExists(atPath: graphFile.path)) + #expect(!FileManager.default.fileExists(atPath: roomFile.path)) + } + + /// A graph file saved before the split carries its posts inline, and loads as it did. + @Test + func aGraphSavedWithTheRoomInlineStillLoadsIt() throws { + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let persistence = ProjectPersistence(baseDirectory: directory) + let path = "/tmp/legacy-\(UUID().uuidString.prefix(6))" + var graph = LoopGraph(project: ProjectRef(path: path, name: "legacy")) + graph.mailroom = [post(1, "from before the split")] + let name = path.replacingOccurrences(of: "/", with: "_") + try JSONEncoder().encode(graph).write( + to: directory.appendingPathComponent("projects/\(name).json")) + + #expect(try #require(persistence.loadGraph(path: path)).mailroom == graph.mailroom) + } + + /// A reader sees the newest snapshot whether or not it has reached the disk yet — + /// what keeps a delete of a closed project from missing loops still queued. + @Test + func aLoadReturnsTheQueuedSnapshotBeforeTheDiskHasIt() throws { + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let persistence = ProjectPersistence(baseDirectory: directory) + let writer = GraphWriter(persistence: persistence) + let path = "/tmp/queued-\(UUID().uuidString.prefix(6))" + var graph = LoopGraph(project: ProjectRef(path: path, name: "queued")) + graph.nodes.append(LoopNode(title: "One", loopType: .turnBased, firstInstruction: "Work")) + writer.save(graph) + writer.flush() + graph.nodes.append(LoopNode(title: "Two", loopType: .turnBased, firstInstruction: "Work")) + graph.nodes.append(LoopNode(title: "Three", loopType: .turnBased, firstInstruction: "Work")) + // Queued but, as far as this test can force it, not yet written: the writer's own + // answer must already be the three-loop graph either way. + writer.save(graph) + #expect(writer.load(path: path)?.nodes.count == 3) + writer.flush() + #expect(persistence.loadGraph(path: path)?.nodes.count == 3) + #expect(writer.load(path: "/tmp/never-saved") == nil) + } + + /// The writer takes a burst and lands the newest snapshot once; `flush` returns with + /// it on disk. + @Test + func theWriterCoalescesABurstAndFlushLandsTheNewest() throws { + let directory = makeDirectory() + defer { try? FileManager.default.removeItem(at: directory) } + let persistence = ProjectPersistence(baseDirectory: directory) + let writer = GraphWriter(persistence: persistence) + let path = "/tmp/burst-\(UUID().uuidString.prefix(6))" + var graph = LoopGraph(project: ProjectRef(path: path, name: "burst")) + graph.nodes.append(LoopNode(title: "v0", loopType: .turnBased, firstInstruction: "Work")) + for version in 1...50 { + graph.nodes[0].title = "v\(version)" + writer.save(graph) + } + writer.flush() + #expect(try #require(persistence.loadGraph(path: path)).nodes[0].title == "v50") + } +} diff --git a/graphcode/Tests/ProjectRegistryTests.swift b/graphcode/Tests/ProjectRegistryTests.swift index 1682d7ec..694a1dab 100644 --- a/graphcode/Tests/ProjectRegistryTests.swift +++ b/graphcode/Tests/ProjectRegistryTests.swift @@ -34,7 +34,8 @@ struct ProjectRegistryTests { let directory = FileManager.default.temporaryDirectory .appendingPathComponent("graphcode-tests-\(UUID().uuidString)", isDirectory: true) return ( - ProjectRegistry(persistenceDirectory: directory), ProjectPersistence(baseDirectory: directory) + ProjectRegistry(persistenceDirectory: directory, persistsSynchronously: true), + ProjectPersistence(baseDirectory: directory) ) } @@ -267,10 +268,12 @@ struct ProjectRegistryTests { let directory = FileManager.default.temporaryDirectory .appendingPathComponent("graphcode-tests-\(UUID().uuidString)", isDirectory: true) let killed = LockIsolated>([]) + // Reads the file straight after each command, so every save is flushed first. let registry = ProjectRegistry( persistenceDirectory: directory, ensureSession: { _, _ in }, - terminateSession: { node, _ in _ = killed.withValue { $0.insert(node.id) } }) + terminateSession: { node, _ in _ = killed.withValue { $0.insert(node.id) } }, + persistsSynchronously: true) let persistence = ProjectPersistence(baseDirectory: directory) let connectionID = UUID() await registry.addConnection(id: connectionID, fileDescriptor: -1) @@ -325,6 +328,9 @@ struct ProjectRegistryTests { command: .createNode( NodeDraft(title: "Watcher", loopType: .timeBased, triggerPrompt: "/loop 1h Check"))), connectionID: connectionID) + // The first daemon's last act on its way out is to flush its writer; the second + // daemon below only ever exists after that. + firstRun.flushPersistence() let persistence = ProjectPersistence(baseDirectory: directory) let nodeID = persistence.loadGraph(path: "/tmp/project-g")?.nodes.first?.id diff --git a/graphcoded/Sources/main.swift b/graphcoded/Sources/main.swift index fd58d82f..defc0493 100644 --- a/graphcoded/Sources/main.swift +++ b/graphcoded/Sources/main.swift @@ -106,6 +106,11 @@ DaemonLog.shared.record( // process just never lived long enough to run it. signal(SIGPIPE, SIG_IGN) +// Before the shutdown handlers below, which flush its writer on the way out. +let registry = ProjectRegistry( + persistenceDirectory: supportDirectory, + reapCondemnedSessions: true) + // Termination is handled on the main queue, not in signal context (#167). The handlers // this replaces called `exit(0)` from inside the signal handler itself, and `exit` is // not async-signal-safe: it runs atexit and runtime teardown after interrupting whatever @@ -124,6 +129,7 @@ signal(SIGINT, SIG_IGN) func makeShutdownSource(for signalNumber: Int32) -> DispatchSourceSignal { let source = DispatchSource.makeSignalSource(signal: signalNumber, queue: .main) source.setEventHandler { + registry.flushPersistence() unlink(path) exit(0) } @@ -156,6 +162,7 @@ func makeStalenessTimer() -> DispatchSourceTimer? { guard let current = ExecutableIdentity.of(path: executablePath), current != launchIdentity else { return } DaemonLog.shared.record("shutdown", [("reason", "binary-replaced")]) + registry.flushPersistence() unlink(path) exit(0) } @@ -165,10 +172,6 @@ func makeStalenessTimer() -> DispatchSourceTimer? { let stalenessTimer = makeStalenessTimer() -let registry = ProjectRegistry( - persistenceDirectory: supportDirectory, - reapCondemnedSessions: true) - /// Bridges a blocking socket read onto a background queue so the `Task` awaiting it /// never blocks Swift concurrency's cooperative thread pool — the whole connection /// handler below is otherwise just async/await hops (this, plus actor calls).