lace_sans_io/core/
direct.rs

1//! Same-runtime direct-pair ownership and atomic policy replacement.
2
3use super::machine::{
4    core_machine_error, endpoint_protocol_phase, integration_machine_error,
5    machine_error_from_logic, store_machine_error, validate_store_result,
6};
7use super::*;
8
9#[derive(Debug)]
10enum DirectPairState {
11    Active,
12    Failed(MachineError),
13    Closed(CloseReason),
14}
15
16#[derive(Debug)]
17pub struct DirectPair {
18    id: InterlaceId,
19    endpoints: [low_level::EndpointMachine; 2],
20    retained: [Option<EndpointMessage>; 2],
21    active_store_call: [Option<StoreCall>; 2],
22    store_response: [Option<Result<StoreResult, StoreError>>; 2],
23    observations: VecDeque<InterlaceObservation>,
24    authorities: [SessionAuthority; 2],
25    limits: InterlaceResourceLimits,
26    staged_replacement: Option<[crate::logic::InterlaceModule; 2]>,
27    state: DirectPairState,
28}
29
30impl DirectPair {
31    pub fn new(config: DirectConfig) -> Result<Self, CoreError> {
32        config.validate()?;
33        let DirectConfig {
34            operands,
35            authorities,
36            limits,
37        } = config;
38        let plan = Arc::new(compile_exchange_plan(operands));
39        let id = InterlaceId(NEXT_INTERLACE_ID.next());
40        Ok(Self {
41            id,
42            endpoints: [
43                low_level::EndpointMachine::new(
44                    id,
45                    SideExecutionConfig::new(
46                        plan.clone(),
47                        crate::logic::ExchangePlanSlot::Slot0,
48                        authorities[0],
49                        limits.clone(),
50                    ),
51                ),
52                low_level::EndpointMachine::new(
53                    id,
54                    SideExecutionConfig::new(
55                        plan,
56                        crate::logic::ExchangePlanSlot::Slot1,
57                        authorities[1],
58                        limits.clone(),
59                    ),
60                ),
61            ],
62            retained: [None, None],
63            active_store_call: [None, None],
64            store_response: [None, None],
65            observations: VecDeque::new(),
66            authorities,
67            limits,
68            staged_replacement: None,
69            state: DirectPairState::Active,
70        })
71    }
72    pub fn id(&self) -> InterlaceId {
73        self.id
74    }
75    pub fn cancel(&mut self) {
76        if matches!(self.state, DirectPairState::Active) {
77            self.set_closed(CloseReason::Cancelled);
78        }
79    }
80    pub fn active_store_call(&self, side: InterlaceSide) -> Option<&StoreCall> {
81        matches!(self.state, DirectPairState::Active)
82            .then(|| self.active_store_call[side.index()].as_ref())
83            .flatten()
84    }
85    pub fn active_store_call_mut(&mut self, side: InterlaceSide) -> Option<&mut StoreCall> {
86        matches!(self.state, DirectPairState::Active)
87            .then(|| self.active_store_call[side.index()].as_mut())
88            .flatten()
89    }
90    pub fn current_round(&self) -> InterlaceRoundId {
91        self.endpoints[0].round().max(self.endpoints[1].round())
92    }
93    pub fn protocol_phase(&self, side: InterlaceSide) -> MachineProtocolPhase {
94        match self.state {
95            DirectPairState::Active => endpoint_protocol_phase(&self.endpoints[side.index()]),
96            DirectPairState::Failed(_) => MachineProtocolPhase::Failed,
97            DirectPairState::Closed(_) => MachineProtocolPhase::Closed,
98        }
99    }
100    #[cfg(test)]
101    pub(super) fn completed_evaluation_work(&self, side: InterlaceSide) -> [u64; 2] {
102        self.endpoints[side.index()].completed_evaluation_work()
103    }
104    pub fn protocol_phase_token(&self, side: InterlaceSide) -> ProtocolPhaseToken {
105        ProtocolPhaseToken::new(Some(self.id), self.protocol_phase(side))
106    }
107
108    pub fn replace_policy(
109        &mut self,
110        operands: [crate::logic::InterlaceModule; 2],
111    ) -> Result<(), CoreError> {
112        if !matches!(self.state, DirectPairState::Active) {
113            return Err(CoreError::ReplacementUnavailable("interlace is not open"));
114        }
115        for operand in &operands {
116            validate_operand_blob_size(operand)?;
117        }
118        let current = [
119            self.endpoints[0].local_operand(),
120            self.endpoints[1].local_operand(),
121        ];
122        if current[0] == &operands[0] && current[1] == &operands[1] {
123            self.staged_replacement = None;
124            return Ok(());
125        }
126        if self.staged_replacement.as_ref() == Some(&operands) {
127            return Ok(());
128        }
129        self.staged_replacement = Some(operands);
130        Ok(())
131    }
132
133    pub fn submit_store_call(&mut self, id: u64) -> Result<(), MachineError> {
134        let index = self
135            .active_store_call
136            .iter()
137            .position(|call| {
138                call.as_ref()
139                    .is_some_and(|call| call.id == id && call.owner == self.id)
140            })
141            .ok_or_else(|| {
142                integration_machine_error("store call does not belong to this direct pair")
143            })?;
144        let mut active = self.active_store_call[index]
145            .take()
146            .expect("located active IO");
147        let response = match active.response.take() {
148            Some(response) => response,
149            None => {
150                return self.fail_store_submission(CoreError::InvalidConfig(
151                    "store-call response is incomplete",
152                ));
153            }
154        };
155        if let Ok(result) = &response
156            && let Err(error) = validate_store_result(active.op(), result)
157        {
158            return self.fail_store_submission(error);
159        }
160        self.endpoints[index].restore_pending_store(active.continuation);
161        self.store_response[index] = Some(response);
162        Ok(())
163    }
164
165    pub fn try_store_position_advanced(
166        &mut self,
167        boundary: &FixedPointWakeBoundary,
168        position: StorePosition,
169    ) -> StorePositionAdvance {
170        if !matches!(self.state, DirectPairState::Active) {
171            return StorePositionAdvance::Stale;
172        }
173        self.consume_staged_replacement();
174        self.endpoints[boundary.side.index()].try_store_position_advanced(boundary, position)
175    }
176    pub fn timeout(&mut self, side: InterlaceSide) {
177        if matches!(self.state, DirectPairState::Active)
178            && self.protocol_phase(side).timeout_phase().is_some()
179        {
180            self.set_failure(MachineError {
181                category: ErrorCategory::Timeout,
182                side: Some(side),
183                record_hash: None,
184                detail: format!("phase timeout in {:?}", self.protocol_phase(side)),
185                source: None,
186            });
187        }
188    }
189
190    pub fn poll(&mut self) -> MachinePoll<'_> {
191        self.consume_staged_replacement();
192        if !matches!(self.state, DirectPairState::Active) {
193            return self.terminal_poll();
194        }
195        if self.staged_replacement.is_some() {
196            return MachinePoll::Effects(self.active_store_effects());
197        }
198        if let Some(observation) = self.observations.pop_front() {
199            return MachinePoll::Observation(observation);
200        }
201        let mut entered_fixed_point = false;
202        for _ in 0..1024 {
203            let mut progressed = false;
204            for index in 0..2 {
205                match self.route(index) {
206                    Ok(routed) => progressed |= routed,
207                    Err(error) => {
208                        self.set_failure(error);
209                        break;
210                    }
211                }
212                if self.active_store_call[index].is_some() || self.retained[index].is_some() {
213                    continue;
214                }
215                let poll = self.endpoints[index].step(&mut self.store_response[index]);
216                match poll {
217                    EndpointPoll::PeerOutput { message } => {
218                        self.retained[index] = Some(message);
219                        progressed = true;
220                    }
221                    EndpointPoll::StoreRequest(request) => {
222                        self.begin_store_call(index, request);
223                        progressed = true;
224                    }
225                    EndpointPoll::Observation(observation) => {
226                        self.observations.push_back(observation);
227                        progressed = true;
228                    }
229                    EndpointPoll::Progress => progressed = true,
230                    EndpointPoll::FixedPoint(_) => {
231                        entered_fixed_point = true;
232                        progressed = true;
233                    }
234                    EndpointPoll::Idle => {}
235                    EndpointPoll::Closed(reason) => {
236                        self.set_closed(reason);
237                        break;
238                    }
239                    EndpointPoll::LogicError(error) => {
240                        self.set_failure(machine_error_from_logic(error));
241                        break;
242                    }
243                    EndpointPoll::StoreFailed {
244                        side,
245                        record_hash,
246                        error,
247                    } => {
248                        self.set_failure(store_machine_error(side, record_hash, error));
249                        break;
250                    }
251                }
252            }
253            if !matches!(self.state, DirectPairState::Active) {
254                return self.terminal_poll();
255            }
256            if let Some(observation) = self.observations.pop_front() {
257                return MachinePoll::Observation(observation);
258            }
259            if !progressed {
260                break;
261            }
262        }
263        if self.both_fixed() {
264            let fixed = self.fixed_point();
265            if entered_fixed_point {
266                return MachinePoll::EnteredFixedPoint(fixed);
267            }
268            return MachinePoll::NeedStoreAdvance(fixed);
269        }
270        let effects = self.active_store_effects();
271        if effects.side0.is_some() || effects.side1.is_some() {
272            MachinePoll::Effects(effects)
273        } else {
274            MachinePoll::NeedPeerInput
275        }
276    }
277
278    fn active_store_effects(&self) -> ReadyEffectsRef<'_> {
279        ReadyEffectsRef {
280            side0: self.active_store_call[0]
281                .as_ref()
282                .map(EndpointEffectRef::StoreCall),
283            side1: self.active_store_call[1]
284                .as_ref()
285                .map(EndpointEffectRef::StoreCall),
286        }
287    }
288    fn consume_staged_replacement(&mut self) {
289        if self.active_store_call.iter().any(Option::is_some) {
290            return;
291        }
292        let Some(operands) = self.staged_replacement.take() else {
293            return;
294        };
295        let config = DirectConfig::new(operands, self.authorities).with_limits(self.limits.clone());
296        match DirectPair::new(config) {
297            Ok(fresh) => *self = fresh,
298            Err(error) => self.set_failure(core_machine_error(error)),
299        }
300    }
301    fn begin_store_call(&mut self, index: usize, request: StoreRequest) {
302        let id = NEXT_STORE_CALL_ID.next();
303        let continuation = self.endpoints[index]
304            .take_pending_store()
305            .expect("StoreRequest owns continuation");
306        self.active_store_call[index] = Some(StoreCall::new(id, self.id, request, continuation));
307    }
308    fn route(&mut self, from: usize) -> Result<bool, MachineError> {
309        let Some(message) = self.retained[from].as_ref() else {
310            return Ok(false);
311        };
312        let target = 1 - from;
313        if self.active_store_call[target].is_some()
314            || !self.endpoints[target].can_accept_input(message)
315        {
316            return Ok(false);
317        }
318        let message = self.retained[from].take().expect("retained output");
319        match self.endpoints[target].push_input(message) {
320            Ok(()) => {}
321            Err(low_level::PushInputError::Unavailable) => {
322                return Err(integration_machine_error(
323                    "direct peer-slot readiness changed without a state transition",
324                ));
325            }
326            Err(low_level::PushInputError::Logic(error)) => {
327                return Err(machine_error_from_logic(error));
328            }
329        }
330        self.endpoints[from].accept_peer_output();
331        Ok(true)
332    }
333    fn fail_store_submission<T>(&mut self, error: CoreError) -> Result<T, MachineError> {
334        let error = core_machine_error(error);
335        self.set_failure(error.clone());
336        Err(error)
337    }
338    fn terminal_poll(&self) -> MachinePoll<'_> {
339        match &self.state {
340            DirectPairState::Failed(error) => MachinePoll::Failed(error),
341            DirectPairState::Closed(reason) => MachinePoll::Closed(reason.clone()),
342            DirectPairState::Active => {
343                unreachable!("terminal poll requires a terminal direct pair")
344            }
345        }
346    }
347    fn retire_effects(&mut self) {
348        self.retained = [None, None];
349        self.active_store_call = [None, None];
350        self.store_response = [None, None];
351        self.observations.clear();
352        self.staged_replacement = None;
353    }
354    fn set_failure(&mut self, error: MachineError) {
355        self.retire_effects();
356        self.state = DirectPairState::Failed(error);
357    }
358    fn set_closed(&mut self, reason: CloseReason) {
359        self.retire_effects();
360        self.state = DirectPairState::Closed(reason);
361    }
362    fn both_fixed(&self) -> bool {
363        self.endpoints
364            .iter()
365            .all(|endpoint| matches!(endpoint.endpoint_state(), EndpointState::FixedPoint(_)))
366    }
367    fn fixed_point(&self) -> FixedPointBoundary {
368        FixedPointBoundary {
369            round: self.current_round(),
370            wake_boundaries: self
371                .endpoints
372                .iter()
373                .map(low_level::EndpointMachine::fixed_point_wake_boundary)
374                .collect(),
375        }
376    }
377}
378
379#[cfg(test)]
380#[path = "direct/exposure_access_tests.rs"]
381mod exposure_access_tests;
382
383#[cfg(test)]
384#[path = "direct/generic_field_vertical_tests.rs"]
385mod generic_field_vertical_tests;
386
387#[cfg(test)]
388#[path = "direct/transfer_tests.rs"]
389mod transfer_tests;
390
391#[cfg(test)]
392#[path = "direct/correctness_tests.rs"]
393mod correctness_tests;
394
395#[cfg(test)]
396#[path = "direct/terminal_store_call_tests.rs"]
397mod terminal_store_call_tests;
398
399#[cfg(test)]
400mod tests {
401    use super::*;
402
403    pub(super) fn module(body: &str) -> crate::logic::InterlaceModule {
404        crate::logic::InterlaceModule::from_text(body).unwrap()
405    }
406
407    fn empty_result(op: &StoreOp) -> StoreResult {
408        match op {
409            StoreOp::CurrentStorePosition => {
410                StoreResult::CurrentStorePosition(StorePosition::BEFORE_FIRST)
411            }
412            StoreOp::ResolveStoreRecordIds { hashes, .. } => StoreResult::ResolvedStoreRecordIds {
413                store_record_ids: vec![None; hashes.len()],
414            },
415            StoreOp::FindRecords { .. } => StoreResult::FoundRecords {
416                records: vec![],
417                complete: true,
418            },
419            StoreOp::LoadRecordsByHash { hashes } => StoreResult::LoadedRecords {
420                outcomes: vec![LoadRecordOutcome::Missing; hashes.len()],
421            },
422            StoreOp::LoadRecordsByStoreId { ids } => StoreResult::LoadedRecords {
423                outcomes: vec![LoadRecordOutcome::Missing; ids.len()],
424            },
425            StoreOp::StoreRecords { records } => StoreResult::StoredRecords {
426                outcomes: records
427                    .iter()
428                    .enumerate()
429                    .map(|(index, _)| StoreRecordOutcome::Stored {
430                        store_record_id: StoreRecordId::new(index as u64 + 1).unwrap(),
431                    })
432                    .collect(),
433                current_store_position: StorePosition::from_raw(records.len() as u64),
434            },
435        }
436    }
437
438    pub(super) fn drive_fixed(pair: &mut DirectPair) -> FixedPointBoundary {
439        for _ in 0..1000 {
440            match pair.poll() {
441                MachinePoll::Effects(effects) => {
442                    let pending = [effects.side0, effects.side1]
443                        .into_iter()
444                        .flatten()
445                        .filter_map(|effect| match effect {
446                            EndpointEffectRef::StoreCall(call) => {
447                                Some((call.side(), call.id(), call.op().clone()))
448                            }
449                            EndpointEffectRef::PeerOutput(_) => None,
450                        })
451                        .collect::<Vec<_>>();
452                    for (side, id, op) in pending {
453                        pair.active_store_call_mut(side)
454                            .unwrap()
455                            .fill_response(Ok(empty_result(&op)))
456                            .unwrap();
457                        pair.submit_store_call(id).unwrap();
458                    }
459                }
460                MachinePoll::EnteredFixedPoint(fixed) => return fixed,
461                MachinePoll::Observation(_) | MachinePoll::NeedPeerInput => {}
462                other => panic!("unexpected direct poll: {other:?}"),
463            }
464        }
465        panic!("direct pair did not settle")
466    }
467
468    #[test]
469    fn phase_tokens_and_store_wakes_follow_core_ownership() {
470        let policy = module("SelectStoredRecord(Record) :- Have(Record).");
471        let mut pair = DirectPair::new(DirectConfig::new(
472            [policy.clone(), policy],
473            [SessionAuthority::default(), SessionAuthority::default()],
474        ))
475        .unwrap();
476        let initial = pair.protocol_phase_token(InterlaceSide::Side0);
477        assert_eq!(initial, pair.protocol_phase_token(InterlaceSide::Side0));
478        assert!(initial.phase().timeout_phase().is_some());
479
480        let fixed = drive_fixed(&mut pair);
481        let fixed_token = pair.protocol_phase_token(InterlaceSide::Side0);
482        assert_ne!(fixed_token, initial);
483        assert!(fixed_token.phase().timeout_phase().is_none());
484        let boundary = &fixed.wake_boundaries[0];
485        assert_eq!(
486            pair.try_store_position_advanced(boundary, boundary.position),
487            StorePositionAdvance::Stale,
488        );
489        assert_eq!(
490            pair.try_store_position_advanced(
491                boundary,
492                StorePosition::from_raw(boundary.position.raw() + 1)
493            ),
494            StorePositionAdvance::Accepted,
495        );
496        assert_ne!(pair.protocol_phase_token(InterlaceSide::Side0), fixed_token);
497    }
498
499    #[test]
500    fn replacement_waits_for_both_store_effects_and_latest_staging_wins() {
501        let initial = module("SelectStoredRecord(Record) :- Have(Record).");
502        let mut pair = DirectPair::new(DirectConfig::new(
503            [initial.clone(), initial],
504            [SessionAuthority::default(), SessionAuthority::default()],
505        ))
506        .unwrap();
507        let first_machine = pair.id();
508        assert!(matches!(pair.poll(), MachinePoll::Effects(_)));
509        assert!(pair.active_store_call.iter().all(Option::is_some));
510
511        let superseded = module("TxDisabled() :- true.");
512        let replacement = module("RxDisabled() :- true.");
513        pair.replace_policy([superseded.clone(), superseded])
514            .unwrap();
515        pair.replace_policy([replacement.clone(), replacement.clone()])
516            .unwrap();
517        assert_eq!(pair.id(), first_machine, "staging mutated the active pair");
518        assert!(matches!(pair.poll(), MachinePoll::Effects(_)));
519        assert_eq!(
520            pair.id(),
521            first_machine,
522            "replacement ran before store call settled"
523        );
524
525        for side in [InterlaceSide::Side0, InterlaceSide::Side1] {
526            let (id, op) = {
527                let call = pair.active_store_call(side).unwrap();
528                (call.id(), call.op().clone())
529            };
530            pair.active_store_call_mut(side)
531                .unwrap()
532                .fill_response(Ok(empty_result(&op)))
533                .unwrap();
534            pair.submit_store_call(id).unwrap();
535            if side == InterlaceSide::Side0 {
536                assert!(matches!(pair.poll(), MachinePoll::Effects(_)));
537                assert_eq!(
538                    pair.id(),
539                    first_machine,
540                    "replacement ran with the sibling store call active"
541                );
542            }
543        }
544
545        assert!(matches!(pair.poll(), MachinePoll::Effects(_)));
546        assert_ne!(pair.id(), first_machine);
547        assert_eq!(pair.endpoints[0].local_operand(), &replacement);
548        assert_eq!(pair.endpoints[1].local_operand(), &replacement);
549    }
550
551    #[test]
552    fn replacement_equal_to_current_cancels_an_unactivated_replacement() {
553        let initial = module("SelectStoredRecord(Record) :- Have(Record).");
554        let mut pair = DirectPair::new(DirectConfig::new(
555            [initial.clone(), initial.clone()],
556            [SessionAuthority::default(), SessionAuthority::default()],
557        ))
558        .unwrap();
559        let first_machine = pair.id();
560        let replacement = module("TxDisabled() :- true.");
561        pair.replace_policy([replacement.clone(), replacement])
562            .unwrap();
563        pair.replace_policy([initial.clone(), initial]).unwrap();
564
565        assert!(matches!(pair.poll(), MachinePoll::Effects(_)));
566        assert_eq!(pair.id(), first_machine);
567    }
568
569    #[test]
570    fn replacement_is_staged_then_atomically_recreates_both_endpoints_and_fixed_point_ids() {
571        let initial = module("SelectStoredRecord(Record) :- Have(Record).");
572        let mut pair = DirectPair::new(DirectConfig::new(
573            [initial.clone(), initial],
574            [SessionAuthority::default(), SessionAuthority::default()],
575        ))
576        .unwrap();
577        let first_machine = pair.id();
578        let first_fixed = drive_fixed(&mut pair);
579        let first_ids = first_fixed
580            .wake_boundaries
581            .iter()
582            .map(|boundary| boundary.fixed_point)
583            .collect::<Vec<_>>();
584
585        let replacement = module("TxDisabled() :- true.");
586        pair.replace_policy([replacement.clone(), replacement])
587            .unwrap();
588        assert_eq!(
589            pair.id(),
590            first_machine,
591            "staging must not step or replace the pair"
592        );
593        assert_eq!(
594            pair.try_store_position_advanced(
595                &first_fixed.wake_boundaries[0],
596                StorePosition::from_raw(1)
597            ),
598            StorePositionAdvance::Stale,
599        );
600        assert_ne!(
601            pair.id(),
602            first_machine,
603            "replacement must precede stale watch delivery"
604        );
605        let second_fixed = drive_fixed(&mut pair);
606        assert_ne!(pair.id(), first_machine);
607        let mut fixed_points = first_ids.clone();
608        fixed_points.extend(
609            second_fixed
610                .wake_boundaries
611                .iter()
612                .map(|boundary| boundary.fixed_point),
613        );
614        let entries = fixed_points.len();
615        fixed_points.sort();
616        fixed_points.dedup();
617        assert_eq!(
618            fixed_points.len(),
619            entries,
620            "each fixed-point entry has a fresh identity"
621        );
622        assert!(
623            second_fixed
624                .wake_boundaries
625                .iter()
626                .all(|boundary| boundary.machine == pair.id())
627        );
628    }
629}
lace_sans_io/core/direct.rs

lace_sans_io/core/
direct.rs
