Harden follow-up delivery and mailroom persistence

GraphcodeKit/Sources/GraphStore.swift

@@ -207,12 +207,14 @@ 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
+  /// 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] = []
+  private var drainInFlight: PendingFollowUp?
+  private var drainDeferred: [PendingFollowUp] = []
 
   /// How long a presence read may take before the store stops waiting on it.
   ///
@@ -2836,25 +2838,44 @@ public actor GraphStore {
           ("held_ms", DaemonLog.milliseconds(heldFor)),
           ("queued", String(pendingFollowUps.count)),
         ])
+      pendingFollowUps = drainDeferred + 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:
     // the walk holds the whole queue in locals, and the stable-release check's reading
     // 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
     }
   }
 

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) {

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"])
   }
 }