⎇ 4d50bfc5879839a835e26ffda3bbc7ae3b927cc7
crates/modules/nakui/core/src/event_log.rs
Ñawpa willay
Pi imata llamkarqan
ch'uya
chore: absorbe nakui (ERP matemático) en modules/nakui
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(
367 &self,
368 executor: &Executor,
369 ) -> Result<(), SnapshotMismatchError> {
370 let Some(snap_hash) = self.schema_hash else {
371 return Ok(());
372 };
373 let current = executor.module_schema_hash();
374 if snap_hash != current {
375 return Err(SnapshotMismatchError::SchemaMismatch {
376 snapshot: snap_hash,
377 current,
378 });
379 }
380 Ok(())
381 }
382
383 /// Atomically write the snapshot to `path`. Writes the bytes to a
384 /// sibling tempfile (`<path>.writing`), fsyncs, renames over the
385 /// target, then fsyncs the parent directory so the rename survives
386 /// a crash. A crash mid-write leaves either the previous snapshot
387 /// at `path` (rename never happened) or the new one (rename
388 /// completed and was durable) — never a truncated file. A stale
389 /// tempfile from a prior crash gets overwritten by `File::create`
390 /// on the next attempt, so writes are also self-healing.
391 pub fn write(&self, path: &Path) -> Result<(), LogError> {
392 let data = serde_json::to_vec_pretty(self).expect("snapshot serializes");
393 let tmp = path.with_extension("writing");
394 {
395 let mut f = std::fs::File::create(&tmp)?;
396 f.write_all(&data)?;
397 f.sync_all()?;
398 }
399 std::fs::rename(&tmp, path)?;
400 sync_parent_dir(path).map_err(LogError::Io)
401 }
402
403 pub fn load(path: &Path) -> Result<Option<Self>, LogError> {
404 if !path.exists() {
405 return Ok(None);
406 }
407 let text = std::fs::read_to_string(path).map_err(LogError::Io)?;
408 let snap: Snapshot = serde_json::from_str(&text).map_err(|e| LogError::Parse {
409 line: 0,
410 source: e,
411 })?;
412 Ok(Some(snap))
413 }
414}
415
416fn read_entries(path: &Path) -> Result<Vec<LogEntry>, LogError> {
417 let f = std::fs::File::open(path)?;
418 let r = BufReader::new(f);
419 let mut out = Vec::new();
420 for (i, line) in r.lines().enumerate() {
421 let line = line?;
422 if line.trim().is_empty() {
423 continue;
424 }
425 let entry: LogEntry = serde_json::from_str(&line).map_err(|e| LogError::Parse {
426 line: i + 1,
427 source: e,
428 })?;
429 out.push(entry);
430 }
431 Ok(out)
432}
433
434/// Seed an entity into the store and persist the event.
435///
436/// WAL order: append to log *first*, then mutate the store. If the log
437/// append fails, the store is untouched and the caller can safely retry.
438/// `Store::seed` is infallible by trait contract — once the log entry is
439/// durable the store update is guaranteed to land for in-memory backends.
440/// For backends with fallible writes (network/disk), failures surface as
441/// a panic during `seed()`; callers that need a fallible seed path should
442/// wrap their own retry/reconcile loop.
443pub fn seed_and_log<S: Store>(
444 executor: &Executor,
445 store: &mut S,
446 log: &mut EventLog,
447 entity: &str,
448 id: Uuid,
449 data: Value,
450) -> Result<(), LogError> {
451 let seq = log.next_seq();
452 log.append(LogEntry::Seed {
453 seq,
454 entity: entity.to_string(),
455 id,
456 data: data.clone(),
457 schema_hash: Some(executor.schema_bundle_hash),
458 })?;
459 store.seed(entity, id, data);
460 // Best-effort: a failure here means next startup does an extra full
461 // replay, never a correctness issue.
462 let _ = store.set_last_applied_seq(seq);
463 Ok(())
464}
465
466/// Run a morphism and persist the event in WAL order:
467/// 1. compute() — pure, no mutation; full kernel validation incl. dry-run.
468/// 2. log.append() — event hits disk *before* the store changes.
469/// 3. store.apply() — materialize the change. By WAL semantics the log
470/// is now the source of truth: if (3) fails, the stale store can be
471/// rebuilt by replaying the log.
472///
473/// The error variants tell the caller exactly which stage failed so they
474/// know whether to retry, recover, or rebuild.
475pub fn execute_and_log<S: Store>(
476 executor: &Executor,
477 store: &mut S,
478 log: &mut EventLog,
479 morphism: &str,
480 inputs: &[(&str, Uuid)],
481 params: Value,
482) -> Result<Vec<FieldOp>, ExecuteError> {
483 let ops = executor.compute(store, morphism, inputs, params.clone())?;
484 let seq = log.next_seq();
485 let entry = LogEntry::Morphism {
486 seq,
487 morphism: morphism.to_string(),
488 inputs: inputs
489 .iter()
490 .map(|(r, id)| (r.to_string(), *id))
491 .collect(),
492 params,
493 ops: ops.clone(),
494 schema_hash: executor.schema_hash(morphism),
495 };
496 log.append(entry)?;
497 store.apply(&ops).map_err(ExecuteError::PostLogStore)?;
498 let _ = store.set_last_applied_seq(seq);
499 Ok(ops)
500}
501
502/// Rebuild a (possibly stale) store from the log. Wipes the store, then
503/// replays every event. Use this after a `PostLogStore` failure: the WAL
504/// fence guarantees the log is the source of truth, so a clean replay
505/// brings the store back into agreement with it.
506///
507/// `execute_and_log_with_recovery` automates this for the common case;
508/// reach for `reconcile` directly when an operator/CLI is doing the
509/// recovery, or when a backend reports drift detected out-of-band.
510pub fn reconcile<S: Store>(store: &mut S, log: &EventLog) -> Result<(), ReconcileError> {
511 store.clear().map_err(ReconcileError::Clear)?;
512 replay_into(log, store)?;
513 Ok(())
514}
515
516/// Like `execute_and_log`, but on `PostLogStore` automatically rebuilds
517/// the store from the log and returns the ops as if the apply had
518/// succeeded. The caller sees a consistent post-state — either the event
519/// landed cleanly, or it landed via reconcile, or `Unrecoverable` (which
520/// means even the rebuild failed and the store must not be trusted).
521///
522/// PreLog and LogAppend are forwarded verbatim: the WAL fence wasn't
523/// crossed, so there's nothing to reconcile.
524pub fn execute_and_log_with_recovery<S: Store>(
525 executor: &Executor,
526 store: &mut S,
527 log: &mut EventLog,
528 morphism: &str,
529 inputs: &[(&str, Uuid)],
530 params: Value,
531) -> Result<Vec<FieldOp>, RecoverableExecuteError> {
532 let ops = executor.compute(store, morphism, inputs, params.clone())?;
533 let seq = log.next_seq();
534 let entry = LogEntry::Morphism {
535 seq,
536 morphism: morphism.to_string(),
537 inputs: inputs
538 .iter()
539 .map(|(r, id)| (r.to_string(), *id))
540 .collect(),
541 params,
542 ops: ops.clone(),
543 schema_hash: executor.schema_hash(morphism),
544 };
545 log.append(entry)?;
546 if let Err(post_log) = store.apply(&ops) {
547 if let Err(reconcile) = reconcile(store, log) {
548 return Err(RecoverableExecuteError::Unrecoverable {
549 post_log,
550 reconcile,
551 });
552 }
553 // After reconcile the store reflects log state up to log.next_seq()-1
554 // (which equals our seq). The reconcile path itself updated the
555 // marker; nothing more to do here.
556 } else {
557 let _ = store.set_last_applied_seq(seq);
558 }
559 Ok(ops)
560}
561
562/// Replay the log into a caller-provided `Store`. The store should be
563/// empty on entry; existing records are not erased. Use this with any
564/// `Store` impl (MemoryStore, SurrealStore, future backends).
565pub fn replay_into<S: Store>(log: &EventLog, store: &mut S) -> Result<(), ReplayError> {
566 replay_with_snapshot_into(log, None, store)
567}
568
569/// Replay starting from a snapshot. If `snapshot` is `Some`, every record
570/// in it is seeded into `store` first, then events with `seq > snapshot.seq`
571/// are applied. The point: replay cost shrinks from O(events) to
572/// O(events_after_snapshot), useful when the log grows large.
573pub fn replay_with_snapshot_into<S: Store>(
574 log: &EventLog,
575 snapshot: Option<&Snapshot>,
576 store: &mut S,
577) -> Result<(), ReplayError> {
578 let start_seq = if let Some(snap) = snapshot {
579 for (entity, recs) in &snap.records {
580 for (id, data) in recs {
581 store.seed(entity, *id, data.clone());
582 }
583 }
584 snap.seq + 1
585 } else {
586 0
587 };
588
589 let mut last_applied: Option<u64> = snapshot.map(|s| s.seq);
590 for entry in log.entries()? {
591 if entry.seq() < start_seq {
592 continue;
593 }
594 let seq = entry.seq();
595 match entry {
596 LogEntry::Seed {
597 entity, id, data, ..
598 } => store.seed(&entity, id, data),
599 LogEntry::Morphism { ops, .. } => store.apply(&ops)?,
600 }
601 last_applied = Some(seq);
602 }
603 if let Some(seq) = last_applied {
604 let _ = store.set_last_applied_seq(seq);
605 }
606 Ok(())
607}
608
609/// Convenience: replay into a fresh `MemoryStore`. The fast path: O(events)
610/// with no Rhai execution.
611pub fn replay(log: &EventLog) -> Result<MemoryStore, ReplayError> {
612 let mut store = MemoryStore::new();
613 replay_into(log, &mut store)?;
614 Ok(store)
615}
616
617/// Re-execute every logged morphism through the kernel and assert the
618/// recomputed ops match the logged ops byte-for-byte. This is the
619/// determinism contract: if it ever fails, a morphism became impure.
620pub fn verify_log(log: &EventLog, executor: &Executor) -> Result<(), VerifyError> {
621 let mut store = MemoryStore::new();
622 for entry in log.entries()? {
623 match entry {
624 LogEntry::Seed {
625 seq,
626 entity,
627 id,
628 data,
629 schema_hash,
630 } => {
631 if let Some(logged_hash) = schema_hash {
632 let current_hash = executor.schema_bundle_hash;
633 if logged_hash != current_hash {
634 return Err(VerifyError::SeedSchemaMismatch {
635 seq,
636 entity,
637 id,
638 logged: logged_hash,
639 current: current_hash,
640 });
641 }
642 }
643 store.seed(&entity, id, data);
644 }
645 LogEntry::Morphism {
646 seq,
647 morphism,
648 inputs,
649 params,
650 ops: logged,
651 schema_hash,
652 } => {
653 // Schema check first: if the rules changed, re-execution
654 // is meaningless — it'd just surface as OpsMismatch with
655 // a less actionable message. Legacy entries with no
656 // hash predate the contract; we let those through.
657 if let Some(logged_hash) = schema_hash {
658 if let Some(current_hash) = executor.schema_hash(&morphism) {
659 if logged_hash != current_hash {
660 return Err(VerifyError::SchemaMismatch {
661 seq,
662 morphism,
663 logged: logged_hash,
664 current: current_hash,
665 });
666 }
667 }
668 }
669 let owned: Vec<(String, Uuid)> = inputs.into_iter().collect();
670 let refs: Vec<(&str, Uuid)> =
671 owned.iter().map(|(r, id)| (r.as_str(), *id)).collect();
672 let recomputed = executor
673 .run(&mut store, &morphism, &refs, params)
674 .map_err(|e| VerifyError::Exec { seq, source: e })?;
675 if recomputed != logged {
676 return Err(VerifyError::OpsMismatch {
677 seq,
678 morphism,
679 expected: logged,
680 actual: recomputed,
681 });
682 }
683 }
684 }
685 }
686 Ok(())
687}