diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 5393b5d8..e6dccfbb 100644 --- a/.github/workflows/tests.yml +++ b/.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 diff --git a/Sources/ColumbaNetworkExtension/NEPythonRNS.swift b/Sources/ColumbaNetworkExtension/NEPythonRNS.swift index 489e2ef3..19a0439d 100644 --- a/Sources/ColumbaNetworkExtension/NEPythonRNS.swift +++ b/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, ...}`. diff --git a/Sources/ColumbaNetworkExtension/PacketTunnelProvider.swift b/Sources/ColumbaNetworkExtension/PacketTunnelProvider.swift index 1b5b1a78..1fa2fc3a 100644 --- a/Sources/ColumbaNetworkExtension/PacketTunnelProvider.swift +++ b/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 diff --git a/Sources/RNSBackendProxy/ProxyRnsBackend.swift b/Sources/RNSBackendProxy/ProxyRnsBackend.swift index 8c21edb2..682a73d9 100644 --- a/Sources/RNSBackendProxy/ProxyRnsBackend.swift +++ b/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() } } } diff --git a/Tests/ColumbaAppTests/ProxyDrainRaceTests.swift b/Tests/ColumbaAppTests/ProxyDrainRaceTests.swift index 4301cf2b..8d9ac6a3 100644 --- a/Tests/ColumbaAppTests/ProxyDrainRaceTests.swift +++ b/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..= 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) + } +} diff --git a/Tests/static/test_rns_bridge_inbox.py b/Tests/static/test_rns_bridge_inbox.py new file mode 100644 index 00000000..abf3ba35 --- /dev/null +++ b/Tests/static/test_rns_bridge_inbox.py @@ -0,0 +1,99 @@ +"""Exercise the real inbox functions with SQLite, without loading iOS/RNS. + +Only the named functions/constants are extracted; their SQL and replay logic run +unchanged. Run: python3 -m unittest Tests.static.test_rns_bridge_inbox +""" +import ast +import json +import sqlite3 +import threading +import unittest +from pathlib import Path +from types import SimpleNamespace + + +class InboxReplayTests(unittest.TestCase): + def setUp(self): + source = ast.parse((Path(__file__).resolve().parents[2] / 'app/rns_bridge.py').read_text()) + names = {'drain_inbox', 'ack_inbox', '_quarantine_inbox_event'} + nodes = [n for n in source.body if + isinstance(n, ast.FunctionDef) and n.name in names or + isinstance(n, ast.Assign) and any(isinstance(t, ast.Name) and t.id.startswith('_INBOX_BATCH_') for t in n.targets)] + self.db = sqlite3.connect(':memory:') + self.db.execute('CREATE TABLE ne_inbox(seq INTEGER PRIMARY KEY, kind TEXT, payload TEXT, created_at REAL)') + self.ns = dict(json=json, time=__import__('time'), _inbox_lock=threading.RLock(), + _open_inbox=lambda: None, _inbox_conn=self.db, + RNS=SimpleNamespace(log=lambda *args: None, LOG_ERROR=1, LOG_WARNING=2)) + exec(compile(ast.Module(body=nodes, type_ignores=[]), 'rns_bridge.py', 'exec'), self.ns) + self.addCleanup(self.db.close) + + def add(self, payload, kind='announce'): + self.db.execute('INSERT INTO ne_inbox(kind,payload,created_at) VALUES(?,?,0)', (kind, payload)) + self.db.commit() + + def drain(self): + return self.ns['drain_inbox']() + + def test_overnight_backlog_is_bounded_and_drains_in_order(self): + for i in range(12604): + self.add(json.dumps(dict(kind='announce', t=i, destination_hash=str(i)))) + seen = [] + while batch := self.drain(): + self.assertLessEqual(len(batch), 128) + self.assertLessEqual(len(json.dumps(batch).encode()), 256 * 1024) + self.assertEqual(batch, self.drain(), 'read must not advance the cursor') + seen.extend(e['seq'] for e in batch) + self.assertTrue(self.ns['ack_inbox'](batch[-1]['seq'])) + self.assertEqual(seen, list(range(1,12605))) + + def test_byte_budget_splits_batch_before_row_limit(self): + for i in range(40): + self.add(json.dumps(dict(kind='inbound', t=i, content='x' * 20000))) + batch = self.drain() + self.assertLess(len(batch), 40) + self.assertLessEqual(len(json.dumps(batch).encode()), 256 * 1024) + + def test_escaped_unicode_counts_as_wire_bytes(self): + for i in range(40): + self.add(json.dumps(dict(kind='inbound', t=i, content='๐Ÿ˜€' * 1000), ensure_ascii=False)) + self.assertLessEqual(len(json.dumps(self.drain()).encode()), 256 * 1024) + + def test_poison_rows_are_preserved_and_do_not_block_following_events(self): + for raw in ('{bad', '[]', json.dumps(dict(kind='inbound', t=0, content='x' * 300000))): + self.add(raw) + self.add(json.dumps(dict(kind='announce', t=1))) + batch = self.drain() + self.assertEqual([e['seq'] for e in batch], [1,2,3,4]) + self.assertEqual([e['kind'] for e in batch], ['inbox_error'] * 3 + ['announce']) + self.assertEqual(self.db.execute('SELECT count(*) FROM ne_inbox').fetchone()[0], 4) + self.assertTrue(self.ns['ack_inbox'](4)) + self.assertEqual(self.db.execute('SELECT count(*) FROM ne_inbox_quarantine').fetchone()[0], 3) + self.assertEqual(self.drain(), []) + + def test_single_event_expanding_past_wire_budget_is_quarantined(self): + self.add(json.dumps(dict(kind='inbound', t=0, content='๐Ÿ˜€' * 30000), ensure_ascii=False)) + batch = self.drain() + self.assertEqual(batch[0]['kind'], 'inbox_error') + self.assertLessEqual(len(json.dumps(batch).encode()), 256 * 1024) + self.assertEqual(self.db.execute('SELECT count(*) FROM ne_inbox_quarantine').fetchone()[0], 1) + + def test_nonfinite_json_cannot_poison_the_swift_decoder(self): + self.add('{"kind":"state","t":NaN}') + self.assertEqual(self.drain()[0]['kind'], 'inbox_error') + + def test_ack_failure_is_reported_and_rows_remain_replayable(self): + self.add(json.dumps(dict(kind='state', t=0))) + self.db.execute("CREATE TRIGGER refuse_delete BEFORE DELETE ON ne_inbox BEGIN SELECT RAISE(ABORT,'test'); END") + self.assertIs(self.ns['ack_inbox'](1), False) + self.assertEqual(self.drain()[0]['seq'], 1) + + def test_ack_leaves_newer_rows_and_is_idempotent(self): + self.add(json.dumps(dict(kind='state', t=0))) + self.add(json.dumps(dict(kind='state', t=1))) + self.assertTrue(self.ns['ack_inbox'](1)) + self.assertTrue(self.ns['ack_inbox'](1)) + self.assertEqual(self.drain()[0]['seq'], 2) + + +if __name__ == '__main__': + unittest.main() diff --git a/Tests/static/test_rns_bridge_persistence.py b/Tests/static/test_rns_bridge_persistence.py index 04bbfbcc..709269f3 100644 --- a/Tests/static/test_rns_bridge_persistence.py +++ b/Tests/static/test_rns_bridge_persistence.py @@ -40,7 +40,14 @@ def _extract(names): tree = ast.parse(BRIDGE.read_text()) wanted = set(names) - return [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in wanted] + # Extract the real bounded-read dependencies as well as the functions. + # Omitting these makes drain_inbox catch NameError and return an empty page. + if "drain_inbox" in wanted: + wanted.update({"_quarantine_inbox_event", "_INBOX_BATCH_ROWS", "_INBOX_BATCH_BYTES"}) + return [n for n in tree.body if + isinstance(n, ast.FunctionDef) and n.name in wanted or + isinstance(n, ast.Assign) and any( + isinstance(t, ast.Name) and t.id in wanted for t in n.targets)] class RnsBridgePersistenceTest(unittest.TestCase): @@ -61,6 +68,7 @@ def _fresh(self): "RNS": types.SimpleNamespace( log=lambda *a, **k: None, LOG_ERROR=0, + LOG_WARNING=0, ), "_inbox_path": None, "_inbox_conn": None, @@ -277,6 +285,7 @@ def _build(self): "RNS": types.SimpleNamespace( log=lambda *a, **k: None, LOG_ERROR=0, + LOG_WARNING=0, Reticulum=lambda *a, **k: (_ for _ in ()).throw( self._StopAfterPersistence() ), @@ -580,6 +589,7 @@ def _fresh(self): "RNS": types.SimpleNamespace( log=lambda *a, **k: None, LOG_ERROR=0, + LOG_WARNING=0, ), "_inbox_path": None, "_inbox_conn": None, diff --git a/app/rns_bridge.py b/app/rns_bridge.py index 0166d409..11429432 100644 --- a/app/rns_bridge.py +++ b/app/rns_bridge.py @@ -586,46 +586,87 @@ def _publish_durable(kind: str, payload: dict) -> None: _coalesced_events_ping() +# Bound both Python objects and the JSON copied/decoded across the NE/app seam. +_INBOX_BATCH_ROWS = 128 +_INBOX_BATCH_BYTES = 256 * 1024 + + +def _quarantine_inbox_event(seq: int, reason: str) -> dict: + """Preserve an unreadable/oversized row on disk, without loading its payload. + + Return a small cursor-bearing marker. Normal ack removes the inbox copy; + the original remains in ne_inbox_quarantine for recovery/inspection. + """ + _inbox_conn.execute( + "CREATE TABLE IF NOT EXISTS ne_inbox_quarantine (" + "seq INTEGER PRIMARY KEY, kind TEXT NOT NULL, payload TEXT NOT NULL, " + "reason TEXT NOT NULL)" + ) + _inbox_conn.execute( + "INSERT OR IGNORE INTO ne_inbox_quarantine(seq,kind,payload,reason) " + "SELECT seq,kind,payload,? FROM ne_inbox WHERE seq=?", (reason, seq) + ) + _inbox_conn.commit() + RNS.log(f"rns_bridge: inbox row {seq} quarantined: {reason}", RNS.LOG_WARNING) + return {"kind": "inbox_error", "t": 0, "seq": seq, "reason": reason} + + def drain_inbox() -> list: - """Read the durable inbox, oldest-first, WITHOUT deleting. Returns the raw - event dicts (kind + t + the original payload fields) PLUS a `_seq` field so - the consumer can ack precisely. The app maps each onto a BackendEvent and - feeds its `backend.events` consumer, then advances the cursor via - `ack_inbox(max_seq)` once it has processed the batch. - - Contract ยง5: the read is bounded and NON-destructive. Deleting on read (the - prior behavior) meant a lost IPC reply or an app termination before the - consumer processed the batch lost delivery-state updates and announces - permanently. With the cursor, unacked rows survive and are re-returned on - the next drain; `ack_inbox` advances past them only after the consumer has - handled them (at-least-once delivery).""" + """Read an ordered, bounded page WITHOUT advancing the durable cursor. + + The app acknowledges the highest seq after yielding the page, then requests + another page until empty. Lost replies and app death leave rows replayable. + Count default json.dumps wire bytes, including ASCII escaping, array brackets + and separators. Inspect sizes in SQLite before fetching payloads so a single + oversized row cannot recreate the startup memory spike. Poison rows are + preserved in quarantine and represented by small ackable error markers. + """ with _inbox_lock: _open_inbox() if _inbox_conn is None: return [] try: - cur = _inbox_conn.execute( - "SELECT seq, kind, payload FROM ne_inbox ORDER BY seq ASC" - ) - rows = cur.fetchall() - if not rows: - return [] + rows = _inbox_conn.execute( + "SELECT seq, length(CAST(payload AS BLOB)) FROM ne_inbox " + "ORDER BY seq ASC LIMIT ?", (_INBOX_BATCH_ROWS,) + ).fetchall() out = [] - for seq, kind, payload in rows: + wire_bytes = 2 # JSON array brackets + for seq, size in rows: + if size is None or size > _INBOX_BATCH_BYTES: + event = _quarantine_inbox_event(seq, "payload exceeds replay byte limit") + else: + payload = _inbox_conn.execute( + "SELECT payload FROM ne_inbox WHERE seq=?", (seq,) + ).fetchone()[0] + try: + event = json.loads(payload) + if not isinstance(event, dict): + raise ValueError("event must be an object") + event["seq"] = seq + except (ValueError, TypeError, RecursionError): + event = _quarantine_inbox_event(seq, "invalid event JSON") try: - event = json.loads(payload) - except Exception: # noqa: BLE001 - continue - event["seq"] = seq + event_bytes = len(json.dumps(event, allow_nan=False).encode("utf-8")) + except (ValueError, TypeError, RecursionError): + event = _quarantine_inbox_event(seq, "event cannot be encoded") + event_bytes = len(json.dumps(event, allow_nan=False).encode("utf-8")) + if event_bytes + 2 > _INBOX_BATCH_BYTES: + event = _quarantine_inbox_event(seq, "encoded event exceeds replay byte limit") + event_bytes = len(json.dumps(event, allow_nan=False).encode("utf-8")) + added_bytes = event_bytes + (2 if out else 0) + if wire_bytes + added_bytes > _INBOX_BATCH_BYTES: + break out.append(event) + wire_bytes += added_bytes return out except Exception as e: # noqa: BLE001 RNS.log(f"rns_bridge: inbox drain failed: {e}", RNS.LOG_ERROR) return [] -def ack_inbox(max_seq: int) -> None: - """Advance the inbox consumer cursor: delete rows at or below `max_seq`. +def ack_inbox(max_seq: int) -> bool: + """Advance the inbox cursor; return True only after a successful commit. Called by the app over IPC AFTER it has processed a drained batch (the highest `_seq` in that batch is the cursor). Rows above the cursor - i.e. @@ -636,14 +677,17 @@ def ack_inbox(max_seq: int) -> None: with _inbox_lock: _open_inbox() if _inbox_conn is None: - return + return False try: _inbox_conn.execute( "DELETE FROM ne_inbox WHERE seq <= ?", (int(max_seq),) ) _inbox_conn.commit() + return True except Exception as e: # noqa: BLE001 + _inbox_conn.rollback() RNS.log(f"rns_bridge: inbox ack failed: {e}", RNS.LOG_ERROR) + return False # โ”€โ”€ Shared GRDB message store (NE โ†’ app, Model B) โ”€โ”€ diff --git a/docs/MODEL_B_BACKGROUND_DELIVERY.md b/docs/MODEL_B_BACKGROUND_DELIVERY.md index 0e356388..5a5ecae8 100644 --- a/docs/MODEL_B_BACKGROUND_DELIVERY.md +++ b/docs/MODEL_B_BACKGROUND_DELIVERY.md @@ -93,6 +93,29 @@ ONE destination on the NE. The app is a thin client that marshals node-owning op This path runs entirely in the extension while `ColumbaModelBApp` is suspended or locked. +## Durable inbox replay limits + +The extension returns at most 128 events and 256 KiB of Python event JSON per +`drainEvents` request. The byte budget includes JSON escaping and separators; +IPC envelopes add overhead. Reads preserve SQLite rows until `ackInbox` commits. +The app then requests the next page until empty, without waiting for another +Darwin notification. Failed or lost acknowledgments trigger a delayed replay; +a stopped or restarted backend invalidates the old catch-up worker. + +Malformed events and individual events exceeding the byte limit are copied to +`ne_inbox_quarantine` in the same identity's `ne-inbox.db`. This table retains +`seq`, `kind`, the original `payload`, and `reason`. A small `inbox_error` marker +lets normal acknowledgment advance past the row. The proxy consumes that +marker without yielding a backend event. A warning records its sequence and +reason; payload contents are not logged. Quarantined rows are retained for +manual recovery, not automatically retried or pruned. Message persistence in +the shared message store is independent of inbox replay. + +Regression checks: `python3 -m unittest Tests.static.test_rns_bridge_inbox` and the +`ColumbaModelBAppTests/ProxyDrainRaceTests` Xcode tests. Device validation must +include backlog catch-up, a cold restart, and background delivery; passing a +short startup test does not establish overnight reliability. + ## Outbound/send flow 1. The app calls `ProxyRnsBackend.sendLxmfMessage(...)` with one app-assigned stable `sendId` per logical send (architecture review P1 #6).