arete-interpreter 0.22.3

AST transformation runtime and VM for Arete streaming pipelines
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

//! Snapshot DTOs for persisting VM state across restarts.
//!
//! The live VM structures (`VmContext`, `StateTable`) hold `DashMap`s,
//! `Mutex<LruCache>`s, and monotonic `Instant`s, so they cannot derive serde
//! directly. These DTOs are explicit, serde-friendly mirrors of the durable
//! subset of that state. Conversion lives on the live types
//! (`VmContext::dump`/`VmContext::hydrate`) so field privacy stays intact.
//!
//! LRU-backed collections are dumped most-recently-used first; hydration
//! inserts them in reverse so eviction order survives the round trip.
//!
//! Intentionally not captured (regenerated by replay from the resume
//! watermark): state-table access order (LRU eviction only), `pending_updates`,
//! `pending_instruction_events`, in-flight resolver requests, and scheduled
//! slot callbacks (documented as non-durable).

use crate::vm::{DeferredWhenOperation, PendingAccountUpdate};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;

/// One temporal-index key with its `(primary_key, timestamp)` history.
pub type TemporalIndexEntries = Vec<(String, Vec<(Value, i64)>)>;

/// Version of the snapshot payload layout. Bump on any incompatible change
/// to these DTOs; readers discard snapshots with a different version.
pub const SNAPSHOT_FORMAT_VERSION: u32 = 1;

/// Serializable dump of one `VmContext`.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct VmSnapshot {
    /// Per-state-id table dumps.
    pub states: HashMap<u32, StateTableSnapshot>,
    /// Resolver cache entries, most-recently-used first. Entries carry an
    /// absolute wall-clock expiry so the existing TTLs are honored across
    /// the restart; hydration drops entries that lapsed during downtime.
    #[serde(default)]
    pub resolver_cache: Vec<ResolverCacheEntrySnapshot>,
}

/// One resolver cache entry with its remaining lifetime pinned to wall clock.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResolverCacheEntrySnapshot {
    pub key: String,
    /// `None` encodes a negative-cache entry.
    pub value: Option<Value>,
    /// Absolute expiry in milliseconds since the Unix epoch.
    pub expires_at_epoch_ms: u64,
}

/// Serializable dump of one `StateTable`.
///
/// The version tracker and instruction dedup entries are required for
/// correctness: they make overlap replay after restore idempotent.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct StateTableSnapshot {
    pub entity_name: String,
    pub data: Vec<(Value, Value)>,
    /// Per index name: `(cache_key, primary_key)` pairs, MRU first.
    #[serde(default)]
    pub lookup_indexes: HashMap<String, Vec<(String, Value)>>,
    /// Per index name: `(cache_key, [(primary_key, timestamp)])`, MRU first.
    #[serde(default)]
    pub temporal_indexes: HashMap<String, TemporalIndexEntries>,
    /// Per index name: `(pda_address, seed_value)` pairs, MRU first.
    #[serde(default)]
    pub pda_reverse_lookups: HashMap<String, Vec<(String, String)>>,
    /// Most recent account data per PDA, used for PDA-remap reprocessing.
    #[serde(default)]
    pub last_account_data: Vec<(String, PendingAccountUpdate)>,
    /// `(key, slot, ordering_value)` entries, MRU first.
    #[serde(default)]
    pub version_tracker: Vec<(String, u64, u64)>,
    /// `(key, slot, txn_index)` entries, MRU first.
    #[serde(default)]
    pub instruction_dedup_cache: Vec<(String, u64, u64)>,
    /// `(signature, instruction names)` entries, MRU first.
    #[serde(default)]
    pub recent_tx_instructions: Vec<(String, Vec<String>)>,
    #[serde(default)]
    pub deferred_when_ops: Vec<((String, String), Vec<DeferredWhenOperation>)>,
}

impl VmSnapshot {
    /// Total number of entity rows across all state tables.
    pub fn total_entries(&self) -> usize {
        self.states.values().map(|table| table.data.len()).sum()
    }

    /// Per-entity row counts, keyed by entity name.
    pub fn entry_counts(&self) -> HashMap<String, u64> {
        self.states
            .values()
            .map(|table| (table.entity_name.clone(), table.data.len() as u64))
            .collect()
    }
}