arete_server/
mutation_batch.rs

1//! MutationBatch - Envelope type for propagating trace context across async boundaries.
2
3use arete_interpreter::Mutation;
4use serde::{Deserialize, Serialize};
5use smallvec::SmallVec;
6use tracing::Span;
7
8/// The producer-local meaning of [`SlotContext::slot_index`].
9///
10/// Values from different domains are deliberately incomparable. In
11/// particular, an async resolver counter must never make a later instruction
12/// transaction index look stale within the same slot.
13#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
14#[serde(rename_all = "camelCase")]
15pub enum SlotIndexDomain {
16    /// Backward-compatible domain for callers of [`SlotContext::new`].
17    #[default]
18    Legacy,
19    Account,
20    Instruction,
21    Resolver,
22}
23
24/// Slot context for ordering mutations by blockchain position.
25/// Used to derive `_seq` field for default recency sorting.
26#[derive(Debug, Clone, Copy, Default)]
27pub struct SlotContext {
28    /// Solana slot number
29    pub slot: u64,
30    /// Index within the slot (write_version for accounts, txn_index for instructions)
31    pub slot_index: u64,
32    /// The producer-local domain in which `slot_index` is ordered.
33    pub slot_index_domain: SlotIndexDomain,
34}
35
36impl SlotContext {
37    pub fn with_domain(slot: u64, slot_index: u64, slot_index_domain: SlotIndexDomain) -> Self {
38        Self {
39            slot,
40            slot_index,
41            slot_index_domain,
42        }
43    }
44
45    /// Construct a context in the legacy generic domain.
46    ///
47    /// Generated ingestion uses the explicit constructors below so account,
48    /// instruction and resolver offsets are never compared accidentally.
49    pub fn new(slot: u64, slot_index: u64) -> Self {
50        Self::with_domain(slot, slot_index, SlotIndexDomain::Legacy)
51    }
52
53    pub fn account(slot: u64, write_version: u64) -> Self {
54        Self::with_domain(slot, write_version, SlotIndexDomain::Account)
55    }
56
57    pub fn instruction(slot: u64, txn_index: u64) -> Self {
58        Self::with_domain(slot, txn_index, SlotIndexDomain::Instruction)
59    }
60
61    pub fn resolver(slot: u64, resolver_index: u64) -> Self {
62        Self::with_domain(slot, resolver_index, SlotIndexDomain::Resolver)
63    }
64
65    /// Compute a monotonic sequence number for sorting.
66    /// Encodes as string to preserve precision in JSON: "{slot}:{slot_index:012}"
67    /// This gives lexicographic ordering that matches (slot, slot_index) tuple ordering.
68    pub fn to_seq_string(&self) -> String {
69        format!("{}:{:012}", self.slot, self.slot_index)
70    }
71}
72
73/// Envelope type that carries mutations along with their originating span context.
74///
75/// This enables trace context propagation across the mpsc channel boundary
76/// from the Vixen parser to the Projector.
77#[derive(Debug)]
78pub struct MutationBatch {
79    /// The span from which these mutations originated
80    pub span: Span,
81    /// The mutations to process
82    pub mutations: SmallVec<[Mutation; 6]>,
83    /// Slot context for ordering (optional for backward compatibility)
84    pub slot_context: Option<SlotContext>,
85    /// Event metadata for logging and diagnostics
86    pub event_context: Option<EventContext>,
87    /// When set, this batch is a flush marker: the projector acknowledges it
88    /// after every batch queued before it has been applied to the caches.
89    /// Used by the snapshot manager to establish a consistency cut.
90    pub flush_ack: Option<tokio::sync::oneshot::Sender<()>>,
91    /// Keeps snapshot capture blocked from the VM update that produced this
92    /// batch until the projector has applied it.
93    pub(crate) snapshot_guard: Option<crate::snapshot::SnapshotProcessingGuard>,
94}
95
96#[derive(Debug, Clone)]
97pub struct EventContext {
98    pub program: String,
99    pub event_kind: String,
100    pub event_type: String,
101    pub account: Option<String>,
102    pub accounts_count: Option<usize>,
103}
104
105impl MutationBatch {
106    pub fn new(mutations: SmallVec<[Mutation; 6]>) -> Self {
107        Self {
108            span: Span::current(),
109            mutations,
110            slot_context: None,
111            event_context: None,
112            flush_ack: None,
113            snapshot_guard: None,
114        }
115    }
116
117    pub fn with_span(span: Span, mutations: SmallVec<[Mutation; 6]>) -> Self {
118        Self {
119            span,
120            mutations,
121            slot_context: None,
122            event_context: None,
123            flush_ack: None,
124            snapshot_guard: None,
125        }
126    }
127
128    pub fn with_slot_context(
129        mutations: SmallVec<[Mutation; 6]>,
130        slot_context: SlotContext,
131    ) -> Self {
132        Self {
133            span: Span::current(),
134            mutations,
135            slot_context: Some(slot_context),
136            event_context: None,
137            flush_ack: None,
138            snapshot_guard: None,
139        }
140    }
141
142    /// An empty batch whose only purpose is to be acknowledged once the
143    /// projector has drained everything queued before it.
144    pub fn flush_marker(ack: tokio::sync::oneshot::Sender<()>) -> Self {
145        Self {
146            span: Span::current(),
147            mutations: SmallVec::new(),
148            slot_context: None,
149            event_context: None,
150            flush_ack: Some(ack),
151            snapshot_guard: None,
152        }
153    }
154
155    /// Transfer a VM processing guard to this batch. The projector retains it
156    /// until cache application and watermark advancement are complete.
157    pub fn with_snapshot_guard(mut self, guard: crate::snapshot::SnapshotProcessingGuard) -> Self {
158        self.snapshot_guard = Some(guard);
159        self
160    }
161
162    pub fn with_event_context(mut self, event_context: EventContext) -> Self {
163        self.event_context = Some(event_context);
164        self
165    }
166
167    pub fn len(&self) -> usize {
168        self.mutations.len()
169    }
170
171    pub fn is_empty(&self) -> bool {
172        self.mutations.is_empty()
173    }
174}
arete_server/mutation_batch.rs

arete_server/
mutation_batch.rs
