rastrillo / native Public

Isolate repeated send throws to their lane and document snapshot availability

Paul Campbell pushed by paul@keymail.dev 1 month ago 0ff1de9d542ce96430e158708e62be04d87d65a4 parent d91d10c

4 files changed, +57 −5

Sources/RastrilloApple/OutboxDrainer.swift +18 −3
Sources/RastrilloNative/Outbox.swift +8 −2
Tests/RastrilloAppleTests/OutboxDrainerTests.swift +22 −0
docs/testing/k3-evidence.md +9 −0

CI failed — run details

diff --git a/Sources/RastrilloApple/OutboxDrainer.swift b/Sources/RastrilloApple/OutboxDrainer.swift

index d3fc1cf..9dde528 100644

--- a/Sources/RastrilloApple/OutboxDrainer.swift

+++ b/Sources/RastrilloApple/OutboxDrainer.swift

@@ -13,7 +13,7 @@ import RastrilloNative

/// | `.unauthorized` | session ended: stop every lane; never hold or drop |

/// | `.rateLimited(after)` | back off at least `after` |

/// | `.server`, `.transport`, `.invalidResponse` | back off, retry the same batch |

-/// | `BlobStoreError.unavailable` | store locked: wait for the next change (foreground, unlock) |

+/// | `BlobStoreError.unavailable` | when thrown by send, retry this lane; hold after three consecutive throws |

/// | `.forbidden`, `.notFound`, `.conflict`, `.tooLarge`, `.rejected` | hold the batch rather than spin |

///

/// A failure to acknowledge or hold after a send is a `.retry`: the server

@@ -46,7 +46,7 @@ import RastrilloNative

case waitingForNetwork

/// Retry after this delay (a connectivity change may cut it short).

case backingOff(TimeInterval)

- /// The server's Retry-After: nothing but "Try again" crosses it.

+ /// The server’s Retry-After: neither connectivity nor "Try again" crosses it.

case rateLimited(TimeInterval)

case sessionEnded

case storeUnavailable

@@ -67,6 +67,7 @@ import RastrilloNative

private let maxBatchBytes: Int?

private var backoff: Backoff

@ObservationIgnored private var knownStatuses: [String: SyncStatus] = [:]

+ private var consecutiveThrows: [String: Int] = [:]

private var draining = false

private var drainWaiters: [CheckedContinuation<Outcome, Never>] = []

private var worker: Task<Void, Never>?

@@ -172,8 +173,12 @@ import RastrilloNative

phase = .sending

do {

let dispositions: [UUID: Disposition]

- do { dispositions = try await send(first.lane, batch, transport) }

+ do {

+ dispositions = try await send(first.lane, batch, transport)

+ consecutiveThrows[first.lane] = nil

+ }

catch let error as NativeClientError {

+ consecutiveThrows[first.lane] = nil

let hold: Hold?

switch error {

case .forbidden, .notFound: hold = Hold(.accessLost)

@@ -185,6 +190,16 @@ import RastrilloNative

guard let hold else { throw error }

dispositions = Dictionary(uniqueKeysWithValues: ids.map { ($0, .hold(hold)) })

}

+ catch {

+ guard !Task.isCancelled else { throw error }

+ let count = (consecutiveThrows[first.lane] ?? 0) + 1

+ consecutiveThrows[first.lane] = count

+ // An app decoder/programming failure belongs to this lane, not the

+ // account. Three consecutive throws require an explicit resolution.

+ let disposition: Disposition = count >= 3 ? .hold(Hold(.rejected("send"))) : .retry

+ dispositions = Dictionary(uniqueKeysWithValues: ids.map { ($0, disposition) })

+ if count >= 3 { consecutiveThrows[first.lane] = nil }

+ }

var mustRetry = false; var progressed = false

for entry in batch {

switch dispositions[entry.id] {

diff --git a/Sources/RastrilloNative/Outbox.swift b/Sources/RastrilloNative/Outbox.swift

index 0ddc905..bf36eed 100644

--- a/Sources/RastrilloNative/Outbox.swift

+++ b/Sources/RastrilloNative/Outbox.swift

@@ -142,10 +142,15 @@ public actor Outbox {

try persist(entry); values.append(entry); next += 1; return entry

}

- public func entries(lane: String) -> [OutboxEntry] {

+ /// A last-known snapshot, including held entries. On an unavailable store

+ /// this may be stale, just like summary (whose unavailable flag is set).

+ /// Use it for display/deduplication hints only; replay must first call load()

+ /// successfully. Every mutation and beginSend independently require load().

+ public func entries(lane: String, excludingInFlight: Bool = false) -> [OutboxEntry] {

try? load()

- return values.filter { $0.lane == lane }

+ return values.filter { $0.lane == lane && (!excludingInFlight || !flying.contains($0.id)) }

}

+ /// Fails closed to an empty batch when loading is unavailable.

public func sendable(lane: String, maxBytes: Int? = nil) -> [OutboxEntry] {

try? load()

guard !unavailable else { return [] }

@@ -241,6 +246,7 @@ public actor Outbox {

}

return result

}

+ /// Last-known counts; unavailable marks retained counts as non-authoritative.

public var summary: OutboxSummary { try? load(); return currentSummary }

private func notify() { for subscriber in subscribers.values { subscriber.yield(currentSummary) } }

private func unsubscribe(_ id: UUID) { subscribers[id] = nil }

diff --git a/Tests/RastrilloAppleTests/OutboxDrainerTests.swift b/Tests/RastrilloAppleTests/OutboxDrainerTests.swift

index b27b4f9..cc1d7f4 100644

--- a/Tests/RastrilloAppleTests/OutboxDrainerTests.swift

+++ b/Tests/RastrilloAppleTests/OutboxDrainerTests.swift

@@ -38,6 +38,28 @@ import XCTest

transport.committed.map { String(decoding: $0.body ?? Data(), as: UTF8.self) }

}

+ func testThrowingLaneDoesNotStarveOtherLanesAndEventuallyHolds() async throws {

+ struct Broken: Error {}

+ let (connectivity, _) = await connectivity()

+ let outbox = try await outbox([("a", "broken"), ("b", "good"), ("c", "good")])

+ let transport = ScriptedTransport([])

+ var calls: [String] = []

+ let drainer = OutboxDrainer(outbox: outbox, transport: { transport }, connectivity: connectivity) { lane, entries, _ in

+ calls.append(lane)

+ if lane == "a" { throw Broken() }

+ return Dictionary(uniqueKeysWithValues: entries.map { ($0.id, .sent) })

+ }

+ await drainer.drain()

+ XCTAssertEqual(calls, ["a", "b", "c"])

+ let remaining = await outbox.summary

+ XCTAssertEqual(remaining.lanes.count, 1)

+ await drainer.drain(); await drainer.drain()

+ let held = await outbox.entries(lane: "a")

+ XCTAssertEqual(held.first?.hold, Hold(.rejected("send")))

+ await drainer.drain()

+ XCTAssertEqual(calls.filter { $0 == "a" }.count, 3, "Three consecutive throws hold the lane instead of spinning")

+ }

func testConcurrentDrainWaitsForTheReservedBatch() async throws {

let (connectivity, _) = await connectivity()

let outbox = try await outbox([("doc:a", "1")])

diff --git a/docs/testing/k3-evidence.md b/docs/testing/k3-evidence.md

index 5de288c..a55964f 100644

--- a/docs/testing/k3-evidence.md

+++ b/docs/testing/k3-evidence.md

@@ -80,3 +80,12 @@ No update or retarget API was added.

The only deliberate ordering difference is app-side discard removing the replica

before its lane; exhaustive crash tests prove this prevents stranded unsent text.

The prior empty-session status exception is removed. No merge or deployment.

+## Fable review follow-up (2026-09-24)

+- `testSealBlocksCacheStateAndSkipUnusedQuotaAfterCrash`: failed against the inherited per-seal implementation (38 high-water assertions and two cache assertions); passes with 1024-slot reservations. Two process lifetimes consume 19 seals each, observe durable marks 1024/2048, and produce 38 distinct 96-bit nonces. A restart abandons 1005 unused slots; key rotation still counts reseals and keeps its previous key until completion. Complete file protection remains in every atomic write.

+- `testThrowingLaneDoesNotStarveOtherLanesAndEventuallyHolds`: failed against the inherited drainer (only lane a called, three lanes remained, no hold, fourth throw); passes with b/c draining on the first attempt and a held after its third consecutive untyped throw. Returning a disposition or a typed transport error resets the throw streak.

+- Full Swift suite: 119 tests passed. Red/green logs: `/tmp/k3-native-review-{red,green}.log` (ephemeral); observed assertions above are the durable record.

+- Reservation API addition: `entries(lane:excludingInFlight:)` adds a filter to the existing snapshot, needed to deduplicate held metadata markers without suppressing a successor to a marker already being sent. No new mutation or reservation operation.

+- Snapshot invariant is now documented: entries/summary can retain stale display/deduplication hints on unavailable storage; summary flags this, sendable fails closed, and load/beginSend/all mutations enforce availability before replay or changes.

+- Corrected unavailable-send and Retry-After comments. The latter remains a deadline that neither connectivity nor Try again bypasses.