1use crate::ast::{
2 self, BinaryOp, ComparisonOp, ComputedExpr, ComputedFieldSpec, FieldPath, ResolveStrategy,
3 ResolverExtractSpec, ResolverType, Transformation,
4};
5use crate::compiler::{MultiEntityBytecode, OpCode};
6use crate::debugger::{VmDebugEvent, VmDebugger, VmLookupHop};
7use crate::Mutation;
8use dashmap::DashMap;
9use lru::LruCache;
10use once_cell::sync::Lazy;
11use serde_json::{json, Value};
12use std::collections::{HashMap, HashSet, VecDeque};
13use std::num::NonZeroUsize;
14use std::sync::Arc;
15use std::time::{Duration, Instant};
16
17#[cfg(feature = "otel")]
18use tracing::instrument;
19#[derive(Debug, Clone, Default)]
22pub struct UpdateContext {
23 pub slot: Option<u64>,
25 pub signature: Option<String>,
27 pub timestamp: Option<i64>,
30 pub write_version: Option<u64>,
33 pub txn_index: Option<u64>,
36 pub skip_resolvers: bool,
40 pub metadata: HashMap<String, Value>,
42}
43
44impl UpdateContext {
45 pub fn new(slot: u64, signature: String) -> Self {
47 Self {
48 slot: Some(slot),
49 signature: Some(signature),
50 timestamp: None,
51 write_version: None,
52 txn_index: None,
53 skip_resolvers: false,
54 metadata: HashMap::new(),
55 }
56 }
57
58 pub fn with_timestamp(slot: u64, signature: String, timestamp: i64) -> Self {
60 Self {
61 slot: Some(slot),
62 signature: Some(signature),
63 timestamp: Some(timestamp),
64 write_version: None,
65 txn_index: None,
66 skip_resolvers: false,
67 metadata: HashMap::new(),
68 }
69 }
70
71 pub fn new_account(slot: u64, signature: String, write_version: u64) -> Self {
73 Self {
74 slot: Some(slot),
75 signature: Some(signature),
76 timestamp: None,
77 write_version: Some(write_version),
78 txn_index: None,
79 skip_resolvers: false,
80 metadata: HashMap::new(),
81 }
82 }
83
84 pub fn new_instruction(slot: u64, signature: String, txn_index: u64) -> Self {
86 Self {
87 slot: Some(slot),
88 signature: Some(signature),
89 timestamp: None,
90 write_version: None,
91 txn_index: Some(txn_index),
92 skip_resolvers: false,
93 metadata: HashMap::new(),
94 }
95 }
96
97 pub fn new_reprocessed(slot: u64, write_version: u64) -> Self {
101 Self {
102 slot: Some(slot),
103 signature: None,
104 timestamp: None,
105 write_version: Some(write_version),
106 txn_index: None,
107 skip_resolvers: true,
108 metadata: HashMap::new(),
109 }
110 }
111
112 pub fn timestamp(&self) -> i64 {
114 self.timestamp.unwrap_or_else(|| {
115 std::time::SystemTime::now()
116 .duration_since(std::time::UNIX_EPOCH)
117 .unwrap()
118 .as_secs() as i64
119 })
120 }
121
122 pub fn empty() -> Self {
124 Self::default()
125 }
126
127 pub fn is_account_update(&self) -> bool {
130 self.write_version.is_some() && self.txn_index.is_none()
131 }
132
133 pub fn is_instruction_update(&self) -> bool {
135 self.txn_index.is_some() && self.write_version.is_none()
136 }
137
138 pub fn with_metadata(mut self, key: String, value: Value) -> Self {
139 self.metadata.insert(key, value);
140 self
141 }
142
143 pub fn get_metadata(&self, key: &str) -> Option<&Value> {
145 self.metadata.get(key)
146 }
147
148 pub fn to_value(&self) -> Value {
150 let mut obj = serde_json::Map::new();
151 if let Some(slot) = self.slot {
152 obj.insert("slot".to_string(), json!(slot));
153 }
154 if let Some(ref sig) = self.signature {
155 obj.insert("signature".to_string(), json!(sig));
156 }
157 obj.insert("timestamp".to_string(), json!(self.timestamp()));
159 for (key, value) in &self.metadata {
160 obj.insert(key.clone(), value.clone());
161 }
162 Value::Object(obj)
163 }
164}
165
166pub type Register = usize;
167pub type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
168pub type ComputedEvaluatorResult =
169 std::result::Result<(), Box<dyn std::error::Error + Send + Sync>>;
170
171pub type RegisterValue = Value;
172
173pub trait ComputedFieldsEvaluator {
176 fn evaluate(&self, state: &mut Value) -> Result<()>;
177}
178
179const MAX_PENDING_UPDATES_TOTAL: usize = 2_500;
181const MAX_PENDING_UPDATES_PER_PDA: usize = 50;
182const PENDING_UPDATE_TTL_SECONDS: i64 = 300; const TEMPORAL_HISTORY_TTL_SECONDS: i64 = 300; const MAX_TEMPORAL_ENTRIES_PER_KEY: usize = 250;
187
188const DEFAULT_MAX_STATE_TABLE_ENTRIES: usize = 2_500;
190const DEFAULT_MAX_ARRAY_LENGTH: usize = 100;
191
192const DEFAULT_MAX_LOOKUP_INDEX_ENTRIES: usize = 2_500;
193
194const DEFAULT_MAX_VERSION_TRACKER_ENTRIES: usize = 2_500;
195
196const DEFAULT_MAX_INSTRUCTION_DEDUP_ENTRIES: usize = 500;
199
200const DEFAULT_MAX_TEMPORAL_INDEX_KEYS: usize = 2_500;
201
202const DEFAULT_MAX_PDA_REVERSE_LOOKUP_ENTRIES: usize = 2_500;
203
204const DEFAULT_MAX_RESOLVER_CACHE_ENTRIES: usize = 20_000;
205const DEFAULT_RESOLVER_CACHE_TTL_SECS: u64 = 3600; const DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS: u64 = 30;
207const DEFAULT_RESOLVER_RETRY_BACKOFF_SECS: u64 = 2;
208const DEFAULT_RESOLVER_MAX_RETRIES: u32 = 1;
209
210static RESOLVER_CACHE_CAPACITY: Lazy<NonZeroUsize> = Lazy::new(|| {
211 NonZeroUsize::new(
212 std::env::var("ARETE_RESOLVER_CACHE_CAPACITY")
213 .ok()
214 .and_then(|value| value.parse::<usize>().ok())
215 .filter(|value| *value > 0)
216 .unwrap_or(DEFAULT_MAX_RESOLVER_CACHE_ENTRIES),
217 )
218 .expect("resolver cache capacity must be > 0")
219});
220
221static RESOLVER_CACHE_TTL: Lazy<Duration> = Lazy::new(|| {
222 let ttl_secs = match std::env::var("ARETE_RESOLVER_CACHE_TTL_SECS") {
223 Ok(value) => match value.parse::<u64>() {
224 Ok(0) => {
225 tracing::warn!(
226 default_ttl_secs = DEFAULT_RESOLVER_CACHE_TTL_SECS,
227 "ARETE_RESOLVER_CACHE_TTL_SECS=0 is not supported; using default"
228 );
229 DEFAULT_RESOLVER_CACHE_TTL_SECS
230 }
231 Ok(value) => value,
232 Err(_) => DEFAULT_RESOLVER_CACHE_TTL_SECS,
233 },
234 Err(_) => DEFAULT_RESOLVER_CACHE_TTL_SECS,
235 };
236
237 Duration::from_secs(ttl_secs)
238});
239
240static NEGATIVE_RESOLVER_CACHE_TTL: Lazy<Duration> = Lazy::new(|| {
241 let ttl_secs = match std::env::var("ARETE_NEGATIVE_RESOLVER_CACHE_TTL_SECS") {
242 Ok(value) => match value.parse::<u64>() {
243 Ok(0) => {
244 tracing::warn!(
245 default_ttl_secs = DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS,
246 "ARETE_NEGATIVE_RESOLVER_CACHE_TTL_SECS=0 is not supported; using default"
247 );
248 DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS
249 }
250 Ok(value) => value,
251 Err(_) => DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS,
252 },
253 Err(_) => DEFAULT_NEGATIVE_RESOLVER_CACHE_TTL_SECS,
254 };
255
256 Duration::from_secs(ttl_secs)
257});
258
259static RESOLVER_RETRY_BACKOFF: Lazy<Duration> = Lazy::new(|| {
260 let secs = std::env::var("ARETE_RESOLVER_RETRY_BACKOFF_SECS")
261 .ok()
262 .and_then(|value| value.parse::<u64>().ok())
263 .filter(|value| *value > 0)
264 .unwrap_or(DEFAULT_RESOLVER_RETRY_BACKOFF_SECS);
265 Duration::from_secs(secs)
266});
267
268static RESOLVER_MAX_RETRIES: Lazy<u32> = Lazy::new(|| {
269 std::env::var("ARETE_RESOLVER_MAX_RETRIES")
270 .ok()
271 .and_then(|value| value.parse::<u32>().ok())
272 .unwrap_or(DEFAULT_RESOLVER_MAX_RETRIES)
273});
274
275#[derive(Debug, Clone)]
276pub(crate) enum CachedResolverValue {
277 Resolved(Value),
278 Negative,
279}
280
281#[derive(Debug, Clone)]
282enum ResolverCacheValue {
283 Resolved(Value),
284 Negative,
285}
286
287#[derive(Debug, Clone)]
288struct ResolverCacheEntry {
289 value: ResolverCacheValue,
290 cached_at: Instant,
291 ttl: Duration,
292}
293
294fn resolver_cache_capacity() -> NonZeroUsize {
295 *RESOLVER_CACHE_CAPACITY
296}
297
298fn resolver_cache_ttl() -> Duration {
299 *RESOLVER_CACHE_TTL
300}
301
302fn negative_resolver_cache_ttl() -> Duration {
303 *NEGATIVE_RESOLVER_CACHE_TTL
304}
305
306fn resolver_retry_backoff() -> Duration {
307 *RESOLVER_RETRY_BACKOFF
308}
309
310fn resolver_max_retries() -> u32 {
311 *RESOLVER_MAX_RETRIES
312}
313
314#[derive(Debug, Clone, Copy, PartialEq, Eq)]
315pub(crate) enum TokenResolverMissAction {
316 RetryScheduled,
317 NegativeCached,
318 Dropped,
319}
320
321fn estimate_json_size(value: &Value) -> usize {
323 match value {
324 Value::Null => 4,
325 Value::Bool(_) => 5,
326 Value::Number(_) => 8,
327 Value::String(s) => s.len() + 2,
328 Value::Array(arr) => 2 + arr.iter().map(|v| estimate_json_size(v) + 1).sum::<usize>(),
329 Value::Object(obj) => {
330 2 + obj
331 .iter()
332 .map(|(k, v)| k.len() + 3 + estimate_json_size(v) + 1)
333 .sum::<usize>()
334 }
335 }
336}
337
338#[derive(Debug, Clone)]
339pub struct CompiledPath {
340 pub segments: std::sync::Arc<[String]>,
341}
342
343impl CompiledPath {
344 pub fn new(path: &str) -> Self {
345 let segments: Vec<String> = path.split('.').map(|s| s.to_string()).collect();
346 CompiledPath {
347 segments: segments.into(),
348 }
349 }
350
351 fn segments(&self) -> &[String] {
352 &self.segments
353 }
354}
355
356#[derive(Debug, Clone)]
359pub enum FieldChange {
360 Replaced,
362 Appended(Vec<Value>),
364}
365
366#[derive(Debug, Clone, Default)]
369pub struct DirtyTracker {
370 changes: HashMap<String, FieldChange>,
371}
372
373impl DirtyTracker {
374 pub fn new() -> Self {
376 Self {
377 changes: HashMap::new(),
378 }
379 }
380
381 pub fn mark_replaced(&mut self, path: &str) {
383 self.changes.insert(path.to_string(), FieldChange::Replaced);
385 }
386
387 pub fn mark_appended(&mut self, path: &str, value: Value) {
389 match self.changes.get_mut(path) {
390 Some(FieldChange::Appended(values)) => {
391 values.push(value);
393 }
394 Some(FieldChange::Replaced) => {
395 }
398 None => {
399 self.changes
401 .insert(path.to_string(), FieldChange::Appended(vec![value]));
402 }
403 }
404 }
405
406 pub fn is_empty(&self) -> bool {
408 self.changes.is_empty()
409 }
410
411 pub fn len(&self) -> usize {
413 self.changes.len()
414 }
415
416 pub fn iter(&self) -> impl Iterator<Item = (&String, &FieldChange)> {
418 self.changes.iter()
419 }
420
421 pub fn dirty_paths(&self) -> HashSet<String> {
423 self.changes.keys().cloned().collect()
424 }
425
426 pub fn into_changes(self) -> HashMap<String, FieldChange> {
428 self.changes
429 }
430
431 pub fn changes(&self) -> &HashMap<String, FieldChange> {
433 &self.changes
434 }
435
436 pub fn appended_paths(&self) -> Vec<String> {
438 self.changes
439 .iter()
440 .filter_map(|(path, change)| match change {
441 FieldChange::Appended(_) => Some(path.clone()),
442 FieldChange::Replaced => None,
443 })
444 .collect()
445 }
446}
447
448pub struct VmContext {
449 registers: Vec<RegisterValue>,
450 states: HashMap<u32, StateTable>,
451 debugger: Option<Arc<dyn VmDebugger>>,
452 pub instructions_executed: u64,
453 pub cache_hits: u64,
454 path_cache: HashMap<String, CompiledPath>,
455 pub pda_cache_hits: u64,
456 pub pda_cache_misses: u64,
457 pub pending_queue_size: u64,
458 resolver_requests: VecDeque<ResolverRequest>,
459 resolver_pending: HashMap<String, PendingResolverEntry>,
460 resolver_cache: LruCache<String, ResolverCacheEntry>,
461 pub resolver_cache_hits: u64,
462 pub resolver_cache_misses: u64,
463 current_context: Option<UpdateContext>,
464 warnings: Vec<String>,
465 last_pda_lookup_miss: Option<String>,
466 last_lookup_index_miss: Option<String>,
467 last_pda_registered: Option<String>,
468 last_lookup_index_keys: Vec<String>,
469 pending_pda_reprocess_updates: Vec<PendingAccountUpdate>,
470 scheduled_callbacks: Vec<(u64, ScheduledCallback)>,
471}
472
473#[derive(Debug)]
474pub struct LookupIndex {
475 index: std::sync::Mutex<LruCache<String, Value>>,
476}
477
478impl LookupIndex {
479 pub fn new() -> Self {
480 Self::with_capacity(DEFAULT_MAX_LOOKUP_INDEX_ENTRIES)
481 }
482
483 pub fn with_capacity(capacity: usize) -> Self {
484 LookupIndex {
485 index: std::sync::Mutex::new(LruCache::new(
486 NonZeroUsize::new(capacity).expect("capacity must be > 0"),
487 )),
488 }
489 }
490
491 pub fn lookup(&self, lookup_value: &Value) -> Option<Value> {
492 let key = value_to_cache_key(lookup_value);
493 self.index.lock().unwrap().get(&key).cloned()
494 }
495
496 pub fn insert(&self, lookup_value: Value, primary_key: Value) {
497 let key = value_to_cache_key(&lookup_value);
498 self.index.lock().unwrap().put(key, primary_key);
499 }
500
501 pub fn remove(&self, lookup_value: &Value) {
504 let key = value_to_cache_key(lookup_value);
505 self.index.lock().unwrap().pop(&key);
506 }
507
508 pub fn len(&self) -> usize {
509 self.index.lock().unwrap().len()
510 }
511
512 pub fn is_empty(&self) -> bool {
513 self.index.lock().unwrap().is_empty()
514 }
515}
516
517impl Default for LookupIndex {
518 fn default() -> Self {
519 Self::new()
520 }
521}
522
523fn value_to_cache_key(value: &Value) -> String {
524 match value {
525 Value::String(s) => s.clone(),
526 Value::Number(n) => n.to_string(),
527 Value::Bool(b) => b.to_string(),
528 Value::Null => "null".to_string(),
529 _ => serde_json::to_string(value).unwrap_or_else(|_| "unknown".to_string()),
530 }
531}
532
533pub(crate) fn resolver_cache_key(resolver: &ResolverType, input: &Value) -> String {
534 match resolver {
535 ResolverType::Token => format!("token:{}", value_to_cache_key(input)),
536 ResolverType::Url(config) => {
537 let method = match config.method {
538 ast::HttpMethod::Get => "get",
539 ast::HttpMethod::Post => "post",
540 };
541 format!("url:{}:{}", method, value_to_cache_key(input))
542 }
543 }
544}
545
546#[derive(Debug)]
547pub struct TemporalIndex {
548 index: std::sync::Mutex<LruCache<String, Vec<(Value, i64)>>>,
549}
550
551impl Default for TemporalIndex {
552 fn default() -> Self {
553 Self::new()
554 }
555}
556
557impl TemporalIndex {
558 pub fn new() -> Self {
559 Self::with_capacity(DEFAULT_MAX_TEMPORAL_INDEX_KEYS)
560 }
561
562 pub fn with_capacity(capacity: usize) -> Self {
563 TemporalIndex {
564 index: std::sync::Mutex::new(LruCache::new(
565 NonZeroUsize::new(capacity).expect("capacity must be > 0"),
566 )),
567 }
568 }
569
570 pub fn lookup(&self, lookup_value: &Value, timestamp: i64) -> Option<Value> {
571 let key = value_to_cache_key(lookup_value);
572 let mut cache = self.index.lock().unwrap();
573 if let Some(entries) = cache.get(&key) {
574 for i in (0..entries.len()).rev() {
575 if entries[i].1 <= timestamp {
576 return Some(entries[i].0.clone());
577 }
578 }
579 }
580 None
581 }
582
583 pub fn lookup_latest(&self, lookup_value: &Value) -> Option<Value> {
584 let key = value_to_cache_key(lookup_value);
585 let mut cache = self.index.lock().unwrap();
586 if let Some(entries) = cache.get(&key) {
587 if let Some(last) = entries.last() {
588 return Some(last.0.clone());
589 }
590 }
591 None
592 }
593
594 pub fn insert(&self, lookup_value: Value, primary_key: Value, timestamp: i64) {
595 let key = value_to_cache_key(&lookup_value);
596 let mut cache = self.index.lock().unwrap();
597
598 let entries = cache.get_or_insert_mut(key, Vec::new);
599 entries.push((primary_key, timestamp));
600 entries.sort_by_key(|(_, ts)| *ts);
601
602 let cutoff = timestamp - TEMPORAL_HISTORY_TTL_SECONDS;
603 entries.retain(|(_, ts)| *ts >= cutoff);
604
605 if entries.len() > MAX_TEMPORAL_ENTRIES_PER_KEY {
606 let excess = entries.len() - MAX_TEMPORAL_ENTRIES_PER_KEY;
607 entries.drain(0..excess);
608 }
609 }
610
611 pub fn len(&self) -> usize {
612 self.index.lock().unwrap().len()
613 }
614
615 pub fn is_empty(&self) -> bool {
616 self.index.lock().unwrap().is_empty()
617 }
618
619 pub fn total_entries(&self) -> usize {
620 self.index
621 .lock()
622 .unwrap()
623 .iter()
624 .map(|(_, entries)| entries.len())
625 .sum()
626 }
627
628 pub fn cleanup_expired(&self, cutoff_timestamp: i64) -> usize {
629 let mut cache = self.index.lock().unwrap();
630 let mut total_removed = 0;
631
632 for (_, entries) in cache.iter_mut() {
633 let original_len = entries.len();
634 entries.retain(|(_, ts)| *ts >= cutoff_timestamp);
635 total_removed += original_len - entries.len();
636 }
637
638 total_removed
639 }
640}
641
642#[derive(Debug)]
643pub struct PdaReverseLookup {
644 index: LruCache<String, String>,
646}
647
648impl PdaReverseLookup {
649 pub fn new(capacity: usize) -> Self {
650 PdaReverseLookup {
651 index: LruCache::new(NonZeroUsize::new(capacity).unwrap()),
652 }
653 }
654
655 pub fn lookup(&mut self, pda_address: &str) -> Option<String> {
656 self.index.get(pda_address).cloned()
657 }
658
659 pub fn insert(&mut self, pda_address: String, seed_value: String) -> Option<String> {
660 let evicted = if self.index.len() >= self.index.cap().get() {
661 self.index.peek_lru().map(|(k, _)| k.clone())
662 } else {
663 None
664 };
665
666 self.index.put(pda_address, seed_value);
667 evicted
668 }
669
670 pub fn len(&self) -> usize {
671 self.index.len()
672 }
673
674 pub fn is_empty(&self) -> bool {
675 self.index.is_empty()
676 }
677
678 pub fn contains(&self, pda_address: &str) -> bool {
679 self.index.peek(pda_address).is_some()
680 }
681}
682
683#[derive(Debug, Clone)]
685pub struct QueuedAccountUpdate {
686 pub pda_address: String,
687 pub account_type: String,
688 pub account_data: Value,
689 pub slot: u64,
690 pub write_version: u64,
691 pub signature: String,
692}
693
694#[derive(Debug, Clone)]
696pub struct PendingAccountUpdate {
697 pub account_type: String,
698 pub pda_address: String,
699 pub account_data: Value,
700 pub slot: u64,
701 pub write_version: u64,
702 pub signature: String,
703 pub queued_at: i64,
704 pub is_stale_reprocess: bool,
710}
711
712#[derive(Debug, Clone)]
714pub struct QueuedInstructionEvent {
715 pub pda_address: String,
716 pub event_type: String,
717 pub event_data: Value,
718 pub slot: u64,
719 pub signature: String,
720}
721
722#[derive(Debug, Clone)]
724pub struct PendingInstructionEvent {
725 pub event_type: String,
726 pub pda_address: String,
727 pub event_data: Value,
728 pub slot: u64,
729 pub signature: String,
730 pub queued_at: i64,
731}
732
733#[derive(Debug, Clone)]
734pub struct DeferredWhenOperation {
735 pub entity_name: String,
736 pub primary_key: Value,
737 pub field_path: String,
738 pub field_value: Value,
739 pub when_instruction: String,
740 pub signature: String,
741 pub slot: u64,
742 pub deferred_at: i64,
743 pub emit: bool,
744}
745
746#[derive(Debug, Clone)]
747pub struct ResolverRequest {
748 pub cache_key: String,
749 pub resolver: ResolverType,
750 pub input: Value,
751}
752
753#[derive(Debug, Clone)]
754pub struct ResolverTarget {
755 pub state_id: u32,
756 pub entity_name: String,
757 pub primary_key: Value,
758 pub extracts: Vec<ResolverExtractSpec>,
759}
760
761#[derive(Debug, Clone)]
762pub struct PendingResolverEntry {
763 pub resolver: ResolverType,
764 pub input: Value,
765 pub targets: Vec<ResolverTarget>,
766 pub queued_at: i64,
767 pub next_retry_at: Instant,
768 pub retry_count: u32,
769 pub queued: bool,
770 pub in_flight: bool,
771}
772
773#[derive(Debug, Clone, PartialEq)]
774pub struct ScheduledCallback {
775 pub state_id: u32,
776 pub entity_name: String,
777 pub primary_key: Value,
778 pub resolver: ResolverType,
779 pub url_template: Option<Vec<ast::UrlTemplatePart>>,
780 pub input_value: Option<Value>,
781 pub input_path: Option<String>,
782 pub condition: Option<ast::ResolverCondition>,
783 pub strategy: ResolveStrategy,
784 pub extracts: Vec<ResolverExtractSpec>,
785 pub retry_count: u32,
786}
787
788impl PendingResolverEntry {
789 fn add_target(&mut self, target: ResolverTarget) {
790 if let Some(existing) = self.targets.iter_mut().find(|t| {
791 t.state_id == target.state_id
792 && t.entity_name == target.entity_name
793 && t.primary_key == target.primary_key
794 }) {
795 let mut seen = HashSet::new();
796 for extract in &existing.extracts {
797 seen.insert((extract.target_path.clone(), extract.source_path.clone()));
798 }
799
800 for extract in target.extracts {
801 let key = (extract.target_path.clone(), extract.source_path.clone());
802 if seen.insert(key) {
803 existing.extracts.push(extract);
804 }
805 }
806 } else {
807 self.targets.push(target);
808 }
809 }
810
811 fn ready_to_queue(&self) -> bool {
812 !self.queued && !self.in_flight && Instant::now() >= self.next_retry_at
813 }
814}
815
816#[derive(Debug, Clone)]
817pub struct PendingQueueStats {
818 pub total_updates: usize,
819 pub unique_pdas: usize,
820 pub oldest_age_seconds: i64,
821 pub largest_pda_queue_size: usize,
822 pub estimated_memory_bytes: usize,
823}
824
825#[derive(Debug, Clone, Default)]
826pub struct VmMemoryStats {
827 pub state_table_entity_count: usize,
828 pub state_table_max_entries: usize,
829 pub state_table_at_capacity: bool,
830 pub lookup_index_count: usize,
831 pub lookup_index_total_entries: usize,
832 pub temporal_index_count: usize,
833 pub temporal_index_total_entries: usize,
834 pub pda_reverse_lookup_count: usize,
835 pub pda_reverse_lookup_total_entries: usize,
836 pub version_tracker_entries: usize,
837 pub pending_queue_stats: Option<PendingQueueStats>,
838 pub path_cache_size: usize,
839}
840
841#[derive(Debug, Clone, Default)]
842pub struct CleanupResult {
843 pub pending_updates_removed: usize,
844 pub temporal_entries_removed: usize,
845}
846
847#[derive(Debug, Clone)]
848pub struct CapacityWarning {
849 pub current_entries: usize,
850 pub max_entries: usize,
851 pub entries_over_limit: usize,
852}
853
854#[derive(Debug, Clone)]
855pub struct StateTableConfig {
856 pub max_entries: usize,
857 pub max_array_length: usize,
858}
859
860impl Default for StateTableConfig {
861 fn default() -> Self {
862 Self {
863 max_entries: DEFAULT_MAX_STATE_TABLE_ENTRIES,
864 max_array_length: DEFAULT_MAX_ARRAY_LENGTH,
865 }
866 }
867}
868
869#[derive(Debug)]
870pub struct VersionTracker {
871 cache: std::sync::Mutex<LruCache<String, (u64, u64)>>,
872}
873
874impl VersionTracker {
875 pub fn new() -> Self {
876 Self::with_capacity(DEFAULT_MAX_VERSION_TRACKER_ENTRIES)
877 }
878
879 pub fn with_capacity(capacity: usize) -> Self {
880 VersionTracker {
881 cache: std::sync::Mutex::new(LruCache::new(
882 NonZeroUsize::new(capacity).expect("capacity must be > 0"),
883 )),
884 }
885 }
886
887 fn make_key(primary_key: &Value, event_type: &str) -> String {
888 format!("{}:{}", primary_key, event_type)
889 }
890
891 pub fn get(&self, primary_key: &Value, event_type: &str) -> Option<(u64, u64)> {
892 let key = Self::make_key(primary_key, event_type);
893 self.cache.lock().unwrap().get(&key).copied()
894 }
895
896 pub fn insert(&self, primary_key: &Value, event_type: &str, slot: u64, ordering_value: u64) {
897 let key = Self::make_key(primary_key, event_type);
898 self.cache.lock().unwrap().put(key, (slot, ordering_value));
899 }
900
901 pub fn len(&self) -> usize {
902 self.cache.lock().unwrap().len()
903 }
904
905 pub fn is_empty(&self) -> bool {
906 self.cache.lock().unwrap().is_empty()
907 }
908}
909
910impl Default for VersionTracker {
911 fn default() -> Self {
912 Self::new()
913 }
914}
915
916#[derive(Debug)]
917pub struct StateTable {
918 pub data: DashMap<Value, Value>,
919 access_times: DashMap<Value, i64>,
920 pub lookup_indexes: HashMap<String, LookupIndex>,
921 pub temporal_indexes: HashMap<String, TemporalIndex>,
922 pub pda_reverse_lookups: HashMap<String, PdaReverseLookup>,
923 pub pending_updates: DashMap<String, Vec<PendingAccountUpdate>>,
924 pub pending_instruction_events: DashMap<String, Vec<PendingInstructionEvent>>,
925 pub last_account_data: DashMap<String, PendingAccountUpdate>,
929 version_tracker: VersionTracker,
930 instruction_dedup_cache: VersionTracker,
931 config: StateTableConfig,
932 #[cfg_attr(not(feature = "otel"), allow(dead_code))]
933 entity_name: String,
934 pub recent_tx_instructions:
935 std::sync::Mutex<lru::LruCache<String, std::collections::HashSet<String>>>,
936 pub deferred_when_ops: DashMap<(String, String), Vec<DeferredWhenOperation>>,
937}
938
939impl StateTable {
940 pub fn is_at_capacity(&self) -> bool {
941 self.data.len() >= self.config.max_entries
942 }
943
944 pub fn entries_over_limit(&self) -> usize {
945 self.data.len().saturating_sub(self.config.max_entries)
946 }
947
948 pub fn max_array_length(&self) -> usize {
949 self.config.max_array_length
950 }
951
952 fn touch(&self, key: &Value) {
953 let now = std::time::SystemTime::now()
954 .duration_since(std::time::UNIX_EPOCH)
955 .unwrap()
956 .as_secs() as i64;
957 self.access_times.insert(key.clone(), now);
958 }
959
960 fn evict_lru(&self, count: usize) -> usize {
961 if count == 0 || self.data.is_empty() {
962 return 0;
963 }
964
965 let mut entries: Vec<(Value, i64)> = self
966 .access_times
967 .iter()
968 .map(|entry| (entry.key().clone(), *entry.value()))
969 .collect();
970
971 entries.sort_by_key(|(_, ts)| *ts);
972
973 let to_evict: Vec<Value> = entries.iter().take(count).map(|(k, _)| k.clone()).collect();
974
975 let mut evicted = 0;
976 for key in to_evict {
977 self.data.remove(&key);
978 self.access_times.remove(&key);
979 evicted += 1;
980 }
981
982 #[cfg(feature = "otel")]
983 if evicted > 0 {
984 crate::vm_metrics::record_state_table_eviction(evicted as u64, &self.entity_name);
985 }
986
987 evicted
988 }
989
990 pub fn insert_with_eviction(&self, key: Value, value: Value) {
991 if self.data.len() >= self.config.max_entries && !self.data.contains_key(&key) {
992 #[cfg(feature = "otel")]
993 crate::vm_metrics::record_state_table_at_capacity(&self.entity_name);
994 let to_evict = (self.data.len() + 1).saturating_sub(self.config.max_entries);
995 self.evict_lru(to_evict.max(1));
996 }
997 self.data.insert(key.clone(), value);
998 self.touch(&key);
999 }
1000
1001 pub fn get_and_touch(&self, key: &Value) -> Option<Value> {
1002 let result = self.data.get(key).map(|v| v.clone());
1003 if result.is_some() {
1004 self.touch(key);
1005 }
1006 result
1007 }
1008
1009 pub fn is_fresh_update(
1016 &self,
1017 primary_key: &Value,
1018 event_type: &str,
1019 slot: u64,
1020 ordering_value: u64,
1021 ) -> bool {
1022 let dominated = self
1023 .version_tracker
1024 .get(primary_key, event_type)
1025 .map(|(last_slot, last_version)| (slot, ordering_value) <= (last_slot, last_version))
1026 .unwrap_or(false);
1027
1028 if dominated {
1029 return false;
1030 }
1031
1032 self.version_tracker
1033 .insert(primary_key, event_type, slot, ordering_value);
1034 true
1035 }
1036
1037 pub fn is_duplicate_instruction(
1045 &self,
1046 primary_key: &Value,
1047 event_type: &str,
1048 slot: u64,
1049 txn_index: u64,
1050 ) -> bool {
1051 let is_duplicate = self
1053 .instruction_dedup_cache
1054 .get(primary_key, event_type)
1055 .map(|(last_slot, last_txn_index)| slot == last_slot && txn_index == last_txn_index)
1056 .unwrap_or(false);
1057
1058 if is_duplicate {
1059 return true;
1060 }
1061
1062 self.instruction_dedup_cache
1064 .insert(primary_key, event_type, slot, txn_index);
1065 false
1066 }
1067}
1068
1069impl VmContext {
1070 pub fn new() -> Self {
1071 let mut vm = VmContext {
1072 registers: vec![Value::Null; 256],
1073 states: HashMap::new(),
1074 debugger: None,
1075 instructions_executed: 0,
1076 cache_hits: 0,
1077 path_cache: HashMap::new(),
1078 pda_cache_hits: 0,
1079 pda_cache_misses: 0,
1080 pending_queue_size: 0,
1081 resolver_requests: VecDeque::new(),
1082 resolver_pending: HashMap::new(),
1083 resolver_cache: LruCache::new(resolver_cache_capacity()),
1084 resolver_cache_hits: 0,
1085 resolver_cache_misses: 0,
1086 current_context: None,
1087 warnings: Vec::new(),
1088 last_pda_lookup_miss: None,
1089 last_lookup_index_miss: None,
1090 last_pda_registered: None,
1091 last_lookup_index_keys: Vec::new(),
1092 pending_pda_reprocess_updates: Vec::new(),
1093 scheduled_callbacks: Vec::new(),
1094 };
1095 vm.states.insert(
1096 0,
1097 StateTable {
1098 data: DashMap::new(),
1099 access_times: DashMap::new(),
1100 lookup_indexes: HashMap::new(),
1101 temporal_indexes: HashMap::new(),
1102 pda_reverse_lookups: HashMap::new(),
1103 pending_updates: DashMap::new(),
1104 pending_instruction_events: DashMap::new(),
1105 last_account_data: DashMap::new(),
1106 version_tracker: VersionTracker::new(),
1107 instruction_dedup_cache: VersionTracker::with_capacity(
1108 DEFAULT_MAX_INSTRUCTION_DEDUP_ENTRIES,
1109 ),
1110 config: StateTableConfig::default(),
1111 entity_name: String::new(),
1112 recent_tx_instructions: std::sync::Mutex::new(LruCache::new(
1113 NonZeroUsize::new(1000).unwrap(),
1114 )),
1115 deferred_when_ops: DashMap::new(),
1116 },
1117 );
1118
1119 vm
1120 }
1121
1122 pub fn set_debugger(&mut self, debugger: Arc<dyn VmDebugger>) {
1123 self.debugger = Some(debugger);
1124 }
1125
1126 pub fn clear_debugger(&mut self) {
1127 self.debugger = None;
1128 }
1129
1130 fn emit_debug<F>(&self, builder: F)
1131 where
1132 F: FnOnce() -> VmDebugEvent,
1133 {
1134 if let Some(debugger) = &self.debugger {
1135 debugger.record(builder());
1136 }
1137 }
1138
1139 fn is_debug_enabled(&self) -> bool {
1140 self.debugger.is_some()
1141 }
1142
1143 fn take_pending_pda_reprocess_updates(&mut self) -> Vec<PendingAccountUpdate> {
1144 std::mem::take(&mut self.pending_pda_reprocess_updates)
1145 }
1146
1147 pub fn new_multi_entity() -> Self {
1149 VmContext {
1150 registers: vec![Value::Null; 256],
1151 states: HashMap::new(),
1152 debugger: None,
1153 instructions_executed: 0,
1154 cache_hits: 0,
1155 path_cache: HashMap::new(),
1156 pda_cache_hits: 0,
1157 pda_cache_misses: 0,
1158 pending_queue_size: 0,
1159 resolver_requests: VecDeque::new(),
1160 resolver_pending: HashMap::new(),
1161 resolver_cache: LruCache::new(resolver_cache_capacity()),
1162 resolver_cache_hits: 0,
1163 resolver_cache_misses: 0,
1164 current_context: None,
1165 warnings: Vec::new(),
1166 last_pda_lookup_miss: None,
1167 last_lookup_index_miss: None,
1168 last_pda_registered: None,
1169 last_lookup_index_keys: Vec::new(),
1170 pending_pda_reprocess_updates: Vec::new(),
1171 scheduled_callbacks: Vec::new(),
1172 }
1173 }
1174
1175 pub fn new_with_config(state_config: StateTableConfig) -> Self {
1176 let mut vm = VmContext {
1177 registers: vec![Value::Null; 256],
1178 states: HashMap::new(),
1179 debugger: None,
1180 instructions_executed: 0,
1181 cache_hits: 0,
1182 path_cache: HashMap::new(),
1183 pda_cache_hits: 0,
1184 pda_cache_misses: 0,
1185 pending_queue_size: 0,
1186 resolver_requests: VecDeque::new(),
1187 resolver_pending: HashMap::new(),
1188 resolver_cache: LruCache::new(resolver_cache_capacity()),
1189 resolver_cache_hits: 0,
1190 resolver_cache_misses: 0,
1191 current_context: None,
1192 warnings: Vec::new(),
1193 last_pda_lookup_miss: None,
1194 last_lookup_index_miss: None,
1195 last_pda_registered: None,
1196 last_lookup_index_keys: Vec::new(),
1197 pending_pda_reprocess_updates: Vec::new(),
1198 scheduled_callbacks: Vec::new(),
1199 };
1200 vm.states.insert(
1201 0,
1202 StateTable {
1203 data: DashMap::new(),
1204 access_times: DashMap::new(),
1205 lookup_indexes: HashMap::new(),
1206 temporal_indexes: HashMap::new(),
1207 pda_reverse_lookups: HashMap::new(),
1208 pending_updates: DashMap::new(),
1209 pending_instruction_events: DashMap::new(),
1210 last_account_data: DashMap::new(),
1211 version_tracker: VersionTracker::new(),
1212 instruction_dedup_cache: VersionTracker::with_capacity(
1213 DEFAULT_MAX_INSTRUCTION_DEDUP_ENTRIES,
1214 ),
1215 config: state_config,
1216 entity_name: "default".to_string(),
1217 recent_tx_instructions: std::sync::Mutex::new(LruCache::new(
1218 NonZeroUsize::new(1000).unwrap(),
1219 )),
1220 deferred_when_ops: DashMap::new(),
1221 },
1222 );
1223 vm
1224 }
1225
1226 pub fn take_resolver_requests(&mut self) -> Vec<ResolverRequest> {
1227 let requests: Vec<ResolverRequest> = self.resolver_requests.drain(..).collect();
1228 for request in &requests {
1229 if let Some(entry) = self.resolver_pending.get_mut(&request.cache_key) {
1230 entry.queued = false;
1231 entry.in_flight = true;
1232 }
1233 }
1234 requests
1235 }
1236
1237 pub fn take_scheduled_callbacks(&mut self) -> Vec<(u64, ScheduledCallback)> {
1238 std::mem::take(&mut self.scheduled_callbacks)
1239 }
1240
1241 pub fn get_entity_state(&self, state_id: u32, key: &Value) -> Option<Value> {
1242 self.states.get(&state_id)?.get_and_touch(key)
1243 }
1244
1245 pub fn snapshot_state_table(&self, state_id: u32) -> Vec<(Value, Value)> {
1246 self.states
1247 .get(&state_id)
1248 .map(|state| {
1249 state
1250 .data
1251 .iter()
1252 .map(|entry| (entry.key().clone(), entry.value().clone()))
1253 .collect()
1254 })
1255 .unwrap_or_default()
1256 }
1257
1258 pub fn restore_resolver_requests(&mut self, requests: Vec<ResolverRequest>) {
1259 if requests.is_empty() {
1260 return;
1261 }
1262
1263 for request in &requests {
1264 if let Some(entry) = self.resolver_pending.get_mut(&request.cache_key) {
1265 entry.in_flight = false;
1266 entry.queued = true;
1267 entry.next_retry_at = Instant::now();
1268 }
1269 }
1270
1271 self.resolver_requests.extend(requests);
1272 }
1273
1274 pub(crate) fn get_cached_resolver_value(
1275 &mut self,
1276 cache_key: &str,
1277 ) -> Option<CachedResolverValue> {
1278 let cached = self.resolver_cache.get(cache_key).cloned();
1279
1280 match cached {
1281 Some(entry) if entry.cached_at.elapsed() <= entry.ttl => {
1282 self.resolver_cache_hits += 1;
1283 Some(match entry.value {
1284 ResolverCacheValue::Resolved(value) => CachedResolverValue::Resolved(value),
1285 ResolverCacheValue::Negative => CachedResolverValue::Negative,
1286 })
1287 }
1288 Some(_) => {
1289 self.resolver_cache.pop(cache_key);
1290 self.resolver_cache_misses += 1;
1291 None
1292 }
1293 None => {
1294 self.resolver_cache_misses += 1;
1295 None
1296 }
1297 }
1298 }
1299
1300 fn cache_resolver_value(
1301 &mut self,
1302 resolver: &ResolverType,
1303 input: &Value,
1304 resolved_value: &Value,
1305 ) {
1306 self.resolver_cache.put(
1307 resolver_cache_key(resolver, input),
1308 ResolverCacheEntry {
1309 value: ResolverCacheValue::Resolved(resolved_value.clone()),
1310 cached_at: Instant::now(),
1311 ttl: resolver_cache_ttl(),
1312 },
1313 );
1314 }
1315
1316 pub(crate) fn cache_negative_resolver_value(&mut self, resolver: &ResolverType, input: &Value) {
1317 self.resolver_cache.put(
1318 resolver_cache_key(resolver, input),
1319 ResolverCacheEntry {
1320 value: ResolverCacheValue::Negative,
1321 cached_at: Instant::now(),
1322 ttl: negative_resolver_cache_ttl(),
1323 },
1324 );
1325 }
1326
1327 pub(crate) fn schedule_resolver_retry(&mut self, cache_key: &str) -> bool {
1328 if let Some(entry) = self.resolver_pending.get_mut(cache_key) {
1329 entry.retry_count = entry.retry_count.saturating_add(1);
1330 entry.next_retry_at = Instant::now() + resolver_retry_backoff();
1331 entry.in_flight = false;
1332 entry.queued = false;
1333 return true;
1334 }
1335 false
1336 }
1337
1338 pub(crate) fn handle_token_resolver_miss(
1339 &mut self,
1340 cache_key: &str,
1341 ) -> TokenResolverMissAction {
1342 let should_retry = self
1343 .resolver_pending
1344 .get(cache_key)
1345 .map(|entry| entry.retry_count < resolver_max_retries())
1346 .unwrap_or(false);
1347
1348 if should_retry && self.schedule_resolver_retry(cache_key) {
1349 return TokenResolverMissAction::RetryScheduled;
1350 }
1351
1352 if let Some(entry) = self.drop_pending_resolver_entry(cache_key) {
1353 self.cache_negative_resolver_value(&entry.resolver, &entry.input);
1354 return TokenResolverMissAction::NegativeCached;
1355 }
1356
1357 TokenResolverMissAction::Dropped
1358 }
1359
1360 pub(crate) fn drop_pending_resolver_entry(
1361 &mut self,
1362 cache_key: &str,
1363 ) -> Option<PendingResolverEntry> {
1364 self.resolver_pending.remove(cache_key)
1365 }
1366
1367 pub fn apply_resolver_result(
1368 &mut self,
1369 bytecode: &MultiEntityBytecode,
1370 cache_key: &str,
1371 resolved_value: Value,
1372 ) -> Result<Vec<Mutation>> {
1373 let entry = match self.resolver_pending.remove(cache_key) {
1374 Some(entry) => entry,
1375 None => return Ok(Vec::new()),
1376 };
1377
1378 self.cache_resolver_value(&entry.resolver, &entry.input, &resolved_value);
1379
1380 let mut mutations = Vec::new();
1381
1382 for target in entry.targets {
1383 if target.primary_key.is_null() {
1384 continue;
1385 }
1386
1387 let entity_bytecode = match bytecode.entities.get(&target.entity_name) {
1388 Some(bc) => bc,
1389 None => continue,
1390 };
1391
1392 let state = match self.states.get(&target.state_id) {
1393 Some(s) => s,
1394 None => continue,
1395 };
1396
1397 let mut entity_state = state
1398 .get_and_touch(&target.primary_key)
1399 .unwrap_or_else(|| json!({}));
1400
1401 let mut dirty_tracker = DirtyTracker::new();
1402 let should_emit = |path: &str| !entity_bytecode.non_emitted_fields.contains(path);
1403
1404 Self::apply_resolver_extractions_to_value(
1405 &mut entity_state,
1406 &resolved_value,
1407 &target.extracts,
1408 &mut dirty_tracker,
1409 &should_emit,
1410 )?;
1411
1412 if let Some(evaluator) = entity_bytecode.computed_fields_evaluator.as_ref() {
1413 let old_values: Vec<_> = entity_bytecode
1414 .computed_paths
1415 .iter()
1416 .map(|path| Self::get_value_at_path(&entity_state, path))
1417 .collect();
1418
1419 let context_slot = self.current_context.as_ref().and_then(|c| c.slot);
1420 let context_timestamp = self
1421 .current_context
1422 .as_ref()
1423 .map(|c| c.timestamp())
1424 .unwrap_or_else(|| {
1425 std::time::SystemTime::now()
1426 .duration_since(std::time::UNIX_EPOCH)
1427 .unwrap()
1428 .as_secs() as i64
1429 });
1430 let eval_result = evaluator(&mut entity_state, context_slot, context_timestamp);
1431
1432 if eval_result.is_ok() {
1433 let mut changed_fields = Vec::new();
1434 for (path, old_value) in
1435 entity_bytecode.computed_paths.iter().zip(old_values.iter())
1436 {
1437 let new_value = Self::get_value_at_path(&entity_state, path);
1438 let changed = new_value != *old_value;
1439 let will_emit = should_emit(path);
1440
1441 if changed && will_emit {
1442 dirty_tracker.mark_replaced(path);
1443 changed_fields.push(path.clone());
1444 }
1445 }
1446 }
1447 }
1448
1449 state.insert_with_eviction(target.primary_key.clone(), entity_state.clone());
1450
1451 if dirty_tracker.is_empty() {
1452 continue;
1453 }
1454
1455 let patch = Self::build_partial_state_from_value(&entity_state, &dirty_tracker)?;
1456
1457 mutations.push(Mutation {
1458 export: target.entity_name.clone(),
1459 key: target.primary_key.clone(),
1460 patch,
1461 append: vec![],
1462 });
1463 }
1464
1465 Ok(mutations)
1466 }
1467
1468 pub fn enqueue_resolver_request(
1469 &mut self,
1470 _cache_key: String,
1471 resolver: ResolverType,
1472 input: Value,
1473 target: ResolverTarget,
1474 ) {
1475 let cache_key = resolver_cache_key(&resolver, &input);
1476
1477 if let Some(entry) = self.resolver_pending.get_mut(&cache_key) {
1478 entry.add_target(target);
1479 if entry.ready_to_queue() {
1480 entry.queued = true;
1481 self.resolver_requests.push_back(ResolverRequest {
1482 cache_key,
1483 resolver,
1484 input,
1485 });
1486 }
1487 return;
1488 }
1489
1490 let queued_at = std::time::SystemTime::now()
1491 .duration_since(std::time::UNIX_EPOCH)
1492 .unwrap()
1493 .as_secs() as i64;
1494
1495 self.resolver_pending.insert(
1496 cache_key.clone(),
1497 PendingResolverEntry {
1498 resolver: resolver.clone(),
1499 input: input.clone(),
1500 targets: vec![target],
1501 queued_at,
1502 next_retry_at: Instant::now(),
1503 retry_count: 0,
1504 queued: true,
1505 in_flight: false,
1506 },
1507 );
1508
1509 self.resolver_requests.push_back(ResolverRequest {
1510 cache_key,
1511 resolver,
1512 input,
1513 });
1514 }
1515
1516 fn apply_resolver_extractions_to_value<F>(
1517 state: &mut Value,
1518 resolved_value: &Value,
1519 extracts: &[ResolverExtractSpec],
1520 dirty_tracker: &mut DirtyTracker,
1521 should_emit: &F,
1522 ) -> Result<()>
1523 where
1524 F: Fn(&str) -> bool,
1525 {
1526 for extract in extracts {
1527 let resolved = match extract.source_path.as_deref() {
1528 Some(path) => Self::get_value_at_path(resolved_value, path),
1529 None => Some(resolved_value.clone()),
1530 };
1531
1532 let Some(value) = resolved else {
1533 continue;
1534 };
1535
1536 let value = if let Some(transform) = &extract.transform {
1537 Self::apply_transformation(&value, transform)?
1538 } else {
1539 value
1540 };
1541
1542 Self::set_nested_field_value(state, &extract.target_path, value)?;
1543 if should_emit(&extract.target_path) {
1544 dirty_tracker.mark_replaced(&extract.target_path);
1545 }
1546 }
1547
1548 Ok(())
1549 }
1550
1551 fn build_partial_state_from_value(state: &Value, tracker: &DirtyTracker) -> Result<Value> {
1552 if tracker.is_empty() {
1553 return Ok(json!({}));
1554 }
1555
1556 let mut partial = serde_json::Map::new();
1557
1558 for (path, change) in tracker.iter() {
1559 let segments: Vec<&str> = path.split('.').collect();
1560
1561 let value_to_insert = match change {
1562 FieldChange::Replaced => {
1563 let mut current = state;
1564 let mut found = true;
1565
1566 for segment in &segments {
1567 match current.get(*segment) {
1568 Some(v) => current = v,
1569 None => {
1570 found = false;
1571 break;
1572 }
1573 }
1574 }
1575
1576 if !found {
1577 continue;
1578 }
1579 current.clone()
1580 }
1581 FieldChange::Appended(values) => Value::Array(values.clone()),
1582 };
1583
1584 let mut target = &mut partial;
1585 for (i, segment) in segments.iter().enumerate() {
1586 if i == segments.len() - 1 {
1587 target.insert(segment.to_string(), value_to_insert.clone());
1588 } else {
1589 target
1590 .entry(segment.to_string())
1591 .or_insert_with(|| json!({}));
1592 target = target
1593 .get_mut(*segment)
1594 .and_then(|v| v.as_object_mut())
1595 .ok_or("Failed to build nested structure")?;
1596 }
1597 }
1598 }
1599
1600 Ok(Value::Object(partial))
1601 }
1602
1603 pub fn get_state_table_mut(&mut self, state_id: u32) -> Option<&mut StateTable> {
1606 self.states.get_mut(&state_id)
1607 }
1608
1609 pub fn registers_mut(&mut self) -> &mut Vec<RegisterValue> {
1611 &mut self.registers
1612 }
1613
1614 pub fn path_cache(&self) -> &HashMap<String, CompiledPath> {
1616 &self.path_cache
1617 }
1618
1619 pub fn current_context(&self) -> Option<&UpdateContext> {
1621 self.current_context.as_ref()
1622 }
1623
1624 pub fn set_current_context(&mut self, context: Option<UpdateContext>) {
1626 self.current_context = context;
1627 }
1628
1629 fn add_warning(&mut self, msg: String) {
1630 self.warnings.push(msg);
1631 }
1632
1633 pub fn take_warnings(&mut self) -> Vec<String> {
1634 std::mem::take(&mut self.warnings)
1635 }
1636
1637 pub fn has_warnings(&self) -> bool {
1638 !self.warnings.is_empty()
1639 }
1640
1641 pub fn update_state_from_register(
1642 &mut self,
1643 state_id: u32,
1644 key: Value,
1645 register: Register,
1646 ) -> Result<()> {
1647 let state = self.states.get(&state_id).ok_or("State table not found")?;
1648 let value = self.registers[register].clone();
1649 state.insert_with_eviction(key, value);
1650 Ok(())
1651 }
1652
1653 fn reset_registers(&mut self) {
1654 for reg in &mut self.registers {
1655 *reg = Value::Null;
1656 }
1657 }
1658
1659 pub fn extract_partial_state(
1661 &self,
1662 state_reg: Register,
1663 dirty_fields: &HashSet<String>,
1664 ) -> Result<Value> {
1665 let full_state = &self.registers[state_reg];
1666
1667 if dirty_fields.is_empty() {
1668 return Ok(json!({}));
1669 }
1670
1671 let mut partial = serde_json::Map::new();
1672
1673 for path in dirty_fields {
1674 let segments: Vec<&str> = path.split('.').collect();
1675
1676 let mut current = full_state;
1677 let mut found = true;
1678
1679 for segment in &segments {
1680 match current.get(segment) {
1681 Some(v) => current = v,
1682 None => {
1683 found = false;
1684 break;
1685 }
1686 }
1687 }
1688
1689 if !found {
1690 continue;
1691 }
1692
1693 let mut target = &mut partial;
1694 for (i, segment) in segments.iter().enumerate() {
1695 if i == segments.len() - 1 {
1696 target.insert(segment.to_string(), current.clone());
1697 } else {
1698 target
1699 .entry(segment.to_string())
1700 .or_insert_with(|| json!({}));
1701 target = target
1702 .get_mut(*segment)
1703 .and_then(|v| v.as_object_mut())
1704 .ok_or("Failed to build nested structure")?;
1705 }
1706 }
1707 }
1708
1709 Ok(Value::Object(partial))
1710 }
1711
1712 pub fn extract_partial_state_with_tracker(
1716 &self,
1717 state_reg: Register,
1718 tracker: &DirtyTracker,
1719 ) -> Result<Value> {
1720 let full_state = &self.registers[state_reg];
1721
1722 if tracker.is_empty() {
1723 return Ok(json!({}));
1724 }
1725
1726 let mut partial = serde_json::Map::new();
1727
1728 for (path, change) in tracker.iter() {
1729 let segments: Vec<&str> = path.split('.').collect();
1730
1731 let value_to_insert = match change {
1732 FieldChange::Replaced => {
1733 let mut current = full_state;
1734 let mut found = true;
1735
1736 for segment in &segments {
1737 match current.get(*segment) {
1738 Some(v) => current = v,
1739 None => {
1740 found = false;
1741 break;
1742 }
1743 }
1744 }
1745
1746 if !found {
1747 continue;
1748 }
1749 current.clone()
1750 }
1751 FieldChange::Appended(values) => Value::Array(values.clone()),
1752 };
1753
1754 let mut target = &mut partial;
1755 for (i, segment) in segments.iter().enumerate() {
1756 if i == segments.len() - 1 {
1757 target.insert(segment.to_string(), value_to_insert.clone());
1758 } else {
1759 target
1760 .entry(segment.to_string())
1761 .or_insert_with(|| json!({}));
1762 target = target
1763 .get_mut(*segment)
1764 .and_then(|v| v.as_object_mut())
1765 .ok_or("Failed to build nested structure")?;
1766 }
1767 }
1768 }
1769
1770 Ok(Value::Object(partial))
1771 }
1772
1773 fn get_compiled_path(&mut self, path: &str) -> CompiledPath {
1774 if let Some(compiled) = self.path_cache.get(path) {
1775 self.cache_hits += 1;
1776 #[cfg(feature = "otel")]
1777 crate::vm_metrics::record_path_cache_hit();
1778 return compiled.clone();
1779 }
1780 #[cfg(feature = "otel")]
1781 crate::vm_metrics::record_path_cache_miss();
1782 let compiled = CompiledPath::new(path);
1783 self.path_cache.insert(path.to_string(), compiled.clone());
1784 compiled
1785 }
1786
1787 #[cfg_attr(feature = "otel", instrument(
1789 name = "vm.process_event",
1790 skip(self, bytecode, event_value, log),
1791 level = "info",
1792 fields(
1793 event_type = %event_type,
1794 slot = context.as_ref().and_then(|c| c.slot),
1795 )
1796 ))]
1797 pub fn process_event(
1798 &mut self,
1799 bytecode: &MultiEntityBytecode,
1800 event_value: Value,
1801 event_type: &str,
1802 context: Option<&UpdateContext>,
1803 mut log: Option<&mut crate::canonical_log::CanonicalLog>,
1804 ) -> Result<Vec<Mutation>> {
1805 self.current_context = context.cloned();
1806 self.emit_debug(|| VmDebugEvent::ProcessEventStart {
1807 event_type: event_type.to_string(),
1808 context: context.map(|ctx| ctx.to_value()),
1809 });
1810
1811 let mut event_value = event_value;
1812 if let Some(ctx) = context {
1813 if let Some(obj) = event_value.as_object_mut() {
1814 obj.insert("__update_context".to_string(), ctx.to_value());
1815 }
1816 }
1817
1818 let mut all_mutations = Vec::new();
1819
1820 if event_type.ends_with("IxState") && bytecode.when_events.contains(event_type) {
1821 if let Some(ctx) = context {
1822 if let Some(signature) = ctx.signature.clone() {
1823 let state_ids: Vec<u32> = self.states.keys().cloned().collect();
1824 for state_id in state_ids {
1825 if let Some(state) = self.states.get(&state_id) {
1826 {
1827 let mut cache = state.recent_tx_instructions.lock().unwrap();
1828 let entry =
1829 cache.get_or_insert_mut(signature.clone(), HashSet::new);
1830 entry.insert(event_type.to_string());
1831 }
1832
1833 let key = (signature.clone(), event_type.to_string());
1834 if let Some((_, deferred_ops)) = state.deferred_when_ops.remove(&key) {
1835 tracing::debug!(
1836 event_type = %event_type,
1837 signature = %signature,
1838 deferred_count = deferred_ops.len(),
1839 "flushing deferred when-ops"
1840 );
1841 for op in deferred_ops {
1842 let (evaluator, computed_paths) = bytecode
1844 .entities
1845 .get(&op.entity_name)
1846 .map(|eb| {
1847 (
1848 eb.computed_fields_evaluator.as_ref(),
1849 eb.computed_paths.as_slice(),
1850 )
1851 })
1852 .unwrap_or((None, &[]));
1853
1854 match self.apply_deferred_when_op(
1855 state_id,
1856 &op,
1857 evaluator,
1858 Some(computed_paths),
1859 ) {
1860 Ok(mutations) => all_mutations.extend(mutations),
1861 Err(e) => tracing::warn!(
1862 "Failed to apply deferred when-op: {}",
1863 e
1864 ),
1865 }
1866 }
1867 }
1868 }
1869 }
1870 }
1871 }
1872 }
1873
1874 if let Some(entity_names) = bytecode.event_routing.get(event_type) {
1875 for entity_name in entity_names {
1876 if let Some(entity_bytecode) = bytecode.entities.get(entity_name) {
1877 if let Some(handler) = entity_bytecode.handlers.get(event_type) {
1878 self.emit_debug(|| VmDebugEvent::HandlerStart {
1879 entity_name: entity_name.clone(),
1880 event_type: event_type.to_string(),
1881 state_id: entity_bytecode.state_id,
1882 });
1883
1884 if let Some(ref mut log) = log {
1885 log.set("entity", entity_name.clone());
1886 log.inc("handlers", 1);
1887 }
1888
1889 let opcodes_before = self.instructions_executed;
1890 let cache_before = self.cache_hits;
1891 let pda_hits_before = self.pda_cache_hits;
1892 let pda_misses_before = self.pda_cache_misses;
1893
1894 let mutations = self.execute_handler(
1895 handler,
1896 &event_value,
1897 event_type,
1898 entity_bytecode.state_id,
1899 entity_name,
1900 entity_bytecode.computed_fields_evaluator.as_ref(),
1901 Some(&entity_bytecode.non_emitted_fields),
1902 )?;
1903 let direct_mutation_count = mutations.len();
1904
1905 if let Some(ref mut log) = log {
1906 log.inc(
1907 "opcodes",
1908 (self.instructions_executed - opcodes_before) as i64,
1909 );
1910 log.inc("cache_hits", (self.cache_hits - cache_before) as i64);
1911 log.inc("pda_hits", (self.pda_cache_hits - pda_hits_before) as i64);
1912 log.inc(
1913 "pda_misses",
1914 (self.pda_cache_misses - pda_misses_before) as i64,
1915 );
1916 }
1917
1918 if mutations.is_empty() {
1919 let is_tx_event =
1922 event_type.ends_with("IxState") || event_type.ends_with("CpiEvent");
1923 if let Some(missed_pda) = self.take_last_pda_lookup_miss() {
1924 if is_tx_event {
1925 let slot = context.and_then(|c| c.slot).unwrap_or(0);
1926 let signature = context
1927 .and_then(|c| c.signature.clone())
1928 .unwrap_or_default();
1929 let _ = self.queue_instruction_event(
1930 entity_bytecode.state_id,
1931 QueuedInstructionEvent {
1932 pda_address: missed_pda.clone(),
1933 event_type: event_type.to_string(),
1934 event_data: event_value.clone(),
1935 slot,
1936 signature,
1937 },
1938 );
1939 self.emit_debug(|| VmDebugEvent::QueueAction {
1940 entity_name: entity_name.clone(),
1941 event_type: event_type.to_string(),
1942 queue_kind: "instruction_pda_lookup_miss".to_string(),
1943 lookup_value: missed_pda,
1944 });
1945 } else {
1946 let slot = context.and_then(|c| c.slot).unwrap_or(0);
1951 let signature = context
1952 .and_then(|c| c.signature.clone())
1953 .unwrap_or_default();
1954 if let Some(write_version) =
1955 context.and_then(|c| c.write_version)
1956 {
1957 let _ = self.queue_account_update(
1958 entity_bytecode.state_id,
1959 QueuedAccountUpdate {
1960 pda_address: missed_pda.clone(),
1961 account_type: event_type.to_string(),
1962 account_data: event_value.clone(),
1963 slot,
1964 write_version,
1965 signature,
1966 },
1967 );
1968 self.emit_debug(|| VmDebugEvent::QueueAction {
1969 entity_name: entity_name.clone(),
1970 event_type: event_type.to_string(),
1971 queue_kind: "account_pda_lookup_miss".to_string(),
1972 lookup_value: missed_pda,
1973 });
1974 } else {
1975 tracing::warn!(
1976 event_type = %event_type,
1977 "Dropping queued account update: write_version missing from context"
1978 );
1979 }
1980 }
1981 }
1982 if let Some(missed_lookup) = self.take_last_lookup_index_miss() {
1983 if !is_tx_event {
1984 let slot = context.and_then(|c| c.slot).unwrap_or(0);
1985 let signature = context
1986 .and_then(|c| c.signature.clone())
1987 .unwrap_or_default();
1988 if let Some(write_version) =
1989 context.and_then(|c| c.write_version)
1990 {
1991 let _ = self.queue_account_update(
1992 entity_bytecode.state_id,
1993 QueuedAccountUpdate {
1994 pda_address: missed_lookup.clone(),
1995 account_type: event_type.to_string(),
1996 account_data: event_value.clone(),
1997 slot,
1998 write_version,
1999 signature,
2000 },
2001 );
2002 self.emit_debug(|| VmDebugEvent::QueueAction {
2003 entity_name: entity_name.clone(),
2004 event_type: event_type.to_string(),
2005 queue_kind: "account_lookup_index_miss".to_string(),
2006 lookup_value: missed_lookup,
2007 });
2008 } else {
2009 tracing::trace!(
2010 event_type = %event_type,
2011 "Discarding lookup_index_miss for tx-scoped event (IxState/CpiEvent do not use lookup-index queuing)"
2012 );
2013 }
2014 }
2015 }
2016 }
2017
2018 all_mutations.extend(mutations);
2019
2020 if event_type.ends_with("IxState") || event_type.ends_with("CpiEvent") {
2021 if let Some(ctx) = context {
2022 if let Some(ref signature) = ctx.signature {
2023 if let Some(state) = self.states.get(&entity_bytecode.state_id)
2024 {
2025 {
2026 let mut cache =
2027 state.recent_tx_instructions.lock().unwrap();
2028 let entry = cache
2029 .get_or_insert_mut(signature.clone(), HashSet::new);
2030 entry.insert(event_type.to_string());
2031 }
2032
2033 let key = (signature.clone(), event_type.to_string());
2034 if let Some((_, deferred_ops)) =
2035 state.deferred_when_ops.remove(&key)
2036 {
2037 tracing::debug!(
2038 event_type = %event_type,
2039 signature = %signature,
2040 deferred_count = deferred_ops.len(),
2041 "flushing deferred when-ops"
2042 );
2043 for op in deferred_ops {
2044 match self.apply_deferred_when_op(
2045 entity_bytecode.state_id,
2046 &op,
2047 entity_bytecode
2048 .computed_fields_evaluator
2049 .as_ref(),
2050 Some(&entity_bytecode.computed_paths),
2051 ) {
2052 Ok(mutations) => {
2053 all_mutations.extend(mutations)
2054 }
2055 Err(e) => {
2056 tracing::warn!(
2057 "Failed to apply deferred when-op: {}",
2058 e
2059 );
2060 }
2061 }
2062 }
2063 }
2064 }
2065 }
2066 }
2067 }
2068
2069 if let Some(registered_pda) = self.take_last_pda_registered() {
2070 let pending_events = self.flush_pending_instruction_events(
2071 entity_bytecode.state_id,
2072 ®istered_pda,
2073 );
2074 let pending_count = pending_events.len();
2075 if pending_count > 0 {
2076 self.emit_debug(|| VmDebugEvent::FlushAction {
2077 entity_name: entity_name.clone(),
2078 event_type: event_type.to_string(),
2079 flush_kind: "pending_instruction_events".to_string(),
2080 trigger: registered_pda.clone(),
2081 count: pending_count,
2082 });
2083 }
2084 for pending in pending_events {
2085 if let Some(pending_handler) =
2086 entity_bytecode.handlers.get(&pending.event_type)
2087 {
2088 if let Ok(reprocessed_mutations) = self.execute_handler(
2089 pending_handler,
2090 &pending.event_data,
2091 &pending.event_type,
2092 entity_bytecode.state_id,
2093 entity_name,
2094 entity_bytecode.computed_fields_evaluator.as_ref(),
2095 Some(&entity_bytecode.non_emitted_fields),
2096 ) {
2097 all_mutations.extend(reprocessed_mutations);
2098 }
2099 }
2100 }
2101 }
2102
2103 for pending in self.take_pending_pda_reprocess_updates() {
2104 if let Some(pending_handler) =
2105 entity_bytecode.handlers.get(&pending.account_type)
2106 {
2107 self.current_context = Some(if pending.is_stale_reprocess {
2108 UpdateContext::new_reprocessed(
2109 pending.slot,
2110 pending.write_version,
2111 )
2112 } else {
2113 UpdateContext::new_account(
2114 pending.slot,
2115 pending.signature.clone(),
2116 pending.write_version,
2117 )
2118 });
2119 match self.execute_handler(
2120 pending_handler,
2121 &pending.account_data,
2122 &pending.account_type,
2123 entity_bytecode.state_id,
2124 entity_name,
2125 entity_bytecode.computed_fields_evaluator.as_ref(),
2126 Some(&entity_bytecode.non_emitted_fields),
2127 ) {
2128 Ok(reprocessed) => {
2129 all_mutations.extend(reprocessed);
2130 }
2131 Err(e) => {
2132 tracing::warn!(
2133 error = %e,
2134 account_type = %pending.account_type,
2135 "PDA remap reprocessing failed"
2136 );
2137 }
2138 }
2139 }
2140 }
2141
2142 let lookup_keys = self.take_last_lookup_index_keys();
2143 if !lookup_keys.is_empty() {
2144 tracing::info!(
2145 keys = ?lookup_keys,
2146 entity = %entity_name,
2147 "vm.process_event: flushing pending updates for lookup_keys"
2148 );
2149 }
2150 for lookup_key in lookup_keys {
2151 if let Ok(pending_updates) =
2152 self.flush_pending_updates(entity_bytecode.state_id, &lookup_key)
2153 {
2154 let pending_count = pending_updates.len();
2155 if pending_count > 0 {
2156 self.emit_debug(|| VmDebugEvent::FlushAction {
2157 entity_name: entity_name.clone(),
2158 event_type: event_type.to_string(),
2159 flush_kind: "pending_account_updates".to_string(),
2160 trigger: lookup_key.clone(),
2161 count: pending_count,
2162 });
2163 }
2164 for pending in pending_updates {
2165 if let Some(pending_handler) =
2166 entity_bytecode.handlers.get(&pending.account_type)
2167 {
2168 self.current_context = Some(UpdateContext::new_account(
2169 pending.slot,
2170 pending.signature.clone(),
2171 pending.write_version,
2172 ));
2173 match self.execute_handler(
2174 pending_handler,
2175 &pending.account_data,
2176 &pending.account_type,
2177 entity_bytecode.state_id,
2178 entity_name,
2179 entity_bytecode.computed_fields_evaluator.as_ref(),
2180 Some(&entity_bytecode.non_emitted_fields),
2181 ) {
2182 Ok(reprocessed) => {
2183 all_mutations.extend(reprocessed);
2184 }
2185 Err(e) => {
2186 tracing::warn!(
2187 error = %e,
2188 account_type = %pending.account_type,
2189 "Flushed event reprocessing failed"
2190 );
2191 }
2192 }
2193 }
2194 }
2195 }
2196 }
2197
2198 self.emit_debug(|| VmDebugEvent::HandlerEnd {
2199 entity_name: entity_name.clone(),
2200 event_type: event_type.to_string(),
2201 mutations: direct_mutation_count,
2202 });
2203 } else if let Some(ref mut log) = log {
2204 log.set("skip_reason", "no_handler");
2205 }
2206 } else if let Some(ref mut log) = log {
2207 log.set("skip_reason", "entity_not_found");
2208 }
2209 }
2210 } else if let Some(ref mut log) = log {
2211 log.set("skip_reason", "no_event_routing");
2212 }
2213
2214 let debug_warnings = self.warnings.clone();
2215
2216 if let Some(log) = log {
2217 log.set("mutations", all_mutations.len() as i64);
2218 if let Some(first) = all_mutations.first() {
2219 if let Some(key_str) = first.key.as_str() {
2220 log.set("primary_key", key_str);
2221 } else if let Some(key_num) = first.key.as_u64() {
2222 log.set("primary_key", key_num as i64);
2223 }
2224 }
2225 if let Some(state) = self.states.get(&0) {
2226 log.set("state_table_size", state.data.len() as i64);
2227 }
2228
2229 let warnings = self.take_warnings();
2230 if !warnings.is_empty() {
2231 log.set("warnings", warnings.len() as i64);
2232 log.set(
2233 "warning_messages",
2234 Value::Array(warnings.into_iter().map(Value::String).collect()),
2235 );
2236 log.set_level(crate::canonical_log::LogLevel::Warn);
2237 }
2238 } else {
2239 self.warnings.clear();
2240 }
2241
2242 self.emit_debug(|| VmDebugEvent::ProcessEventEnd {
2243 event_type: event_type.to_string(),
2244 mutations: all_mutations.len(),
2245 warnings: debug_warnings,
2246 });
2247
2248 if self.instructions_executed.is_multiple_of(1000) {
2249 let state_ids: Vec<u32> = self.states.keys().cloned().collect();
2250 for state_id in state_ids {
2251 let expired = self.cleanup_expired_when_ops(state_id, 60);
2252 if expired > 0 {
2253 tracing::debug!(
2254 "Cleaned up {} expired deferred when-ops for state {}",
2255 expired,
2256 state_id
2257 );
2258 }
2259 }
2260 }
2261
2262 Ok(all_mutations)
2263 }
2264
2265 pub fn process_any(
2266 &mut self,
2267 bytecode: &MultiEntityBytecode,
2268 any: prost_types::Any,
2269 ) -> Result<Vec<Mutation>> {
2270 let (event_value, event_type) = bytecode.proto_router.decode(any)?;
2271 self.process_event(bytecode, event_value, &event_type, None, None)
2272 }
2273
2274 #[cfg_attr(feature = "otel", instrument(
2275 name = "vm.execute_handler",
2276 skip(self, handler, event_value, entity_evaluator),
2277 level = "debug",
2278 fields(
2279 event_type = %event_type,
2280 handler_opcodes = handler.len(),
2281 )
2282 ))]
2283 #[allow(clippy::type_complexity, clippy::too_many_arguments)]
2284 fn execute_handler(
2285 &mut self,
2286 handler: &[OpCode],
2287 event_value: &Value,
2288 event_type: &str,
2289 override_state_id: u32,
2290 entity_name: &str,
2291 entity_evaluator: Option<
2292 &Box<dyn Fn(&mut Value, Option<u64>, i64) -> ComputedEvaluatorResult + Send + Sync>,
2293 >,
2294 non_emitted_fields: Option<&HashSet<String>>,
2295 ) -> Result<Vec<Mutation>> {
2296 self.reset_registers();
2297 self.last_pda_lookup_miss = None;
2298
2299 let mut pc: usize = 0;
2300 let mut output = Vec::new();
2301 let mut dirty_tracker = DirtyTracker::new();
2302 let should_emit = |path: &str| {
2303 non_emitted_fields
2304 .map(|fields| !fields.contains(path))
2305 .unwrap_or(true)
2306 };
2307
2308 while pc < handler.len() {
2309 match &handler[pc] {
2310 OpCode::LoadEventField {
2311 path,
2312 dest,
2313 default,
2314 } => {
2315 let value = self.load_field(event_value, path, default.as_ref())?;
2316 if self.is_debug_enabled() {
2317 self.emit_debug(|| VmDebugEvent::LoadEventField {
2318 entity_name: entity_name.to_string(),
2319 event_type: event_type.to_string(),
2320 path: path.segments.clone(),
2321 value: value.clone(),
2322 });
2323 }
2324 self.registers[*dest] = value;
2325 pc += 1;
2326 }
2327 OpCode::LoadConstant { value, dest } => {
2328 self.registers[*dest] = value.clone();
2329 pc += 1;
2330 }
2331 OpCode::CopyRegister { source, dest } => {
2332 self.registers[*dest] = self.registers[*source].clone();
2333 pc += 1;
2334 }
2335 OpCode::CopyRegisterIfNull { source, dest } => {
2336 if self.registers[*dest].is_null() {
2337 self.registers[*dest] = self.registers[*source].clone();
2338 }
2339 pc += 1;
2340 }
2341 OpCode::GetEventType { dest } => {
2342 self.registers[*dest] = json!(event_type);
2343 pc += 1;
2344 }
2345 OpCode::CreateObject { dest } => {
2346 self.registers[*dest] = json!({});
2347 pc += 1;
2348 }
2349 OpCode::SetField {
2350 object,
2351 path,
2352 value,
2353 } => {
2354 let old_value = self
2355 .is_debug_enabled()
2356 .then(|| Self::get_value_at_path(&self.registers[*object], path))
2357 .flatten();
2358 self.set_field_auto_vivify(*object, path, *value)?;
2359 if should_emit(path) {
2360 dirty_tracker.mark_replaced(path);
2361 }
2362 if self.is_debug_enabled() {
2363 self.emit_debug(|| VmDebugEvent::FieldWrite {
2364 entity_name: entity_name.to_string(),
2365 event_type: event_type.to_string(),
2366 op: "set_field".to_string(),
2367 path: path.clone(),
2368 old_value,
2369 new_value: Self::get_value_at_path(&self.registers[*object], path),
2370 applied: true,
2371 reason: None,
2372 });
2373 }
2374 pc += 1;
2375 }
2376 OpCode::SetFields { object, fields } => {
2377 for (path, value_reg) in fields {
2378 let old_value = self
2379 .is_debug_enabled()
2380 .then(|| Self::get_value_at_path(&self.registers[*object], path))
2381 .flatten();
2382 self.set_field_auto_vivify(*object, path, *value_reg)?;
2383 if should_emit(path) {
2384 dirty_tracker.mark_replaced(path);
2385 }
2386 if self.is_debug_enabled() {
2387 self.emit_debug(|| VmDebugEvent::FieldWrite {
2388 entity_name: entity_name.to_string(),
2389 event_type: event_type.to_string(),
2390 op: "set_fields".to_string(),
2391 path: path.clone(),
2392 old_value,
2393 new_value: Self::get_value_at_path(&self.registers[*object], path),
2394 applied: true,
2395 reason: None,
2396 });
2397 }
2398 }
2399 pc += 1;
2400 }
2401 OpCode::GetField { object, path, dest } => {
2402 let value = self.get_field(*object, path)?;
2403 self.registers[*dest] = value;
2404 pc += 1;
2405 }
2406 OpCode::AbortIfNullKey {
2407 key,
2408 is_account_event,
2409 } => {
2410 let key_value = &self.registers[*key];
2411 if key_value.is_null() && *is_account_event {
2412 self.emit_debug(|| VmDebugEvent::ReadOrInitState {
2413 entity_name: entity_name.to_string(),
2414 event_type: event_type.to_string(),
2415 key: key_value.clone(),
2416 existing_state: None,
2417 loaded_state: Value::Null,
2418 skipped_reason: Some("null_primary_key".to_string()),
2419 });
2420 tracing::debug!(
2421 event_type = %event_type,
2422 "AbortIfNullKey: key is null for account state event, \
2423 returning empty mutations for queueing"
2424 );
2425 return Ok(Vec::new());
2426 }
2427
2428 pc += 1;
2429 }
2430 OpCode::ReadOrInitState {
2431 state_id: _,
2432 key,
2433 default,
2434 dest,
2435 } => {
2436 let actual_state_id = override_state_id;
2437 let entity_name_owned = entity_name.to_string();
2438 self.states
2439 .entry(actual_state_id)
2440 .or_insert_with(|| StateTable {
2441 data: DashMap::new(),
2442 access_times: DashMap::new(),
2443 lookup_indexes: HashMap::new(),
2444 temporal_indexes: HashMap::new(),
2445 pda_reverse_lookups: HashMap::new(),
2446 pending_updates: DashMap::new(),
2447 pending_instruction_events: DashMap::new(),
2448 last_account_data: DashMap::new(),
2449 version_tracker: VersionTracker::new(),
2450 instruction_dedup_cache: VersionTracker::with_capacity(
2451 DEFAULT_MAX_INSTRUCTION_DEDUP_ENTRIES,
2452 ),
2453 config: StateTableConfig::default(),
2454 entity_name: entity_name_owned,
2455 recent_tx_instructions: std::sync::Mutex::new(LruCache::new(
2456 NonZeroUsize::new(1000).unwrap(),
2457 )),
2458 deferred_when_ops: DashMap::new(),
2459 });
2460 let key_value = self.registers[*key].clone();
2461 let warn_null_key = key_value.is_null()
2463 && event_type.ends_with("State")
2464 && !event_type.ends_with("IxState")
2465 && !event_type.ends_with("CpiEvent");
2466
2467 if warn_null_key {
2468 self.add_warning(format!(
2469 "ReadOrInitState: key register {} is NULL for account state, event_type={}",
2470 key, event_type
2471 ));
2472 }
2473
2474 let state = self
2475 .states
2476 .get(&actual_state_id)
2477 .ok_or("State table not found")?;
2478
2479 if !key_value.is_null() {
2480 if let Some(ctx) = &self.current_context {
2481 if ctx.is_account_update() {
2486 if let (Some(slot), Some(write_version)) =
2487 (ctx.slot, ctx.write_version)
2488 {
2489 let account_address = event_value
2491 .get("__account_address")
2492 .cloned()
2493 .unwrap_or_else(|| key_value.clone());
2494
2495 if !state.is_fresh_update(
2496 &account_address,
2497 event_type,
2498 slot,
2499 write_version,
2500 ) {
2501 self.emit_debug(|| VmDebugEvent::ReadOrInitState {
2502 entity_name: entity_name.to_string(),
2503 event_type: event_type.to_string(),
2504 key: key_value.clone(),
2505 existing_state: None,
2506 loaded_state: Value::Null,
2507 skipped_reason: Some(
2508 "stale_account_update".to_string(),
2509 ),
2510 });
2511 self.add_warning(format!(
2512 "Stale account update skipped: slot={}, write_version={}, account={}",
2513 slot, write_version, account_address
2514 ));
2515 return Ok(Vec::new());
2516 }
2517 }
2518 }
2519 else if ctx.is_instruction_update() {
2521 if let (Some(slot), Some(txn_index)) = (ctx.slot, ctx.txn_index) {
2522 if state.is_duplicate_instruction(
2523 &key_value, event_type, slot, txn_index,
2524 ) {
2525 self.emit_debug(|| VmDebugEvent::ReadOrInitState {
2526 entity_name: entity_name.to_string(),
2527 event_type: event_type.to_string(),
2528 key: key_value.clone(),
2529 existing_state: None,
2530 loaded_state: Value::Null,
2531 skipped_reason: Some(
2532 "duplicate_instruction".to_string(),
2533 ),
2534 });
2535 self.add_warning(format!(
2536 "Duplicate instruction skipped: slot={}, txn_index={}",
2537 slot, txn_index
2538 ));
2539 return Ok(Vec::new());
2540 }
2541 }
2542 }
2543 }
2544 }
2545 let existing_state = state.get_and_touch(&key_value);
2546 let value = existing_state.clone().unwrap_or_else(|| default.clone());
2547
2548 self.emit_debug(|| VmDebugEvent::ReadOrInitState {
2549 entity_name: entity_name.to_string(),
2550 event_type: event_type.to_string(),
2551 key: key_value,
2552 existing_state,
2553 loaded_state: value.clone(),
2554 skipped_reason: None,
2555 });
2556
2557 self.registers[*dest] = value;
2558 pc += 1;
2559 }
2560 OpCode::UpdateState {
2561 state_id: _,
2562 key,
2563 value,
2564 } => {
2565 let actual_state_id = override_state_id;
2566 let state = self
2567 .states
2568 .get(&actual_state_id)
2569 .ok_or("State table not found")?;
2570 let key_value = self.registers[*key].clone();
2571 let value_data = self.registers[*value].clone();
2572
2573 state.insert_with_eviction(key_value, value_data);
2574 pc += 1;
2575 }
2576 OpCode::AppendToArray {
2577 object,
2578 path,
2579 value,
2580 } => {
2581 let appended_value = self.registers[*value].clone();
2582 let max_len = self
2583 .states
2584 .get(&override_state_id)
2585 .map(|s| s.max_array_length())
2586 .unwrap_or(DEFAULT_MAX_ARRAY_LENGTH);
2587 self.append_to_array(*object, path, *value, max_len)?;
2588 if should_emit(path) {
2589 dirty_tracker.mark_appended(path, appended_value);
2590 }
2591 pc += 1;
2592 }
2593 OpCode::GetCurrentTimestamp { dest } => {
2594 let timestamp = std::time::SystemTime::now()
2595 .duration_since(std::time::UNIX_EPOCH)
2596 .unwrap()
2597 .as_secs() as i64;
2598 self.registers[*dest] = json!(timestamp);
2599 pc += 1;
2600 }
2601 OpCode::CreateEvent { dest, event_value } => {
2602 let timestamp = std::time::SystemTime::now()
2603 .duration_since(std::time::UNIX_EPOCH)
2604 .unwrap()
2605 .as_secs() as i64;
2606
2607 let mut event_data = self.registers[*event_value].clone();
2609 if let Some(obj) = event_data.as_object_mut() {
2610 obj.remove("__update_context");
2611 }
2612
2613 let mut event = serde_json::Map::new();
2615 event.insert("timestamp".to_string(), json!(timestamp));
2616 event.insert("data".to_string(), event_data);
2617
2618 if let Some(ref ctx) = self.current_context {
2620 if let Some(slot) = ctx.slot {
2621 event.insert("slot".to_string(), json!(slot));
2622 }
2623 if let Some(ref signature) = ctx.signature {
2624 event.insert("signature".to_string(), json!(signature));
2625 }
2626 }
2627
2628 self.registers[*dest] = Value::Object(event);
2629 pc += 1;
2630 }
2631 OpCode::CreateCapture {
2632 dest,
2633 capture_value,
2634 } => {
2635 let timestamp = std::time::SystemTime::now()
2636 .duration_since(std::time::UNIX_EPOCH)
2637 .unwrap()
2638 .as_secs() as i64;
2639
2640 let capture_data = self.registers[*capture_value].clone();
2642
2643 let account_address = event_value
2645 .get("__account_address")
2646 .and_then(|v| v.as_str())
2647 .unwrap_or("")
2648 .to_string();
2649
2650 let mut capture = serde_json::Map::new();
2652 capture.insert("timestamp".to_string(), json!(timestamp));
2653 capture.insert("account_address".to_string(), json!(account_address));
2654 capture.insert("data".to_string(), capture_data);
2655
2656 if let Some(ref ctx) = self.current_context {
2658 if let Some(slot) = ctx.slot {
2659 capture.insert("slot".to_string(), json!(slot));
2660 }
2661 if let Some(ref signature) = ctx.signature {
2662 capture.insert("signature".to_string(), json!(signature));
2663 }
2664 }
2665
2666 self.registers[*dest] = Value::Object(capture);
2667 pc += 1;
2668 }
2669 OpCode::Transform {
2670 source,
2671 dest,
2672 transformation,
2673 } => {
2674 if source == dest {
2675 self.transform_in_place(*source, transformation)?;
2676 } else {
2677 let source_value = &self.registers[*source];
2678 let value = Self::apply_transformation(source_value, transformation)?;
2679 self.registers[*dest] = value;
2680 }
2681 pc += 1;
2682 }
2683 OpCode::EmitMutation {
2684 entity_name,
2685 key,
2686 state,
2687 } => {
2688 let primary_key = self.registers[*key].clone();
2689 let dirty_fields: Vec<String> =
2690 dirty_tracker.dirty_paths().into_iter().collect();
2691
2692 if primary_key.is_null() || dirty_tracker.is_empty() {
2693 let reason = if dirty_tracker.is_empty() {
2694 "no_fields_modified"
2695 } else {
2696 "null_primary_key"
2697 };
2698 self.emit_debug(|| VmDebugEvent::EmitMutation {
2699 entity_name: entity_name.clone(),
2700 event_type: event_type.to_string(),
2701 key: primary_key.clone(),
2702 emitted: false,
2703 reason: Some(reason.to_string()),
2704 patch: None,
2705 dirty_fields,
2706 });
2707 self.add_warning(format!(
2708 "Skipping mutation for entity '{}': {} (dirty_fields={})",
2709 entity_name,
2710 reason,
2711 dirty_tracker.len()
2712 ));
2713 } else {
2714 let patch =
2715 self.extract_partial_state_with_tracker(*state, &dirty_tracker)?;
2716
2717 let append = dirty_tracker.appended_paths();
2718 let mutation = Mutation {
2719 export: entity_name.clone(),
2720 key: primary_key,
2721 patch,
2722 append,
2723 };
2724 self.emit_debug(|| VmDebugEvent::EmitMutation {
2725 entity_name: entity_name.clone(),
2726 event_type: event_type.to_string(),
2727 key: mutation.key.clone(),
2728 emitted: true,
2729 reason: None,
2730 patch: Some(mutation.patch.clone()),
2731 dirty_fields,
2732 });
2733 output.push(mutation);
2734 }
2735 pc += 1;
2736 }
2737 OpCode::SetFieldIfNull {
2738 object,
2739 path,
2740 value,
2741 } => {
2742 let old_value = self
2743 .is_debug_enabled()
2744 .then(|| Self::get_value_at_path(&self.registers[*object], path))
2745 .flatten();
2746 let was_set = self.set_field_if_null(*object, path, *value)?;
2747 if was_set && should_emit(path) {
2748 dirty_tracker.mark_replaced(path);
2749 }
2750 if self.is_debug_enabled() {
2751 self.emit_debug(|| VmDebugEvent::FieldWrite {
2752 entity_name: entity_name.to_string(),
2753 event_type: event_type.to_string(),
2754 op: "set_field_if_null".to_string(),
2755 path: path.clone(),
2756 old_value,
2757 new_value: Self::get_value_at_path(&self.registers[*object], path),
2758 applied: was_set,
2759 reason: (!was_set).then_some("already_set".to_string()),
2760 });
2761 }
2762 pc += 1;
2763 }
2764 OpCode::SetFieldMax {
2765 object,
2766 path,
2767 value,
2768 } => {
2769 let was_updated = self.set_field_max(*object, path, *value)?;
2770 if was_updated && should_emit(path) {
2771 dirty_tracker.mark_replaced(path);
2772 }
2773 pc += 1;
2774 }
2775 OpCode::UpdateTemporalIndex {
2776 state_id: _,
2777 index_name,
2778 lookup_value,
2779 primary_key,
2780 timestamp,
2781 } => {
2782 let actual_state_id = override_state_id;
2783 let state = self
2784 .states
2785 .get_mut(&actual_state_id)
2786 .ok_or("State table not found")?;
2787 let index = state
2788 .temporal_indexes
2789 .entry(index_name.clone())
2790 .or_insert_with(TemporalIndex::new);
2791
2792 let lookup_val = self.registers[*lookup_value].clone();
2793 let pk_val = self.registers[*primary_key].clone();
2794 let ts_val = if let Some(val) = self.registers[*timestamp].as_i64() {
2795 val
2796 } else if let Some(val) = self.registers[*timestamp].as_u64() {
2797 val as i64
2798 } else {
2799 return Err(format!(
2800 "Timestamp must be a number (i64 or u64), got: {:?}",
2801 self.registers[*timestamp]
2802 )
2803 .into());
2804 };
2805
2806 index.insert(lookup_val, pk_val, ts_val);
2807 pc += 1;
2808 }
2809 OpCode::LookupTemporalIndex {
2810 state_id: _,
2811 index_name,
2812 lookup_value,
2813 timestamp,
2814 dest,
2815 } => {
2816 let actual_state_id = override_state_id;
2817 let state = self
2818 .states
2819 .get(&actual_state_id)
2820 .ok_or("State table not found")?;
2821 let lookup_val = &self.registers[*lookup_value];
2822
2823 let result = if self.registers[*timestamp].is_null() {
2824 if let Some(index) = state.temporal_indexes.get(index_name) {
2825 index.lookup_latest(lookup_val).unwrap_or(Value::Null)
2826 } else {
2827 Value::Null
2828 }
2829 } else {
2830 let ts_val = if let Some(val) = self.registers[*timestamp].as_i64() {
2831 val
2832 } else if let Some(val) = self.registers[*timestamp].as_u64() {
2833 val as i64
2834 } else {
2835 return Err(format!(
2836 "Timestamp must be a number (i64 or u64), got: {:?}",
2837 self.registers[*timestamp]
2838 )
2839 .into());
2840 };
2841
2842 if let Some(index) = state.temporal_indexes.get(index_name) {
2843 index.lookup(lookup_val, ts_val).unwrap_or(Value::Null)
2844 } else {
2845 Value::Null
2846 }
2847 };
2848
2849 self.registers[*dest] = result;
2850 pc += 1;
2851 }
2852 OpCode::UpdateLookupIndex {
2853 state_id: _,
2854 index_name,
2855 lookup_value,
2856 primary_key,
2857 } => {
2858 let actual_state_id = override_state_id;
2859 let state = self
2860 .states
2861 .get_mut(&actual_state_id)
2862 .ok_or("State table not found")?;
2863 let index = state
2864 .lookup_indexes
2865 .entry(index_name.clone())
2866 .or_insert_with(LookupIndex::new);
2867
2868 let lookup_val = self.registers[*lookup_value].clone();
2869 let pk_val = self.registers[*primary_key].clone();
2870
2871 index.insert(lookup_val.clone(), pk_val);
2872
2873 if let Some(key_str) = lookup_val.as_str() {
2875 self.last_lookup_index_keys.push(key_str.to_string());
2876 }
2877
2878 pc += 1;
2879 }
2880 OpCode::LookupIndex {
2881 state_id: _,
2882 index_name,
2883 lookup_value,
2884 dest,
2885 } => {
2886 let actual_state_id = override_state_id;
2887 let mut current_value = self.registers[*lookup_value].clone();
2888 let original_lookup_value = current_value.clone();
2889 let mut hops = if self.is_debug_enabled() {
2890 Some(Vec::<VmLookupHop>::new())
2891 } else {
2892 None
2893 };
2894 let mut miss_kind: Option<String> = None;
2895
2896 const MAX_CHAIN_DEPTH: usize = 5;
2897 let mut iterations = 0;
2898
2899 let final_result = if self.states.contains_key(&actual_state_id) {
2900 loop {
2901 iterations += 1;
2902 if iterations > MAX_CHAIN_DEPTH {
2903 break current_value;
2904 }
2905
2906 let resolved = self
2907 .states
2908 .get(&actual_state_id)
2909 .and_then(|state| {
2910 if let Some(index) = state.lookup_indexes.get(index_name) {
2911 if let Some(found) = index.lookup(¤t_value) {
2912 return Some(found);
2913 }
2914 }
2915
2916 for (name, index) in state.lookup_indexes.iter() {
2917 if name == index_name {
2918 continue;
2919 }
2920 if let Some(found) = index.lookup(¤t_value) {
2921 return Some(found);
2922 }
2923 }
2924
2925 None
2926 })
2927 .unwrap_or(Value::Null);
2928
2929 let lookup_result = resolved.clone();
2930 if let Some(hops) = hops.as_mut() {
2931 hops.push(VmLookupHop {
2932 source: "lookup_index".to_string(),
2933 input: current_value.clone(),
2934 result: lookup_result.clone(),
2935 chained: false,
2936 });
2937 }
2938
2939 let mut resolved_from_pda = false;
2940 let resolved = if resolved.is_null() {
2941 if let Some(pda_str) = current_value.as_str() {
2942 resolved_from_pda = true;
2943 let pda_result = self
2944 .states
2945 .get_mut(&actual_state_id)
2946 .and_then(|state_mut| {
2947 state_mut
2948 .pda_reverse_lookups
2949 .get_mut("default_pda_lookup")
2950 })
2951 .and_then(|pda_lookup| pda_lookup.lookup(pda_str))
2952 .map(Value::String)
2953 .unwrap_or(Value::Null);
2954 if let Some(hops) = hops.as_mut() {
2955 hops.push(VmLookupHop {
2956 source: "pda_reverse_lookup".to_string(),
2957 input: current_value.clone(),
2958 result: pda_result.clone(),
2959 chained: false,
2960 });
2961 }
2962 pda_result
2963 } else {
2964 Value::Null
2965 }
2966 } else {
2967 resolved
2968 };
2969
2970 if resolved.is_null() {
2971 if iterations == 1 {
2972 if let Some(pda_str) = current_value.as_str() {
2973 self.last_pda_lookup_miss = Some(pda_str.to_string());
2974 miss_kind = Some("pda_lookup_miss".to_string());
2975 } else {
2976 miss_kind = Some("lookup_index_miss".to_string());
2977 }
2978 }
2979 break Value::Null;
2980 }
2981
2982 let can_chain =
2983 self.can_resolve_further(&resolved, actual_state_id, index_name);
2984
2985 if !can_chain {
2986 if resolved_from_pda {
2987 if let Some(resolved_str) = resolved.as_str() {
2988 self.last_lookup_index_miss =
2989 Some(resolved_str.to_string());
2990 }
2991 miss_kind = Some("lookup_chain_incomplete".to_string());
2992 break Value::Null;
2993 }
2994 break resolved;
2995 }
2996
2997 if let Some(hops) = hops.as_mut() {
2998 if let Some(last) = hops.last_mut() {
2999 last.chained = true;
3000 }
3001 }
3002
3003 current_value = resolved;
3004 }
3005 } else {
3006 miss_kind = Some("state_not_initialized".to_string());
3007 Value::Null
3008 };
3009
3010 self.emit_debug(|| VmDebugEvent::LookupIndex {
3011 entity_name: entity_name.to_string(),
3012 event_type: event_type.to_string(),
3013 index_name: index_name.clone(),
3014 lookup_value: original_lookup_value,
3015 hops: hops.unwrap_or_default(),
3016 final_result: final_result.clone(),
3017 miss_kind,
3018 });
3019
3020 self.registers[*dest] = final_result;
3021 pc += 1;
3022 }
3023 OpCode::SetFieldSum {
3024 object,
3025 path,
3026 value,
3027 } => {
3028 let was_updated = self.set_field_sum(*object, path, *value)?;
3029 if was_updated && should_emit(path) {
3030 dirty_tracker.mark_replaced(path);
3031 }
3032 pc += 1;
3033 }
3034 OpCode::SetFieldIncrement { object, path } => {
3035 let was_updated = self.set_field_increment(*object, path)?;
3036 if was_updated && should_emit(path) {
3037 dirty_tracker.mark_replaced(path);
3038 }
3039 pc += 1;
3040 }
3041 OpCode::SetFieldMin {
3042 object,
3043 path,
3044 value,
3045 } => {
3046 let was_updated = self.set_field_min(*object, path, *value)?;
3047 if was_updated && should_emit(path) {
3048 dirty_tracker.mark_replaced(path);
3049 }
3050 pc += 1;
3051 }
3052 OpCode::AddToUniqueSet {
3053 state_id: _,
3054 set_name,
3055 value,
3056 count_object,
3057 count_path,
3058 } => {
3059 let value_to_add = self.registers[*value].clone();
3060
3061 let set_field_path = format!("__unique_set:{}", set_name);
3064
3065 let mut set: HashSet<Value> =
3067 if let Ok(existing) = self.get_field(*count_object, &set_field_path) {
3068 if !existing.is_null() {
3069 serde_json::from_value(existing).unwrap_or_default()
3070 } else {
3071 HashSet::new()
3072 }
3073 } else {
3074 HashSet::new()
3075 };
3076
3077 let was_new = set.insert(value_to_add);
3079
3080 let set_as_vec: Vec<Value> = set.iter().cloned().collect();
3082 self.registers[100] = serde_json::to_value(set_as_vec)?;
3083 self.set_field_auto_vivify(*count_object, &set_field_path, 100)?;
3084
3085 if was_new {
3087 self.registers[100] = Value::Number(serde_json::Number::from(set.len()));
3088 self.set_field_auto_vivify(*count_object, count_path, 100)?;
3089 if should_emit(count_path) {
3090 dirty_tracker.mark_replaced(count_path);
3091 }
3092 }
3093
3094 pc += 1;
3095 }
3096 OpCode::ConditionalSetField {
3097 object,
3098 path,
3099 value,
3100 condition_field,
3101 condition_op,
3102 condition_value,
3103 } => {
3104 let field_value = self.load_field(event_value, condition_field, None)?;
3105 let condition_met =
3106 self.evaluate_comparison(&field_value, condition_op, condition_value)?;
3107
3108 if condition_met {
3109 self.set_field_auto_vivify(*object, path, *value)?;
3110 if should_emit(path) {
3111 dirty_tracker.mark_replaced(path);
3112 }
3113 }
3114 pc += 1;
3115 }
3116 OpCode::SetFieldWhen {
3117 object,
3118 path,
3119 value,
3120 when_instruction,
3121 entity_name,
3122 key_reg,
3123 condition_field,
3124 condition_op,
3125 condition_value,
3126 } => {
3127 let actual_state_id = override_state_id;
3128 let condition_met = if let (Some(field), Some(op), Some(cond_value)) = (
3129 condition_field.as_ref(),
3130 condition_op.as_ref(),
3131 condition_value.as_ref(),
3132 ) {
3133 let field_value = self.load_field(event_value, field, None)?;
3134 self.evaluate_comparison(&field_value, op, cond_value)?
3135 } else {
3136 true
3137 };
3138
3139 if !condition_met {
3140 pc += 1;
3141 continue;
3142 }
3143
3144 let signature = self
3145 .current_context
3146 .as_ref()
3147 .and_then(|c| c.signature.clone())
3148 .unwrap_or_default();
3149
3150 let emit = should_emit(path);
3151
3152 let instruction_seen = if !signature.is_empty() {
3153 if let Some(state) = self.states.get(&actual_state_id) {
3154 let mut cache = state.recent_tx_instructions.lock().unwrap();
3155 cache
3156 .get(&signature)
3157 .map(|set| set.contains(when_instruction))
3158 .unwrap_or(false)
3159 } else {
3160 false
3161 }
3162 } else {
3163 false
3164 };
3165
3166 if instruction_seen {
3167 self.set_field_auto_vivify(*object, path, *value)?;
3168 if emit {
3169 dirty_tracker.mark_replaced(path);
3170 }
3171 } else if !signature.is_empty() {
3172 let deferred = DeferredWhenOperation {
3173 entity_name: entity_name.clone(),
3174 primary_key: self.registers[*key_reg].clone(),
3175 field_path: path.clone(),
3176 field_value: self.registers[*value].clone(),
3177 when_instruction: when_instruction.clone(),
3178 signature: signature.clone(),
3179 slot: self
3180 .current_context
3181 .as_ref()
3182 .and_then(|c| c.slot)
3183 .unwrap_or(0),
3184 deferred_at: std::time::SystemTime::now()
3185 .duration_since(std::time::UNIX_EPOCH)
3186 .unwrap()
3187 .as_secs() as i64,
3188 emit,
3189 };
3190
3191 if let Some(state) = self.states.get(&actual_state_id) {
3192 let key = (signature, when_instruction.clone());
3193 state
3194 .deferred_when_ops
3195 .entry(key)
3196 .or_insert_with(Vec::new)
3197 .push(deferred);
3198 }
3199 }
3200
3201 pc += 1;
3202 }
3203 OpCode::SetFieldUnlessStopped {
3204 object,
3205 path,
3206 value,
3207 stop_field,
3208 stop_instruction,
3209 entity_name,
3210 key_reg: _,
3211 } => {
3212 let stop_value = self.get_field(*object, stop_field).unwrap_or(Value::Null);
3213 let stopped = stop_value.as_bool().unwrap_or(false);
3214 let old_value = self
3215 .is_debug_enabled()
3216 .then(|| Self::get_value_at_path(&self.registers[*object], path))
3217 .flatten();
3218
3219 if stopped {
3220 self.emit_debug(|| VmDebugEvent::FieldWrite {
3221 entity_name: entity_name.clone(),
3222 event_type: event_type.to_string(),
3223 op: "set_field_unless_stopped".to_string(),
3224 path: path.clone(),
3225 old_value,
3226 new_value: Self::get_value_at_path(&self.registers[*object], path),
3227 applied: false,
3228 reason: Some(format!("stop_flag:{}", stop_instruction)),
3229 });
3230 tracing::debug!(
3231 entity = %entity_name,
3232 field = %path,
3233 stop_field = %stop_field,
3234 stop_instruction = %stop_instruction,
3235 "stop flag set; skipping field update"
3236 );
3237 pc += 1;
3238 continue;
3239 }
3240
3241 self.set_field_auto_vivify(*object, path, *value)?;
3242 if should_emit(path) {
3243 dirty_tracker.mark_replaced(path);
3244 }
3245 if self.is_debug_enabled() {
3246 self.emit_debug(|| VmDebugEvent::FieldWrite {
3247 entity_name: entity_name.clone(),
3248 event_type: event_type.to_string(),
3249 op: "set_field_unless_stopped".to_string(),
3250 path: path.clone(),
3251 old_value,
3252 new_value: Self::get_value_at_path(&self.registers[*object], path),
3253 applied: true,
3254 reason: None,
3255 });
3256 }
3257 pc += 1;
3258 }
3259 OpCode::ConditionalIncrement {
3260 object,
3261 path,
3262 condition_field,
3263 condition_op,
3264 condition_value,
3265 } => {
3266 let field_value = self.load_field(event_value, condition_field, None)?;
3267 let condition_met =
3268 self.evaluate_comparison(&field_value, condition_op, condition_value)?;
3269
3270 if condition_met {
3271 let was_updated = self.set_field_increment(*object, path)?;
3272 if was_updated && should_emit(path) {
3273 dirty_tracker.mark_replaced(path);
3274 }
3275 }
3276 pc += 1;
3277 }
3278 OpCode::EvaluateComputedFields {
3279 state,
3280 computed_paths,
3281 } => {
3282 if let Some(evaluator) = entity_evaluator {
3283 let old_values: Vec<_> = computed_paths
3284 .iter()
3285 .map(|path| Self::get_value_at_path(&self.registers[*state], path))
3286 .collect();
3287
3288 let state_value = &mut self.registers[*state];
3289 let context_slot = self.current_context.as_ref().and_then(|c| c.slot);
3290 let context_timestamp = self
3291 .current_context
3292 .as_ref()
3293 .map(|c| c.timestamp())
3294 .unwrap_or_else(|| {
3295 std::time::SystemTime::now()
3296 .duration_since(std::time::UNIX_EPOCH)
3297 .unwrap()
3298 .as_secs() as i64
3299 });
3300 let eval_result = evaluator(state_value, context_slot, context_timestamp);
3301
3302 if eval_result.is_ok() {
3303 for (path, old_value) in computed_paths.iter().zip(old_values.iter()) {
3304 let new_value =
3305 Self::get_value_at_path(&self.registers[*state], path);
3306
3307 if new_value != *old_value && should_emit(path) {
3308 dirty_tracker.mark_replaced(path);
3309 }
3310 }
3311 }
3312 }
3313 pc += 1;
3314 }
3315 OpCode::QueueResolver {
3316 state_id: _,
3317 entity_name,
3318 resolver,
3319 input_path,
3320 input_value,
3321 url_template,
3322 strategy,
3323 extracts,
3324 condition,
3325 schedule_at,
3326 state,
3327 key,
3328 } => {
3329 let actual_state_id = override_state_id;
3330
3331 let skip_immediate_resolver = self
3334 .current_context
3335 .as_ref()
3336 .map(|c| c.skip_resolvers)
3337 .unwrap_or(false);
3338
3339 if let Some(cond) = condition {
3341 let field_val =
3342 Self::get_value_at_path(&self.registers[*state], &cond.field_path)
3343 .unwrap_or(Value::Null);
3344 let condition_met =
3345 self.evaluate_comparison(&field_val, &cond.op, &cond.value)?;
3346
3347 if !condition_met {
3348 pc += 1;
3349 continue;
3350 }
3351 }
3352
3353 if let Some(schedule_path) = schedule_at {
3355 let target_val =
3356 Self::get_value_at_path(&self.registers[*state], schedule_path);
3357
3358 match target_val.as_ref().and_then(|value| {
3359 value.as_u64().or_else(|| {
3360 value.as_str().and_then(|text| text.parse::<u64>().ok())
3361 })
3362 }) {
3363 Some(target_slot) => {
3364 let current_slot = self
3365 .current_context
3366 .as_ref()
3367 .and_then(|ctx| ctx.slot)
3368 .unwrap_or(0);
3369 if current_slot < target_slot {
3370 let key_value = &self.registers[*key];
3371 if !key_value.is_null() {
3372 self.scheduled_callbacks.push((
3373 target_slot,
3374 ScheduledCallback {
3375 state_id: actual_state_id,
3376 entity_name: entity_name.clone(),
3377 primary_key: key_value.clone(),
3378 resolver: resolver.clone(),
3379 url_template: url_template.clone(),
3380 input_value: input_value.clone(),
3381 input_path: input_path.clone(),
3382 condition: condition.clone(),
3383 strategy: strategy.clone(),
3384 extracts: extracts.clone(),
3385 retry_count: 0,
3386 },
3387 ));
3388 }
3389 pc += 1;
3390 continue;
3391 }
3392 }
3394 None => {
3395 pc += 1;
3399 continue;
3400 }
3401 }
3402 }
3403
3404 if skip_immediate_resolver {
3405 pc += 1;
3406 continue;
3407 }
3408
3409 let resolved_input = if let Some(template) = url_template {
3411 crate::scheduler::build_url_from_template(template, &self.registers[*state])
3412 .map(Value::String)
3413 } else if let Some(value) = input_value {
3414 Some(value.clone())
3415 } else if let Some(path) = input_path.as_ref() {
3416 Self::get_value_at_path(&self.registers[*state], path)
3417 } else {
3418 None
3419 };
3420
3421 if let Some(input) = resolved_input {
3422 let key_value = &self.registers[*key];
3423
3424 if input.is_null() || key_value.is_null() {
3425 pc += 1;
3426 continue;
3427 }
3428
3429 if matches!(strategy, ResolveStrategy::SetOnce)
3430 && extracts.iter().all(|extract| {
3434 match Self::get_value_at_path(
3435 &self.registers[*state],
3436 &extract.target_path,
3437 ) {
3438 Some(value) => !value.is_null(),
3439 None => false,
3440 }
3441 })
3442 {
3443 pc += 1;
3444 continue;
3445 }
3446
3447 let cache_key = resolver_cache_key(resolver, &input);
3448
3449 match self.get_cached_resolver_value(&cache_key) {
3450 Some(CachedResolverValue::Resolved(cached)) => {
3451 Self::apply_resolver_extractions_to_value(
3452 &mut self.registers[*state],
3453 &cached,
3454 extracts,
3455 &mut dirty_tracker,
3456 &should_emit,
3457 )?;
3458 }
3459 Some(CachedResolverValue::Negative) => {}
3460 None => {
3461 let target = ResolverTarget {
3462 state_id: actual_state_id,
3463 entity_name: entity_name.clone(),
3464 primary_key: self.registers[*key].clone(),
3465 extracts: extracts.clone(),
3466 };
3467
3468 self.enqueue_resolver_request(
3469 cache_key,
3470 resolver.clone(),
3471 input,
3472 target,
3473 );
3474 }
3475 }
3476 }
3477
3478 pc += 1;
3479 }
3480 OpCode::UpdatePdaReverseLookup {
3481 state_id: _,
3482 lookup_name,
3483 pda_address,
3484 primary_key,
3485 } => {
3486 let actual_state_id = override_state_id;
3487 let pda_val = self.registers[*pda_address].clone();
3488 let pk_val = self.registers[*primary_key].clone();
3489
3490 if let (Some(pda_str), Some(pk_str)) = (pda_val.as_str(), pk_val.as_str()) {
3491 let pending = self.update_pda_reverse_lookup(
3492 actual_state_id,
3493 lookup_name,
3494 pda_str.to_string(),
3495 pk_str.to_string(),
3496 )?;
3497 self.pending_pda_reprocess_updates.extend(pending);
3498 self.last_pda_registered = Some(pda_str.to_string());
3499 self.emit_debug(|| VmDebugEvent::PdaReverseLookupUpdate {
3500 entity_name: entity_name.to_string(),
3501 event_type: event_type.to_string(),
3502 lookup_name: lookup_name.clone(),
3503 pda_address: pda_str.to_string(),
3504 primary_key: pk_val.clone(),
3505 });
3506 } else if !pk_val.is_null() {
3507 if let Some(pk_num) = pk_val.as_u64() {
3508 if let Some(pda_str) = pda_val.as_str() {
3509 let pending = self.update_pda_reverse_lookup(
3510 actual_state_id,
3511 lookup_name,
3512 pda_str.to_string(),
3513 pk_num.to_string(),
3514 )?;
3515 self.pending_pda_reprocess_updates.extend(pending);
3516 self.last_pda_registered = Some(pda_str.to_string());
3517 self.emit_debug(|| VmDebugEvent::PdaReverseLookupUpdate {
3518 entity_name: entity_name.to_string(),
3519 event_type: event_type.to_string(),
3520 lookup_name: lookup_name.clone(),
3521 pda_address: pda_str.to_string(),
3522 primary_key: pk_val.clone(),
3523 });
3524 }
3525 }
3526 }
3527
3528 pc += 1;
3529 }
3530 }
3531
3532 self.instructions_executed += 1;
3533 }
3534
3535 Ok(output)
3536 }
3537
3538 fn load_field(
3539 &self,
3540 event_value: &Value,
3541 path: &FieldPath,
3542 default: Option<&Value>,
3543 ) -> Result<Value> {
3544 if path.segments.is_empty() {
3545 if let Some(obj) = event_value.as_object() {
3546 let filtered: serde_json::Map<String, Value> = obj
3547 .iter()
3548 .filter(|(k, _)| !k.starts_with("__"))
3549 .map(|(k, v)| (k.clone(), v.clone()))
3550 .collect();
3551 return Ok(Value::Object(filtered));
3552 }
3553 return Ok(event_value.clone());
3554 }
3555
3556 let mut current = event_value;
3557 for segment in path.segments.iter() {
3558 current = match current.get(segment) {
3559 Some(v) => v,
3560 None => {
3561 tracing::trace!(
3562 "load_field: segment={:?} not found in {:?}, returning default",
3563 segment,
3564 current
3565 );
3566 return Ok(default.cloned().unwrap_or(Value::Null));
3567 }
3568 };
3569 }
3570
3571 tracing::trace!("load_field: path={:?}, result={:?}", path.segments, current);
3572 Ok(current.clone())
3573 }
3574
3575 fn get_value_at_path(value: &Value, path: &str) -> Option<Value> {
3576 let mut current = value;
3577 for segment in path.split('.') {
3578 current = current.get(segment)?;
3579 }
3580 Some(current.clone())
3581 }
3582
3583 fn set_field_auto_vivify(
3584 &mut self,
3585 object_reg: Register,
3586 path: &str,
3587 value_reg: Register,
3588 ) -> Result<()> {
3589 let compiled = self.get_compiled_path(path);
3590 let segments = compiled.segments();
3591 let value = self.registers[value_reg].clone();
3592
3593 if !self.registers[object_reg].is_object() {
3594 self.registers[object_reg] = json!({});
3595 }
3596
3597 let obj = self.registers[object_reg]
3598 .as_object_mut()
3599 .ok_or("Not an object")?;
3600
3601 let mut current = obj;
3602 for (i, segment) in segments.iter().enumerate() {
3603 if i == segments.len() - 1 {
3604 current.insert(segment.to_string(), value);
3605 return Ok(());
3606 } else {
3607 current
3608 .entry(segment.to_string())
3609 .or_insert_with(|| json!({}));
3610 current = current
3611 .get_mut(segment)
3612 .and_then(|v| v.as_object_mut())
3613 .ok_or("Path collision: expected object")?;
3614 }
3615 }
3616
3617 Ok(())
3618 }
3619
3620 fn set_field_if_null(
3621 &mut self,
3622 object_reg: Register,
3623 path: &str,
3624 value_reg: Register,
3625 ) -> Result<bool> {
3626 let compiled = self.get_compiled_path(path);
3627 let segments = compiled.segments();
3628 let value = self.registers[value_reg].clone();
3629
3630 if value.is_null() {
3634 return Ok(false);
3635 }
3636
3637 if !self.registers[object_reg].is_object() {
3638 self.registers[object_reg] = json!({});
3639 }
3640
3641 let obj = self.registers[object_reg]
3642 .as_object_mut()
3643 .ok_or("Not an object")?;
3644
3645 let mut current = obj;
3646 for (i, segment) in segments.iter().enumerate() {
3647 if i == segments.len() - 1 {
3648 if !current.contains_key(segment) || current.get(segment).unwrap().is_null() {
3649 current.insert(segment.to_string(), value);
3650 return Ok(true);
3651 }
3652 return Ok(false);
3653 } else {
3654 current
3655 .entry(segment.to_string())
3656 .or_insert_with(|| json!({}));
3657 current = current
3658 .get_mut(segment)
3659 .and_then(|v| v.as_object_mut())
3660 .ok_or("Path collision: expected object")?;
3661 }
3662 }
3663
3664 Ok(false)
3665 }
3666
3667 fn set_field_max(
3668 &mut self,
3669 object_reg: Register,
3670 path: &str,
3671 value_reg: Register,
3672 ) -> Result<bool> {
3673 let compiled = self.get_compiled_path(path);
3674 let segments = compiled.segments();
3675 let new_value = self.registers[value_reg].clone();
3676
3677 if !self.registers[object_reg].is_object() {
3678 self.registers[object_reg] = json!({});
3679 }
3680
3681 let obj = self.registers[object_reg]
3682 .as_object_mut()
3683 .ok_or("Not an object")?;
3684
3685 let mut current = obj;
3686 for (i, segment) in segments.iter().enumerate() {
3687 if i == segments.len() - 1 {
3688 let should_update = if let Some(current_value) = current.get(segment) {
3689 if current_value.is_null() {
3690 true
3691 } else {
3692 match (current_value.as_i64(), new_value.as_i64()) {
3693 (Some(current_val), Some(new_val)) => new_val > current_val,
3694 (Some(current_val), None) if new_value.as_u64().is_some() => {
3695 new_value.as_u64().unwrap() as i64 > current_val
3696 }
3697 (None, Some(new_val)) if current_value.as_u64().is_some() => {
3698 new_val > current_value.as_u64().unwrap() as i64
3699 }
3700 (None, None) => match (current_value.as_u64(), new_value.as_u64()) {
3701 (Some(current_val), Some(new_val)) => new_val > current_val,
3702 _ => match (current_value.as_f64(), new_value.as_f64()) {
3703 (Some(current_val), Some(new_val)) => new_val > current_val,
3704 _ => false,
3705 },
3706 },
3707 _ => false,
3708 }
3709 }
3710 } else {
3711 true
3712 };
3713
3714 if should_update {
3715 current.insert(segment.to_string(), new_value);
3716 return Ok(true);
3717 }
3718 return Ok(false);
3719 } else {
3720 current
3721 .entry(segment.to_string())
3722 .or_insert_with(|| json!({}));
3723 current = current
3724 .get_mut(segment)
3725 .and_then(|v| v.as_object_mut())
3726 .ok_or("Path collision: expected object")?;
3727 }
3728 }
3729
3730 Ok(false)
3731 }
3732
3733 fn set_field_sum(
3734 &mut self,
3735 object_reg: Register,
3736 path: &str,
3737 value_reg: Register,
3738 ) -> Result<bool> {
3739 let compiled = self.get_compiled_path(path);
3740 let segments = compiled.segments();
3741 let new_value = &self.registers[value_reg];
3742
3743 tracing::trace!(
3745 "set_field_sum: path={:?}, value={:?}, value_type={}",
3746 path,
3747 new_value,
3748 match new_value {
3749 serde_json::Value::Null => "null",
3750 serde_json::Value::Bool(_) => "bool",
3751 serde_json::Value::Number(_) => "number",
3752 serde_json::Value::String(_) => "string",
3753 serde_json::Value::Array(_) => "array",
3754 serde_json::Value::Object(_) => "object",
3755 }
3756 );
3757 let new_val_num = new_value
3758 .as_i64()
3759 .or_else(|| new_value.as_u64().map(|n| n as i64))
3760 .ok_or("Sum requires numeric value")?;
3761
3762 if !self.registers[object_reg].is_object() {
3763 self.registers[object_reg] = json!({});
3764 }
3765
3766 let obj = self.registers[object_reg]
3767 .as_object_mut()
3768 .ok_or("Not an object")?;
3769
3770 let mut current = obj;
3771 for (i, segment) in segments.iter().enumerate() {
3772 if i == segments.len() - 1 {
3773 let current_val = current
3774 .get(segment)
3775 .and_then(|v| {
3776 if v.is_null() {
3777 None
3778 } else {
3779 v.as_i64().or_else(|| v.as_u64().map(|n| n as i64))
3780 }
3781 })
3782 .unwrap_or(0);
3783
3784 let sum = current_val + new_val_num;
3785 current.insert(segment.to_string(), json!(sum));
3786 return Ok(true);
3787 } else {
3788 current
3789 .entry(segment.to_string())
3790 .or_insert_with(|| json!({}));
3791 current = current
3792 .get_mut(segment)
3793 .and_then(|v| v.as_object_mut())
3794 .ok_or("Path collision: expected object")?;
3795 }
3796 }
3797
3798 Ok(false)
3799 }
3800
3801 fn set_field_increment(&mut self, object_reg: Register, path: &str) -> Result<bool> {
3802 let compiled = self.get_compiled_path(path);
3803 let segments = compiled.segments();
3804
3805 if !self.registers[object_reg].is_object() {
3806 self.registers[object_reg] = json!({});
3807 }
3808
3809 let obj = self.registers[object_reg]
3810 .as_object_mut()
3811 .ok_or("Not an object")?;
3812
3813 let mut current = obj;
3814 for (i, segment) in segments.iter().enumerate() {
3815 if i == segments.len() - 1 {
3816 let current_val = current
3818 .get(segment)
3819 .and_then(|v| {
3820 if v.is_null() {
3821 None
3822 } else {
3823 v.as_i64().or_else(|| v.as_u64().map(|n| n as i64))
3824 }
3825 })
3826 .unwrap_or(0);
3827
3828 let incremented = current_val + 1;
3829 current.insert(segment.to_string(), json!(incremented));
3830 return Ok(true);
3831 } else {
3832 current
3833 .entry(segment.to_string())
3834 .or_insert_with(|| json!({}));
3835 current = current
3836 .get_mut(segment)
3837 .and_then(|v| v.as_object_mut())
3838 .ok_or("Path collision: expected object")?;
3839 }
3840 }
3841
3842 Ok(false)
3843 }
3844
3845 fn set_field_min(
3846 &mut self,
3847 object_reg: Register,
3848 path: &str,
3849 value_reg: Register,
3850 ) -> Result<bool> {
3851 let compiled = self.get_compiled_path(path);
3852 let segments = compiled.segments();
3853 let new_value = self.registers[value_reg].clone();
3854
3855 if !self.registers[object_reg].is_object() {
3856 self.registers[object_reg] = json!({});
3857 }
3858
3859 let obj = self.registers[object_reg]
3860 .as_object_mut()
3861 .ok_or("Not an object")?;
3862
3863 let mut current = obj;
3864 for (i, segment) in segments.iter().enumerate() {
3865 if i == segments.len() - 1 {
3866 let should_update = if let Some(current_value) = current.get(segment) {
3867 if current_value.is_null() {
3868 true
3869 } else {
3870 match (current_value.as_i64(), new_value.as_i64()) {
3871 (Some(current_val), Some(new_val)) => new_val < current_val,
3872 (Some(current_val), None) if new_value.as_u64().is_some() => {
3873 (new_value.as_u64().unwrap() as i64) < current_val
3874 }
3875 (None, Some(new_val)) if current_value.as_u64().is_some() => {
3876 new_val < current_value.as_u64().unwrap() as i64
3877 }
3878 (None, None) => match (current_value.as_u64(), new_value.as_u64()) {
3879 (Some(current_val), Some(new_val)) => new_val < current_val,
3880 _ => match (current_value.as_f64(), new_value.as_f64()) {
3881 (Some(current_val), Some(new_val)) => new_val < current_val,
3882 _ => false,
3883 },
3884 },
3885 _ => false,
3886 }
3887 }
3888 } else {
3889 true
3890 };
3891
3892 if should_update {
3893 current.insert(segment.to_string(), new_value);
3894 return Ok(true);
3895 }
3896 return Ok(false);
3897 } else {
3898 current
3899 .entry(segment.to_string())
3900 .or_insert_with(|| json!({}));
3901 current = current
3902 .get_mut(segment)
3903 .and_then(|v| v.as_object_mut())
3904 .ok_or("Path collision: expected object")?;
3905 }
3906 }
3907
3908 Ok(false)
3909 }
3910
3911 fn get_field(&mut self, object_reg: Register, path: &str) -> Result<Value> {
3912 let compiled = self.get_compiled_path(path);
3913 let segments = compiled.segments();
3914 let mut current = &self.registers[object_reg];
3915
3916 for segment in segments {
3917 current = current
3918 .get(segment)
3919 .ok_or_else(|| format!("Field not found: {}", segment))?;
3920 }
3921
3922 Ok(current.clone())
3923 }
3924
3925 fn append_to_array(
3926 &mut self,
3927 object_reg: Register,
3928 path: &str,
3929 value_reg: Register,
3930 max_length: usize,
3931 ) -> Result<()> {
3932 let compiled = self.get_compiled_path(path);
3933 let segments = compiled.segments();
3934 let value = self.registers[value_reg].clone();
3935
3936 if !self.registers[object_reg].is_object() {
3937 self.registers[object_reg] = json!({});
3938 }
3939
3940 let obj = self.registers[object_reg]
3941 .as_object_mut()
3942 .ok_or("Not an object")?;
3943
3944 let mut current = obj;
3945 for (i, segment) in segments.iter().enumerate() {
3946 if i == segments.len() - 1 {
3947 current
3948 .entry(segment.to_string())
3949 .or_insert_with(|| json!([]));
3950 let arr = current
3951 .get_mut(segment)
3952 .and_then(|v| v.as_array_mut())
3953 .ok_or("Path is not an array")?;
3954 arr.push(value.clone());
3955
3956 if arr.len() > max_length {
3957 let excess = arr.len() - max_length;
3958 arr.drain(0..excess);
3959 }
3960 } else {
3961 current
3962 .entry(segment.to_string())
3963 .or_insert_with(|| json!({}));
3964 current = current
3965 .get_mut(segment)
3966 .and_then(|v| v.as_object_mut())
3967 .ok_or("Path collision: expected object")?;
3968 }
3969 }
3970
3971 Ok(())
3972 }
3973
3974 fn transform_in_place(&mut self, reg: Register, transformation: &Transformation) -> Result<()> {
3975 let value = &self.registers[reg];
3976 let transformed = Self::apply_transformation(value, transformation)?;
3977 self.registers[reg] = transformed;
3978 Ok(())
3979 }
3980
3981 fn apply_transformation(value: &Value, transformation: &Transformation) -> Result<Value> {
3982 match transformation {
3983 Transformation::HexEncode => {
3984 if let Some(arr) = value.as_array() {
3985 let bytes: Vec<u8> = arr
3986 .iter()
3987 .filter_map(|v| v.as_u64().map(|n| n as u8))
3988 .collect();
3989 let hex = hex::encode(&bytes);
3990 Ok(json!(hex))
3991 } else if value.is_string() {
3992 Ok(value.clone())
3993 } else {
3994 Err("HexEncode requires an array of numbers".into())
3995 }
3996 }
3997 Transformation::HexDecode => {
3998 if let Some(s) = value.as_str() {
3999 let s = s.strip_prefix("0x").unwrap_or(s);
4000 let bytes = hex::decode(s).map_err(|e| format!("Hex decode error: {}", e))?;
4001 Ok(json!(bytes))
4002 } else {
4003 Err("HexDecode requires a string".into())
4004 }
4005 }
4006 Transformation::Base58Encode => {
4007 if let Some(arr) = value.as_array() {
4008 let bytes: Vec<u8> = arr
4009 .iter()
4010 .filter_map(|v| v.as_u64().map(|n| n as u8))
4011 .collect();
4012 let encoded = bs58::encode(&bytes).into_string();
4013 Ok(json!(encoded))
4014 } else if value.is_string() {
4015 Ok(value.clone())
4016 } else {
4017 Err("Base58Encode requires an array of numbers".into())
4018 }
4019 }
4020 Transformation::Base58Decode => {
4021 if let Some(s) = value.as_str() {
4022 let bytes = bs58::decode(s)
4023 .into_vec()
4024 .map_err(|e| format!("Base58 decode error: {}", e))?;
4025 Ok(json!(bytes))
4026 } else {
4027 Err("Base58Decode requires a string".into())
4028 }
4029 }
4030 Transformation::ToString => Ok(json!(value.to_string())),
4031 Transformation::ToNumber => {
4032 if let Some(s) = value.as_str() {
4033 let n = s
4034 .parse::<i64>()
4035 .map_err(|e| format!("Parse error: {}", e))?;
4036 Ok(json!(n))
4037 } else {
4038 Ok(value.clone())
4039 }
4040 }
4041 }
4042 }
4043
4044 fn evaluate_comparison(
4045 &self,
4046 field_value: &Value,
4047 op: &ComparisonOp,
4048 condition_value: &Value,
4049 ) -> Result<bool> {
4050 use ComparisonOp::*;
4051
4052 match op {
4053 Equal => Ok(field_value == condition_value),
4054 NotEqual => Ok(field_value != condition_value),
4055 GreaterThan => {
4056 match (field_value.as_i64(), condition_value.as_i64()) {
4058 (Some(a), Some(b)) => Ok(a > b),
4059 _ => match (field_value.as_u64(), condition_value.as_u64()) {
4060 (Some(a), Some(b)) => Ok(a > b),
4061 _ => match (field_value.as_f64(), condition_value.as_f64()) {
4062 (Some(a), Some(b)) => Ok(a > b),
4063 _ => Err("Cannot compare non-numeric values with GreaterThan".into()),
4064 },
4065 },
4066 }
4067 }
4068 GreaterThanOrEqual => match (field_value.as_i64(), condition_value.as_i64()) {
4069 (Some(a), Some(b)) => Ok(a >= b),
4070 _ => match (field_value.as_u64(), condition_value.as_u64()) {
4071 (Some(a), Some(b)) => Ok(a >= b),
4072 _ => match (field_value.as_f64(), condition_value.as_f64()) {
4073 (Some(a), Some(b)) => Ok(a >= b),
4074 _ => {
4075 Err("Cannot compare non-numeric values with GreaterThanOrEqual".into())
4076 }
4077 },
4078 },
4079 },
4080 LessThan => match (field_value.as_i64(), condition_value.as_i64()) {
4081 (Some(a), Some(b)) => Ok(a < b),
4082 _ => match (field_value.as_u64(), condition_value.as_u64()) {
4083 (Some(a), Some(b)) => Ok(a < b),
4084 _ => match (field_value.as_f64(), condition_value.as_f64()) {
4085 (Some(a), Some(b)) => Ok(a < b),
4086 _ => Err("Cannot compare non-numeric values with LessThan".into()),
4087 },
4088 },
4089 },
4090 LessThanOrEqual => match (field_value.as_i64(), condition_value.as_i64()) {
4091 (Some(a), Some(b)) => Ok(a <= b),
4092 _ => match (field_value.as_u64(), condition_value.as_u64()) {
4093 (Some(a), Some(b)) => Ok(a <= b),
4094 _ => match (field_value.as_f64(), condition_value.as_f64()) {
4095 (Some(a), Some(b)) => Ok(a <= b),
4096 _ => Err("Cannot compare non-numeric values with LessThanOrEqual".into()),
4097 },
4098 },
4099 },
4100 }
4101 }
4102
4103 fn can_resolve_further(&self, value: &Value, state_id: u32, index_name: &str) -> bool {
4104 if let Some(state) = self.states.get(&state_id) {
4105 if let Some(index) = state.lookup_indexes.get(index_name) {
4106 if index.lookup(value).is_some() {
4107 return true;
4108 }
4109 }
4110
4111 for (name, index) in state.lookup_indexes.iter() {
4112 if name == index_name {
4113 continue;
4114 }
4115 if index.lookup(value).is_some() {
4116 return true;
4117 }
4118 }
4119
4120 if let Some(pda_str) = value.as_str() {
4121 if let Some(pda_lookup) = state.pda_reverse_lookups.get("default_pda_lookup") {
4122 if pda_lookup.contains(pda_str) {
4123 return true;
4124 }
4125 }
4126 }
4127 }
4128
4129 false
4130 }
4131
4132 #[allow(clippy::type_complexity)]
4133 fn apply_deferred_when_op(
4134 &mut self,
4135 state_id: u32,
4136 op: &DeferredWhenOperation,
4137 entity_evaluator: Option<
4138 &Box<dyn Fn(&mut Value, Option<u64>, i64) -> ComputedEvaluatorResult + Send + Sync>,
4139 >,
4140 computed_paths: Option<&[String]>,
4141 ) -> Result<Vec<Mutation>> {
4142 let state = self.states.get(&state_id).ok_or("State not found")?;
4143
4144 if op.primary_key.is_null() {
4145 return Ok(vec![]);
4146 }
4147
4148 let mut entity_state = state
4149 .get_and_touch(&op.primary_key)
4150 .unwrap_or_else(|| json!({}));
4151
4152 let old_computed_values: Vec<_> = computed_paths
4154 .map(|paths| {
4155 paths
4156 .iter()
4157 .map(|path| Self::get_value_at_path(&entity_state, path))
4158 .collect()
4159 })
4160 .unwrap_or_default();
4161
4162 Self::set_nested_field_value(&mut entity_state, &op.field_path, op.field_value.clone())?;
4163
4164 if let Some(evaluator) = entity_evaluator {
4166 let context_slot = self.current_context.as_ref().and_then(|c| c.slot);
4167 let context_timestamp = self
4168 .current_context
4169 .as_ref()
4170 .map(|c| c.timestamp())
4171 .unwrap_or_else(|| {
4172 std::time::SystemTime::now()
4173 .duration_since(std::time::UNIX_EPOCH)
4174 .unwrap()
4175 .as_secs() as i64
4176 });
4177
4178 tracing::debug!(
4179 entity_name = %op.entity_name,
4180 primary_key = %op.primary_key,
4181 field_path = %op.field_path,
4182 "Re-evaluating computed fields after deferred when-op"
4183 );
4184
4185 if let Err(e) = evaluator(&mut entity_state, context_slot, context_timestamp) {
4186 tracing::warn!(
4187 entity_name = %op.entity_name,
4188 primary_key = %op.primary_key,
4189 error = %e,
4190 "Failed to evaluate computed fields after deferred when-op"
4191 );
4192 }
4193 }
4194
4195 state.insert_with_eviction(op.primary_key.clone(), entity_state.clone());
4196
4197 if !op.emit {
4198 return Ok(vec![]);
4199 }
4200
4201 let mut patch = json!({});
4202 Self::set_nested_field_value(&mut patch, &op.field_path, op.field_value.clone())?;
4203
4204 if let Some(paths) = computed_paths {
4206 tracing::debug!(
4207 entity_name = %op.entity_name,
4208 primary_key = %op.primary_key,
4209 computed_paths_count = paths.len(),
4210 "Checking computed fields for changes after deferred when-op"
4211 );
4212 for (path, old_value) in paths.iter().zip(old_computed_values.iter()) {
4213 let new_value = Self::get_value_at_path(&entity_state, path);
4214 tracing::debug!(
4215 entity_name = %op.entity_name,
4216 primary_key = %op.primary_key,
4217 field_path = %path,
4218 old_value = ?old_value,
4219 new_value = ?new_value,
4220 "Comparing computed field values"
4221 );
4222 if let Some(ref new_val) = new_value {
4223 if Some(new_val) != old_value.as_ref() {
4224 Self::set_nested_field_value(&mut patch, path, new_val.clone())?;
4225 tracing::info!(
4226 entity_name = %op.entity_name,
4227 primary_key = %op.primary_key,
4228 field_path = %path,
4229 "Computed field changed after deferred when-op, including in mutation"
4230 );
4231 }
4232 }
4233 }
4234 }
4235
4236 Ok(vec![Mutation {
4237 export: op.entity_name.clone(),
4238 key: op.primary_key.clone(),
4239 patch,
4240 append: vec![],
4241 }])
4242 }
4243
4244 fn set_nested_field_value(obj: &mut Value, path: &str, value: Value) -> Result<()> {
4245 let parts: Vec<&str> = path.split('.').collect();
4246 let mut current = obj;
4247
4248 for (i, part) in parts.iter().enumerate() {
4249 if i == parts.len() - 1 {
4250 if let Some(map) = current.as_object_mut() {
4251 map.insert(part.to_string(), value);
4252 return Ok(());
4253 }
4254 return Err("Cannot set field on non-object".into());
4255 }
4256
4257 if current.get(*part).is_none() || !current.get(*part).unwrap().is_object() {
4258 if let Some(map) = current.as_object_mut() {
4259 map.insert(part.to_string(), json!({}));
4260 }
4261 }
4262
4263 current = current.get_mut(*part).ok_or("Path navigation failed")?;
4264 }
4265
4266 Ok(())
4267 }
4268
4269 pub fn cleanup_expired_when_ops(&mut self, state_id: u32, max_age_secs: i64) -> usize {
4270 let now = std::time::SystemTime::now()
4271 .duration_since(std::time::UNIX_EPOCH)
4272 .unwrap()
4273 .as_secs() as i64;
4274
4275 let state = match self.states.get(&state_id) {
4276 Some(s) => s,
4277 None => return 0,
4278 };
4279
4280 let mut removed = 0;
4281 state.deferred_when_ops.retain(|_, ops| {
4282 let before = ops.len();
4283 ops.retain(|op| now - op.deferred_at < max_age_secs);
4284 removed += before - ops.len();
4285 !ops.is_empty()
4286 });
4287
4288 removed
4289 }
4290
4291 #[cfg_attr(feature = "otel", instrument(
4300 name = "vm.update_pda_lookup",
4301 skip(self),
4302 fields(
4303 pda = %pda_address,
4304 seed = %seed_value,
4305 )
4306 ))]
4307 pub fn update_pda_reverse_lookup(
4308 &mut self,
4309 state_id: u32,
4310 lookup_name: &str,
4311 pda_address: String,
4312 seed_value: String,
4313 ) -> Result<Vec<PendingAccountUpdate>> {
4314 let state = self
4315 .states
4316 .get_mut(&state_id)
4317 .ok_or("State table not found")?;
4318
4319 let lookup = state
4320 .pda_reverse_lookups
4321 .entry(lookup_name.to_string())
4322 .or_insert_with(|| PdaReverseLookup::new(DEFAULT_MAX_PDA_REVERSE_LOOKUP_ENTRIES));
4323
4324 let old_seed = lookup.index.peek(&pda_address).cloned();
4328 let mapping_changed = old_seed
4329 .as_ref()
4330 .map(|old| old != &seed_value)
4331 .unwrap_or(false);
4332
4333 if !mapping_changed && old_seed.is_none() {
4334 tracing::info!(
4335 pda = %pda_address,
4336 seed = %seed_value,
4337 "[PDA] First-time PDA reverse lookup established"
4338 );
4339 } else if !mapping_changed {
4340 tracing::debug!(
4341 pda = %pda_address,
4342 seed = %seed_value,
4343 "[PDA] PDA reverse lookup re-registered (same mapping)"
4344 );
4345 }
4346
4347 let evicted_pda = lookup.insert(pda_address.clone(), seed_value.clone());
4348
4349 if let Some(ref evicted) = evicted_pda {
4350 if let Some((_, evicted_updates)) = state.pending_updates.remove(evicted) {
4351 let count = evicted_updates.len();
4352 self.pending_queue_size = self.pending_queue_size.saturating_sub(count as u64);
4353 }
4354 }
4355
4356 let mut pending = self.flush_pending_updates(state_id, &pda_address)?;
4358
4359 if mapping_changed {
4367 if let Some(state) = self.states.get(&state_id) {
4368 for index in state.lookup_indexes.values() {
4370 index.remove(&Value::String(pda_address.clone()));
4371 }
4372
4373 if let Some((_, mut cached)) = state.last_account_data.remove(&pda_address) {
4374 tracing::info!(
4375 pda = %pda_address,
4376 old_seed = ?old_seed,
4377 new_seed = %seed_value,
4378 account_type = %cached.account_type,
4379 "PDA mapping changed — clearing stale indexes and reprocessing cached data"
4380 );
4381 cached.is_stale_reprocess = true;
4382 pending.push(cached);
4383 }
4384 }
4385 }
4386
4387 Ok(pending)
4388 }
4389
4390 pub fn cleanup_expired_pending_updates(&mut self, state_id: u32) -> usize {
4395 let state = match self.states.get_mut(&state_id) {
4396 Some(s) => s,
4397 None => return 0,
4398 };
4399
4400 let now = std::time::SystemTime::now()
4401 .duration_since(std::time::UNIX_EPOCH)
4402 .unwrap()
4403 .as_secs() as i64;
4404
4405 let mut removed_count = 0;
4406
4407 state.pending_updates.retain(|_pda_address, updates| {
4409 let original_len = updates.len();
4410
4411 updates.retain(|update| {
4412 let age = now - update.queued_at;
4413 age <= PENDING_UPDATE_TTL_SECONDS
4414 });
4415
4416 removed_count += original_len - updates.len();
4417
4418 !updates.is_empty()
4420 });
4421
4422 self.pending_queue_size = self.pending_queue_size.saturating_sub(removed_count as u64);
4424
4425 if removed_count > 0 {
4426 #[cfg(feature = "otel")]
4427 crate::vm_metrics::record_pending_updates_expired(
4428 removed_count as u64,
4429 &state.entity_name,
4430 );
4431 }
4432
4433 removed_count
4434 }
4435
4436 #[cfg_attr(feature = "otel", instrument(
4468 name = "vm.queue_account_update",
4469 skip(self, update),
4470 fields(
4471 pda = %update.pda_address,
4472 account_type = %update.account_type,
4473 slot = update.slot,
4474 )
4475 ))]
4476 pub fn queue_account_update(
4477 &mut self,
4478 state_id: u32,
4479 update: QueuedAccountUpdate,
4480 ) -> Result<()> {
4481 if self.pending_queue_size >= MAX_PENDING_UPDATES_TOTAL as u64 {
4482 self.cleanup_expired_pending_updates(state_id);
4483 if self.pending_queue_size >= MAX_PENDING_UPDATES_TOTAL as u64 {
4484 self.drop_oldest_pending_update(state_id)?;
4485 }
4486 }
4487
4488 let state = self
4489 .states
4490 .get_mut(&state_id)
4491 .ok_or("State table not found")?;
4492
4493 let pending = PendingAccountUpdate {
4494 account_type: update.account_type,
4495 pda_address: update.pda_address.clone(),
4496 account_data: update.account_data,
4497 slot: update.slot,
4498 write_version: update.write_version,
4499 signature: update.signature,
4500 queued_at: std::time::SystemTime::now()
4501 .duration_since(std::time::UNIX_EPOCH)
4502 .unwrap()
4503 .as_secs() as i64,
4504 is_stale_reprocess: false,
4505 };
4506
4507 let pda_address = pending.pda_address.clone();
4508 let slot = pending.slot;
4509
4510 let mut updates = state
4511 .pending_updates
4512 .entry(pda_address.clone())
4513 .or_insert_with(Vec::new);
4514
4515 let original_len = updates.len();
4516 updates.retain(|existing| existing.slot > slot);
4517 let removed_by_dedup = original_len - updates.len();
4518
4519 if removed_by_dedup > 0 {
4520 self.pending_queue_size = self
4521 .pending_queue_size
4522 .saturating_sub(removed_by_dedup as u64);
4523 }
4524
4525 if updates.len() >= MAX_PENDING_UPDATES_PER_PDA {
4526 updates.remove(0);
4527 self.pending_queue_size = self.pending_queue_size.saturating_sub(1);
4528 }
4529
4530 updates.push(pending);
4531 #[cfg(feature = "otel")]
4532 crate::vm_metrics::record_pending_update_queued(&state.entity_name);
4533
4534 Ok(())
4535 }
4536
4537 pub fn queue_instruction_event(
4538 &mut self,
4539 state_id: u32,
4540 event: QueuedInstructionEvent,
4541 ) -> Result<()> {
4542 let state = self
4543 .states
4544 .get_mut(&state_id)
4545 .ok_or("State table not found")?;
4546
4547 let pda_address = event.pda_address.clone();
4548
4549 let pending = PendingInstructionEvent {
4550 event_type: event.event_type,
4551 pda_address: event.pda_address,
4552 event_data: event.event_data,
4553 slot: event.slot,
4554 signature: event.signature,
4555 queued_at: std::time::SystemTime::now()
4556 .duration_since(std::time::UNIX_EPOCH)
4557 .unwrap()
4558 .as_secs() as i64,
4559 };
4560
4561 let mut events = state
4562 .pending_instruction_events
4563 .entry(pda_address)
4564 .or_insert_with(Vec::new);
4565
4566 if events.len() >= MAX_PENDING_UPDATES_PER_PDA {
4567 events.remove(0);
4568 }
4569
4570 events.push(pending);
4571
4572 Ok(())
4573 }
4574
4575 pub fn take_last_pda_lookup_miss(&mut self) -> Option<String> {
4576 self.last_pda_lookup_miss.take()
4577 }
4578
4579 pub fn take_last_lookup_index_miss(&mut self) -> Option<String> {
4580 self.last_lookup_index_miss.take()
4581 }
4582
4583 pub fn take_last_pda_registered(&mut self) -> Option<String> {
4584 self.last_pda_registered.take()
4585 }
4586
4587 pub fn take_last_lookup_index_keys(&mut self) -> Vec<String> {
4588 std::mem::take(&mut self.last_lookup_index_keys)
4589 }
4590
4591 pub fn flush_pending_instruction_events(
4592 &mut self,
4593 state_id: u32,
4594 pda_address: &str,
4595 ) -> Vec<PendingInstructionEvent> {
4596 let state = match self.states.get_mut(&state_id) {
4597 Some(s) => s,
4598 None => return Vec::new(),
4599 };
4600
4601 if let Some((_, events)) = state.pending_instruction_events.remove(pda_address) {
4602 events
4603 } else {
4604 Vec::new()
4605 }
4606 }
4607
4608 pub fn get_pending_queue_stats(&self, state_id: u32) -> Option<PendingQueueStats> {
4610 let state = self.states.get(&state_id)?;
4611
4612 let now = std::time::SystemTime::now()
4613 .duration_since(std::time::UNIX_EPOCH)
4614 .unwrap()
4615 .as_secs() as i64;
4616
4617 let mut total_updates = 0;
4618 let mut oldest_timestamp = now;
4619 let mut largest_pda_queue = 0;
4620 let mut estimated_memory = 0;
4621
4622 for entry in state.pending_updates.iter() {
4623 let (_, updates) = entry.pair();
4624 total_updates += updates.len();
4625 largest_pda_queue = largest_pda_queue.max(updates.len());
4626
4627 for update in updates.iter() {
4628 oldest_timestamp = oldest_timestamp.min(update.queued_at);
4629 estimated_memory += update.account_type.len() +
4631 update.pda_address.len() +
4632 update.signature.len() +
4633 16 + estimate_json_size(&update.account_data);
4635 }
4636 }
4637
4638 Some(PendingQueueStats {
4639 total_updates,
4640 unique_pdas: state.pending_updates.len(),
4641 oldest_age_seconds: now - oldest_timestamp,
4642 largest_pda_queue_size: largest_pda_queue,
4643 estimated_memory_bytes: estimated_memory,
4644 })
4645 }
4646
4647 pub fn get_memory_stats(&self, state_id: u32) -> VmMemoryStats {
4648 let mut stats = VmMemoryStats {
4649 path_cache_size: self.path_cache.len(),
4650 ..Default::default()
4651 };
4652
4653 if let Some(state) = self.states.get(&state_id) {
4654 stats.state_table_entity_count = state.data.len();
4655 stats.state_table_max_entries = state.config.max_entries;
4656 stats.state_table_at_capacity = state.is_at_capacity();
4657
4658 stats.lookup_index_count = state.lookup_indexes.len();
4659 stats.lookup_index_total_entries =
4660 state.lookup_indexes.values().map(|idx| idx.len()).sum();
4661
4662 stats.temporal_index_count = state.temporal_indexes.len();
4663 stats.temporal_index_total_entries = state
4664 .temporal_indexes
4665 .values()
4666 .map(|idx| idx.total_entries())
4667 .sum();
4668
4669 stats.pda_reverse_lookup_count = state.pda_reverse_lookups.len();
4670 stats.pda_reverse_lookup_total_entries = state
4671 .pda_reverse_lookups
4672 .values()
4673 .map(|lookup| lookup.len())
4674 .sum();
4675
4676 stats.version_tracker_entries = state.version_tracker.len();
4677
4678 stats.pending_queue_stats = self.get_pending_queue_stats(state_id);
4679 }
4680
4681 stats
4682 }
4683
4684 pub fn cleanup_all_expired(&mut self, state_id: u32) -> CleanupResult {
4685 let pending_removed = self.cleanup_expired_pending_updates(state_id);
4686 let temporal_removed = self.cleanup_temporal_indexes(state_id);
4687
4688 #[cfg(feature = "otel")]
4689 if let Some(state) = self.states.get(&state_id) {
4690 crate::vm_metrics::record_cleanup(
4691 pending_removed,
4692 temporal_removed,
4693 &state.entity_name,
4694 );
4695 }
4696
4697 CleanupResult {
4698 pending_updates_removed: pending_removed,
4699 temporal_entries_removed: temporal_removed,
4700 }
4701 }
4702
4703 fn cleanup_temporal_indexes(&mut self, state_id: u32) -> usize {
4704 let state = match self.states.get_mut(&state_id) {
4705 Some(s) => s,
4706 None => return 0,
4707 };
4708
4709 let now = std::time::SystemTime::now()
4710 .duration_since(std::time::UNIX_EPOCH)
4711 .unwrap()
4712 .as_secs() as i64;
4713
4714 let cutoff = now - TEMPORAL_HISTORY_TTL_SECONDS;
4715 let mut total_removed = 0;
4716
4717 for index in state.temporal_indexes.values_mut() {
4718 total_removed += index.cleanup_expired(cutoff);
4719 }
4720
4721 total_removed
4722 }
4723
4724 pub fn check_state_table_capacity(&self, state_id: u32) -> Option<CapacityWarning> {
4725 let state = self.states.get(&state_id)?;
4726
4727 if state.is_at_capacity() {
4728 Some(CapacityWarning {
4729 current_entries: state.data.len(),
4730 max_entries: state.config.max_entries,
4731 entries_over_limit: state.entries_over_limit(),
4732 })
4733 } else {
4734 None
4735 }
4736 }
4737
4738 fn drop_oldest_pending_update(&mut self, state_id: u32) -> Result<()> {
4740 let state = self
4741 .states
4742 .get_mut(&state_id)
4743 .ok_or("State table not found")?;
4744
4745 let mut oldest_pda: Option<String> = None;
4746 let mut oldest_timestamp = i64::MAX;
4747
4748 for entry in state.pending_updates.iter() {
4750 let (pda, updates) = entry.pair();
4751 if let Some(update) = updates.first() {
4752 if update.queued_at < oldest_timestamp {
4753 oldest_timestamp = update.queued_at;
4754 oldest_pda = Some(pda.clone());
4755 }
4756 }
4757 }
4758
4759 if let Some(pda) = oldest_pda {
4761 if let Some(mut updates) = state.pending_updates.get_mut(&pda) {
4762 if !updates.is_empty() {
4763 updates.remove(0);
4764 self.pending_queue_size = self.pending_queue_size.saturating_sub(1);
4765
4766 if updates.is_empty() {
4768 drop(updates);
4769 state.pending_updates.remove(&pda);
4770 }
4771 }
4772 }
4773 }
4774
4775 Ok(())
4776 }
4777
4778 fn flush_pending_updates(
4783 &mut self,
4784 state_id: u32,
4785 pda_address: &str,
4786 ) -> Result<Vec<PendingAccountUpdate>> {
4787 let state = self
4788 .states
4789 .get_mut(&state_id)
4790 .ok_or("State table not found")?;
4791
4792 if let Some((_, pending_updates)) = state.pending_updates.remove(pda_address) {
4793 let count = pending_updates.len();
4794 self.pending_queue_size = self.pending_queue_size.saturating_sub(count as u64);
4795 #[cfg(feature = "otel")]
4796 crate::vm_metrics::record_pending_updates_flushed(count as u64, &state.entity_name);
4797 Ok(pending_updates)
4798 } else {
4799 Ok(Vec::new())
4800 }
4801 }
4802
4803 pub fn cache_last_account_data(
4809 &mut self,
4810 state_id: u32,
4811 pda_address: &str,
4812 update: PendingAccountUpdate,
4813 ) {
4814 if let Some(state) = self.states.get(&state_id) {
4815 state
4816 .last_account_data
4817 .insert(pda_address.to_string(), update);
4818 }
4819 }
4820
4821 pub fn try_pda_reverse_lookup(
4823 &mut self,
4824 state_id: u32,
4825 lookup_name: &str,
4826 pda_address: &str,
4827 ) -> Option<String> {
4828 let state = self.states.get_mut(&state_id)?;
4829
4830 if let Some(lookup) = state.pda_reverse_lookups.get_mut(lookup_name) {
4831 if let Some(value) = lookup.lookup(pda_address) {
4832 self.pda_cache_hits += 1;
4833 return Some(value);
4834 }
4835 }
4836
4837 self.pda_cache_misses += 1;
4838 None
4839 }
4840
4841 pub fn try_lookup_index_resolution(&self, state_id: u32, value: &Value) -> Option<Value> {
4844 let state = self.states.get(&state_id)?;
4845
4846 for index in state.lookup_indexes.values() {
4847 if let Some(resolved) = index.lookup(value) {
4848 return Some(resolved);
4849 }
4850 }
4851
4852 None
4853 }
4854
4855 pub fn try_chained_pda_lookup(
4860 &mut self,
4861 state_id: u32,
4862 lookup_name: &str,
4863 pda_address: &str,
4864 ) -> Option<String> {
4865 let pda_result = self.try_pda_reverse_lookup(state_id, lookup_name, pda_address)?;
4867
4868 let pda_value = Value::String(pda_result.clone());
4871 if let Some(resolved) = self.try_lookup_index_resolution(state_id, &pda_value) {
4872 resolved.as_str().map(|s| s.to_string())
4873 } else {
4874 pda_value.as_str().map(|s| s.to_string())
4876 }
4877 }
4878
4879 pub fn evaluate_computed_expr(&self, expr: &ComputedExpr, state: &Value) -> Result<Value> {
4886 self.evaluate_computed_expr_with_env(expr, state, &std::collections::HashMap::new())
4887 }
4888
4889 fn evaluate_computed_expr_with_env(
4891 &self,
4892 expr: &ComputedExpr,
4893 state: &Value,
4894 env: &std::collections::HashMap<String, Value>,
4895 ) -> Result<Value> {
4896 match expr {
4897 ComputedExpr::FieldRef { path } => self.get_field_from_state(state, path),
4898
4899 ComputedExpr::Var { name } => env
4900 .get(name)
4901 .cloned()
4902 .ok_or_else(|| format!("Undefined variable: {}", name).into()),
4903
4904 ComputedExpr::Let { name, value, body } => {
4905 let val = self.evaluate_computed_expr_with_env(value, state, env)?;
4906 let mut new_env = env.clone();
4907 new_env.insert(name.clone(), val);
4908 self.evaluate_computed_expr_with_env(body, state, &new_env)
4909 }
4910
4911 ComputedExpr::If {
4912 condition,
4913 then_branch,
4914 else_branch,
4915 } => {
4916 let cond_val = self.evaluate_computed_expr_with_env(condition, state, env)?;
4917 if self.value_to_bool(&cond_val) {
4918 self.evaluate_computed_expr_with_env(then_branch, state, env)
4919 } else {
4920 self.evaluate_computed_expr_with_env(else_branch, state, env)
4921 }
4922 }
4923
4924 ComputedExpr::None => Ok(Value::Null),
4925
4926 ComputedExpr::Some { value } => self.evaluate_computed_expr_with_env(value, state, env),
4927
4928 ComputedExpr::Slice { expr, start, end } => {
4929 let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
4930 match val {
4931 Value::Array(arr) => {
4932 let slice: Vec<Value> = arr.get(*start..*end).unwrap_or(&[]).to_vec();
4933 Ok(Value::Array(slice))
4934 }
4935 _ => Err(format!("Cannot slice non-array value: {:?}", val).into()),
4936 }
4937 }
4938
4939 ComputedExpr::Index { expr, index } => {
4940 let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
4941 match val {
4942 Value::Array(arr) => Ok(arr.get(*index).cloned().unwrap_or(Value::Null)),
4943 _ => Err(format!("Cannot index non-array value: {:?}", val).into()),
4944 }
4945 }
4946
4947 ComputedExpr::U64FromLeBytes { bytes } => {
4948 let val = self.evaluate_computed_expr_with_env(bytes, state, env)?;
4949 let byte_vec = self.value_to_bytes(&val)?;
4950 if byte_vec.len() < 8 {
4951 return Err(format!(
4952 "u64::from_le_bytes requires 8 bytes, got {}",
4953 byte_vec.len()
4954 )
4955 .into());
4956 }
4957 let arr: [u8; 8] = byte_vec[..8]
4958 .try_into()
4959 .map_err(|_| "Failed to convert to [u8; 8]")?;
4960 Ok(Value::Number(serde_json::Number::from(u64::from_le_bytes(
4961 arr,
4962 ))))
4963 }
4964
4965 ComputedExpr::U64FromBeBytes { bytes } => {
4966 let val = self.evaluate_computed_expr_with_env(bytes, state, env)?;
4967 let byte_vec = self.value_to_bytes(&val)?;
4968 if byte_vec.len() < 8 {
4969 return Err(format!(
4970 "u64::from_be_bytes requires 8 bytes, got {}",
4971 byte_vec.len()
4972 )
4973 .into());
4974 }
4975 let arr: [u8; 8] = byte_vec[..8]
4976 .try_into()
4977 .map_err(|_| "Failed to convert to [u8; 8]")?;
4978 Ok(Value::Number(serde_json::Number::from(u64::from_be_bytes(
4979 arr,
4980 ))))
4981 }
4982
4983 ComputedExpr::ByteArray { bytes } => {
4984 Ok(Value::Array(bytes.iter().map(|b| json!(*b)).collect()))
4985 }
4986
4987 ComputedExpr::Closure { param, body } => {
4988 Ok(json!({
4991 "__closure": {
4992 "param": param,
4993 "body": serde_json::to_value(body).unwrap_or(Value::Null)
4994 }
4995 }))
4996 }
4997
4998 ComputedExpr::Unary { op, expr } => {
4999 let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5000 self.apply_unary_op(op, &val)
5001 }
5002
5003 ComputedExpr::JsonToBytes { expr } => {
5004 let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5005 let bytes = self.value_to_bytes(&val)?;
5007 Ok(Value::Array(bytes.iter().map(|b| json!(*b)).collect()))
5008 }
5009
5010 ComputedExpr::UnwrapOr { expr, default } => {
5011 let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5012 if val.is_null() {
5013 Ok(default.clone())
5014 } else {
5015 Ok(val)
5016 }
5017 }
5018
5019 ComputedExpr::Binary { op, left, right } => {
5020 let l = self.evaluate_computed_expr_with_env(left, state, env)?;
5021 let r = self.evaluate_computed_expr_with_env(right, state, env)?;
5022 self.apply_binary_op(op, &l, &r)
5023 }
5024
5025 ComputedExpr::Cast { expr, to_type } => {
5026 let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5027 self.apply_cast(&val, to_type)
5028 }
5029
5030 ComputedExpr::ResolverComputed {
5031 resolver,
5032 method,
5033 args,
5034 } => {
5035 let evaluated_args: Vec<Value> = args
5036 .iter()
5037 .map(|arg| self.evaluate_computed_expr_with_env(arg, state, env))
5038 .collect::<Result<Vec<_>>>()?;
5039 crate::resolvers::evaluate_resolver_computed(resolver, method, &evaluated_args)
5040 }
5041
5042 ComputedExpr::MethodCall { expr, method, args } => {
5043 let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5044 if (method == "map" || method == "filter") && args.len() == 1 {
5046 if let ComputedExpr::Closure { param, body } = &args[0] {
5047 if val.is_null() {
5049 return Ok(Value::Null);
5050 }
5051
5052 if method == "map" {
5053 if let Value::Array(arr) = &val {
5054 let results: Result<Vec<Value>> = arr
5055 .iter()
5056 .map(|elem| {
5057 let mut closure_env = env.clone();
5058 closure_env.insert(param.clone(), elem.clone());
5059 self.evaluate_computed_expr_with_env(
5060 body,
5061 state,
5062 &closure_env,
5063 )
5064 })
5065 .collect();
5066 return Ok(Value::Array(results?));
5067 }
5068
5069 let mut closure_env = env.clone();
5070 closure_env.insert(param.clone(), val);
5071 return self.evaluate_computed_expr_with_env(body, state, &closure_env);
5072 }
5073
5074 if let Value::Array(arr) = &val {
5075 let results: Result<Vec<Value>> = arr
5076 .iter()
5077 .filter_map(|elem| {
5078 let mut closure_env = env.clone();
5079 closure_env.insert(param.clone(), elem.clone());
5080 match self.evaluate_computed_expr_with_env(
5081 body,
5082 state,
5083 &closure_env,
5084 ) {
5085 Ok(predicate) if self.value_to_bool(&predicate) => {
5086 Some(Ok(elem.clone()))
5087 }
5088 Ok(_) => None,
5089 Err(error) => Some(Err(error)),
5090 }
5091 })
5092 .collect();
5093 return Ok(Value::Array(results?));
5094 }
5095
5096 let mut closure_env = env.clone();
5097 closure_env.insert(param.clone(), val.clone());
5098 let predicate =
5099 self.evaluate_computed_expr_with_env(body, state, &closure_env)?;
5100 return Ok(if self.value_to_bool(&predicate) {
5101 val
5102 } else {
5103 Value::Null
5104 });
5105 }
5106 }
5107 let evaluated_args: Vec<Value> = args
5108 .iter()
5109 .map(|a| self.evaluate_computed_expr_with_env(a, state, env))
5110 .collect::<Result<Vec<_>>>()?;
5111 self.apply_method_call(&val, method, &evaluated_args)
5112 }
5113
5114 ComputedExpr::Literal { value } => Ok(value.clone()),
5115
5116 ComputedExpr::Paren { expr } => self.evaluate_computed_expr_with_env(expr, state, env),
5117
5118 ComputedExpr::ContextSlot => Ok(self
5119 .current_context
5120 .as_ref()
5121 .and_then(|ctx| ctx.slot)
5122 .map(|s| json!(s))
5123 .unwrap_or(Value::Null)),
5124
5125 ComputedExpr::ContextTimestamp => Ok(self
5126 .current_context
5127 .as_ref()
5128 .map(|ctx| json!(ctx.timestamp()))
5129 .unwrap_or(Value::Null)),
5130
5131 ComputedExpr::Keccak256 { expr } => {
5132 let val = self.evaluate_computed_expr_with_env(expr, state, env)?;
5133 let bytes = self.value_to_bytes(&val)?;
5134 use sha3::{Digest, Keccak256};
5135 let hash = Keccak256::digest(&bytes);
5136 Ok(Value::Array(
5137 hash.to_vec().iter().map(|b| json!(*b)).collect(),
5138 ))
5139 }
5140 }
5141 }
5142
5143 fn value_to_bytes(&self, val: &Value) -> Result<Vec<u8>> {
5145 match val {
5146 Value::Array(arr) => arr
5147 .iter()
5148 .map(|v| {
5149 v.as_u64()
5150 .map(|n| n as u8)
5151 .ok_or_else(|| "Array element not a valid byte".into())
5152 })
5153 .collect(),
5154 Value::String(s) => {
5155 if s.starts_with("0x") || s.starts_with("0X") {
5157 hex::decode(&s[2..]).map_err(|e| format!("Invalid hex string: {}", e).into())
5158 } else {
5159 hex::decode(s).map_err(|e| format!("Invalid hex string: {}", e).into())
5160 }
5161 }
5162 _ => Err(format!("Cannot convert {:?} to bytes", val).into()),
5163 }
5164 }
5165
5166 fn apply_unary_op(&self, op: &crate::ast::UnaryOp, val: &Value) -> Result<Value> {
5168 use crate::ast::UnaryOp;
5169 match op {
5170 UnaryOp::Not => Ok(json!(!self.value_to_bool(val))),
5171 UnaryOp::ReverseBits => match val.as_u64() {
5172 Some(n) => Ok(json!(n.reverse_bits())),
5173 None => match val.as_i64() {
5174 Some(n) => Ok(json!((n as u64).reverse_bits())),
5175 None => Err("reverse_bits requires an integer".into()),
5176 },
5177 },
5178 }
5179 }
5180
5181 fn get_field_from_state(&self, state: &Value, path: &str) -> Result<Value> {
5183 let segments: Vec<&str> = path.split('.').collect();
5184 let mut current = state;
5185
5186 for segment in segments {
5187 match current.get(segment) {
5188 Some(v) => current = v,
5189 None => return Ok(Value::Null),
5190 }
5191 }
5192
5193 Ok(current.clone())
5194 }
5195
5196 fn apply_binary_op(&self, op: &BinaryOp, left: &Value, right: &Value) -> Result<Value> {
5198 match op {
5199 BinaryOp::Add => self.numeric_op(left, right, |a, b| a + b, |a, b| a + b),
5201 BinaryOp::Sub => self.numeric_op(left, right, |a, b| a - b, |a, b| a - b),
5202 BinaryOp::Mul => self.numeric_op(left, right, |a, b| a * b, |a, b| a * b),
5203 BinaryOp::Div => {
5204 if let Some(r) = right.as_i64() {
5206 if r == 0 {
5207 return Err("Division by zero".into());
5208 }
5209 }
5210 if let Some(r) = right.as_f64() {
5211 if r == 0.0 {
5212 return Err("Division by zero".into());
5213 }
5214 }
5215 self.numeric_op(left, right, |a, b| a / b, |a, b| a / b)
5216 }
5217 BinaryOp::Mod => {
5218 match (left.as_i64(), right.as_i64()) {
5220 (Some(a), Some(b)) if b != 0 => Ok(json!(a % b)),
5221 (None, _) | (_, None) => match (left.as_u64(), right.as_u64()) {
5222 (Some(a), Some(b)) if b != 0 => Ok(json!(a % b)),
5223 _ => Err("Modulo requires non-zero integer operands".into()),
5224 },
5225 _ => Err("Modulo by zero".into()),
5226 }
5227 }
5228
5229 BinaryOp::Gt => self.comparison_op(left, right, |a, b| a > b, |a, b| a > b),
5231 BinaryOp::Lt => self.comparison_op(left, right, |a, b| a < b, |a, b| a < b),
5232 BinaryOp::Gte => self.comparison_op(left, right, |a, b| a >= b, |a, b| a >= b),
5233 BinaryOp::Lte => self.comparison_op(left, right, |a, b| a <= b, |a, b| a <= b),
5234 BinaryOp::Eq => Ok(json!(left == right)),
5235 BinaryOp::Ne => Ok(json!(left != right)),
5236
5237 BinaryOp::And => {
5239 let l_bool = self.value_to_bool(left);
5240 let r_bool = self.value_to_bool(right);
5241 Ok(json!(l_bool && r_bool))
5242 }
5243 BinaryOp::Or => {
5244 let l_bool = self.value_to_bool(left);
5245 let r_bool = self.value_to_bool(right);
5246 Ok(json!(l_bool || r_bool))
5247 }
5248
5249 BinaryOp::Xor => match (left.as_u64(), right.as_u64()) {
5251 (Some(a), Some(b)) => Ok(json!(a ^ b)),
5252 _ => match (left.as_i64(), right.as_i64()) {
5253 (Some(a), Some(b)) => Ok(json!(a ^ b)),
5254 _ => Err("XOR requires integer operands".into()),
5255 },
5256 },
5257 BinaryOp::BitAnd => match (left.as_u64(), right.as_u64()) {
5258 (Some(a), Some(b)) => Ok(json!(a & b)),
5259 _ => match (left.as_i64(), right.as_i64()) {
5260 (Some(a), Some(b)) => Ok(json!(a & b)),
5261 _ => Err("BitAnd requires integer operands".into()),
5262 },
5263 },
5264 BinaryOp::BitOr => match (left.as_u64(), right.as_u64()) {
5265 (Some(a), Some(b)) => Ok(json!(a | b)),
5266 _ => match (left.as_i64(), right.as_i64()) {
5267 (Some(a), Some(b)) => Ok(json!(a | b)),
5268 _ => Err("BitOr requires integer operands".into()),
5269 },
5270 },
5271 BinaryOp::Shl => match (left.as_u64(), right.as_u64()) {
5272 (Some(a), Some(b)) => Ok(json!(a << b)),
5273 _ => match (left.as_i64(), right.as_i64()) {
5274 (Some(a), Some(b)) => Ok(json!(a << b)),
5275 _ => Err("Shl requires integer operands".into()),
5276 },
5277 },
5278 BinaryOp::Shr => match (left.as_u64(), right.as_u64()) {
5279 (Some(a), Some(b)) => Ok(json!(a >> b)),
5280 _ => match (left.as_i64(), right.as_i64()) {
5281 (Some(a), Some(b)) => Ok(json!(a >> b)),
5282 _ => Err("Shr requires integer operands".into()),
5283 },
5284 },
5285 }
5286 }
5287
5288 fn numeric_op<F1, F2>(
5290 &self,
5291 left: &Value,
5292 right: &Value,
5293 int_op: F1,
5294 float_op: F2,
5295 ) -> Result<Value>
5296 where
5297 F1: Fn(i64, i64) -> i64,
5298 F2: Fn(f64, f64) -> f64,
5299 {
5300 if let (Some(a), Some(b)) = (left.as_i64(), right.as_i64()) {
5302 return Ok(json!(int_op(a, b)));
5303 }
5304
5305 if let (Some(a), Some(b)) = (left.as_u64(), right.as_u64()) {
5307 return Ok(json!(int_op(a as i64, b as i64)));
5309 }
5310
5311 if let (Some(a), Some(b)) = (left.as_f64(), right.as_f64()) {
5313 return Ok(json!(float_op(a, b)));
5314 }
5315
5316 if left.is_null() || right.is_null() {
5318 return Ok(Value::Null);
5319 }
5320
5321 Err(format!(
5322 "Cannot perform numeric operation on {:?} and {:?}",
5323 left, right
5324 )
5325 .into())
5326 }
5327
5328 fn comparison_op<F1, F2>(
5330 &self,
5331 left: &Value,
5332 right: &Value,
5333 int_cmp: F1,
5334 float_cmp: F2,
5335 ) -> Result<Value>
5336 where
5337 F1: Fn(i64, i64) -> bool,
5338 F2: Fn(f64, f64) -> bool,
5339 {
5340 if let (Some(a), Some(b)) = (left.as_i64(), right.as_i64()) {
5342 return Ok(json!(int_cmp(a, b)));
5343 }
5344
5345 if let (Some(a), Some(b)) = (left.as_u64(), right.as_u64()) {
5347 return Ok(json!(int_cmp(a as i64, b as i64)));
5348 }
5349
5350 if let (Some(a), Some(b)) = (left.as_f64(), right.as_f64()) {
5352 return Ok(json!(float_cmp(a, b)));
5353 }
5354
5355 if left.is_null() || right.is_null() {
5357 return Ok(json!(false));
5358 }
5359
5360 Err(format!("Cannot compare {:?} and {:?}", left, right).into())
5361 }
5362
5363 fn value_to_bool(&self, value: &Value) -> bool {
5365 match value {
5366 Value::Null => false,
5367 Value::Bool(b) => *b,
5368 Value::Number(n) => {
5369 if let Some(i) = n.as_i64() {
5370 i != 0
5371 } else if let Some(f) = n.as_f64() {
5372 f != 0.0
5373 } else {
5374 true
5375 }
5376 }
5377 Value::String(s) => !s.is_empty(),
5378 Value::Array(arr) => !arr.is_empty(),
5379 Value::Object(obj) => !obj.is_empty(),
5380 }
5381 }
5382
5383 fn apply_cast(&self, value: &Value, to_type: &str) -> Result<Value> {
5385 match to_type {
5386 "i8" | "i16" | "i32" | "i64" | "isize" => {
5387 if let Some(n) = value.as_i64() {
5388 Ok(json!(n))
5389 } else if let Some(n) = value.as_u64() {
5390 Ok(json!(n as i64))
5391 } else if let Some(n) = value.as_f64() {
5392 Ok(json!(n as i64))
5393 } else if let Some(s) = value.as_str() {
5394 s.parse::<i64>()
5395 .map(|n| json!(n))
5396 .map_err(|e| format!("Cannot parse '{}' as integer: {}", s, e).into())
5397 } else {
5398 Err(format!("Cannot cast {:?} to {}", value, to_type).into())
5399 }
5400 }
5401 "u8" | "u16" | "u32" | "u64" | "usize" => {
5402 if let Some(n) = value.as_u64() {
5403 Ok(json!(n))
5404 } else if let Some(n) = value.as_i64() {
5405 Ok(json!(n as u64))
5406 } else if let Some(n) = value.as_f64() {
5407 Ok(json!(n as u64))
5408 } else if let Some(s) = value.as_str() {
5409 s.parse::<u64>().map(|n| json!(n)).map_err(|e| {
5410 format!("Cannot parse '{}' as unsigned integer: {}", s, e).into()
5411 })
5412 } else {
5413 Err(format!("Cannot cast {:?} to {}", value, to_type).into())
5414 }
5415 }
5416 "f32" | "f64" => {
5417 if let Some(n) = value.as_f64() {
5418 Ok(json!(n))
5419 } else if let Some(n) = value.as_i64() {
5420 Ok(json!(n as f64))
5421 } else if let Some(n) = value.as_u64() {
5422 Ok(json!(n as f64))
5423 } else if let Some(s) = value.as_str() {
5424 s.parse::<f64>()
5425 .map(|n| json!(n))
5426 .map_err(|e| format!("Cannot parse '{}' as float: {}", s, e).into())
5427 } else {
5428 Err(format!("Cannot cast {:?} to {}", value, to_type).into())
5429 }
5430 }
5431 "String" | "string" => Ok(json!(value.to_string())),
5432 "bool" => Ok(json!(self.value_to_bool(value))),
5433 _ => {
5434 Ok(value.clone())
5436 }
5437 }
5438 }
5439
5440 fn apply_method_call(&self, value: &Value, method: &str, args: &[Value]) -> Result<Value> {
5442 match method {
5443 "unwrap_or" => {
5444 if value.is_null() && !args.is_empty() {
5445 Ok(args[0].clone())
5446 } else {
5447 Ok(value.clone())
5448 }
5449 }
5450 "unwrap_or_default" => {
5451 if value.is_null() {
5452 Ok(json!(0))
5454 } else {
5455 Ok(value.clone())
5456 }
5457 }
5458 "is_some" => Ok(json!(!value.is_null())),
5459 "is_none" => Ok(json!(value.is_null())),
5460 "is_null" => Ok(json!(value.is_null())),
5461 "or" => {
5462 if args.len() != 1 {
5463 return Err("or() requires exactly one argument".into());
5464 }
5465 if value.is_null() {
5466 Ok(args[0].clone())
5467 } else {
5468 Ok(value.clone())
5469 }
5470 }
5471 "abs" => {
5472 if let Some(n) = value.as_i64() {
5473 Ok(json!(n.abs()))
5474 } else if let Some(n) = value.as_f64() {
5475 Ok(json!(n.abs()))
5476 } else {
5477 Err(format!("Cannot call abs() on {:?}", value).into())
5478 }
5479 }
5480 "len" => {
5481 if let Some(s) = value.as_str() {
5482 Ok(json!(s.len()))
5483 } else if let Some(arr) = value.as_array() {
5484 Ok(json!(arr.len()))
5485 } else if let Some(obj) = value.as_object() {
5486 Ok(json!(obj.len()))
5487 } else {
5488 Err(format!("Cannot call len() on {:?}", value).into())
5489 }
5490 }
5491 "sum" => {
5492 if !args.is_empty() {
5493 return Err("sum() does not accept arguments".into());
5494 }
5495 if value.is_null() {
5496 return Ok(Value::Null);
5497 }
5498
5499 let values = value
5500 .as_array()
5501 .ok_or_else(|| format!("Cannot call sum() on {:?}", value))?;
5502
5503 let unsigned_values = values
5504 .iter()
5505 .map(|item| {
5506 item.as_u64()
5507 .or_else(|| item.as_str().and_then(|text| text.parse::<u64>().ok()))
5508 })
5509 .collect::<Option<Vec<_>>>();
5510 if let Some(unsigned_values) = unsigned_values {
5511 let total = unsigned_values
5512 .iter()
5513 .try_fold(0_u64, |total, item| total.checked_add(*item));
5514 return total
5515 .map(|total| json!(total))
5516 .ok_or_else(|| "sum() unsigned integer overflow".into());
5517 }
5518
5519 let signed_values = values
5520 .iter()
5521 .map(|item| {
5522 item.as_i64()
5523 .or_else(|| item.as_str().and_then(|text| text.parse::<i64>().ok()))
5524 })
5525 .collect::<Option<Vec<_>>>();
5526 if let Some(signed_values) = signed_values {
5527 let total = signed_values
5528 .iter()
5529 .try_fold(0_i64, |total, item| total.checked_add(*item));
5530 return total
5531 .map(|total| json!(total))
5532 .ok_or_else(|| "sum() signed integer overflow".into());
5533 }
5534
5535 let total = values.iter().try_fold(0.0_f64, |total, item| {
5536 item.as_f64()
5537 .or_else(|| item.as_str().and_then(|text| text.parse::<f64>().ok()))
5538 .map(|value| total + value)
5539 });
5540 match total {
5541 Some(total) if total.is_finite() => serde_json::Number::from_f64(total)
5542 .map(Value::Number)
5543 .ok_or_else(|| "Failed to serialize sum() result".into()),
5544 Some(_) => Err("sum() result is not finite".into()),
5545 None => Err(format!("Cannot sum non-numeric array {:?}", value).into()),
5546 }
5547 }
5548 "to_string" => Ok(json!(value.to_string())),
5549 "min" => {
5550 if args.is_empty() {
5551 return Err("min() requires an argument".into());
5552 }
5553 let other = &args[0];
5554 if let (Some(a), Some(b)) = (value.as_i64(), other.as_i64()) {
5555 Ok(json!(a.min(b)))
5556 } else if let (Some(a), Some(b)) = (value.as_f64(), other.as_f64()) {
5557 Ok(json!(a.min(b)))
5558 } else {
5559 Err(format!("Cannot call min() on {:?} and {:?}", value, other).into())
5560 }
5561 }
5562 "max" => {
5563 if args.is_empty() {
5564 return Err("max() requires an argument".into());
5565 }
5566 let other = &args[0];
5567 if let (Some(a), Some(b)) = (value.as_i64(), other.as_i64()) {
5568 Ok(json!(a.max(b)))
5569 } else if let (Some(a), Some(b)) = (value.as_f64(), other.as_f64()) {
5570 Ok(json!(a.max(b)))
5571 } else {
5572 Err(format!("Cannot call max() on {:?} and {:?}", value, other).into())
5573 }
5574 }
5575 "saturating_add" => {
5576 if args.is_empty() {
5577 return Err("saturating_add() requires an argument".into());
5578 }
5579 let other = &args[0];
5580 if let (Some(a), Some(b)) = (value.as_i64(), other.as_i64()) {
5581 Ok(json!(a.saturating_add(b)))
5582 } else if let (Some(a), Some(b)) = (value.as_u64(), other.as_u64()) {
5583 Ok(json!(a.saturating_add(b)))
5584 } else {
5585 Err(format!(
5586 "Cannot call saturating_add() on {:?} and {:?}",
5587 value, other
5588 )
5589 .into())
5590 }
5591 }
5592 "saturating_sub" => {
5593 if args.is_empty() {
5594 return Err("saturating_sub() requires an argument".into());
5595 }
5596 let other = &args[0];
5597 if let (Some(a), Some(b)) = (value.as_i64(), other.as_i64()) {
5598 Ok(json!(a.saturating_sub(b)))
5599 } else if let (Some(a), Some(b)) = (value.as_u64(), other.as_u64()) {
5600 Ok(json!(a.saturating_sub(b)))
5601 } else {
5602 Err(format!(
5603 "Cannot call saturating_sub() on {:?} and {:?}",
5604 value, other
5605 )
5606 .into())
5607 }
5608 }
5609 _ => Err(format!("Unknown method call: {}()", method).into()),
5610 }
5611 }
5612
5613 pub fn evaluate_computed_fields_from_ast(
5616 &self,
5617 state: &mut Value,
5618 computed_field_specs: &[ComputedFieldSpec],
5619 ) -> Result<Vec<String>> {
5620 let mut updated_paths = Vec::new();
5621
5622 crate::resolvers::validate_resolver_computed_specs(computed_field_specs)?;
5623
5624 for spec in computed_field_specs {
5625 match self.evaluate_computed_expr(&spec.expression, state) {
5626 Ok(result) => {
5627 self.set_field_in_state(state, &spec.target_path, result)?;
5628 updated_paths.push(spec.target_path.clone());
5629 }
5630 Err(e) => {
5631 tracing::warn!(
5632 target_path = %spec.target_path,
5633 error = %e,
5634 "Failed to evaluate computed field"
5635 );
5636 }
5637 }
5638 }
5639
5640 Ok(updated_paths)
5641 }
5642
5643 fn set_field_in_state(&self, state: &mut Value, path: &str, value: Value) -> Result<()> {
5645 let segments: Vec<&str> = path.split('.').collect();
5646
5647 if segments.is_empty() {
5648 return Err("Empty path".into());
5649 }
5650
5651 let mut current = state;
5653 for (i, segment) in segments.iter().enumerate() {
5654 if i == segments.len() - 1 {
5655 if let Some(obj) = current.as_object_mut() {
5657 obj.insert(segment.to_string(), value);
5658 return Ok(());
5659 } else {
5660 return Err(format!("Cannot set field '{}' on non-object", segment).into());
5661 }
5662 } else {
5663 if !current.is_object() {
5665 *current = json!({});
5666 }
5667 let obj = current.as_object_mut().unwrap();
5668 current = obj.entry(segment.to_string()).or_insert_with(|| json!({}));
5669 }
5670 }
5671
5672 Ok(())
5673 }
5674
5675 pub fn create_evaluator_from_specs(
5678 specs: Vec<ComputedFieldSpec>,
5679 ) -> impl Fn(&mut Value, Option<u64>, i64) -> ComputedEvaluatorResult + Send + Sync + 'static
5680 {
5681 move |state: &mut Value, context_slot: Option<u64>, context_timestamp: i64| {
5682 let mut vm = VmContext::new();
5683 vm.current_context = Some(UpdateContext {
5684 slot: context_slot,
5685 timestamp: Some(context_timestamp),
5686 ..Default::default()
5687 });
5688 vm.evaluate_computed_fields_from_ast(state, &specs)
5689 .map_err(|error| {
5690 Box::<dyn std::error::Error + Send + Sync>::from(std::io::Error::other(
5691 error.to_string(),
5692 ))
5693 })?;
5694 Ok(())
5695 }
5696 }
5697}
5698
5699impl Default for VmContext {
5700 fn default() -> Self {
5701 Self::new()
5702 }
5703}
5704
5705impl crate::resolvers::ReverseLookupUpdater for VmContext {
5707 fn update(&mut self, pda_address: String, seed_value: String) -> Vec<PendingAccountUpdate> {
5708 self.update_pda_reverse_lookup(0, "default_pda_lookup", pda_address, seed_value)
5710 .unwrap_or_else(|e| {
5711 tracing::error!("Failed to update PDA reverse lookup: {}", e);
5712 Vec::new()
5713 })
5714 }
5715
5716 fn flush_pending(&mut self, pda_address: &str) -> Vec<PendingAccountUpdate> {
5717 self.flush_pending_updates(0, pda_address)
5719 .unwrap_or_else(|e| {
5720 tracing::error!("Failed to flush pending updates: {}", e);
5721 Vec::new()
5722 })
5723 }
5724}
5725
5726#[cfg(test)]
5727mod tests {
5728 use super::*;
5729 use crate::ast::{
5730 BinaryOp, ComputedExpr, ComputedFieldSpec, HttpMethod, UrlResolverConfig, UrlSource,
5731 };
5732
5733 #[test]
5734 fn test_url_resolver_cache_key_uses_method_and_resolved_url() {
5735 let field_path_resolver = ResolverType::Url(UrlResolverConfig {
5736 url_source: UrlSource::FieldPath("metadata_uri".to_string()),
5737 method: HttpMethod::Get,
5738 extract_path: None,
5739 });
5740 let template_resolver = ResolverType::Url(UrlResolverConfig {
5741 url_source: UrlSource::Template(vec![ast::UrlTemplatePart::Literal(
5742 "https://example.com/metadata".to_string(),
5743 )]),
5744 method: HttpMethod::Get,
5745 extract_path: Some("data".to_string()),
5746 });
5747 let input = json!("https://cdn.example.com/token.json");
5748
5749 assert_eq!(
5750 resolver_cache_key(&field_path_resolver, &input),
5751 resolver_cache_key(&template_resolver, &input)
5752 );
5753 }
5754
5755 #[test]
5756 fn test_url_resolver_cache_key_distinguishes_http_method() {
5757 let get_resolver = ResolverType::Url(UrlResolverConfig {
5758 url_source: UrlSource::FieldPath("metadata_uri".to_string()),
5759 method: HttpMethod::Get,
5760 extract_path: None,
5761 });
5762 let post_resolver = ResolverType::Url(UrlResolverConfig {
5763 url_source: UrlSource::FieldPath("metadata_uri".to_string()),
5764 method: HttpMethod::Post,
5765 extract_path: None,
5766 });
5767 let input = json!("https://api.example.com/round");
5768
5769 assert_ne!(
5770 resolver_cache_key(&get_resolver, &input),
5771 resolver_cache_key(&post_resolver, &input)
5772 );
5773 }
5774
5775 #[test]
5776 fn test_expired_resolver_cache_entry_is_dropped() {
5777 let mut vm = VmContext::new();
5778 let resolver = ResolverType::Url(UrlResolverConfig {
5779 url_source: UrlSource::FieldPath("metadata_uri".to_string()),
5780 method: HttpMethod::Get,
5781 extract_path: None,
5782 });
5783 let input = json!("https://cdn.example.com/token.json");
5784 let cache_key = resolver_cache_key(&resolver, &input);
5785
5786 vm.resolver_cache.put(
5787 cache_key.clone(),
5788 ResolverCacheEntry {
5789 value: ResolverCacheValue::Resolved(json!({ "name": "Token" })),
5790 cached_at: Instant::now() - resolver_cache_ttl() - Duration::from_secs(1),
5791 ttl: resolver_cache_ttl(),
5792 },
5793 );
5794
5795 assert!(vm.get_cached_resolver_value(&cache_key).is_none());
5796 assert!(vm.resolver_cache.get(&cache_key).is_none());
5797 }
5798
5799 #[test]
5800 fn test_in_flight_resolver_request_collects_targets_without_requeueing() {
5801 let mut vm = VmContext::new();
5802 let resolver = ResolverType::Token;
5803 let input = json!("mint_123");
5804
5805 vm.enqueue_resolver_request(
5806 "ignored".to_string(),
5807 resolver.clone(),
5808 input.clone(),
5809 ResolverTarget {
5810 state_id: 0,
5811 entity_name: "TestEntity".to_string(),
5812 primary_key: json!("entity_1"),
5813 extracts: vec![],
5814 },
5815 );
5816
5817 let first_batch = vm.take_resolver_requests();
5818 assert_eq!(first_batch.len(), 1);
5819
5820 let cache_key = resolver_cache_key(&resolver, &input);
5821 assert!(vm.resolver_pending.get(&cache_key).unwrap().in_flight);
5822
5823 vm.enqueue_resolver_request(
5824 "ignored-again".to_string(),
5825 resolver,
5826 input,
5827 ResolverTarget {
5828 state_id: 0,
5829 entity_name: "TestEntity".to_string(),
5830 primary_key: json!("entity_2"),
5831 extracts: vec![],
5832 },
5833 );
5834
5835 assert!(vm.take_resolver_requests().is_empty());
5836 let pending = vm.resolver_pending.get(&cache_key).unwrap();
5837 assert!(pending.in_flight);
5838 assert_eq!(pending.targets.len(), 2);
5839 }
5840
5841 #[test]
5842 fn test_computed_field_preserves_integer_type() {
5843 let vm = VmContext::new();
5844
5845 let mut state = serde_json::json!({
5846 "trading": {
5847 "total_buy_volume": 20000000000_i64,
5848 "total_sell_volume": 17951316474_i64
5849 }
5850 });
5851
5852 let spec = ComputedFieldSpec {
5853 target_path: "trading.total_volume".to_string(),
5854 result_type: "Option<u64>".to_string(),
5855 expression: ComputedExpr::Binary {
5856 op: BinaryOp::Add,
5857 left: Box::new(ComputedExpr::UnwrapOr {
5858 expr: Box::new(ComputedExpr::FieldRef {
5859 path: "trading.total_buy_volume".to_string(),
5860 }),
5861 default: serde_json::json!(0),
5862 }),
5863 right: Box::new(ComputedExpr::UnwrapOr {
5864 expr: Box::new(ComputedExpr::FieldRef {
5865 path: "trading.total_sell_volume".to_string(),
5866 }),
5867 default: serde_json::json!(0),
5868 }),
5869 },
5870 };
5871
5872 vm.evaluate_computed_fields_from_ast(&mut state, &[spec])
5873 .unwrap();
5874
5875 let total_volume = state
5876 .get("trading")
5877 .and_then(|t| t.get("total_volume"))
5878 .expect("total_volume should exist");
5879
5880 let serialized = serde_json::to_string(total_volume).unwrap();
5881 assert!(
5882 !serialized.contains('.'),
5883 "Integer should not have decimal point: {}",
5884 serialized
5885 );
5886 assert_eq!(
5887 total_volume.as_i64(),
5888 Some(37951316474),
5889 "Value should be correct sum"
5890 );
5891 }
5892
5893 #[test]
5894 fn test_computed_array_sum_preserves_u64() {
5895 let vm = VmContext::new();
5896 let mut state = json!({
5897 "state": {
5898 "deployed": [422874702_u64, 569506895_u64, 573312366_u64]
5899 }
5900 });
5901 let spec = ComputedFieldSpec {
5902 target_path: "state.total_deployed".to_string(),
5903 result_type: "Option<u64>".to_string(),
5904 expression: ComputedExpr::MethodCall {
5905 expr: Box::new(ComputedExpr::FieldRef {
5906 path: "state.deployed".to_string(),
5907 }),
5908 method: "sum".to_string(),
5909 args: Vec::new(),
5910 },
5911 };
5912
5913 vm.evaluate_computed_fields_from_ast(&mut state, &[spec])
5914 .unwrap();
5915
5916 assert_eq!(state["state"]["total_deployed"], json!(1_565_693_963_u64));
5917 }
5918
5919 #[test]
5920 fn test_computed_array_sum_accepts_u64_strings() {
5921 let vm = VmContext::new();
5922 let mut state = json!({
5923 "state": {
5924 "deployed": ["422874702", "569506895", "573312366"]
5925 }
5926 });
5927 let spec = ComputedFieldSpec {
5928 target_path: "state.total_deployed".to_string(),
5929 result_type: "Option<u64>".to_string(),
5930 expression: ComputedExpr::MethodCall {
5931 expr: Box::new(ComputedExpr::FieldRef {
5932 path: "state.deployed".to_string(),
5933 }),
5934 method: "sum".to_string(),
5935 args: Vec::new(),
5936 },
5937 };
5938
5939 vm.evaluate_computed_fields_from_ast(&mut state, &[spec])
5940 .unwrap();
5941
5942 assert_eq!(state["state"]["total_deployed"], json!(1_565_693_963_u64));
5943 }
5944
5945 #[test]
5946 fn test_computed_array_sum_propagates_null() {
5947 let vm = VmContext::new();
5948 let mut state = json!({ "state": { "deployed": null } });
5949 let spec = ComputedFieldSpec {
5950 target_path: "state.total_deployed".to_string(),
5951 result_type: "Option<u64>".to_string(),
5952 expression: ComputedExpr::MethodCall {
5953 expr: Box::new(ComputedExpr::FieldRef {
5954 path: "state.deployed".to_string(),
5955 }),
5956 method: "sum".to_string(),
5957 args: Vec::new(),
5958 },
5959 };
5960
5961 vm.evaluate_computed_fields_from_ast(&mut state, &[spec])
5962 .unwrap();
5963
5964 assert!(state["state"]["total_deployed"].is_null());
5965 }
5966
5967 #[test]
5968 fn test_computed_option_filter_or_fallback() {
5969 fn specs() -> Vec<ComputedFieldSpec> {
5970 vec![
5971 ComputedFieldSpec {
5972 target_path: "results.pre_reveal_rng".to_string(),
5973 result_type: "Option<u64>".to_string(),
5974 expression: ComputedExpr::MethodCall {
5975 expr: Box::new(ComputedExpr::MethodCall {
5976 expr: Box::new(ComputedExpr::FieldRef {
5977 path: "results.rng".to_string(),
5978 }),
5979 method: "filter".to_string(),
5980 args: vec![ComputedExpr::Closure {
5981 param: "rng".to_string(),
5982 body: Box::new(ComputedExpr::Unary {
5983 op: crate::ast::UnaryOp::Not,
5984 expr: Box::new(ComputedExpr::MethodCall {
5985 expr: Box::new(ComputedExpr::Var {
5986 name: "rng".to_string(),
5987 }),
5988 method: "is_null".to_string(),
5989 args: Vec::new(),
5990 }),
5991 }),
5992 }],
5993 }),
5994 method: "or".to_string(),
5995 args: vec![ComputedExpr::FieldRef {
5996 path: "results.pre_reveal_rng_candidate".to_string(),
5997 }],
5998 },
5999 },
6000 ComputedFieldSpec {
6001 target_path: "results.pre_reveal_winning_square".to_string(),
6002 result_type: "Option<u64>".to_string(),
6003 expression: ComputedExpr::MethodCall {
6004 expr: Box::new(ComputedExpr::FieldRef {
6005 path: "results.pre_reveal_rng".to_string(),
6006 }),
6007 method: "map".to_string(),
6008 args: vec![ComputedExpr::Closure {
6009 param: "rng".to_string(),
6010 body: Box::new(ComputedExpr::Binary {
6011 op: BinaryOp::Mod,
6012 left: Box::new(ComputedExpr::Var {
6013 name: "rng".to_string(),
6014 }),
6015 right: Box::new(ComputedExpr::Literal { value: json!(25) }),
6016 }),
6017 }],
6018 },
6019 },
6020 ]
6021 }
6022
6023 let vm = VmContext::new();
6024 for (rng, candidate, expected_rng, expected_square) in [
6025 (Value::Null, json!(42_u64), json!(42_u64), json!(17_u64)),
6026 (json!(7_u64), json!(42_u64), json!(7_u64), json!(7_u64)),
6027 ] {
6028 let mut state = json!({
6029 "results": {
6030 "rng": rng,
6031 "pre_reveal_rng_candidate": candidate,
6032 }
6033 });
6034
6035 vm.evaluate_computed_fields_from_ast(&mut state, &specs())
6036 .unwrap();
6037
6038 assert_eq!(state["results"]["pre_reveal_rng"], expected_rng);
6039 assert_eq!(
6040 state["results"]["pre_reveal_winning_square"],
6041 expected_square
6042 );
6043 }
6044 }
6045
6046 #[test]
6047 fn test_set_field_sum_preserves_integer_type() {
6048 let mut vm = VmContext::new();
6049 vm.registers[0] = serde_json::json!({});
6050 vm.registers[1] = serde_json::json!(20000000000_i64);
6051 vm.registers[2] = serde_json::json!(17951316474_i64);
6052
6053 vm.set_field_sum(0, "trading.total_buy_volume", 1).unwrap();
6054 vm.set_field_sum(0, "trading.total_sell_volume", 2).unwrap();
6055
6056 let state = &vm.registers[0];
6057 let buy_vol = state
6058 .get("trading")
6059 .and_then(|t| t.get("total_buy_volume"))
6060 .unwrap();
6061 let sell_vol = state
6062 .get("trading")
6063 .and_then(|t| t.get("total_sell_volume"))
6064 .unwrap();
6065
6066 let buy_serialized = serde_json::to_string(buy_vol).unwrap();
6067 let sell_serialized = serde_json::to_string(sell_vol).unwrap();
6068
6069 assert!(
6070 !buy_serialized.contains('.'),
6071 "Buy volume should not have decimal: {}",
6072 buy_serialized
6073 );
6074 assert!(
6075 !sell_serialized.contains('.'),
6076 "Sell volume should not have decimal: {}",
6077 sell_serialized
6078 );
6079 }
6080
6081 #[test]
6082 fn test_lookup_index_chaining() {
6083 let mut vm = VmContext::new();
6084
6085 let state = vm.states.get_mut(&0).unwrap();
6086
6087 state
6088 .pda_reverse_lookups
6089 .entry("default_pda_lookup".to_string())
6090 .or_insert_with(|| PdaReverseLookup::new(1000))
6091 .insert("pda_123".to_string(), "addr_456".to_string());
6092
6093 state
6094 .lookup_indexes
6095 .entry("round_address_lookup_index".to_string())
6096 .or_insert_with(LookupIndex::new)
6097 .insert(json!("addr_456"), json!(789));
6098
6099 let handler = vec![
6100 OpCode::LoadConstant {
6101 value: json!("pda_123"),
6102 dest: 0,
6103 },
6104 OpCode::LookupIndex {
6105 state_id: 0,
6106 index_name: "round_address_lookup_index".to_string(),
6107 lookup_value: 0,
6108 dest: 1,
6109 },
6110 ];
6111
6112 vm.execute_handler(&handler, &json!({}), "test", 0, "TestEntity", None, None)
6113 .unwrap();
6114
6115 assert_eq!(vm.registers[1], json!(789));
6116 }
6117
6118 #[test]
6119 fn test_lookup_index_no_chain() {
6120 let mut vm = VmContext::new();
6121
6122 let state = vm.states.get_mut(&0).unwrap();
6123 state
6124 .lookup_indexes
6125 .entry("test_index".to_string())
6126 .or_insert_with(LookupIndex::new)
6127 .insert(json!("key_abc"), json!(42));
6128
6129 let handler = vec![
6130 OpCode::LoadConstant {
6131 value: json!("key_abc"),
6132 dest: 0,
6133 },
6134 OpCode::LookupIndex {
6135 state_id: 0,
6136 index_name: "test_index".to_string(),
6137 lookup_value: 0,
6138 dest: 1,
6139 },
6140 ];
6141
6142 vm.execute_handler(&handler, &json!({}), "test", 0, "TestEntity", None, None)
6143 .unwrap();
6144
6145 assert_eq!(vm.registers[1], json!(42));
6146 }
6147
6148 #[test]
6149 fn test_conditional_set_field_with_zero_array() {
6150 let mut vm = VmContext::new();
6151
6152 let event_zeros = json!({
6153 "value": [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
6154 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
6155 });
6156
6157 let event_nonzero = json!({
6158 "value": [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16,
6159 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32]
6160 });
6161
6162 let zero_32: Value = json!([
6163 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
6164 0, 0, 0
6165 ]);
6166
6167 let handler = vec![
6168 OpCode::CreateObject { dest: 2 },
6169 OpCode::LoadEventField {
6170 path: FieldPath::new(&["value"]),
6171 dest: 10,
6172 default: None,
6173 },
6174 OpCode::ConditionalSetField {
6175 object: 2,
6176 path: "captured_value".to_string(),
6177 value: 10,
6178 condition_field: FieldPath::new(&["value"]),
6179 condition_op: ComparisonOp::NotEqual,
6180 condition_value: zero_32,
6181 },
6182 ];
6183
6184 vm.execute_handler(&handler, &event_zeros, "test", 0, "Test", None, None)
6185 .unwrap();
6186 assert!(
6187 vm.registers[2].get("captured_value").is_none(),
6188 "Field should not be set when value is all zeros"
6189 );
6190
6191 vm.reset_registers();
6192 vm.execute_handler(&handler, &event_nonzero, "test", 0, "Test", None, None)
6193 .unwrap();
6194 assert!(
6195 vm.registers[2].get("captured_value").is_some(),
6196 "Field should be set when value is non-zero"
6197 );
6198 }
6199
6200 #[test]
6201 fn test_when_instruction_arrives_first() {
6202 let mut vm = VmContext::new();
6203
6204 let signature = "test_sig_123".to_string();
6205
6206 {
6207 let state = vm.states.get(&0).unwrap();
6208 let mut cache = state.recent_tx_instructions.lock().unwrap();
6209 let mut set = HashSet::new();
6210 set.insert("RevealIxState".to_string());
6211 cache.put(signature.clone(), set);
6212 }
6213
6214 vm.current_context = Some(UpdateContext::new(100, signature.clone()));
6215
6216 let handler = vec![
6217 OpCode::CreateObject { dest: 2 },
6218 OpCode::LoadConstant {
6219 value: json!("primary_key_value"),
6220 dest: 1,
6221 },
6222 OpCode::LoadConstant {
6223 value: json!("the_revealed_value"),
6224 dest: 10,
6225 },
6226 OpCode::SetFieldWhen {
6227 object: 2,
6228 path: "entropy_value".to_string(),
6229 value: 10,
6230 when_instruction: "RevealIxState".to_string(),
6231 entity_name: "TestEntity".to_string(),
6232 key_reg: 1,
6233 condition_field: None,
6234 condition_op: None,
6235 condition_value: None,
6236 },
6237 ];
6238
6239 vm.execute_handler(
6240 &handler,
6241 &json!({}),
6242 "VarState",
6243 0,
6244 "TestEntity",
6245 None,
6246 None,
6247 )
6248 .unwrap();
6249
6250 assert_eq!(
6251 vm.registers[2].get("entropy_value").unwrap(),
6252 "the_revealed_value",
6253 "Field should be set when instruction was already seen"
6254 );
6255 }
6256
6257 #[test]
6258 fn test_when_account_arrives_first() {
6259 let mut vm = VmContext::new();
6260
6261 let signature = "test_sig_456".to_string();
6262
6263 vm.current_context = Some(UpdateContext::new(100, signature.clone()));
6264
6265 let handler = vec![
6266 OpCode::CreateObject { dest: 2 },
6267 OpCode::LoadConstant {
6268 value: json!("pk_123"),
6269 dest: 1,
6270 },
6271 OpCode::LoadConstant {
6272 value: json!("deferred_value"),
6273 dest: 10,
6274 },
6275 OpCode::SetFieldWhen {
6276 object: 2,
6277 path: "entropy_value".to_string(),
6278 value: 10,
6279 when_instruction: "RevealIxState".to_string(),
6280 entity_name: "TestEntity".to_string(),
6281 key_reg: 1,
6282 condition_field: None,
6283 condition_op: None,
6284 condition_value: None,
6285 },
6286 ];
6287
6288 vm.execute_handler(
6289 &handler,
6290 &json!({}),
6291 "VarState",
6292 0,
6293 "TestEntity",
6294 None,
6295 None,
6296 )
6297 .unwrap();
6298
6299 assert!(
6300 vm.registers[2].get("entropy_value").is_none(),
6301 "Field should not be set when instruction hasn't been seen"
6302 );
6303
6304 let state = vm.states.get(&0).unwrap();
6305 let key = (signature.clone(), "RevealIxState".to_string());
6306 assert!(
6307 state.deferred_when_ops.contains_key(&key),
6308 "Operation should be queued"
6309 );
6310
6311 {
6312 let mut cache = state.recent_tx_instructions.lock().unwrap();
6313 let mut set = HashSet::new();
6314 set.insert("RevealIxState".to_string());
6315 cache.put(signature.clone(), set);
6316 }
6317
6318 let deferred = state.deferred_when_ops.remove(&key).unwrap().1;
6319 for op in deferred {
6320 vm.apply_deferred_when_op(0, &op, None, None).unwrap();
6321 }
6322
6323 let state = vm.states.get(&0).unwrap();
6324 let entity = state.data.get(&json!("pk_123")).unwrap();
6325 assert_eq!(
6326 entity.get("entropy_value").unwrap(),
6327 "deferred_value",
6328 "Field should be set after instruction arrives"
6329 );
6330 }
6331
6332 #[test]
6333 fn test_when_cleanup_expired() {
6334 let mut vm = VmContext::new();
6335
6336 let state = vm.states.get(&0).unwrap();
6337 let key = ("old_sig".to_string(), "SomeIxState".to_string());
6338 state.deferred_when_ops.insert(
6339 key,
6340 vec![DeferredWhenOperation {
6341 entity_name: "Test".to_string(),
6342 primary_key: json!("pk"),
6343 field_path: "field".to_string(),
6344 field_value: json!("value"),
6345 when_instruction: "SomeIxState".to_string(),
6346 signature: "old_sig".to_string(),
6347 slot: 0,
6348 deferred_at: 0,
6349 emit: true,
6350 }],
6351 );
6352
6353 let removed = vm.cleanup_expired_when_ops(0, 60);
6354
6355 assert_eq!(removed, 1, "Should have removed 1 expired op");
6356 assert!(
6357 vm.states.get(&0).unwrap().deferred_when_ops.is_empty(),
6358 "Deferred ops should be empty after cleanup"
6359 );
6360 }
6361
6362 #[test]
6363 fn test_deferred_when_op_recomputes_dependent_fields() {
6364 use crate::ast::{BinaryOp, ComputedExpr, ComputedFieldSpec};
6365
6366 let mut vm = VmContext::new();
6367
6368 let computed_specs = vec![
6372 ComputedFieldSpec {
6373 target_path: "results.pre_reveal_rng".to_string(),
6374 result_type: "Option<u64>".to_string(),
6375 expression: ComputedExpr::FieldRef {
6376 path: "entropy.base_value".to_string(),
6377 },
6378 },
6379 ComputedFieldSpec {
6380 target_path: "results.pre_reveal_winning_square".to_string(),
6381 result_type: "Option<u64>".to_string(),
6382 expression: ComputedExpr::MethodCall {
6383 expr: Box::new(ComputedExpr::FieldRef {
6384 path: "results.pre_reveal_rng".to_string(),
6385 }),
6386 method: "map".to_string(),
6387 args: vec![ComputedExpr::Closure {
6388 param: "r".to_string(),
6389 body: Box::new(ComputedExpr::Binary {
6390 op: BinaryOp::Mod,
6391 left: Box::new(ComputedExpr::Var {
6392 name: "r".to_string(),
6393 }),
6394 right: Box::new(ComputedExpr::Literal {
6395 value: serde_json::json!(25),
6396 }),
6397 }),
6398 }],
6399 },
6400 },
6401 ];
6402
6403 let evaluator: Box<
6404 dyn Fn(&mut Value, Option<u64>, i64) -> ComputedEvaluatorResult + Send + Sync,
6405 > = Box::new(VmContext::create_evaluator_from_specs(computed_specs));
6406
6407 let primary_key = json!("test_pk");
6410 let op = DeferredWhenOperation {
6411 entity_name: "TestEntity".to_string(),
6412 primary_key: primary_key.clone(),
6413 field_path: "entropy.base_value".to_string(),
6414 field_value: json!(100),
6415 when_instruction: "TestIxState".to_string(),
6416 signature: "test_sig".to_string(),
6417 slot: 100,
6418 deferred_at: 0,
6419 emit: true,
6420 };
6421
6422 let initial_state = json!({
6424 "results": {}
6425 });
6426 vm.states
6427 .get(&0)
6428 .unwrap()
6429 .insert_with_eviction(primary_key.clone(), initial_state);
6430
6431 let mutations = vm
6433 .apply_deferred_when_op(
6434 0,
6435 &op,
6436 Some(&evaluator),
6437 Some(&[
6438 "results.pre_reveal_rng".to_string(),
6439 "results.pre_reveal_winning_square".to_string(),
6440 ]),
6441 )
6442 .unwrap();
6443
6444 let state = vm.states.get(&0).unwrap();
6446 let entity = state.data.get(&primary_key).unwrap();
6447
6448 println!(
6449 "Entity state: {}",
6450 serde_json::to_string_pretty(&*entity).unwrap()
6451 );
6452
6453 assert_eq!(
6455 entity.get("entropy").and_then(|e| e.get("base_value")),
6456 Some(&json!(100)),
6457 "Base value should be set"
6458 );
6459
6460 let pre_reveal_rng = entity
6462 .get("results")
6463 .and_then(|r| r.get("pre_reveal_rng"))
6464 .cloned();
6465 let pre_reveal_winning_square = entity
6466 .get("results")
6467 .and_then(|r| r.get("pre_reveal_winning_square"))
6468 .cloned();
6469
6470 assert_eq!(
6471 pre_reveal_rng,
6472 Some(json!(100)),
6473 "pre_reveal_rng should be computed"
6474 );
6475 assert_eq!(
6476 pre_reveal_winning_square,
6477 Some(json!(0)),
6478 "pre_reveal_winning_square should be 100 % 25 = 0"
6479 );
6480
6481 assert!(!mutations.is_empty(), "Should have mutations");
6483 let mutation = &mutations[0];
6484 let patch = &mutation.patch;
6485
6486 assert!(
6487 patch
6488 .get("results")
6489 .and_then(|r| r.get("pre_reveal_rng"))
6490 .is_some(),
6491 "Mutation should include pre_reveal_rng"
6492 );
6493 assert!(
6494 patch
6495 .get("results")
6496 .and_then(|r| r.get("pre_reveal_winning_square"))
6497 .is_some(),
6498 "Mutation should include pre_reveal_winning_square"
6499 );
6500 }
6501}