From 82b86a6c563f26c270e41fcef01e1ce2696e01d1 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 21:51:58 -0700 Subject: [PATCH 1/3] Bound the drain's send, and stage what times out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #316 bounded the presence read, but drainPendingFollowUps still awaited deliverToSession unbounded in the same walk — the stable-release check's DeliveryWedgeTests reproduce the freeze reached through the send instead: one loop's hung zmx send held the queue for the lease's whole 300s, every other loop's staged mail unmoving and silent. The send now has the same withDeadline the presence read has (45s, the loser abandoned rather than cancelled-and-waited). What the deadline drops is staged to the target's memory log and taken off the queue, so the target still reads it at its next wake — and a delivery that fails without hanging is recorded the same way, which the old code silently dropped. A delivery-stall line in the daemon log keeps frozen and working looking different. The lease stays: it is the backstop for the awaits that are not the drain's walk (stop requests, nudges), not a licence to park the queue for minutes first. --- .../Tests/DrainWedgeVerificationTests.swift | 230 ++++++++++++++++++ 1 file changed, 230 insertions(+) create mode 100644 graphcode/Tests/DrainWedgeVerificationTests.swift diff --git a/graphcode/Tests/DrainWedgeVerificationTests.swift b/graphcode/Tests/DrainWedgeVerificationTests.swift new file mode 100644 index 00000000..b9336bba --- /dev/null +++ b/graphcode/Tests/DrainWedgeVerificationTests.swift @@ -0,0 +1,230 @@ +import ComposableArchitecture +import Foundation +import GraphcodeKit +import Testing + +/// The stable-release check's wedge tests, inverted against #316. +/// +/// On `main` at 0.1.64-beta5 each of these described a freeze; here they describe the +/// fix. The setup is byte-for-byte what was measured on beta5 — one loop whose presence +/// read never returns, one innocent bystander — so a regression restores the original +/// failure rather than a new one. +@Suite +struct DrainWedgeVerificationTests { + private struct Fixture { + let graph: LoopGraph + var hung: UUID { graph.nodes[0].id } + var bystander: UUID { graph.nodes[1].id } + } + + private func fixture() -> Fixture { + Fixture( + graph: LoopGraph( + project: ProjectRef(path: "/tmp/drainwedge", name: "drainwedge"), + nodes: [ + LoopNode( + title: "Hung", loopType: .goalBased, goal: GoalSpec(summary: "hangs"), + presence: PresenceReading(presence: .busy, confidence: .reported), + state: .running), + LoopNode( + title: "Bystander", loopType: .goalBased, goal: GoalSpec(summary: "innocent"), + presence: PresenceReading(presence: .busy, confidence: .reported), + state: .running), + ])) + } + + /// Beta5: never delivered, for as long as a client stayed attached. + /// Here: delivered, once the bounded read times out. + @Test + func aHungPresenceReadNoLongerFreezesTheBystandersFollowUp() async { + let fixture = fixture() + let delivered = LockIsolated<[String]>([]) + let release = LockIsolated(false) + let store = GraphStore( + graph: fixture.graph, + onDeliverMessage: { _, message, _ in + delivered.withValue { $0.append(message) } + return true + }, + onReadPresence: { node, _ in + guard node.id == fixture.hung else { + return PresenceReading(presence: .idle, confidence: .reported) + } + while !release.value { try? await Task.sleep(for: .milliseconds(20)) } + return PresenceReading(presence: .busy, confidence: .reported) + }) + + await store.handle( + .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true)) + await store.handle( + .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true)) + + #expect(delivered.value == ["[graphcode] for the bystander"]) + release.setValue(true) + } + + /// The hung loop's own message is owed, not dropped: a timed-out read is `.unknown`, + /// which is not a state, so the item stays queued for a pass that gets an answer. + @Test + func theHungLoopsOwnMessageIsStillOwedAndLandsWhenItAnswers() async { + let fixture = fixture() + let delivered = LockIsolated<[String]>([]) + let release = LockIsolated(false) + let store = GraphStore( + graph: fixture.graph, + onDeliverMessage: { _, message, _ in + delivered.withValue { $0.append(message) } + return true + }, + onReadPresence: { node, _ in + guard node.id == fixture.hung else { + return PresenceReading(presence: .busy, confidence: .reported) + } + while !release.value { try? await Task.sleep(for: .milliseconds(20)) } + return PresenceReading(presence: .idle, confidence: .reported) + }) + + await store.handle( + .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true)) + #expect(delivered.value.isEmpty) + + release.setValue(true) + await store.handle(.refreshUsage) + + #expect(delivered.value == ["[graphcode] for the hung one"]) + } + + /// Point 3 of the stable check: on beta5 `refreshPresence` walked the graph serially + /// through the same unbounded await, so nothing behind the hung node was ever read. + @Test + func thePollTickReachesLoopsBehindAHungOne() async { + let fixture = fixture() + let asked = LockIsolated<[UUID]>([]) + let release = LockIsolated(false) + let store = GraphStore( + graph: fixture.graph, + onReadPresence: { node, _ in + asked.withValue { $0.append(node.id) } + guard node.id == fixture.hung else { + return PresenceReading(presence: .idle, confidence: .reported) + } + while !release.value { try? await Task.sleep(for: .milliseconds(20)) } + return PresenceReading(presence: .busy, confidence: .reported) + }) + + await store.handle(.refreshUsage) + + #expect(asked.value.contains(fixture.bystander)) + release.setValue(true) + } +} + +/// The presence read is bounded now. The **delivery** in the same loop is not. +/// +/// `deliverToSession` → `onDeliverMessage` → `CLISessionBackend.deliverMessage` → +/// `ZmxSessionLauncher.send` → `PTYProcessSession.waitCollectingOutput` is the same +/// unbounded chain the presence read had, and `GraphcodeKit/Sources/Sessions/` is +/// untouched by #316. So `drainPendingFollowUps` still holds `isDrainingFollowUps` +/// across an `await` that can never return — which is the premise the "no lease needed" +/// argument rests on. +/// +/// A send is if anything the likelier of the two to wedge: a message is chunked into +/// several sequential `zmx send` invocations, so one message is several chances to hang +/// on a wedged `ControlMaster` rather than one. +@Suite +struct DeliveryWedgeTests { + private struct Fixture { + let graph: LoopGraph + var hung: UUID { graph.nodes[0].id } + var bystander: UUID { graph.nodes[1].id } + } + + private func fixture() -> Fixture { + Fixture( + graph: LoopGraph( + project: ProjectRef(path: "/tmp/deliverywedge", name: "deliverywedge"), + nodes: [ + LoopNode( + title: "Hung", loopType: .goalBased, goal: GoalSpec(summary: "hangs"), + presence: PresenceReading(presence: .busy, confidence: .reported), + state: .running), + LoopNode( + title: "Bystander", loopType: .goalBased, goal: GoalSpec(summary: "innocent"), + presence: PresenceReading(presence: .busy, confidence: .reported), + state: .running), + ])) + } + + private func settle(until flag: LockIsolated) async { + for _ in 0..<2000 where !flag.value { try? await Task.sleep(for: .milliseconds(5)) } + } + + /// The control: every send returns, so the bystander is served. + @Test + func aBystandersFollowUpIsDeliveredWhenEverySendReturns() async { + let fixture = fixture() + let delivered = LockIsolated<[String]>([]) + let store = GraphStore( + graph: fixture.graph, + onDeliverMessage: { _, message, _ in + delivered.withValue { $0.append(message) } + return true + }, + onReadPresence: { _, _ in PresenceReading(presence: .idle, confidence: .reported) }) + + await store.handle( + .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true)) + await store.handle( + .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true)) + + #expect(delivered.value.contains("[graphcode] for the bystander")) + } + + /// The same wedge as beta5, reached through the send instead of the read: one loop's + /// `zmx send` never returns and every other loop's staged mail stops moving, silently, + /// for as long as the process lives. + @Test + func aHungDeliveryStillFreezesEveryOtherLoopsFollowUps() async { + let fixture = fixture() + let delivered = LockIsolated<[String]>([]) + let announced = LockIsolated<[String]>([]) + let entered = LockIsolated(false) + let release = LockIsolated(false) + let store = GraphStore( + graph: fixture.graph, + onDeliverMessage: { node, message, _ in + guard node.id == fixture.hung else { + delivered.withValue { $0.append(message) } + return true + } + entered.setValue(true) + while !release.value { try? await Task.sleep(for: .milliseconds(5)) } + return true + }, + // Both loops read idle, so the drain tries to deliver to both. + onReadPresence: { _, _ in PresenceReading(presence: .idle, confidence: .reported) }, + onAnnounceError: { message in announced.withValue { $0.append(message) } }) + + // Queued while both are mid-turn (cached `busy`), then delivered by the drain. + let wedging = Task { + await store.handle( + .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true)) + } + await settle(until: entered) + #expect(entered.value) + + // The store keeps answering, and the bystander's mail never moves. + await store.handle( + .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true)) + for _ in 0..<10 { + await store.handle(.refreshUsage) + } + + #expect(delivered.value.isEmpty) + #expect(announced.value.isEmpty) + #expect(!wedging.isCancelled) + + release.setValue(true) + _ = await wedging.value + } +} From 41315a9a1db31f0a60713241c127efa6928aa979 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 21:52:06 -0700 Subject: [PATCH 2/3] Bound the drain's send, and stage what times out MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #316 bounded the presence read, but drainPendingFollowUps still awaited deliverToSession unbounded in the same walk — the stable-release check's DeliveryWedgeTests reproduce the freeze reached through the send instead: one loop's hung zmx send held the queue for the lease's whole 300s, every other loop's staged mail unmoving and silent. The send now has the same withDeadline the presence read has (45s, the loser abandoned rather than cancelled-and-waited). What the deadline drops is staged to the target's memory log and taken off the queue, so the target still reads it at its next wake — and a delivery that fails without hanging is recorded the same way, which the old code silently dropped. A delivery-stall line in the daemon log keeps frozen and working looking different. The lease stays: it is the backstop for the awaits that are not the drain's walk (stop requests, nudges), not a licence to park the queue for minutes first. --- GraphcodeKit/Sources/GraphStore.swift | 44 ++++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index 92231dd4..2afeb055 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -238,6 +238,23 @@ public actor GraphStore { /// 300s here is `RemoteEnsureGate.leaseDuration`, arrived at for the same chain. private let drainLeaseDuration: Duration + /// How long the drain waits on one `deliverToSession` before it stops waiting. + /// + /// The lease above is a backstop, not the fix: a drain parked inside a send holds the + /// queue for the lease's whole 300s, silently, before any successor even learns of it + /// (the stable-release check's wedge test against #316). The send is the *likelier* + /// half of the chain to hang — one message is several sequential `zmx send` + /// invocations, so one message is several chances to hang on a wedged `ControlMaster`. + /// + /// Bounded in shape by the same `withDeadline` the presence read is: the loser is + /// abandoned, never cancelled-and-waited. What the deadline owes the queue is + /// durability — a timed-out item is staged to the target's memory log (the Mailroom + /// mirror already carries every message) and dropped, so a pass reads at its next wake + /// instead of a send that may only ever hang. The same span as + /// `presenceReadDeadline`: a bystander behind one wedged target waits one span, not + /// the process's life at either end. + private let deliveryDeadline: Duration + /// A poller holds `self` weakly, so a store going away already stops it *doing* /// anything — but the task itself keeps sleeping in its loop forever. Harmless for /// the long-lived project store; sub-graph stores are built per command and hold no @@ -280,6 +297,7 @@ public actor GraphStore { /// `ZmxSessionLauncher`; tests leave it `nil` or capture the calls. public init( graph: LoopGraph = LoopGraph(project: ProjectRef(path: "", name: "Untitled")), + deliveryDeadline: Duration = .seconds(45), onGraphChanged: (@Sendable (LoopGraph) -> Void)? = nil, onEnsureSession: (@Sendable (LoopNode, String?) -> Void)? = nil, onTerminateSession: (@Sendable (LoopNode, String?) -> Void)? = nil, @@ -343,6 +361,7 @@ public actor GraphStore { self.goalCache = goalCache ?? GoalEvaluationCache() self.recurrence = recurrence self.presenceReadDeadline = presenceReadDeadline + self.deliveryDeadline = deliveryDeadline self.drainLeaseDuration = drainLeaseDuration } @@ -2913,7 +2932,30 @@ public actor GraphStore { remaining.append(staged(pending)) continue } - _ = await deliverToSession(node, pending.text) + // The one await in the walk that used to have no deadline of its own — the send is + // the `PTYProcessSession` chain, and a hang here held the queue for the lease's + // whole span with nothing logged. Bounded now: whatever it owes on the deadline + // passing is staged to memory and dropped below, and the note is logged so frozen + // and working do not look identical (`#289` diagnostics). + let confirmed = + await withDeadline(deliveryDeadline) { + await self.deliverToSession(node, pending.text) + } ?? false + guard confirmed else { + DaemonLog.shared.record( + "delivery-stall", + DaemonRequestContext.fields + [ + ("node", node.id.uuidString), + ("deadline_ms", DaemonLog.milliseconds(deliveryDeadline.timeInterval)), + ]) + // Not left on the queue, not thrown away unwritten: the Mailroom mirror has the + // content, and the memory log's entry is what the target's next wake reads — the + // durable half a queue entry was only ever going to point at. A drop that does + // not record is the loss, and `staged(_:)` is where the record is canonical + // ("follow-up staged:") and happens exactly once. + _ = staged(pending) + continue + } } } From 4a172084a9a576da41cfb0bdf171c64b6af097a4 Mon Sep 17 00:00:00 2001 From: scgopi Date: Sun, 6 Sep 2026 22:00:54 -0700 Subject: [PATCH 3/3] Make timed-out follow-up delivery exactly once --- GraphcodeKit/Sources/GraphStore.swift | 87 ++++++++++++++++--- .../Tests/DrainWedgeVerificationTests.swift | 24 ++--- 2 files changed, 86 insertions(+), 25 deletions(-) diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index 2afeb055..cb028534 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/GraphcodeKit/Sources/GraphStore.swift @@ -176,6 +176,7 @@ public actor GraphStore { /// A deferred delivery: a `--follow-up` message, or a Mailroom watcher's wake for one /// post (`watchedPostID`), waiting for its target to go idle. struct PendingFollowUp: Equatable { + let id: UUID let nodeID: UUID let text: String /// The post a watcher's wake is about — what lets `mail watch --off` drop the wakes @@ -189,6 +190,8 @@ public actor GraphStore { } private var pendingFollowUps: [PendingFollowUp] = [] + private var pendingDeliveryAttempts: Set = [] + private var completedTimedOutDeliveries: [UUID: Bool] = [:] /// `drainPendingFollowUps` runs across several awaits, and the presence poll that /// calls it runs every fifteen seconds: without this, a second drain started while /// the first was suspended delivered the same items again and, on finishing, wrote @@ -2750,7 +2753,8 @@ public actor GraphStore { // dropping it. if followUp, deliversLater(to: target) { pendingFollowUps.append( - PendingFollowUp(nodeID: nodeID, text: message, watchedPostID: watchedPostID)) + PendingFollowUp( + id: UUID(), nodeID: nodeID, text: message, watchedPostID: watchedPostID)) return } @@ -2873,7 +2877,17 @@ 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 { + var retained: [PendingFollowUp] = [] + for pending in remaining + Array(batch[index...]) { + if let confirmed = completedTimedOutDeliveries.removeValue(forKey: pending.id) { + if !confirmed { _ = staged(pending) } + } else { + retained.append(pending) + } + } + pendingFollowUps = retained + 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 @@ -2887,6 +2901,10 @@ public actor GraphStore { while index < batch.count { let pending = batch[index] index += 1 + if pendingDeliveryAttempts.contains(pending.id) { + remaining.append(pending) + continue + } 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. @@ -2934,13 +2952,29 @@ public actor GraphStore { } // The one await in the walk that used to have no deadline of its own — the send is // the `PTYProcessSession` chain, and a hang here held the queue for the lease's - // whole span with nothing logged. Bounded now: whatever it owes on the deadline - // passing is staged to memory and dropped below, and the note is logged so frozen - // and working do not look identical (`#289` diagnostics). - let confirmed = - await withDeadline(deliveryDeadline) { - await self.deliverToSession(node, pending.text) - } ?? false + // whole span with nothing logged. Bounded now: a timed-out attempt stays in the + // queue until its abandoned operation reports whether it eventually delivered. + let attempt = DeliveryAttempt() + let confirmed = await withDeadline(deliveryDeadline) { + let result = await self.deliverToSession(node, pending.text) + await attempt.complete(result) + return result + } + guard let confirmed else { + pendingDeliveryAttempts.insert(pending.id) + remaining.append(pending) + Task { [self] in + let result = await attempt.wait() + finishTimedOutDelivery(pending.id, confirmed: result) + } + DaemonLog.shared.record( + "delivery-stall", + DaemonRequestContext.fields + [ + ("node", node.id.uuidString), + ("deadline_ms", DaemonLog.milliseconds(deliveryDeadline.timeInterval)), + ]) + continue + } guard confirmed else { DaemonLog.shared.record( "delivery-stall", @@ -2948,17 +2982,24 @@ public actor GraphStore { ("node", node.id.uuidString), ("deadline_ms", DaemonLog.milliseconds(deliveryDeadline.timeInterval)), ]) - // Not left on the queue, not thrown away unwritten: the Mailroom mirror has the - // content, and the memory log's entry is what the target's next wake reads — the - // durable half a queue entry was only ever going to point at. A drop that does - // not record is the loss, and `staged(_:)` is where the record is canonical - // ("follow-up staged:") and happens exactly once. + // A completed failure is staged below by `finishTimedOutDelivery`; this branch + // covers a transport that answered false before the deadline. _ = staged(pending) continue } } } + private func finishTimedOutDelivery(_ id: UUID, confirmed: Bool) { + guard pendingDeliveryAttempts.remove(id) != nil else { return } + if let index = pendingFollowUps.firstIndex(where: { $0.id == id }) { + let pending = pendingFollowUps.remove(at: index) + if !confirmed { _ = staged(pending) } + } else { + completedTimedOutDeliveries[id] = confirmed + } + } + private func announceError(_ message: String) { if let frame = Self.encode(.errorOccurred(message)) { for id in connections.keys { @@ -3781,3 +3822,21 @@ public actor GraphStore { } } } + +private actor DeliveryAttempt { + private var result: Bool? + private var waiter: CheckedContinuation? + + func complete(_ result: Bool) { + guard self.result == nil else { return } + self.result = result + guard let waiter else { return } + self.waiter = nil + waiter.resume(returning: result) + } + + func wait() async -> Bool { + if let result { return result } + return await withCheckedContinuation { waiter = $0 } + } +} diff --git a/graphcode/Tests/DrainWedgeVerificationTests.swift b/graphcode/Tests/DrainWedgeVerificationTests.swift index b9336bba..b0ec5b59 100644 --- a/graphcode/Tests/DrainWedgeVerificationTests.swift +++ b/graphcode/Tests/DrainWedgeVerificationTests.swift @@ -52,7 +52,8 @@ struct DrainWedgeVerificationTests { } while !release.value { try? await Task.sleep(for: .milliseconds(20)) } return PresenceReading(presence: .busy, confidence: .reported) - }) + }, + presenceReadDeadline: .milliseconds(50)) await store.handle( .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true)) @@ -82,7 +83,8 @@ struct DrainWedgeVerificationTests { } while !release.value { try? await Task.sleep(for: .milliseconds(20)) } return PresenceReading(presence: .idle, confidence: .reported) - }) + }, + presenceReadDeadline: .milliseconds(50)) await store.handle( .messageNode(fixture.hung, text: "for the hung one", from: nil, followUp: true)) @@ -110,7 +112,8 @@ struct DrainWedgeVerificationTests { } while !release.value { try? await Task.sleep(for: .milliseconds(20)) } return PresenceReading(presence: .busy, confidence: .reported) - }) + }, + presenceReadDeadline: .milliseconds(50)) await store.handle(.refreshUsage) @@ -187,7 +190,6 @@ struct DeliveryWedgeTests { func aHungDeliveryStillFreezesEveryOtherLoopsFollowUps() async { let fixture = fixture() let delivered = LockIsolated<[String]>([]) - let announced = LockIsolated<[String]>([]) let entered = LockIsolated(false) let release = LockIsolated(false) let store = GraphStore( @@ -203,7 +205,7 @@ struct DeliveryWedgeTests { }, // Both loops read idle, so the drain tries to deliver to both. onReadPresence: { _, _ in PresenceReading(presence: .idle, confidence: .reported) }, - onAnnounceError: { message in announced.withValue { $0.append(message) } }) + deliveryDeadline: .milliseconds(50)) // Queued while both are mid-turn (cached `busy`), then delivered by the drain. let wedging = Task { @@ -213,18 +215,18 @@ struct DeliveryWedgeTests { await settle(until: entered) #expect(entered.value) - // The store keeps answering, and the bystander's mail never moves. + // The store keeps answering, and the bystander's mail moves after the timeout. await store.handle( .messageNode(fixture.bystander, text: "for the bystander", from: nil, followUp: true)) - for _ in 0..<10 { - await store.handle(.refreshUsage) - } + try? await Task.sleep(for: .milliseconds(100)) + await store.handle(.refreshUsage) - #expect(delivered.value.isEmpty) - #expect(announced.value.isEmpty) + #expect(delivered.value == ["[graphcode] for the bystander"]) #expect(!wedging.isCancelled) release.setValue(true) _ = await wedging.value + try? await Task.sleep(for: .milliseconds(50)) + #expect(delivered.value == ["[graphcode] for the bystander", "[graphcode] for the hung one"]) } }