Refactor actor delivery and domain storage boundaries

benches/tier3_system_kv.rs

@@ -39,7 +39,7 @@ fn begin_transaction(
             resource,
         ),
         mode,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
 
     match response {

benches/tier4_kv_support.rs

@@ -111,7 +111,7 @@ fn direct_lifecycle(actor: &mut DirectKvActor, commit: bool, key: Bytes, value:
     let begin = actor.actor.handle(KvMessage::Begin {
         scope: scope.clone(),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::sync(),
+        write_options: cntryl_midge::WriteOptions::sync().into(),
     });
     let KvResponse::BeginOk { tx_id } = begin else {
         panic!("KV begin failed: {begin:?}")

docs/development/architecture.md

@@ -240,6 +240,7 @@ shard. The transport/router edge only enqueues work:
 - Shards drain ready families round-robin so one noisy family cannot monopolize
   a worker.
 #### Domain Handles
+See [runtime and storage boundaries](runtime-storage-boundaries.md) for delivery error contracts, optional mailbox priority, KV write policy, and Queue recovery ownership.
 `DomainHandles` owns the concrete domain sinks but keeps those fields private. Boot, background maintenance, metrics, and admin query code must use explicit handle or `Runtime::*` facade methods so concrete sink internals do not become a public mutable API.
 There is no active `runtime::Scheduler` API. The legacy scheduler module is
 test-only while managed domain actors are migrated to family-owned workers.

docs/development/runtime-storage-boundaries.md

@@ -0,0 +1,59 @@
+# Runtime and storage boundaries
+
+## Delivery failures
+
+`ActorRef::send_detailed` and the `Context::*_detailed` sending methods return
+`RouteError` with the exact destination and delivery failure. A timeout does not
+establish that an actor stopped. A payload rejection does not establish that a
+sink panicked. Invalid-payload errors retain the actual size and wire limit.
+
+The existing `send`, `send_untracked`, `publish_event`, and `reply` methods retain
+the original four-variant `SendError` and its legacy classification for source
+and behavior compatibility. Callers that need to distinguish timeout, invalid
+payload, and unsupported payload must use the detailed methods.
+
+`MailboxSink::deliver_high_priority` remains required. Every sink explicitly
+chooses its handling: managed actor mailboxes use their separate bounded control
+lane, while single-lane sinks such as session outbound transport explicitly
+forward to `deliver`. The trait provides no automatic priority fallback. Code
+requiring reserved control capacity must use an implementation that provides it.
+
+## KV write policy
+
+`KvMessage::Begin` carries `domains::WritePolicy`, a Fitz-owned guarantee type.
+The wire codec produces `Buffered` for flag 0 and `Sync` for flag 1. The domain
+sink maps those requests to the broker's configured local or cloud policies;
+explicit `BestEffort`, `CloudAsync`, and `CloudStrict` requests retain their
+meaning. No policy has a default. The wire inventory and configuration resolver
+live together in `domains/kv/write_policy.rs`; the codec and sink share them.
+
+Conversions to and from Midge `WriteOptions` live in `src/storage/write_policy.rs`.
+Existing engine-based construction and broker configuration methods continue
+accepting Midge options. The actor converts the resolved policy when creating
+its engine transaction state. Midge remains the concrete storage engine.
+
+## Queue recovery
+
+`QueueRecoveryStore` owns recovery transactions, index queries, row decoding,
+and atomic index replacement. One store is created per actor; both normal writes
+and recovery reuse its cached keys and reference-counted scan prefixes. Recovery
+clones one store handle, without copying queue identity strings or rebuilding
+prefixes on its error and fallback paths.
+
+Index metadata, ID reservation, ready/delayed/dead-letter rows, and fallback
+headers all use the same read snapshot. If index metadata is invalid, the
+reserved-ID row supplies the fallback floor; invalid index metadata is not an
+authority for that floor. Ordinary typed iterators decode each scan lazily and
+borrow the snapshot. Header recovery no longer retains a vector of complete
+header records. The live recovered state and ready-ID sorting still scale with
+the recovered queue; this is not a constant-memory recovery guarantee.
+
+`QueueActor` owns index-counter validation, fallback selection, live ready and
+delayed state reconstruction, and the recovered ID boundary. The store receives
+a borrowed index-rebuild description and commits stale-index deletion, new
+entries, and metadata together. A failed replacement commit does not publish a
+partially replaced index. Recovery assumes the existing single-owner queue
+lifecycle; snapshot consistency does not authorize concurrent queue writers.
+
+Storage formats, acknowledgement guarantees, RouteFamily isolation, and
+ephemeral inflight ownership are unchanged.

docs/operations/migration-guide.md

@@ -2,6 +2,31 @@
 
 This guide covers safe upgrades between Fitz releases.
 
+## Rust embedding API: write policies and delivery errors
+
+`KvMessage::Begin::write_options` now takes `fitz::domains::WritePolicy` so the
+domain message no longer exposes the storage engine's option type. Use a variant
+such as `WritePolicy::Buffered`, or add `.into()` to an existing Midge expression:
+
+```rust,ignore
+write_options: cntryl_midge::WriteOptions::buffered().into(),
+```
+
+Broker configuration methods such as `KvDomainSink::with_write_options` retain
+their Midge option parameters. KV wire flags and persisted data formats do not
+change; network SDK consumers need no migration for this refactor.
+
+Existing exhaustive matches on `SendError` remain compatible: its variants and
+legacy mappings are unchanged. Use `ActorRef::send_detailed` or the corresponding
+`Context::*_detailed` methods for lossless `RouteError` values. Their
+`DeliveryFailed(destination, cause)` preserves `DeliveryError::Timeout`,
+`InvalidPayload { len, max }`, and `UnsupportedPayload`. The original send
+methods retain their legacy stopped-actor/panic classifications for these cases.
+
+`MailboxSink::deliver_high_priority` remains required. Implementations with one
+lane explicitly forward to `deliver`; managed mailboxes explicitly use their
+separate control lane. No priority behavior changes implicitly.
+
 ## Upgrade Strategy
 
 1. Read [development/format-compatibility.md](../development/format-compatibility.md).

src/api/outbound.rs

@@ -117,6 +117,7 @@ impl MailboxSink for SessionOutboundSink {
     }
 
     fn deliver_high_priority(&self, envelope: Envelope) -> Result<(), DeliveryError> {
+        // Session transport has one lane; choose ordinary delivery explicitly.
         self.deliver(envelope)
     }
 }

src/boot/storage/tests.rs

@@ -138,20 +138,21 @@ fn should_apply_cloud_throughput_defaults_when_memtable_is_auto() {
     )
     .memory_budget(MemoryBudget::Bytes(512 * 1024 * 1024));
     let memory_budget_bytes = 512 * 1024 * 1024usize;
-    let transaction_pool_bytes = memory_budget_bytes / 10;
-    let compaction_pool_bytes = memory_budget_bytes / 10;
-    let expected_memtable_bytes = memory_budget_bytes
-        .saturating_sub(transaction_pool_bytes)
-        .saturating_sub(compaction_pool_bytes)
-        / 2;
 
     // Act
     let tuned = build_midge_open_options(open_options, &config).expect("build cloud options");
 
     // Assert
     assert_eq!(tuned.goal(), Goal::Throughput);
     assert_eq!(tuned.workload(), WorkloadProfile::WriteHeavy);
-    assert_eq!(tuned.memtable_size_limit(), expected_memtable_bytes);
+    assert!(tuned.memtable_size_limit() > 0);
+    assert!(tuned.block_cache_size() > 0);
+    assert!(
+        tuned.transaction_memory_pool_size()
+            + tuned.memtable_size_limit() * 2
+            + tuned.block_cache_size()
+            <= memory_budget_bytes
+    );
     assert_eq!(tuned.wal_buffer_size(), 1024 * 1024);
     assert_eq!(tuned.target_sst_size(), 512 * 1024 * 1024);
 }

src/domains/kv/actor/mod.rs

@@ -69,7 +69,7 @@ impl KvActor {
                 scope,
                 mode,
                 write_options,
-            } => self.handle_begin(scope, mode, write_options),
+            } => self.handle_begin(scope, mode, write_options.into()),
             KvMessage::Commit { tx_id, scope } => self.handle_commit(tx_id, &scope),
             KvMessage::Rollback { tx_id, scope } => self.handle_rollback(tx_id, &scope),
             KvMessage::Get { tx_id, scope, key } => self.handle_get(tx_id, &scope, &key),

src/domains/kv/actor/tests.rs

@@ -19,7 +19,7 @@ pub(super) fn begin_with_scope(actor: &mut KvActor, scope: KvResourceScope) -> u
     let response = actor.handle(KvMessage::Begin {
         scope,
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id } = response else {
         panic!("expected transaction begin, got {response:?}");

src/domains/kv/actor/tests/conflict_and_error_paths.rs

@@ -13,7 +13,7 @@ fn should_handle_concurrent_puts_with_conflict_detection() {
             "concurrent".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id: tx1 } = b1 else {
         panic!("Expected BeginOk");
@@ -27,7 +27,7 @@ fn should_handle_concurrent_puts_with_conflict_detection() {
             "concurrent".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id: tx2 } = b2 else {
         panic!("Expected BeginOk");
@@ -80,7 +80,7 @@ fn should_handle_concurrent_puts_with_conflict_detection() {
             "concurrent".to_string(),
         ),
         mode: TxMode::ReadOnly,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id: tx3 } = b3 else {
         panic!("Begin failed");
@@ -121,7 +121,7 @@ fn should_reject_operations_from_wrong_area() {
             "shared".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id: tx1 } = r1 else {
         panic!("Expected BeginOk");
@@ -135,7 +135,7 @@ fn should_reject_operations_from_wrong_area() {
             "shared".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id: tx2 } = r2 else {
         panic!("Expected BeginOk");
@@ -201,7 +201,7 @@ fn should_return_not_found_when_key_never_written() {
             "table1".to_string(),
         ),
         mode: TxMode::ReadOnly,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id } = begin_response else {
         panic!("Expected BeginOk");
@@ -236,7 +236,7 @@ fn should_delete_nonexistent_key_without_error() {
             "table1".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id } = begin_response else {
         panic!("Expected BeginOk");
@@ -265,7 +265,7 @@ fn should_scan_empty_table_returns_empty_result() {
             "empty_table".to_string(),
         ),
         mode: TxMode::ReadOnly,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id } = begin_response else {
         panic!("Expected BeginOk");
@@ -307,7 +307,7 @@ fn should_reject_begin_with_empty_realm() {
             "table1".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
 
     // Assert
@@ -365,7 +365,7 @@ fn should_reject_begin_with_realm_containing_spaces() {
             "table1".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
 
     // Assert
@@ -389,7 +389,7 @@ fn should_reject_commit_on_already_committed_txid() {
             "table1".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id } = begin_response else {
         panic!("Expected BeginOk");
@@ -426,7 +426,7 @@ fn should_reject_rollback_on_already_rolled_back_txid() {
             "table1".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id } = begin_response else {
         panic!("Expected BeginOk");
@@ -463,7 +463,7 @@ fn should_reject_empty_resource_in_follow_up_scope() {
             "table1".to_string(),
         ),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::buffered(),
+        write_options: cntryl_midge::WriteOptions::buffered().into(),
     });
     let KvResponse::BeginOk { tx_id } = begin_response else {
         panic!("Expected BeginOk");

src/domains/kv/actor/tests/inventory.rs

@@ -54,7 +54,7 @@ fn should_persist_inventory_estimate_after_commit_in_cloud_mode() {
     let KvResponse::BeginOk { tx_id } = actor.handle(KvMessage::Begin {
         scope: scope.clone(),
         mode: TxMode::ReadWrite,
-        write_options: cntryl_midge::WriteOptions::cloud_async(),
+        write_options: cntryl_midge::WriteOptions::cloud_async().into(),
     }) else {
         panic!("transaction should begin");
     };
@@ -103,7 +103,7 @@ fn should_commit_disjoint_writes_without_inventory_conflict() {
         let KvResponse::BeginOk { tx_id } = actor.handle(KvMessage::Begin {
             scope: scope.clone(),
             mode: TxMode::ReadWrite,
-            write_options: cntryl_midge::WriteOptions::buffered(),
+            write_options: cntryl_midge::WriteOptions::buffered().into(),
         }) else {
             panic!("transaction should begin");
         };