sectorsync_runtime/
lib.rs

1//! Multi-station orchestration helpers for `SectorSync`.
2
3#![forbid(unsafe_code)]
4
5pub mod deployment;
6#[cfg(feature = "parallel")]
7mod parallel;
8
9use std::collections::{BTreeMap, HashMap, HashSet};
10
11use sectorsync_core::prelude::{
12    BarrierId, BarrierScope, BarrierState, CellCoord3, CellIndex, CellLoadSample, CellOccupancy,
13    ClientId, CommandEnvelope, CommandId, CommandIngress, CommandQueueError, CommandQueueMode,
14    CommandQueues, ComponentStore, EntityHandle, EntityId, EventQueueError, EventQueueLimits,
15    EventQueues, GatewayError, GatewaySessionTable, HandoffTransfer, HotspotDecision,
16    HotspotPlanner, HotspotSeverity, HotspotSplitScratch, HotspotThresholds, NodeId, OwnerEpoch,
17    PolicyTable, PushOutcome, ReplicationBudget, ReplicationPlan, ReplicationPlanner,
18    ReplicationScratch, RuntimeBarrier, RuntimeUpgradeHook, SnapshotVersion, SplitProposal,
19    Station, StationError, StationEvent, StationId, StationLoadSample, StationSnapshot, Tick,
20    ViewerQuery, VisibilityFilter,
21};
22use sectorsync_transport::{
23    InboundPacket, OutboundPacket, StationOutboundPacket, StationTransportReceiver,
24    StationTransportSink, TransportReceiver, TransportSink,
25};
26use sectorsync_wire::{
27    BarrierFrame, BinaryDecodeError, BinaryEncodeError, BinaryFrameDecoder, BinaryFrameEncoder,
28    CommandAckFrame, CommandDispatchFrame, CommandFrame, ComponentSelection, FrameDecoder,
29    FrameEncoder, ReplicationFrame, ReplicationFrameBuildStats, ReplicationFrameBuilder,
30    ReplicationFrameRef, ReplicationFrameRefDecodeError, RuntimeFrame, StationEventFrame,
31};
32
33pub use deployment::{
34    DeploymentConfig, DeploymentError, DeploymentNodeRoute, DeploymentNodeState,
35    DeploymentRouteTable, DeploymentStationMove, DeploymentStationRoute, DeploymentStats,
36    GatewayDeliveryError, GatewayDeliveryRoute,
37};
38#[cfg(feature = "parallel")]
39pub use parallel::{
40    ParallelReplicationResult, ParallelReplicationScratch, ParallelReplicationView,
41    ReplicationThreadPool, ReplicationThreadPoolBuildError, ReplicationThreadPoolConfig,
42    StationReplicationBatch,
43};
44
45/// Client replication transport bridge configuration.
46#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
47pub struct ReplicationTransportConfig {
48    /// Planner budget used for every viewer unless the caller builds frames manually.
49    pub budget: ReplicationBudget,
50    /// Whether to send replication frames with no encoded entity deltas.
51    pub send_empty_frames: bool,
52}
53
54/// Client replication transport bridge statistics.
55#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
56pub struct ReplicationTransportStats {
57    /// Viewer queries planned.
58    pub viewers_planned: usize,
59    /// Frames skipped because they had no encoded entity deltas.
60    pub frames_skipped_empty: usize,
61    /// Frames encoded and sent to client transport.
62    pub frames_sent: usize,
63    /// Bytes sent through client transport.
64    pub bytes_sent: usize,
65    /// Initial packet byte capacity requested from bounded wire hints.
66    pub packet_capacity_hint_bytes: usize,
67    /// Largest entity capacity retained by the reusable single-viewer plan.
68    pub planning_entity_capacity_max: usize,
69    /// Entities selected by AOI planning.
70    pub entities_selected: usize,
71    /// Entities skipped by replication planner budget.
72    pub entities_skipped_by_budget: usize,
73    /// Entities skipped by replication planner cadence.
74    pub entities_skipped_by_cadence: usize,
75    /// Entity deltas encoded into replication frames.
76    pub entities_encoded: usize,
77    /// Component deltas encoded into replication frames.
78    pub components_encoded: usize,
79    /// Entities rolled back because the concrete frame byte budget filled.
80    pub entities_skipped_by_frame_bytes: usize,
81}
82
83/// Result of one viewer replication send attempt.
84#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
85pub struct ReplicationTransportReport {
86    /// Target client.
87    pub client_id: ClientId,
88    /// Candidate entities selected by the replication planner.
89    pub selected_entities: usize,
90    /// Entity deltas encoded into the frame.
91    pub encoded_entities: usize,
92    /// Component deltas encoded into the frame.
93    pub encoded_components: usize,
94    /// Estimated bytes from the replication planner.
95    pub estimated_plan_bytes: usize,
96    /// Candidate entities skipped because the planner budget was exhausted.
97    pub skipped_by_budget: usize,
98    /// Candidate entities skipped because cadence had not elapsed.
99    pub skipped_by_cadence: usize,
100    /// Entities rolled back because the concrete frame byte budget filled.
101    pub skipped_by_frame_bytes: usize,
102    /// Encoded wire bytes submitted to transport.
103    pub bytes_sent: usize,
104    /// Whether a frame was sent.
105    pub sent: bool,
106}
107
108/// Error produced while planning, building, encoding, or sending replication.
109#[derive(Clone, Debug, PartialEq, Eq)]
110pub enum ReplicationTransportError<E> {
111    /// Wire encoding failed.
112    Encode(BinaryEncodeError),
113    /// Underlying client transport failed.
114    Transport(E),
115}
116
117impl<E: core::fmt::Display> core::fmt::Display for ReplicationTransportError<E> {
118    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
119        match self {
120            Self::Encode(error) => write!(f, "{error}"),
121            Self::Transport(error) => write!(f, "{error}"),
122        }
123    }
124}
125
126impl<E> std::error::Error for ReplicationTransportError<E>
127where
128    E: std::error::Error + 'static,
129{
130    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
131        match self {
132            Self::Encode(error) => Some(error),
133            Self::Transport(error) => Some(error),
134        }
135    }
136}
137
138impl<E> From<BinaryEncodeError> for ReplicationTransportError<E> {
139    fn from(value: BinaryEncodeError) -> Self {
140        Self::Encode(value)
141    }
142}
143
144/// Bridge between replication planning/frame building and client packet transport.
145#[derive(Clone, Debug)]
146pub struct ReplicationTransportBridge {
147    config: ReplicationTransportConfig,
148    builder: ReplicationFrameBuilder,
149    planning_scratch: Option<ReplicationScratch>,
150    planning_output: ReplicationPlan,
151    stats: ReplicationTransportStats,
152}
153
154impl ReplicationTransportBridge {
155    /// Creates a replication transport bridge.
156    pub const fn new(config: ReplicationTransportConfig, builder: ReplicationFrameBuilder) -> Self {
157        Self {
158            config,
159            builder,
160            planning_scratch: None,
161            planning_output: ReplicationPlan {
162                entities: Vec::new(),
163                stats: sectorsync_core::prelude::ReplicationStats {
164                    candidates: 0,
165                    considered: 0,
166                    selected: 0,
167                    skipped_by_budget: 0,
168                    unexamined_after_budget: 0,
169                    skipped_by_cadence: 0,
170                    estimated_bytes: 0,
171                },
172            },
173            stats: ReplicationTransportStats {
174                viewers_planned: 0,
175                frames_skipped_empty: 0,
176                frames_sent: 0,
177                bytes_sent: 0,
178                packet_capacity_hint_bytes: 0,
179                planning_entity_capacity_max: 0,
180                entities_selected: 0,
181                entities_skipped_by_budget: 0,
182                entities_skipped_by_cadence: 0,
183                entities_encoded: 0,
184                components_encoded: 0,
185                entities_skipped_by_frame_bytes: 0,
186            },
187        }
188    }
189
190    /// Returns configuration.
191    pub const fn config(&self) -> ReplicationTransportConfig {
192        self.config
193    }
194
195    /// Returns frame builder configuration.
196    pub const fn builder(&self) -> ReplicationFrameBuilder {
197        self.builder
198    }
199
200    /// Returns accumulated statistics.
201    pub const fn stats(&self) -> ReplicationTransportStats {
202        self.stats
203    }
204
205    /// Plans, builds, encodes, and sends one viewer replication frame.
206    #[allow(clippy::too_many_arguments)]
207    pub fn send_viewer<T, F>(
208        &mut self,
209        transport: &mut T,
210        station: &Station,
211        index: &CellIndex,
212        policies: &PolicyTable,
213        components: &ComponentStore,
214        selection: &ComponentSelection,
215        viewer: &ViewerQuery,
216        filter: &F,
217    ) -> Result<ReplicationTransportReport, ReplicationTransportError<T::Error>>
218    where
219        T: TransportSink,
220        F: VisibilityFilter,
221    {
222        let mut plan = core::mem::take(&mut self.planning_output);
223        ReplicationPlanner::plan_for_viewer_with_scratch_into(
224            station,
225            index,
226            policies,
227            viewer,
228            filter,
229            self.config.budget,
230            self.planning_scratch.get_or_insert_default(),
231            &mut plan,
232        );
233        self.record_planning_capacity(&plan);
234        let result = self.send_plan(
235            transport,
236            viewer.client_id,
237            station.tick(),
238            station,
239            components,
240            selection,
241            &plan,
242        );
243        self.planning_output = plan;
244        result
245    }
246
247    /// Plans with cadence, builds, encodes, and sends one viewer replication frame.
248    #[allow(clippy::too_many_arguments)]
249    pub fn send_viewer_with_cadence<T, F, L>(
250        &mut self,
251        transport: &mut T,
252        station: &Station,
253        index: &CellIndex,
254        policies: &PolicyTable,
255        components: &ComponentStore,
256        selection: &ComponentSelection,
257        viewer: &ViewerQuery,
258        filter: &F,
259        last_sent: L,
260    ) -> Result<ReplicationTransportReport, ReplicationTransportError<T::Error>>
261    where
262        T: TransportSink,
263        F: VisibilityFilter,
264        L: Fn(EntityHandle) -> Option<Tick>,
265    {
266        let mut plan = core::mem::take(&mut self.planning_output);
267        ReplicationPlanner::plan_for_viewer_with_cadence_and_scratch_into(
268            station,
269            index,
270            policies,
271            viewer,
272            filter,
273            self.config.budget,
274            last_sent,
275            self.planning_scratch.get_or_insert_default(),
276            &mut plan,
277        );
278        self.record_planning_capacity(&plan);
279        let result = self.send_plan(
280            transport,
281            viewer.client_id,
282            station.tick(),
283            station,
284            components,
285            selection,
286            &plan,
287        );
288        self.planning_output = plan;
289        result
290    }
291
292    /// Plans by priority, builds, encodes, and sends one viewer replication frame.
293    #[allow(clippy::too_many_arguments)]
294    pub fn send_viewer_prioritized<T, F>(
295        &mut self,
296        transport: &mut T,
297        station: &Station,
298        index: &CellIndex,
299        policies: &PolicyTable,
300        components: &ComponentStore,
301        selection: &ComponentSelection,
302        viewer: &ViewerQuery,
303        filter: &F,
304    ) -> Result<ReplicationTransportReport, ReplicationTransportError<T::Error>>
305    where
306        T: TransportSink,
307        F: VisibilityFilter,
308    {
309        let mut plan = core::mem::take(&mut self.planning_output);
310        ReplicationPlanner::plan_for_viewer_prioritized_with_scratch_into(
311            station,
312            index,
313            policies,
314            viewer,
315            filter,
316            self.config.budget,
317            self.planning_scratch.get_or_insert_default(),
318            &mut plan,
319        );
320        self.record_planning_capacity(&plan);
321        let result = self.send_plan(
322            transport,
323            viewer.client_id,
324            station.tick(),
325            station,
326            components,
327            selection,
328            &plan,
329        );
330        self.planning_output = plan;
331        result
332    }
333
334    /// Plans by priority and cadence, builds, encodes, and sends one viewer replication frame.
335    #[allow(clippy::too_many_arguments)]
336    pub fn send_viewer_prioritized_with_cadence<T, F, L>(
337        &mut self,
338        transport: &mut T,
339        station: &Station,
340        index: &CellIndex,
341        policies: &PolicyTable,
342        components: &ComponentStore,
343        selection: &ComponentSelection,
344        viewer: &ViewerQuery,
345        filter: &F,
346        last_sent: L,
347    ) -> Result<ReplicationTransportReport, ReplicationTransportError<T::Error>>
348    where
349        T: TransportSink,
350        F: VisibilityFilter,
351        L: Fn(EntityHandle) -> Option<Tick>,
352    {
353        let mut plan = core::mem::take(&mut self.planning_output);
354        ReplicationPlanner::plan_for_viewer_prioritized_with_cadence_and_scratch_into(
355            station,
356            index,
357            policies,
358            viewer,
359            filter,
360            self.config.budget,
361            last_sent,
362            self.planning_scratch.get_or_insert_default(),
363            &mut plan,
364        );
365        self.record_planning_capacity(&plan);
366        let result = self.send_plan(
367            transport,
368            viewer.client_id,
369            station.tick(),
370            station,
371            components,
372            selection,
373            &plan,
374        );
375        self.planning_output = plan;
376        result
377    }
378
379    fn record_planning_capacity(&mut self, plan: &ReplicationPlan) {
380        self.stats.planning_entity_capacity_max = self
381            .stats
382            .planning_entity_capacity_max
383            .max(plan.entities.capacity());
384    }
385
386    /// Builds, encodes, and sends a caller-provided replication plan.
387    #[allow(clippy::too_many_arguments)]
388    pub fn send_plan<T>(
389        &mut self,
390        transport: &mut T,
391        client_id: ClientId,
392        server_tick: Tick,
393        station: &Station,
394        components: &ComponentStore,
395        selection: &ComponentSelection,
396        plan: &ReplicationPlan,
397    ) -> Result<ReplicationTransportReport, ReplicationTransportError<T::Error>>
398    where
399        T: TransportSink,
400    {
401        self.stats.viewers_planned = self.stats.viewers_planned.saturating_add(1);
402        self.stats.entities_selected = self
403            .stats
404            .entities_selected
405            .saturating_add(plan.stats.selected);
406        self.stats.entities_skipped_by_budget = self
407            .stats
408            .entities_skipped_by_budget
409            .saturating_add(plan.stats.skipped_by_budget);
410        self.stats.entities_skipped_by_cadence = self
411            .stats
412            .entities_skipped_by_cadence
413            .saturating_add(plan.stats.skipped_by_cadence);
414
415        let capacity_hint = self
416            .builder
417            .sampled_binary_capacity_hint(station, plan, components, selection)
418            .min(self.config.budget.max_bytes);
419        let mut bytes = Vec::with_capacity(capacity_hint);
420        self.stats.packet_capacity_hint_bytes = self
421            .stats
422            .packet_capacity_hint_bytes
423            .saturating_add(capacity_hint);
424        let build_stats = self.builder.encode_binary_bounded_into(
425            client_id,
426            server_tick,
427            station,
428            plan,
429            components,
430            selection,
431            self.config.budget.max_bytes,
432            &mut bytes,
433        )?;
434        self.stats.entities_encoded = self
435            .stats
436            .entities_encoded
437            .saturating_add(build_stats.encoded_entities);
438        self.stats.components_encoded = self
439            .stats
440            .components_encoded
441            .saturating_add(build_stats.encoded_components);
442        self.stats.entities_skipped_by_frame_bytes = self
443            .stats
444            .entities_skipped_by_frame_bytes
445            .saturating_add(build_stats.skipped_entities_by_frame_bytes);
446
447        if build_stats.encoded_entities == 0 && !self.config.send_empty_frames {
448            self.stats.frames_skipped_empty = self.stats.frames_skipped_empty.saturating_add(1);
449            return Ok(replication_report(client_id, plan, build_stats, 0, false));
450        }
451
452        let byte_len = bytes.len();
453        transport
454            .send(OutboundPacket { client_id, bytes })
455            .map_err(ReplicationTransportError::Transport)?;
456        self.stats.frames_sent = self.stats.frames_sent.saturating_add(1);
457        self.stats.bytes_sent = self.stats.bytes_sent.saturating_add(byte_len);
458
459        Ok(replication_report(
460            client_id,
461            plan,
462            build_stats,
463            byte_len,
464            true,
465        ))
466    }
467}
468
469impl Default for ReplicationTransportBridge {
470    fn default() -> Self {
471        Self::new(
472            ReplicationTransportConfig::default(),
473            ReplicationFrameBuilder::default(),
474        )
475    }
476}
477
478fn replication_report(
479    client_id: ClientId,
480    plan: &ReplicationPlan,
481    build_stats: ReplicationFrameBuildStats,
482    bytes_sent: usize,
483    sent: bool,
484) -> ReplicationTransportReport {
485    ReplicationTransportReport {
486        client_id,
487        selected_entities: plan.stats.selected,
488        encoded_entities: build_stats.encoded_entities,
489        encoded_components: build_stats.encoded_components,
490        estimated_plan_bytes: plan.stats.estimated_bytes,
491        skipped_by_budget: plan.stats.skipped_by_budget,
492        skipped_by_cadence: plan.stats.skipped_by_cadence,
493        skipped_by_frame_bytes: build_stats.skipped_entities_by_frame_bytes,
494        bytes_sent,
495        sent,
496    }
497}
498
499/// Replication receive bridge configuration.
500#[derive(Clone, Copy, Debug, PartialEq, Eq)]
501pub struct ReplicationReceiveConfig {
502    /// Local client id expected inside replication frames.
503    pub client_id: ClientId,
504    /// Expected remote sender identity when the transport can identify it.
505    pub expected_source: Option<ClientId>,
506}
507
508impl ReplicationReceiveConfig {
509    /// Creates receive configuration for a local client.
510    pub const fn new(client_id: ClientId) -> Self {
511        Self {
512            client_id,
513            expected_source: None,
514        }
515    }
516
517    /// Returns a copy that expects packets from `source`.
518    #[must_use]
519    pub const fn with_expected_source(mut self, source: ClientId) -> Self {
520        self.expected_source = Some(source);
521        self
522    }
523}
524
525/// Replication receive bridge statistics.
526#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
527pub struct ReplicationReceiveStats {
528    /// Packets consumed from client transport.
529    pub packets_received: usize,
530    /// Bytes consumed from client transport.
531    pub bytes_received: usize,
532    /// Replication frames decoded and accepted.
533    pub frames_received: usize,
534    /// Packets rejected by wire decoding.
535    pub frames_rejected_decode: usize,
536    /// Packets rejected because they were not replication frames.
537    pub frames_rejected_unexpected: usize,
538    /// Packets rejected because the transport source did not match.
539    pub frames_rejected_source: usize,
540    /// Packets rejected because the frame target did not match this client.
541    pub frames_rejected_target: usize,
542    /// Entity deltas received.
543    pub entities_received: usize,
544    /// Component deltas received.
545    pub components_received: usize,
546}
547
548/// Result of pumping replication packets.
549#[derive(Clone, Debug, Default, PartialEq, Eq)]
550pub struct ReplicationReceivePump {
551    /// Packets consumed from client transport.
552    pub packets_received: usize,
553    /// Bytes consumed from client transport.
554    pub bytes_received: usize,
555    /// Decoded replication frames.
556    pub frames: Vec<ReplicationFrame>,
557}
558
559impl ReplicationReceivePump {
560    /// Returns accepted frame count.
561    pub fn frames_received(&self) -> usize {
562        self.frames.len()
563    }
564
565    /// Returns received entity delta count.
566    pub fn entities_received(&self) -> usize {
567        self.frames.iter().map(|frame| frame.entities.len()).sum()
568    }
569
570    /// Returns received component delta count.
571    pub fn components_received(&self) -> usize {
572        self.frames
573            .iter()
574            .flat_map(|frame| &frame.entities)
575            .map(|entity| entity.components.len())
576            .sum()
577    }
578}
579
580/// Allocation-free summary from visiting replication packets immediately.
581#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
582pub struct ReplicationReceiveVisitReport {
583    /// Packets consumed from client transport.
584    pub packets_received: usize,
585    /// Bytes consumed from client transport.
586    pub bytes_received: usize,
587    /// Replication frames accepted and passed to the visitor.
588    pub frames_received: usize,
589    /// Entity deltas observed across accepted frames.
590    pub entities_received: usize,
591    /// Component deltas observed across accepted frames.
592    pub components_received: usize,
593}
594
595/// Error produced while receiving replication frames.
596#[derive(Clone, Debug, PartialEq, Eq)]
597pub enum ReplicationReceiveError<E> {
598    /// Underlying client transport failed.
599    Transport(E),
600    /// Wire decoding failed.
601    Decode(BinaryDecodeError),
602    /// Packet decoded as a non-replication frame.
603    UnexpectedFrame,
604    /// Packet source did not match expected remote.
605    SourceMismatch {
606        /// Expected source.
607        expected: ClientId,
608        /// Actual source if transport identified one.
609        actual: Option<ClientId>,
610    },
611    /// Replication frame targeted another client.
612    TargetMismatch {
613        /// Expected local client id.
614        expected: ClientId,
615        /// Actual frame target.
616        actual: ClientId,
617    },
618}
619
620impl<E: core::fmt::Display> core::fmt::Display for ReplicationReceiveError<E> {
621    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
622        match self {
623            Self::Transport(error) => write!(f, "{error}"),
624            Self::Decode(error) => write!(f, "{error}"),
625            Self::UnexpectedFrame => f.write_str("client packet was not a replication frame"),
626            Self::SourceMismatch { expected, actual } => write!(
627                f,
628                "replication source mismatch: expected {}, actual {:?}",
629                expected.get(),
630                actual.map(ClientId::get)
631            ),
632            Self::TargetMismatch { expected, actual } => write!(
633                f,
634                "replication target mismatch: expected {}, actual {}",
635                expected.get(),
636                actual.get()
637            ),
638        }
639    }
640}
641
642impl<E> std::error::Error for ReplicationReceiveError<E>
643where
644    E: std::error::Error + 'static,
645{
646    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
647        match self {
648            Self::Transport(error) => Some(error),
649            Self::Decode(error) => Some(error),
650            Self::UnexpectedFrame | Self::SourceMismatch { .. } | Self::TargetMismatch { .. } => {
651                None
652            }
653        }
654    }
655}
656
657impl<E> From<BinaryDecodeError> for ReplicationReceiveError<E> {
658    fn from(value: BinaryDecodeError) -> Self {
659        Self::Decode(value)
660    }
661}
662
663/// Error produced while visiting borrowed replication frames.
664#[derive(Clone, Debug, PartialEq, Eq)]
665pub enum ReplicationReceiveVisitError<T, V> {
666    /// Packet receive, validation, or frame decoding failed.
667    Receive(ReplicationReceiveError<T>),
668    /// The caller-provided frame visitor failed.
669    Visitor(V),
670}
671
672impl<T: core::fmt::Display, V: core::fmt::Display> core::fmt::Display
673    for ReplicationReceiveVisitError<T, V>
674{
675    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
676        match self {
677            Self::Receive(error) => error.fmt(f),
678            Self::Visitor(error) => write!(f, "replication frame visitor failed: {error}"),
679        }
680    }
681}
682
683impl<T, V> std::error::Error for ReplicationReceiveVisitError<T, V>
684where
685    T: std::error::Error + 'static,
686    V: std::error::Error + 'static,
687{
688    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
689        match self {
690            Self::Receive(error) => Some(error),
691            Self::Visitor(error) => Some(error),
692        }
693    }
694}
695
696/// Bridge between client packet transport and decoded replication frames.
697#[derive(Clone, Debug)]
698pub struct ReplicationReceiveBridge {
699    config: ReplicationReceiveConfig,
700    stats: ReplicationReceiveStats,
701}
702
703impl ReplicationReceiveBridge {
704    /// Creates a receive bridge.
705    pub const fn new(config: ReplicationReceiveConfig) -> Self {
706        Self {
707            config,
708            stats: ReplicationReceiveStats {
709                packets_received: 0,
710                bytes_received: 0,
711                frames_received: 0,
712                frames_rejected_decode: 0,
713                frames_rejected_unexpected: 0,
714                frames_rejected_source: 0,
715                frames_rejected_target: 0,
716                entities_received: 0,
717                components_received: 0,
718            },
719        }
720    }
721
722    /// Returns configuration.
723    pub const fn config(&self) -> ReplicationReceiveConfig {
724        self.config
725    }
726
727    /// Returns accumulated statistics.
728    pub const fn stats(&self) -> ReplicationReceiveStats {
729        self.stats
730    }
731
732    /// Receives and decodes up to `max_packets` replication frames.
733    pub fn pump<T>(
734        &mut self,
735        transport: &mut T,
736        max_packets: usize,
737    ) -> Result<ReplicationReceivePump, ReplicationReceiveError<T::Error>>
738    where
739        T: TransportReceiver,
740    {
741        let mut pump = ReplicationReceivePump::default();
742        for _ in 0..max_packets {
743            let Some(packet) = transport
744                .try_recv()
745                .map_err(ReplicationReceiveError::Transport)?
746            else {
747                break;
748            };
749            self.stats.packets_received = self.stats.packets_received.saturating_add(1);
750            self.stats.bytes_received =
751                self.stats.bytes_received.saturating_add(packet.bytes.len());
752            pump.packets_received = pump.packets_received.saturating_add(1);
753            pump.bytes_received = pump.bytes_received.saturating_add(packet.bytes.len());
754
755            if let Some(expected) = self.config.expected_source
756                && packet.client_id != Some(expected)
757            {
758                self.stats.frames_rejected_source =
759                    self.stats.frames_rejected_source.saturating_add(1);
760                return Err(ReplicationReceiveError::SourceMismatch {
761                    expected,
762                    actual: packet.client_id,
763                });
764            }
765
766            let decoded = match BinaryFrameDecoder.decode(&packet.bytes) {
767                Ok(decoded) => decoded,
768                Err(error) => {
769                    self.stats.frames_rejected_decode =
770                        self.stats.frames_rejected_decode.saturating_add(1);
771                    return Err(ReplicationReceiveError::Decode(error));
772                }
773            };
774            let RuntimeFrame::Replication(frame) = decoded else {
775                self.stats.frames_rejected_unexpected =
776                    self.stats.frames_rejected_unexpected.saturating_add(1);
777                return Err(ReplicationReceiveError::UnexpectedFrame);
778            };
779            if frame.client_id != self.config.client_id {
780                self.stats.frames_rejected_target =
781                    self.stats.frames_rejected_target.saturating_add(1);
782                return Err(ReplicationReceiveError::TargetMismatch {
783                    expected: self.config.client_id,
784                    actual: frame.client_id,
785                });
786            }
787
788            self.stats.frames_received = self.stats.frames_received.saturating_add(1);
789            self.stats.entities_received = self
790                .stats
791                .entities_received
792                .saturating_add(frame.entities.len());
793            let components = frame
794                .entities
795                .iter()
796                .map(|entity| entity.components.len())
797                .sum::<usize>();
798            self.stats.components_received =
799                self.stats.components_received.saturating_add(components);
800            pump.frames.push(frame);
801        }
802        Ok(pump)
803    }
804
805    /// Receives validated replication frames and visits borrowed deltas without
806    /// materializing nested owned frame storage.
807    ///
808    /// The visitor must consume borrowed component bytes before returning. Its
809    /// error is surfaced separately from transport and frame-validation errors.
810    /// A frame is counted as accepted after source/wire/target validation and
811    /// before visitor invocation; accumulated bridge statistics are not rolled
812    /// back when the visitor returns an error.
813    pub fn pump_visit<T, F, V>(
814        &mut self,
815        transport: &mut T,
816        max_packets: usize,
817        mut visitor: F,
818    ) -> Result<ReplicationReceiveVisitReport, ReplicationReceiveVisitError<T::Error, V>>
819    where
820        T: TransportReceiver,
821        F: for<'frame> FnMut(ReplicationFrameRef<'frame>) -> Result<(), V>,
822    {
823        let mut report = ReplicationReceiveVisitReport::default();
824        for _ in 0..max_packets {
825            let Some(packet) = transport.try_recv().map_err(|error| {
826                ReplicationReceiveVisitError::Receive(ReplicationReceiveError::Transport(error))
827            })?
828            else {
829                break;
830            };
831            self.stats.packets_received = self.stats.packets_received.saturating_add(1);
832            self.stats.bytes_received =
833                self.stats.bytes_received.saturating_add(packet.bytes.len());
834            report.packets_received = report.packets_received.saturating_add(1);
835            report.bytes_received = report.bytes_received.saturating_add(packet.bytes.len());
836
837            if let Some(expected) = self.config.expected_source
838                && packet.client_id != Some(expected)
839            {
840                self.stats.frames_rejected_source =
841                    self.stats.frames_rejected_source.saturating_add(1);
842                return Err(ReplicationReceiveVisitError::Receive(
843                    ReplicationReceiveError::SourceMismatch {
844                        expected,
845                        actual: packet.client_id,
846                    },
847                ));
848            }
849
850            let frame = match BinaryFrameDecoder.decode_replication_ref(&packet.bytes) {
851                Ok(frame) => frame,
852                Err(ReplicationFrameRefDecodeError::Binary(error)) => {
853                    self.stats.frames_rejected_decode =
854                        self.stats.frames_rejected_decode.saturating_add(1);
855                    return Err(ReplicationReceiveVisitError::Receive(
856                        ReplicationReceiveError::Decode(error),
857                    ));
858                }
859                Err(ReplicationFrameRefDecodeError::UnexpectedFrameKind(_)) => {
860                    self.stats.frames_rejected_unexpected =
861                        self.stats.frames_rejected_unexpected.saturating_add(1);
862                    return Err(ReplicationReceiveVisitError::Receive(
863                        ReplicationReceiveError::UnexpectedFrame,
864                    ));
865                }
866            };
867            if frame.client_id != self.config.client_id {
868                self.stats.frames_rejected_target =
869                    self.stats.frames_rejected_target.saturating_add(1);
870                return Err(ReplicationReceiveVisitError::Receive(
871                    ReplicationReceiveError::TargetMismatch {
872                        expected: self.config.client_id,
873                        actual: frame.client_id,
874                    },
875                ));
876            }
877
878            let entities = frame.encoded_entity_count();
879            let components = frame
880                .entities()
881                .map(sectorsync_wire::EntityDeltaRef::encoded_component_count)
882                .sum::<usize>();
883            self.stats.frames_received = self.stats.frames_received.saturating_add(1);
884            self.stats.entities_received = self.stats.entities_received.saturating_add(entities);
885            self.stats.components_received =
886                self.stats.components_received.saturating_add(components);
887            report.frames_received = report.frames_received.saturating_add(1);
888            report.entities_received = report.entities_received.saturating_add(entities);
889            report.components_received = report.components_received.saturating_add(components);
890            visitor(frame).map_err(ReplicationReceiveVisitError::Visitor)?;
891        }
892        Ok(report)
893    }
894}
895
896/// Low-level client transport bridge configuration.
897#[derive(Clone, Copy, Debug, PartialEq, Eq)]
898pub struct ClientTransportConfig {
899    /// Local client id expected inside client-bound frames.
900    pub client_id: ClientId,
901    /// Remote server/gateway id used as the transport packet target for commands.
902    pub server_id: ClientId,
903    /// Expected remote sender identity when the transport can identify it.
904    pub expected_source: Option<ClientId>,
905}
906
907impl ClientTransportConfig {
908    /// Creates client transport configuration for a server/gateway target.
909    pub const fn new(client_id: ClientId, server_id: ClientId) -> Self {
910        Self {
911            client_id,
912            server_id,
913            expected_source: None,
914        }
915    }
916
917    /// Returns a copy that expects inbound packets from `source`.
918    #[must_use]
919    pub const fn with_expected_source(mut self, source: ClientId) -> Self {
920        self.expected_source = Some(source);
921        self
922    }
923}
924
925/// Low-level client transport bridge statistics.
926#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
927pub struct ClientTransportStats {
928    /// Command frames encoded and submitted to transport.
929    pub commands_sent: usize,
930    /// Command bytes submitted to transport.
931    pub command_bytes_sent: usize,
932    /// Packets consumed from transport.
933    pub packets_received: usize,
934    /// Bytes consumed from transport.
935    pub bytes_received: usize,
936    /// Command ACK frames decoded and accepted.
937    pub command_acks_received: usize,
938    /// Replication frames decoded and accepted.
939    pub replication_frames_received: usize,
940    /// Barrier frames decoded and accepted.
941    pub barrier_frames_received: usize,
942    /// Packets rejected by wire decoding.
943    pub frames_rejected_decode: usize,
944    /// Packets rejected because they were not client-bound frames.
945    pub frames_rejected_unexpected: usize,
946    /// Packets rejected because the transport source did not match.
947    pub frames_rejected_source: usize,
948    /// Packets rejected because the frame target did not match this client.
949    pub frames_rejected_target: usize,
950    /// Entity deltas received in replication frames.
951    pub entities_received: usize,
952    /// Component deltas received in replication frames.
953    pub components_received: usize,
954}
955
956/// Result of sending one command frame.
957#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
958pub struct ClientCommandSendReport {
959    /// Submitted command id.
960    pub command_id: CommandId,
961    /// Encoded command bytes submitted to transport.
962    pub bytes_sent: usize,
963}
964
965/// Client-bound frame categories accepted by `ClientTransportBridge`.
966#[derive(Clone, Copy, Debug, PartialEq, Eq)]
967pub enum ClientInboundFrameKind {
968    /// Command acknowledgement.
969    CommandAck,
970    /// Replication update.
971    Replication,
972    /// Runtime barrier notification.
973    Barrier,
974}
975
976/// Result of pumping client-bound frames.
977#[derive(Clone, Debug, Default, PartialEq, Eq)]
978pub struct ClientTransportPump {
979    /// Packets consumed from transport.
980    pub packets_received: usize,
981    /// Bytes consumed from transport.
982    pub bytes_received: usize,
983    /// Command acknowledgements decoded and accepted.
984    pub command_acks: Vec<CommandAckFrame>,
985    /// Replication frames decoded and accepted.
986    pub replication_frames: Vec<ReplicationFrame>,
987    /// Barrier frames decoded and accepted.
988    pub barriers: Vec<BarrierFrame>,
989}
990
991impl ClientTransportPump {
992    /// Returns received command ACK count.
993    pub fn command_acks_received(&self) -> usize {
994        self.command_acks.len()
995    }
996
997    /// Returns accepted replication frame count.
998    pub fn replication_frames_received(&self) -> usize {
999        self.replication_frames.len()
1000    }
1001
1002    /// Returns accepted barrier frame count.
1003    pub fn barrier_frames_received(&self) -> usize {
1004        self.barriers.len()
1005    }
1006
1007    /// Returns received entity delta count.
1008    pub fn entities_received(&self) -> usize {
1009        self.replication_frames
1010            .iter()
1011            .map(|frame| frame.entities.len())
1012            .sum()
1013    }
1014
1015    /// Returns received component delta count.
1016    pub fn components_received(&self) -> usize {
1017        self.replication_frames
1018            .iter()
1019            .flat_map(|frame| &frame.entities)
1020            .map(|entity| entity.components.len())
1021            .sum()
1022    }
1023}
1024
1025/// Client-bound frame visited without materializing replication delta storage.
1026#[derive(Clone, Debug)]
1027pub enum ClientInboundFrameRef<'a> {
1028    /// Decoded command acknowledgement value.
1029    CommandAck(CommandAckFrame),
1030    /// Validated replication frame borrowing entity/component payload bytes.
1031    Replication(ReplicationFrameRef<'a>),
1032    /// Decoded runtime barrier notification value.
1033    Barrier(BarrierFrame),
1034}
1035
1036/// Fixed-size result of visiting client-bound transport frames.
1037#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
1038pub struct ClientTransportVisitReport {
1039    /// Packets consumed from transport.
1040    pub packets_received: usize,
1041    /// Bytes consumed from transport.
1042    pub bytes_received: usize,
1043    /// Command acknowledgements decoded and accepted.
1044    pub command_acks_received: usize,
1045    /// Replication frames decoded and accepted.
1046    pub replication_frames_received: usize,
1047    /// Barrier frames decoded and accepted.
1048    pub barrier_frames_received: usize,
1049    /// Entity deltas visited in replication frames.
1050    pub entities_received: usize,
1051    /// Component deltas visited in replication frames.
1052    pub components_received: usize,
1053}
1054
1055/// Error produced while visiting client-bound transport frames.
1056#[derive(Clone, Debug, PartialEq, Eq)]
1057pub enum ClientTransportVisitError<T, V> {
1058    /// Packet receive, validation, or frame decoding failed.
1059    Receive(ClientTransportBridgeError<T>),
1060    /// The caller-provided frame visitor failed.
1061    Visitor(V),
1062}
1063
1064impl<T: core::fmt::Display, V: core::fmt::Display> core::fmt::Display
1065    for ClientTransportVisitError<T, V>
1066{
1067    fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1068        match self {
1069            Self::Receive(error) => error.fmt(formatter),
1070            Self::Visitor(error) => write!(formatter, "client frame visitor failed: {error}"),
1071        }
1072    }
1073}
1074
1075impl<T, V> std::error::Error for ClientTransportVisitError<T, V>
1076where
1077    T: std::error::Error + 'static,
1078    V: std::error::Error + 'static,
1079{
1080    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1081        match self {
1082            Self::Receive(error) => Some(error),
1083            Self::Visitor(error) => Some(error),
1084        }
1085    }
1086}
1087
1088/// Error produced by the low-level client transport bridge.
1089#[derive(Clone, Debug, PartialEq, Eq)]
1090pub enum ClientTransportBridgeError<E> {
1091    /// Outbound command used a different client id than the bridge config.
1092    CommandClientMismatch {
1093        /// Expected local client id.
1094        expected: ClientId,
1095        /// Actual command client id.
1096        actual: ClientId,
1097    },
1098    /// Wire encoding failed.
1099    Encode(BinaryEncodeError),
1100    /// Underlying client transport failed.
1101    Transport(E),
1102    /// Wire decoding failed.
1103    Decode(BinaryDecodeError),
1104    /// Packet decoded as a frame that is not client-bound.
1105    UnexpectedFrame,
1106    /// Packet source did not match expected remote.
1107    SourceMismatch {
1108        /// Expected source.
1109        expected: ClientId,
1110        /// Actual source if transport identified one.
1111        actual: Option<ClientId>,
1112    },
1113    /// Client-bound frame targeted another client.
1114    TargetMismatch {
1115        /// Frame category.
1116        kind: ClientInboundFrameKind,
1117        /// Expected local client id.
1118        expected: ClientId,
1119        /// Actual frame target.
1120        actual: ClientId,
1121    },
1122}
1123
1124impl<E: core::fmt::Display> core::fmt::Display for ClientTransportBridgeError<E> {
1125    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1126        match self {
1127            Self::CommandClientMismatch { expected, actual } => write!(
1128                f,
1129                "command client mismatch: expected {}, actual {}",
1130                expected.get(),
1131                actual.get()
1132            ),
1133            Self::Encode(error) => write!(f, "{error}"),
1134            Self::Transport(error) => write!(f, "{error}"),
1135            Self::Decode(error) => write!(f, "{error}"),
1136            Self::UnexpectedFrame => f.write_str("packet was not a client-bound frame"),
1137            Self::SourceMismatch { expected, actual } => write!(
1138                f,
1139                "client packet source mismatch: expected {}, actual {:?}",
1140                expected.get(),
1141                actual.map(ClientId::get)
1142            ),
1143            Self::TargetMismatch {
1144                kind,
1145                expected,
1146                actual,
1147            } => write!(
1148                f,
1149                "client {:?} frame target mismatch: expected {}, actual {}",
1150                kind,
1151                expected.get(),
1152                actual.get()
1153            ),
1154        }
1155    }
1156}
1157
1158impl<E> std::error::Error for ClientTransportBridgeError<E>
1159where
1160    E: std::error::Error + 'static,
1161{
1162    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1163        match self {
1164            Self::Encode(error) => Some(error),
1165            Self::Transport(error) => Some(error),
1166            Self::Decode(error) => Some(error),
1167            Self::CommandClientMismatch { .. }
1168            | Self::UnexpectedFrame
1169            | Self::SourceMismatch { .. }
1170            | Self::TargetMismatch { .. } => None,
1171        }
1172    }
1173}
1174
1175impl<E> From<BinaryEncodeError> for ClientTransportBridgeError<E> {
1176    fn from(value: BinaryEncodeError) -> Self {
1177        Self::Encode(value)
1178    }
1179}
1180
1181impl<E> From<BinaryDecodeError> for ClientTransportBridgeError<E> {
1182    fn from(value: BinaryDecodeError) -> Self {
1183        Self::Decode(value)
1184    }
1185}
1186
1187/// Low-level bridge for client command send and client-bound frame receive.
1188#[derive(Clone, Debug)]
1189pub struct ClientTransportBridge {
1190    config: ClientTransportConfig,
1191    stats: ClientTransportStats,
1192}
1193
1194impl ClientTransportBridge {
1195    /// Creates a client transport bridge.
1196    pub const fn new(config: ClientTransportConfig) -> Self {
1197        Self {
1198            config,
1199            stats: ClientTransportStats {
1200                commands_sent: 0,
1201                command_bytes_sent: 0,
1202                packets_received: 0,
1203                bytes_received: 0,
1204                command_acks_received: 0,
1205                replication_frames_received: 0,
1206                barrier_frames_received: 0,
1207                frames_rejected_decode: 0,
1208                frames_rejected_unexpected: 0,
1209                frames_rejected_source: 0,
1210                frames_rejected_target: 0,
1211                entities_received: 0,
1212                components_received: 0,
1213            },
1214        }
1215    }
1216
1217    /// Returns configuration.
1218    pub const fn config(&self) -> ClientTransportConfig {
1219        self.config
1220    }
1221
1222    /// Returns accumulated statistics.
1223    pub const fn stats(&self) -> ClientTransportStats {
1224        self.stats
1225    }
1226
1227    /// Encodes and sends a client command frame to the configured server id.
1228    pub fn send_command_frame<T>(
1229        &mut self,
1230        transport: &mut T,
1231        frame: &CommandFrame,
1232    ) -> Result<ClientCommandSendReport, ClientTransportBridgeError<T::Error>>
1233    where
1234        T: TransportSink,
1235    {
1236        if frame.client_id != self.config.client_id {
1237            return Err(ClientTransportBridgeError::CommandClientMismatch {
1238                expected: self.config.client_id,
1239                actual: frame.client_id,
1240            });
1241        }
1242
1243        let mut bytes = Vec::new();
1244        BinaryFrameEncoder.encode_command(frame, &mut bytes)?;
1245        let bytes_sent = bytes.len();
1246        transport
1247            .send(OutboundPacket {
1248                client_id: self.config.server_id,
1249                bytes,
1250            })
1251            .map_err(ClientTransportBridgeError::Transport)?;
1252        self.stats.commands_sent = self.stats.commands_sent.saturating_add(1);
1253        self.stats.command_bytes_sent = self.stats.command_bytes_sent.saturating_add(bytes_sent);
1254
1255        Ok(ClientCommandSendReport {
1256            command_id: frame.command_id,
1257            bytes_sent,
1258        })
1259    }
1260
1261    /// Receives and decodes up to `max_packets` client-bound frames.
1262    pub fn pump<T>(
1263        &mut self,
1264        transport: &mut T,
1265        max_packets: usize,
1266    ) -> Result<ClientTransportPump, ClientTransportBridgeError<T::Error>>
1267    where
1268        T: TransportReceiver,
1269    {
1270        let mut pump = ClientTransportPump::default();
1271        for _ in 0..max_packets {
1272            let Some(packet) = transport
1273                .try_recv()
1274                .map_err(ClientTransportBridgeError::Transport)?
1275            else {
1276                break;
1277            };
1278            self.stats.packets_received = self.stats.packets_received.saturating_add(1);
1279            self.stats.bytes_received =
1280                self.stats.bytes_received.saturating_add(packet.bytes.len());
1281            pump.packets_received = pump.packets_received.saturating_add(1);
1282            pump.bytes_received = pump.bytes_received.saturating_add(packet.bytes.len());
1283
1284            if let Some(expected) = self.config.expected_source
1285                && packet.client_id != Some(expected)
1286            {
1287                self.stats.frames_rejected_source =
1288                    self.stats.frames_rejected_source.saturating_add(1);
1289                return Err(ClientTransportBridgeError::SourceMismatch {
1290                    expected,
1291                    actual: packet.client_id,
1292                });
1293            }
1294
1295            let decoded = match BinaryFrameDecoder.decode(&packet.bytes) {
1296                Ok(decoded) => decoded,
1297                Err(error) => {
1298                    self.stats.frames_rejected_decode =
1299                        self.stats.frames_rejected_decode.saturating_add(1);
1300                    return Err(ClientTransportBridgeError::Decode(error));
1301                }
1302            };
1303            match decoded {
1304                RuntimeFrame::CommandAck(frame) => {
1305                    self.validate_client_target(
1306                        ClientInboundFrameKind::CommandAck,
1307                        frame.client_id,
1308                    )?;
1309                    self.stats.command_acks_received =
1310                        self.stats.command_acks_received.saturating_add(1);
1311                    pump.command_acks.push(frame);
1312                }
1313                RuntimeFrame::Replication(frame) => {
1314                    self.validate_client_target(
1315                        ClientInboundFrameKind::Replication,
1316                        frame.client_id,
1317                    )?;
1318                    self.stats.replication_frames_received =
1319                        self.stats.replication_frames_received.saturating_add(1);
1320                    self.stats.entities_received = self
1321                        .stats
1322                        .entities_received
1323                        .saturating_add(frame.entities.len());
1324                    let components = frame
1325                        .entities
1326                        .iter()
1327                        .map(|entity| entity.components.len())
1328                        .sum::<usize>();
1329                    self.stats.components_received =
1330                        self.stats.components_received.saturating_add(components);
1331                    pump.replication_frames.push(frame);
1332                }
1333                RuntimeFrame::Barrier(frame) => {
1334                    self.validate_client_target(ClientInboundFrameKind::Barrier, frame.client_id)?;
1335                    self.stats.barrier_frames_received =
1336                        self.stats.barrier_frames_received.saturating_add(1);
1337                    pump.barriers.push(frame);
1338                }
1339                RuntimeFrame::Command(_)
1340                | RuntimeFrame::CommandDispatch(_)
1341                | RuntimeFrame::StationEvent(_) => {
1342                    self.stats.frames_rejected_unexpected =
1343                        self.stats.frames_rejected_unexpected.saturating_add(1);
1344                    return Err(ClientTransportBridgeError::UnexpectedFrame);
1345                }
1346            }
1347        }
1348        Ok(pump)
1349    }
1350
1351    /// Receives mixed client-bound frames and visits them without materializing
1352    /// nested replication frame storage.
1353    ///
1354    /// ACK and barrier values are copied into the visitor enum. Replication
1355    /// component bytes borrow the current transport packet and must be consumed
1356    /// before the visitor returns. Accepted statistics are recorded before
1357    /// visitor invocation and are not rolled back if the visitor fails.
1358    #[allow(clippy::too_many_lines)]
1359    pub fn pump_visit<T, F, V>(
1360        &mut self,
1361        transport: &mut T,
1362        max_packets: usize,
1363        mut visitor: F,
1364    ) -> Result<ClientTransportVisitReport, ClientTransportVisitError<T::Error, V>>
1365    where
1366        T: TransportReceiver,
1367        F: for<'frame> FnMut(ClientInboundFrameRef<'frame>) -> Result<(), V>,
1368    {
1369        let mut report = ClientTransportVisitReport::default();
1370        for _ in 0..max_packets {
1371            let Some(packet) = transport.try_recv().map_err(|error| {
1372                ClientTransportVisitError::Receive(ClientTransportBridgeError::Transport(error))
1373            })?
1374            else {
1375                break;
1376            };
1377            self.stats.packets_received = self.stats.packets_received.saturating_add(1);
1378            self.stats.bytes_received =
1379                self.stats.bytes_received.saturating_add(packet.bytes.len());
1380            report.packets_received = report.packets_received.saturating_add(1);
1381            report.bytes_received = report.bytes_received.saturating_add(packet.bytes.len());
1382
1383            if let Some(expected) = self.config.expected_source
1384                && packet.client_id != Some(expected)
1385            {
1386                self.stats.frames_rejected_source =
1387                    self.stats.frames_rejected_source.saturating_add(1);
1388                return Err(ClientTransportVisitError::Receive(
1389                    ClientTransportBridgeError::SourceMismatch {
1390                        expected,
1391                        actual: packet.client_id,
1392                    },
1393                ));
1394            }
1395
1396            match BinaryFrameDecoder.decode_replication_ref(&packet.bytes) {
1397                Ok(frame) => {
1398                    self.validate_client_target::<T::Error>(
1399                        ClientInboundFrameKind::Replication,
1400                        frame.client_id,
1401                    )
1402                    .map_err(ClientTransportVisitError::Receive)?;
1403                    let entities = frame.encoded_entity_count();
1404                    let components = frame
1405                        .entities()
1406                        .map(sectorsync_wire::EntityDeltaRef::encoded_component_count)
1407                        .sum::<usize>();
1408                    self.stats.replication_frames_received =
1409                        self.stats.replication_frames_received.saturating_add(1);
1410                    self.stats.entities_received =
1411                        self.stats.entities_received.saturating_add(entities);
1412                    self.stats.components_received =
1413                        self.stats.components_received.saturating_add(components);
1414                    report.replication_frames_received =
1415                        report.replication_frames_received.saturating_add(1);
1416                    report.entities_received = report.entities_received.saturating_add(entities);
1417                    report.components_received =
1418                        report.components_received.saturating_add(components);
1419                    visitor(ClientInboundFrameRef::Replication(frame))
1420                        .map_err(ClientTransportVisitError::Visitor)?;
1421                }
1422                Err(ReplicationFrameRefDecodeError::Binary(error)) => {
1423                    self.stats.frames_rejected_decode =
1424                        self.stats.frames_rejected_decode.saturating_add(1);
1425                    return Err(ClientTransportVisitError::Receive(
1426                        ClientTransportBridgeError::Decode(error),
1427                    ));
1428                }
1429                Err(ReplicationFrameRefDecodeError::UnexpectedFrameKind(_)) => {
1430                    let decoded = BinaryFrameDecoder.decode(&packet.bytes).map_err(|error| {
1431                        self.stats.frames_rejected_decode =
1432                            self.stats.frames_rejected_decode.saturating_add(1);
1433                        ClientTransportVisitError::Receive(ClientTransportBridgeError::Decode(
1434                            error,
1435                        ))
1436                    })?;
1437                    match decoded {
1438                        RuntimeFrame::CommandAck(frame) => {
1439                            self.validate_client_target::<T::Error>(
1440                                ClientInboundFrameKind::CommandAck,
1441                                frame.client_id,
1442                            )
1443                            .map_err(ClientTransportVisitError::Receive)?;
1444                            self.stats.command_acks_received =
1445                                self.stats.command_acks_received.saturating_add(1);
1446                            report.command_acks_received =
1447                                report.command_acks_received.saturating_add(1);
1448                            visitor(ClientInboundFrameRef::CommandAck(frame))
1449                                .map_err(ClientTransportVisitError::Visitor)?;
1450                        }
1451                        RuntimeFrame::Barrier(frame) => {
1452                            self.validate_client_target::<T::Error>(
1453                                ClientInboundFrameKind::Barrier,
1454                                frame.client_id,
1455                            )
1456                            .map_err(ClientTransportVisitError::Receive)?;
1457                            self.stats.barrier_frames_received =
1458                                self.stats.barrier_frames_received.saturating_add(1);
1459                            report.barrier_frames_received =
1460                                report.barrier_frames_received.saturating_add(1);
1461                            visitor(ClientInboundFrameRef::Barrier(frame))
1462                                .map_err(ClientTransportVisitError::Visitor)?;
1463                        }
1464                        RuntimeFrame::Replication(_)
1465                        | RuntimeFrame::Command(_)
1466                        | RuntimeFrame::CommandDispatch(_)
1467                        | RuntimeFrame::StationEvent(_) => {
1468                            self.stats.frames_rejected_unexpected =
1469                                self.stats.frames_rejected_unexpected.saturating_add(1);
1470                            return Err(ClientTransportVisitError::Receive(
1471                                ClientTransportBridgeError::UnexpectedFrame,
1472                            ));
1473                        }
1474                    }
1475                }
1476            }
1477        }
1478        Ok(report)
1479    }
1480
1481    fn validate_client_target<E>(
1482        &mut self,
1483        kind: ClientInboundFrameKind,
1484        actual: ClientId,
1485    ) -> Result<(), ClientTransportBridgeError<E>> {
1486        if actual == self.config.client_id {
1487            return Ok(());
1488        }
1489        self.stats.frames_rejected_target = self.stats.frames_rejected_target.saturating_add(1);
1490        Err(ClientTransportBridgeError::TargetMismatch {
1491            kind,
1492            expected: self.config.client_id,
1493            actual,
1494        })
1495    }
1496}
1497
1498/// Runtime barrier notification transport statistics.
1499#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
1500pub struct BarrierTransportStats {
1501    /// Barrier frames encoded and submitted to client transport.
1502    pub notifications_sent: usize,
1503    /// Client targets submitted to transport.
1504    pub clients_notified: usize,
1505    /// Encoded bytes submitted to transport.
1506    pub bytes_sent: usize,
1507}
1508
1509/// Result of one barrier notification broadcast.
1510#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1511pub struct BarrierTransportReport {
1512    /// Barrier id.
1513    pub barrier_id: BarrierId,
1514    /// Barrier state sent to clients.
1515    pub state: BarrierState,
1516    /// Server tick associated with the notification.
1517    pub server_tick: Tick,
1518    /// Client targets requested by the caller.
1519    pub clients_requested: usize,
1520    /// Client targets successfully submitted to transport.
1521    pub clients_sent: usize,
1522    /// Encoded bytes submitted to transport.
1523    pub bytes_sent: usize,
1524}
1525
1526/// Error produced while encoding or sending barrier notifications.
1527#[derive(Clone, Debug, PartialEq, Eq)]
1528pub enum BarrierTransportError<E> {
1529    /// Wire encoding failed.
1530    Encode(BinaryEncodeError),
1531    /// Underlying client transport failed.
1532    Transport(E),
1533}
1534
1535impl<E: core::fmt::Display> core::fmt::Display for BarrierTransportError<E> {
1536    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1537        match self {
1538            Self::Encode(error) => write!(f, "{error}"),
1539            Self::Transport(error) => write!(f, "{error}"),
1540        }
1541    }
1542}
1543
1544impl<E> std::error::Error for BarrierTransportError<E>
1545where
1546    E: std::error::Error + 'static,
1547{
1548    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1549        match self {
1550            Self::Encode(error) => Some(error),
1551            Self::Transport(error) => Some(error),
1552        }
1553    }
1554}
1555
1556impl<E> From<BinaryEncodeError> for BarrierTransportError<E> {
1557    fn from(value: BinaryEncodeError) -> Self {
1558        Self::Encode(value)
1559    }
1560}
1561
1562/// Low-level bridge for sending runtime barrier notifications to clients.
1563#[derive(Clone, Debug, Default)]
1564pub struct BarrierTransportBridge {
1565    stats: BarrierTransportStats,
1566}
1567
1568impl BarrierTransportBridge {
1569    /// Creates a barrier notification transport bridge.
1570    pub const fn new() -> Self {
1571        Self {
1572            stats: BarrierTransportStats {
1573                notifications_sent: 0,
1574                clients_notified: 0,
1575                bytes_sent: 0,
1576            },
1577        }
1578    }
1579
1580    /// Returns accumulated statistics.
1581    pub const fn stats(&self) -> BarrierTransportStats {
1582        self.stats
1583    }
1584
1585    /// Sends one barrier notification to one client.
1586    pub fn send_state<T>(
1587        &mut self,
1588        transport: &mut T,
1589        client_id: ClientId,
1590        barrier_id: BarrierId,
1591        server_tick: Tick,
1592        state: BarrierState,
1593    ) -> Result<usize, BarrierTransportError<T::Error>>
1594    where
1595        T: TransportSink,
1596    {
1597        let frame = BarrierFrame {
1598            client_id,
1599            barrier_id,
1600            server_tick,
1601            state,
1602        };
1603        let mut bytes = Vec::new();
1604        BinaryFrameEncoder.encode_barrier(&frame, &mut bytes)?;
1605        let bytes_sent = bytes.len();
1606        transport
1607            .send(OutboundPacket { client_id, bytes })
1608            .map_err(BarrierTransportError::Transport)?;
1609        self.stats.notifications_sent = self.stats.notifications_sent.saturating_add(1);
1610        self.stats.clients_notified = self.stats.clients_notified.saturating_add(1);
1611        self.stats.bytes_sent = self.stats.bytes_sent.saturating_add(bytes_sent);
1612        Ok(bytes_sent)
1613    }
1614
1615    /// Sends one runtime barrier notification to one client.
1616    pub fn send_barrier<T>(
1617        &mut self,
1618        transport: &mut T,
1619        client_id: ClientId,
1620        barrier: RuntimeBarrier,
1621    ) -> Result<usize, BarrierTransportError<T::Error>>
1622    where
1623        T: TransportSink,
1624    {
1625        self.send_state(
1626            transport,
1627            client_id,
1628            barrier.id,
1629            barrier.target_tick,
1630            barrier.state,
1631        )
1632    }
1633
1634    /// Broadcasts one barrier state to a bounded caller-provided client list.
1635    pub fn broadcast_state<T, I>(
1636        &mut self,
1637        transport: &mut T,
1638        clients: I,
1639        barrier_id: BarrierId,
1640        server_tick: Tick,
1641        state: BarrierState,
1642    ) -> Result<BarrierTransportReport, BarrierTransportError<T::Error>>
1643    where
1644        T: TransportSink,
1645        I: IntoIterator<Item = ClientId>,
1646    {
1647        let mut report = BarrierTransportReport {
1648            barrier_id,
1649            state,
1650            server_tick,
1651            clients_requested: 0,
1652            clients_sent: 0,
1653            bytes_sent: 0,
1654        };
1655        for client_id in clients {
1656            report.clients_requested = report.clients_requested.saturating_add(1);
1657            let bytes_sent =
1658                self.send_state(transport, client_id, barrier_id, server_tick, state)?;
1659            report.clients_sent = report.clients_sent.saturating_add(1);
1660            report.bytes_sent = report.bytes_sent.saturating_add(bytes_sent);
1661        }
1662        Ok(report)
1663    }
1664
1665    /// Broadcasts one runtime barrier to a bounded caller-provided client list.
1666    pub fn broadcast_barrier<T, I>(
1667        &mut self,
1668        transport: &mut T,
1669        clients: I,
1670        barrier: RuntimeBarrier,
1671    ) -> Result<BarrierTransportReport, BarrierTransportError<T::Error>>
1672    where
1673        T: TransportSink,
1674        I: IntoIterator<Item = ClientId>,
1675    {
1676        self.broadcast_state(
1677            transport,
1678            clients,
1679            barrier.id,
1680            barrier.target_tick,
1681            barrier.state,
1682        )
1683    }
1684}
1685
1686/// Accepted command ACK reason code.
1687pub const GATEWAY_COMMAND_ACK_ACCEPTED: u16 = 0;
1688/// Command was rejected by generic gateway/session state.
1689pub const GATEWAY_COMMAND_ACK_GATEWAY_REJECTED: u16 = 1;
1690/// Command was rejected by gateway rate limiting.
1691pub const GATEWAY_COMMAND_ACK_RATE_LIMITED: u16 = 2;
1692/// Command was rejected as stale or replayed.
1693pub const GATEWAY_COMMAND_ACK_REPLAY_OR_STALE: u16 = 3;
1694/// Command could not be queued because a target station queue was full.
1695pub const GATEWAY_COMMAND_ACK_QUEUE_FULL: u16 = 4;
1696/// Command was rejected by the station barrier ingress policy.
1697pub const GATEWAY_COMMAND_ACK_BARRIER_REJECTED: u16 = 5;
1698/// Command route pointed at a station queue that was not registered.
1699pub const GATEWAY_COMMAND_ACK_MISSING_QUEUE: u16 = 6;
1700/// Command could not be resolved through deployment metadata.
1701pub const GATEWAY_COMMAND_ACK_DEPLOYMENT_REJECTED: u16 = 7;
1702
1703/// Gateway command pipeline configuration.
1704#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1705pub struct GatewayCommandPipelineConfig {
1706    /// Encode negative ACKs for gateway/queue rejections.
1707    pub ack_rejections: bool,
1708}
1709
1710impl Default for GatewayCommandPipelineConfig {
1711    fn default() -> Self {
1712        Self {
1713            ack_rejections: true,
1714        }
1715    }
1716}
1717
1718/// Gateway command pipeline statistics.
1719#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
1720pub struct GatewayCommandPipelineStats {
1721    /// Command frames decoded.
1722    pub command_frames_decoded: usize,
1723    /// Frames rejected by the binary decoder.
1724    pub frames_rejected_decode: usize,
1725    /// Non-command frames rejected by this pipeline.
1726    pub frames_rejected_non_command: usize,
1727    /// Commands admitted by gateway/session metadata.
1728    pub commands_admitted: usize,
1729    /// Commands enqueued into target station queues.
1730    pub commands_enqueued: usize,
1731    /// Commands rejected by gateway/session metadata.
1732    pub commands_rejected_gateway: usize,
1733    /// Commands rejected by station queue or station queue lookup.
1734    pub commands_rejected_queue: usize,
1735    /// Commands resolved to deployment node delivery routes.
1736    pub commands_routed_deployment: usize,
1737    /// Commands rejected by deployment node/station route metadata.
1738    pub commands_rejected_deployment: usize,
1739    /// ACK frames encoded.
1740    pub acks_encoded: usize,
1741}
1742
1743/// Gateway command pipeline error.
1744#[derive(Clone, Debug, PartialEq, Eq)]
1745pub enum GatewayCommandPipelineError {
1746    /// Wire decode failed.
1747    Decode(BinaryDecodeError),
1748    /// Frame decoded correctly but was not a command frame.
1749    NonCommandFrame,
1750    /// Gateway/session metadata rejected the command.
1751    Gateway(GatewayError),
1752    /// Gateway route pointed at a missing station queue.
1753    MissingQueue(StationId),
1754    /// Target station queue rejected the command.
1755    Queue(CommandQueueError),
1756    /// Deployment route metadata rejected command delivery.
1757    Deployment(DeploymentError),
1758    /// ACK encode failed.
1759    Encode(BinaryEncodeError),
1760}
1761
1762impl core::fmt::Display for GatewayCommandPipelineError {
1763    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
1764        match self {
1765            Self::Decode(error) => write!(f, "{error}"),
1766            Self::NonCommandFrame => f.write_str("gateway command pipeline expected command frame"),
1767            Self::Gateway(error) => write!(f, "{error}"),
1768            Self::MissingQueue(station_id) => write!(
1769                f,
1770                "gateway command route target station {} has no queue",
1771                station_id.get()
1772            ),
1773            Self::Queue(error) => write!(f, "{error}"),
1774            Self::Deployment(error) => write!(f, "{error}"),
1775            Self::Encode(error) => write!(f, "{error}"),
1776        }
1777    }
1778}
1779
1780impl std::error::Error for GatewayCommandPipelineError {
1781    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1782        match self {
1783            Self::Decode(error) => Some(error),
1784            Self::Gateway(error) => Some(error),
1785            Self::Queue(error) => Some(error),
1786            Self::Deployment(error) => Some(error),
1787            Self::Encode(error) => Some(error),
1788            Self::NonCommandFrame | Self::MissingQueue(_) => None,
1789        }
1790    }
1791}
1792
1793/// Gateway command pipeline result.
1794#[derive(Clone, Debug, Default, PartialEq, Eq)]
1795pub struct GatewayCommandPipelineReport {
1796    /// Client id when a command frame was decoded.
1797    pub client_id: Option<ClientId>,
1798    /// Command id when a command frame was decoded.
1799    pub command_id: Option<CommandId>,
1800    /// Target station when gateway routing succeeded.
1801    pub station_id: Option<StationId>,
1802    /// Target node when deployment routing succeeded.
1803    pub node_id: Option<NodeId>,
1804    /// Resolved deployment delivery route.
1805    pub delivery: Option<GatewayDeliveryRoute>,
1806    /// Stamped command envelope for external dispatch, when not enqueued locally.
1807    pub command: Option<CommandEnvelope>,
1808    /// Whether the command was queued for station application.
1809    pub accepted: bool,
1810    /// ACK reason code. Zero means accepted.
1811    pub reason_code: u16,
1812    /// Encoded ACK bytes, when an ACK was produced.
1813    pub ack_bytes: Option<Vec<u8>>,
1814    /// Decode, gateway, queue, or encode error detail.
1815    pub error: Option<GatewayCommandPipelineError>,
1816}
1817
1818/// Business-agnostic gateway command frame pipeline.
1819#[derive(Clone, Debug)]
1820pub struct GatewayCommandPipeline {
1821    config: GatewayCommandPipelineConfig,
1822    decoder: BinaryFrameDecoder,
1823    encoder: BinaryFrameEncoder,
1824    stats: GatewayCommandPipelineStats,
1825}
1826
1827impl GatewayCommandPipeline {
1828    /// Creates a pipeline.
1829    pub fn new(config: GatewayCommandPipelineConfig) -> Self {
1830        Self {
1831            config,
1832            decoder: BinaryFrameDecoder,
1833            encoder: BinaryFrameEncoder,
1834            stats: GatewayCommandPipelineStats::default(),
1835        }
1836    }
1837
1838    /// Returns configuration.
1839    pub const fn config(&self) -> GatewayCommandPipelineConfig {
1840        self.config
1841    }
1842
1843    /// Returns statistics.
1844    pub const fn stats(&self) -> GatewayCommandPipelineStats {
1845        self.stats
1846    }
1847
1848    /// Processes one decoded-transport command packet and optionally produces
1849    /// an encoded command ACK.
1850    pub fn process(
1851        &mut self,
1852        gateway: &mut GatewaySessionTable,
1853        station_queues: &mut BTreeMap<StationId, CommandQueues>,
1854        input: &[u8],
1855        now: Tick,
1856        ingress: CommandIngress,
1857    ) -> GatewayCommandPipelineReport {
1858        let command_frame = match self.decode_command_frame(input) {
1859            Ok(command_frame) => command_frame,
1860            Err(error) => {
1861                return GatewayCommandPipelineReport {
1862                    error: Some(error),
1863                    ..GatewayCommandPipelineReport::default()
1864                };
1865            }
1866        };
1867
1868        self.process_command_frame(gateway, station_queues, command_frame, now, ingress)
1869    }
1870
1871    /// Decodes and admits one command packet, then resolves a deployment route
1872    /// for external node dispatch without touching local station queues.
1873    pub fn dispatch(
1874        &mut self,
1875        gateway: &mut GatewaySessionTable,
1876        deployment: &DeploymentRouteTable,
1877        input: &[u8],
1878        now: Tick,
1879    ) -> GatewayCommandPipelineReport {
1880        let command_frame = match self.decode_command_frame(input) {
1881            Ok(command_frame) => command_frame,
1882            Err(error) => {
1883                return GatewayCommandPipelineReport {
1884                    error: Some(error),
1885                    ..GatewayCommandPipelineReport::default()
1886                };
1887            }
1888        };
1889
1890        self.dispatch_command_frame(gateway, deployment, command_frame, now)
1891    }
1892
1893    fn decode_command_frame(
1894        &mut self,
1895        input: &[u8],
1896    ) -> Result<CommandFrame, GatewayCommandPipelineError> {
1897        let frame = match self.decoder.decode(input) {
1898            Ok(frame) => frame,
1899            Err(error) => {
1900                self.stats.frames_rejected_decode =
1901                    self.stats.frames_rejected_decode.saturating_add(1);
1902                return Err(GatewayCommandPipelineError::Decode(error));
1903            }
1904        };
1905
1906        let RuntimeFrame::Command(command_frame) = frame else {
1907            self.stats.frames_rejected_non_command =
1908                self.stats.frames_rejected_non_command.saturating_add(1);
1909            return Err(GatewayCommandPipelineError::NonCommandFrame);
1910        };
1911        self.stats.command_frames_decoded = self.stats.command_frames_decoded.saturating_add(1);
1912
1913        Ok(command_frame)
1914    }
1915
1916    fn process_command_frame(
1917        &mut self,
1918        gateway: &mut GatewaySessionTable,
1919        station_queues: &mut BTreeMap<StationId, CommandQueues>,
1920        command_frame: CommandFrame,
1921        now: Tick,
1922        ingress: CommandIngress,
1923    ) -> GatewayCommandPipelineReport {
1924        let client_id = command_frame.client_id;
1925        let command_id = command_frame.command_id;
1926        let command = command_frame.into_envelope(now);
1927        let admission = match gateway.admit_command(&command) {
1928            Ok(admission) => {
1929                self.stats.commands_admitted = self.stats.commands_admitted.saturating_add(1);
1930                admission
1931            }
1932            Err(error) => {
1933                self.stats.commands_rejected_gateway =
1934                    self.stats.commands_rejected_gateway.saturating_add(1);
1935                return self.rejected_report(
1936                    client_id,
1937                    command_id,
1938                    None,
1939                    now,
1940                    gateway_reject_reason_code(error),
1941                    GatewayCommandPipelineError::Gateway(error),
1942                );
1943            }
1944        };
1945
1946        let station_id = admission.route.station_id;
1947        let Some(queue) = station_queues.get_mut(&station_id) else {
1948            self.stats.commands_rejected_queue =
1949                self.stats.commands_rejected_queue.saturating_add(1);
1950            return self.rejected_report(
1951                client_id,
1952                command_id,
1953                Some(station_id),
1954                now,
1955                GATEWAY_COMMAND_ACK_MISSING_QUEUE,
1956                GatewayCommandPipelineError::MissingQueue(station_id),
1957            );
1958        };
1959
1960        if let Err(error) = queue.push(command, ingress) {
1961            self.stats.commands_rejected_queue =
1962                self.stats.commands_rejected_queue.saturating_add(1);
1963            return self.rejected_report(
1964                client_id,
1965                command_id,
1966                Some(station_id),
1967                now,
1968                queue_reject_reason_code(error),
1969                GatewayCommandPipelineError::Queue(error),
1970            );
1971        }
1972
1973        self.stats.commands_enqueued = self.stats.commands_enqueued.saturating_add(1);
1974        self.accepted_report(client_id, command_id, station_id, now)
1975    }
1976
1977    fn dispatch_command_frame(
1978        &mut self,
1979        gateway: &mut GatewaySessionTable,
1980        deployment: &DeploymentRouteTable,
1981        command_frame: CommandFrame,
1982        now: Tick,
1983    ) -> GatewayCommandPipelineReport {
1984        let client_id = command_frame.client_id;
1985        let command_id = command_frame.command_id;
1986        let command = command_frame.into_envelope(now);
1987        let admission = match gateway.admit_command(&command) {
1988            Ok(admission) => {
1989                self.stats.commands_admitted = self.stats.commands_admitted.saturating_add(1);
1990                admission
1991            }
1992            Err(error) => {
1993                self.stats.commands_rejected_gateway =
1994                    self.stats.commands_rejected_gateway.saturating_add(1);
1995                return self.rejected_report(
1996                    client_id,
1997                    command_id,
1998                    None,
1999                    now,
2000                    gateway_reject_reason_code(error),
2001                    GatewayCommandPipelineError::Gateway(error),
2002                );
2003            }
2004        };
2005
2006        let delivery = match deployment.resolve_gateway_route(admission.route) {
2007            Ok(delivery) => {
2008                self.stats.commands_routed_deployment =
2009                    self.stats.commands_routed_deployment.saturating_add(1);
2010                delivery
2011            }
2012            Err(error) => {
2013                self.stats.commands_rejected_deployment =
2014                    self.stats.commands_rejected_deployment.saturating_add(1);
2015                return self.rejected_report(
2016                    client_id,
2017                    command_id,
2018                    Some(admission.route.station_id),
2019                    now,
2020                    GATEWAY_COMMAND_ACK_DEPLOYMENT_REJECTED,
2021                    GatewayCommandPipelineError::Deployment(error),
2022                );
2023            }
2024        };
2025
2026        self.dispatch_report(command, delivery, now)
2027    }
2028
2029    fn accepted_report(
2030        &mut self,
2031        client_id: ClientId,
2032        command_id: CommandId,
2033        station_id: StationId,
2034        now: Tick,
2035    ) -> GatewayCommandPipelineReport {
2036        let ack = CommandAckFrame {
2037            client_id,
2038            command_id,
2039            server_tick: now,
2040            accepted: true,
2041            reason_code: GATEWAY_COMMAND_ACK_ACCEPTED,
2042        };
2043        match self.encode_ack(&ack) {
2044            Ok(ack_bytes) => GatewayCommandPipelineReport {
2045                client_id: Some(client_id),
2046                command_id: Some(command_id),
2047                station_id: Some(station_id),
2048                node_id: None,
2049                delivery: None,
2050                command: None,
2051                accepted: true,
2052                reason_code: GATEWAY_COMMAND_ACK_ACCEPTED,
2053                ack_bytes: Some(ack_bytes),
2054                error: None,
2055            },
2056            Err(error) => GatewayCommandPipelineReport {
2057                client_id: Some(client_id),
2058                command_id: Some(command_id),
2059                station_id: Some(station_id),
2060                node_id: None,
2061                delivery: None,
2062                command: None,
2063                accepted: false,
2064                reason_code: GATEWAY_COMMAND_ACK_ACCEPTED,
2065                ack_bytes: None,
2066                error: Some(GatewayCommandPipelineError::Encode(error)),
2067            },
2068        }
2069    }
2070
2071    fn dispatch_report(
2072        &mut self,
2073        command: CommandEnvelope,
2074        delivery: GatewayDeliveryRoute,
2075        now: Tick,
2076    ) -> GatewayCommandPipelineReport {
2077        let ack = CommandAckFrame {
2078            client_id: command.client_id,
2079            command_id: command.id,
2080            server_tick: now,
2081            accepted: true,
2082            reason_code: GATEWAY_COMMAND_ACK_ACCEPTED,
2083        };
2084        match self.encode_ack(&ack) {
2085            Ok(ack_bytes) => GatewayCommandPipelineReport {
2086                client_id: Some(command.client_id),
2087                command_id: Some(command.id),
2088                station_id: Some(delivery.station_id),
2089                node_id: Some(delivery.node_id),
2090                delivery: Some(delivery),
2091                command: Some(command),
2092                accepted: true,
2093                reason_code: GATEWAY_COMMAND_ACK_ACCEPTED,
2094                ack_bytes: Some(ack_bytes),
2095                error: None,
2096            },
2097            Err(error) => GatewayCommandPipelineReport {
2098                client_id: Some(command.client_id),
2099                command_id: Some(command.id),
2100                station_id: Some(delivery.station_id),
2101                node_id: Some(delivery.node_id),
2102                delivery: Some(delivery),
2103                command: None,
2104                accepted: false,
2105                reason_code: GATEWAY_COMMAND_ACK_ACCEPTED,
2106                ack_bytes: None,
2107                error: Some(GatewayCommandPipelineError::Encode(error)),
2108            },
2109        }
2110    }
2111
2112    fn rejected_report(
2113        &mut self,
2114        client_id: ClientId,
2115        command_id: CommandId,
2116        station_id: Option<StationId>,
2117        now: Tick,
2118        reason_code: u16,
2119        error: GatewayCommandPipelineError,
2120    ) -> GatewayCommandPipelineReport {
2121        let ack_bytes = if self.config.ack_rejections {
2122            let ack = CommandAckFrame {
2123                client_id,
2124                command_id,
2125                server_tick: now,
2126                accepted: false,
2127                reason_code,
2128            };
2129            match self.encode_ack(&ack) {
2130                Ok(bytes) => Some(bytes),
2131                Err(encode_error) => {
2132                    return GatewayCommandPipelineReport {
2133                        client_id: Some(client_id),
2134                        command_id: Some(command_id),
2135                        station_id,
2136                        node_id: None,
2137                        delivery: None,
2138                        command: None,
2139                        accepted: false,
2140                        reason_code,
2141                        ack_bytes: None,
2142                        error: Some(GatewayCommandPipelineError::Encode(encode_error)),
2143                    };
2144                }
2145            }
2146        } else {
2147            None
2148        };
2149
2150        GatewayCommandPipelineReport {
2151            client_id: Some(client_id),
2152            command_id: Some(command_id),
2153            station_id,
2154            node_id: None,
2155            delivery: None,
2156            command: None,
2157            accepted: false,
2158            reason_code,
2159            ack_bytes,
2160            error: Some(error),
2161        }
2162    }
2163
2164    fn encode_ack(&mut self, ack: &CommandAckFrame) -> Result<Vec<u8>, BinaryEncodeError> {
2165        let mut out = Vec::new();
2166        self.encoder.encode_command_ack(ack, &mut out)?;
2167        self.stats.acks_encoded = self.stats.acks_encoded.saturating_add(1);
2168        Ok(out)
2169    }
2170}
2171
2172impl Default for GatewayCommandPipeline {
2173    fn default() -> Self {
2174        Self::new(GatewayCommandPipelineConfig::default())
2175    }
2176}
2177
2178const fn gateway_reject_reason_code(error: GatewayError) -> u16 {
2179    match error {
2180        GatewayError::ReplayOrStale { .. } => GATEWAY_COMMAND_ACK_REPLAY_OR_STALE,
2181        GatewayError::RateLimited { .. } => GATEWAY_COMMAND_ACK_RATE_LIMITED,
2182        GatewayError::MissingSession(_)
2183        | GatewayError::SessionDisconnected { .. }
2184        | GatewayError::BadGeneration { .. }
2185        | GatewayError::CapacityFull { .. } => GATEWAY_COMMAND_ACK_GATEWAY_REJECTED,
2186    }
2187}
2188
2189const fn queue_reject_reason_code(error: CommandQueueError) -> u16 {
2190    match error {
2191        CommandQueueError::QueueFull(_) => GATEWAY_COMMAND_ACK_QUEUE_FULL,
2192        CommandQueueError::RejectedByBarrier(_) => GATEWAY_COMMAND_ACK_BARRIER_REJECTED,
2193    }
2194}
2195
2196/// Gateway-side client command transport bridge statistics.
2197#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
2198pub struct GatewayClientTransportStats {
2199    /// Client packets consumed from transport.
2200    pub packets_received: usize,
2201    /// Client packet bytes consumed from transport.
2202    pub bytes_received: usize,
2203    /// Command frames decoded from client packets.
2204    pub command_frames_received: usize,
2205    /// Packets rejected because transport source and command client differed.
2206    pub source_mismatches: usize,
2207    /// Commands accepted by the gateway command pipeline.
2208    pub commands_accepted: usize,
2209    /// Commands rejected by the gateway command pipeline.
2210    pub commands_rejected: usize,
2211    /// Command ACK frames submitted to client transport.
2212    pub acks_sent: usize,
2213    /// Command ACK bytes submitted to client transport.
2214    pub ack_bytes_sent: usize,
2215}
2216
2217/// Result of pumping gateway-side client command packets.
2218#[derive(Clone, Debug, Default, PartialEq, Eq)]
2219pub struct GatewayClientTransportPump {
2220    /// Client packets consumed from transport.
2221    pub packets_received: usize,
2222    /// Client packet bytes consumed from transport.
2223    pub bytes_received: usize,
2224    /// Gateway pipeline reports produced from accepted or rejected commands.
2225    pub reports: Vec<GatewayCommandPipelineReport>,
2226    /// Command ACK frames submitted to client transport.
2227    pub acks_sent: usize,
2228    /// Command ACK bytes submitted to client transport.
2229    pub ack_bytes_sent: usize,
2230}
2231
2232impl GatewayClientTransportPump {
2233    /// Returns processed command count.
2234    pub fn commands_processed(&self) -> usize {
2235        self.reports.len()
2236    }
2237
2238    /// Returns accepted command count.
2239    pub fn commands_accepted(&self) -> usize {
2240        self.reports.iter().filter(|report| report.accepted).count()
2241    }
2242
2243    /// Returns rejected command count.
2244    pub fn commands_rejected(&self) -> usize {
2245        self.reports
2246            .iter()
2247            .filter(|report| !report.accepted)
2248            .count()
2249    }
2250}
2251
2252/// Compact result of pumping gateway commands without retaining per-command reports.
2253#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
2254pub struct GatewayClientTransportSummary {
2255    /// Client packets consumed from transport.
2256    pub packets_received: usize,
2257    /// Client packet bytes consumed from transport.
2258    pub bytes_received: usize,
2259    /// Commands accepted by the gateway pipeline.
2260    pub commands_accepted: usize,
2261    /// Commands rejected by the gateway pipeline.
2262    pub commands_rejected: usize,
2263    /// Command ACK frames submitted to client transport.
2264    pub acks_sent: usize,
2265    /// Command ACK bytes submitted to client transport.
2266    pub ack_bytes_sent: usize,
2267}
2268
2269/// Error produced while pumping gateway-side client command packets.
2270#[derive(Clone, Debug, PartialEq, Eq)]
2271pub enum GatewayClientTransportError<E> {
2272    /// Underlying client transport failed while receiving or sending acknowledgements.
2273    Transport(E),
2274    /// Wire decoding failed.
2275    Decode(BinaryDecodeError),
2276    /// Packet decoded as a non-command frame.
2277    NonCommandFrame,
2278    /// Transport source client and command frame client disagreed.
2279    SourceMismatch {
2280        /// Client identified by the transport.
2281        packet_client_id: ClientId,
2282        /// Client encoded inside the command frame.
2283        frame_client_id: ClientId,
2284    },
2285}
2286
2287impl<E: core::fmt::Display> core::fmt::Display for GatewayClientTransportError<E> {
2288    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
2289        match self {
2290            Self::Transport(error) => write!(f, "{error}"),
2291            Self::Decode(error) => write!(f, "{error}"),
2292            Self::NonCommandFrame => f.write_str("gateway client transport expected command frame"),
2293            Self::SourceMismatch {
2294                packet_client_id,
2295                frame_client_id,
2296            } => write!(
2297                f,
2298                "gateway client source mismatch: packet {}, frame {}",
2299                packet_client_id.get(),
2300                frame_client_id.get()
2301            ),
2302        }
2303    }
2304}
2305
2306impl<E> std::error::Error for GatewayClientTransportError<E>
2307where
2308    E: std::error::Error + 'static,
2309{
2310    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
2311        match self {
2312            Self::Transport(error) => Some(error),
2313            Self::Decode(error) => Some(error),
2314            Self::NonCommandFrame | Self::SourceMismatch { .. } => None,
2315        }
2316    }
2317}
2318
2319/// Low-level bridge from client packet transport into the gateway command pipeline.
2320#[derive(Clone, Debug, Default)]
2321pub struct GatewayClientTransportBridge {
2322    stats: GatewayClientTransportStats,
2323}
2324
2325impl GatewayClientTransportBridge {
2326    /// Creates a gateway client transport bridge.
2327    pub const fn new() -> Self {
2328        Self {
2329            stats: GatewayClientTransportStats {
2330                packets_received: 0,
2331                bytes_received: 0,
2332                command_frames_received: 0,
2333                source_mismatches: 0,
2334                commands_accepted: 0,
2335                commands_rejected: 0,
2336                acks_sent: 0,
2337                ack_bytes_sent: 0,
2338            },
2339        }
2340    }
2341
2342    /// Returns accumulated statistics.
2343    pub const fn stats(&self) -> GatewayClientTransportStats {
2344        self.stats
2345    }
2346
2347    /// Pumps up to `max_packets` client command packets into station queues and
2348    /// sends produced ACKs back through the same bounded client transport.
2349    #[allow(clippy::too_many_arguments)]
2350    pub fn pump_ingress<T, E>(
2351        &mut self,
2352        transport: &mut T,
2353        pipeline: &mut GatewayCommandPipeline,
2354        gateway: &mut GatewaySessionTable,
2355        station_queues: &mut BTreeMap<StationId, CommandQueues>,
2356        now: Tick,
2357        ingress: CommandIngress,
2358        max_packets: usize,
2359    ) -> Result<GatewayClientTransportPump, GatewayClientTransportError<E>>
2360    where
2361        T: TransportReceiver<Error = E> + TransportSink<Error = E>,
2362    {
2363        let mut pump = GatewayClientTransportPump::default();
2364        for _ in 0..max_packets {
2365            let Some(packet) = transport
2366                .try_recv()
2367                .map_err(GatewayClientTransportError::Transport)?
2368            else {
2369                break;
2370            };
2371            let (ack_client_id, packet_bytes, report) = self.process_ingress_packet::<E>(
2372                pipeline,
2373                gateway,
2374                station_queues,
2375                &packet,
2376                now,
2377                ingress,
2378            )?;
2379            pump.packets_received = pump.packets_received.saturating_add(1);
2380            pump.bytes_received = pump.bytes_received.saturating_add(packet_bytes);
2381
2382            if let Some(bytes) = &report.ack_bytes {
2383                let ack_len = bytes.len();
2384                transport
2385                    .send(OutboundPacket {
2386                        client_id: ack_client_id,
2387                        bytes: bytes.clone(),
2388                    })
2389                    .map_err(GatewayClientTransportError::Transport)?;
2390                self.stats.acks_sent = self.stats.acks_sent.saturating_add(1);
2391                self.stats.ack_bytes_sent = self.stats.ack_bytes_sent.saturating_add(ack_len);
2392                pump.acks_sent = pump.acks_sent.saturating_add(1);
2393                pump.ack_bytes_sent = pump.ack_bytes_sent.saturating_add(ack_len);
2394            }
2395            pump.reports.push(report);
2396        }
2397        Ok(pump)
2398    }
2399
2400    /// Pumps commands and moves ACK buffers directly into transport without
2401    /// retaining per-command pipeline reports.
2402    ///
2403    /// Use [`Self::pump_ingress`] when the caller needs detailed error reports
2404    /// or encoded ACK bytes after sending.
2405    #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
2406    pub fn pump_ingress_compact<T, E>(
2407        &mut self,
2408        transport: &mut T,
2409        pipeline: &mut GatewayCommandPipeline,
2410        gateway: &mut GatewaySessionTable,
2411        station_queues: &mut BTreeMap<StationId, CommandQueues>,
2412        now: Tick,
2413        ingress: CommandIngress,
2414        max_packets: usize,
2415    ) -> Result<GatewayClientTransportSummary, GatewayClientTransportError<E>>
2416    where
2417        T: TransportReceiver<Error = E> + TransportSink<Error = E>,
2418    {
2419        let mut summary = GatewayClientTransportSummary::default();
2420        for _ in 0..max_packets {
2421            let Some(packet) = transport
2422                .try_recv()
2423                .map_err(GatewayClientTransportError::Transport)?
2424            else {
2425                break;
2426            };
2427            let (ack_client_id, packet_bytes, mut report) = self.process_ingress_packet::<E>(
2428                pipeline,
2429                gateway,
2430                station_queues,
2431                &packet,
2432                now,
2433                ingress,
2434            )?;
2435            summary.packets_received = summary.packets_received.saturating_add(1);
2436            summary.bytes_received = summary.bytes_received.saturating_add(packet_bytes);
2437            if report.accepted {
2438                summary.commands_accepted = summary.commands_accepted.saturating_add(1);
2439            } else {
2440                summary.commands_rejected = summary.commands_rejected.saturating_add(1);
2441            }
2442
2443            if let Some(bytes) = report.ack_bytes.take() {
2444                let ack_len = bytes.len();
2445                transport
2446                    .send(OutboundPacket {
2447                        client_id: ack_client_id,
2448                        bytes,
2449                    })
2450                    .map_err(GatewayClientTransportError::Transport)?;
2451                self.stats.acks_sent = self.stats.acks_sent.saturating_add(1);
2452                self.stats.ack_bytes_sent = self.stats.ack_bytes_sent.saturating_add(ack_len);
2453                summary.acks_sent = summary.acks_sent.saturating_add(1);
2454                summary.ack_bytes_sent = summary.ack_bytes_sent.saturating_add(ack_len);
2455            }
2456        }
2457        Ok(summary)
2458    }
2459
2460    #[allow(clippy::too_many_arguments)]
2461    fn process_ingress_packet<E>(
2462        &mut self,
2463        pipeline: &mut GatewayCommandPipeline,
2464        gateway: &mut GatewaySessionTable,
2465        station_queues: &mut BTreeMap<StationId, CommandQueues>,
2466        packet: &InboundPacket,
2467        now: Tick,
2468        ingress: CommandIngress,
2469    ) -> Result<(ClientId, usize, GatewayCommandPipelineReport), GatewayClientTransportError<E>>
2470    {
2471        let packet_bytes = packet.bytes.len();
2472        self.stats.packets_received = self.stats.packets_received.saturating_add(1);
2473        self.stats.bytes_received = self.stats.bytes_received.saturating_add(packet_bytes);
2474        let command_frame = match pipeline.decode_command_frame(&packet.bytes) {
2475            Ok(command_frame) => command_frame,
2476            Err(GatewayCommandPipelineError::Decode(error)) => {
2477                return Err(GatewayClientTransportError::Decode(error));
2478            }
2479            Err(GatewayCommandPipelineError::NonCommandFrame) => {
2480                return Err(GatewayClientTransportError::NonCommandFrame);
2481            }
2482            Err(error) => {
2483                unreachable!(
2484                    "decode_command_frame only returns decode/non-command errors: {error}"
2485                );
2486            }
2487        };
2488        self.stats.command_frames_received = self.stats.command_frames_received.saturating_add(1);
2489        if let Some(packet_client_id) = packet.client_id
2490            && packet_client_id != command_frame.client_id
2491        {
2492            self.stats.source_mismatches = self.stats.source_mismatches.saturating_add(1);
2493            return Err(GatewayClientTransportError::SourceMismatch {
2494                packet_client_id,
2495                frame_client_id: command_frame.client_id,
2496            });
2497        }
2498        let ack_client_id = command_frame.client_id;
2499        let report =
2500            pipeline.process_command_frame(gateway, station_queues, command_frame, now, ingress);
2501        if report.accepted {
2502            self.stats.commands_accepted = self.stats.commands_accepted.saturating_add(1);
2503        } else {
2504            self.stats.commands_rejected = self.stats.commands_rejected.saturating_add(1);
2505        }
2506        Ok((ack_client_id, packet_bytes, report))
2507    }
2508}
2509
2510// Linear scans win for small registries; larger sets amortize an ID index.
2511const STATION_LOOKUP_INDEX_THRESHOLD: usize = 64;
2512
2513/// Small ordered in-process Station collection for simulations and embedders.
2514#[derive(Clone, Debug, Default)]
2515pub struct StationSet {
2516    stations: Vec<Station>,
2517    positions: HashMap<StationId, usize>,
2518}
2519
2520impl StationSet {
2521    /// Creates an empty collection with capacity for `capacity` Stations.
2522    pub fn with_capacity(capacity: usize) -> Self {
2523        Self {
2524            stations: Vec::with_capacity(capacity),
2525            positions: if capacity >= STATION_LOOKUP_INDEX_THRESHOLD {
2526                HashMap::with_capacity(capacity)
2527            } else {
2528                HashMap::new()
2529            },
2530        }
2531    }
2532
2533    /// Reserves capacity for at least `additional` more Stations and lookup entries.
2534    pub fn reserve(&mut self, additional: usize) {
2535        self.stations.reserve(additional);
2536        if !self.positions.is_empty() {
2537            self.positions.reserve(additional);
2538        } else if self.stations.len().saturating_add(additional) >= STATION_LOOKUP_INDEX_THRESHOLD {
2539            self.positions.reserve(self.stations.len() + additional);
2540        }
2541    }
2542
2543    /// Adds a station to the collection.
2544    pub fn push(&mut self, station: Station) {
2545        let station_id = station.config().station_id;
2546        self.activate_lookup_for(self.stations.len().saturating_add(1));
2547        if !self.positions.is_empty() {
2548            self.positions
2549                .entry(station_id)
2550                .or_insert(self.stations.len());
2551        }
2552        self.stations.push(station);
2553    }
2554
2555    /// Removes and returns the first Station with `station_id`.
2556    ///
2557    /// Remaining Stations retain their iteration order. Lookup storage remains
2558    /// allocated for reuse when the indexed path is active.
2559    pub fn remove(&mut self, station_id: StationId) -> Option<Station> {
2560        let position = self.position(station_id)?;
2561        let station = self.stations.remove(position);
2562        self.rebuild_positions();
2563        Some(station)
2564    }
2565
2566    /// Gets a station by id.
2567    pub fn get(&self, station_id: StationId) -> Option<&Station> {
2568        self.position(station_id)
2569            .and_then(|index| self.stations.get(index))
2570    }
2571
2572    /// Gets a mutable station by id.
2573    pub fn get_mut(&mut self, station_id: StationId) -> Option<&mut Station> {
2574        let index = self.position(station_id)?;
2575        self.stations.get_mut(index)
2576    }
2577
2578    /// Gets two distinct mutable stations by id.
2579    pub fn get_pair_mut(
2580        &mut self,
2581        left_id: StationId,
2582        right_id: StationId,
2583    ) -> Option<(&mut Station, &mut Station)> {
2584        if left_id == right_id {
2585            return None;
2586        }
2587
2588        let left_index = self.position(left_id)?;
2589        let right_index = self.position(right_id)?;
2590
2591        if left_index < right_index {
2592            let (left, right) = self.stations.split_at_mut(right_index);
2593            Some((&mut left[left_index], &mut right[0]))
2594        } else {
2595            let (left, right) = self.stations.split_at_mut(left_index);
2596            Some((&mut right[0], &mut left[right_index]))
2597        }
2598    }
2599
2600    /// Iterates over stations.
2601    pub fn iter(&self) -> impl Iterator<Item = &Station> {
2602        self.stations.iter()
2603    }
2604
2605    /// Iterates mutably over stations.
2606    pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut Station> {
2607        self.stations.iter_mut()
2608    }
2609
2610    /// Returns station ids matching a barrier scope.
2611    pub fn station_ids_in_scope(&self, scope: BarrierScope) -> Vec<StationId> {
2612        self.stations
2613            .iter()
2614            .filter(|station| match scope {
2615                BarrierScope::Instance(instance_id) => station.config().instance_id == instance_id,
2616                BarrierScope::Station(station_id) => station.config().station_id == station_id,
2617            })
2618            .map(|station| station.config().station_id)
2619            .collect()
2620    }
2621
2622    /// Number of stations.
2623    pub fn len(&self) -> usize {
2624        self.stations.len()
2625    }
2626
2627    /// Station slots retained without growing the ordered collection.
2628    pub fn station_capacity(&self) -> usize {
2629        self.stations.capacity()
2630    }
2631
2632    /// Lookup entries retained without rehashing the Station index.
2633    pub fn lookup_capacity(&self) -> usize {
2634        self.positions.capacity()
2635    }
2636
2637    /// Returns whether Station lookup currently uses the indexed path.
2638    pub fn lookup_index_active(&self) -> bool {
2639        !self.positions.is_empty()
2640    }
2641
2642    /// Returns whether no stations are registered.
2643    pub fn is_empty(&self) -> bool {
2644        self.stations.is_empty()
2645    }
2646
2647    fn position(&self, station_id: StationId) -> Option<usize> {
2648        if self.positions.is_empty() {
2649            self.stations
2650                .iter()
2651                .position(|station| station.config().station_id == station_id)
2652        } else {
2653            self.positions.get(&station_id).copied()
2654        }
2655    }
2656
2657    fn activate_lookup_for(&mut self, new_len: usize) {
2658        if new_len < STATION_LOOKUP_INDEX_THRESHOLD || !self.positions.is_empty() {
2659            return;
2660        }
2661        self.positions.reserve(new_len);
2662        for (index, station) in self.stations.iter().enumerate() {
2663            self.positions
2664                .entry(station.config().station_id)
2665                .or_insert(index);
2666        }
2667    }
2668
2669    fn rebuild_positions(&mut self) {
2670        if self.positions.is_empty() {
2671            return;
2672        }
2673        self.positions.clear();
2674        for (index, station) in self.stations.iter().enumerate() {
2675            self.positions
2676                .entry(station.config().station_id)
2677                .or_insert(index);
2678        }
2679    }
2680}
2681
2682/// Station-local spatial indexes keyed by station id.
2683#[derive(Clone, Debug, Default)]
2684pub struct StationIndexSet {
2685    indexes: Vec<(StationId, CellIndex)>,
2686    positions: HashMap<StationId, usize>,
2687}
2688
2689impl StationIndexSet {
2690    /// Creates an empty collection with capacity for `capacity` Station indexes.
2691    pub fn with_capacity(capacity: usize) -> Self {
2692        Self {
2693            indexes: Vec::with_capacity(capacity),
2694            positions: if capacity >= STATION_LOOKUP_INDEX_THRESHOLD {
2695                HashMap::with_capacity(capacity)
2696            } else {
2697                HashMap::new()
2698            },
2699        }
2700    }
2701
2702    /// Reserves capacity for at least `additional` more indexes and lookup entries.
2703    pub fn reserve(&mut self, additional: usize) {
2704        self.indexes.reserve(additional);
2705        if !self.positions.is_empty() {
2706            self.positions.reserve(additional);
2707        } else if self.indexes.len().saturating_add(additional) >= STATION_LOOKUP_INDEX_THRESHOLD {
2708            self.positions.reserve(self.indexes.len() + additional);
2709        }
2710    }
2711
2712    /// Adds or replaces one station index.
2713    pub fn insert(&mut self, station_id: StationId, index: CellIndex) {
2714        if let Some(position) = self.position(station_id) {
2715            self.indexes[position].1 = index;
2716        } else {
2717            self.activate_lookup_for(self.indexes.len().saturating_add(1));
2718            if !self.positions.is_empty() {
2719                self.positions.insert(station_id, self.indexes.len());
2720            }
2721            self.indexes.push((station_id, index));
2722        }
2723    }
2724
2725    /// Removes and returns one Station-local spatial index.
2726    ///
2727    /// Remaining indexes retain their iteration order and indexed lookup
2728    /// storage remains allocated for later registrations.
2729    pub fn remove(&mut self, station_id: StationId) -> Option<CellIndex> {
2730        let position = self.position(station_id)?;
2731        let (_, index) = self.indexes.remove(position);
2732        self.rebuild_positions();
2733        Some(index)
2734    }
2735
2736    /// Gets one station index.
2737    pub fn get(&self, station_id: StationId) -> Option<&CellIndex> {
2738        self.position(station_id)
2739            .and_then(|position| self.indexes.get(position))
2740            .map(|(_, index)| index)
2741    }
2742
2743    /// Gets one mutable station index.
2744    pub fn get_mut(&mut self, station_id: StationId) -> Option<&mut CellIndex> {
2745        let position = self.position(station_id)?;
2746        self.indexes.get_mut(position).map(|(_, index)| index)
2747    }
2748
2749    /// Gets two distinct mutable station indexes.
2750    pub fn get_pair_mut(
2751        &mut self,
2752        left_id: StationId,
2753        right_id: StationId,
2754    ) -> Option<(&mut CellIndex, &mut CellIndex)> {
2755        if left_id == right_id {
2756            return None;
2757        }
2758
2759        let left_index = self.position(left_id)?;
2760        let right_index = self.position(right_id)?;
2761
2762        if left_index < right_index {
2763            let (left, right) = self.indexes.split_at_mut(right_index);
2764            Some((&mut left[left_index].1, &mut right[0].1))
2765        } else {
2766            let (left, right) = self.indexes.split_at_mut(left_index);
2767            Some((&mut right[0].1, &mut left[right_index].1))
2768        }
2769    }
2770
2771    /// Number of indexes.
2772    pub fn len(&self) -> usize {
2773        self.indexes.len()
2774    }
2775
2776    /// Index slots retained without growing the ordered collection.
2777    pub fn index_capacity(&self) -> usize {
2778        self.indexes.capacity()
2779    }
2780
2781    /// Lookup entries retained without rehashing the Station index.
2782    pub fn lookup_capacity(&self) -> usize {
2783        self.positions.capacity()
2784    }
2785
2786    /// Returns whether Station-index lookup currently uses the indexed path.
2787    pub fn lookup_index_active(&self) -> bool {
2788        !self.positions.is_empty()
2789    }
2790
2791    /// Iterates over registered indexes.
2792    pub fn iter(&self) -> impl Iterator<Item = (StationId, &CellIndex)> {
2793        self.indexes
2794            .iter()
2795            .map(|(station_id, index)| (*station_id, index))
2796    }
2797
2798    /// Returns whether no indexes are registered.
2799    pub fn is_empty(&self) -> bool {
2800        self.indexes.is_empty()
2801    }
2802
2803    fn position(&self, station_id: StationId) -> Option<usize> {
2804        if self.positions.is_empty() {
2805            self.indexes.iter().position(|(id, _)| *id == station_id)
2806        } else {
2807            self.positions.get(&station_id).copied()
2808        }
2809    }
2810
2811    fn activate_lookup_for(&mut self, new_len: usize) {
2812        if new_len < STATION_LOOKUP_INDEX_THRESHOLD || !self.positions.is_empty() {
2813            return;
2814        }
2815        self.positions.reserve(new_len);
2816        for (index, (station_id, _)) in self.indexes.iter().enumerate() {
2817            self.positions.entry(*station_id).or_insert(index);
2818        }
2819    }
2820
2821    fn rebuild_positions(&mut self) {
2822        if self.positions.is_empty() {
2823            return;
2824        }
2825        self.positions.clear();
2826        for (index, (station_id, _)) in self.indexes.iter().enumerate() {
2827            self.positions.entry(*station_id).or_insert(index);
2828        }
2829    }
2830}
2831
2832/// Lightweight coefficients used to derive hotspot/scheduler load samples.
2833#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2834pub struct StationLoadSamplerConfig {
2835    /// Estimated bytes contributed by one stored entity in the measurement window.
2836    pub estimated_bytes_per_entity: usize,
2837    /// Estimated bytes contributed by one subscriber routed to a station.
2838    pub estimated_bytes_per_subscriber: usize,
2839    /// Estimated bytes contributed by one queued station event.
2840    pub estimated_bytes_per_event: usize,
2841    /// Runtime cost units assigned to one authoritative entity.
2842    pub tick_cost_per_owned_entity: u64,
2843    /// Runtime cost units assigned to one read-only ghost entity.
2844    pub tick_cost_per_ghost_entity: u64,
2845    /// Runtime cost units assigned to one occupied spatial cell.
2846    pub tick_cost_per_occupied_cell: u64,
2847    /// Runtime cost units assigned to one queued station event.
2848    pub tick_cost_per_queued_event: u64,
2849}
2850
2851impl Default for StationLoadSamplerConfig {
2852    fn default() -> Self {
2853        Self {
2854            estimated_bytes_per_entity: 48,
2855            estimated_bytes_per_subscriber: 16,
2856            estimated_bytes_per_event: 32,
2857            tick_cost_per_owned_entity: 2,
2858            tick_cost_per_ghost_entity: 1,
2859            tick_cost_per_occupied_cell: 1,
2860            tick_cost_per_queued_event: 1,
2861        }
2862    }
2863}
2864
2865/// Runtime helper that derives `StationLoadSample` from existing low-level state.
2866#[derive(Clone, Copy, Debug, PartialEq, Eq)]
2867pub struct StationLoadSampler {
2868    config: StationLoadSamplerConfig,
2869}
2870
2871/// Caller-owned reusable storage for periodic station load sampling.
2872#[derive(Clone, Debug, Default)]
2873pub struct StationLoadSamplerScratch {
2874    subscribers_by_station: HashMap<StationId, usize>,
2875    occupancy: Vec<CellOccupancy>,
2876    samples: Vec<StationLoadSample>,
2877}
2878
2879impl StationLoadSamplerScratch {
2880    /// Creates empty sampling storage.
2881    pub fn new() -> Self {
2882        Self::default()
2883    }
2884
2885    /// Hash-table capacity retained for subscriber aggregation.
2886    pub fn retained_subscriber_capacity(&self) -> usize {
2887        self.subscribers_by_station.capacity()
2888    }
2889
2890    /// Temporary occupancy capacity retained across station scans.
2891    pub fn retained_occupancy_capacity(&self) -> usize {
2892        self.occupancy.capacity()
2893    }
2894
2895    /// Station output slots retained across sampling passes.
2896    pub fn retained_sample_slots(&self) -> usize {
2897        self.samples.len()
2898    }
2899
2900    /// Total cell output capacity retained across station slots.
2901    pub fn retained_cell_capacity(&self) -> usize {
2902        self.samples
2903            .iter()
2904            .map(|sample| sample.cells.capacity())
2905            .sum()
2906    }
2907}
2908
2909impl StationLoadSampler {
2910    /// Creates a station load sampler.
2911    pub const fn new(config: StationLoadSamplerConfig) -> Self {
2912        Self { config }
2913    }
2914
2915    /// Returns the sampler configuration.
2916    pub const fn config(&self) -> StationLoadSamplerConfig {
2917        self.config
2918    }
2919
2920    /// Samples one station from its storage, optional index, queued event count,
2921    /// and caller-provided subscriber estimate.
2922    pub fn sample_station(
2923        &self,
2924        station: &Station,
2925        index: Option<&CellIndex>,
2926        queued_events: usize,
2927        subscribers: usize,
2928    ) -> StationLoadSample {
2929        let mut sample = StationLoadSample::default();
2930        let mut occupancy = Vec::new();
2931        self.sample_station_into(
2932            station,
2933            index,
2934            queued_events,
2935            subscribers,
2936            &mut occupancy,
2937            &mut sample,
2938        );
2939        sample
2940    }
2941
2942    /// Samples every station in deterministic station-set order.
2943    ///
2944    /// `subscriber_counts` is explicit integration input: `SectorSync` can use it
2945    /// for load decisions, but does not own gateway/client/session business state.
2946    /// Counts with the same station id are aggregated with saturating arithmetic.
2947    /// Because the event router and subscriber input are station-scoped, their
2948    /// pressure stays on the station sample instead of being invented per cell.
2949    pub fn sample_all(
2950        &self,
2951        stations: &StationSet,
2952        indexes: &StationIndexSet,
2953        router: &EventRouter,
2954        subscriber_counts: &[(StationId, usize)],
2955    ) -> Vec<StationLoadSample> {
2956        let subscribers_by_station = station_count_map(subscriber_counts);
2957        stations
2958            .iter()
2959            .map(|station| {
2960                let station_id = station.config().station_id;
2961                self.sample_station(
2962                    station,
2963                    indexes.get(station_id),
2964                    router.queued_len(station_id).unwrap_or(0),
2965                    subscribers_by_station
2966                        .get(&station_id)
2967                        .copied()
2968                        .unwrap_or(0),
2969                )
2970            })
2971            .collect()
2972    }
2973
2974    /// Samples every station into caller-owned storage and returns its active prefix.
2975    ///
2976    /// Station order and aggregation semantics match [`Self::sample_all`]. Reusing
2977    /// the same scratch retains subscriber, occupancy, station, and per-cell storage.
2978    pub fn sample_all_into<'a>(
2979        &self,
2980        stations: &StationSet,
2981        indexes: &StationIndexSet,
2982        router: &EventRouter,
2983        subscriber_counts: &[(StationId, usize)],
2984        scratch: &'a mut StationLoadSamplerScratch,
2985    ) -> &'a [StationLoadSample] {
2986        scratch.subscribers_by_station.clear();
2987        for (station_id, count) in subscriber_counts {
2988            let entry = scratch
2989                .subscribers_by_station
2990                .entry(*station_id)
2991                .or_insert(0);
2992            *entry = entry.saturating_add(*count);
2993        }
2994
2995        let station_count = stations.len();
2996        if scratch.samples.len() < station_count {
2997            scratch
2998                .samples
2999                .resize_with(station_count, StationLoadSample::default);
3000        }
3001        for (station, sample) in stations
3002            .iter()
3003            .zip(scratch.samples[..station_count].iter_mut())
3004        {
3005            let station_id = station.config().station_id;
3006            self.sample_station_into(
3007                station,
3008                indexes.get(station_id),
3009                router.queued_len(station_id).unwrap_or(0),
3010                scratch
3011                    .subscribers_by_station
3012                    .get(&station_id)
3013                    .copied()
3014                    .unwrap_or(0),
3015                &mut scratch.occupancy,
3016                sample,
3017            );
3018        }
3019        &scratch.samples[..station_count]
3020    }
3021
3022    fn sample_station_into(
3023        &self,
3024        station: &Station,
3025        index: Option<&CellIndex>,
3026        queued_events: usize,
3027        subscribers: usize,
3028        occupancy: &mut Vec<CellOccupancy>,
3029        sample: &mut StationLoadSample,
3030    ) {
3031        let (owned_entities, ghost_entities) = count_station_roles(station);
3032        sample.cells.clear();
3033        if let Some(index) = index {
3034            index.cell_occupancy_into(occupancy);
3035            for occupancy in occupancy.iter() {
3036                let mut cell_owned_entities = 0usize;
3037                let mut cell_ghost_entities = 0usize;
3038                for handle in index.handles_in_cell_slice(occupancy.cell) {
3039                    if let Some(record) = station.get(*handle) {
3040                        if record.is_owned() {
3041                            cell_owned_entities = cell_owned_entities.saturating_add(1);
3042                        } else {
3043                            cell_ghost_entities = cell_ghost_entities.saturating_add(1);
3044                        }
3045                    }
3046                }
3047                let entities = cell_owned_entities.saturating_add(cell_ghost_entities);
3048                sample.cells.push(CellLoadSample {
3049                    cell: occupancy.cell,
3050                    owned_entities: cell_owned_entities,
3051                    ghost_entities: cell_ghost_entities,
3052                    subscribers: 0,
3053                    estimated_updates: entities,
3054                    estimated_bytes: entities
3055                        .saturating_mul(self.config.estimated_bytes_per_entity),
3056                    tick_cost_units: self.estimate_tick_cost(
3057                        cell_owned_entities,
3058                        cell_ghost_entities,
3059                        1,
3060                        0,
3061                    ),
3062                    event_pressure: 0,
3063                });
3064            }
3065        } else {
3066            occupancy.clear();
3067        }
3068        sample.station_id = station.config().station_id;
3069        sample.owned_entities = owned_entities;
3070        sample.ghost_entities = ghost_entities;
3071        sample.subscribers = subscribers;
3072        sample.queued_events = queued_events;
3073        sample.estimated_bytes =
3074            self.estimate_station_bytes(owned_entities, ghost_entities, subscribers, queued_events);
3075        sample.tick_cost_units = self.estimate_tick_cost(
3076            owned_entities,
3077            ghost_entities,
3078            sample.cells.len(),
3079            queued_events,
3080        );
3081    }
3082
3083    fn estimate_station_bytes(
3084        &self,
3085        owned_entities: usize,
3086        ghost_entities: usize,
3087        subscribers: usize,
3088        queued_events: usize,
3089    ) -> usize {
3090        owned_entities
3091            .saturating_add(ghost_entities)
3092            .saturating_mul(self.config.estimated_bytes_per_entity)
3093            .saturating_add(subscribers.saturating_mul(self.config.estimated_bytes_per_subscriber))
3094            .saturating_add(queued_events.saturating_mul(self.config.estimated_bytes_per_event))
3095    }
3096
3097    fn estimate_tick_cost(
3098        &self,
3099        owned_entities: usize,
3100        ghost_entities: usize,
3101        occupied_cells: usize,
3102        queued_events: usize,
3103    ) -> u64 {
3104        (owned_entities as u64)
3105            .saturating_mul(self.config.tick_cost_per_owned_entity)
3106            .saturating_add(
3107                (ghost_entities as u64).saturating_mul(self.config.tick_cost_per_ghost_entity),
3108            )
3109            .saturating_add(
3110                (occupied_cells as u64).saturating_mul(self.config.tick_cost_per_occupied_cell),
3111            )
3112            .saturating_add(
3113                (queued_events as u64).saturating_mul(self.config.tick_cost_per_queued_event),
3114            )
3115    }
3116}
3117
3118impl Default for StationLoadSampler {
3119    fn default() -> Self {
3120        Self::new(StationLoadSamplerConfig::default())
3121    }
3122}
3123
3124fn count_station_roles(station: &Station) -> (usize, usize) {
3125    let mut owned_entities = 0usize;
3126    let mut ghost_entities = 0usize;
3127    for record in station.iter() {
3128        if record.is_owned() {
3129            owned_entities = owned_entities.saturating_add(1);
3130        } else {
3131            ghost_entities = ghost_entities.saturating_add(1);
3132        }
3133    }
3134    (owned_entities, ghost_entities)
3135}
3136
3137fn station_count_map(counts: &[(StationId, usize)]) -> BTreeMap<StationId, usize> {
3138    let mut map = BTreeMap::new();
3139    for (station_id, count) in counts {
3140        let entry = map.entry(*station_id).or_insert(0usize);
3141        *entry = entry.saturating_add(*count);
3142    }
3143    map
3144}
3145
3146/// Result of an in-process entity owner migration.
3147#[derive(Clone, Debug, PartialEq)]
3148pub struct EntityMigrationReport {
3149    /// Transfer payload used for the migration.
3150    pub transfer: HandoffTransfer,
3151    /// Source-side ghost handle after commit.
3152    pub source_ghost: EntityHandle,
3153    /// Target-side authoritative handle after commit.
3154    pub target_owner: EntityHandle,
3155}
3156
3157/// Entity migration error.
3158#[derive(Clone, Copy, Debug, PartialEq, Eq)]
3159pub enum EntityMigrationError {
3160    /// Source and target station ids must differ.
3161    SameSourceAndTarget(StationId),
3162    /// Source station was not found.
3163    MissingSource(StationId),
3164    /// Target station was not found.
3165    MissingTarget(StationId),
3166    /// Station-level operation failed.
3167    Station(StationError),
3168}
3169
3170impl core::fmt::Display for EntityMigrationError {
3171    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
3172        match self {
3173            Self::SameSourceAndTarget(id) => {
3174                write!(f, "source and target station are both {}", id.get())
3175            }
3176            Self::MissingSource(id) => write!(f, "source station {} is missing", id.get()),
3177            Self::MissingTarget(id) => write!(f, "target station {} is missing", id.get()),
3178            Self::Station(error) => write!(f, "{error}"),
3179        }
3180    }
3181}
3182
3183impl std::error::Error for EntityMigrationError {}
3184
3185impl From<StationError> for EntityMigrationError {
3186    fn from(value: StationError) -> Self {
3187        Self::Station(value)
3188    }
3189}
3190
3191/// Runtime helper for in-process station-to-station owner migration.
3192#[derive(Clone, Copy, Debug, Default)]
3193pub struct EntityMigrationExecutor;
3194
3195impl EntityMigrationExecutor {
3196    /// Migrates one authoritative entity from source station to target station.
3197    pub fn migrate_entity(
3198        stations: &mut StationSet,
3199        entity_id: EntityId,
3200        source_station: StationId,
3201        target_station: StationId,
3202        ghost_ttl_ticks: u64,
3203    ) -> Result<EntityMigrationReport, EntityMigrationError> {
3204        if source_station == target_station {
3205            return Err(EntityMigrationError::SameSourceAndTarget(source_station));
3206        }
3207
3208        if stations.get(source_station).is_none() {
3209            return Err(EntityMigrationError::MissingSource(source_station));
3210        }
3211        if stations.get(target_station).is_none() {
3212            return Err(EntityMigrationError::MissingTarget(target_station));
3213        }
3214
3215        let (source, target) = stations
3216            .get_pair_mut(source_station, target_station)
3217            .expect("stations were checked above");
3218        let target_epoch = next_target_epoch(target);
3219        let source_ghost_expires_at =
3220            Tick::new(source.tick().get().saturating_add(ghost_ttl_ticks));
3221        let transfer = source.prepare_outgoing_handoff(
3222            entity_id,
3223            target_station,
3224            target_epoch,
3225            source_ghost_expires_at,
3226        )?;
3227        target.prewarm_handoff_ghost(&transfer)?;
3228        let target_owner = target.commit_incoming_handoff(transfer.clone())?;
3229        let source_ghost = source.commit_outgoing_handoff(&transfer)?;
3230
3231        Ok(EntityMigrationReport {
3232            transfer,
3233            source_ghost,
3234            target_owner,
3235        })
3236    }
3237}
3238
3239fn next_target_epoch(station: &mut Station) -> OwnerEpoch {
3240    station.next_owner_epoch()
3241}
3242
3243/// Dynamic ownership table for fixed 3D cells.
3244#[derive(Clone, Debug, Default, PartialEq, Eq)]
3245pub struct CellOwnershipTable {
3246    owners: BTreeMap<CellCoord3, StationId>,
3247}
3248
3249impl CellOwnershipTable {
3250    /// Assigns one cell to a station and returns the previous owner.
3251    pub fn assign(&mut self, cell: CellCoord3, station_id: StationId) -> Option<StationId> {
3252        self.owners.insert(cell, station_id)
3253    }
3254
3255    /// Returns the current owner for one cell.
3256    pub fn owner_of(&self, cell: CellCoord3) -> Option<StationId> {
3257        self.owners.get(&cell).copied()
3258    }
3259
3260    /// Applies a split proposal by assigning all proposed cells to `target_station`.
3261    pub fn apply_split(
3262        &mut self,
3263        proposal: &SplitProposal,
3264        target_station: StationId,
3265    ) -> CellOwnershipUpdate {
3266        let mut update = CellOwnershipUpdate::default();
3267        self.apply_split_into(proposal, target_station, &mut update);
3268        update
3269    }
3270
3271    /// Applies a split into caller-owned reusable update storage.
3272    pub fn apply_split_into(
3273        &mut self,
3274        proposal: &SplitProposal,
3275        target_station: StationId,
3276        update: &mut CellOwnershipUpdate,
3277    ) {
3278        update.source_station = proposal.source_station;
3279        update.target_station = target_station;
3280        update.moved_cells.clear();
3281        for cell in &proposal.cells_to_move {
3282            let previous = self.assign(*cell, target_station);
3283            if previous != Some(target_station) {
3284                update.moved_cells.push(*cell);
3285            }
3286        }
3287    }
3288
3289    /// Number of explicitly assigned cells.
3290    pub fn len(&self) -> usize {
3291        self.owners.len()
3292    }
3293
3294    /// Returns whether no cells are explicitly assigned.
3295    pub fn is_empty(&self) -> bool {
3296        self.owners.is_empty()
3297    }
3298}
3299
3300/// Result of applying cell ownership changes.
3301#[derive(Clone, Debug, Default, PartialEq, Eq)]
3302pub struct CellOwnershipUpdate {
3303    /// Previous/source station.
3304    pub source_station: StationId,
3305    /// New/target station.
3306    pub target_station: StationId,
3307    /// Cells whose owner changed.
3308    pub moved_cells: Vec<CellCoord3>,
3309}
3310
3311/// Result of migrating entities indexed by moved cells.
3312#[derive(Clone, Debug, Default, PartialEq)]
3313pub struct CellMigrationReport {
3314    /// Source station.
3315    pub source_station: StationId,
3316    /// Target station.
3317    pub target_station: StationId,
3318    /// Cells scanned for owner entities.
3319    pub scanned_cells: Vec<CellCoord3>,
3320    /// Entity migrations that were committed.
3321    pub entity_migrations: Vec<EntityMigrationReport>,
3322    /// Candidate handles that no longer resolved to an entity.
3323    pub skipped_missing_handles: usize,
3324    /// Candidate entities skipped because they were ghosts or non-authoritative.
3325    pub skipped_non_owned: usize,
3326    /// Duplicate candidate entities skipped after first occurrence.
3327    pub skipped_duplicate_entities: usize,
3328}
3329
3330/// Caller-owned working storage for repeated cell migration passes.
3331#[derive(Clone, Debug, Default)]
3332pub struct CellMigrationScratch {
3333    seen_handles: HashSet<EntityHandle>,
3334    seen_entities: HashSet<EntityId>,
3335    entity_ids: Vec<EntityId>,
3336}
3337
3338impl CellMigrationScratch {
3339    /// Creates empty migration scratch storage.
3340    pub fn new() -> Self {
3341        Self::default()
3342    }
3343
3344    /// Reserves deduplication and candidate storage for an expected pass size.
3345    pub fn reserve(&mut self, handles: usize, entities: usize) {
3346        if self.seen_handles.capacity() < handles {
3347            self.seen_handles
3348                .reserve(handles.saturating_sub(self.seen_handles.len()));
3349        }
3350        if self.seen_entities.capacity() < entities {
3351            self.seen_entities
3352                .reserve(entities.saturating_sub(self.seen_entities.len()));
3353        }
3354        if self.entity_ids.capacity() < entities {
3355            self.entity_ids
3356                .reserve(entities.saturating_sub(self.entity_ids.len()));
3357        }
3358    }
3359
3360    /// Retained handle-deduplication capacity.
3361    pub fn handle_capacity(&self) -> usize {
3362        self.seen_handles.capacity()
3363    }
3364
3365    /// Retained entity-deduplication capacity.
3366    pub fn entity_capacity(&self) -> usize {
3367        self.seen_entities.capacity()
3368    }
3369
3370    /// Retained candidate entity capacity.
3371    pub fn candidate_capacity(&self) -> usize {
3372        self.entity_ids.capacity()
3373    }
3374
3375    fn clear(&mut self) {
3376        self.seen_handles.clear();
3377        self.seen_entities.clear();
3378        self.entity_ids.clear();
3379    }
3380}
3381
3382/// Cell-level migration error.
3383#[derive(Clone, Copy, Debug, PartialEq, Eq)]
3384pub enum CellMigrationError {
3385    /// Entity migration failed.
3386    Entity(EntityMigrationError),
3387    /// Target owner record was not found after a successful migration.
3388    MissingTargetRecord(EntityId),
3389    /// Source ghost record was not found after a successful migration.
3390    MissingSourceRecord(EntityId),
3391}
3392
3393impl core::fmt::Display for CellMigrationError {
3394    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
3395        match self {
3396            Self::Entity(error) => write!(f, "{error}"),
3397            Self::MissingTargetRecord(id) => {
3398                write!(f, "target owner record for entity {} is missing", id.get())
3399            }
3400            Self::MissingSourceRecord(id) => {
3401                write!(f, "source ghost record for entity {} is missing", id.get())
3402            }
3403        }
3404    }
3405}
3406
3407impl std::error::Error for CellMigrationError {}
3408
3409impl From<EntityMigrationError> for CellMigrationError {
3410    fn from(value: EntityMigrationError) -> Self {
3411        Self::Entity(value)
3412    }
3413}
3414
3415/// Executes cell-level ownership migration using station-local indexes.
3416#[derive(Clone, Copy, Debug, Default)]
3417pub struct CellMigrationExecutor;
3418
3419impl CellMigrationExecutor {
3420    /// Migrates owned entities found in `cells` from source station to target station.
3421    pub fn migrate_cells(
3422        stations: &mut StationSet,
3423        source_index: &mut CellIndex,
3424        target_index: &mut CellIndex,
3425        source_station: StationId,
3426        target_station: StationId,
3427        cells: &[CellCoord3],
3428        ghost_ttl_ticks: u64,
3429    ) -> Result<CellMigrationReport, CellMigrationError> {
3430        let mut report = CellMigrationReport::default();
3431        let mut scratch = CellMigrationScratch::new();
3432        Self::migrate_cells_into(
3433            stations,
3434            source_index,
3435            target_index,
3436            source_station,
3437            target_station,
3438            cells,
3439            ghost_ttl_ticks,
3440            &mut scratch,
3441            &mut report,
3442        )?;
3443        Ok(report)
3444    }
3445
3446    /// Migrates cells using caller-owned reusable working and report storage.
3447    ///
3448    /// The report is reset before processing. As with [`Self::migrate_cells`],
3449    /// an error may occur after earlier entities have already migrated.
3450    #[allow(clippy::too_many_arguments)]
3451    pub fn migrate_cells_into(
3452        stations: &mut StationSet,
3453        source_index: &mut CellIndex,
3454        target_index: &mut CellIndex,
3455        source_station: StationId,
3456        target_station: StationId,
3457        cells: &[CellCoord3],
3458        ghost_ttl_ticks: u64,
3459        scratch: &mut CellMigrationScratch,
3460        report: &mut CellMigrationReport,
3461    ) -> Result<(), CellMigrationError> {
3462        report.source_station = source_station;
3463        report.target_station = target_station;
3464        report.scanned_cells.clear();
3465        report.scanned_cells.extend_from_slice(cells);
3466        report.entity_migrations.clear();
3467        report.skipped_missing_handles = 0;
3468        report.skipped_non_owned = 0;
3469        report.skipped_duplicate_entities = 0;
3470        scratch.clear();
3471
3472        {
3473            let source = stations
3474                .get(source_station)
3475                .ok_or(EntityMigrationError::MissingSource(source_station))?;
3476            for cell in cells {
3477                for &handle in source_index.handles_in_cell_slice(*cell) {
3478                    if !scratch.seen_handles.insert(handle) {
3479                        report.skipped_duplicate_entities += 1;
3480                        continue;
3481                    }
3482                    let Some(record) = source.get(handle) else {
3483                        report.skipped_missing_handles += 1;
3484                        continue;
3485                    };
3486                    if record.is_owned() {
3487                        scratch.entity_ids.push(record.id);
3488                    } else {
3489                        report.skipped_non_owned += 1;
3490                    }
3491                }
3492            }
3493        }
3494
3495        for &entity_id in &scratch.entity_ids {
3496            if !scratch.seen_entities.insert(entity_id) {
3497                report.skipped_duplicate_entities += 1;
3498                continue;
3499            }
3500            let migration = EntityMigrationExecutor::migrate_entity(
3501                stations,
3502                entity_id,
3503                source_station,
3504                target_station,
3505                ghost_ttl_ticks,
3506            )?;
3507
3508            {
3509                let target = stations
3510                    .get(target_station)
3511                    .ok_or(EntityMigrationError::MissingTarget(target_station))?;
3512                let target_record = target
3513                    .get(migration.target_owner)
3514                    .ok_or(CellMigrationError::MissingTargetRecord(entity_id))?;
3515                target_index.upsert(
3516                    migration.target_owner,
3517                    target_record.position,
3518                    target_record.bounds,
3519                );
3520            }
3521
3522            {
3523                let source = stations
3524                    .get(source_station)
3525                    .ok_or(EntityMigrationError::MissingSource(source_station))?;
3526                let source_record = source
3527                    .get(migration.source_ghost)
3528                    .ok_or(CellMigrationError::MissingSourceRecord(entity_id))?;
3529                source_index.upsert(
3530                    migration.source_ghost,
3531                    source_record.position,
3532                    source_record.bounds,
3533                );
3534            }
3535
3536            report.entity_migrations.push(migration);
3537        }
3538
3539        Ok(())
3540    }
3541}
3542
3543/// Automatic split scheduler configuration.
3544#[derive(Clone, Copy, Debug, PartialEq, Eq)]
3545pub struct SplitSchedulerConfig {
3546    /// Hotspot thresholds.
3547    pub thresholds: HotspotThresholds,
3548    /// Maximum split actions to create per scheduling pass.
3549    pub max_actions_per_pass: usize,
3550    /// Maximum cells to move in each split action.
3551    pub max_cells_per_action: usize,
3552    /// Source ghost TTL used during migration execution.
3553    pub ghost_ttl_ticks: u64,
3554    /// Minimum load-score gap required between source and target.
3555    pub min_score_improvement: u64,
3556    /// Maximum permitted target load score after moved cell pressure is added.
3557    pub max_target_score_after_move: u64,
3558    /// Ticks a source station must wait before another split can be planned.
3559    pub split_cooldown_ticks: u64,
3560    /// Whether warm target stations may receive split cells.
3561    pub allow_warm_targets: bool,
3562}
3563
3564impl Default for SplitSchedulerConfig {
3565    fn default() -> Self {
3566        Self {
3567            thresholds: HotspotThresholds::default(),
3568            max_actions_per_pass: 4,
3569            max_cells_per_action: 4,
3570            ghost_ttl_ticks: 4,
3571            min_score_improvement: 1,
3572            max_target_score_after_move: u64::MAX,
3573            split_cooldown_ticks: 0,
3574            allow_warm_targets: true,
3575        }
3576    }
3577}
3578
3579/// One scheduled split action.
3580#[derive(Clone, Debug, Default, PartialEq, Eq)]
3581pub struct SplitAction {
3582    /// Source station selected for split.
3583    pub source_station: StationId,
3584    /// Target station selected to receive cells.
3585    pub target_station: StationId,
3586    /// Cell split proposal.
3587    pub proposal: SplitProposal,
3588    /// Source load score observed when planning.
3589    pub source_score: u64,
3590    /// Target load score observed when planning.
3591    pub target_score: u64,
3592    /// Estimated target score after moving proposed cell pressure.
3593    pub estimated_target_score_after_move: u64,
3594}
3595
3596/// Split schedule produced from a load snapshot.
3597#[derive(Clone, Debug, Default, PartialEq, Eq)]
3598pub struct SplitSchedule {
3599    /// Hotspot decisions produced for every input station.
3600    pub decisions: Vec<HotspotDecision>,
3601    /// Actions selected for execution.
3602    pub actions: Vec<SplitAction>,
3603    /// Hot stations skipped because no distinct target existed.
3604    pub skipped_no_target: usize,
3605    /// Hot stations skipped because no cells were proposed.
3606    pub skipped_no_cells: usize,
3607    /// Hot stations skipped because source station is inside split cooldown.
3608    pub skipped_cooldown: usize,
3609    /// Hot stations skipped because all targets were too warm or hot.
3610    pub skipped_target_severity: usize,
3611    /// Hot stations skipped because target capacity would be exceeded.
3612    pub skipped_target_capacity: usize,
3613    /// Hot stations skipped because target score improvement was too small.
3614    pub skipped_insufficient_improvement: usize,
3615}
3616
3617/// Borrowed split schedule produced from reusable scheduler output slots.
3618#[derive(Clone, Copy, Debug, PartialEq, Eq)]
3619pub struct SplitScheduleView<'a> {
3620    /// Hotspot decisions in input sample order.
3621    pub decisions: &'a [HotspotDecision],
3622    /// Admitted split actions in deterministic source order.
3623    pub actions: &'a [SplitAction],
3624    /// Hot stations skipped because no distinct target existed.
3625    pub skipped_no_target: usize,
3626    /// Hot stations skipped because no cells were proposed.
3627    pub skipped_no_cells: usize,
3628    /// Hot stations skipped because source station is inside split cooldown.
3629    pub skipped_cooldown: usize,
3630    /// Hot stations skipped because all targets were too warm or hot.
3631    pub skipped_target_severity: usize,
3632    /// Hot stations skipped because target capacity would be exceeded.
3633    pub skipped_target_capacity: usize,
3634    /// Hot stations skipped because target score improvement was too small.
3635    pub skipped_insufficient_improvement: usize,
3636}
3637
3638impl From<SplitScheduleView<'_>> for SplitSchedule {
3639    fn from(view: SplitScheduleView<'_>) -> Self {
3640        Self {
3641            decisions: view.decisions.to_vec(),
3642            actions: view.actions.to_vec(),
3643            skipped_no_target: view.skipped_no_target,
3644            skipped_no_cells: view.skipped_no_cells,
3645            skipped_cooldown: view.skipped_cooldown,
3646            skipped_target_severity: view.skipped_target_severity,
3647            skipped_target_capacity: view.skipped_target_capacity,
3648            skipped_insufficient_improvement: view.skipped_insufficient_improvement,
3649        }
3650    }
3651}
3652
3653/// Caller-owned reusable output and working storage for split scheduling.
3654#[derive(Clone, Debug, Default)]
3655pub struct SplitSchedulerScratch {
3656    decisions: Vec<HotspotDecision>,
3657    active_decisions: usize,
3658    actions: Vec<SplitAction>,
3659    active_actions: usize,
3660    hotspot: HotspotSplitScratch,
3661    proposal: SplitProposal,
3662    skipped_no_target: usize,
3663    skipped_no_cells: usize,
3664    skipped_cooldown: usize,
3665    skipped_target_severity: usize,
3666    skipped_target_capacity: usize,
3667    skipped_insufficient_improvement: usize,
3668}
3669
3670impl SplitSchedulerScratch {
3671    /// Creates empty scheduler storage.
3672    pub fn new() -> Self {
3673        Self::default()
3674    }
3675
3676    /// Decision slots retained across passes.
3677    pub fn retained_decision_slots(&self) -> usize {
3678        self.decisions.len()
3679    }
3680
3681    /// Action slots retained across passes.
3682    pub fn retained_action_slots(&self) -> usize {
3683        self.actions.len()
3684    }
3685
3686    /// Total reason capacity retained across decision slots.
3687    pub fn retained_reason_capacity(&self) -> usize {
3688        self.decisions
3689            .iter()
3690            .map(|decision| decision.reasons.capacity())
3691            .sum()
3692    }
3693
3694    /// Total cell-coordinate capacity retained across action proposal slots.
3695    pub fn retained_action_cell_capacity(&self) -> usize {
3696        self.actions
3697            .iter()
3698            .map(|action| action.proposal.cells_to_move.capacity())
3699            .sum()
3700    }
3701
3702    /// Hotspot candidate capacity retained across passes.
3703    pub fn retained_candidate_capacity(&self) -> usize {
3704        self.hotspot.candidate_capacity()
3705    }
3706
3707    fn prepare(&mut self, decisions: usize) {
3708        if self.decisions.len() < decisions {
3709            self.decisions
3710                .resize_with(decisions, HotspotDecision::default);
3711        }
3712        self.active_decisions = decisions;
3713        self.active_actions = 0;
3714        self.skipped_no_target = 0;
3715        self.skipped_no_cells = 0;
3716        self.skipped_cooldown = 0;
3717        self.skipped_target_severity = 0;
3718        self.skipped_target_capacity = 0;
3719        self.skipped_insufficient_improvement = 0;
3720    }
3721
3722    fn view(&self) -> SplitScheduleView<'_> {
3723        SplitScheduleView {
3724            decisions: &self.decisions[..self.active_decisions],
3725            actions: &self.actions[..self.active_actions],
3726            skipped_no_target: self.skipped_no_target,
3727            skipped_no_cells: self.skipped_no_cells,
3728            skipped_cooldown: self.skipped_cooldown,
3729            skipped_target_severity: self.skipped_target_severity,
3730            skipped_target_capacity: self.skipped_target_capacity,
3731            skipped_insufficient_improvement: self.skipped_insufficient_improvement,
3732        }
3733    }
3734}
3735
3736/// Mutable planning state for conservative split scheduling.
3737#[derive(Clone, Debug, Default, PartialEq, Eq)]
3738pub struct SplitSchedulerState {
3739    last_split_at: BTreeMap<StationId, Tick>,
3740}
3741
3742impl SplitSchedulerState {
3743    /// Returns the last split tick for a source station.
3744    pub fn last_split_at(&self, station_id: StationId) -> Option<Tick> {
3745        self.last_split_at.get(&station_id).copied()
3746    }
3747
3748    /// Records one executed or externally accepted split action.
3749    pub fn record_action(&mut self, action: &SplitAction, tick: Tick) {
3750        self.last_split_at.insert(action.source_station, tick);
3751    }
3752
3753    /// Records all actions in a schedule at the same tick.
3754    pub fn record_schedule(&mut self, schedule: &SplitSchedule, tick: Tick) {
3755        for action in &schedule.actions {
3756            self.record_action(action, tick);
3757        }
3758    }
3759
3760    /// Records all actions from a borrowed reusable schedule view.
3761    pub fn record_schedule_view(&mut self, schedule: SplitScheduleView<'_>, tick: Tick) {
3762        for action in schedule.actions {
3763            self.record_action(action, tick);
3764        }
3765    }
3766
3767    /// Returns whether a station is inside split cooldown.
3768    pub fn is_in_cooldown(
3769        &self,
3770        station_id: StationId,
3771        current_tick: Tick,
3772        cooldown_ticks: u64,
3773    ) -> bool {
3774        if cooldown_ticks == 0 {
3775            return false;
3776        }
3777        let Some(last_split) = self.last_split_at(station_id) else {
3778            return false;
3779        };
3780        current_tick.get().saturating_sub(last_split.get()) < cooldown_ticks
3781    }
3782}
3783
3784/// Result of executing a split schedule.
3785#[derive(Clone, Debug, Default, PartialEq)]
3786pub struct SplitScheduleExecutionReport {
3787    /// Ownership changes applied.
3788    pub ownership_updates: Vec<CellOwnershipUpdate>,
3789    /// Cell migration reports.
3790    pub cell_migrations: Vec<CellMigrationReport>,
3791}
3792
3793/// Borrowed result of executing a split schedule into reusable storage.
3794#[derive(Clone, Copy, Debug, PartialEq)]
3795pub struct SplitScheduleExecutionView<'a> {
3796    /// Ownership changes applied during this pass.
3797    pub ownership_updates: &'a [CellOwnershipUpdate],
3798    /// Cell migration reports produced during this pass.
3799    pub cell_migrations: &'a [CellMigrationReport],
3800}
3801
3802/// Caller-owned reusable output and working storage for split execution.
3803#[derive(Clone, Debug, Default)]
3804pub struct SplitScheduleExecutionScratch {
3805    ownership_updates: Vec<CellOwnershipUpdate>,
3806    cell_migrations: Vec<CellMigrationReport>,
3807    active_actions: usize,
3808    migration: CellMigrationScratch,
3809}
3810
3811impl SplitScheduleExecutionScratch {
3812    /// Creates empty split-execution storage.
3813    pub fn new() -> Self {
3814        Self::default()
3815    }
3816
3817    /// Reserves outer action slots and per-action cell/entity report capacity.
3818    pub fn reserve(&mut self, actions: usize, cells_per_action: usize, entities_per_action: usize) {
3819        while self.ownership_updates.len() < actions {
3820            self.ownership_updates.push(CellOwnershipUpdate::default());
3821            self.cell_migrations.push(CellMigrationReport::default());
3822        }
3823        for update in &mut self.ownership_updates[..actions] {
3824            if update.moved_cells.capacity() < cells_per_action {
3825                update
3826                    .moved_cells
3827                    .reserve(cells_per_action.saturating_sub(update.moved_cells.len()));
3828            }
3829        }
3830        for report in &mut self.cell_migrations[..actions] {
3831            if report.scanned_cells.capacity() < cells_per_action {
3832                report
3833                    .scanned_cells
3834                    .reserve(cells_per_action.saturating_sub(report.scanned_cells.len()));
3835            }
3836            if report.entity_migrations.capacity() < entities_per_action {
3837                report
3838                    .entity_migrations
3839                    .reserve(entities_per_action.saturating_sub(report.entity_migrations.len()));
3840            }
3841        }
3842        self.migration
3843            .reserve(entities_per_action, entities_per_action);
3844    }
3845
3846    /// Ownership-update slots retained across passes.
3847    pub fn retained_ownership_slots(&self) -> usize {
3848        self.ownership_updates.len()
3849    }
3850
3851    /// Cell-migration report slots retained across passes.
3852    pub fn retained_migration_slots(&self) -> usize {
3853        self.cell_migrations.len()
3854    }
3855
3856    /// Total moved-cell capacity retained across ownership updates.
3857    pub fn retained_update_cell_capacity(&self) -> usize {
3858        self.ownership_updates
3859            .iter()
3860            .map(|update| update.moved_cells.capacity())
3861            .sum()
3862    }
3863
3864    /// Total entity-report capacity retained across cell migrations.
3865    pub fn retained_entity_migration_capacity(&self) -> usize {
3866        self.cell_migrations
3867            .iter()
3868            .map(|report| report.entity_migrations.capacity())
3869            .sum()
3870    }
3871
3872    /// Retained candidate entity capacity in the shared migration scratch.
3873    pub fn retained_candidate_capacity(&self) -> usize {
3874        self.migration.candidate_capacity()
3875    }
3876
3877    fn prepare(&mut self, actions: usize) {
3878        while self.ownership_updates.len() < actions {
3879            self.ownership_updates.push(CellOwnershipUpdate::default());
3880            self.cell_migrations.push(CellMigrationReport::default());
3881        }
3882        self.active_actions = 0;
3883    }
3884
3885    fn view(&self) -> SplitScheduleExecutionView<'_> {
3886        SplitScheduleExecutionView {
3887            ownership_updates: &self.ownership_updates[..self.active_actions],
3888            cell_migrations: &self.cell_migrations[..self.active_actions],
3889        }
3890    }
3891}
3892
3893/// Split schedule execution error.
3894#[derive(Clone, Copy, Debug, PartialEq, Eq)]
3895pub enum SplitScheduleExecutionError {
3896    /// Source index is missing.
3897    MissingSourceIndex(StationId),
3898    /// Target index is missing.
3899    MissingTargetIndex(StationId),
3900    /// Cell migration failed.
3901    CellMigration(CellMigrationError),
3902}
3903
3904impl core::fmt::Display for SplitScheduleExecutionError {
3905    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
3906        match self {
3907            Self::MissingSourceIndex(id) => write!(f, "source index {} is missing", id.get()),
3908            Self::MissingTargetIndex(id) => write!(f, "target index {} is missing", id.get()),
3909            Self::CellMigration(error) => write!(f, "{error}"),
3910        }
3911    }
3912}
3913
3914impl std::error::Error for SplitScheduleExecutionError {}
3915
3916impl From<CellMigrationError> for SplitScheduleExecutionError {
3917    fn from(value: CellMigrationError) -> Self {
3918        Self::CellMigration(value)
3919    }
3920}
3921
3922/// Conservative automatic split scheduler.
3923#[derive(Clone, Copy, Debug)]
3924pub struct SplitScheduler {
3925    /// Scheduler configuration.
3926    pub config: SplitSchedulerConfig,
3927}
3928
3929impl SplitScheduler {
3930    /// Creates a split scheduler.
3931    pub const fn new(config: SplitSchedulerConfig) -> Self {
3932        Self { config }
3933    }
3934
3935    /// Plans split actions from station load samples.
3936    pub fn plan(&self, samples: &[StationLoadSample]) -> SplitSchedule {
3937        let mut scratch = SplitSchedulerScratch::new();
3938        self.plan_into(samples, &mut scratch).into()
3939    }
3940
3941    /// Plans into fully reusable caller-owned output and working storage.
3942    pub fn plan_into<'a>(
3943        &self,
3944        samples: &[StationLoadSample],
3945        scratch: &'a mut SplitSchedulerScratch,
3946    ) -> SplitScheduleView<'a> {
3947        self.plan_with_state_into(samples, None, Tick::new(0), scratch)
3948    }
3949
3950    /// Plans split actions using reusable hotspot cell candidate storage.
3951    pub fn plan_with_scratch(
3952        &self,
3953        samples: &[StationLoadSample],
3954        scratch: &mut HotspotSplitScratch,
3955    ) -> SplitSchedule {
3956        self.plan_with_state_and_scratch(samples, None, Tick::new(0), scratch)
3957    }
3958
3959    /// Plans split actions using optional cooldown state.
3960    pub fn plan_with_state(
3961        &self,
3962        samples: &[StationLoadSample],
3963        state: Option<&SplitSchedulerState>,
3964        current_tick: Tick,
3965    ) -> SplitSchedule {
3966        let mut scratch = SplitSchedulerScratch::new();
3967        self.plan_with_state_into(samples, state, current_tick, &mut scratch)
3968            .into()
3969    }
3970
3971    /// Plans with optional cooldown state and reusable hotspot candidate storage.
3972    pub fn plan_with_state_and_scratch(
3973        &self,
3974        samples: &[StationLoadSample],
3975        state: Option<&SplitSchedulerState>,
3976        current_tick: Tick,
3977        scratch: &mut HotspotSplitScratch,
3978    ) -> SplitSchedule {
3979        let mut scheduler_scratch = SplitSchedulerScratch::new();
3980        core::mem::swap(&mut scheduler_scratch.hotspot, scratch);
3981        let schedule = self
3982            .plan_with_state_into(samples, state, current_tick, &mut scheduler_scratch)
3983            .into();
3984        core::mem::swap(&mut scheduler_scratch.hotspot, scratch);
3985        schedule
3986    }
3987
3988    /// Plans with optional cooldown state into fully reusable scheduler storage.
3989    pub fn plan_with_state_into<'a>(
3990        &self,
3991        samples: &[StationLoadSample],
3992        state: Option<&SplitSchedulerState>,
3993        current_tick: Tick,
3994        scratch: &'a mut SplitSchedulerScratch,
3995    ) -> SplitScheduleView<'a> {
3996        scratch.prepare(samples.len());
3997        for (decision, sample) in scratch.decisions[..samples.len()].iter_mut().zip(samples) {
3998            HotspotPlanner::evaluate_into(sample, self.config.thresholds, decision);
3999        }
4000
4001        for (source_index, source) in samples.iter().enumerate() {
4002            if scratch.active_actions >= self.config.max_actions_per_pass {
4003                break;
4004            }
4005            let source_decision = &scratch.decisions[source_index];
4006            if source_decision.severity != HotspotSeverity::Hot {
4007                continue;
4008            }
4009            if state.is_some_and(|state| {
4010                state.is_in_cooldown(
4011                    source.station_id,
4012                    current_tick,
4013                    self.config.split_cooldown_ticks,
4014                )
4015            }) {
4016                scratch.skipped_cooldown = scratch.skipped_cooldown.saturating_add(1);
4017                continue;
4018            }
4019
4020            HotspotPlanner::propose_cell_split_into(
4021                source,
4022                self.config.max_cells_per_action,
4023                &mut scratch.hotspot,
4024                &mut scratch.proposal,
4025            );
4026            if scratch.proposal.cells_to_move.is_empty() {
4027                scratch.skipped_no_cells = scratch.skipped_no_cells.saturating_add(1);
4028                continue;
4029            }
4030            let target_selection = select_split_target(
4031                source,
4032                &scratch.proposal,
4033                samples,
4034                &scratch.decisions[..scratch.active_decisions],
4035                self.config,
4036            );
4037            let Some(target) = target_selection.target else {
4038                if target_selection.considered_targets == 0 {
4039                    scratch.skipped_no_target = scratch.skipped_no_target.saturating_add(1);
4040                } else {
4041                    scratch.skipped_target_severity = scratch
4042                        .skipped_target_severity
4043                        .saturating_add(usize::from(target_selection.rejected_by_severity > 0));
4044                    scratch.skipped_target_capacity = scratch
4045                        .skipped_target_capacity
4046                        .saturating_add(usize::from(target_selection.rejected_by_capacity > 0));
4047                    scratch.skipped_insufficient_improvement = scratch
4048                        .skipped_insufficient_improvement
4049                        .saturating_add(usize::from(target_selection.rejected_by_improvement > 0));
4050                }
4051                continue;
4052            };
4053            let target_score = station_load_score(target);
4054            let estimated_target_score_after_move =
4055                target_score.saturating_add(scratch.proposal.moved_pressure_score);
4056            let action_index = scratch.active_actions;
4057            if action_index == scratch.actions.len() {
4058                scratch.actions.push(SplitAction::default());
4059            }
4060            let action = &mut scratch.actions[action_index];
4061            action.source_station = source.station_id;
4062            action.target_station = target.station_id;
4063            action.proposal.source_station = scratch.proposal.source_station;
4064            action.proposal.cells_to_move.clear();
4065            action
4066                .proposal
4067                .cells_to_move
4068                .extend_from_slice(&scratch.proposal.cells_to_move);
4069            action.proposal.moved_pressure_score = scratch.proposal.moved_pressure_score;
4070            action.source_score = station_load_score(source);
4071            action.target_score = target_score;
4072            action.estimated_target_score_after_move = estimated_target_score_after_move;
4073            scratch.active_actions = scratch.active_actions.saturating_add(1);
4074        }
4075
4076        scratch.view()
4077    }
4078
4079    /// Executes a split schedule by applying ownership updates and migrating entities.
4080    pub fn execute(
4081        &self,
4082        schedule: &SplitSchedule,
4083        stations: &mut StationSet,
4084        indexes: &mut StationIndexSet,
4085        ownership: &mut CellOwnershipTable,
4086    ) -> Result<SplitScheduleExecutionReport, SplitScheduleExecutionError> {
4087        self.execute_actions(&schedule.actions, stations, indexes, ownership)
4088    }
4089
4090    /// Executes actions directly from a borrowed reusable schedule view.
4091    pub fn execute_view(
4092        &self,
4093        schedule: SplitScheduleView<'_>,
4094        stations: &mut StationSet,
4095        indexes: &mut StationIndexSet,
4096        ownership: &mut CellOwnershipTable,
4097    ) -> Result<SplitScheduleExecutionReport, SplitScheduleExecutionError> {
4098        self.execute_actions(schedule.actions, stations, indexes, ownership)
4099    }
4100
4101    /// Executes an owned schedule into fully reusable caller-owned storage.
4102    pub fn execute_into<'a>(
4103        &self,
4104        schedule: &SplitSchedule,
4105        stations: &mut StationSet,
4106        indexes: &mut StationIndexSet,
4107        ownership: &mut CellOwnershipTable,
4108        scratch: &'a mut SplitScheduleExecutionScratch,
4109    ) -> Result<SplitScheduleExecutionView<'a>, SplitScheduleExecutionError> {
4110        self.execute_actions_into(&schedule.actions, stations, indexes, ownership, scratch)
4111    }
4112
4113    /// Executes a borrowed schedule into fully reusable caller-owned storage.
4114    pub fn execute_view_into<'a>(
4115        &self,
4116        schedule: SplitScheduleView<'_>,
4117        stations: &mut StationSet,
4118        indexes: &mut StationIndexSet,
4119        ownership: &mut CellOwnershipTable,
4120        scratch: &'a mut SplitScheduleExecutionScratch,
4121    ) -> Result<SplitScheduleExecutionView<'a>, SplitScheduleExecutionError> {
4122        self.execute_actions_into(schedule.actions, stations, indexes, ownership, scratch)
4123    }
4124
4125    fn execute_actions_into<'a>(
4126        &self,
4127        actions: &[SplitAction],
4128        stations: &mut StationSet,
4129        indexes: &mut StationIndexSet,
4130        ownership: &mut CellOwnershipTable,
4131        scratch: &'a mut SplitScheduleExecutionScratch,
4132    ) -> Result<SplitScheduleExecutionView<'a>, SplitScheduleExecutionError> {
4133        scratch.prepare(actions.len());
4134
4135        for action in actions {
4136            if indexes.get(action.source_station).is_none() {
4137                return Err(SplitScheduleExecutionError::MissingSourceIndex(
4138                    action.source_station,
4139                ));
4140            }
4141            if indexes.get(action.target_station).is_none() {
4142                return Err(SplitScheduleExecutionError::MissingTargetIndex(
4143                    action.target_station,
4144                ));
4145            }
4146
4147            let action_index = scratch.active_actions;
4148            let update = &mut scratch.ownership_updates[action_index];
4149            ownership.apply_split_into(&action.proposal, action.target_station, update);
4150            let (source_index, target_index) = indexes
4151                .get_pair_mut(action.source_station, action.target_station)
4152                .expect("indexes were checked above");
4153            CellMigrationExecutor::migrate_cells_into(
4154                stations,
4155                source_index,
4156                target_index,
4157                action.source_station,
4158                action.target_station,
4159                &update.moved_cells,
4160                self.config.ghost_ttl_ticks,
4161                &mut scratch.migration,
4162                &mut scratch.cell_migrations[action_index],
4163            )?;
4164            scratch.active_actions = scratch.active_actions.saturating_add(1);
4165        }
4166
4167        Ok(scratch.view())
4168    }
4169
4170    fn execute_actions(
4171        &self,
4172        actions: &[SplitAction],
4173        stations: &mut StationSet,
4174        indexes: &mut StationIndexSet,
4175        ownership: &mut CellOwnershipTable,
4176    ) -> Result<SplitScheduleExecutionReport, SplitScheduleExecutionError> {
4177        let mut report = SplitScheduleExecutionReport::default();
4178
4179        for action in actions {
4180            if indexes.get(action.source_station).is_none() {
4181                return Err(SplitScheduleExecutionError::MissingSourceIndex(
4182                    action.source_station,
4183                ));
4184            }
4185            if indexes.get(action.target_station).is_none() {
4186                return Err(SplitScheduleExecutionError::MissingTargetIndex(
4187                    action.target_station,
4188                ));
4189            }
4190
4191            let update = ownership.apply_split(&action.proposal, action.target_station);
4192            let (source_index, target_index) = indexes
4193                .get_pair_mut(action.source_station, action.target_station)
4194                .expect("indexes were checked above");
4195            let migration = CellMigrationExecutor::migrate_cells(
4196                stations,
4197                source_index,
4198                target_index,
4199                action.source_station,
4200                action.target_station,
4201                &update.moved_cells,
4202                self.config.ghost_ttl_ticks,
4203            )?;
4204            report.ownership_updates.push(update);
4205            report.cell_migrations.push(migration);
4206        }
4207
4208        Ok(report)
4209    }
4210}
4211
4212impl Default for SplitScheduler {
4213    fn default() -> Self {
4214        Self::new(SplitSchedulerConfig::default())
4215    }
4216}
4217
4218#[derive(Clone, Copy, Debug, Default)]
4219struct SplitTargetSelection<'a> {
4220    target: Option<&'a StationLoadSample>,
4221    target_key: Option<(u8, u64, u32)>,
4222    considered_targets: usize,
4223    rejected_by_severity: usize,
4224    rejected_by_capacity: usize,
4225    rejected_by_improvement: usize,
4226}
4227
4228fn select_split_target<'a>(
4229    source: &StationLoadSample,
4230    proposal: &SplitProposal,
4231    samples: &'a [StationLoadSample],
4232    decisions: &[HotspotDecision],
4233    config: SplitSchedulerConfig,
4234) -> SplitTargetSelection<'a> {
4235    let mut selection = SplitTargetSelection::default();
4236    let source_score = station_load_score(source);
4237
4238    for (target, decision) in samples.iter().zip(decisions) {
4239        if target.station_id == source.station_id {
4240            continue;
4241        }
4242        selection.considered_targets += 1;
4243
4244        debug_assert_eq!(decision.station_id, target.station_id);
4245        let severity = decision.severity;
4246        if severity == HotspotSeverity::Hot
4247            || (severity == HotspotSeverity::Warm && !config.allow_warm_targets)
4248        {
4249            selection.rejected_by_severity += 1;
4250            continue;
4251        }
4252
4253        let target_score = station_load_score(target);
4254        if source_score.saturating_sub(target_score) < config.min_score_improvement {
4255            selection.rejected_by_improvement += 1;
4256            continue;
4257        }
4258        if target_score.saturating_add(proposal.moved_pressure_score)
4259            > config.max_target_score_after_move
4260        {
4261            selection.rejected_by_capacity += 1;
4262            continue;
4263        }
4264
4265        let target_key = (
4266            severity_rank(severity),
4267            target_score,
4268            target.station_id.get(),
4269        );
4270        if selection
4271            .target_key
4272            .is_none_or(|current_key| target_key < current_key)
4273        {
4274            selection.target = Some(target);
4275            selection.target_key = Some(target_key);
4276        }
4277    }
4278
4279    selection
4280}
4281
4282fn severity_rank(severity: HotspotSeverity) -> u8 {
4283    match severity {
4284        HotspotSeverity::Normal => 0,
4285        HotspotSeverity::Warm => 1,
4286        HotspotSeverity::Hot => 2,
4287    }
4288}
4289
4290fn station_load_score(sample: &StationLoadSample) -> u64 {
4291    (sample.total_entities() as u64)
4292        .saturating_mul(8)
4293        .saturating_add((sample.subscribers as u64).saturating_mul(4))
4294        .saturating_add(sample.queued_events as u64)
4295        .saturating_add((sample.estimated_bytes / 256) as u64)
4296        .saturating_add(sample.tick_cost_units)
4297}
4298
4299/// Event router statistics.
4300#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
4301pub struct EventRouterStats {
4302    /// Events accepted by target queues.
4303    pub routed_events: usize,
4304    /// Ready events drained for station application.
4305    pub drained_events: usize,
4306    /// Best-effort events dropped by bounded target queues.
4307    pub dropped_best_effort_events: usize,
4308}
4309
4310/// Event router error.
4311#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4312pub enum EventRouterError {
4313    /// Target station was not registered with the router.
4314    MissingTarget(StationId),
4315    /// Underlying target queue rejected the event.
4316    Queue(EventQueueError),
4317}
4318
4319impl core::fmt::Display for EventRouterError {
4320    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
4321        match self {
4322            Self::MissingTarget(id) => write!(f, "event target station {} is missing", id.get()),
4323            Self::Queue(error) => write!(f, "{error}"),
4324        }
4325    }
4326}
4327
4328impl std::error::Error for EventRouterError {}
4329
4330impl From<EventQueueError> for EventRouterError {
4331    fn from(value: EventQueueError) -> Self {
4332        Self::Queue(value)
4333    }
4334}
4335
4336/// In-process station event router.
4337#[derive(Clone, Debug)]
4338pub struct EventRouter {
4339    limits: EventQueueLimits,
4340    queues: BTreeMap<StationId, EventQueues>,
4341    stats: EventRouterStats,
4342}
4343
4344impl EventRouter {
4345    /// Creates an empty event router.
4346    pub fn new(limits: EventQueueLimits) -> Self {
4347        Self {
4348            limits,
4349            queues: BTreeMap::new(),
4350            stats: EventRouterStats::default(),
4351        }
4352    }
4353
4354    /// Registers a station target queue.
4355    pub fn register_station(&mut self, station_id: StationId) {
4356        self.queues
4357            .entry(station_id)
4358            .or_insert_with(|| EventQueues::new(self.limits));
4359    }
4360
4361    /// Registers all stations in a set.
4362    pub fn register_stations(&mut self, stations: &StationSet) {
4363        for station in stations.iter() {
4364            self.register_station(station.config().station_id);
4365        }
4366    }
4367
4368    /// Unregisters a station queue and returns the number of queued events dropped.
4369    pub fn unregister_station(&mut self, station_id: StationId) -> Option<usize> {
4370        self.queues.remove(&station_id).map(|queue| queue.len())
4371    }
4372
4373    /// Routes an event to its target station queue.
4374    pub fn route(&mut self, event: StationEvent) -> Result<PushOutcome, EventRouterError> {
4375        let queue = self
4376            .queues
4377            .get_mut(&event.target)
4378            .ok_or(EventRouterError::MissingTarget(event.target))?;
4379        let outcome = queue.push(event)?;
4380        self.stats.routed_events += 1;
4381        if outcome == PushOutcome::DroppedOldestBestEffort {
4382            self.stats.dropped_best_effort_events += 1;
4383        }
4384        Ok(outcome)
4385    }
4386
4387    /// Drains events whose `target_tick` is ready for application.
4388    pub fn drain_ready(
4389        &mut self,
4390        station_id: StationId,
4391        current_tick: Tick,
4392    ) -> Result<Vec<StationEvent>, EventRouterError> {
4393        let mut ready = Vec::new();
4394        self.drain_ready_into(station_id, current_tick, &mut ready)?;
4395        Ok(ready)
4396    }
4397
4398    /// Drains ready events into caller-owned storage while retaining its capacity.
4399    pub fn drain_ready_into(
4400        &mut self,
4401        station_id: StationId,
4402        current_tick: Tick,
4403        ready: &mut Vec<StationEvent>,
4404    ) -> Result<(), EventRouterError> {
4405        ready.clear();
4406        self.append_ready(station_id, current_tick, ready)?;
4407        Ok(())
4408    }
4409
4410    fn append_ready(
4411        &mut self,
4412        station_id: StationId,
4413        current_tick: Tick,
4414        ready: &mut Vec<StationEvent>,
4415    ) -> Result<(), EventRouterError> {
4416        let queue = self
4417            .queues
4418            .get_mut(&station_id)
4419            .ok_or(EventRouterError::MissingTarget(station_id))?;
4420        let drained = queue.drain_ready_into(current_tick, ready);
4421        self.stats.drained_events = self.stats.drained_events.saturating_add(drained);
4422        Ok(())
4423    }
4424
4425    /// Returns queued event count for one station.
4426    pub fn queued_len(&self, station_id: StationId) -> Option<usize> {
4427        self.queues.get(&station_id).map(EventQueues::len)
4428    }
4429
4430    /// Returns router statistics.
4431    pub const fn stats(&self) -> EventRouterStats {
4432        self.stats
4433    }
4434}
4435
4436impl Default for EventRouter {
4437    fn default() -> Self {
4438        Self::new(EventQueueLimits::default())
4439    }
4440}
4441
4442/// Statistics for station event transport bridging.
4443#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
4444pub struct StationEventTransportStats {
4445    /// Events encoded and submitted to station transport.
4446    pub events_sent: usize,
4447    /// Bytes submitted to station transport.
4448    pub bytes_sent: usize,
4449    /// Packets received from station transport.
4450    pub packets_received: usize,
4451    /// Bytes received from station transport.
4452    pub bytes_received: usize,
4453    /// Events decoded and accepted by the target router.
4454    pub events_routed: usize,
4455}
4456
4457/// Result of pumping station event packets for one target.
4458#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
4459pub struct StationEventPumpReport {
4460    /// Target station pumped.
4461    pub target_station: StationId,
4462    /// Packets consumed from the station transport.
4463    pub packets_received: usize,
4464    /// Bytes consumed from the station transport.
4465    pub bytes_received: usize,
4466    /// Events accepted by the target router.
4467    pub events_routed: usize,
4468}
4469
4470/// Error produced while bridging station events through packet transport.
4471#[derive(Clone, Debug, PartialEq, Eq)]
4472pub enum StationEventTransportError<E> {
4473    /// Underlying station transport failed.
4474    Transport(E),
4475    /// Wire encoding failed.
4476    Encode(BinaryEncodeError),
4477    /// Wire decoding failed.
4478    Decode(BinaryDecodeError),
4479    /// Packet decoded as a non-event frame.
4480    UnexpectedFrame,
4481    /// Packet envelope and decoded event disagreed about endpoints.
4482    EndpointMismatch {
4483        /// Packet source station.
4484        packet_source: StationId,
4485        /// Packet target station.
4486        packet_target: StationId,
4487        /// Decoded event source station.
4488        event_source: StationId,
4489        /// Decoded event target station.
4490        event_target: StationId,
4491    },
4492    /// Event router rejected the decoded event.
4493    Router(EventRouterError),
4494}
4495
4496impl<E: core::fmt::Display> core::fmt::Display for StationEventTransportError<E> {
4497    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
4498        match self {
4499            Self::Transport(error) => write!(f, "{error}"),
4500            Self::Encode(error) => write!(f, "{error}"),
4501            Self::Decode(error) => write!(f, "{error}"),
4502            Self::UnexpectedFrame => f.write_str("station transport packet was not an event frame"),
4503            Self::EndpointMismatch {
4504                packet_source,
4505                packet_target,
4506                event_source,
4507                event_target,
4508            } => write!(
4509                f,
4510                "station event endpoint mismatch: packet {}->{}, event {}->{}",
4511                packet_source.get(),
4512                packet_target.get(),
4513                event_source.get(),
4514                event_target.get()
4515            ),
4516            Self::Router(error) => write!(f, "{error}"),
4517        }
4518    }
4519}
4520
4521impl<E> std::error::Error for StationEventTransportError<E>
4522where
4523    E: std::error::Error + 'static,
4524{
4525    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
4526        match self {
4527            Self::Transport(error) => Some(error),
4528            Self::Encode(error) => Some(error),
4529            Self::Decode(error) => Some(error),
4530            Self::UnexpectedFrame | Self::EndpointMismatch { .. } => None,
4531            Self::Router(error) => Some(error),
4532        }
4533    }
4534}
4535
4536impl<E> From<BinaryEncodeError> for StationEventTransportError<E> {
4537    fn from(value: BinaryEncodeError) -> Self {
4538        Self::Encode(value)
4539    }
4540}
4541
4542impl<E> From<BinaryDecodeError> for StationEventTransportError<E> {
4543    fn from(value: BinaryDecodeError) -> Self {
4544        Self::Decode(value)
4545    }
4546}
4547
4548impl<E> From<EventRouterError> for StationEventTransportError<E> {
4549    fn from(value: EventRouterError) -> Self {
4550        Self::Router(value)
4551    }
4552}
4553
4554/// Bridge between typed station events and bounded station packet transport.
4555#[derive(Clone, Debug, Default)]
4556pub struct StationEventTransportBridge {
4557    stats: StationEventTransportStats,
4558}
4559
4560impl StationEventTransportBridge {
4561    /// Returns bridge statistics.
4562    pub const fn stats(&self) -> StationEventTransportStats {
4563        self.stats
4564    }
4565
4566    /// Encodes and sends one station event through the station transport.
4567    pub fn send_event<T>(
4568        &mut self,
4569        transport: &mut T,
4570        event: &StationEvent,
4571    ) -> Result<(), StationEventTransportError<T::Error>>
4572    where
4573        T: StationTransportSink,
4574    {
4575        let frame = StationEventFrame::from_event(event);
4576        let mut bytes = Vec::with_capacity(64);
4577        BinaryFrameEncoder.encode_station_event(&frame, &mut bytes)?;
4578        let byte_len = bytes.len();
4579        transport
4580            .send_station(StationOutboundPacket {
4581                source_station: event.source,
4582                target_station: event.target,
4583                bytes,
4584            })
4585            .map_err(StationEventTransportError::Transport)?;
4586        self.stats.events_sent = self.stats.events_sent.saturating_add(1);
4587        self.stats.bytes_sent = self.stats.bytes_sent.saturating_add(byte_len);
4588        Ok(())
4589    }
4590
4591    /// Receives up to `max_packets` for `target_station`, decodes station
4592    /// events, and routes them into `router`.
4593    pub fn pump_target<T>(
4594        &mut self,
4595        transport: &mut T,
4596        router: &mut EventRouter,
4597        target_station: StationId,
4598        max_packets: usize,
4599    ) -> Result<StationEventPumpReport, StationEventTransportError<T::Error>>
4600    where
4601        T: StationTransportReceiver,
4602    {
4603        let mut report = StationEventPumpReport {
4604            target_station,
4605            ..StationEventPumpReport::default()
4606        };
4607        for _ in 0..max_packets {
4608            let Some(packet) = transport
4609                .try_recv_station(target_station)
4610                .map_err(StationEventTransportError::Transport)?
4611            else {
4612                break;
4613            };
4614            report.packets_received = report.packets_received.saturating_add(1);
4615            report.bytes_received = report.bytes_received.saturating_add(packet.bytes.len());
4616
4617            let decoded = BinaryFrameDecoder.decode(&packet.bytes)?;
4618            let RuntimeFrame::StationEvent(frame) = decoded else {
4619                return Err(StationEventTransportError::UnexpectedFrame);
4620            };
4621            if frame.source_station != packet.source_station
4622                || frame.target_station != packet.target_station
4623            {
4624                return Err(StationEventTransportError::EndpointMismatch {
4625                    packet_source: packet.source_station,
4626                    packet_target: packet.target_station,
4627                    event_source: frame.source_station,
4628                    event_target: frame.target_station,
4629                });
4630            }
4631
4632            router.route(frame.into_event())?;
4633            report.events_routed = report.events_routed.saturating_add(1);
4634        }
4635
4636        self.stats.packets_received = self
4637            .stats
4638            .packets_received
4639            .saturating_add(report.packets_received);
4640        self.stats.bytes_received = self
4641            .stats
4642            .bytes_received
4643            .saturating_add(report.bytes_received);
4644        self.stats.events_routed = self
4645            .stats
4646            .events_routed
4647            .saturating_add(report.events_routed);
4648        Ok(report)
4649    }
4650}
4651
4652/// Statistics for command dispatch transport bridging.
4653#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
4654pub struct CommandDispatchTransportStats {
4655    /// Commands encoded and submitted to station transport.
4656    pub commands_sent: usize,
4657    /// Bytes submitted to station transport.
4658    pub bytes_sent: usize,
4659    /// Packets received from station transport.
4660    pub packets_received: usize,
4661    /// Bytes received from station transport.
4662    pub bytes_received: usize,
4663    /// Commands decoded and enqueued at the target station.
4664    pub commands_enqueued: usize,
4665    /// Commands rejected by target station queues.
4666    pub commands_rejected_queue: usize,
4667}
4668
4669/// Result of pumping command dispatch packets for one target.
4670#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
4671pub struct CommandDispatchPumpReport {
4672    /// Target station pumped.
4673    pub target_station: StationId,
4674    /// Packets consumed from station transport.
4675    pub packets_received: usize,
4676    /// Bytes consumed from station transport.
4677    pub bytes_received: usize,
4678    /// Commands enqueued into the target queue.
4679    pub commands_enqueued: usize,
4680}
4681
4682/// Error produced while bridging command dispatch frames through station packet transport.
4683#[derive(Clone, Debug, PartialEq, Eq)]
4684pub enum CommandDispatchTransportError<E> {
4685    /// Underlying station transport failed.
4686    Transport(E),
4687    /// Wire encoding failed.
4688    Encode(BinaryEncodeError),
4689    /// Wire decoding failed.
4690    Decode(BinaryDecodeError),
4691    /// Packet decoded as a non-command-dispatch frame.
4692    UnexpectedFrame,
4693    /// Packet envelope and decoded dispatch frame disagreed about the target.
4694    EndpointMismatch {
4695        /// Packet source station.
4696        packet_source: StationId,
4697        /// Packet target station.
4698        packet_target: StationId,
4699        /// Decoded command dispatch target station.
4700        dispatch_target: StationId,
4701    },
4702    /// Target station queue was not registered.
4703    MissingQueue(StationId),
4704    /// Target station queue rejected the command.
4705    Queue(CommandQueueError),
4706}
4707
4708impl<E: core::fmt::Display> core::fmt::Display for CommandDispatchTransportError<E> {
4709    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
4710        match self {
4711            Self::Transport(error) => write!(f, "{error}"),
4712            Self::Encode(error) => write!(f, "{error}"),
4713            Self::Decode(error) => write!(f, "{error}"),
4714            Self::UnexpectedFrame => {
4715                f.write_str("station transport packet was not a command dispatch frame")
4716            }
4717            Self::EndpointMismatch {
4718                packet_source,
4719                packet_target,
4720                dispatch_target,
4721            } => write!(
4722                f,
4723                "command dispatch endpoint mismatch: packet {}->{}, dispatch target {}",
4724                packet_source.get(),
4725                packet_target.get(),
4726                dispatch_target.get()
4727            ),
4728            Self::MissingQueue(station_id) => {
4729                write!(
4730                    f,
4731                    "command dispatch target station {} has no queue",
4732                    station_id.get()
4733                )
4734            }
4735            Self::Queue(error) => write!(f, "{error}"),
4736        }
4737    }
4738}
4739
4740impl<E> std::error::Error for CommandDispatchTransportError<E>
4741where
4742    E: std::error::Error + 'static,
4743{
4744    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
4745        match self {
4746            Self::Transport(error) => Some(error),
4747            Self::Encode(error) => Some(error),
4748            Self::Decode(error) => Some(error),
4749            Self::UnexpectedFrame | Self::EndpointMismatch { .. } | Self::MissingQueue(_) => None,
4750            Self::Queue(error) => Some(error),
4751        }
4752    }
4753}
4754
4755impl<E> From<BinaryEncodeError> for CommandDispatchTransportError<E> {
4756    fn from(value: BinaryEncodeError) -> Self {
4757        Self::Encode(value)
4758    }
4759}
4760
4761impl<E> From<BinaryDecodeError> for CommandDispatchTransportError<E> {
4762    fn from(value: BinaryDecodeError) -> Self {
4763        Self::Decode(value)
4764    }
4765}
4766
4767impl<E> From<CommandQueueError> for CommandDispatchTransportError<E> {
4768    fn from(value: CommandQueueError) -> Self {
4769        Self::Queue(value)
4770    }
4771}
4772
4773/// Bridge between stamped command envelopes and bounded station packet transport.
4774#[derive(Clone, Debug, Default)]
4775pub struct CommandDispatchTransportBridge {
4776    stats: CommandDispatchTransportStats,
4777}
4778
4779impl CommandDispatchTransportBridge {
4780    /// Returns bridge statistics.
4781    pub const fn stats(&self) -> CommandDispatchTransportStats {
4782        self.stats
4783    }
4784
4785    /// Encodes and sends a stamped command envelope to a station node.
4786    pub fn send_envelope<T>(
4787        &mut self,
4788        transport: &mut T,
4789        source_station: StationId,
4790        target_station: StationId,
4791        command: &CommandEnvelope,
4792    ) -> Result<(), CommandDispatchTransportError<T::Error>>
4793    where
4794        T: StationTransportSink,
4795    {
4796        let mut bytes = Vec::with_capacity(64_usize.saturating_add(command.payload.len()));
4797        BinaryFrameEncoder.encode_command_dispatch_envelope(target_station, command, &mut bytes)?;
4798        self.send_encoded(transport, source_station, target_station, bytes)
4799    }
4800
4801    /// Encodes and sends a command dispatch frame to its target station.
4802    pub fn send_frame<T>(
4803        &mut self,
4804        transport: &mut T,
4805        source_station: StationId,
4806        frame: &CommandDispatchFrame,
4807    ) -> Result<(), CommandDispatchTransportError<T::Error>>
4808    where
4809        T: StationTransportSink,
4810    {
4811        let mut bytes = Vec::with_capacity(64);
4812        BinaryFrameEncoder.encode_command_dispatch(frame, &mut bytes)?;
4813        self.send_encoded(transport, source_station, frame.station_id, bytes)
4814    }
4815
4816    fn send_encoded<T>(
4817        &mut self,
4818        transport: &mut T,
4819        source_station: StationId,
4820        target_station: StationId,
4821        bytes: Vec<u8>,
4822    ) -> Result<(), CommandDispatchTransportError<T::Error>>
4823    where
4824        T: StationTransportSink,
4825    {
4826        let byte_len = bytes.len();
4827        transport
4828            .send_station(StationOutboundPacket {
4829                source_station,
4830                target_station,
4831                bytes,
4832            })
4833            .map_err(CommandDispatchTransportError::Transport)?;
4834        self.stats.commands_sent = self.stats.commands_sent.saturating_add(1);
4835        self.stats.bytes_sent = self.stats.bytes_sent.saturating_add(byte_len);
4836        Ok(())
4837    }
4838
4839    /// Receives up to `max_packets` for `target_station`, decodes command
4840    /// dispatch frames, and enqueues stamped commands into `station_queues`.
4841    pub fn pump_target<T>(
4842        &mut self,
4843        transport: &mut T,
4844        station_queues: &mut BTreeMap<StationId, CommandQueues>,
4845        target_station: StationId,
4846        max_packets: usize,
4847        ingress: CommandIngress,
4848    ) -> Result<CommandDispatchPumpReport, CommandDispatchTransportError<T::Error>>
4849    where
4850        T: StationTransportReceiver,
4851    {
4852        let mut report = CommandDispatchPumpReport {
4853            target_station,
4854            ..CommandDispatchPumpReport::default()
4855        };
4856        for _ in 0..max_packets {
4857            let Some(packet) = transport
4858                .try_recv_station(target_station)
4859                .map_err(CommandDispatchTransportError::Transport)?
4860            else {
4861                break;
4862            };
4863            report.packets_received = report.packets_received.saturating_add(1);
4864            report.bytes_received = report.bytes_received.saturating_add(packet.bytes.len());
4865
4866            let decoded = BinaryFrameDecoder.decode(&packet.bytes)?;
4867            let RuntimeFrame::CommandDispatch(frame) = decoded else {
4868                return Err(CommandDispatchTransportError::UnexpectedFrame);
4869            };
4870            if frame.station_id != packet.target_station {
4871                return Err(CommandDispatchTransportError::EndpointMismatch {
4872                    packet_source: packet.source_station,
4873                    packet_target: packet.target_station,
4874                    dispatch_target: frame.station_id,
4875                });
4876            }
4877
4878            let queue = station_queues.get_mut(&frame.station_id).ok_or(
4879                CommandDispatchTransportError::MissingQueue(frame.station_id),
4880            )?;
4881            match queue.push(frame.into_envelope(), ingress) {
4882                Ok(_) => {
4883                    report.commands_enqueued = report.commands_enqueued.saturating_add(1);
4884                }
4885                Err(error) => {
4886                    self.stats.commands_rejected_queue =
4887                        self.stats.commands_rejected_queue.saturating_add(1);
4888                    return Err(CommandDispatchTransportError::Queue(error));
4889                }
4890            }
4891        }
4892
4893        self.stats.packets_received = self
4894            .stats
4895            .packets_received
4896            .saturating_add(report.packets_received);
4897        self.stats.bytes_received = self
4898            .stats
4899            .bytes_received
4900            .saturating_add(report.bytes_received);
4901        self.stats.commands_enqueued = self
4902            .stats
4903            .commands_enqueued
4904            .saturating_add(report.commands_enqueued);
4905        Ok(report)
4906    }
4907}
4908
4909/// Budget for a load-aware scheduler step.
4910#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4911pub struct StationScheduleConfig {
4912    /// Maximum stations that may advance during one scheduler step.
4913    pub max_station_advances_per_step: usize,
4914}
4915
4916impl Default for StationScheduleConfig {
4917    fn default() -> Self {
4918        Self {
4919            max_station_advances_per_step: usize::MAX,
4920        }
4921    }
4922}
4923
4924/// Candidate selected by the load-aware station scheduler.
4925#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4926pub struct StationScheduleCandidate {
4927    /// Station selected for advancement.
4928    pub station_id: StationId,
4929    /// Deterministic pressure score derived from the latest load sample.
4930    pub load_score: u64,
4931    /// How far this station is behind the most advanced station in the set.
4932    pub tick_lag: u64,
4933}
4934
4935/// Result of one load-aware station scheduling pass.
4936#[derive(Clone, Debug, Default, PartialEq, Eq)]
4937pub struct StationSchedulePlan {
4938    /// Stations considered by this pass.
4939    pub candidates_considered: usize,
4940    /// Stations selected by this pass.
4941    pub stations_selected: usize,
4942    /// Station tick advances requested by this pass.
4943    pub total_advances: usize,
4944    /// Selected stations in deterministic execution order.
4945    pub selected: Vec<StationScheduleCandidate>,
4946}
4947
4948/// Borrowed deterministic result from reusable Station scheduling storage.
4949#[derive(Clone, Copy, Debug, PartialEq, Eq)]
4950pub struct StationScheduleView<'a> {
4951    /// Stations considered by this pass.
4952    pub candidates_considered: usize,
4953    /// Stations selected by this pass.
4954    pub stations_selected: usize,
4955    /// Station tick advances requested by this pass.
4956    pub total_advances: usize,
4957    /// Selected stations in deterministic execution order.
4958    pub selected: &'a [StationScheduleCandidate],
4959}
4960
4961/// Caller-owned reusable storage for load-aware Station scheduling.
4962///
4963/// Storage contains only derived pressure scores and stateless candidates. It
4964/// does not retain scheduling decisions, cooldowns, or gameplay state.
4965#[derive(Clone, Debug, Default)]
4966pub struct StationScheduleScratch {
4967    scores: HashMap<StationId, u64>,
4968    candidates: Vec<StationScheduleCandidate>,
4969}
4970
4971impl StationScheduleScratch {
4972    /// Creates empty scheduling scratch.
4973    pub fn new() -> Self {
4974        Self::default()
4975    }
4976
4977    /// Hash-table capacity retained for sampled Station scores.
4978    pub fn score_capacity(&self) -> usize {
4979        self.scores.capacity()
4980    }
4981
4982    /// Candidate capacity retained across scheduling passes.
4983    pub fn candidate_capacity(&self) -> usize {
4984        self.candidates.capacity()
4985    }
4986}
4987
4988/// Basic in-process station scheduler.
4989#[derive(Clone, Debug, Default)]
4990pub struct StationScheduler {
4991    /// Total station ticks advanced by this scheduler.
4992    pub advanced_ticks: u64,
4993}
4994
4995impl StationScheduler {
4996    /// Advances every station by one tick.
4997    pub fn advance_all(&mut self, stations: &mut StationSet) {
4998        for station in stations.iter_mut() {
4999            station.advance_tick();
5000            self.advanced_ticks = self.advanced_ticks.saturating_add(1);
5001        }
5002    }
5003
5004    /// Plans a bounded station advancement pass from load samples.
5005    pub fn plan_loaded(
5006        &self,
5007        stations: &StationSet,
5008        samples: &[StationLoadSample],
5009        config: StationScheduleConfig,
5010    ) -> StationSchedulePlan {
5011        let mut scratch = StationScheduleScratch::default();
5012        let view = self.plan_loaded_into(stations, samples, config, &mut scratch);
5013        StationSchedulePlan {
5014            candidates_considered: view.candidates_considered,
5015            stations_selected: view.stations_selected,
5016            total_advances: view.total_advances,
5017            selected: view.selected.to_vec(),
5018        }
5019    }
5020
5021    /// Plans into caller-owned storage and returns a borrowed deterministic top-k view.
5022    pub fn plan_loaded_into<'a>(
5023        &self,
5024        stations: &StationSet,
5025        samples: &[StationLoadSample],
5026        config: StationScheduleConfig,
5027        scratch: &'a mut StationScheduleScratch,
5028    ) -> StationScheduleView<'a> {
5029        let candidates_considered = stations.len();
5030        let limit = config
5031            .max_station_advances_per_step
5032            .min(candidates_considered);
5033        let max_tick = stations
5034            .iter()
5035            .map(|station| station.tick().get())
5036            .max()
5037            .unwrap_or(0);
5038        scratch.scores.clear();
5039        scratch.scores.reserve(samples.len());
5040        for sample in samples {
5041            scratch
5042                .scores
5043                .insert(sample.station_id, station_schedule_score(sample));
5044        }
5045        scratch.candidates.clear();
5046        scratch.candidates.reserve(candidates_considered);
5047        scratch.candidates.extend(stations.iter().map(|station| {
5048            let station_id = station.config().station_id;
5049            StationScheduleCandidate {
5050                station_id,
5051                load_score: scratch.scores.get(&station_id).copied().unwrap_or(0),
5052                tick_lag: max_tick.saturating_sub(station.tick().get()),
5053            }
5054        }));
5055        prioritize_station_candidates(&mut scratch.candidates, limit);
5056        let selected = &scratch.candidates[..limit];
5057
5058        StationScheduleView {
5059            candidates_considered,
5060            stations_selected: selected.len(),
5061            total_advances: selected.len(),
5062            selected,
5063        }
5064    }
5065
5066    /// Advances a bounded set of high-load stations by one tick each.
5067    pub fn advance_loaded(
5068        &mut self,
5069        stations: &mut StationSet,
5070        samples: &[StationLoadSample],
5071        config: StationScheduleConfig,
5072    ) -> StationSchedulePlan {
5073        let plan = self.plan_loaded(stations, samples, config);
5074        for candidate in &plan.selected {
5075            if let Some(station) = stations.get_mut(candidate.station_id) {
5076                station.advance_tick();
5077                self.advanced_ticks = self.advanced_ticks.saturating_add(1);
5078            }
5079        }
5080        plan
5081    }
5082
5083    /// Advances a bounded top-k set using reusable caller-owned scheduling storage.
5084    pub fn advance_loaded_into<'a>(
5085        &mut self,
5086        stations: &mut StationSet,
5087        samples: &[StationLoadSample],
5088        config: StationScheduleConfig,
5089        scratch: &'a mut StationScheduleScratch,
5090    ) -> StationScheduleView<'a> {
5091        let plan = self.plan_loaded_into(stations, samples, config, scratch);
5092        for candidate in plan.selected {
5093            if let Some(station) = stations.get_mut(candidate.station_id) {
5094                station.advance_tick();
5095                self.advanced_ticks = self.advanced_ticks.saturating_add(1);
5096            }
5097        }
5098        plan
5099    }
5100
5101    /// Drains router events ready for each station's current tick.
5102    pub fn drain_ready_events(
5103        &mut self,
5104        stations: &StationSet,
5105        router: &mut EventRouter,
5106    ) -> Result<Vec<StationEvent>, EventRouterError> {
5107        let mut events = Vec::new();
5108        self.drain_ready_events_into(stations, router, &mut events)?;
5109        Ok(events)
5110    }
5111
5112    /// Drains all Station-ready events into reusable caller-owned output.
5113    pub fn drain_ready_events_into(
5114        &mut self,
5115        stations: &StationSet,
5116        router: &mut EventRouter,
5117        events: &mut Vec<StationEvent>,
5118    ) -> Result<(), EventRouterError> {
5119        events.clear();
5120        for station in stations.iter() {
5121            router.append_ready(station.config().station_id, station.tick(), events)?;
5122        }
5123        Ok(())
5124    }
5125}
5126
5127fn compare_station_schedule_candidates(
5128    left: &StationScheduleCandidate,
5129    right: &StationScheduleCandidate,
5130) -> core::cmp::Ordering {
5131    right
5132        .load_score
5133        .cmp(&left.load_score)
5134        .then_with(|| right.tick_lag.cmp(&left.tick_lag))
5135        .then_with(|| left.station_id.cmp(&right.station_id))
5136}
5137
5138fn prioritize_station_candidates(candidates: &mut [StationScheduleCandidate], limit: usize) {
5139    if limit == 0 {
5140        return;
5141    }
5142    if limit.saturating_mul(2) < candidates.len() {
5143        candidates.select_nth_unstable_by(limit, compare_station_schedule_candidates);
5144        candidates[..limit].sort_by(compare_station_schedule_candidates);
5145    } else {
5146        candidates.sort_by(compare_station_schedule_candidates);
5147    }
5148}
5149
5150fn station_schedule_score(sample: &StationLoadSample) -> u64 {
5151    station_load_score(sample).saturating_add(sample.max_cell_pressure())
5152}
5153
5154/// Per-station progress inside a full runtime barrier.
5155#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5156pub enum StationBarrierPhase {
5157    /// Station is part of the barrier but has not reached the target tick.
5158    WaitingTick,
5159    /// Station reached the target tick and is frozen.
5160    Frozen,
5161    /// Station has resumed.
5162    Resumed,
5163}
5164
5165/// Barrier progress summary.
5166#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5167pub struct BarrierProgress {
5168    /// Barrier state.
5169    pub state: BarrierState,
5170    /// Number of stations covered by the barrier.
5171    pub station_count: usize,
5172    /// Number of stations frozen.
5173    pub frozen_count: usize,
5174    /// Target tick selected for the barrier.
5175    pub target_tick: Tick,
5176}
5177
5178/// Runtime barrier metrics.
5179#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
5180pub struct BarrierMetrics {
5181    /// Number of stations covered by this barrier.
5182    pub station_count: usize,
5183    /// Number of snapshots exported while frozen.
5184    pub snapshots_exported: usize,
5185    /// Number of times polling observed at least one station still waiting.
5186    pub waiting_polls: u64,
5187    /// Number of times polling observed a fully frozen barrier.
5188    pub frozen_polls: u64,
5189}
5190
5191/// Runtime barrier execution error.
5192#[derive(Clone, Copy, Debug, PartialEq, Eq)]
5193pub enum BarrierRuntimeError {
5194    /// A barrier is already active.
5195    AlreadyActive(BarrierId),
5196    /// No barrier is active.
5197    NoActiveBarrier,
5198    /// Barrier scope matched no stations.
5199    EmptyScope(BarrierScope),
5200    /// Requested operation requires frozen state.
5201    NotFrozen(BarrierState),
5202    /// A station covered by the barrier is missing.
5203    MissingStation(StationId),
5204}
5205
5206/// Caller-owned reusable station snapshot slots for frozen barrier exports.
5207#[derive(Clone, Debug, Default)]
5208pub struct BarrierSnapshotScratch {
5209    snapshots: Vec<StationSnapshot>,
5210    active_snapshots: usize,
5211}
5212
5213impl BarrierSnapshotScratch {
5214    /// Creates empty barrier snapshot storage.
5215    pub fn new() -> Self {
5216        Self::default()
5217    }
5218
5219    /// Reserves Station slots and per-Station entity capacity.
5220    pub fn reserve(&mut self, stations: usize, entities_per_station: usize) {
5221        if self.snapshots.len() < stations {
5222            self.snapshots
5223                .resize_with(stations, StationSnapshot::default);
5224        }
5225        for snapshot in &mut self.snapshots[..stations] {
5226            if snapshot.entities.capacity() < entities_per_station {
5227                snapshot
5228                    .entities
5229                    .reserve(entities_per_station.saturating_sub(snapshot.entities.len()));
5230            }
5231        }
5232    }
5233
5234    /// Snapshot slots retained across exports.
5235    pub fn retained_snapshot_slots(&self) -> usize {
5236        self.snapshots.len()
5237    }
5238
5239    /// Total entity capacity retained across Station snapshot slots.
5240    pub fn retained_entity_capacity(&self) -> usize {
5241        self.snapshots
5242            .iter()
5243            .map(|snapshot| snapshot.entities.capacity())
5244            .sum()
5245    }
5246
5247    fn prepare(&mut self, stations: usize) {
5248        if self.snapshots.len() < stations {
5249            self.snapshots
5250                .resize_with(stations, StationSnapshot::default);
5251        }
5252        self.active_snapshots = stations;
5253    }
5254
5255    fn active(&self) -> &[StationSnapshot] {
5256        &self.snapshots[..self.active_snapshots]
5257    }
5258}
5259
5260impl core::fmt::Display for BarrierRuntimeError {
5261    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
5262        match self {
5263            Self::AlreadyActive(id) => write!(f, "barrier {} is already active", id.get()),
5264            Self::NoActiveBarrier => f.write_str("no active barrier"),
5265            Self::EmptyScope(scope) => write!(f, "barrier scope {scope:?} contains no stations"),
5266            Self::NotFrozen(state) => {
5267                write!(f, "barrier operation requires Frozen state, got {state:?}")
5268            }
5269            Self::MissingStation(id) => write!(f, "barrier station {} is missing", id.get()),
5270        }
5271    }
5272}
5273
5274impl std::error::Error for BarrierRuntimeError {}
5275
5276/// Full runtime barrier executor for in-process station sets.
5277#[derive(Clone, Debug, Default)]
5278pub struct BarrierController {
5279    active: Option<RuntimeBarrier>,
5280    phases: BTreeMap<StationId, StationBarrierPhase>,
5281    metrics: BarrierMetrics,
5282}
5283
5284impl BarrierController {
5285    /// Returns the active barrier, if any.
5286    pub const fn active(&self) -> Option<RuntimeBarrier> {
5287        self.active
5288    }
5289
5290    /// Requests a barrier over stations matching `scope`.
5291    pub fn request(
5292        &mut self,
5293        stations: &StationSet,
5294        id: BarrierId,
5295        scope: BarrierScope,
5296        target_tick: Tick,
5297        command_mode: CommandQueueMode,
5298    ) -> Result<BarrierProgress, BarrierRuntimeError> {
5299        if let Some(active) = self.active {
5300            return Err(BarrierRuntimeError::AlreadyActive(active.id));
5301        }
5302
5303        let station_ids = stations.station_ids_in_scope(scope);
5304        if station_ids.is_empty() {
5305            return Err(BarrierRuntimeError::EmptyScope(scope));
5306        }
5307
5308        let requested_at = station_ids
5309            .iter()
5310            .filter_map(|station_id| stations.get(*station_id).map(Station::tick))
5311            .map(Tick::get)
5312            .max()
5313            .map_or(Tick::new(0), Tick::new);
5314
5315        let mut barrier =
5316            RuntimeBarrier::requested(id, scope, requested_at, target_tick, command_mode);
5317        barrier.wait_for_tick_boundary();
5318
5319        self.metrics = BarrierMetrics {
5320            station_count: station_ids.len(),
5321            ..BarrierMetrics::default()
5322        };
5323        self.phases.clear();
5324        for station_id in station_ids {
5325            self.phases
5326                .insert(station_id, StationBarrierPhase::WaitingTick);
5327        }
5328        self.active = Some(barrier);
5329
5330        Ok(self.progress())
5331    }
5332
5333    /// Polls station ticks and freezes the barrier once all covered stations are aligned.
5334    pub fn poll(&mut self, stations: &StationSet) -> Result<BarrierProgress, BarrierRuntimeError> {
5335        let Some(mut barrier) = self.active else {
5336            return Err(BarrierRuntimeError::NoActiveBarrier);
5337        };
5338
5339        if matches!(barrier.state, BarrierState::Frozen) {
5340            self.metrics.frozen_polls = self.metrics.frozen_polls.saturating_add(1);
5341            return Ok(self.progress());
5342        }
5343
5344        let mut all_ready = true;
5345        for (station_id, phase) in &mut self.phases {
5346            let station = stations
5347                .get(*station_id)
5348                .ok_or(BarrierRuntimeError::MissingStation(*station_id))?;
5349            if station.tick() >= barrier.target_tick {
5350                *phase = StationBarrierPhase::Frozen;
5351            } else {
5352                all_ready = false;
5353            }
5354        }
5355
5356        if all_ready {
5357            barrier.freeze();
5358            self.active = Some(barrier);
5359            self.metrics.frozen_polls = self.metrics.frozen_polls.saturating_add(1);
5360        } else {
5361            self.metrics.waiting_polls = self.metrics.waiting_polls.saturating_add(1);
5362        }
5363
5364        Ok(self.progress())
5365    }
5366
5367    /// Exports station snapshots while the barrier is frozen.
5368    pub fn export_snapshots(
5369        &mut self,
5370        stations: &StationSet,
5371        version: SnapshotVersion,
5372    ) -> Result<Vec<StationSnapshot>, BarrierRuntimeError> {
5373        let barrier = self.active.ok_or(BarrierRuntimeError::NoActiveBarrier)?;
5374        if barrier.state != BarrierState::Frozen {
5375            return Err(BarrierRuntimeError::NotFrozen(barrier.state));
5376        }
5377
5378        let mut snapshots = Vec::with_capacity(self.phases.len());
5379        for station_id in self.phases.keys().copied() {
5380            let station = stations
5381                .get(station_id)
5382                .ok_or(BarrierRuntimeError::MissingStation(station_id))?;
5383            snapshots.push(station.snapshot(version));
5384        }
5385        self.metrics.snapshots_exported = self
5386            .metrics
5387            .snapshots_exported
5388            .saturating_add(snapshots.len());
5389        Ok(snapshots)
5390    }
5391
5392    /// Exports frozen snapshots into caller-owned reusable Station slots.
5393    pub fn export_snapshots_into<'a>(
5394        &mut self,
5395        stations: &StationSet,
5396        version: SnapshotVersion,
5397        scratch: &'a mut BarrierSnapshotScratch,
5398    ) -> Result<&'a [StationSnapshot], BarrierRuntimeError> {
5399        let barrier = self.active.ok_or(BarrierRuntimeError::NoActiveBarrier)?;
5400        if barrier.state != BarrierState::Frozen {
5401            return Err(BarrierRuntimeError::NotFrozen(barrier.state));
5402        }
5403
5404        scratch.prepare(self.phases.len());
5405        for (snapshot, station_id) in scratch.snapshots[..scratch.active_snapshots]
5406            .iter_mut()
5407            .zip(self.phases.keys().copied())
5408        {
5409            let station = stations
5410                .get(station_id)
5411                .ok_or(BarrierRuntimeError::MissingStation(station_id))?;
5412            station.snapshot_into(version, snapshot);
5413        }
5414        self.metrics.snapshots_exported = self
5415            .metrics
5416            .snapshots_exported
5417            .saturating_add(scratch.active_snapshots);
5418        Ok(scratch.active())
5419    }
5420
5421    /// Resumes all stations covered by the barrier and returns final metrics.
5422    pub fn resume(&mut self) -> Result<BarrierMetrics, BarrierRuntimeError> {
5423        let Some(mut barrier) = self.active else {
5424            return Err(BarrierRuntimeError::NoActiveBarrier);
5425        };
5426        if barrier.state != BarrierState::Frozen {
5427            return Err(BarrierRuntimeError::NotFrozen(barrier.state));
5428        }
5429
5430        barrier.resume();
5431        for phase in self.phases.values_mut() {
5432            *phase = StationBarrierPhase::Resumed;
5433        }
5434        barrier.finish();
5435        let metrics = self.metrics;
5436        self.active = None;
5437        self.phases.clear();
5438        self.metrics = BarrierMetrics::default();
5439        Ok(metrics)
5440    }
5441
5442    /// Returns current barrier progress.
5443    pub fn progress(&self) -> BarrierProgress {
5444        let state = self
5445            .active
5446            .map_or(BarrierState::Running, |barrier| barrier.state);
5447        let target_tick = self
5448            .active
5449            .map_or(Tick::new(0), |barrier| barrier.target_tick);
5450        let frozen_count = self
5451            .phases
5452            .values()
5453            .filter(|phase| matches!(phase, StationBarrierPhase::Frozen))
5454            .count();
5455
5456        BarrierProgress {
5457            state,
5458            station_count: self.phases.len(),
5459            frozen_count,
5460            target_tick,
5461        }
5462    }
5463}
5464
5465/// Report produced after applying an external upgrade hook to frozen snapshots.
5466#[derive(Clone, Debug, PartialEq, Eq)]
5467pub struct BarrierUpgradeReport {
5468    /// Snapshot version requested for export.
5469    pub version: SnapshotVersion,
5470    /// Snapshots passed through the upgrade hook.
5471    pub snapshots_migrated: usize,
5472    /// Stations restored from migrated snapshots.
5473    pub stations_restored: usize,
5474    /// Entity records restored across all stations.
5475    pub entities_restored: usize,
5476}
5477
5478/// Error produced while applying an upgrade hook around a frozen barrier.
5479#[derive(Clone, Debug, PartialEq, Eq)]
5480pub enum BarrierUpgradeError {
5481    /// Barrier was missing or not frozen.
5482    Barrier(BarrierRuntimeError),
5483    /// Station disappeared between snapshot export and restore.
5484    MissingStation(StationId),
5485    /// Restoring a migrated snapshot failed.
5486    Restore {
5487        /// Station being restored.
5488        station_id: StationId,
5489        /// Restore error.
5490        error: StationError,
5491    },
5492}
5493
5494impl core::fmt::Display for BarrierUpgradeError {
5495    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
5496        match self {
5497            Self::Barrier(error) => write!(f, "{error}"),
5498            Self::MissingStation(station_id) => {
5499                write!(f, "upgrade station {} is missing", station_id.get())
5500            }
5501            Self::Restore { station_id, error } => {
5502                write!(
5503                    f,
5504                    "upgrade restore for station {} failed: {error}",
5505                    station_id.get()
5506                )
5507            }
5508        }
5509    }
5510}
5511
5512impl std::error::Error for BarrierUpgradeError {
5513    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
5514        match self {
5515            Self::Barrier(error) => Some(error),
5516            Self::Restore { error, .. } => Some(error),
5517            Self::MissingStation(_) => None,
5518        }
5519    }
5520}
5521
5522impl From<BarrierRuntimeError> for BarrierUpgradeError {
5523    fn from(value: BarrierRuntimeError) -> Self {
5524        Self::Barrier(value)
5525    }
5526}
5527
5528/// Applies an external in-memory upgrade hook while a runtime barrier is frozen.
5529#[derive(Clone, Copy, Debug, Default)]
5530pub struct BarrierUpgradeExecutor;
5531
5532impl BarrierUpgradeExecutor {
5533    /// Exports frozen station snapshots, lets `hook` migrate them, and restores
5534    /// every station only after all migrated snapshots are valid.
5535    pub fn migrate_frozen<H>(
5536        controller: &mut BarrierController,
5537        stations: &mut StationSet,
5538        version: SnapshotVersion,
5539        hook: &mut H,
5540    ) -> Result<BarrierUpgradeReport, BarrierUpgradeError>
5541    where
5542        H: RuntimeUpgradeHook,
5543    {
5544        let report_version = version;
5545        let snapshots = controller.export_snapshots(stations, version)?;
5546        let mut restored = Vec::with_capacity(snapshots.len());
5547        let mut entities_restored = 0usize;
5548
5549        for snapshot in snapshots {
5550            let station_id = snapshot.meta.station_id;
5551            let config = stations
5552                .get(station_id)
5553                .ok_or(BarrierUpgradeError::MissingStation(station_id))?
5554                .config();
5555            hook.pre_upgrade(&snapshot.meta);
5556            let migrated = hook.migrate_state(snapshot);
5557            let migrated_meta = migrated.meta.clone();
5558            let restored_station = Station::restore(config, migrated)
5559                .map_err(|error| BarrierUpgradeError::Restore { station_id, error })?;
5560            entities_restored = entities_restored.saturating_add(restored_station.len());
5561            hook.post_upgrade(&migrated_meta);
5562            restored.push((station_id, restored_station));
5563        }
5564
5565        let stations_restored = restored.len();
5566        for (station_id, restored_station) in restored {
5567            let station = stations
5568                .get_mut(station_id)
5569                .ok_or(BarrierUpgradeError::MissingStation(station_id))?;
5570            *station = restored_station;
5571        }
5572
5573        Ok(BarrierUpgradeReport {
5574            version: report_version,
5575            snapshots_migrated: stations_restored,
5576            stations_restored,
5577            entities_restored,
5578        })
5579    }
5580}
5581
5582#[cfg(test)]
5583mod tests {
5584    use super::*;
5585    use sectorsync_core::prelude::{
5586        Bounds, CellCoord3, CellLoadSample, CommandEnvelope, CommandPriority, CommandQueueLimits,
5587        CompiledSyncPolicy, ComponentDescriptor, ComponentId, ComponentMigrationMode,
5588        ComponentSyncMode, EventId, EventKind, EventPriority, GatewayConfig, GridSpec,
5589        HotspotThresholds, InstanceId, NodeId, PolicyId, Position3, RangeOnlyVisibility,
5590        SnapshotMeta, StationConfig, StationLoadSample, U32LeCodec,
5591    };
5592    use sectorsync_transport::{
5593        BudgetedTransport, ClientTransportLimits, FakeTransport, InMemoryStationTransport,
5594        InMemoryTransportHub, OutboundPacket, StationOutboundPacket, StationTransportSink,
5595        TransportReceiver, TransportSink,
5596    };
5597    use sectorsync_wire::{
5598        BarrierFrame, BinaryFrameDecoder, BinaryFrameEncoder, CommandAckFrame,
5599        CommandDispatchFrame, CommandFrame, ComponentDelta, EntityDelta, FrameDecoder,
5600        FrameEncoder, ReplicationFrame,
5601    };
5602
5603    fn station(station_id: u32, instance_id: u64) -> Station {
5604        Station::new(StationConfig {
5605            station_id: StationId::new(station_id),
5606            node_id: NodeId::new(0),
5607            instance_id: InstanceId::new(instance_id),
5608            tick_rate_hz: 20,
5609        })
5610    }
5611
5612    fn encode_command_frame(sequence: u64) -> Vec<u8> {
5613        let frame = CommandFrame {
5614            client_id: ClientId::new(7),
5615            command_id: CommandId::new(sequence),
5616            entity_id: EntityId::new(100),
5617            sequence,
5618            kind: 1,
5619            priority: CommandPriority::High,
5620            payload: b"move:north".to_vec(),
5621        };
5622        let mut bytes = Vec::new();
5623        BinaryFrameEncoder
5624            .encode_command(&frame, &mut bytes)
5625            .expect("command should encode");
5626        bytes
5627    }
5628
5629    fn command_queues() -> CommandQueues {
5630        CommandQueues::new(CommandQueueLimits {
5631            high: 4,
5632            normal: 4,
5633            low: 4,
5634        })
5635    }
5636
5637    fn gateway(max_commands_per_tick: usize) -> GatewaySessionTable {
5638        GatewaySessionTable::new(GatewayConfig {
5639            max_sessions: 8,
5640            reconnect_grace_ticks: 10,
5641            max_commands_per_tick,
5642        })
5643    }
5644
5645    #[test]
5646    fn station_set_indexes_first_slot_and_reserves_both_storage_classes() {
5647        let mut stations = StationSet::with_capacity(3);
5648        let mut duplicate = station(1, 99);
5649        duplicate.advance_tick();
5650        stations.push(station(1, 10));
5651        stations.push(duplicate);
5652        stations.push(station(2, 10));
5653
5654        assert!(stations.station_capacity() >= 3);
5655        assert!(!stations.lookup_index_active());
5656        assert_eq!(
5657            stations
5658                .get(StationId::new(1))
5659                .expect("first exists")
5660                .tick(),
5661            Tick::new(0)
5662        );
5663        let (first, second) = stations
5664            .get_pair_mut(StationId::new(1), StationId::new(2))
5665            .expect("distinct indexed Stations should borrow");
5666        first.advance_tick();
5667        second.advance_tick();
5668        assert_eq!(
5669            stations
5670                .get(StationId::new(1))
5671                .expect("first exists")
5672                .tick(),
5673            Tick::new(1)
5674        );
5675        assert_eq!(
5676            stations
5677                .get(StationId::new(2))
5678                .expect("second exists")
5679                .tick(),
5680            Tick::new(1)
5681        );
5682
5683        let lookup_capacity = stations.lookup_capacity();
5684        stations.reserve(4);
5685        assert!(stations.station_capacity() >= stations.len().saturating_add(4));
5686        assert!(stations.lookup_capacity() >= lookup_capacity);
5687    }
5688
5689    #[test]
5690    fn station_index_set_replaces_in_place_and_indexes_mutable_pairs() {
5691        let grid = GridSpec::new(10.0).expect("grid should build");
5692        let first_id = StationId::new(1);
5693        let second_id = StationId::new(2);
5694        let first_handle = EntityHandle::new(1, 0);
5695        let second_handle = EntityHandle::new(2, 0);
5696        let mut indexes = StationIndexSet::with_capacity(2);
5697        indexes.insert(first_id, CellIndex::new(grid));
5698        indexes.insert(second_id, CellIndex::new(grid));
5699
5700        let mut replacement = CellIndex::new(grid);
5701        replacement.upsert(first_handle, Position3::new(1.0, 0.0, 0.0), Bounds::Point);
5702        indexes.insert(first_id, replacement);
5703        assert_eq!(indexes.len(), 2);
5704        assert_eq!(
5705            indexes.iter().map(|(id, _)| id).collect::<Vec<_>>(),
5706            vec![first_id, second_id]
5707        );
5708        assert_eq!(
5709            indexes
5710                .get(first_id)
5711                .expect("first index exists")
5712                .entity_count(),
5713            1
5714        );
5715
5716        let (first, second) = indexes
5717            .get_pair_mut(first_id, second_id)
5718            .expect("distinct indexed cells should borrow");
5719        first.remove(first_handle);
5720        second.upsert(second_handle, Position3::new(11.0, 0.0, 0.0), Bounds::Point);
5721        assert_eq!(
5722            indexes
5723                .get(first_id)
5724                .expect("first index exists")
5725                .entity_count(),
5726            0
5727        );
5728        assert_eq!(
5729            indexes
5730                .get(second_id)
5731                .expect("second index exists")
5732                .entity_count(),
5733            1
5734        );
5735        assert!(indexes.index_capacity() >= 2);
5736        assert!(!indexes.lookup_index_active());
5737    }
5738
5739    #[test]
5740    fn station_registries_activate_lookup_index_at_adaptive_threshold() {
5741        let grid = GridSpec::new(10.0).expect("grid should build");
5742        let mut stations = StationSet::with_capacity(STATION_LOOKUP_INDEX_THRESHOLD);
5743        let mut indexes = StationIndexSet::with_capacity(STATION_LOOKUP_INDEX_THRESHOLD);
5744        for raw_id in 1..=STATION_LOOKUP_INDEX_THRESHOLD {
5745            let station_id = StationId::new(u32::try_from(raw_id).expect("threshold fits u32"));
5746            stations.push(station(station_id.get(), 10));
5747            indexes.insert(station_id, CellIndex::new(grid));
5748            if raw_id < STATION_LOOKUP_INDEX_THRESHOLD {
5749                assert!(!stations.lookup_index_active());
5750                assert!(!indexes.lookup_index_active());
5751            }
5752        }
5753
5754        assert!(stations.lookup_index_active());
5755        assert!(indexes.lookup_index_active());
5756        assert!(stations.lookup_capacity() >= STATION_LOOKUP_INDEX_THRESHOLD);
5757        assert!(indexes.lookup_capacity() >= STATION_LOOKUP_INDEX_THRESHOLD);
5758        let last = StationId::new(
5759            u32::try_from(STATION_LOOKUP_INDEX_THRESHOLD).expect("threshold fits u32"),
5760        );
5761        assert_eq!(
5762            stations
5763                .get(last)
5764                .expect("last Station exists")
5765                .config()
5766                .station_id,
5767            last
5768        );
5769        assert!(indexes.get(last).is_some());
5770
5771        let removed_id = StationId::new(2);
5772        let removed_station = stations.remove(removed_id).expect("Station should remove");
5773        let removed_index = indexes.remove(removed_id).expect("index should remove");
5774        assert_eq!(removed_station.config().station_id, removed_id);
5775        assert_eq!(removed_index.entity_count(), 0);
5776        assert!(stations.get(removed_id).is_none());
5777        assert!(indexes.get(removed_id).is_none());
5778        assert_eq!(
5779            stations
5780                .get(last)
5781                .expect("shifted Station resolves")
5782                .config()
5783                .station_id,
5784            last
5785        );
5786        assert!(indexes.get(last).is_some());
5787        assert_eq!(
5788            stations
5789                .iter()
5790                .map(|station| station.config().station_id)
5791                .nth(1),
5792            Some(StationId::new(3))
5793        );
5794        assert_eq!(
5795            indexes.iter().nth(1).map(|(id, _)| id),
5796            Some(StationId::new(3))
5797        );
5798        assert!(stations.lookup_index_active());
5799        assert!(indexes.lookup_index_active());
5800    }
5801
5802    #[test]
5803    fn barrier_freezes_snapshots_and_resumes_instance_scope() {
5804        let mut stations = StationSet::default();
5805        stations.push(station(1, 10));
5806        stations.push(station(2, 10));
5807
5808        for station in stations.iter_mut() {
5809            station.advance_tick();
5810            station.advance_tick();
5811        }
5812
5813        let mut controller = BarrierController::default();
5814        let requested = controller
5815            .request(
5816                &stations,
5817                BarrierId::new(7),
5818                BarrierScope::Instance(InstanceId::new(10)),
5819                Tick::new(2),
5820                CommandQueueMode::Buffer,
5821            )
5822            .expect("request should work");
5823        assert_eq!(requested.state, BarrierState::WaitingTickBoundary);
5824
5825        let frozen = controller.poll(&stations).expect("poll should work");
5826        assert_eq!(frozen.state, BarrierState::Frozen);
5827        assert_eq!(frozen.frozen_count, 2);
5828
5829        let mut scratch = BarrierSnapshotScratch::new();
5830        scratch.reserve(2, 1);
5831        let snapshots = controller
5832            .export_snapshots_into(&stations, SnapshotVersion::default(), &mut scratch)
5833            .expect("reusable snapshot should work while frozen");
5834        assert_eq!(snapshots.len(), 2);
5835        let retained_slots = scratch.retained_snapshot_slots();
5836        let retained_entities = scratch.retained_entity_capacity();
5837        scratch.reserve(2, 1);
5838        let snapshots = controller
5839            .export_snapshots_into(&stations, SnapshotVersion::default(), &mut scratch)
5840            .expect("second reusable snapshot should work while frozen");
5841        assert_eq!(snapshots.len(), 2);
5842        assert_eq!(scratch.retained_snapshot_slots(), retained_slots);
5843        assert_eq!(scratch.retained_entity_capacity(), retained_entities);
5844
5845        let metrics = controller.resume().expect("resume should work");
5846        assert_eq!(metrics.station_count, 2);
5847        assert_eq!(metrics.snapshots_exported, 4);
5848        assert_eq!(controller.progress().state, BarrierState::Running);
5849    }
5850
5851    #[derive(Default)]
5852    struct MoveSnapshotUpgrade {
5853        pre: usize,
5854        migrations: usize,
5855        post: usize,
5856    }
5857
5858    impl RuntimeUpgradeHook for MoveSnapshotUpgrade {
5859        fn pre_upgrade(&mut self, meta: &SnapshotMeta) {
5860            self.pre = self.pre.saturating_add(1);
5861            assert_eq!(meta.version.runtime_version, 2);
5862        }
5863
5864        fn migrate_state(&mut self, mut snapshot: StationSnapshot) -> StationSnapshot {
5865            self.migrations = self.migrations.saturating_add(1);
5866            for entity in &mut snapshot.entities {
5867                entity.position.x += 10.0;
5868            }
5869            snapshot
5870        }
5871
5872        fn post_upgrade(&mut self, meta: &SnapshotMeta) {
5873            self.post = self.post.saturating_add(1);
5874            assert_eq!(meta.version.runtime_version, 2);
5875        }
5876    }
5877
5878    #[test]
5879    fn barrier_upgrade_executor_migrates_and_restores_frozen_snapshots() {
5880        let mut first = station(1, 10);
5881        first
5882            .spawn_owned(
5883                EntityId::new(100),
5884                Position3::new(1.0, 2.0, 3.0),
5885                Bounds::Point,
5886                PolicyId::new(0),
5887            )
5888            .expect("spawn should work");
5889        let mut stations = StationSet::default();
5890        stations.push(first);
5891        stations.push(station(2, 10));
5892
5893        for station in stations.iter_mut() {
5894            station.advance_tick();
5895            station.advance_tick();
5896        }
5897
5898        let mut controller = BarrierController::default();
5899        controller
5900            .request(
5901                &stations,
5902                BarrierId::new(8),
5903                BarrierScope::Instance(InstanceId::new(10)),
5904                Tick::new(2),
5905                CommandQueueMode::Buffer,
5906            )
5907            .expect("request should work");
5908        assert_eq!(
5909            controller.poll(&stations).expect("poll should work").state,
5910            BarrierState::Frozen
5911        );
5912
5913        let mut hook = MoveSnapshotUpgrade::default();
5914        let version = SnapshotVersion {
5915            runtime_version: 2,
5916            ..SnapshotVersion::default()
5917        };
5918        let report = BarrierUpgradeExecutor::migrate_frozen(
5919            &mut controller,
5920            &mut stations,
5921            version,
5922            &mut hook,
5923        )
5924        .expect("upgrade should migrate frozen snapshots");
5925
5926        assert_eq!(report.version, version);
5927        assert_eq!(report.snapshots_migrated, 2);
5928        assert_eq!(report.stations_restored, 2);
5929        assert_eq!(report.entities_restored, 1);
5930        assert_eq!(hook.pre, 2);
5931        assert_eq!(hook.migrations, 2);
5932        assert_eq!(hook.post, 2);
5933        let moved = stations
5934            .get(StationId::new(1))
5935            .expect("station should exist")
5936            .get_by_id(EntityId::new(100))
5937            .expect("entity should restore");
5938        assert_eq!(moved.position, Position3::new(11.0, 2.0, 3.0));
5939        assert_eq!(controller.progress().state, BarrierState::Frozen);
5940
5941        let metrics = controller.resume().expect("resume should work");
5942        assert_eq!(metrics.snapshots_exported, 2);
5943        assert_eq!(controller.progress().state, BarrierState::Running);
5944    }
5945
5946    #[test]
5947    fn barrier_transport_bridge_broadcasts_client_notifications() {
5948        let server_id = ClientId::new(0);
5949        let clients = [ClientId::new(7), ClientId::new(8)];
5950        let hub = InMemoryTransportHub::new(ClientTransportLimits {
5951            max_queued_packets_per_client: 4,
5952            max_packet_bytes: 512,
5953        });
5954        let mut server_transport = hub
5955            .endpoint(server_id, "127.0.0.1:23400".parse().expect("server addr"))
5956            .expect("server endpoint should register");
5957        let mut client_transports = clients
5958            .into_iter()
5959            .enumerate()
5960            .map(|(index, client_id)| {
5961                hub.endpoint(
5962                    client_id,
5963                    format!("127.0.0.1:{}", 23407 + index)
5964                        .parse()
5965                        .expect("client addr"),
5966                )
5967                .expect("client endpoint should register")
5968            })
5969            .collect::<Vec<_>>();
5970        let mut barrier = RuntimeBarrier::requested(
5971            BarrierId::new(5),
5972            BarrierScope::Instance(InstanceId::new(10)),
5973            Tick::new(10),
5974            Tick::new(12),
5975            CommandQueueMode::Buffer,
5976        );
5977        barrier.wait_for_tick_boundary();
5978        barrier.freeze();
5979
5980        let mut bridge = BarrierTransportBridge::default();
5981        let report = bridge
5982            .broadcast_barrier(&mut server_transport, clients, barrier)
5983            .expect("barrier should broadcast");
5984
5985        assert_eq!(report.barrier_id, barrier.id);
5986        assert_eq!(report.state, BarrierState::Frozen);
5987        assert_eq!(report.server_tick, Tick::new(12));
5988        assert_eq!(report.clients_requested, 2);
5989        assert_eq!(report.clients_sent, 2);
5990        assert!(report.bytes_sent > 0);
5991        assert_eq!(bridge.stats().notifications_sent, 2);
5992        assert_eq!(bridge.stats().clients_notified, 2);
5993        assert_eq!(bridge.stats().bytes_sent, report.bytes_sent);
5994
5995        for (index, client_id) in clients.into_iter().enumerate() {
5996            let mut client_bridge = ClientTransportBridge::new(
5997                ClientTransportConfig::new(client_id, server_id).with_expected_source(server_id),
5998            );
5999            let pump = client_bridge
6000                .pump(&mut client_transports[index], 2)
6001                .expect("client should receive barrier");
6002            assert_eq!(pump.barrier_frames_received(), 1);
6003            assert_eq!(
6004                pump.barriers[0],
6005                BarrierFrame {
6006                    client_id,
6007                    barrier_id: barrier.id,
6008                    server_tick: barrier.target_tick,
6009                    state: BarrierState::Frozen,
6010                }
6011            );
6012        }
6013    }
6014
6015    #[test]
6016    fn replication_transport_bridge_sends_planned_frame() {
6017        let mut station = station(1, 10);
6018        let mut index = CellIndex::new(GridSpec::new(64.0).expect("grid is valid"));
6019        let mut policies = PolicyTable::default();
6020        policies.set(CompiledSyncPolicy::new(PolicyId::new(0), 1, 20, 256.0));
6021        let handle = station
6022            .spawn_owned(
6023                EntityId::new(100),
6024                Position3::new(0.0, 0.0, 0.0),
6025                Bounds::Point,
6026                PolicyId::new(0),
6027            )
6028            .expect("spawn should work");
6029        index.upsert(handle, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
6030        let descriptor = ComponentDescriptor::sparse_blob(
6031            ComponentId::new(1),
6032            "health",
6033            ComponentSyncMode::Delta,
6034            ComponentMigrationMode::Copy,
6035            4,
6036        );
6037        let mut components = ComponentStore::default();
6038        components
6039            .set_typed(&descriptor, handle, 1, &U32LeCodec, &100)
6040            .expect("component should write");
6041        let selection = ComponentSelection {
6042            component_ids: vec![ComponentId::new(1)],
6043        };
6044        let viewer = ViewerQuery {
6045            client_id: ClientId::new(7),
6046            position: Position3::new(0.0, 0.0, 0.0),
6047            radius: 256.0,
6048            max_entities: 32,
6049        };
6050        let mut bridge = ReplicationTransportBridge::default();
6051        let mut transport = FakeTransport::default();
6052
6053        let report = bridge
6054            .send_viewer(
6055                &mut transport,
6056                &station,
6057                &index,
6058                &policies,
6059                &components,
6060                &selection,
6061                &viewer,
6062                &RangeOnlyVisibility,
6063            )
6064            .expect("replication should send");
6065
6066        assert!(report.sent);
6067        assert_eq!(report.client_id, viewer.client_id);
6068        assert_eq!(report.selected_entities, 1);
6069        assert_eq!(report.encoded_entities, 1);
6070        assert_eq!(report.encoded_components, 1);
6071        assert_eq!(transport.packets_sent(), 1);
6072        assert_eq!(transport.bytes_sent(), report.bytes_sent);
6073        assert_eq!(bridge.stats().frames_sent, 1);
6074        assert_eq!(bridge.stats().entities_selected, 1);
6075        assert_eq!(bridge.stats().components_encoded, 1);
6076        assert!(bridge.stats().packet_capacity_hint_bytes >= report.bytes_sent);
6077        assert!(bridge.stats().planning_entity_capacity_max >= 1);
6078        assert!(bridge.planning_scratch.is_some());
6079
6080        let retained_entities = bridge.planning_output.entities.as_ptr();
6081        let mut rejecting = BudgetedTransport::new(FakeTransport::default(), 0, 0);
6082        bridge
6083            .send_viewer(
6084                &mut rejecting,
6085                &station,
6086                &index,
6087                &policies,
6088                &components,
6089                &selection,
6090                &viewer,
6091                &RangeOnlyVisibility,
6092            )
6093            .expect_err("transport budget should reject the packet");
6094        assert_eq!(bridge.planning_output.entities.as_ptr(), retained_entities);
6095    }
6096
6097    #[test]
6098    fn replication_transport_bridge_skips_empty_frames_by_default() {
6099        let mut station = station(1, 10);
6100        let mut index = CellIndex::new(GridSpec::new(64.0).expect("grid is valid"));
6101        let mut policies = PolicyTable::default();
6102        policies.set(CompiledSyncPolicy::new(PolicyId::new(0), 1, 20, 256.0));
6103        let handle = station
6104            .spawn_owned(
6105                EntityId::new(100),
6106                Position3::new(0.0, 0.0, 0.0),
6107                Bounds::Point,
6108                PolicyId::new(0),
6109            )
6110            .expect("spawn should work");
6111        index.upsert(handle, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
6112        let components = ComponentStore::default();
6113        let selection = ComponentSelection {
6114            component_ids: vec![ComponentId::new(1)],
6115        };
6116        let viewer = ViewerQuery {
6117            client_id: ClientId::new(7),
6118            position: Position3::new(0.0, 0.0, 0.0),
6119            radius: 256.0,
6120            max_entities: 32,
6121        };
6122        let mut bridge = ReplicationTransportBridge::default();
6123        let mut transport = FakeTransport::default();
6124
6125        let report = bridge
6126            .send_viewer(
6127                &mut transport,
6128                &station,
6129                &index,
6130                &policies,
6131                &components,
6132                &selection,
6133                &viewer,
6134                &RangeOnlyVisibility,
6135            )
6136            .expect("empty replication should skip");
6137
6138        assert!(!report.sent);
6139        assert_eq!(report.selected_entities, 1);
6140        assert_eq!(report.encoded_entities, 0);
6141        assert_eq!(transport.packets_sent(), 0);
6142        assert_eq!(bridge.stats().frames_skipped_empty, 1);
6143        assert_eq!(bridge.stats().entities_selected, 1);
6144    }
6145
6146    #[test]
6147    #[allow(clippy::too_many_lines)]
6148    fn replication_transport_bridge_prioritized_reports_budget_skips() {
6149        let client_id = ClientId::new(7);
6150        let server_id = ClientId::new(0);
6151        let mut station = station(1, 10);
6152        let mut index = CellIndex::new(GridSpec::new(64.0).expect("grid is valid"));
6153        let mut policies = PolicyTable::default();
6154        let mut low = CompiledSyncPolicy::new(PolicyId::new(1), 1, 20, 256.0);
6155        low.priority_weight = 1;
6156        let mut high = CompiledSyncPolicy::new(PolicyId::new(2), 1, 20, 256.0);
6157        high.priority_weight = 10;
6158        policies.set(low);
6159        policies.set(high);
6160
6161        let low_handle = station
6162            .spawn_owned(
6163                EntityId::new(100),
6164                Position3::new(0.0, 0.0, 0.0),
6165                Bounds::Point,
6166                PolicyId::new(1),
6167            )
6168            .expect("spawn low should work");
6169        let high_handle = station
6170            .spawn_owned(
6171                EntityId::new(200),
6172                Position3::new(128.0, 0.0, 0.0),
6173                Bounds::Point,
6174                PolicyId::new(2),
6175            )
6176            .expect("spawn high should work");
6177        index.upsert(low_handle, Position3::new(0.0, 0.0, 0.0), Bounds::Point);
6178        index.upsert(high_handle, Position3::new(128.0, 0.0, 0.0), Bounds::Point);
6179
6180        let descriptor = ComponentDescriptor::sparse_blob(
6181            ComponentId::new(1),
6182            "health",
6183            ComponentSyncMode::Delta,
6184            ComponentMigrationMode::Copy,
6185            4,
6186        );
6187        let mut components = ComponentStore::default();
6188        components
6189            .set_typed(&descriptor, low_handle, 1, &U32LeCodec, &100)
6190            .expect("low component should write");
6191        components
6192            .set_typed(&descriptor, high_handle, 1, &U32LeCodec, &200)
6193            .expect("high component should write");
6194        let selection = ComponentSelection {
6195            component_ids: vec![ComponentId::new(1)],
6196        };
6197        let viewer = ViewerQuery {
6198            client_id,
6199            position: Position3::new(0.0, 0.0, 0.0),
6200            radius: 256.0,
6201            max_entities: 1,
6202        };
6203        let hub = InMemoryTransportHub::new(ClientTransportLimits {
6204            max_queued_packets_per_client: 4,
6205            max_packet_bytes: 512,
6206        });
6207        let mut client_transport = hub
6208            .endpoint(client_id, "127.0.0.1:23107".parse().expect("client addr"))
6209            .expect("client endpoint should register");
6210        let mut server_transport = hub
6211            .endpoint(server_id, "127.0.0.1:23100".parse().expect("server addr"))
6212            .expect("server endpoint should register");
6213        let mut bridge = ReplicationTransportBridge::new(
6214            ReplicationTransportConfig {
6215                budget: ReplicationBudget {
6216                    max_entities: 1,
6217                    max_bytes: 128,
6218                    estimated_entity_bytes: 32,
6219                },
6220                send_empty_frames: false,
6221            },
6222            ReplicationFrameBuilder::default(),
6223        );
6224
6225        let report = bridge
6226            .send_viewer_prioritized(
6227                &mut server_transport,
6228                &station,
6229                &index,
6230                &policies,
6231                &components,
6232                &selection,
6233                &viewer,
6234                &RangeOnlyVisibility,
6235            )
6236            .expect("prioritized replication should send");
6237
6238        assert!(report.sent);
6239        assert_eq!(report.selected_entities, 1);
6240        assert_eq!(report.skipped_by_budget, 1);
6241        assert_eq!(bridge.stats().entities_skipped_by_budget, 1);
6242
6243        let packet = client_transport
6244            .try_recv()
6245            .expect("receive should work")
6246            .expect("packet should exist");
6247        let RuntimeFrame::Replication(frame) = BinaryFrameDecoder
6248            .decode(&packet.bytes)
6249            .expect("frame decodes")
6250        else {
6251            panic!("expected replication frame");
6252        };
6253        assert_eq!(frame.entities.len(), 1);
6254        assert_eq!(frame.entities[0].entity_id, EntityId::new(200));
6255    }
6256
6257    #[test]
6258    fn replication_receive_bridge_decodes_target_frames() {
6259        let client_id = ClientId::new(7);
6260        let server_id = ClientId::new(0);
6261        let hub = InMemoryTransportHub::new(ClientTransportLimits {
6262            max_queued_packets_per_client: 4,
6263            max_packet_bytes: 512,
6264        });
6265        let mut client_transport = hub
6266            .endpoint(client_id, "127.0.0.1:23007".parse().expect("client addr"))
6267            .expect("client endpoint should register");
6268        let mut server_transport = hub
6269            .endpoint(server_id, "127.0.0.1:23000".parse().expect("server addr"))
6270            .expect("server endpoint should register");
6271        let frame = ReplicationFrame {
6272            client_id,
6273            server_tick: Tick::new(12),
6274            entity_count: 1,
6275            estimated_payload_bytes: 4,
6276            entities: vec![EntityDelta {
6277                entity_id: EntityId::new(100),
6278                owner_epoch: OwnerEpoch::new(1),
6279                components: vec![ComponentDelta {
6280                    component_id: ComponentId::new(1),
6281                    version: 1,
6282                    flags: 0,
6283                    bytes: 100_u32.to_le_bytes().to_vec(),
6284                }],
6285            }],
6286        };
6287        let mut bytes = Vec::new();
6288        BinaryFrameEncoder
6289            .encode_replication(&frame, &mut bytes)
6290            .expect("replication should encode");
6291        server_transport
6292            .send(OutboundPacket {
6293                client_id,
6294                bytes: bytes.clone(),
6295            })
6296            .expect("replication packet should send");
6297
6298        let mut receive = ReplicationReceiveBridge::new(
6299            ReplicationReceiveConfig::new(client_id).with_expected_source(server_id),
6300        );
6301        let pump = receive
6302            .pump(&mut client_transport, 4)
6303            .expect("replication packet should receive");
6304
6305        assert_eq!(pump.frames_received(), 1);
6306        assert_eq!(pump.entities_received(), 1);
6307        assert_eq!(pump.components_received(), 1);
6308        assert_eq!(pump.frames[0].client_id, client_id);
6309        assert_eq!(receive.stats().packets_received, 1);
6310        assert_eq!(receive.stats().frames_received, 1);
6311        assert_eq!(receive.stats().entities_received, 1);
6312        assert_eq!(receive.stats().components_received, 1);
6313        assert!(receive.stats().bytes_received > 0);
6314
6315        server_transport
6316            .send(OutboundPacket {
6317                client_id,
6318                bytes: bytes.clone(),
6319            })
6320            .expect("visitor packet should send");
6321        let mut visited_payload = 0_u32;
6322        let visit = receive
6323            .pump_visit(&mut client_transport, 4, |borrowed| {
6324                assert_eq!(borrowed.client_id, client_id);
6325                for entity in borrowed.entities() {
6326                    assert_eq!(entity.entity_id, EntityId::new(100));
6327                    for component in entity.components() {
6328                        visited_payload = u32::from_le_bytes(
6329                            component
6330                                .bytes
6331                                .try_into()
6332                                .expect("health payload should be four bytes"),
6333                        );
6334                    }
6335                }
6336                Ok::<_, core::convert::Infallible>(())
6337            })
6338            .expect("borrowed replication packet should visit");
6339        assert_eq!(visited_payload, 100);
6340        assert_eq!(visit.packets_received, 1);
6341        assert_eq!(visit.frames_received, 1);
6342        assert_eq!(visit.entities_received, 1);
6343        assert_eq!(visit.components_received, 1);
6344
6345        server_transport
6346            .send(OutboundPacket { client_id, bytes })
6347            .expect("visitor failure packet should send");
6348        let error = receive
6349            .pump_visit(&mut client_transport, 4, |_| Err("apply failed"))
6350            .expect_err("visitor failure should surface separately");
6351        assert!(matches!(
6352            error,
6353            ReplicationReceiveVisitError::Visitor("apply failed")
6354        ));
6355        assert_eq!(receive.stats().packets_received, 3);
6356        assert_eq!(receive.stats().frames_received, 3);
6357        assert_eq!(receive.stats().entities_received, 3);
6358        assert_eq!(receive.stats().components_received, 3);
6359    }
6360
6361    #[test]
6362    fn replication_receive_bridge_rejects_wrong_target() {
6363        let client_id = ClientId::new(7);
6364        let server_id = ClientId::new(0);
6365        let wrong_client_id = ClientId::new(99);
6366        let hub = InMemoryTransportHub::new(ClientTransportLimits {
6367            max_queued_packets_per_client: 4,
6368            max_packet_bytes: 512,
6369        });
6370        let mut client_transport = hub
6371            .endpoint(client_id, "127.0.0.1:23107".parse().expect("client addr"))
6372            .expect("client endpoint should register");
6373        let mut server_transport = hub
6374            .endpoint(server_id, "127.0.0.1:23100".parse().expect("server addr"))
6375            .expect("server endpoint should register");
6376        let frame = ReplicationFrame {
6377            client_id: wrong_client_id,
6378            server_tick: Tick::new(12),
6379            entity_count: 1,
6380            estimated_payload_bytes: 4,
6381            entities: vec![EntityDelta {
6382                entity_id: EntityId::new(100),
6383                owner_epoch: OwnerEpoch::new(1),
6384                components: vec![ComponentDelta {
6385                    component_id: ComponentId::new(1),
6386                    version: 1,
6387                    flags: 0,
6388                    bytes: 100_u32.to_le_bytes().to_vec(),
6389                }],
6390            }],
6391        };
6392        let mut bytes = Vec::new();
6393        BinaryFrameEncoder
6394            .encode_replication(&frame, &mut bytes)
6395            .expect("replication should encode");
6396        server_transport
6397            .send(OutboundPacket { client_id, bytes })
6398            .expect("replication packet should send");
6399
6400        let mut receive = ReplicationReceiveBridge::new(
6401            ReplicationReceiveConfig::new(client_id).with_expected_source(server_id),
6402        );
6403        let error = receive
6404            .pump(&mut client_transport, 4)
6405            .expect_err("wrong target should be rejected");
6406
6407        assert!(matches!(
6408            error,
6409            ReplicationReceiveError::TargetMismatch {
6410                expected,
6411                actual,
6412            } if expected == client_id && actual == wrong_client_id
6413        ));
6414        assert_eq!(receive.stats().packets_received, 1);
6415        assert_eq!(receive.stats().frames_received, 0);
6416        assert_eq!(receive.stats().frames_rejected_target, 1);
6417    }
6418
6419    #[test]
6420    #[allow(clippy::too_many_lines)]
6421    fn client_transport_bridge_sends_command_and_receives_client_frames() {
6422        let client_id = ClientId::new(7);
6423        let server_id = ClientId::new(0);
6424        let hub = InMemoryTransportHub::new(ClientTransportLimits {
6425            max_queued_packets_per_client: 8,
6426            max_packet_bytes: 512,
6427        });
6428        let mut client_transport = hub
6429            .endpoint(client_id, "127.0.0.1:23207".parse().expect("client addr"))
6430            .expect("client endpoint should register");
6431        let mut server_transport = hub
6432            .endpoint(server_id, "127.0.0.1:23200".parse().expect("server addr"))
6433            .expect("server endpoint should register");
6434        let mut bridge = ClientTransportBridge::new(
6435            ClientTransportConfig::new(client_id, server_id).with_expected_source(server_id),
6436        );
6437        let command = CommandFrame {
6438            client_id,
6439            command_id: CommandId::new(42),
6440            entity_id: EntityId::new(100),
6441            sequence: 9,
6442            kind: 1,
6443            priority: CommandPriority::High,
6444            payload: b"move:north".to_vec(),
6445        };
6446
6447        let send = bridge
6448            .send_command_frame(&mut client_transport, &command)
6449            .expect("command should send");
6450        assert_eq!(send.command_id, command.command_id);
6451        assert!(send.bytes_sent > 0);
6452        assert_eq!(bridge.stats().commands_sent, 1);
6453        assert_eq!(bridge.stats().command_bytes_sent, send.bytes_sent);
6454        let inbound = server_transport
6455            .try_recv()
6456            .expect("server receive should work")
6457            .expect("command packet should arrive");
6458        assert_eq!(inbound.client_id, Some(client_id));
6459        let RuntimeFrame::Command(decoded) = BinaryFrameDecoder
6460            .decode(&inbound.bytes)
6461            .expect("command should decode")
6462        else {
6463            panic!("expected command frame");
6464        };
6465        assert_eq!(decoded, command);
6466
6467        let ack = CommandAckFrame {
6468            client_id,
6469            command_id: command.command_id,
6470            server_tick: Tick::new(12),
6471            accepted: true,
6472            reason_code: GATEWAY_COMMAND_ACK_ACCEPTED,
6473        };
6474        let mut ack_bytes = Vec::new();
6475        BinaryFrameEncoder
6476            .encode_command_ack(&ack, &mut ack_bytes)
6477            .expect("ACK should encode");
6478        server_transport
6479            .send(OutboundPacket {
6480                client_id,
6481                bytes: ack_bytes.clone(),
6482            })
6483            .expect("ACK should send");
6484
6485        let replication = ReplicationFrame {
6486            client_id,
6487            server_tick: Tick::new(12),
6488            entity_count: 1,
6489            estimated_payload_bytes: 4,
6490            entities: vec![EntityDelta {
6491                entity_id: EntityId::new(100),
6492                owner_epoch: OwnerEpoch::new(1),
6493                components: vec![ComponentDelta {
6494                    component_id: ComponentId::new(1),
6495                    version: 1,
6496                    flags: 0,
6497                    bytes: 100_u32.to_le_bytes().to_vec(),
6498                }],
6499            }],
6500        };
6501        let mut replication_bytes = Vec::new();
6502        BinaryFrameEncoder
6503            .encode_replication(&replication, &mut replication_bytes)
6504            .expect("replication should encode");
6505        server_transport
6506            .send(OutboundPacket {
6507                client_id,
6508                bytes: replication_bytes.clone(),
6509            })
6510            .expect("replication should send");
6511
6512        let barrier = BarrierFrame {
6513            client_id,
6514            barrier_id: BarrierId::new(5),
6515            server_tick: Tick::new(12),
6516            state: BarrierState::Frozen,
6517        };
6518        let mut barrier_bytes = Vec::new();
6519        BinaryFrameEncoder
6520            .encode_barrier(&barrier, &mut barrier_bytes)
6521            .expect("barrier should encode");
6522        server_transport
6523            .send(OutboundPacket {
6524                client_id,
6525                bytes: barrier_bytes.clone(),
6526            })
6527            .expect("barrier should send");
6528
6529        let pump = bridge
6530            .pump(&mut client_transport, 8)
6531            .expect("client frames should receive");
6532
6533        assert_eq!(pump.packets_received, 3);
6534        assert_eq!(pump.command_acks_received(), 1);
6535        assert_eq!(pump.replication_frames_received(), 1);
6536        assert_eq!(pump.barrier_frames_received(), 1);
6537        assert_eq!(pump.entities_received(), 1);
6538        assert_eq!(pump.components_received(), 1);
6539        assert_eq!(pump.command_acks[0], ack);
6540        assert_eq!(pump.replication_frames[0], replication);
6541        assert_eq!(pump.barriers[0], barrier);
6542        assert_eq!(bridge.stats().packets_received, 3);
6543        assert_eq!(bridge.stats().command_acks_received, 1);
6544        assert_eq!(bridge.stats().replication_frames_received, 1);
6545        assert_eq!(bridge.stats().barrier_frames_received, 1);
6546        assert_eq!(bridge.stats().entities_received, 1);
6547        assert_eq!(bridge.stats().components_received, 1);
6548
6549        for bytes in [ack_bytes.clone(), replication_bytes, barrier_bytes] {
6550            server_transport
6551                .send(OutboundPacket { client_id, bytes })
6552                .expect("visitor packet should send");
6553        }
6554        let mut visited_ack = 0_usize;
6555        let mut visited_replication = 0_usize;
6556        let mut visited_barrier = 0_usize;
6557        let mut payload_checksum = 0_u64;
6558        let visit = bridge
6559            .pump_visit(&mut client_transport, 8, |frame| {
6560                match frame {
6561                    ClientInboundFrameRef::CommandAck(frame) => {
6562                        assert_eq!(frame, ack);
6563                        visited_ack = visited_ack.saturating_add(1);
6564                    }
6565                    ClientInboundFrameRef::Replication(frame) => {
6566                        assert_eq!(frame.client_id, client_id);
6567                        assert_eq!(frame.encoded_entity_count(), 1);
6568                        for entity in frame.entities() {
6569                            for component in entity.components() {
6570                                payload_checksum = payload_checksum.saturating_add(
6571                                    component.bytes.iter().map(|byte| u64::from(*byte)).sum(),
6572                                );
6573                            }
6574                        }
6575                        visited_replication = visited_replication.saturating_add(1);
6576                    }
6577                    ClientInboundFrameRef::Barrier(frame) => {
6578                        assert_eq!(frame, barrier);
6579                        visited_barrier = visited_barrier.saturating_add(1);
6580                    }
6581                }
6582                Ok::<(), &'static str>(())
6583            })
6584            .expect("mixed visitor pump should work");
6585        assert_eq!(visit.packets_received, 3);
6586        assert_eq!(visit.command_acks_received, 1);
6587        assert_eq!(visit.replication_frames_received, 1);
6588        assert_eq!(visit.barrier_frames_received, 1);
6589        assert_eq!(visit.entities_received, 1);
6590        assert_eq!(visit.components_received, 1);
6591        assert_eq!(
6592            (visited_ack, visited_replication, visited_barrier),
6593            (1, 1, 1)
6594        );
6595        assert_eq!(payload_checksum, 100);
6596        assert_eq!(bridge.stats().packets_received, 6);
6597
6598        server_transport
6599            .send(OutboundPacket {
6600                client_id,
6601                bytes: ack_bytes,
6602            })
6603            .expect("failing visitor packet should send");
6604        let visitor_error = bridge
6605            .pump_visit(&mut client_transport, 1, |_| Err("apply failed"))
6606            .expect_err("visitor failure should propagate");
6607        assert_eq!(
6608            visitor_error,
6609            ClientTransportVisitError::Visitor("apply failed")
6610        );
6611        assert_eq!(bridge.stats().packets_received, 7);
6612        assert_eq!(bridge.stats().command_acks_received, 3);
6613    }
6614
6615    #[test]
6616    fn client_transport_bridge_rejects_wrong_ack_target() {
6617        let client_id = ClientId::new(7);
6618        let server_id = ClientId::new(0);
6619        let wrong_client_id = ClientId::new(99);
6620        let hub = InMemoryTransportHub::new(ClientTransportLimits {
6621            max_queued_packets_per_client: 4,
6622            max_packet_bytes: 512,
6623        });
6624        let mut client_transport = hub
6625            .endpoint(client_id, "127.0.0.1:23307".parse().expect("client addr"))
6626            .expect("client endpoint should register");
6627        let mut server_transport = hub
6628            .endpoint(server_id, "127.0.0.1:23300".parse().expect("server addr"))
6629            .expect("server endpoint should register");
6630        let mut bridge = ClientTransportBridge::new(
6631            ClientTransportConfig::new(client_id, server_id).with_expected_source(server_id),
6632        );
6633        let ack = CommandAckFrame {
6634            client_id: wrong_client_id,
6635            command_id: CommandId::new(42),
6636            server_tick: Tick::new(12),
6637            accepted: true,
6638            reason_code: GATEWAY_COMMAND_ACK_ACCEPTED,
6639        };
6640        let mut ack_bytes = Vec::new();
6641        BinaryFrameEncoder
6642            .encode_command_ack(&ack, &mut ack_bytes)
6643            .expect("ACK should encode");
6644        server_transport
6645            .send(OutboundPacket {
6646                client_id,
6647                bytes: ack_bytes,
6648            })
6649            .expect("ACK should send");
6650
6651        let error = bridge
6652            .pump(&mut client_transport, 4)
6653            .expect_err("wrong target should be rejected");
6654
6655        assert!(matches!(
6656            error,
6657            ClientTransportBridgeError::TargetMismatch {
6658                kind: ClientInboundFrameKind::CommandAck,
6659                expected,
6660                actual,
6661            } if expected == client_id && actual == wrong_client_id
6662        ));
6663        assert_eq!(bridge.stats().packets_received, 1);
6664        assert_eq!(bridge.stats().command_acks_received, 0);
6665        assert_eq!(bridge.stats().frames_rejected_target, 1);
6666    }
6667
6668    #[test]
6669    #[allow(clippy::too_many_lines)]
6670    fn gateway_client_transport_bridge_queues_command_and_sends_ack() {
6671        let client_id = ClientId::new(7);
6672        let server_id = ClientId::new(0);
6673        let station_id = StationId::new(1);
6674        let hub = InMemoryTransportHub::new(ClientTransportLimits {
6675            max_queued_packets_per_client: 8,
6676            max_packet_bytes: 512,
6677        });
6678        let mut client_transport = hub
6679            .endpoint(client_id, "127.0.0.1:23507".parse().expect("client addr"))
6680            .expect("client endpoint should register");
6681        let mut server_transport = hub
6682            .endpoint(server_id, "127.0.0.1:23500".parse().expect("server addr"))
6683            .expect("server endpoint should register");
6684        let mut client_bridge = ClientTransportBridge::new(
6685            ClientTransportConfig::new(client_id, server_id).with_expected_source(server_id),
6686        );
6687        let command = CommandFrame {
6688            client_id,
6689            command_id: CommandId::new(42),
6690            entity_id: EntityId::new(100),
6691            sequence: 9,
6692            kind: 1,
6693            priority: CommandPriority::High,
6694            payload: b"move:north".to_vec(),
6695        };
6696        client_bridge
6697            .send_command_frame(&mut client_transport, &command)
6698            .expect("client command should send");
6699
6700        let mut gateway = gateway(4);
6701        gateway
6702            .connect(client_id, station_id, Tick::new(10))
6703            .expect("client should connect");
6704        let mut station_queues = BTreeMap::from([(station_id, command_queues())]);
6705        let mut pipeline = GatewayCommandPipeline::default();
6706        let mut gateway_bridge = GatewayClientTransportBridge::default();
6707
6708        let pump = gateway_bridge
6709            .pump_ingress(
6710                &mut server_transport,
6711                &mut pipeline,
6712                &mut gateway,
6713                &mut station_queues,
6714                Tick::new(10),
6715                CommandIngress::RUNNING,
6716                4,
6717            )
6718            .expect("gateway client transport should pump");
6719
6720        assert_eq!(pump.packets_received, 1);
6721        assert_eq!(pump.commands_processed(), 1);
6722        assert_eq!(pump.commands_accepted(), 1);
6723        assert_eq!(pump.acks_sent, 1);
6724        assert_eq!(gateway_bridge.stats().packets_received, 1);
6725        assert_eq!(gateway_bridge.stats().command_frames_received, 1);
6726        assert_eq!(gateway_bridge.stats().commands_accepted, 1);
6727        assert_eq!(gateway_bridge.stats().acks_sent, 1);
6728        let queued = station_queues
6729            .get_mut(&station_id)
6730            .expect("station queue should exist")
6731            .pop_next()
6732            .expect("command should queue");
6733        assert_eq!(queued.id, command.command_id);
6734
6735        let ack_pump = client_bridge
6736            .pump(&mut client_transport, 4)
6737            .expect("client should receive ACK");
6738        assert_eq!(ack_pump.command_acks_received(), 1);
6739        assert!(ack_pump.command_acks[0].accepted);
6740        assert_eq!(ack_pump.command_acks[0].command_id, command.command_id);
6741
6742        let compact_command = CommandFrame {
6743            command_id: CommandId::new(43),
6744            sequence: 10,
6745            ..command
6746        };
6747        client_bridge
6748            .send_command_frame(&mut client_transport, &compact_command)
6749            .expect("second client command should send");
6750        let summary = gateway_bridge
6751            .pump_ingress_compact(
6752                &mut server_transport,
6753                &mut pipeline,
6754                &mut gateway,
6755                &mut station_queues,
6756                Tick::new(11),
6757                CommandIngress::RUNNING,
6758                4,
6759            )
6760            .expect("compact gateway transport should pump");
6761        assert_eq!(summary.packets_received, 1);
6762        assert_eq!(summary.commands_accepted, 1);
6763        assert_eq!(summary.commands_rejected, 0);
6764        assert_eq!(summary.acks_sent, 1);
6765        assert!(summary.ack_bytes_sent > 0);
6766        let compact_queued = station_queues
6767            .get_mut(&station_id)
6768            .expect("station queue should exist")
6769            .pop_next()
6770            .expect("compact command should queue");
6771        assert_eq!(compact_queued.id, compact_command.command_id);
6772        let compact_ack = client_bridge
6773            .pump(&mut client_transport, 4)
6774            .expect("client should receive compact ACK");
6775        assert_eq!(compact_ack.command_acks_received(), 1);
6776        assert_eq!(
6777            compact_ack.command_acks[0].command_id,
6778            compact_command.command_id
6779        );
6780        assert_eq!(gateway_bridge.stats().packets_received, 2);
6781        assert_eq!(gateway_bridge.stats().commands_accepted, 2);
6782        assert_eq!(gateway_bridge.stats().acks_sent, 2);
6783    }
6784
6785    #[test]
6786    fn gateway_client_transport_bridge_rejects_source_mismatch_before_admission() {
6787        let packet_client_id = ClientId::new(7);
6788        let frame_client_id = ClientId::new(8);
6789        let server_id = ClientId::new(0);
6790        let station_id = StationId::new(1);
6791        let hub = InMemoryTransportHub::new(ClientTransportLimits {
6792            max_queued_packets_per_client: 4,
6793            max_packet_bytes: 512,
6794        });
6795        let mut packet_client_transport = hub
6796            .endpoint(
6797                packet_client_id,
6798                "127.0.0.1:23607".parse().expect("client addr"),
6799            )
6800            .expect("client endpoint should register");
6801        let mut server_transport = hub
6802            .endpoint(server_id, "127.0.0.1:23600".parse().expect("server addr"))
6803            .expect("server endpoint should register");
6804        let command = CommandFrame {
6805            client_id: frame_client_id,
6806            command_id: CommandId::new(42),
6807            entity_id: EntityId::new(100),
6808            sequence: 9,
6809            kind: 1,
6810            priority: CommandPriority::High,
6811            payload: b"move:north".to_vec(),
6812        };
6813        let mut bytes = Vec::new();
6814        BinaryFrameEncoder
6815            .encode_command(&command, &mut bytes)
6816            .expect("command should encode");
6817        packet_client_transport
6818            .send(OutboundPacket {
6819                client_id: server_id,
6820                bytes,
6821            })
6822            .expect("packet should send");
6823
6824        let mut gateway = gateway(4);
6825        gateway
6826            .connect(frame_client_id, station_id, Tick::new(10))
6827            .expect("frame client should connect");
6828        let mut station_queues = BTreeMap::from([(station_id, command_queues())]);
6829        let mut pipeline = GatewayCommandPipeline::default();
6830        let mut gateway_bridge = GatewayClientTransportBridge::default();
6831
6832        let error = gateway_bridge
6833            .pump_ingress(
6834                &mut server_transport,
6835                &mut pipeline,
6836                &mut gateway,
6837                &mut station_queues,
6838                Tick::new(10),
6839                CommandIngress::RUNNING,
6840                4,
6841            )
6842            .expect_err("source mismatch should reject before admission");
6843
6844        assert!(matches!(
6845            error,
6846            GatewayClientTransportError::SourceMismatch {
6847                packet_client_id: actual_packet,
6848                frame_client_id: actual_frame,
6849            } if actual_packet == packet_client_id && actual_frame == frame_client_id
6850        ));
6851        assert_eq!(gateway_bridge.stats().source_mismatches, 1);
6852        assert_eq!(gateway_bridge.stats().commands_accepted, 0);
6853        assert_eq!(pipeline.stats().commands_admitted, 0);
6854        assert_eq!(
6855            station_queues
6856                .get(&station_id)
6857                .expect("station queue should exist")
6858                .total_len(),
6859            0
6860        );
6861    }
6862
6863    #[test]
6864    fn gateway_command_pipeline_queues_command_and_encodes_ack() {
6865        let client_id = ClientId::new(7);
6866        let station_id = StationId::new(1);
6867        let mut gateway = gateway(4);
6868        gateway
6869            .connect(client_id, station_id, Tick::new(10))
6870            .expect("client should connect");
6871        let mut station_queues = BTreeMap::from([(station_id, command_queues())]);
6872        let mut pipeline = GatewayCommandPipeline::default();
6873
6874        let report = pipeline.process(
6875            &mut gateway,
6876            &mut station_queues,
6877            &encode_command_frame(1),
6878            Tick::new(10),
6879            CommandIngress::RUNNING,
6880        );
6881
6882        assert!(report.accepted);
6883        assert_eq!(report.reason_code, GATEWAY_COMMAND_ACK_ACCEPTED);
6884        assert_eq!(report.station_id, Some(station_id));
6885        assert!(report.error.is_none());
6886        let ack_bytes = report.ack_bytes.expect("ACK should encode");
6887        let RuntimeFrame::CommandAck(ack) = BinaryFrameDecoder
6888            .decode(&ack_bytes)
6889            .expect("ACK should decode")
6890        else {
6891            panic!("expected command ACK");
6892        };
6893        assert!(ack.accepted);
6894        assert_eq!(ack.command_id, CommandId::new(1));
6895        let queued = station_queues
6896            .get_mut(&station_id)
6897            .expect("queue should exist")
6898            .pop_next()
6899            .expect("command should queue");
6900        assert_eq!(queued.id, CommandId::new(1));
6901        assert_eq!(pipeline.stats().commands_admitted, 1);
6902        assert_eq!(pipeline.stats().commands_enqueued, 1);
6903        assert_eq!(pipeline.stats().acks_encoded, 1);
6904    }
6905
6906    #[test]
6907    fn gateway_command_pipeline_negative_acks_rate_limit() {
6908        let client_id = ClientId::new(7);
6909        let station_id = StationId::new(1);
6910        let mut gateway = gateway(1);
6911        gateway
6912            .connect(client_id, station_id, Tick::new(10))
6913            .expect("client should connect");
6914        let mut station_queues = BTreeMap::from([(station_id, command_queues())]);
6915        let mut pipeline = GatewayCommandPipeline::default();
6916
6917        assert!(
6918            pipeline
6919                .process(
6920                    &mut gateway,
6921                    &mut station_queues,
6922                    &encode_command_frame(1),
6923                    Tick::new(10),
6924                    CommandIngress::RUNNING,
6925                )
6926                .accepted
6927        );
6928        let rejected = pipeline.process(
6929            &mut gateway,
6930            &mut station_queues,
6931            &encode_command_frame(2),
6932            Tick::new(10),
6933            CommandIngress::RUNNING,
6934        );
6935
6936        assert!(!rejected.accepted);
6937        assert_eq!(rejected.reason_code, GATEWAY_COMMAND_ACK_RATE_LIMITED);
6938        assert!(matches!(
6939            rejected.error,
6940            Some(GatewayCommandPipelineError::Gateway(
6941                GatewayError::RateLimited { .. }
6942            ))
6943        ));
6944        let RuntimeFrame::CommandAck(ack) = BinaryFrameDecoder
6945            .decode(&rejected.ack_bytes.expect("rejection ACK should encode"))
6946            .expect("ACK should decode")
6947        else {
6948            panic!("expected command ACK");
6949        };
6950        assert!(!ack.accepted);
6951        assert_eq!(ack.reason_code, GATEWAY_COMMAND_ACK_RATE_LIMITED);
6952        assert_eq!(pipeline.stats().commands_rejected_gateway, 1);
6953    }
6954
6955    #[test]
6956    fn gateway_command_pipeline_rejects_missing_station_queue() {
6957        let client_id = ClientId::new(7);
6958        let station_id = StationId::new(1);
6959        let mut gateway = gateway(4);
6960        gateway
6961            .connect(client_id, station_id, Tick::new(10))
6962            .expect("client should connect");
6963        let mut station_queues = BTreeMap::new();
6964        let mut pipeline = GatewayCommandPipeline::default();
6965
6966        let report = pipeline.process(
6967            &mut gateway,
6968            &mut station_queues,
6969            &encode_command_frame(1),
6970            Tick::new(10),
6971            CommandIngress::RUNNING,
6972        );
6973
6974        assert!(!report.accepted);
6975        assert_eq!(report.station_id, Some(station_id));
6976        assert_eq!(report.reason_code, GATEWAY_COMMAND_ACK_MISSING_QUEUE);
6977        assert!(matches!(
6978            report.error,
6979            Some(GatewayCommandPipelineError::MissingQueue(id)) if id == station_id
6980        ));
6981        assert_eq!(pipeline.stats().commands_admitted, 1);
6982        assert_eq!(pipeline.stats().commands_rejected_queue, 1);
6983    }
6984
6985    #[test]
6986    fn gateway_command_pipeline_dispatches_to_deployment_route() {
6987        let client_id = ClientId::new(7);
6988        let station_id = StationId::new(1);
6989        let node_id = NodeId::new(9);
6990        let mut gateway = gateway(4);
6991        gateway
6992            .connect(client_id, station_id, Tick::new(10))
6993            .expect("client should connect");
6994        let mut deployment = DeploymentRouteTable::new(DeploymentConfig {
6995            max_nodes: 4,
6996            max_stations_per_node: 4,
6997            stale_after_ticks: 10,
6998        });
6999        deployment
7000            .register_node(node_id, 4, Tick::new(10))
7001            .expect("node should register");
7002        deployment
7003            .assign_station(station_id, node_id, Tick::new(10))
7004            .expect("station should assign");
7005        let mut pipeline = GatewayCommandPipeline::default();
7006
7007        let report = pipeline.dispatch(
7008            &mut gateway,
7009            &deployment,
7010            &encode_command_frame(1),
7011            Tick::new(12),
7012        );
7013
7014        assert!(report.accepted);
7015        assert_eq!(report.station_id, Some(station_id));
7016        assert_eq!(report.node_id, Some(node_id));
7017        let delivery = report.delivery.expect("delivery should resolve");
7018        assert_eq!(delivery.client_id, client_id);
7019        assert_eq!(delivery.station_id, station_id);
7020        assert_eq!(delivery.node_id, node_id);
7021        assert_eq!(delivery.station_route_epoch, 1);
7022        assert_eq!(
7023            report
7024                .command
7025                .expect("command should be returned")
7026                .received_at,
7027            Tick::new(12)
7028        );
7029        let RuntimeFrame::CommandAck(ack) = BinaryFrameDecoder
7030            .decode(&report.ack_bytes.expect("ACK should encode"))
7031            .expect("ACK should decode")
7032        else {
7033            panic!("expected command ACK");
7034        };
7035        assert!(ack.accepted);
7036        assert_eq!(pipeline.stats().commands_routed_deployment, 1);
7037    }
7038
7039    #[test]
7040    fn gateway_command_pipeline_negative_acks_missing_deployment_route() {
7041        let client_id = ClientId::new(7);
7042        let station_id = StationId::new(1);
7043        let mut gateway = gateway(4);
7044        gateway
7045            .connect(client_id, station_id, Tick::new(10))
7046            .expect("client should connect");
7047        let deployment = DeploymentRouteTable::default();
7048        let mut pipeline = GatewayCommandPipeline::default();
7049
7050        let report = pipeline.dispatch(
7051            &mut gateway,
7052            &deployment,
7053            &encode_command_frame(1),
7054            Tick::new(12),
7055        );
7056
7057        assert!(!report.accepted);
7058        assert_eq!(report.station_id, Some(station_id));
7059        assert_eq!(report.reason_code, GATEWAY_COMMAND_ACK_DEPLOYMENT_REJECTED);
7060        assert!(matches!(
7061            report.error,
7062            Some(GatewayCommandPipelineError::Deployment(
7063                DeploymentError::MissingStation(id)
7064            )) if id == station_id
7065        ));
7066        let RuntimeFrame::CommandAck(ack) = BinaryFrameDecoder
7067            .decode(&report.ack_bytes.expect("rejection ACK should encode"))
7068            .expect("ACK should decode")
7069        else {
7070            panic!("expected command ACK");
7071        };
7072        assert!(!ack.accepted);
7073        assert_eq!(ack.reason_code, GATEWAY_COMMAND_ACK_DEPLOYMENT_REJECTED);
7074        assert_eq!(pipeline.stats().commands_rejected_deployment, 1);
7075    }
7076
7077    #[test]
7078    fn gateway_command_pipeline_rejects_non_command_frame() {
7079        let ack = CommandAckFrame {
7080            client_id: ClientId::new(7),
7081            command_id: CommandId::new(1),
7082            server_tick: Tick::new(10),
7083            accepted: true,
7084            reason_code: 0,
7085        };
7086        let mut bytes = Vec::new();
7087        BinaryFrameEncoder
7088            .encode_command_ack(&ack, &mut bytes)
7089            .expect("ACK should encode");
7090        let mut gateway = gateway(4);
7091        let mut station_queues = BTreeMap::new();
7092        let mut pipeline = GatewayCommandPipeline::default();
7093
7094        let report = pipeline.process(
7095            &mut gateway,
7096            &mut station_queues,
7097            &bytes,
7098            Tick::new(10),
7099            CommandIngress::RUNNING,
7100        );
7101
7102        assert!(!report.accepted);
7103        assert!(report.ack_bytes.is_none());
7104        assert_eq!(
7105            report.error,
7106            Some(GatewayCommandPipelineError::NonCommandFrame)
7107        );
7108        assert_eq!(pipeline.stats().frames_rejected_non_command, 1);
7109    }
7110
7111    #[test]
7112    fn migration_executor_moves_owner_and_leaves_source_ghost() {
7113        let mut stations = StationSet::default();
7114        let mut source = station(1, 10);
7115        source
7116            .spawn_owned(
7117                EntityId::new(99),
7118                Position3::new(1.0, 2.0, 3.0),
7119                Bounds::Point,
7120                PolicyId::new(0),
7121            )
7122            .expect("spawn should work");
7123        stations.push(source);
7124        stations.push(station(2, 10));
7125
7126        let report = EntityMigrationExecutor::migrate_entity(
7127            &mut stations,
7128            EntityId::new(99),
7129            StationId::new(1),
7130            StationId::new(2),
7131            4,
7132        )
7133        .expect("migration should work");
7134
7135        assert_eq!(report.transfer.target_station, StationId::new(2));
7136        assert!(
7137            !stations
7138                .get(StationId::new(1))
7139                .expect("source")
7140                .get_by_id(EntityId::new(99))
7141                .expect("source ghost")
7142                .is_owned()
7143        );
7144        assert!(
7145            stations
7146                .get(StationId::new(2))
7147                .expect("target")
7148                .get_by_id(EntityId::new(99))
7149                .expect("target owner")
7150                .is_owned()
7151        );
7152    }
7153
7154    #[test]
7155    fn event_router_delays_until_target_tick_and_scheduler_drains() {
7156        let mut stations = StationSet::default();
7157        stations.push(station(1, 10));
7158        stations.push(station(2, 10));
7159
7160        let mut router = EventRouter::default();
7161        router.register_stations(&stations);
7162        router
7163            .route(StationEvent {
7164                id: EventId::new(1),
7165                source: StationId::new(1),
7166                target: StationId::new(2),
7167                source_tick: Tick::new(0),
7168                target_tick: Tick::new(2),
7169                priority: EventPriority::Critical,
7170                kind: EventKind::Custom(7),
7171            })
7172            .expect("route should work");
7173
7174        let mut scheduler = StationScheduler::default();
7175        let mut drained = Vec::new();
7176        scheduler.advance_all(&mut stations);
7177        scheduler
7178            .drain_ready_events_into(&stations, &mut router, &mut drained)
7179            .expect("drain should work");
7180        assert!(drained.is_empty());
7181
7182        scheduler.advance_all(&mut stations);
7183        scheduler
7184            .drain_ready_events_into(&stations, &mut router, &mut drained)
7185            .expect("drain should work");
7186        assert_eq!(drained.len(), 1);
7187        let retained_capacity = drained.capacity();
7188
7189        scheduler
7190            .drain_ready_events_into(&stations, &mut router, &mut drained)
7191            .expect("empty drain should work");
7192        assert!(drained.is_empty());
7193        assert_eq!(drained.capacity(), retained_capacity);
7194        assert_eq!(router.stats().routed_events, 1);
7195        assert_eq!(router.stats().drained_events, 1);
7196    }
7197
7198    #[test]
7199    fn event_router_unregisters_station_and_discards_queued_events() {
7200        let station_id = StationId::new(2);
7201        let mut router = EventRouter::default();
7202        router.register_station(station_id);
7203        router
7204            .route(StationEvent {
7205                id: EventId::new(1),
7206                source: StationId::new(1),
7207                target: station_id,
7208                source_tick: Tick::new(0),
7209                target_tick: Tick::new(10),
7210                priority: EventPriority::Important,
7211                kind: EventKind::Custom(1),
7212            })
7213            .expect("event should queue");
7214
7215        assert_eq!(router.unregister_station(station_id), Some(1));
7216        assert_eq!(router.unregister_station(station_id), None);
7217        assert_eq!(router.queued_len(station_id), None);
7218        assert_eq!(
7219            router.drain_ready(station_id, Tick::new(10)),
7220            Err(EventRouterError::MissingTarget(station_id))
7221        );
7222    }
7223
7224    #[test]
7225    fn station_scheduler_prioritizes_loaded_stations_with_budget() {
7226        let mut stations = StationSet::default();
7227        stations.push(station(1, 10));
7228        stations.push(station(2, 10));
7229        stations.push(station(3, 10));
7230
7231        let samples = vec![
7232            StationLoadSample {
7233                station_id: StationId::new(1),
7234                owned_entities: 1,
7235                ..StationLoadSample::default()
7236            },
7237            StationLoadSample {
7238                station_id: StationId::new(2),
7239                owned_entities: 100,
7240                subscribers: 40,
7241                queued_events: 20,
7242                tick_cost_units: 500,
7243                cells: vec![CellLoadSample {
7244                    cell: CellCoord3::new(0, 0, 0),
7245                    owned_entities: 90,
7246                    subscribers: 40,
7247                    event_pressure: 10,
7248                    ..CellLoadSample::default()
7249                }],
7250                ..StationLoadSample::default()
7251            },
7252            StationLoadSample {
7253                station_id: StationId::new(3),
7254                owned_entities: 25,
7255                subscribers: 10,
7256                queued_events: 5,
7257                tick_cost_units: 50,
7258                ..StationLoadSample::default()
7259            },
7260        ];
7261
7262        let mut scheduler = StationScheduler::default();
7263        let plan = scheduler.advance_loaded(
7264            &mut stations,
7265            &samples,
7266            StationScheduleConfig {
7267                max_station_advances_per_step: 2,
7268            },
7269        );
7270
7271        assert_eq!(plan.candidates_considered, 3);
7272        assert_eq!(plan.stations_selected, 2);
7273        assert_eq!(plan.total_advances, 2);
7274        assert_eq!(
7275            plan.selected
7276                .iter()
7277                .map(|candidate| candidate.station_id)
7278                .collect::<Vec<_>>(),
7279            vec![StationId::new(2), StationId::new(3)]
7280        );
7281        assert_eq!(scheduler.advanced_ticks, 2);
7282        assert_eq!(
7283            stations.get(StationId::new(1)).expect("station").tick(),
7284            Tick::new(0)
7285        );
7286        assert_eq!(
7287            stations.get(StationId::new(2)).expect("station").tick(),
7288            Tick::new(1)
7289        );
7290        assert_eq!(
7291            stations.get(StationId::new(3)).expect("station").tick(),
7292            Tick::new(1)
7293        );
7294    }
7295
7296    #[test]
7297    fn station_scheduler_top_k_matches_full_sort_for_budget_edges() {
7298        let candidates = (0_u32..257)
7299            .map(|index| StationScheduleCandidate {
7300                station_id: StationId::new(index),
7301                load_score: u64::from(index.wrapping_mul(37) % 23),
7302                tick_lag: u64::from(index.wrapping_mul(19) % 11),
7303            })
7304            .collect::<Vec<_>>();
7305
7306        for requested in [0, 1, 7, 64, 128, 129, 256, 257, 300] {
7307            let limit = requested.min(candidates.len());
7308            let mut expected = candidates.clone();
7309            expected.sort_by(compare_station_schedule_candidates);
7310            expected.truncate(limit);
7311            let mut actual = candidates.clone();
7312            prioritize_station_candidates(&mut actual, limit);
7313
7314            assert_eq!(&actual[..limit], expected.as_slice());
7315        }
7316    }
7317
7318    #[test]
7319    fn station_schedule_scratch_reuses_capacity_and_last_sample_wins() {
7320        let mut stations = StationSet::default();
7321        for station_id in 1..=8 {
7322            stations.push(station(station_id, 10));
7323        }
7324        let samples = [
7325            StationLoadSample {
7326                station_id: StationId::new(3),
7327                owned_entities: 1,
7328                ..StationLoadSample::default()
7329            },
7330            StationLoadSample {
7331                station_id: StationId::new(3),
7332                owned_entities: 500,
7333                ..StationLoadSample::default()
7334            },
7335        ];
7336        let scheduler = StationScheduler::default();
7337        let mut scratch = StationScheduleScratch::new();
7338
7339        {
7340            let plan = scheduler.plan_loaded_into(
7341                &stations,
7342                &samples,
7343                StationScheduleConfig {
7344                    max_station_advances_per_step: 2,
7345                },
7346                &mut scratch,
7347            );
7348            assert_eq!(plan.candidates_considered, 8);
7349            assert_eq!(plan.selected[0].station_id, StationId::new(3));
7350            assert_eq!(
7351                plan.selected[0].load_score,
7352                station_schedule_score(&samples[1])
7353            );
7354        }
7355        let score_capacity = scratch.score_capacity();
7356        let candidate_capacity = scratch.candidate_capacity();
7357
7358        let plan = scheduler.plan_loaded_into(
7359            &stations,
7360            &samples[..1],
7361            StationScheduleConfig {
7362                max_station_advances_per_step: 1,
7363            },
7364            &mut scratch,
7365        );
7366        assert_eq!(plan.selected.len(), 1);
7367        assert_eq!(scratch.score_capacity(), score_capacity);
7368        assert_eq!(scratch.candidate_capacity(), candidate_capacity);
7369    }
7370
7371    #[test]
7372    fn station_load_sampler_derives_cells_router_and_subscribers() {
7373        let station_id = StationId::new(1);
7374        let owner_position = Position3::new(1.0, 0.0, 0.0);
7375        let ghost_position = Position3::new(12.0, 0.0, 0.0);
7376        let policy_id = PolicyId::new(1);
7377        let mut station = station(1, 10);
7378        let owner = station
7379            .spawn_owned(EntityId::new(10), owner_position, Bounds::Point, policy_id)
7380            .expect("owner should spawn");
7381        let ghost = station.upsert_ghost(
7382            EntityId::new(20),
7383            ghost_position,
7384            Bounds::Point,
7385            policy_id,
7386            StationId::new(9),
7387            OwnerEpoch::new(3),
7388            Tick::new(30),
7389        );
7390
7391        let grid = GridSpec::new(10.0).expect("grid should build");
7392        let mut index = CellIndex::new(grid);
7393        index.upsert(owner, owner_position, Bounds::Point);
7394        index.upsert(ghost, ghost_position, Bounds::Point);
7395        let mut indexes = StationIndexSet::default();
7396        indexes.insert(station_id, index);
7397
7398        let mut stations = StationSet::default();
7399        stations.push(station);
7400        let mut router = EventRouter::default();
7401        router.register_station(station_id);
7402        for (event_id, kind) in [(1_u64, 1_u32), (2, 2)] {
7403            router
7404                .route(StationEvent {
7405                    id: EventId::new(event_id),
7406                    source: StationId::new(9),
7407                    target: station_id,
7408                    source_tick: Tick::new(0),
7409                    target_tick: Tick::new(4),
7410                    priority: EventPriority::Important,
7411                    kind: EventKind::Custom(kind),
7412                })
7413                .expect("event should queue");
7414        }
7415
7416        assert_eq!(indexes.iter().count(), 1);
7417        let load_sampler = StationLoadSampler::default();
7418        let samples = load_sampler.sample_all(
7419            &stations,
7420            &indexes,
7421            &router,
7422            &[(station_id, 2), (station_id, 3)],
7423        );
7424
7425        assert_eq!(samples.len(), 1);
7426        let sample = &samples[0];
7427        assert_eq!(sample.station_id, station_id);
7428        assert_eq!(sample.owned_entities, 1);
7429        assert_eq!(sample.ghost_entities, 1);
7430        assert_eq!(sample.subscribers, 5);
7431        assert_eq!(sample.queued_events, 2);
7432        assert_eq!(sample.estimated_bytes, 240);
7433        assert_eq!(sample.tick_cost_units, 7);
7434        assert_eq!(
7435            sample.cells,
7436            vec![
7437                CellLoadSample {
7438                    cell: grid.cell_at(owner_position),
7439                    owned_entities: 1,
7440                    ghost_entities: 0,
7441                    subscribers: 0,
7442                    estimated_updates: 1,
7443                    estimated_bytes: 48,
7444                    tick_cost_units: 3,
7445                    event_pressure: 0,
7446                },
7447                CellLoadSample {
7448                    cell: grid.cell_at(ghost_position),
7449                    owned_entities: 0,
7450                    ghost_entities: 1,
7451                    subscribers: 0,
7452                    estimated_updates: 1,
7453                    estimated_bytes: 48,
7454                    tick_cost_units: 2,
7455                    event_pressure: 0,
7456                },
7457            ]
7458        );
7459
7460        assert_load_sampler_scratch_reuse(
7461            &load_sampler,
7462            &stations,
7463            &indexes,
7464            &router,
7465            station_id,
7466            &samples,
7467        );
7468    }
7469
7470    fn assert_load_sampler_scratch_reuse(
7471        load_sampler: &StationLoadSampler,
7472        stations: &StationSet,
7473        indexes: &StationIndexSet,
7474        router: &EventRouter,
7475        station_id: StationId,
7476        samples: &[StationLoadSample],
7477    ) {
7478        let mut scratch = StationLoadSamplerScratch::new();
7479        let (sample_ptr, cell_ptr) = {
7480            let reused = load_sampler.sample_all_into(
7481                stations,
7482                indexes,
7483                router,
7484                &[(station_id, 2), (station_id, 3)],
7485                &mut scratch,
7486            );
7487            assert_eq!(reused, samples);
7488            (reused.as_ptr(), reused[0].cells.as_ptr())
7489        };
7490        let subscriber_capacity = scratch.retained_subscriber_capacity();
7491        let occupancy_capacity = scratch.retained_occupancy_capacity();
7492        let cell_capacity = scratch.retained_cell_capacity();
7493
7494        let reused = load_sampler.sample_all_into(
7495            stations,
7496            indexes,
7497            router,
7498            &[(station_id, 2), (station_id, 3)],
7499            &mut scratch,
7500        );
7501        assert_eq!(reused, samples);
7502        assert_eq!(reused.as_ptr(), sample_ptr);
7503        assert_eq!(reused[0].cells.as_ptr(), cell_ptr);
7504        assert_eq!(scratch.retained_sample_slots(), 1);
7505        assert_eq!(scratch.retained_subscriber_capacity(), subscriber_capacity);
7506        assert_eq!(scratch.retained_occupancy_capacity(), occupancy_capacity);
7507        assert_eq!(scratch.retained_cell_capacity(), cell_capacity);
7508    }
7509
7510    #[test]
7511    fn station_event_transport_bridge_routes_events_through_bounded_packets() {
7512        let mut stations = StationSet::default();
7513        stations.push(station(1, 10));
7514        stations.push(station(2, 10));
7515
7516        let mut router = EventRouter::default();
7517        router.register_stations(&stations);
7518        let mut transport = InMemoryStationTransport::default();
7519        transport.register_station(StationId::new(2));
7520        let mut bridge = StationEventTransportBridge::default();
7521        let event = StationEvent {
7522            id: EventId::new(7),
7523            source: StationId::new(1),
7524            target: StationId::new(2),
7525            source_tick: Tick::new(0),
7526            target_tick: Tick::new(1),
7527            priority: EventPriority::Important,
7528            kind: EventKind::Custom(99),
7529        };
7530
7531        bridge
7532            .send_event(&mut transport, &event)
7533            .expect("event should encode and send");
7534        assert_eq!(transport.queued_len(StationId::new(2)), Some(1));
7535
7536        let report = bridge
7537            .pump_target(&mut transport, &mut router, StationId::new(2), 4)
7538            .expect("event should pump into router");
7539        assert_eq!(report.packets_received, 1);
7540        assert_eq!(report.events_routed, 1);
7541        assert_eq!(router.queued_len(StationId::new(2)), Some(1));
7542
7543        let mut scheduler = StationScheduler::default();
7544        scheduler.advance_all(&mut stations);
7545        let drained = scheduler
7546            .drain_ready_events(&stations, &mut router)
7547            .expect("drain should work");
7548        assert_eq!(drained, vec![event]);
7549        assert_eq!(bridge.stats().events_sent, 1);
7550        assert_eq!(bridge.stats().events_routed, 1);
7551        assert_eq!(transport.stats().packets_sent, 1);
7552        assert_eq!(transport.stats().packets_received, 1);
7553    }
7554
7555    #[test]
7556    fn command_dispatch_transport_bridge_enqueues_stamped_command() {
7557        let gateway_station = StationId::new(0);
7558        let target_station = StationId::new(2);
7559        let command = CommandEnvelope {
7560            id: CommandId::new(42),
7561            client_id: ClientId::new(7),
7562            entity_id: EntityId::new(100),
7563            sequence: 42,
7564            received_at: Tick::new(12),
7565            kind: 1,
7566            priority: CommandPriority::High,
7567            payload: b"move:north".to_vec(),
7568        };
7569        let mut transport = InMemoryStationTransport::default();
7570        transport.register_station(target_station);
7571        let mut queues = BTreeMap::from([(target_station, command_queues())]);
7572        let mut bridge = CommandDispatchTransportBridge::default();
7573
7574        bridge
7575            .send_envelope(&mut transport, gateway_station, target_station, &command)
7576            .expect("command dispatch should send");
7577        assert_eq!(transport.queued_len(target_station), Some(1));
7578        let report = bridge
7579            .pump_target(
7580                &mut transport,
7581                &mut queues,
7582                target_station,
7583                4,
7584                CommandIngress::RUNNING,
7585            )
7586            .expect("command dispatch should pump");
7587
7588        assert_eq!(report.packets_received, 1);
7589        assert_eq!(report.commands_enqueued, 1);
7590        let queued_command = queues
7591            .get_mut(&target_station)
7592            .expect("queue should exist")
7593            .pop_next()
7594            .expect("command should queue");
7595        assert_eq!(queued_command, command);
7596        assert_eq!(bridge.stats().commands_sent, 1);
7597        assert_eq!(bridge.stats().commands_enqueued, 1);
7598        assert_eq!(transport.stats().packets_sent, 1);
7599        assert_eq!(transport.stats().packets_received, 1);
7600    }
7601
7602    #[test]
7603    fn command_dispatch_transport_bridge_rejects_endpoint_mismatch() {
7604        let packet_target = StationId::new(2);
7605        let frame_target = StationId::new(3);
7606        let mut transport = InMemoryStationTransport::default();
7607        transport.register_station(packet_target);
7608        let frame = CommandDispatchFrame {
7609            station_id: frame_target,
7610            client_id: ClientId::new(7),
7611            command_id: CommandId::new(42),
7612            entity_id: EntityId::new(100),
7613            sequence: 42,
7614            received_at: Tick::new(12),
7615            kind: 1,
7616            priority: CommandPriority::High,
7617            payload: Vec::new(),
7618        };
7619        let mut bytes = Vec::new();
7620        BinaryFrameEncoder
7621            .encode_command_dispatch(&frame, &mut bytes)
7622            .expect("frame should encode");
7623        transport
7624            .send_station(StationOutboundPacket {
7625                source_station: StationId::new(0),
7626                target_station: packet_target,
7627                bytes,
7628            })
7629            .expect("bad packet should enter transport");
7630        let mut queues = BTreeMap::from([(packet_target, command_queues())]);
7631        let mut bridge = CommandDispatchTransportBridge::default();
7632
7633        let error = bridge
7634            .pump_target(
7635                &mut transport,
7636                &mut queues,
7637                packet_target,
7638                4,
7639                CommandIngress::RUNNING,
7640            )
7641            .expect_err("endpoint mismatch should reject");
7642
7643        assert!(matches!(
7644            error,
7645            CommandDispatchTransportError::EndpointMismatch {
7646                packet_source,
7647                packet_target: observed_packet_target,
7648                dispatch_target,
7649            } if packet_source == StationId::new(0)
7650                && observed_packet_target == packet_target
7651                && dispatch_target == frame_target
7652        ));
7653        assert!(
7654            queues
7655                .get_mut(&packet_target)
7656                .expect("queue should exist")
7657                .pop_next()
7658                .is_none()
7659        );
7660    }
7661
7662    #[test]
7663    fn cell_migration_moves_owned_entities_and_updates_indexes() {
7664        let grid = GridSpec::new(16.0).expect("valid grid");
7665        let cell = CellCoord3::new(0, 0, 0);
7666        let mut stations = StationSet::default();
7667        let mut source = station(1, 10);
7668        let first = source
7669            .spawn_owned(
7670                EntityId::new(1),
7671                Position3::new(1.0, 1.0, 1.0),
7672                Bounds::Point,
7673                PolicyId::new(0),
7674            )
7675            .expect("first spawn should work");
7676        let second = source
7677            .spawn_owned(
7678                EntityId::new(2),
7679                Position3::new(2.0, 1.0, 1.0),
7680                Bounds::Point,
7681                PolicyId::new(0),
7682            )
7683            .expect("second spawn should work");
7684        stations.push(source);
7685        stations.push(station(2, 10));
7686
7687        let mut source_index = CellIndex::new(grid);
7688        source_index.upsert(first, Position3::new(1.0, 1.0, 1.0), Bounds::Point);
7689        source_index.upsert(second, Position3::new(2.0, 1.0, 1.0), Bounds::Point);
7690        let mut target_index = CellIndex::new(grid);
7691
7692        let mut ownership = CellOwnershipTable::default();
7693        ownership.assign(cell, StationId::new(1));
7694        let update = ownership.apply_split(
7695            &SplitProposal {
7696                source_station: StationId::new(1),
7697                cells_to_move: vec![cell],
7698                moved_pressure_score: 10,
7699            },
7700            StationId::new(2),
7701        );
7702        assert_eq!(ownership.owner_of(cell), Some(StationId::new(2)));
7703        assert_eq!(update.moved_cells, vec![cell]);
7704
7705        let mut scratch = CellMigrationScratch::new();
7706        scratch.reserve(2, 2);
7707        let mut report = CellMigrationReport::default();
7708        report.scanned_cells.reserve(1);
7709        report.entity_migrations.reserve(2);
7710        CellMigrationExecutor::migrate_cells_into(
7711            &mut stations,
7712            &mut source_index,
7713            &mut target_index,
7714            StationId::new(1),
7715            StationId::new(2),
7716            &update.moved_cells,
7717            4,
7718            &mut scratch,
7719            &mut report,
7720        )
7721        .expect("cell migration should work");
7722
7723        assert_eq!(report.entity_migrations.len(), 2);
7724        assert_eq!(target_index.entity_count(), 2);
7725        assert!(
7726            !stations
7727                .get(StationId::new(1))
7728                .expect("source")
7729                .get_by_id(EntityId::new(1))
7730                .expect("source ghost")
7731                .is_owned()
7732        );
7733        assert!(
7734            stations
7735                .get(StationId::new(2))
7736                .expect("target")
7737                .get_by_id(EntityId::new(1))
7738                .expect("target owner")
7739                .is_owned()
7740        );
7741
7742        let retained_handle_capacity = scratch.handle_capacity();
7743        let retained_entity_capacity = scratch.entity_capacity();
7744        let retained_candidate_capacity = scratch.candidate_capacity();
7745        let retained_scanned_capacity = report.scanned_cells.capacity();
7746        let retained_migration_capacity = report.entity_migrations.capacity();
7747        CellMigrationExecutor::migrate_cells_into(
7748            &mut stations,
7749            &mut source_index,
7750            &mut target_index,
7751            StationId::new(1),
7752            StationId::new(2),
7753            &[],
7754            4,
7755            &mut scratch,
7756            &mut report,
7757        )
7758        .expect("empty reusable migration should work");
7759        assert!(report.scanned_cells.is_empty());
7760        assert!(report.entity_migrations.is_empty());
7761        assert_eq!(report.scanned_cells.capacity(), retained_scanned_capacity);
7762        assert_eq!(
7763            report.entity_migrations.capacity(),
7764            retained_migration_capacity
7765        );
7766        assert_eq!(scratch.handle_capacity(), retained_handle_capacity);
7767        assert_eq!(scratch.entity_capacity(), retained_entity_capacity);
7768        assert_eq!(scratch.candidate_capacity(), retained_candidate_capacity);
7769    }
7770
7771    #[test]
7772    #[allow(clippy::too_many_lines)]
7773    fn split_scheduler_plans_and_executes_hot_cell_move() {
7774        let grid = GridSpec::new(16.0).expect("valid grid");
7775        let hot_cell = CellCoord3::new(0, 0, 0);
7776        let mut stations = StationSet::default();
7777        let mut source = station(1, 10);
7778        let handle = source
7779            .spawn_owned(
7780                EntityId::new(1),
7781                Position3::new(1.0, 1.0, 1.0),
7782                Bounds::Point,
7783                PolicyId::new(0),
7784            )
7785            .expect("spawn should work");
7786        stations.push(source);
7787        stations.push(station(2, 10));
7788
7789        let mut source_index = CellIndex::new(grid);
7790        source_index.upsert(handle, Position3::new(1.0, 1.0, 1.0), Bounds::Point);
7791        let mut indexes = StationIndexSet::default();
7792        indexes.insert(StationId::new(1), source_index);
7793        indexes.insert(StationId::new(2), CellIndex::new(grid));
7794
7795        let samples = vec![
7796            StationLoadSample {
7797                station_id: StationId::new(1),
7798                owned_entities: 100,
7799                subscribers: 100,
7800                tick_cost_units: 1000,
7801                cells: vec![CellLoadSample {
7802                    cell: hot_cell,
7803                    owned_entities: 100,
7804                    subscribers: 100,
7805                    event_pressure: 10,
7806                    ..CellLoadSample::default()
7807                }],
7808                ..StationLoadSample::default()
7809            },
7810            StationLoadSample {
7811                station_id: StationId::new(2),
7812                owned_entities: 1,
7813                cells: vec![CellLoadSample {
7814                    cell: CellCoord3::new(10, 0, 0),
7815                    owned_entities: 1,
7816                    ..CellLoadSample::default()
7817                }],
7818                ..StationLoadSample::default()
7819            },
7820        ];
7821        let scheduler = SplitScheduler::new(SplitSchedulerConfig {
7822            thresholds: HotspotThresholds {
7823                max_station_entities: 10,
7824                max_station_subscribers: 10,
7825                max_cell_pressure: 10,
7826                ..HotspotThresholds::default()
7827            },
7828            max_actions_per_pass: 1,
7829            max_cells_per_action: 1,
7830            ghost_ttl_ticks: 4,
7831            ..SplitSchedulerConfig::default()
7832        });
7833        let schedule = scheduler.plan(&samples);
7834        assert_eq!(schedule.actions.len(), 1);
7835        assert_eq!(schedule.actions[0].target_station, StationId::new(2));
7836
7837        let mut ownership = CellOwnershipTable::default();
7838        ownership.assign(hot_cell, StationId::new(1));
7839        let mut execution_scratch = SplitScheduleExecutionScratch::new();
7840        execution_scratch.reserve(1, 1, 1);
7841        {
7842            let report = scheduler
7843                .execute_into(
7844                    &schedule,
7845                    &mut stations,
7846                    &mut indexes,
7847                    &mut ownership,
7848                    &mut execution_scratch,
7849                )
7850                .expect("reusable execute should work");
7851            assert_eq!(report.cell_migrations.len(), 1);
7852            assert_eq!(report.cell_migrations[0].entity_migrations.len(), 1);
7853        }
7854
7855        assert_eq!(ownership.owner_of(hot_cell), Some(StationId::new(2)));
7856        assert_eq!(
7857            indexes
7858                .get(StationId::new(2))
7859                .expect("target index")
7860                .entity_count(),
7861            1
7862        );
7863
7864        let retained_ownership_slots = execution_scratch.retained_ownership_slots();
7865        let retained_migration_slots = execution_scratch.retained_migration_slots();
7866        let retained_update_cells = execution_scratch.retained_update_cell_capacity();
7867        let retained_entity_migrations = execution_scratch.retained_entity_migration_capacity();
7868        let retained_candidates = execution_scratch.retained_candidate_capacity();
7869        execution_scratch.reserve(1, 1, 1);
7870        assert_eq!(
7871            execution_scratch.retained_update_cell_capacity(),
7872            retained_update_cells
7873        );
7874        assert_eq!(
7875            execution_scratch.retained_entity_migration_capacity(),
7876            retained_entity_migrations
7877        );
7878        assert_eq!(
7879            execution_scratch.retained_candidate_capacity(),
7880            retained_candidates
7881        );
7882        let empty = SplitSchedule::default();
7883        let empty_report = scheduler
7884            .execute_into(
7885                &empty,
7886                &mut stations,
7887                &mut indexes,
7888                &mut ownership,
7889                &mut execution_scratch,
7890            )
7891            .expect("empty reusable execute should work");
7892        assert!(empty_report.ownership_updates.is_empty());
7893        assert!(empty_report.cell_migrations.is_empty());
7894        assert_eq!(
7895            execution_scratch.retained_ownership_slots(),
7896            retained_ownership_slots
7897        );
7898        assert_eq!(
7899            execution_scratch.retained_migration_slots(),
7900            retained_migration_slots
7901        );
7902        assert_eq!(
7903            execution_scratch.retained_update_cell_capacity(),
7904            retained_update_cells
7905        );
7906        assert_eq!(
7907            execution_scratch.retained_entity_migration_capacity(),
7908            retained_entity_migrations
7909        );
7910        assert_eq!(
7911            execution_scratch.retained_candidate_capacity(),
7912            retained_candidates
7913        );
7914    }
7915
7916    #[test]
7917    fn split_scheduler_respects_source_cooldown() {
7918        let hot_cell = CellCoord3::new(0, 0, 0);
7919        let samples = split_test_samples(hot_cell);
7920        let scheduler = SplitScheduler::new(SplitSchedulerConfig {
7921            thresholds: split_test_thresholds(),
7922            max_actions_per_pass: 1,
7923            max_cells_per_action: 1,
7924            split_cooldown_ticks: 10,
7925            ..SplitSchedulerConfig::default()
7926        });
7927        let mut state = SplitSchedulerState::default();
7928
7929        let initial = scheduler.plan_with_state(&samples, Some(&state), Tick::new(5));
7930        assert_eq!(initial.actions.len(), 1);
7931        state.record_schedule(&initial, Tick::new(5));
7932
7933        let cooled_down = scheduler.plan_with_state(&samples, Some(&state), Tick::new(8));
7934        assert!(cooled_down.actions.is_empty());
7935        assert_eq!(cooled_down.skipped_cooldown, 1);
7936
7937        let after_cooldown = scheduler.plan_with_state(&samples, Some(&state), Tick::new(16));
7938        assert_eq!(after_cooldown.actions.len(), 1);
7939    }
7940
7941    #[test]
7942    fn split_scheduler_reports_capacity_and_improvement_skips() {
7943        let hot_cell = CellCoord3::new(0, 0, 0);
7944        let samples = split_test_samples(hot_cell);
7945
7946        let capacity_guard = SplitScheduler::new(SplitSchedulerConfig {
7947            thresholds: split_test_thresholds(),
7948            max_actions_per_pass: 1,
7949            max_cells_per_action: 1,
7950            max_target_score_after_move: 1,
7951            ..SplitSchedulerConfig::default()
7952        });
7953        let capacity_schedule = capacity_guard.plan(&samples);
7954        assert!(capacity_schedule.actions.is_empty());
7955        assert_eq!(capacity_schedule.skipped_target_capacity, 1);
7956
7957        let improvement_guard = SplitScheduler::new(SplitSchedulerConfig {
7958            thresholds: split_test_thresholds(),
7959            max_actions_per_pass: 1,
7960            max_cells_per_action: 1,
7961            min_score_improvement: u64::MAX,
7962            ..SplitSchedulerConfig::default()
7963        });
7964        let improvement_schedule = improvement_guard.plan(&samples);
7965        assert!(improvement_schedule.actions.is_empty());
7966        assert_eq!(improvement_schedule.skipped_insufficient_improvement, 1);
7967    }
7968
7969    #[test]
7970    fn split_scheduler_view_matches_owned_and_retains_nested_capacity() {
7971        let hot_cell = CellCoord3::new(0, 0, 0);
7972        let samples = split_test_samples(hot_cell);
7973        let scheduler = SplitScheduler::new(SplitSchedulerConfig {
7974            thresholds: split_test_thresholds(),
7975            max_actions_per_pass: 1,
7976            max_cells_per_action: 1,
7977            ..SplitSchedulerConfig::default()
7978        });
7979        let expected = scheduler.plan(&samples);
7980        let mut scratch = SplitSchedulerScratch::new();
7981
7982        {
7983            let view = scheduler.plan_into(&samples, &mut scratch);
7984            assert_eq!(SplitSchedule::from(view), expected);
7985        }
7986        let decision_slots = scratch.retained_decision_slots();
7987        let action_slots = scratch.retained_action_slots();
7988        let reason_capacity = scratch.retained_reason_capacity();
7989        let action_cell_capacity = scratch.retained_action_cell_capacity();
7990        let candidate_capacity = scratch.retained_candidate_capacity();
7991        assert_eq!(decision_slots, samples.len());
7992        assert_eq!(action_slots, 1);
7993        assert!(reason_capacity > 0);
7994        assert!(action_cell_capacity > 0);
7995        assert!(candidate_capacity > 0);
7996
7997        let reduced = scheduler.plan_into(&samples[1..], &mut scratch);
7998        assert_eq!(reduced.decisions.len(), 1);
7999        assert!(reduced.actions.is_empty());
8000        assert_eq!(scratch.retained_decision_slots(), decision_slots);
8001        assert_eq!(scratch.retained_action_slots(), action_slots);
8002        assert_eq!(scratch.retained_reason_capacity(), reason_capacity);
8003        assert_eq!(
8004            scratch.retained_action_cell_capacity(),
8005            action_cell_capacity
8006        );
8007        assert_eq!(scratch.retained_candidate_capacity(), candidate_capacity);
8008    }
8009
8010    fn split_test_thresholds() -> HotspotThresholds {
8011        HotspotThresholds {
8012            max_station_entities: 10,
8013            max_station_subscribers: 10,
8014            max_cell_pressure: 10,
8015            ..HotspotThresholds::default()
8016        }
8017    }
8018
8019    fn split_test_samples(hot_cell: CellCoord3) -> Vec<StationLoadSample> {
8020        vec![
8021            StationLoadSample {
8022                station_id: StationId::new(1),
8023                owned_entities: 100,
8024                subscribers: 100,
8025                tick_cost_units: 1000,
8026                cells: vec![CellLoadSample {
8027                    cell: hot_cell,
8028                    owned_entities: 100,
8029                    subscribers: 100,
8030                    event_pressure: 10,
8031                    ..CellLoadSample::default()
8032                }],
8033                ..StationLoadSample::default()
8034            },
8035            StationLoadSample {
8036                station_id: StationId::new(2),
8037                owned_entities: 1,
8038                cells: vec![CellLoadSample {
8039                    cell: CellCoord3::new(10, 0, 0),
8040                    owned_entities: 1,
8041                    ..CellLoadSample::default()
8042                }],
8043                ..StationLoadSample::default()
8044            },
8045        ]
8046    }
8047}
sectorsync_runtime/lib.rs

sectorsync_runtime/
lib.rs
