diff --git a/GraphcodeKit/Sources/GraphStore.swift b/GraphcodeKit/Sources/GraphStore.swift index da05006c..d9dc444c 100644 --- a/GraphcodeKit/Sources/GraphStore.swift +++ b/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) { diff --git a/graphcode/Tests/FollowUpDrainTests.swift b/graphcode/Tests/FollowUpDrainTests.swift new file mode 100644 index 00000000..7e38f335 --- /dev/null +++ b/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 + 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) + } +}