⎇ 78fbde12b4b118275c300ee1be8c370ee3a0c03b
crates/modules/nakui/core/src/event_log.rs
Historia
Quién tocó qué
crudo
feat(nakui): módulo crm — clientes, pipeline de ventas, interacciones
1//! Append-only event log for deterministic replay.
2//!
3//! Two entry kinds:
4//! - `Seed`: an externally-provided initial record (the system boundary).
5//! - `Morphism`: a successful kernel-validated morphism call, with the
6//! produced ops attached.
7//!
8//! `replay()` reconstructs a store by reading the log and applying ops
9//! directly — fast, no script execution. `verify_log()` re-runs every
10//! morphism through the kernel and asserts the recomputed ops match the
11//! logged ones, which is the operational definition of determinism.
12//!
13//! Failures are never logged: a rejected morphism produces no event.
14
15use serde::{Deserialize, Serialize};
16use serde_json::Value;
17use std::collections::{BTreeMap, HashMap};
18use std::fs::OpenOptions;
19use std::io::{BufRead, BufReader, Write};
20use std::path::{Path, PathBuf};
21use thiserror::Error;
22use uuid::Uuid;
23
24use crate::delta::FieldOp;
25use crate::executor::{ExecError, Executor};
26use crate::store::{MemoryStore, Store, StoreError};
27
28#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
29#[serde(tag = "kind", rename_all = "snake_case")]
30pub enum LogEntry {
31 Seed {
32 seq: u64,
33 entity: String,
34 id: Uuid,
35 data: Value,
36 /// Bundle hash (just the KCL schemas) at the moment this seed
37 /// was logged. `None` for pre-versioning entries — `verify_log`
38 /// skips the schema check on those. New writes always populate
39 /// it via `seed_and_log`.
40 #[serde(default, skip_serializing_if = "Option::is_none")]
41 schema_hash: Option<[u8; 32]>,
42 },
43 Morphism {
44 seq: u64,
45 morphism: String,
46 inputs: BTreeMap<String, Uuid>,
47 params: Value,
48 ops: Vec<FieldOp>,
49 /// Hash of (kcl bundle | manifest spec | rhai script bytes) at
50 /// the moment this event was logged. `None` for pre-versioning
51 /// entries — `verify_log` skips the schema check on those (they
52 /// predate the contract). New writes always populate it.
53 #[serde(default, skip_serializing_if = "Option::is_none")]
54 schema_hash: Option<[u8; 32]>,
55 },
56}
57
58impl LogEntry {
59 pub fn seq(&self) -> u64 {
60 match self {
61 LogEntry::Seed { seq, .. } => *seq,
62 LogEntry::Morphism { seq, .. } => *seq,
63 }
64 }
65}
66
67#[derive(Debug, Error)]
68pub enum LogError {
69 #[error("io: {0}")]
70 Io(#[from] std::io::Error),
71 #[error("parse at line {line}: {source}")]
72 Parse {
73 line: usize,
74 #[source]
75 source: serde_json::Error,
76 },
77 #[error("non-monotonic seq: got {got}, expected {expected}")]
78 NonMonotonic { got: u64, expected: u64 },
79}
80
81/// Errors from `execute_and_log`. The variants distinguish *when in the
82/// pipeline* the failure occurred — which determines whether the log was
83/// updated and whether the live store is still consistent.
84#[derive(Debug, Error)]
85pub enum ExecuteError {
86 /// Failure before the log was written. Store untouched, log untouched.
87 /// Safe to retry with the same inputs.
88 #[error("pre-log validation failed: {0}")]
89 PreLog(#[from] ExecError),
90
91 /// Log append failed (typically IO). Store untouched, log untouched.
92 /// Safe to retry once the log backend recovers.
93 #[error("log append failed: {0}")]
94 LogAppend(#[from] LogError),
95
96 /// Apply to the store failed AFTER the event was logged. The log is
97 /// canonical; the live store is now stale and should be rebuilt by
98 /// replaying the log. Retrying the same morphism is incorrect — the
99 /// event is already on disk.
100 #[error("store apply failed after log was committed (log is canonical, store stale): {0}")]
101 PostLogStore(crate::store::StoreError),
102}
103
104#[derive(Debug, Error)]
105pub enum ReplayError {
106 #[error("log: {0}")]
107 Log(#[from] LogError),
108 #[error("store: {0}")]
109 Store(#[from] StoreError),
110}
111
112/// A reconcile rebuilds a stale store from the log. Either the wipe step
113/// or the replay step can fail.
114#[derive(Debug, Error)]
115pub enum ReconcileError {
116 #[error("clearing store before replay failed: {0}")]
117 Clear(#[source] StoreError),
118 #[error("replay into cleared store failed: {0}")]
119 Replay(#[from] ReplayError),
120}
121
122/// Outcome of `execute_and_log_with_recovery`. PreLog/LogAppend mirror the
123/// pre-WAL-fence variants of `ExecuteError` — the store is untouched and
124/// the caller can retry. `Unrecoverable` means the WAL fence was crossed
125/// (event is canonical on disk) but reconcile *also* failed: the operator
126/// must intervene before any further writes.
127#[derive(Debug, Error)]
128pub enum RecoverableExecuteError {
129 #[error("pre-log validation failed: {0}")]
130 PreLog(#[from] ExecError),
131 #[error("log append failed: {0}")]
132 LogAppend(#[from] LogError),
133 #[error(
134 "store apply failed AND reconcile failed — log is canonical, store is in an unknown state. apply: {post_log}; reconcile: {reconcile}"
135 )]
136 Unrecoverable {
137 #[source]
138 post_log: StoreError,
139 reconcile: ReconcileError,
140 },
141}
142
143#[derive(Debug, Error)]
144pub enum VerifyError {
145 #[error("log: {0}")]
146 Log(#[from] LogError),
147 #[error("morphism replay failed at seq {seq}: {source}")]
148 Exec {
149 seq: u64,
150 #[source]
151 source: ExecError,
152 },
153 #[error(
154 "non-determinism at seq {seq} morphism `{morphism}`: recomputed ops differ from logged ops"
155 )]
156 OpsMismatch {
157 seq: u64,
158 morphism: String,
159 expected: Vec<FieldOp>,
160 actual: Vec<FieldOp>,
161 },
162 /// The morphism was logged under a different schema/script bundle
163 /// than the one currently loaded. Re-executing it would (likely)
164 /// produce different ops, but the more specific signal is "the
165 /// rules changed since this was logged" — actionable: migrate the
166 /// log, or pin the executor to a compatible version.
167 #[error(
168 "schema mismatch at seq {seq} morphism `{morphism}`: logged schema_hash differs from current executor"
169 )]
170 SchemaMismatch {
171 seq: u64,
172 morphism: String,
173 logged: [u8; 32],
174 current: [u8; 32],
175 },
176 /// A `Seed` entry was logged under a different KCL bundle than the
177 /// one currently loaded. The seed's data may no longer fit the
178 /// entity definition. Coarser than `SchemaMismatch` (any change
179 /// to any schema file flips it, even one that doesn't affect the
180 /// seeded entity) but the operator still wants to know.
181 #[error(
182 "seed schema mismatch at seq {seq} entity `{entity}` id {id}: logged bundle hash differs from current executor"
183 )]
184 SeedSchemaMismatch {
185 seq: u64,
186 entity: String,
187 id: Uuid,
188 logged: [u8; 32],
189 current: [u8; 32],
190 },
191}
192
193pub struct EventLog {
194 path: PathBuf,
195 next_seq: u64,
196}
197
198impl EventLog {
199 /// Open or create a log at `path`. Reads existing entries to compute
200 /// `next_seq` and validate monotonicity. The first entry can start at
201 /// any seq (compacted logs are rooted at seq > 0); subsequent entries
202 /// must be strictly contiguous.
203 pub fn open(path: impl Into<PathBuf>) -> Result<Self, LogError> {
204 let path = path.into();
205 let mut next_seq: u64 = 0;
206 if path.exists() {
207 let entries = read_entries(&path)?;
208 let mut iter = entries.iter();
209 if let Some(first) = iter.next() {
210 next_seq = first.seq() + 1;
211 for e in iter {
212 if e.seq() != next_seq {
213 return Err(LogError::NonMonotonic {
214 got: e.seq(),
215 expected: next_seq,
216 });
217 }
218 next_seq = e.seq() + 1;
219 }
220 }
221 }
222 Ok(Self { path, next_seq })
223 }
224
225 pub fn next_seq(&self) -> u64 {
226 self.next_seq
227 }
228
229 pub fn path(&self) -> &Path {
230 &self.path
231 }
232
233 /// Append an entry. Calls `sync_all()` so the entry is durable on disk
234 /// before returning Ok — this is the WAL fence: by the time the caller
235 /// proceeds to mutate the store, the event is recoverable from a power
236 /// loss.
237 pub fn append(&mut self, entry: LogEntry) -> Result<(), LogError> {
238 if entry.seq() != self.next_seq {
239 return Err(LogError::NonMonotonic {
240 got: entry.seq(),
241 expected: self.next_seq,
242 });
243 }
244 let mut f = OpenOptions::new()
245 .create(true)
246 .append(true)
247 .open(&self.path)?;
248 let s = serde_json::to_string(&entry).expect("LogEntry serializes");
249 f.write_all(s.as_bytes())?;
250 f.write_all(b"\n")?;
251 f.sync_all()?;
252 self.next_seq += 1;
253 Ok(())
254 }
255
256 pub fn entries(&self) -> Result<Vec<LogEntry>, LogError> {
257 if !self.path.exists() {
258 return Ok(Vec::new());
259 }
260 read_entries(&self.path)
261 }
262
263 /// Truncate the log to drop entries with `seq <= through_seq`.
264 /// IRREVERSIBLE: caller must verify a Snapshot covering `through_seq`
265 /// exists on durable storage before calling this — once the entries
266 /// are gone, replay can only start from the snapshot.
267 ///
268 /// Atomic at the filesystem level: writes survivors to a sibling
269 /// tempfile then renames over the original.
270 pub fn compact_through(&mut self, through_seq: u64) -> Result<(), LogError> {
271 let survivors: Vec<LogEntry> = self
272 .entries()?
273 .into_iter()
274 .filter(|e| e.seq() > through_seq)
275 .collect();
276
277 let tmp = self.path.with_extension("compacting");
278 {
279 let mut f = std::fs::File::create(&tmp)?;
280 for e in &survivors {
281 let s = serde_json::to_string(e).expect("LogEntry serializes");
282 f.write_all(s.as_bytes())?;
283 f.write_all(b"\n")?;
284 }
285 f.sync_all()?;
286 }
287 std::fs::rename(&tmp, &self.path)?;
288 sync_parent_dir(&self.path)?;
289 Ok(())
290 }
291}
292
293/// Open and fsync the parent directory of `target`. After an atomic
294/// rename, the directory entry change isn't durable until the directory
295/// itself is fsynced — without this, a kernel/power crash between the
296/// rename and the next disk flush could leave the directory in a state
297/// where the rename never happened (depending on filesystem journal
298/// mode). With it, the rename survives.
299///
300/// Best-effort on platforms where opening a directory for sync isn't
301/// permitted: the syscalls are POSIX-portable across Linux, macOS, and
302/// the BSDs (the OSes Nakui targets), so this generally succeeds. A
303/// failure here is propagated as an IO error so the caller can choose
304/// to surface it; we prefer "loud" over "silent" for durability code.
305fn sync_parent_dir(target: &Path) -> std::io::Result<()> {
306 let parent = target.parent().unwrap_or_else(|| Path::new("."));
307 let dir = std::fs::File::open(parent)?;
308 dir.sync_all()
309}
310
311/// A snapshot of a `Store`'s state at a particular log seq. Lets us short-
312/// circuit replay: load the snapshot, then apply only the events with
313/// `seq > snapshot.seq`. MemoryStore-specific for V1 — backends that
314/// already persist (SurrealStore + RocksDB) don't need this layer.
315#[derive(Debug, Clone, Serialize, Deserialize)]
316pub struct Snapshot {
317 /// The last log seq this snapshot subsumes. `replay` resumes at seq+1.
318 pub seq: u64,
319 /// Full state at that seq, in MemoryStore's native shape.
320 pub records: HashMap<String, HashMap<Uuid, Value>>,
321 /// Module schema hash at capture time. `Some` for snapshots taken
322 /// via `capture(_, _, executor)`; `None` for those taken via the
323 /// hash-unaware `from_memory_store`. Loaders use this to refuse a
324 /// snapshot produced under a different bundle.
325 #[serde(default, skip_serializing_if = "Option::is_none")]
326 pub schema_hash: Option<[u8; 32]>,
327}
328
329#[derive(Debug, Error)]
330pub enum SnapshotMismatchError {
331 #[error(
332 "snapshot schema_hash differs from current executor; refusing to load (snapshot was taken under a different module bundle)"
333 )]
334 SchemaMismatch {
335 snapshot: [u8; 32],
336 current: [u8; 32],
337 },
338}
339
340impl Snapshot {
341 /// Capture the in-memory store's current state without binding to a
342 /// schema bundle. Test fixtures and ad-hoc tooling call this; the
343 /// production path uses `capture` so the snapshot can be validated
344 /// against the executor on load.
345 pub fn from_memory_store(store: &MemoryStore, seq: u64) -> Self {
346 Self {
347 seq,
348 records: store.records().clone(),
349 schema_hash: None,
350 }
351 }
352
353 /// Production capture: stamp the snapshot with the executor's
354 /// `module_schema_hash` so future loads can refuse a mismatch.
355 pub fn capture(store: &MemoryStore, seq: u64, executor: &Executor) -> Self {
356 Self {
357 seq,
358 records: store.records().clone(),
359 schema_hash: Some(executor.module_schema_hash()),
360 }
361 }
362
363 /// Verify the snapshot was produced under a bundle compatible with
364 /// `executor`. Snapshots without a hash (legacy / `from_memory_store`)
365 /// pass — the operator opted out of this check at capture time.
366 pub fn ensure_compatible_with(&self, executor: &Executor) -> Result<(), SnapshotMismatchError> {
367 let Some(snap_hash) = self.schema_hash else {
368 return Ok(());
369 };
370 let current = executor.module_schema_hash();
371 if snap_hash != current {
372 return Err(SnapshotMismatchError::SchemaMismatch {
373 snapshot: snap_hash,
374 current,
375 });
376 }
377 Ok(())
378 }
379
380 /// Atomically write the snapshot to `path`. Writes the bytes to a
381 /// sibling tempfile (`<path>.writing`), fsyncs, renames over the
382 /// target, then fsyncs the parent directory so the rename survives
383 /// a crash. A crash mid-write leaves either the previous snapshot
384 /// at `path` (rename never happened) or the new one (rename
385 /// completed and was durable) — never a truncated file. A stale
386 /// tempfile from a prior crash gets overwritten by `File::create`
387 /// on the next attempt, so writes are also self-healing.
388 pub fn write(&self, path: &Path) -> Result<(), LogError> {
389 let data = serde_json::to_vec_pretty(self).expect("snapshot serializes");
390 let tmp = path.with_extension("writing");
391 {
392 let mut f = std::fs::File::create(&tmp)?;
393 f.write_all(&data)?;
394 f.sync_all()?;
395 }
396 std::fs::rename(&tmp, path)?;
397 sync_parent_dir(path).map_err(LogError::Io)
398 }
399
400 pub fn load(path: &Path) -> Result<Option<Self>, LogError> {
401 if !path.exists() {
402 return Ok(None);
403 }
404 let text = std::fs::read_to_string(path).map_err(LogError::Io)?;
405 let snap: Snapshot =
406 serde_json::from_str(&text).map_err(|e| LogError::Parse { line: 0, source: e })?;
407 Ok(Some(snap))
408 }
409}
410
411fn read_entries(path: &Path) -> Result<Vec<LogEntry>, LogError> {
412 let f = std::fs::File::open(path)?;
413 let r = BufReader::new(f);
414 let mut out = Vec::new();
415 for (i, line) in r.lines().enumerate() {
416 let line = line?;
417 if line.trim().is_empty() {
418 continue;
419 }
420 let entry: LogEntry = serde_json::from_str(&line).map_err(|e| LogError::Parse {
421 line: i + 1,
422 source: e,
423 })?;
424 out.push(entry);
425 }
426 Ok(out)
427}
428
429/// Seed an entity into the store and persist the event.
430///
431/// WAL order: append to log *first*, then mutate the store. If the log
432/// append fails, the store is untouched and the caller can safely retry.
433/// `Store::seed` is infallible by trait contract — once the log entry is
434/// durable the store update is guaranteed to land for in-memory backends.
435/// For backends with fallible writes (network/disk), failures surface as
436/// a panic during `seed()`; callers that need a fallible seed path should
437/// wrap their own retry/reconcile loop.
438pub fn seed_and_log<S: Store>(
439 executor: &Executor,
440 store: &mut S,
441 log: &mut EventLog,
442 entity: &str,
443 id: Uuid,
444 data: Value,
445) -> Result<(), LogError> {
446 let seq = log.next_seq();
447 log.append(LogEntry::Seed {
448 seq,
449 entity: entity.to_string(),
450 id,
451 data: data.clone(),
452 schema_hash: Some(executor.schema_bundle_hash),
453 })?;
454 store.seed(entity, id, data);
455 // Best-effort: a failure here means next startup does an extra full
456 // replay, never a correctness issue.
457 let _ = store.set_last_applied_seq(seq);
458 Ok(())
459}
460
461/// Run a morphism and persist the event in WAL order:
462/// 1. compute() — pure, no mutation; full kernel validation incl. dry-run.
463/// 2. log.append() — event hits disk *before* the store changes.
464/// 3. store.apply() — materialize the change. By WAL semantics the log
465/// is now the source of truth: if (3) fails, the stale store can be
466/// rebuilt by replaying the log.
467///
468/// The error variants tell the caller exactly which stage failed so they
469/// know whether to retry, recover, or rebuild.
470pub fn execute_and_log<S: Store>(
471 executor: &Executor,
472 store: &mut S,
473 log: &mut EventLog,
474 morphism: &str,
475 inputs: &[(&str, Uuid)],
476 params: Value,
477) -> Result<Vec<FieldOp>, ExecuteError> {
478 let ops = executor.compute(store, morphism, inputs, params.clone())?;
479 let seq = log.next_seq();
480 let entry = LogEntry::Morphism {
481 seq,
482 morphism: morphism.to_string(),
483 inputs: inputs.iter().map(|(r, id)| (r.to_string(), *id)).collect(),
484 params,
485 ops: ops.clone(),
486 schema_hash: executor.schema_hash(morphism),
487 };
488 log.append(entry)?;
489 store.apply(&ops).map_err(ExecuteError::PostLogStore)?;
490 let _ = store.set_last_applied_seq(seq);
491 Ok(ops)
492}
493
494/// Rebuild a (possibly stale) store from the log. Wipes the store, then
495/// replays every event. Use this after a `PostLogStore` failure: the WAL
496/// fence guarantees the log is the source of truth, so a clean replay
497/// brings the store back into agreement with it.
498///
499/// `execute_and_log_with_recovery` automates this for the common case;
500/// reach for `reconcile` directly when an operator/CLI is doing the
501/// recovery, or when a backend reports drift detected out-of-band.
502pub fn reconcile<S: Store>(store: &mut S, log: &EventLog) -> Result<(), ReconcileError> {
503 store.clear().map_err(ReconcileError::Clear)?;
504 replay_into(log, store)?;
505 Ok(())
506}
507
508/// Like `execute_and_log`, but on `PostLogStore` automatically rebuilds
509/// the store from the log and returns the ops as if the apply had
510/// succeeded. The caller sees a consistent post-state — either the event
511/// landed cleanly, or it landed via reconcile, or `Unrecoverable` (which
512/// means even the rebuild failed and the store must not be trusted).
513///
514/// PreLog and LogAppend are forwarded verbatim: the WAL fence wasn't
515/// crossed, so there's nothing to reconcile.
516pub fn execute_and_log_with_recovery<S: Store>(
517 executor: &Executor,
518 store: &mut S,
519 log: &mut EventLog,
520 morphism: &str,
521 inputs: &[(&str, Uuid)],
522 params: Value,
523) -> Result<Vec<FieldOp>, RecoverableExecuteError> {
524 let ops = executor.compute(store, morphism, inputs, params.clone())?;
525 let seq = log.next_seq();
526 let entry = LogEntry::Morphism {
527 seq,
528 morphism: morphism.to_string(),
529 inputs: inputs.iter().map(|(r, id)| (r.to_string(), *id)).collect(),
530 params,
531 ops: ops.clone(),
532 schema_hash: executor.schema_hash(morphism),
533 };
534 log.append(entry)?;
535 if let Err(post_log) = store.apply(&ops) {
536 if let Err(reconcile) = reconcile(store, log) {
537 return Err(RecoverableExecuteError::Unrecoverable {
538 post_log,
539 reconcile,
540 });
541 }
542 // After reconcile the store reflects log state up to log.next_seq()-1
543 // (which equals our seq). The reconcile path itself updated the
544 // marker; nothing more to do here.
545 } else {
546 let _ = store.set_last_applied_seq(seq);
547 }
548 Ok(ops)
549}
550
551/// Replay the log into a caller-provided `Store`. The store should be
552/// empty on entry; existing records are not erased. Use this with any
553/// `Store` impl (MemoryStore, SurrealStore, future backends).
554pub fn replay_into<S: Store>(log: &EventLog, store: &mut S) -> Result<(), ReplayError> {
555 replay_with_snapshot_into(log, None, store)
556}
557
558/// Replay starting from a snapshot. If `snapshot` is `Some`, every record
559/// in it is seeded into `store` first, then events with `seq > snapshot.seq`
560/// are applied. The point: replay cost shrinks from O(events) to
561/// O(events_after_snapshot), useful when the log grows large.
562pub fn replay_with_snapshot_into<S: Store>(
563 log: &EventLog,
564 snapshot: Option<&Snapshot>,
565 store: &mut S,
566) -> Result<(), ReplayError> {
567 let start_seq = if let Some(snap) = snapshot {
568 for (entity, recs) in &snap.records {
569 for (id, data) in recs {
570 store.seed(entity, *id, data.clone());
571 }
572 }
573 snap.seq + 1
574 } else {
575 0
576 };
577
578 let mut last_applied: Option<u64> = snapshot.map(|s| s.seq);
579 for entry in log.entries()? {
580 if entry.seq() < start_seq {
581 continue;
582 }
583 let seq = entry.seq();
584 match entry {
585 LogEntry::Seed {
586 entity, id, data, ..
587 } => store.seed(&entity, id, data),
588 LogEntry::Morphism { ops, .. } => store.apply(&ops)?,
589 }
590 last_applied = Some(seq);
591 }
592 if let Some(seq) = last_applied {
593 let _ = store.set_last_applied_seq(seq);
594 }
595 Ok(())
596}
597
598/// Convenience: replay into a fresh `MemoryStore`. The fast path: O(events)
599/// with no Rhai execution.
600pub fn replay(log: &EventLog) -> Result<MemoryStore, ReplayError> {
601 let mut store = MemoryStore::new();
602 replay_into(log, &mut store)?;
603 Ok(store)
604}
605
606/// Re-execute every logged morphism through the kernel and assert the
607/// recomputed ops match the logged ops byte-for-byte. This is the
608/// determinism contract: if it ever fails, a morphism became impure.
609pub fn verify_log(log: &EventLog, executor: &Executor) -> Result<(), VerifyError> {
610 let mut store = MemoryStore::new();
611 for entry in log.entries()? {
612 match entry {
613 LogEntry::Seed {
614 seq,
615 entity,
616 id,
617 data,
618 schema_hash,
619 } => {
620 if let Some(logged_hash) = schema_hash {
621 let current_hash = executor.schema_bundle_hash;
622 if logged_hash != current_hash {
623 return Err(VerifyError::SeedSchemaMismatch {
624 seq,
625 entity,
626 id,
627 logged: logged_hash,
628 current: current_hash,
629 });
630 }
631 }
632 store.seed(&entity, id, data);
633 }
634 LogEntry::Morphism {
635 seq,
636 morphism,
637 inputs,
638 params,
639 ops: logged,
640 schema_hash,
641 } => {
642 // Schema check first: if the rules changed, re-execution
643 // is meaningless — it'd just surface as OpsMismatch with
644 // a less actionable message. Legacy entries with no
645 // hash predate the contract; we let those through.
646 if let Some(logged_hash) = schema_hash {
647 if let Some(current_hash) = executor.schema_hash(&morphism) {
648 if logged_hash != current_hash {
649 return Err(VerifyError::SchemaMismatch {
650 seq,
651 morphism,
652 logged: logged_hash,
653 current: current_hash,
654 });
655 }
656 }
657 }
658 let owned: Vec<(String, Uuid)> = inputs.into_iter().collect();
659 let refs: Vec<(&str, Uuid)> =
660 owned.iter().map(|(r, id)| (r.as_str(), *id)).collect();
661 let recomputed = executor
662 .run(&mut store, &morphism, &refs, params)
663 .map_err(|e| VerifyError::Exec { seq, source: e })?;
664 if recomputed != logged {
665 return Err(VerifyError::OpsMismatch {
666 seq,
667 morphism,
668 expected: logged,
669 actual: recomputed,
670 });
671 }
672 }
673 }
674 }
675 Ok(())
676}