From 010350931557d991cf4c5efc3b2caa5ae7d7ad02 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 21:33:39 -0700 Subject: [PATCH 1/2] Harden follow-up delivery and mailroom persistence --- GraphcodeKit/Sources/GraphStore.swift | 61 ++++++++++++++++--- GraphcodeKit/Sources/ProjectPersistence.swift | 37 ++++++++--- 2 files changed, 82 insertions(+), 16 deletions(-) diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index 92231dd4..495b7659 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -207,12 +207,13 @@ public actor GraphStore { /// rejects the plain flag for this same chain and for this same reason. /// /// So the guard expires. A drain that outlives its lease is a wedge by definition, and - /// the next one says so in the daemon log and takes over. Taking over cannot duplicate - /// anything — the batch was taken and cleared in one actor step, so the successor - /// finds only what was queued after it — and if the wedged drain ever returns, its - /// `defer` folds its own remainder back on. Order can slip in that recovery; a queue - /// that moves again beats one frozen for ever. + /// the next one says so in the daemon log and takes over. The successor recovers the + /// suspended batch and invalidates the old owner before taking the queue on. private var drainLease: Date? + private var drainOwner: UUID? + private var drainBatch: [PendingFollowUp] = [] + private var drainInFlight: PendingFollowUp? + private var drainDeferred: [PendingFollowUp] = [] /// How long a presence read may take before the store stops waiting on it. /// @@ -2836,17 +2837,33 @@ public actor GraphStore { ("held_ms", DaemonLog.milliseconds(heldFor)), ("queued", String(pendingFollowUps.count)), ]) + pendingFollowUps = drainDeferred + (drainInFlight.map { [$0] } ?? []) + drainBatch + + pendingFollowUps + drainBatch = [] + drainInFlight = nil + drainDeferred = [] } + let owner = UUID() drainLease = taken + drainOwner = owner // Released only if it is still ours: a successor that took over after this lease // expired must not have its own lease cleared by the drain it replaced finally // returning — the mistake `RemoteEnsureGate.end(_:token:)` documents. - defer { if drainLease == taken { drainLease = nil } } + defer { + if drainOwner == owner { + drainOwner = nil + drainLease = nil + drainBatch = [] + drainInFlight = nil + drainDeferred = [] + } + } // Taken and cleared in one actor step, delivered from the local batch. Anything // queued while a delivery below is suspended lands in `pendingFollowUps` untouched // and is folded back in behind the retries at the end — never overwritten. let batch = pendingFollowUps pendingFollowUps = [] + drainBatch = batch var remaining: [PendingFollowUp] = [] var index = 0 // Folded back on the way out of every path, not only the one that runs to the end: @@ -2854,7 +2871,11 @@ public actor GraphStore { // of the wedge was that a drain which never returns strands them there. It cannot // now — the reading below is bounded — but what the pass did not resolve belongs on // the queue rather than in a variable about to go out of scope, whatever the exit. - defer { pendingFollowUps = remaining + Array(batch[index...]) + pendingFollowUps } + defer { + if drainOwner == owner { + pendingFollowUps = remaining + Array(batch[index...]) + pendingFollowUps + } + } // One reading per target per pass, taken the first time this pass reaches that // target and reused for the rest of its queue. Reading again between items is what // let a turn ending *mid-drain* reorder the queue: three messages for one loop went @@ -2866,14 +2887,18 @@ public actor GraphStore { // costs one probe per target rather than one per message. var readings: [UUID: Presence] = [:] while index < batch.count { + guard drainOwner == owner else { return } let pending = batch[index] index += 1 + drainBatch = Array(batch[index...]) + drainInFlight = pending guard let node = graph.nodes[id: pending.nodeID], !node.isResolved else { // Its work is over, but the message must not go with the queue entry: the log is // what the loop's next wake reads, and for a resolved loop that is all there is. if !pending.recorded, graph.nodes[id: pending.nodeID] != nil { recordMemory(pending.nodeID, "while you were away: \(pending.text)") } + drainInFlight = nil continue } // A watcher's wake is only owed while the watch that asked for it stands and the @@ -2885,16 +2910,22 @@ public actor GraphStore { if let postID = pending.watchedPostID { guard let watch = node.mailroomWatch, (node.lastMailroomRead ?? 0) < postID, let post = graph.mailroom.first(where: { $0.id == postID }), watch.matches(post.topic) - else { continue } + else { + drainInFlight = nil + continue + } } switch MessageBus.deliverability(to: node) { case .targetBusyWithACheck: remaining.append(staged(pending)) + drainDeferred = remaining + drainInFlight = nil continue case .some: if !pending.recorded { recordMemory(pending.nodeID, "while you were away: \(pending.text)") } + drainInFlight = nil continue case nil: break @@ -2911,9 +2942,21 @@ public actor GraphStore { // is not a state, so the message stays queued for a pass that gets an answer. guard presence == .idle else { remaining.append(staged(pending)) + drainDeferred = remaining + drainInFlight = nil continue } - _ = await deliverToSession(node, pending.text) + let delivered = await deliverToSession(node, pending.text) + guard drainOwner == owner else { return } + if !delivered { + remaining.append(staged(pending)) + drainDeferred = remaining + if !pending.recorded { + announceError( + "delivery to \(node.title)'s session failed — follow-up retained for retry") + } + } + drainInFlight = nil } } diff --git a/GraphcodeKit/Sources/ProjectPersistence.swift b/GraphcodeKit/Sources/ProjectPersistence.swift index 1c4501fd..00bed7f3 100644 --- a/GraphcodeKit/Sources/ProjectPersistence.swift +++ b/GraphcodeKit/Sources/ProjectPersistence.swift @@ -51,15 +51,34 @@ public struct ProjectPersistence: Sendable { 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) + do { + try data.write(to: fileURL(forProjectPath: graph.project.path), options: .atomic) + } catch { + return + } let roomURL = mailroomURL(forProjectPath: graph.project.path) let digest = MailroomDigest(of: graph.mailroom) - guard Self.roomDigests.changed(to: digest, for: roomURL.path) else { return } + guard !Self.roomDigests.matches(digest, for: roomURL.path) + || !FileManager.default.fileExists(atPath: roomURL.path) + else { return } if graph.mailroom.isEmpty { - try? FileManager.default.removeItem(at: roomURL) + do { + try FileManager.default.removeItem(at: roomURL) + } catch where !FileManager.default.fileExists(atPath: roomURL.path) { + Self.roomDigests.set(digest, for: roomURL.path) + } catch { + return + } } else if let room = try? JSONEncoder().encode(graph.mailroom) { - try? room.write(to: roomURL, options: .atomic) + do { + try room.write(to: roomURL, options: .atomic) + } catch { + return + } + } else { + return } + Self.roomDigests.set(digest, for: roomURL.path) } /// What the room last written for each project looked like, so an unchanged room is @@ -76,12 +95,16 @@ public struct ProjectPersistence: Sendable { private let lock = NSLock() private var digests: [String: MailroomDigest] = [:] - func changed(to digest: MailroomDigest, for path: String) -> Bool { + func matches(_ digest: MailroomDigest, for path: String) -> Bool { lock.lock() defer { lock.unlock() } - guard digests[path] != digest else { return false } + return digests[path] == digest + } + + func set(_ digest: MailroomDigest, for path: String) { + lock.lock() digests[path] = digest - return true + lock.unlock() } func forget(_ path: String) { From 5a5c0551b1a00728ab17f27d7eb1dfe4bebf471f Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 21:38:43 -0700 Subject: [PATCH 2/2] Avoid retrying indeterminate deliveries --- GraphcodeKit/Sources/GraphStore.swift | 6 +++--- graphcode/Tests/DrainWedgeTests.swift | 4 ++++ 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index 495b7659..b84252ca 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -208,7 +208,8 @@ public actor GraphStore { /// /// So the guard expires. A drain that outlives its lease is a wedge by definition, and /// the next one says so in the daemon log and takes over. The successor recovers the - /// suspended batch and invalidates the old owner before taking the queue on. + /// deferred and not-yet-started batch, but leaves an in-flight send alone because its + /// side effect may already have happened even though the callback never returned. private var drainLease: Date? private var drainOwner: UUID? private var drainBatch: [PendingFollowUp] = [] @@ -2837,8 +2838,7 @@ public actor GraphStore { ("held_ms", DaemonLog.milliseconds(heldFor)), ("queued", String(pendingFollowUps.count)), ]) - pendingFollowUps = drainDeferred + (drainInFlight.map { [$0] } ?? []) + drainBatch - + pendingFollowUps + pendingFollowUps = drainDeferred + drainBatch + pendingFollowUps drainBatch = [] drainInFlight = nil drainDeferred = [] diff --git a/graphcode/Tests/DrainWedgeTests.swift b/graphcode/Tests/DrainWedgeTests.swift index 310af5cb..45ad7497 100644 --- a/graphcode/Tests/DrainWedgeTests.swift +++ b/graphcode/Tests/DrainWedgeTests.swift @@ -158,6 +158,7 @@ struct DrainWedgeTests { func aHungDeliveryReleasesTheQueueWhenItsLeaseExpires() async { let fixture = fixture() let delivered = LockIsolated<[String]>([]) + let typed = LockIsolated<[String]>([]) let lines = LockIsolated<[String]>([]) let entered = LockIsolated(false) let release = LockIsolated(false) @@ -166,6 +167,7 @@ struct DrainWedgeTests { let store = GraphStore( graph: fixture.graph, onDeliverMessage: { node, message, _ in + typed.withValue { $0.append(message) } guard node.id == fixture.hung else { delivered.withValue { $0.append(message) } return true @@ -195,8 +197,10 @@ struct DrainWedgeTests { await store.handle(.memoNode(fixture.bystander, text: "settle", from: fixture.bystander)) #expect(delivered.value == ["[graphcode] for the bystander"]) + #expect(typed.value == ["[graphcode] for the hung one", "[graphcode] for the bystander"]) #expect(lines.value.contains { $0.contains("event=drain-stall") }) release.setValue(true) _ = await wedging.value + #expect(typed.value == ["[graphcode] for the hung one", "[graphcode] for the bystander"]) } }