Drain staged follow-ups once, in order, and let watch --off take its wakes (#304)

GraphcodeKit/Sources/GraphStore.swift

@@ -173,7 +173,24 @@ public actor GraphStore {
   /// presence poll. The content is in the target's memory log from the moment it was
   /// queued, so losing this queue to a restart delays the message to the next wake
   /// rather than dropping it.
-  private var pendingFollowUps: [(nodeID: UUID, text: String)] = []
+  /// 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 nodeID: UUID
+    let text: String
+    /// The post a watcher's wake is about — what lets `mail watch --off` drop the wakes
+    /// still queued, and lets a wake for a post the reader has since read go unsent.
+    /// `nil` for a message a peer or a human sent.
+    let watchedPostID: Int?
+  }
+
+  private var pendingFollowUps: [PendingFollowUp] = []
+  /// `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
+  /// its own idea of what remained over the first's — dropping whatever had been
+  /// queued in between (issue #304: duplicates, lost mail, and out-of-order delivery).
+  private var isDrainingFollowUps = false
 
   /// 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
@@ -1682,7 +1699,8 @@ public actor GraphStore {
         "mailroom — new post #\(post.id)\(topicSuffix(post)) from \(post.author): "
         + "\(preview) — read it with: graphcode mail inbox \(graph.project.path)"
       await deliverAdHocMessage(
-        to: node.id, text: nudge, from: nil, followUp: true, mirror: false)
+        to: node.id, text: nudge, from: nil, followUp: true, mirror: false,
+        watchedPostID: post.id)
     }
   }
 
@@ -1824,6 +1842,9 @@ public actor GraphStore {
         recordMemory(watcherID, "mailroom: stopped watching")
       }
       graph.nodes[id: watcherID]?.mailroomWatch = nil
+      // What `--off` is for: the wakes already staged for this watcher go with the
+      // watch. A peer's `--follow-up` message to the same loop is not a wake and stays.
+      pendingFollowUps.removeAll { $0.nodeID == watcherID && $0.watchedPostID != nil }
     }
   }
 
@@ -2601,7 +2622,7 @@ public actor GraphStore {
   /// it landed is the one wrong answer.
   private func deliverAdHocMessage(
     to nodeID: UUID, text: String, from senderID: UUID?, followUp: Bool = false,
-    mirror: Bool = true
+    mirror: Bool = true, watchedPostID: Int? = nil
   ) async {
     guard let target = graph.nodes[id: nodeID] else {
       announceError("message not delivered: no loop \(nodeID) in this graph")
@@ -2633,7 +2654,8 @@ public actor GraphStore {
     // dropping it.
     if followUp, deliversLater(to: target) {
       recordMemory(nodeID, "follow-up staged: \(message)")
-      pendingFollowUps.append((nodeID: nodeID, text: message))
+      pendingFollowUps.append(
+        PendingFollowUp(nodeID: nodeID, text: message, watchedPostID: watchedPostID))
       return
     }
 
@@ -2698,10 +2720,23 @@ public actor GraphStore {
   /// the reading this waits for. A target that resolved or died is simply dropped from
   /// the queue: its memory log has carried the message since it was staged.
   private func drainPendingFollowUps() async {
-    guard !pendingFollowUps.isEmpty else { return }
-    var remaining: [(nodeID: UUID, text: String)] = []
-    for pending in pendingFollowUps {
+    guard !pendingFollowUps.isEmpty, !isDrainingFollowUps else { return }
+    isDrainingFollowUps = true
+    defer { isDrainingFollowUps = false }
+    // 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 = []
+    var remaining: [PendingFollowUp] = []
+    for pending in batch {
       guard let node = graph.nodes[id: pending.nodeID], !node.isResolved else { continue }
+      // A watcher's wake is only owed while the watch stands and the post is unread:
+      // `mail watch --off` after the wake was staged, or an inbox that has since read
+      // past the post, means the wake has nothing left to say.
+      if let postID = pending.watchedPostID {
+        guard node.mailroomWatch != nil, (node.lastMailroomRead ?? 0) < postID else { continue }
+      }
       switch MessageBus.deliverability(to: node) {
       case .targetBusyWithACheck:
         remaining.append(pending)
@@ -2723,7 +2758,7 @@ public actor GraphStore {
       }
       _ = await deliverToSession(node, pending.text)
     }
-    pendingFollowUps = remaining
+    pendingFollowUps = remaining + pendingFollowUps
   }
 
   private func announceError(_ message: String) {

graphcode/Tests/FollowUpDrainTests.swift

@@ -0,0 +1,156 @@
+import ComposableArchitecture
+import Foundation
+import GraphcodeKit
+import MailroomKit
+import Testing
+
+#if canImport(Darwin)
+  import Darwin
+#endif
+
+/// Issue #304: the drain of staged follow-ups runs across awaits and the presence poll
+/// re-enters it every fifteen seconds. Two drains overlapping delivered the same items
+/// twice and the later one's bookkeeping overwrote the earlier's — mail duplicated, mail
+/// lost, mail out of order. And `mail watch --off` left the wakes already staged to
+/// arrive anyway, past the reader's cursor.
+@Suite
+struct FollowUpDrainTests {
+  /// What the target's session appears to be doing; the drain asks each time.
+  private actor Readings {
+    var presence: Presence = .busy
+    var delay: Duration = .zero
+    func set(_ presence: Presence, delay: Duration = .zero) {
+      self.presence = presence
+      self.delay = delay
+    }
+    func read() async -> PresenceReading {
+      if delay > .zero { try? await Task.sleep(for: delay) }
+      return PresenceReading(presence: presence, confidence: .reported)
+    }
+  }
+
+  /// A connection with a channel lifecycle of its own and a reader draining it: a bare
+  /// `/dev/null` descriptor can be dropped as "disconnected" when its number is reused
+  /// across the suite, and then the poll never drains.
+  private final class Attachment: @unchecked Sendable {
+    let daemonEnd: Int32
+    private let peer: Int32
+    private let drainer: Task<Void, Never>
+    init() {
+      var pair: [Int32] = [0, 0]
+      _ = socketpair(AF_UNIX, SOCK_STREAM, 0, &pair)
+      daemonEnd = pair[0]
+      peer = pair[1]
+      let reading = pair[1]
+      drainer = Task.detached {
+        var sink = [UInt8](repeating: 0, count: 65536)
+        while recv(reading, &sink, sink.count, 0) > 0 {}
+      }
+    }
+    deinit {
+      OutboundChannels.close(daemonEnd)
+      close(peer)
+      drainer.cancel()
+    }
+  }
+
+  private func makeStore(
+    readings: Readings, delivered: LockIsolated<[String]>, attachment: Attachment
+  ) async -> (GraphStore, target: UUID, sender: UUID) {
+    let store = GraphStore(
+      onEnsureSession: { _, _ in },
+      onDeliverMessage: { _, message, _ in
+        delivered.withValue { $0.append(message) }
+        return true
+      },
+      onReadPresence: { _, _ in await readings.read() },
+      onMailroomEnabled: { true })
+    await store.handle(
+      .createNode(NodeDraft(title: "Target", loopType: .turnBased, firstInstruction: "Work")))
+    await store.handle(
+      .createNode(NodeDraft(title: "Sender", loopType: .turnBased, firstInstruction: "Work")))
+    let ids = await store.graph.nodes.map(\.id)
+    // A connection, so the presence poll runs.
+    await store.addConnection(id: UUID(), fileDescriptor: attachment.daemonEnd)
+    // Start the poll once so the target carries a `busy` reading: that is what stages a
+    // follow-up instead of typing it in.
+    await store.pollPresence()
+    return (store, ids[0], ids[1])
+  }
+
+  @Test
+  func overlappingDrainsDeliverEachMessageOnceAndLoseNothing() async {
+    let readings = Readings()
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, target, sender) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    for index in 1...3 {
+      await store.handle(
+        .messageNode(target, text: "staged \(index)", from: sender, followUp: true))
+    }
+    #expect(delivered.value.isEmpty)
+
+    // Idle now, and slow to answer, so a second poll lands while the first drain is
+    // suspended mid-delivery — the overlap that duplicated and dropped mail.
+    await readings.set(.idle, delay: .milliseconds(40))
+    async let first: Void = store.pollPresence()
+    try? await Task.sleep(for: .milliseconds(20))
+    async let second: Void = store.pollPresence()
+    // And one more staged while both are in flight: it must survive the overlap.
+    try? await Task.sleep(for: .milliseconds(10))
+    await store.handle(.messageNode(target, text: "staged late", from: sender, followUp: true))
+    _ = await (first, second)
+    await store.pollPresence()
+
+    let texts = delivered.value.map {
+      $0.replacingOccurrences(of: "[graphcode] Sender: ", with: "")
+    }
+    #expect(texts.filter { $0 == "staged 1" }.count == 1)
+    #expect(texts.filter { $0 == "staged 2" }.count == 1)
+    #expect(texts.filter { $0 == "staged 3" }.count == 1)
+    #expect(texts.filter { $0 == "staged late" }.count == 1)
+    #expect(Array(texts.prefix(3)) == ["staged 1", "staged 2", "staged 3"])
+  }
+
+  @Test
+  func watchOffDropsStagedWakesButNotAPeersFollowUp() async {
+    let readings = Readings()
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, target, sender) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    await store.handle(.mailroomWatch(on: true, topic: nil, from: target))
+    await store.handle(
+      .mailroomPost(text: "a post the watcher will be woken for", topic: nil, from: sender))
+    await store.handle(.messageNode(target, text: "a peer's word", from: sender, followUp: true))
+    #expect(delivered.value.isEmpty)
+
+    await store.handle(.mailroomWatch(on: false, topic: nil, from: target))
+    await readings.set(.idle)
+    await store.pollPresence()
+
+    #expect(delivered.value.count == 1)
+    #expect(delivered.value.first?.contains("a peer's word") == true)
+    #expect(!delivered.value.contains { $0.contains("new post #") })
+  }
+
+  @Test
+  func aWakeForAPostAlreadyReadIsNotSent() async throws {
+    let readings = Readings()
+    let delivered = LockIsolated<[String]>([])
+    let attachment = Attachment()
+    let (store, target, sender) = await makeStore(
+      readings: readings, delivered: delivered, attachment: attachment)
+    await store.handle(.mailroomWatch(on: true, topic: nil, from: target))
+    await store.handle(
+      .mailroomPost(text: "read before the wake could land", topic: nil, from: sender))
+    // The watcher reads its inbox — the cursor passes the post — before it goes idle.
+    _ = try await store.mailbox(
+      MailboxQuery(selection: .unread(reader: target), advanceCursor: true))
+    await readings.set(.idle)
+    await store.pollPresence()
+
+    #expect(delivered.value.isEmpty)
+  }
+}