Complete SOLID domain ownership remediation

.github/workflows/ci-backend.yml

@@ -36,7 +36,15 @@ jobs:
         run: rustup update stable
 
       - name: Install lld linker
-        run: sudo apt-get update && sudo apt-get install -y lld
+        run: |
+          # The hosted image may have a stale Google Chrome index while its
+          # release metadata is being refreshed. It is unrelated to this job;
+          # remove only that optional source so Ubuntu package installation is
+          # deterministic.
+          sudo grep -RIl 'dl.google.com/linux/chrome' /etc/apt/sources.list.d \
+            | xargs -r sudo rm -f
+          sudo apt-get update
+          sudo apt-get install -y lld
 
       - name: Cache Rust dependencies
         uses: Swatinem/rust-cache@v2

benches/tier2_subsystem_lease.rs

@@ -3,11 +3,11 @@ use bytes::Bytes;
 mod tier2_stress;
 
 use cntryl_stress::{black_box, stress, stress_main, StressContext};
+use fitz::benchkit::BenchDomainHandle;
 use fitz::benchkit::{
     create_bench_lease_sink, register_session_counting_sink, route_frame,
     wait_for_counting_sinks_each_count, CountingSink,
 };
-use fitz::domains::lease::sink::LeaseDomainSink;
 use fitz::protocol::frame::ChannelId;
 use fitz::protocol::lease_codec::msg_type;
 use fitz::protocol::payload_codec::PayloadEncoder;
@@ -24,7 +24,7 @@ const EXACT_ROUTE_64_NOTIFY_REPEAT_COUNT: u64 = 131_072;
 const NOTIFY_CHUNK_SIZE: u64 = 512;
 
 struct PreparedLeaseNotifyCase {
-    sink: Arc<LeaseDomainSink>,
+    sink: Arc<BenchDomainHandle>,
     destination: RouteAddress,
     event: DomainPublishEvent,
     watcher_sinks: Vec<Arc<CountingSink>>,

benches/tier2_subsystem_notice.rs

@@ -4,11 +4,11 @@ use bytes::Bytes;
 mod tier2_stress;
 
 use cntryl_stress::{black_box, stress, stress_main, StressContext};
+use fitz::benchkit::BenchDomainHandle;
 use fitz::benchkit::{
     build_notice_subscribe, create_bench_notice_sink, extract_single_tlv_field,
     register_session_counting_sink, route_frame, wait_for_counting_sinks_each_count, CountingSink,
 };
-use fitz::domains::notice::sink::NoticeDomainSink;
 use fitz::protocol::frame::ChannelId;
 use fitz::runtime::domain_event::DomainPublishEvent;
 use fitz::runtime::envelope::Envelope;
@@ -30,7 +30,7 @@ const PUBLISH_CHUNK_SIZE: u64 = 64;
 const DELIVERY_DRAIN_TIMEOUT: Duration = Duration::from_secs(5);
 
 struct NoticePublishCase {
-    sink: Arc<NoticeDomainSink>,
+    sink: Arc<BenchDomainHandle>,
     destination: RouteAddress,
     event: DomainPublishEvent,
     subscriber_sinks: Vec<Arc<CountingSink>>,

benches/tier2_subsystem_schedule_churn.rs

@@ -7,9 +7,10 @@ mod tier2_stress;
 
 use cntryl_stress::{black_box, stress, stress_main, StressContext};
 use fitz::domains::schedule::protocol::validate_concrete_schedule_route;
-use fitz::domains::schedule::{ScheduleActor, ScheduleMessage, ScheduleResponse};
+use fitz::domains::schedule::{ScheduleMessage, ScheduleResponse};
 use fitz::runtime::routing::RouteFamily;
 use fitz::testkit::create_test_engine_with_cfs;
+use fitz::testkit::domain_internals::schedule::{ScheduleActor, ScheduleStore};
 use std::time::{Duration, Instant};
 
 const DELETE_CHURN_OPERATION_COUNT: u64 = 1024;
@@ -19,8 +20,8 @@ fn create_test_actor() -> ScheduleActor {
     let store = create_test_engine_with_cfs(vec![1, 2, 3, 4, 5]);
     ScheduleActor::new(
         RouteFamily::new(1),
-        store,
-        cntryl_midge::WriteOptions::buffered(),
+        ScheduleStore::new(store),
+        fitz::domains::WritePolicy::Buffered,
     )
 }
 

benches/tier2_subsystem_schedule_create.rs

@@ -5,12 +5,14 @@ mod tier2_stress;
 
 use cntryl_stress::{black_box, stress, stress_main, StressContext};
 use fitz::benchkit::create_bench_store;
-use fitz::domains::schedule::actor::ScheduleActor;
 use fitz::domains::schedule::protocol::{
     validate_concrete_schedule_route, CronSchedule, ScheduleCreateEntry, ScheduleDeliveryMode,
 };
-use fitz::domains::schedule::store::{ScheduleBatchInsert, ScheduleInsert, ScheduleStore};
 use fitz::runtime::routing::RouteFamily;
+use fitz::testkit::domain_internals::schedule::ScheduleActor;
+use fitz::testkit::domain_internals::schedule::{
+    ScheduleBatchInsert, ScheduleInsert, ScheduleStore,
+};
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 const CREATE_BATCH_SIZE: usize = 32;
@@ -128,8 +130,8 @@ fn create_actor_case(fixtures: &ScheduleCreateFixtures) -> ActorCreateCase {
     ActorCreateCase {
         actor: ScheduleActor::new(
             RouteFamily::new(1),
-            create_bench_store(),
-            cntryl_midge::WriteOptions::buffered(),
+            ScheduleStore::new(create_bench_store()),
+            fitz::domains::WritePolicy::Buffered,
         ),
         routes: fixtures.routes[..CREATE_BATCH_SIZE].to_vec(),
         payloads: fixtures.payloads[..CREATE_BATCH_SIZE].to_vec(),
@@ -229,7 +231,7 @@ fn should_store_insert_unique_inmemory_32(ctx: &mut StressContext) {
                                     last_fire_ms: None,
                                     executions_total: 0,
                                 },
-                                cntryl_midge::WriteOptions::buffered(),
+                                fitz::domains::WritePolicy::Buffered,
                             )
                             .expect("schedule insert"),
                     );
@@ -270,7 +272,7 @@ fn should_store_insert_batch_unique_inmemory_32(ctx: &mut StressContext) {
         || {
             for (case, items) in cases.iter().zip(&items_by_case) {
                 case.store
-                    .insert_batch(1, items, cntryl_midge::WriteOptions::buffered())
+                    .insert_batch(1, items, fitz::domains::WritePolicy::Buffered)
                     .expect("schedule insert batch");
                 black_box(());
             }

benches/tier2_subsystem_schedule_fire.rs

@@ -4,17 +4,18 @@ use bytes::Bytes;
 mod tier2_stress;
 
 use cntryl_stress::{black_box, stress, stress_main, StressContext};
+use fitz::benchkit::BenchDomainHandle;
 use fitz::benchkit::{
     create_bench_schedule_sink, create_bench_store_with_cfs, register_session_counting_sink,
     route_frame, wait_for_counting_sinks_each_count, CountingSink,
 };
 use fitz::domains::schedule::protocol::{validate_concrete_schedule_route, Clock};
-use fitz::domains::schedule::sink::ScheduleDomainSink;
-use fitz::domains::schedule::{ScheduleActor, ScheduleMessage, ScheduleResponse};
+use fitz::domains::schedule::{ScheduleMessage, ScheduleResponse};
 use fitz::protocol::frame::ChannelId;
 use fitz::protocol::payload_codec::PayloadEncoder;
 use fitz::runtime::routing::{Route, RouteFamily};
 use fitz::runtime::{DomainPublishEvent, Router};
+use fitz::testkit::domain_internals::schedule::{ScheduleActor, ScheduleStore};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -59,8 +60,8 @@ fn create_test_actor(clock: Arc<dyn Clock>) -> ScheduleActor {
     let store = create_bench_store_with_cfs([1, 2, 3, 4, 5]);
     ScheduleActor::new_with_clock(
         RouteFamily::new(1),
-        store,
-        cntryl_midge::WriteOptions::buffered(),
+        ScheduleStore::new(store),
+        fitz::domains::WritePolicy::Buffered,
         clock,
     )
 }
@@ -160,7 +161,7 @@ fn encode_schedule_subscribe(pattern: &str) -> Bytes {
 fn create_publish_case(
     subscriber_count: usize,
 ) -> (
-    Arc<ScheduleDomainSink>,
+    Arc<BenchDomainHandle>,
     DomainPublishEvent,
     Vec<Arc<CountingSink>>,
 ) {

benches/tier2_subsystem_schedule_scan.rs

@@ -10,8 +10,9 @@ mod tier2_stress;
 use cntryl_stress::{black_box, stress, stress_main, StressContext};
 use fitz::benchkit::create_bench_store_with_cfs;
 use fitz::domains::schedule::protocol::{validate_concrete_schedule_route, Clock};
-use fitz::domains::schedule::{ScheduleActor, ScheduleMessage, ScheduleResponse};
+use fitz::domains::schedule::{ScheduleMessage, ScheduleResponse};
 use fitz::runtime::routing::RouteFamily;
+use fitz::testkit::domain_internals::schedule::{ScheduleActor, ScheduleStore};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -53,8 +54,8 @@ fn create_test_actor(clock: Arc<dyn Clock>) -> ScheduleActor {
     let store = create_bench_store_with_cfs([1, 2, 3, 4, 5]);
     ScheduleActor::new_with_clock(
         RouteFamily::new(1),
-        store,
-        cntryl_midge::WriteOptions::buffered(),
+        ScheduleStore::new(store),
+        fitz::domains::WritePolicy::Buffered,
         clock,
     )
 }

benches/tier2_subsystem_stream.rs

@@ -4,12 +4,12 @@ use bytes::Bytes;
 mod tier2_stress;
 
 use cntryl_stress::{black_box, stress, stress_main, StressContext};
+use fitz::benchkit::BenchDomainHandle;
 use fitz::benchkit::{
     build_stream_subscribe, create_bench_stream_sink, drain_frame_queue_sinks_after_each_count,
     extract_single_tlv_field, register_session_counting_sink, register_session_queue_sink,
     route_frame, wait_for_counting_sinks_each_count, CountingSink, FrameQueueSink,
 };
-use fitz::domains::stream::sink::StreamDomainSink;
 use fitz::protocol::frame::ChannelId;
 use fitz::runtime::domain_event::DomainPublishEvent;
 use fitz::runtime::envelope::Envelope;
@@ -30,7 +30,7 @@ const SUBSCRIBE_DESTINATION: &str = "stream://realm/area/control/append";
 const COMMIT_NOTIFY_ROUTE: &str = "stream://realm/area/orders";
 
 struct PreparedStreamNotifyCase {
-    sink: Arc<StreamDomainSink>,
+    sink: Arc<BenchDomainHandle>,
     destination: RouteAddress,
     event: DomainPublishEvent,
     subscriber_sinks: Vec<Arc<CountingSink>>,

benches/tier3_system_kv.rs

@@ -14,9 +14,10 @@ use stress_config::StressContextExt;
 
 use bytes::Bytes;
 use cntryl_stress::{stress, stress_main, StressContext};
-use fitz::domains::kv::{KvActor, KvMessage, KvResourceScope, KvResponse, TxMode};
+use fitz::domains::kv::{KvMessage, KvResourceScope, KvResponse, TxMode};
 use fitz::runtime::routing::RouteFamily;
 use fitz::testkit::create_test_engine_with_cfs;
+use fitz::testkit::domain_internals::kv::KvActor;
 
 const TRIPLE_FAMILY_PUTS_PER_FAMILY: u64 = 25;
 
@@ -39,7 +40,7 @@ fn begin_transaction(
             resource,
         ),
         mode,
-        write_options: cntryl_midge::WriteOptions::buffered().into(),
+        write_options: fitz::domains::WritePolicy::Buffered,
     });
 
     match response {

benches/tier3_system_lease.rs

@@ -1,7 +1,7 @@
 //! Lease domain tier 3 system benchmarks using live domain sinks.
 //!
 //! Concurrent lease contention and route isolation measurement.
-//! Tests the same `FrameContext` -> `LeaseDomainSink` path used by the live server.
+//! Tests the same `FrameContext` -> `LeaseDomain` path used by the live server.
 //!
 //! Each test measures a single operation with all setup/teardown outside the measurement loop.
 //! Target: ops/sec via `record_completed(count)`

benches/tier3_system_queue.rs

@@ -11,10 +11,11 @@ use fitz::benchkit::{
     extract_single_tlv_field, register_session_counting_sink, register_session_queue_sink,
     route_frame, CountingSink, FrameQueueSink,
 };
-use fitz::domains::queue::{Clock, QueueActor, QueueKey, QueueResponse};
+use fitz::domains::queue::{Clock, QueueKey, QueueResponse};
 use fitz::protocol::frame::ChannelId;
 use fitz::runtime::router::{MailboxSink, Router};
 use fitz::runtime::routing::{RouteAddress, RouteFamily};
+use fitz::testkit::domain_internals::queue::QueueActor;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
@@ -230,14 +231,14 @@ fn should_complete_capacity_mixed_workload(ctx: &mut StressContext) {
         resource: "queue".to_string(),
     };
     let store = create_write_heavy_bench_store();
-    let mut actor = QueueActor::with_clock_and_write_options(
+    let mut actor = QueueActor::with_clock_and_write_policy(
         RouteFamily::new(1),
         queue_key,
         store,
         Box::new(clock.clone()),
         Some(3),
         fitz::utils::idempotency::default_dedup_store(),
-        cntryl_midge::WriteOptions::best_effort(),
+        fitz::domains::WritePolicy::BestEffort,
     );
     let payload = Bytes::from_static(b"mixed workload message");
 

benches/tier3_system_rpc.rs

@@ -1,7 +1,7 @@
 #![allow(deprecated)]
 //! RPC domain tier 3 system benchmarks using the live RPC domain sink.
 //!
-//! Measures the real in-proc path: requester frame -> `RpcDomainSink`
+//! Measures the real in-proc path: requester frame -> `RpcDomain`
 //! -> worker inbox delivery -> worker response frame -> requester inbox.
 
 #[path = "stress_config.rs"]

benches/tier4_kv_support.rs

@@ -9,9 +9,10 @@ use fitz::benchkit::{
     build_kv_begin, build_kv_commit, build_kv_put, create_local_bench_store,
     create_write_heavy_bench_store, parse_kv_response, parse_kv_tx_id, shared_bench_runtime,
 };
-use fitz::domains::kv::{KvActor, KvMessage, KvResourceScope, KvResponse, TxMode};
+use fitz::domains::kv::{KvMessage, KvResourceScope, KvResponse, TxMode};
 use fitz::protocol::kv_codec::parse_request as parse_kv_request;
 use fitz::runtime::routing::RouteFamily;
+use fitz::testkit::domain_internals::kv::KvActor;
 use fitz::testkit::transport::TlvFrameParser;
 use fitz::testkit::{TestClient, TestServer, TestWebSocketClient};
 use futures_util::future::join_all;
@@ -111,7 +112,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().into(),
+        write_options: fitz::domains::WritePolicy::Sync,
     });
     let KvResponse::BeginOk { tx_id } = begin else {
         panic!("KV begin failed: {begin:?}")

benches/tier4_lease_support.rs

@@ -5,12 +5,12 @@ use crate::tier4_support::{
 };
 use bytes::Bytes;
 use cntryl_stress::StressContext;
+use fitz::benchkit::BenchDomainHandle;
 use fitz::benchkit::{
     build_lease_acquire_immediate, build_lease_release, create_bench_lease_sink,
     extract_single_tlv_field, register_session_queue_sink, route_frame_to_address,
     shared_bench_runtime, DirectLeaseAcquireRelease, FrameQueueSink,
 };
-use fitz::domains::lease::sink::LeaseDomainSink;
 use fitz::protocol::frame::ChannelId;
 use fitz::runtime::router::{MailboxSink, Router};
 use fitz::runtime::routing::{Route, RouteAddress, RouteFamily};
@@ -231,7 +231,7 @@ pub(crate) fn measure_direct(ctx: &mut StressContext, measurement: &'static str)
 
 struct EncodedFixture {
     router: Arc<Router>,
-    sink: Arc<LeaseDomainSink>,
+    sink: Arc<BenchDomainHandle>,
     source: RouteAddress,
     destination: RouteAddress,
     inbox: Arc<FrameQueueSink>,

benches/tier4_notice_support.rs

@@ -1,14 +1,14 @@
 #![allow(dead_code)] // Standalone Notice targets use focused subsets of this fixture API.
 
 use bytes::Bytes;
+use fitz::benchkit::BenchDomainHandle;
 use fitz::benchkit::{
     create_bench_notice_sink, parse_notice_delivery, register_session_queue_sink,
     route_frame_to_address, shared_bench_runtime, FrameQueueSink,
 };
 use fitz::domains::notice::protocol::{
     NoticeClientRequest, NotificationMessage, PublishMessage, SubscribeMessage, UnsubscribeMessage,
 };
-use fitz::domains::notice::sink::NoticeDomainSink;
 use fitz::protocol::frame::ChannelId;
 use fitz::protocol::frame_context::FrameContext;
 use fitz::protocol::payload_codec::PayloadDecoder;
@@ -188,7 +188,7 @@ pub(crate) async fn complete_network_control_lifecycle(
 }
 
 pub(crate) struct InProcessNoticeFixture {
-    sink: Arc<NoticeDomainSink>,
+    sink: Arc<BenchDomainHandle>,
     router: Arc<Router>,
     family: RouteFamily,
     destination: RouteAddress,
@@ -308,7 +308,7 @@ impl InProcessNoticeFixture {
 }
 
 pub(crate) struct InProcessNoticeControlFixture {
-    sink: Arc<NoticeDomainSink>,
+    sink: Arc<BenchDomainHandle>,
     router: Arc<Router>,
     family: RouteFamily,
     destination: RouteAddress,