rastrillo / native Public

Preserve session status priority and await concurrent drain completion

Paul Campbell pushed by paul@keymail.dev 1 month ago 3e93ef916f2bb5469fd09375e0f6253938da8650 parent 6b6b198

5 files changed, +49 −5

Sources/RastrilloApple/OutboxDrainer.swift +9 −2
Sources/RastrilloNative/SyncStatus.swift +1 −1
Tests/RastrilloAppleTests/OutboxDrainerTests.swift +22 −0
Tests/RastrilloNativeTests/SyncStatusTests.swift +2 −2
docs/testing/k3-evidence.md +15 −0

CI failed — run details

Intent

full ledger: feat/k3-outbox

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

index 604392f..aab2fc0 100644

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

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

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

private let maxBatchBytes: Int?

private var backoff: Backoff

private var draining = false

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

private var worker: Task<Void, Never>?

private var observation: Task<Void, Never>?

private var networkObservation: Task<Void, Never>?

@@ -133,9 +134,15 @@ import RastrilloNative

private func retry() -> Outcome { phase = .backingOff; return .backingOff(backoff.next()) }

@discardableResult public func drain() async -> Outcome {

- guard !draining else { return .idle }

+ if draining { return await withCheckedContinuation { drainWaiters.append($0) } }

draining = true

- defer { draining = false }

+ let outcome = await drainOnce()

+ draining = false

+ let waiters = drainWaiters; drainWaiters.removeAll()

+ for waiter in waiters { waiter.resume(returning: outcome) }

+ return outcome

+ }

+ private func drainOnce() async -> Outcome {

do { try await outbox.load() }

catch { summary = await outbox.summary; phase = .storeUnavailable; return .storeUnavailable }

summary = await outbox.summary

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

index f97f943..a3cde75 100644

--- a/Sources/RastrilloNative/SyncStatus.swift

+++ b/Sources/RastrilloNative/SyncStatus.swift

@@ -49,8 +49,8 @@ public enum SyncStatus: Sendable, Equatable {

self = .failed(pending: pending, reason: .held(held))

} else if let localFailure { self = .failed(pending: pending, reason: localFailure) }

else if summary.unavailable { self = .failed(pending: pending, reason: .cannotSaveOnDevice) }

- else if summary.isEmpty && summary.inFlight == 0 { self = .saved }

else if sessionEnded { self = .needsSignIn(pending: pending) }

+ else if summary.isEmpty && summary.inFlight == 0 { self = .saved }

else if sending || summary.inFlight > 0 { self = .saving }

else if offline { self = .offline(pending: pending) }

else if backingOff { self = .retrying(pending: pending) }

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

index aa3df64..163de59 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 testConcurrentDrainWaitsForTheReservedBatch() async throws {

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

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

+ let transport = ScriptedTransport([])

+ var release: CheckedContinuation<Void, Never>?

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

+ await withCheckedContinuation { release = $0 }

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

+ }

+ let first = Task { await drainer.drain() }

+ await settle { release != nil }

+ var finished = false

+ let second = Task { let outcome = await drainer.drain(); finished = true; return outcome }

+ try await Task.sleep(for: .milliseconds(30))

+ XCTAssertFalse(finished, "A concurrent close-save cannot report idle before acknowledgement")

+ release?.resume()

+ let outcomes = await (first.value, second.value)

+ XCTAssertEqual(outcomes.0, .idle); XCTAssertEqual(outcomes.1, .idle)

+ let summary = await outbox.summary

+ XCTAssertTrue(summary.isEmpty)

+ }

func testDrainsEachLaneInOrderAndAcknowledges() async throws {

let (connectivity, _) = await connectivity()

let transport = ScriptedTransport([.respond(status: 204), .respond(status: 204)])

diff --git a/Tests/RastrilloNativeTests/SyncStatusTests.swift b/Tests/RastrilloNativeTests/SyncStatusTests.swift

index ca3c9c8..14021c0 100644

--- a/Tests/RastrilloNativeTests/SyncStatusTests.swift

+++ b/Tests/RastrilloNativeTests/SyncStatusTests.swift

@@ -42,8 +42,8 @@ final class SyncStatusTests: XCTestCase {

func testOfflineWithNothingPendingIsSilentlySaved() {

let status = SyncStatus(summary: OutboxSummary(), sending: false, sessionEnded: false, offline: true, backingOff: false)

XCTAssertEqual(status, .saved, "\"Offline · 0 pending\" must never appear")

- XCTAssertEqual(SyncStatus(summary: OutboxSummary(), sending: false, sessionEnded: true, offline: true, backingOff: false), .saved,

- "\"Sign in to send 0 pending\" must never appear either")

+ XCTAssertEqual(SyncStatus(summary: OutboxSummary(), sending: false, sessionEnded: true, offline: true, backingOff: false), .needsSignIn(pending: 0),

+ "Session-ended has priority over an empty outbox (spec 3.4)")

}

func testNCountsLanesByDefaultAndWhatTheAppChooses() {

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

index 8165ae0..c463970 100644

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

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

@@ -45,3 +45,18 @@ This branch is WIP and must not merge.

Docs adoption is saved on oficina/docs `feat/k3-outbox-lane`, discussion #69.

Its evidence file lists unresolved title/lifecycle issues and remaining gates.

+## Resume: status priority and concurrent close-save

+Spec §3.4 gives session-ended priority 2 and empty priority 3. Pending entries

+already produced needsSignIn(n); the empty-session exception is now removed too.

+A concurrent drain now awaits the reserved batch's result, so close-save cannot

+mistake an in-flight batch for idle. New checks failed on the previous behavior

+(2 assertions, /tmp/k3-status-drain-red.log); all 20 status/drainer tests then

+passed (/tmp/k3-status-drain-green.log).

+Reservation API rationale: load surfaces unavailable storage rather than an

+empty queue; entries includes held entries for offline replay; beginSend makes

+selection and reservation one actor operation so compaction cannot delete a

+batch in flight; endSend releases a retry/cancelled reservation without ack.

+No update or retarget API was added.