arete_server/
mutation_batch.rs

1//! MutationBatch - Envelope type for propagating trace context across async boundaries.
2
3use arete_interpreter::Mutation;
4use smallvec::SmallVec;
5use tracing::Span;
6
7/// Slot context for ordering mutations by blockchain position.
8/// Used to derive `_seq` field for default recency sorting.
9#[derive(Debug, Clone, Copy, Default)]
10pub struct SlotContext {
11    /// Solana slot number
12    pub slot: u64,
13    /// Index within the slot (write_version for accounts, txn_index for instructions)
14    pub slot_index: u64,
15}
16
17impl SlotContext {
18    pub fn new(slot: u64, slot_index: u64) -> Self {
19        Self { slot, slot_index }
20    }
21
22    /// Compute a monotonic sequence number for sorting.
23    /// Encodes as string to preserve precision in JSON: "{slot}:{slot_index:012}"
24    /// This gives lexicographic ordering that matches (slot, slot_index) tuple ordering.
25    pub fn to_seq_string(&self) -> String {
26        format!("{}:{:012}", self.slot, self.slot_index)
27    }
28}
29
30/// Envelope type that carries mutations along with their originating span context.
31///
32/// This enables trace context propagation across the mpsc channel boundary
33/// from the Vixen parser to the Projector.
34#[derive(Debug)]
35pub struct MutationBatch {
36    /// The span from which these mutations originated
37    pub span: Span,
38    /// The mutations to process
39    pub mutations: SmallVec<[Mutation; 6]>,
40    /// Slot context for ordering (optional for backward compatibility)
41    pub slot_context: Option<SlotContext>,
42    /// Event metadata for logging and diagnostics
43    pub event_context: Option<EventContext>,
44    /// When set, this batch is a flush marker: the projector acknowledges it
45    /// after every batch queued before it has been applied to the caches.
46    /// Used by the snapshot manager to establish a consistency cut.
47    pub flush_ack: Option<tokio::sync::oneshot::Sender<()>>,
48    /// Keeps snapshot capture blocked from the VM update that produced this
49    /// batch until the projector has applied it.
50    pub(crate) snapshot_guard: Option<crate::snapshot::SnapshotProcessingGuard>,
51}
52
53#[derive(Debug, Clone)]
54pub struct EventContext {
55    pub program: String,
56    pub event_kind: String,
57    pub event_type: String,
58    pub account: Option<String>,
59    pub accounts_count: Option<usize>,
60}
61
62impl MutationBatch {
63    pub fn new(mutations: SmallVec<[Mutation; 6]>) -> Self {
64        Self {
65            span: Span::current(),
66            mutations,
67            slot_context: None,
68            event_context: None,
69            flush_ack: None,
70            snapshot_guard: None,
71        }
72    }
73
74    pub fn with_span(span: Span, mutations: SmallVec<[Mutation; 6]>) -> Self {
75        Self {
76            span,
77            mutations,
78            slot_context: None,
79            event_context: None,
80            flush_ack: None,
81            snapshot_guard: None,
82        }
83    }
84
85    pub fn with_slot_context(
86        mutations: SmallVec<[Mutation; 6]>,
87        slot_context: SlotContext,
88    ) -> Self {
89        Self {
90            span: Span::current(),
91            mutations,
92            slot_context: Some(slot_context),
93            event_context: None,
94            flush_ack: None,
95            snapshot_guard: None,
96        }
97    }
98
99    /// An empty batch whose only purpose is to be acknowledged once the
100    /// projector has drained everything queued before it.
101    pub fn flush_marker(ack: tokio::sync::oneshot::Sender<()>) -> Self {
102        Self {
103            span: Span::current(),
104            mutations: SmallVec::new(),
105            slot_context: None,
106            event_context: None,
107            flush_ack: Some(ack),
108            snapshot_guard: None,
109        }
110    }
111
112    /// Transfer a VM processing guard to this batch. The projector retains it
113    /// until cache application and watermark advancement are complete.
114    pub fn with_snapshot_guard(mut self, guard: crate::snapshot::SnapshotProcessingGuard) -> Self {
115        self.snapshot_guard = Some(guard);
116        self
117    }
118
119    pub fn with_event_context(mut self, event_context: EventContext) -> Self {
120        self.event_context = Some(event_context);
121        self
122    }
123
124    pub fn len(&self) -> usize {
125        self.mutations.len()
126    }
127
128    pub fn is_empty(&self) -> bool {
129        self.mutations.is_empty()
130    }
131}
arete_server/mutation_batch.rs

arete_server/
mutation_batch.rs
