Bound NE inbox replay and continue acknowledged catch-up

.github/workflows/tests.yml

@@ -88,6 +88,7 @@ jobs:
             Tests.static.test_discovered_interfaces_contract \
             Tests.static.test_ios_rnode_driver \
             Tests.static.test_rns_bridge_persistence \
+            Tests.static.test_rns_bridge_inbox \
             Tests.static.test_ios_rnode_stale_bond_contract \
             Tests.static.test_network_card_interface_list \
             Tests.static.test_tag_version_stamp

Sources/ColumbaNetworkExtension/NEPythonRNS.swift

@@ -322,11 +322,10 @@ final class NEPythonRNS: @unchecked Sendable {
     /// `rns_bridge.ack_inbox(max_seq)` - advance the durable-inbox consumer
     /// cursor (Contract §5). The app calls this after it has processed a drained
     /// batch, passing the highest `seq` in that batch; the NE deletes only rows
-    /// at or below the cursor. Best-effort (the Python ack is idempotent and
-    /// safe below the floor); a failure just means the rows are re-returned on
-    /// the next drain (at-least-once).
-    func ackInbox(maxSeq: Int) {
-        invoke("ack_inbox", kwargs: ["max_seq": maxSeq])
+    /// at or below the cursor. Return true only after Python commits. A failure
+    /// leaves the page replayable and the app retries with a delay.
+    func ackInbox(maxSeq: Int) -> Bool {
+        call("ack_inbox", kwargs: ["max_seq": maxSeq]) == "true"
     }
 
     /// `rns_bridge.persist()` → `{ok, ...}`.

Sources/ColumbaNetworkExtension/PacketTunnelProvider.swift

@@ -511,7 +511,7 @@ class PacketTunnelProvider: NEPacketTunnelProvider {
             return .ok(try? JSONEncoder().encode(announces))
 
         case .drainEvents:
-            // The Model B event bridge: drain the FULL queue (inbound, delivery,
+            // The Model B event bridge: read a bounded page of the queue (inbound, delivery,
             // state, link, announce) and map every event onto ProxyEvent so the
             // app's proxy poller can re-emit each as a BackendEvent. (The legacy
             // .heardAnnounces op above only maps announces and discards the rest;
@@ -527,8 +527,8 @@ class PacketTunnelProvider: NEPacketTunnelProvider {
             // Unacked rows survive a lost drain reply / app death and are
             // re-returned on the next drain (at-least-once; announce dedup in the
             // app makes re-delivery idempotent).
-            engine.ackInbox(maxSeq: maxSeq)
-            return .ok(try? JSONEncoder().encode(true))
+            let acknowledged = engine.ackInbox(maxSeq: maxSeq)
+            return .ok(try? JSONEncoder().encode(acknowledged))
 
         case .bleConnections:
             // Phase 2: the mesh CoreBluetooth radio now runs IN-PROCESS in the

Sources/RNSBackendProxy/ProxyRnsBackend.swift

@@ -331,29 +331,39 @@ public final class ProxyRnsBackend: RnsBackend, @unchecked Sendable {
         stateLock.unlock()
         // Initial catch-up drain (anything delivered while the app was suspended or
         // before this start). The ping-driven drains keep it current from here.
-        requestDrain()
+        requestDrain(expectedGeneration: expectedGeneration)
     }
 
     /// Coalesce ping-driven drains into at most one in-flight + one trailing pass.
-    private func requestDrain() {
+    private func requestDrain(expectedGeneration: Int? = nil) {
         stateLock.lock()
+        guard cachedLocalInfo != nil,
+              expectedGeneration == nil || expectedGeneration == startGeneration else {
+            stateLock.unlock()
+            return
+        }
         if draining {
             rescanRequested = true
             stateLock.unlock()
             return
         }
         draining = true
+        let drainGeneration = startGeneration
         stateLock.unlock()
         Task { [weak self] in
             defer {
                 self?.stateLock.lock()
-                self?.draining = false
-                let again = self?.rescanRequested ?? false
-                self?.rescanRequested = false
+                // A stopped worker must not clear a restarted worker's state.
+                let ownsDrain = self?.startGeneration == drainGeneration
+                let again = ownsDrain && (self?.rescanRequested ?? false)
+                if ownsDrain {
+                    self?.draining = false
+                    self?.rescanRequested = false
+                }
                 self?.stateLock.unlock()
-                if again { self?.requestDrain() }
+                if again { self?.requestDrain(expectedGeneration: drainGeneration) }
             }
-            await self?.drainNow()
+            await self?.drainNow(expectedGeneration: drainGeneration)
         }
     }
 
@@ -365,10 +375,11 @@ public final class ProxyRnsBackend: RnsBackend, @unchecked Sendable {
     /// the generation BEFORE the await and re-check it UNDER the lock after, so a
     /// stale drain discards its payload instead of yielding events or resurrecting
     /// `lastSeenAnnounce` for a node that is stopped.
-    private func drainNow() async {
+    private func drainNow(expectedGeneration myGeneration: Int) async {
         stateLock.lock()
-        let myGeneration = startGeneration
+        let isLive = myGeneration == startGeneration
         stateLock.unlock()
+        guard isLive else { return }
         guard let response = try? await roundTrip(.drainEvents, op: "drainEvents"),
               case .ok(let payload) = response, let payload,
               let events = try? JSONDecoder().decode([ProxyEvent].self, from: payload)
@@ -385,8 +396,7 @@ public final class ProxyRnsBackend: RnsBackend, @unchecked Sendable {
         // we have re-emitted the whole batch on the event stream, ack the highest
         // `seq` so the NE advances its cursor and deletes only the processed
         // rows. Ack the max over EVERY event in the batch (not just the ones we
-        // yielded): the NE already emitted each row, so a deduped announce we
-        // skip here is still consumed. An ack is fire-and-forget: if it is lost,
+        // yielded): a deduped announce is still consumed. If an ack is lost,
         // or the app dies before it lands, the unacked rows survive and are
         // re-returned on the next drain (at-least-once; announce dedup + the
         // UI's message-hash idempotency make the re-delivery harmless).
@@ -485,7 +495,22 @@ public final class ProxyRnsBackend: RnsBackend, @unchecked Sendable {
             let stillLive = (myGeneration == startGeneration)
             stateLock.unlock()
             if stillLive {
-                _ = try? await roundTrip(.ackInbox(maxSeq: ackSeq), op: "ackInbox")
+                let reply = try? await roundTrip(.ackInbox(maxSeq: ackSeq), op: "ackInbox")
+                let acknowledged: Bool
+                if case .ok(let payload) = reply, let payload {
+                    acknowledged = (try? JSONDecoder().decode(Bool.self, from: payload)) == true
+                } else {
+                    acknowledged = false
+                }
+                // A lost/failed ack must replay the page, not spin IPC at full
+                // speed. Successful acks immediately fetch the next page, even
+                // when no new Darwin ping arrives while the app catches up.
+                if !acknowledged {
+                    try? await Task.sleep(nanoseconds: 1_000_000_000)
+                }
+                stateLock.lock()
+                if myGeneration == startGeneration { rescanRequested = true }
+                stateLock.unlock()
             }
         }
     }

Tests/ColumbaAppTests/ProxyDrainRaceTests.swift

@@ -1,5 +1,6 @@
 import XCTest
 import Foundation
+import CoreFoundation
 import RNSAPI
 @testable import ColumbaModelBApp
 
@@ -185,3 +186,191 @@ final class ProxyDrainRaceTests: XCTestCase {
                        "a drain that resumed after stop() must not yield events (stale incarnation): \(snapshot)")
     }
 }
+
+extension ProxyDrainRaceTests {
+    /// Catch-up must cross page boundaries without another Darwin notification.
+    func testCatchUpContinuesAfterAckUntilEmpty() async throws {
+        let finished = expectation(description: "all pages acknowledged and empty read reached")
+        let lock = NSLock()
+        var acknowledged: [Int] = []
+        var reads = 0
+        let backend = ProxyRnsBackend(send: { wire in
+            guard let request = try? ProxyIPC.decodeRequest(wire) else { return nil }
+            switch request {
+            case .start:
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode(
+                    ProxyLocalInfo(identityHash: "11", destinationHash: "22"))))
+            case .drainEvents:
+                lock.lock()
+                reads += 1
+                let page = reads
+                let acks = acknowledged
+                lock.unlock()
+                if page <= 2 {
+                    XCTAssertEqual(acks, Array(1..<page), "next page must follow successful ack")
+                    let event = ProxyEvent(kind: "state", t: Double(page), state: "running", seq: page)
+                    return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode([event])))
+                }
+                XCTAssertEqual(page, 3, "an empty page must end catch-up")
+                finished.fulfill()
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode([ProxyEvent]())))
+            case .ackInbox(let seq):
+                lock.lock(); acknowledged.append(seq); lock.unlock()
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode(true)))
+            default:
+                return ProxyIPC.encodeResponse(.ok(nil))
+            }
+        })
+        _ = try await backend.start(startParams())
+        await fulfillment(of: [finished], timeout: 5)
+        await backend.stop()
+        XCTAssertEqual(acknowledged, [1, 2])
+    }
+
+    /// A failed ack retains/replays the page, with a delay to prevent a hot loop.
+    func testFailedAckRetriesSamePageBeforeAdvancing() async throws {
+        let finished = expectation(description: "failed ack retried and catch-up finished")
+        let lock = NSLock()
+        var ackAttempts = 0
+        var reads = 0
+        var failedAt: Date?
+        let backend = ProxyRnsBackend(send: { wire in
+            guard let request = try? ProxyIPC.decodeRequest(wire) else { return nil }
+            switch request {
+            case .start:
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode(
+                    ProxyLocalInfo(identityHash: "11", destinationHash: "22"))))
+            case .drainEvents:
+                lock.lock()
+                reads += 1
+                let attempts = ackAttempts
+                let previousFailure = failedAt
+                lock.unlock()
+                if attempts >= 2 {
+                    finished.fulfill()
+                    return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode([ProxyEvent]())))
+                }
+                if let previousFailure {
+                    XCTAssertGreaterThanOrEqual(Date().timeIntervalSince(previousFailure), 0.9)
+                }
+                let event = ProxyEvent(kind: "state", t: 1, state: "running", seq: 1)
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode([event])))
+            case .ackInbox(let seq):
+                XCTAssertEqual(seq, 1)
+                lock.lock()
+                ackAttempts += 1
+                let succeeds = ackAttempts > 1
+                if !succeeds { failedAt = Date() }
+                lock.unlock()
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode(succeeds)))
+            default:
+                return ProxyIPC.encodeResponse(.ok(nil))
+            }
+        })
+        _ = try await backend.start(startParams())
+        await fulfillment(of: [finished], timeout: 5)
+        await backend.stop()
+        XCTAssertEqual(reads, 3)
+        XCTAssertEqual(ackAttempts, 2)
+    }
+}
+
+extension ProxyDrainRaceTests {
+    func testStopDuringAckDoesNotScheduleAnotherPage() async throws {
+        let ackGate = DrainGate()
+        let lock = NSLock()
+        var reads = 0
+        let backend = ProxyRnsBackend(send: { wire in
+            guard let request = try? ProxyIPC.decodeRequest(wire) else { return nil }
+            switch request {
+            case .start:
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode(
+                    ProxyLocalInfo(identityHash: "11", destinationHash: "22"))))
+            case .drainEvents:
+                lock.lock(); reads += 1; lock.unlock()
+                let event = ProxyEvent(kind: "state", t: 1, state: "running", seq: 1)
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode([event])))
+            case .ackInbox:
+                return await ackGate.request()
+            default:
+                return ProxyIPC.encodeResponse(.ok(nil))
+            }
+        })
+        _ = try await backend.start(startParams())
+        let entered = await ackGate.waitEntered(timeout: 5)
+        XCTAssertTrue(entered)
+        await backend.stop()
+        ackGate.resolve(ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode(true))))
+        try await Task.sleep(nanoseconds: 300_000_000)
+        lock.lock(); let count = reads; lock.unlock()
+        XCTAssertEqual(count, 1, "stopped incarnation must not continue catch-up")
+    }
+}
+
+extension ProxyDrainRaceTests {
+    /// Releasing the old worker must preserve BOTH flags owned by its replacement.
+    /// A notification while the replacement is blocked must coalesce, and its
+    /// trailing pass must still run when the replacement eventually completes.
+    func testRestartWhileOldAckBlockedPreservesNewDrainOwnership() async throws {
+        let oldAck = DrainGate()
+        let newDrain = DrainGate()
+        let caughtUp = expectation(description: "new worker retains its trailing drain")
+        let lock = NSLock()
+        var reads = 0
+        let empty = ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode([ProxyEvent]())))
+        let ack = ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode(true)))
+        defer { oldAck.resolve(ack); newDrain.resolve(empty) }
+        let backend = ProxyRnsBackend(send: { wire in
+            guard let request = try? ProxyIPC.decodeRequest(wire) else { return nil }
+            switch request {
+            case .start:
+                return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode(
+                    ProxyLocalInfo(identityHash: "11", destinationHash: "22"))))
+            case .drainEvents:
+                lock.lock(); reads += 1; let page = reads; lock.unlock()
+                if page == 1 {
+                    let event = ProxyEvent(kind: "state", t: 1, state: "running", seq: 1)
+                    return ProxyIPC.encodeResponse(.ok(try? JSONEncoder().encode([event])))
+                }
+                if page == 2 { return await newDrain.request() }
+                XCTAssertEqual(page, 3, "old worker must not create an extra drain")
+                caughtUp.fulfill()
+                return empty
+            case .ackInbox:
+                return await oldAck.request()
+            default:
+                return ProxyIPC.encodeResponse(.ok(nil))
+            }
+        })
+        _ = try await backend.start(startParams())
+        let oldEntered = await oldAck.waitEntered(timeout: 5)
+        XCTAssertTrue(oldEntered)
+        await backend.stop()
+        _ = try await backend.start(startParams())
+        let newEntered = await newDrain.waitEntered(timeout: 5)
+        XCTAssertTrue(newEntered)
+
+        func ping() {
+            CFNotificationCenterPostNotification(
+                CFNotificationCenterGetDarwinNotifyCenter(),
+                CFNotificationName("network.columba.events" as CFString), nil, nil, true)
+        }
+        // Set the new worker's trailing-pass flag, then finish the obsolete worker.
+        ping()
+        try await Task.sleep(nanoseconds: 100_000_000)
+        oldAck.resolve(ack)
+        try await Task.sleep(nanoseconds: 300_000_000)
+        lock.lock(); let afterOldReply = reads; lock.unlock()
+        XCTAssertEqual(afterOldReply, 2, "old completion must not start a drain")
+        // If the old defer cleared `draining`, this ping starts a concurrent read.
+        ping()
+        try await Task.sleep(nanoseconds: 100_000_000)
+        lock.lock(); let whileNewBlocked = reads; lock.unlock()
+        XCTAssertEqual(whileNewBlocked, 2, "replacement must still own draining")
+        newDrain.resolve(empty)
+        await fulfillment(of: [caughtUp], timeout: 5)
+        await backend.stop()
+        lock.lock(); let finalReads = reads; lock.unlock()
+        XCTAssertEqual(finalReads, 3)
+    }
+}