canwu_sim/runtime/
replay.rs

1use super::{
2    BoundaryRecord, BoundaryRequest, COMMITMENT_FORMAT_VERSION, CanwuError, CauseRef,
3    CommandAttemptOutcome, CommandAttemptRecord, CommandIngress, CommandOutcome, CommandRecord,
4    ENGINE_VERSION, ErrorCode, IngressPayload, IngressRecord, PluginArchiveObjectProvider,
5    PluginIngressRequest, PluginRegistry, ReplayJournal, SNAPSHOT_FORMAT_VERSION,
6    STATE_REVISION_FORMAT_VERSION, SimDuration, SimTime, Simulation, SimulationPlugin,
7    authoritative_revision_count, authoritative_run_identity, boundary_state_hash_format,
8    is_canonical_hash, manifest,
9};
10#[cfg(test)]
11use super::{RunConfiguration, RunManifest, Scenario};
12use std::{collections::BTreeSet, rc::Rc};
13
14impl Simulation {
15    /// Reconstructs caller-supplied core commands without proving a recorded
16    /// package environment. Use [`Self::replay_from_journal`] for exact replay.
17    #[cfg(test)]
18    pub(crate) fn replay(
19        seed: u64,
20        scenario: Scenario,
21        commands: &[CommandRecord],
22        final_time: SimTime,
23    ) -> Result<Self, CanwuError> {
24        Self::replay_with_plugins(seed, scenario, &[], commands, final_time)
25    }
26
27    /// Reconstructs caller-supplied inputs under caller-supplied plugins.
28    /// This is not an exact replay identity check.
29    #[cfg(test)]
30    pub(crate) fn replay_with_plugins(
31        seed: u64,
32        scenario: Scenario,
33        plugins: &[&dyn SimulationPlugin],
34        commands: &[CommandRecord],
35        final_time: SimTime,
36    ) -> Result<Self, CanwuError> {
37        Self::replay_with_boundaries(seed, scenario, plugins, commands, &[], final_time)
38    }
39
40    /// Reconstructs caller-supplied inputs and compares supplied boundaries.
41    /// Use [`Self::replay_from_journal`] when command-only runs must also bind
42    /// their recorded run and plugin identities.
43    #[cfg(test)]
44    pub(crate) fn replay_with_boundaries(
45        seed: u64,
46        scenario: Scenario,
47        plugins: &[&dyn SimulationPlugin],
48        commands: &[CommandRecord],
49        boundaries: &[BoundaryRecord],
50        final_time: SimTime,
51    ) -> Result<Self, CanwuError> {
52        let run_manifest = RunManifest::for_scenario("canwu.inline", "scenario", "1", &scenario)?;
53        Self::replay_with_run_manifest(
54            seed,
55            scenario,
56            run_manifest,
57            plugins,
58            commands,
59            boundaries,
60            final_time,
61        )
62    }
63
64    /// Reconstructs caller-supplied inputs under a caller-supplied run manifest.
65    /// This is useful for fixtures; it does not establish recorded identity.
66    #[cfg(test)]
67    pub(crate) fn replay_with_run_manifest(
68        seed: u64,
69        scenario: Scenario,
70        run_manifest: RunManifest,
71        plugins: &[&dyn SimulationPlugin],
72        commands: &[CommandRecord],
73        boundaries: &[BoundaryRecord],
74        final_time: SimTime,
75    ) -> Result<Self, CanwuError> {
76        let simulation =
77            Self::new_with_manifest_and_plugins(seed, scenario, run_manifest, plugins)?;
78        Self::replay_records(simulation, commands, &[], &[], boundaries, final_time)
79    }
80
81    /// Reconstructs caller-supplied inputs under a caller-supplied declared run
82    /// configuration. This does not establish recorded-environment identity.
83    #[allow(clippy::too_many_arguments)]
84    #[cfg(test)]
85    pub(crate) fn replay_with_run_configuration(
86        seed: u64,
87        scenario: Scenario,
88        run_manifest: RunManifest,
89        run_configuration: RunConfiguration,
90        plugins: &[&dyn SimulationPlugin],
91        commands: &[CommandRecord],
92        command_attempts: &[CommandAttemptRecord],
93        boundaries: &[BoundaryRecord],
94        final_time: SimTime,
95    ) -> Result<Self, CanwuError> {
96        if command_attempts
97            .iter()
98            .any(|attempt| attempt.ingress == CommandIngress::FrozenReplay)
99        {
100            return Err(CanwuError::new(
101                ErrorCode::ReplayEnvironmentMismatch,
102                "frozen replay attempts require an environment-bound replay journal",
103            ));
104        }
105        let simulation = Self::new_with_run_configuration_and_plugins(
106            seed,
107            scenario,
108            run_manifest,
109            run_configuration,
110            plugins,
111        )?;
112        Self::replay_records(
113            simulation,
114            commands,
115            command_attempts,
116            &[],
117            boundaries,
118            final_time,
119        )
120    }
121
122    /// Replays only after the recorded engine, run, seed, and plugin manifests
123    /// match, then verifies the final checkpoint commitment.
124    pub fn replay_from_journal(
125        plugins: &[&dyn SimulationPlugin],
126        journal: &ReplayJournal,
127    ) -> Result<Self, CanwuError> {
128        Self::replay_from_journal_with_archive_provider(plugins, journal, Rc::new(()))
129    }
130
131    /// Replays with caller-owned package archive storage attached before any
132    /// recorded boundary is evaluated. Cold idempotency and continuation
133    /// checks therefore use the same authenticated provider during replay as
134    /// they do during a live run.
135    pub fn replay_from_journal_with_archive_provider(
136        plugins: &[&dyn SimulationPlugin],
137        journal: &ReplayJournal,
138        archive_provider: Rc<dyn PluginArchiveObjectProvider>,
139    ) -> Result<Self, CanwuError> {
140        if journal.commitment_format_version != COMMITMENT_FORMAT_VERSION {
141            return Err(CanwuError::new(
142                ErrorCode::ReplayEnvironmentMismatch,
143                format!(
144                    "replay journal commitment format {} is unsupported; this engine reads format {COMMITMENT_FORMAT_VERSION}",
145                    journal.commitment_format_version
146                ),
147            ));
148        }
149        if journal.revision_format_version != STATE_REVISION_FORMAT_VERSION {
150            return Err(CanwuError::new(
151                ErrorCode::ReplayEnvironmentMismatch,
152                format!(
153                    "replay journal revision format {} is unsupported; this engine reads format {STATE_REVISION_FORMAT_VERSION}",
154                    journal.revision_format_version
155                ),
156            ));
157        }
158        if journal.authority_root_seed == 0 {
159            return Err(CanwuError::new(
160                ErrorCode::ReplayEnvironmentMismatch,
161                "replay journal is missing its persisted authority root",
162            ));
163        }
164        let expected_final_revision = authoritative_revision_count(
165            journal.commands.len(),
166            journal.command_attempts.len(),
167            journal.boundaries.len(),
168        )?;
169        if journal.final_revision != expected_final_revision {
170            return Err(CanwuError::new(
171                ErrorCode::ReplayEnvironmentMismatch,
172                "replay journal final revision is inconsistent with its committed evidence",
173            ));
174        }
175        let normalized = journal;
176        let scenario = normalized.initial_scenario.clone();
177        manifest::validate(&normalized.run_manifest, Some(&scenario))?;
178        let expected_manifest_hash = manifest::hash(&normalized.run_manifest)?;
179        if normalized.run_manifest_hash != expected_manifest_hash {
180            return Err(CanwuError::new(
181                ErrorCode::ReplayEnvironmentMismatch,
182                "replay journal run manifest hash is inconsistent",
183            ));
184        }
185        manifest::validate_run_configuration(
186            &normalized.run_manifest,
187            &normalized.run_configuration,
188        )?;
189        if normalized.engine_version != ENGINE_VERSION
190            || normalized.snapshot_format_version != SNAPSHOT_FORMAT_VERSION
191            || !is_canonical_hash(&normalized.run_manifest_hash)
192            || manifest::hash(&normalized.run_manifest)? != normalized.run_manifest_hash
193            || !is_canonical_hash(&normalized.checkpoint_hash)
194        {
195            return Err(CanwuError::new(
196                ErrorCode::ReplayEnvironmentMismatch,
197                "replay journal engine, format, or run identity does not match this runtime",
198            ));
199        }
200        let (_, authority_manifest_hash) = authoritative_run_identity(
201            &normalized.run_manifest,
202            &normalized.run_manifest_hash,
203            &normalized.run_configuration,
204        )?;
205        if normalized.authority_root_seed
206            != super::fresh_authority_root_seed(normalized.root_seed, &authority_manifest_hash)?
207        {
208            return Err(CanwuError::new(
209                ErrorCode::ReplayEnvironmentMismatch,
210                "replay journal authority root is not bound to its run identity",
211            ));
212        }
213        PluginRegistry::from_descriptors(normalized.plugin_descriptors.clone()).map_err(
214            |error| {
215                CanwuError::new(
216                    ErrorCode::ReplayEnvironmentMismatch,
217                    format!("replay journal plugin manifest is invalid: {error}"),
218                )
219            },
220        )?;
221
222        let mut simulation = Self::new_with_configuration_snapshot(
223            normalized.root_seed,
224            scenario,
225            normalized.run_manifest.clone(),
226            normalized.run_configuration.clone(),
227        )?;
228        simulation.set_plugin_archive_object_provider(archive_provider);
229        simulation.state.current.authority_root_seed = normalized.authority_root_seed;
230        let simulation = Self::activate_initial_plugins(simulation, plugins)?;
231        let actual_descriptors: Vec<_> = simulation.plugin_descriptors().cloned().collect();
232        if actual_descriptors != normalized.plugin_descriptors {
233            return Err(CanwuError::new(
234                ErrorCode::ReplayEnvironmentMismatch,
235                "active plugin identities and contracts do not match the replay journal",
236            ));
237        }
238
239        let mut simulation = Self::replay_records(
240            simulation,
241            &normalized.commands,
242            &normalized.command_attempts,
243            &normalized.ingress,
244            &normalized.boundaries,
245            normalized.final_time,
246        )?;
247        if normalized.plugin_registration_closed
248            && !simulation.state.metadata.plugin_registration_closed
249        {
250            simulation.advance(SimDuration::ZERO)?;
251        }
252        if simulation.state.metadata.plugin_registration_closed
253            != normalized.plugin_registration_closed
254        {
255            return Err(CanwuError::new(
256                ErrorCode::ReplayMismatch,
257                "replayed plugin-registration lifecycle does not match the recorded journal",
258            ));
259        }
260        if simulation.revision() != normalized.final_revision {
261            return Err(CanwuError::new(
262                ErrorCode::ReplayMismatch,
263                "replayed final state revision does not match the recorded journal",
264            ));
265        }
266        if simulation.checkpoint_hash() != normalized.checkpoint_hash {
267            return Err(CanwuError::new(
268                ErrorCode::ReplayMismatch,
269                "replayed final checkpoint does not match the recorded journal",
270            ));
271        }
272        Ok(simulation)
273    }
274
275    /// Deserializes a Format 8 replay journal with recursive unknown-field
276    /// rejection, then performs the exact environment-bound replay.
277    pub fn replay_from_journal_json(
278        plugins: &[&dyn SimulationPlugin],
279        json: &str,
280    ) -> Result<Self, CanwuError> {
281        let journal: ReplayJournal = super::deserialize_current_json(json, "replay journal")?;
282        Self::replay_from_journal(plugins, &journal)
283    }
284
285    #[cfg(test)]
286    #[allow(clippy::needless_pass_by_value)]
287    pub(crate) fn replay_from_journal_with_scenario(
288        scenario: Scenario,
289        plugins: &[&dyn SimulationPlugin],
290        journal: &ReplayJournal,
291    ) -> Result<Self, CanwuError> {
292        if scenario != journal.initial_scenario {
293            return Err(CanwuError::new(
294                ErrorCode::ReplayEnvironmentMismatch,
295                "test replay scenario disagrees with the self-contained journal scenario",
296            ));
297        }
298        Self::replay_from_journal(plugins, journal)
299    }
300
301    fn replay_records(
302        mut simulation: Self,
303        commands: &[CommandRecord],
304        attempts: &[CommandAttemptRecord],
305        ingress: &[IngressRecord],
306        boundaries: &[BoundaryRecord],
307        final_time: SimTime,
308    ) -> Result<Self, CanwuError> {
309        simulation.ensure_runtime_ready()?;
310        if !attempts.is_empty() {
311            return Self::replay_attempt_records(
312                simulation, commands, attempts, ingress, boundaries, final_time,
313            );
314        }
315        let mut next_ingress = 0;
316        let mut next_command = 0;
317        for (boundary_index, expected_boundary) in boundaries.iter().enumerate() {
318            enqueue_replay_ingress_cut(
319                &mut simulation,
320                ingress,
321                &mut next_ingress,
322                boundary_index,
323            )?;
324            for admitted in &expected_boundary.admitted_commands {
325                let Some(record) = commands.get(next_command) else {
326                    return Err(CanwuError::new(
327                        ErrorCode::ReplayMismatch,
328                        "boundary replay admits a command absent from the journal",
329                    ));
330                };
331                if record.id != *admitted {
332                    return Err(CanwuError::new(
333                        ErrorCode::ReplayMismatch,
334                        "boundary replay command admission does not match journal order",
335                    ));
336                }
337                replay_command_record(&mut simulation, record, expected_boundary.at)?;
338                next_command += 1;
339            }
340            let receipt = simulation.settle_boundary_with_state_hash_format(
341                BoundaryRequest {
342                    at: expected_boundary.at,
343                    cadences: expected_boundary.cadences.clone(),
344                },
345                boundary_state_hash_format(expected_boundary.state_hash.as_deref())?,
346            )?;
347            let Some(actual_boundary) = simulation.boundaries().last() else {
348                return Err(CanwuError::new(
349                    ErrorCode::ReplayMismatch,
350                    "boundary replay did not append settlement evidence",
351                ));
352            };
353            if receipt.boundary_id != expected_boundary.id || actual_boundary != expected_boundary {
354                return Err(CanwuError::new(
355                    ErrorCode::ReplayMismatch,
356                    format!(
357                        "regenerated boundary {} did not match its journal evidence",
358                        expected_boundary.id
359                    ),
360                ));
361            }
362        }
363        for record in &commands[next_command..] {
364            replay_command_record(&mut simulation, record, final_time)?;
365        }
366        enqueue_replay_ingress_cut(
367            &mut simulation,
368            ingress,
369            &mut next_ingress,
370            boundaries.len(),
371        )?;
372        if next_ingress != ingress.len() {
373            return Err(CanwuError::new(
374                ErrorCode::ReplayMismatch,
375                "ingress journal contains an impossible future boundary issue cut",
376            ));
377        }
378        if final_time < simulation.time() {
379            return Err(CanwuError::new(
380                ErrorCode::InvalidDuration,
381                "replay final time cannot precede the last command",
382            ));
383        }
384        if final_time > simulation.time() {
385            simulation.ensure_legacy_advance_does_not_cross_ingress(final_time)?;
386            simulation.advance_to(final_time)?;
387        }
388        Ok(simulation)
389    }
390
391    fn replay_attempt_records(
392        mut simulation: Self,
393        commands: &[CommandRecord],
394        attempts: &[CommandAttemptRecord],
395        ingress: &[IngressRecord],
396        boundaries: &[BoundaryRecord],
397        final_time: SimTime,
398    ) -> Result<Self, CanwuError> {
399        let command_ingress_requests: BTreeSet<_> = ingress
400            .iter()
401            .filter_map(|record| match &record.payload {
402                IngressPayload::Command { request } => Some(request.request_id),
403                IngressPayload::Decision { request } => {
404                    request.command.as_ref().map(|command| command.request_id)
405                }
406                IngressPayload::Plugin { .. }
407                | IngressPayload::Calendar { .. }
408                | IngressPayload::Maintenance { .. }
409                | IngressPayload::PluginCancellation { .. } => None,
410            })
411            .collect();
412        let mut next_ingress = 0;
413        let mut next_attempt = 0;
414        for (boundary_index, expected_boundary) in boundaries.iter().enumerate() {
415            enqueue_replay_ingress_cut(
416                &mut simulation,
417                ingress,
418                &mut next_ingress,
419                boundary_index,
420            )?;
421            let mut admitted_commands = Vec::new();
422            for admitted in &expected_boundary.admitted_attempts {
423                let Some(record) = attempts.get(next_attempt) else {
424                    return Err(CanwuError::new(
425                        ErrorCode::ReplayMismatch,
426                        "boundary replay admits a command attempt absent from the journal",
427                    ));
428                };
429                if record.id != *admitted {
430                    return Err(CanwuError::new(
431                        ErrorCode::ReplayMismatch,
432                        "boundary replay attempt admission does not match journal order",
433                    ));
434                }
435                let queued = record
436                    .request_id
437                    .is_some_and(|request| command_ingress_requests.contains(&request));
438                if !queued {
439                    replay_attempt_record(&mut simulation, record, commands, expected_boundary.at)?;
440                }
441                if let CommandAttemptOutcome::Accepted { command_id } = record.outcome {
442                    admitted_commands.push(command_id);
443                }
444                next_attempt += 1;
445            }
446            if admitted_commands != expected_boundary.admitted_commands {
447                return Err(CanwuError::new(
448                    ErrorCode::ReplayMismatch,
449                    "boundary replay accepted-command cut disagrees with admitted attempts",
450                ));
451            }
452            let receipt = simulation.settle_boundary_with_state_hash_format(
453                BoundaryRequest {
454                    at: expected_boundary.at,
455                    cadences: expected_boundary.cadences.clone(),
456                },
457                boundary_state_hash_format(expected_boundary.state_hash.as_deref())?,
458            )?;
459            let Some(actual_boundary) = simulation.boundaries().last() else {
460                return Err(CanwuError::new(
461                    ErrorCode::ReplayMismatch,
462                    "boundary replay did not append settlement evidence",
463                ));
464            };
465            if receipt.boundary_id != expected_boundary.id || actual_boundary != expected_boundary {
466                return Err(CanwuError::new(
467                    ErrorCode::ReplayMismatch,
468                    format!(
469                        "regenerated boundary {} did not match its journal evidence",
470                        expected_boundary.id
471                    ),
472                ));
473            }
474        }
475        for record in &attempts[next_attempt..] {
476            replay_attempt_record(&mut simulation, record, commands, final_time)?;
477        }
478        enqueue_replay_ingress_cut(
479            &mut simulation,
480            ingress,
481            &mut next_ingress,
482            boundaries.len(),
483        )?;
484        if next_ingress != ingress.len() {
485            return Err(CanwuError::new(
486                ErrorCode::ReplayMismatch,
487                "ingress journal contains an impossible future boundary issue cut",
488            ));
489        }
490        if simulation.command_log() != commands {
491            return Err(CanwuError::new(
492                ErrorCode::ReplayMismatch,
493                "replayed accepted command journal does not match its recorded evidence",
494            ));
495        }
496        if final_time < simulation.time() {
497            return Err(CanwuError::new(
498                ErrorCode::InvalidDuration,
499                "replay final time cannot precede the last command attempt",
500            ));
501        }
502        if final_time > simulation.time() {
503            simulation.ensure_legacy_advance_does_not_cross_ingress(final_time)?;
504            simulation.advance_to(final_time)?;
505        }
506        Ok(simulation)
507    }
508}
509
510fn enqueue_replay_ingress_cut(
511    simulation: &mut Simulation,
512    ingress: &[IngressRecord],
513    next_ingress: &mut usize,
514    boundary_count: usize,
515) -> Result<(), CanwuError> {
516    let expected_boundary_count = u64::try_from(boundary_count).map_err(|_| {
517        CanwuError::new(
518            ErrorCode::ReplayMismatch,
519            "boundary count exceeds ingress range",
520        )
521    })?;
522    while let Some(record) = ingress.get(*next_ingress) {
523        if record.eligible_boundary_count < expected_boundary_count {
524            return Err(CanwuError::new(
525                ErrorCode::ReplayMismatch,
526                "ingress journal skipped its recorded issue boundary",
527            ));
528        }
529        if record.eligible_boundary_count > expected_boundary_count {
530            break;
531        }
532        if record.issued_at < simulation.time() {
533            return Err(CanwuError::new(
534                ErrorCode::ReplayMismatch,
535                "ingress journal issue time precedes replay state",
536            ));
537        }
538        if record.issued_at > simulation.time() {
539            simulation.ensure_legacy_advance_does_not_cross_ingress(record.issued_at)?;
540            simulation.advance_to(record.issued_at)?;
541        }
542        if let Some(actual) = simulation.state.evidence.ingress.get(*next_ingress) {
543            if actual != record {
544                return Err(CanwuError::new(
545                    ErrorCode::ReplayMismatch,
546                    "plugin-generated ingress does not match journal evidence",
547                ));
548            }
549            *next_ingress += 1;
550            continue;
551        }
552        if matches!(record.cause, Some(CauseRef::Boundary(_))) {
553            return Err(CanwuError::new(
554                ErrorCode::ReplayMismatch,
555                "recorded boundary-generated ingress was not reproduced by its plugin system",
556            ));
557        }
558        let receipt = match &record.payload {
559            IngressPayload::Command { request } => simulation.enqueue_command(
560                record.due_at,
561                record.priority,
562                request.as_ref().clone(),
563            )?,
564            IngressPayload::Plugin {
565                plugin,
566                packet_type,
567                payload,
568                affected_entities,
569                archive_retention,
570            } => {
571                let mut request = PluginIngressRequest::new(
572                    plugin.clone(),
573                    packet_type.clone(),
574                    record.due_at,
575                    payload.clone(),
576                )
577                .with_priority(record.priority);
578                request.affected_entities.clone_from(affected_entities);
579                request.cause.clone_from(&record.cause);
580                request.archive_retention.clone_from(archive_retention);
581                simulation.replay_plugin_ingress(request)?
582            }
583            IngressPayload::Calendar { cadences } => {
584                simulation.schedule_calendar_boundary(record.due_at, cadences.clone())?
585            }
586            IngressPayload::Decision { request } => simulation.enqueue_decision(
587                record.due_at,
588                record.priority,
589                request.as_ref().clone(),
590            )?,
591            IngressPayload::Maintenance { request } => match request.as_ref() {
592                super::MaintenanceIngressRequest::DecisionArchive { commit } => simulation
593                    .enqueue_decision_archive_commit(
594                        record.due_at,
595                        record.priority,
596                        commit.clone(),
597                    )?,
598                super::MaintenanceIngressRequest::OwnerAuthorized { commit } => simulation
599                    .enqueue_owner_authorized_maintenance(
600                        record.due_at,
601                        record.priority,
602                        commit.clone(),
603                    )?,
604            },
605            IngressPayload::PluginCancellation {
606                cancelled,
607                authority,
608                reason,
609            } => simulation.replay_plugin_ingress_cancellation(
610                *cancelled,
611                *authority,
612                reason.clone(),
613            )?,
614        };
615        if receipt.ingress_id != record.id
616            || simulation.state.evidence.ingress.last() != Some(record)
617        {
618            return Err(CanwuError::new(
619                ErrorCode::ReplayMismatch,
620                "regenerated ingress record does not match journal evidence",
621            ));
622        }
623        *next_ingress += 1;
624    }
625    Ok(())
626}
627
628fn replay_command_record(
629    simulation: &mut Simulation,
630    record: &CommandRecord,
631    latest_time: SimTime,
632) -> Result<(), CanwuError> {
633    if record.accepted_at < simulation.time() || record.accepted_at > latest_time {
634        return Err(CanwuError::new(
635            ErrorCode::ReplayMismatch,
636            "replay command timestamps do not match authoritative operation order",
637        ));
638    }
639    simulation.ensure_legacy_advance_does_not_cross_ingress(record.accepted_at)?;
640    simulation.advance_to(record.accepted_at)?;
641    let CommandOutcome::Accepted { receipt } = simulation.admit_command(
642        None,
643        None,
644        record.envelope.clone(),
645        CommandIngress::LegacyDirect,
646        None,
647        false,
648    )?
649    else {
650        return Err(CanwuError::new(
651            ErrorCode::ReplayMismatch,
652            "legacy replay command was rejected",
653        ));
654    };
655    if receipt.command_id != record.id {
656        return Err(CanwuError::new(
657            ErrorCode::ReplayMismatch,
658            "replay command IDs did not match the journal",
659        ));
660    }
661    Ok(())
662}
663
664fn replay_attempt_record(
665    simulation: &mut Simulation,
666    record: &CommandAttemptRecord,
667    commands: &[CommandRecord],
668    latest_time: SimTime,
669) -> Result<(), CanwuError> {
670    if record.at < simulation.time() || record.at > latest_time {
671        return Err(CanwuError::new(
672            ErrorCode::ReplayMismatch,
673            "replay command-attempt timestamps do not match authoritative operation order",
674        ));
675    }
676    simulation.ensure_legacy_advance_does_not_cross_ingress(record.at)?;
677    simulation.advance_to(record.at)?;
678    let outcome = simulation.admit_command(
679        record.request_id,
680        record.expected_revision,
681        record.envelope.clone(),
682        record.ingress,
683        None,
684        true,
685    )?;
686    if simulation.command_attempts().last() != Some(record) {
687        return Err(CanwuError::new(
688            ErrorCode::ReplayMismatch,
689            format!(
690                "regenerated command attempt {} did not match its journal evidence",
691                record.id
692            ),
693        ));
694    }
695    match (&record.outcome, outcome) {
696        (CommandAttemptOutcome::Accepted { command_id }, CommandOutcome::Accepted { receipt })
697            if receipt.command_id == *command_id =>
698        {
699            let index = usize::try_from(command_id.get().saturating_sub(1)).map_err(|_| {
700                CanwuError::new(
701                    ErrorCode::ReplayMismatch,
702                    "replayed command ID exceeds the journal index range",
703                )
704            })?;
705            if simulation.command_log().last() != commands.get(index) {
706                return Err(CanwuError::new(
707                    ErrorCode::ReplayMismatch,
708                    "regenerated command record did not match its journal evidence",
709                ));
710            }
711        }
712        (
713            CommandAttemptOutcome::Rejected { error: expected },
714            CommandOutcome::Rejected { rejection },
715        ) if rejection.error == *expected => {}
716        _ => {
717            return Err(CanwuError::new(
718                ErrorCode::ReplayMismatch,
719                "replayed command-attempt outcome differs from its journal evidence",
720            ));
721        }
722    }
723    Ok(())
724}
canwu_sim/runtime/replay.rs

canwu_sim/runtime/
replay.rs
