linera_execution/
transaction_tracker.rs

1// Copyright (c) Zefchain Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{
5    collections::{BTreeMap, BTreeSet},
6    future::Future,
7    mem, vec,
8};
9
10use custom_debug_derive::Debug;
11use linera_base::{
12    crypto::CryptoHash,
13    data_types::{Blob, BlobContent, Cursor, Event, OracleResponse, StreamUpdate, Timestamp},
14    ensure,
15    identifiers::{ApplicationId, BlobId, ChainId, StreamId},
16};
17
18use crate::{ExecutionError, OutgoingMessage};
19
20/// Maps a (publishing chain, stream) to its `(previous_index, first_index, next_index)`.
21type AppStreamUpdates = BTreeMap<(ChainId, StreamId), (u32, u32, u32)>;
22
23/// Tracks oracle responses and execution outcomes of an ongoing transaction execution, as well
24/// as replayed oracle responses.
25#[derive(Debug, Default)]
26pub struct TransactionTracker {
27    #[debug(skip_if = Option::is_none)]
28    replaying_oracle_responses: Option<vec::IntoIter<OracleResponse>>,
29    #[debug(skip_if = Vec::is_empty)]
30    oracle_responses: Vec<OracleResponse>,
31    #[debug(skip_if = Vec::is_empty)]
32    outgoing_messages: Vec<OutgoingMessage>,
33    /// The current local time.
34    local_time: Timestamp,
35    /// The index of the current transaction in the block.
36    transaction_index: u32,
37    next_application_index: u32,
38    next_chain_index: u32,
39    /// Events recorded by contracts' `emit` calls.
40    events: Vec<Event>,
41    /// Blobs created by contracts.
42    ///
43    /// As of right now, blobs created by the contracts are one of the following types:
44    /// - [`Data`]
45    /// - [`ContractBytecode`]
46    /// - [`ServiceBytecode`]
47    /// - [`EvmBytecode`]
48    /// - [`ApplicationDescription`]
49    /// - [`ChainDescription`]
50    blobs: BTreeMap<BlobId, BlobContent>,
51    /// The blobs created in the previous transactions.
52    previously_created_blobs: BTreeMap<BlobId, BlobContent>,
53    /// Operation result.
54    operation_result: Option<Vec<u8>>,
55    /// Streams that have been updated but not yet processed during this transaction.
56    streams_to_process: BTreeMap<ApplicationId, AppStreamUpdates>,
57    /// Published blobs this transaction refers to by [`BlobId`].
58    blobs_published: BTreeSet<BlobId>,
59    /// Blob IDs created or published by free apps (fees waived).
60    free_blob_ids: BTreeSet<BlobId>,
61    /// Inputs computed pre-block by the checkpoint pre-hook, stashed here so that the
62    /// matching `SystemOperation::Checkpoint` operation handler can use them when it
63    /// runs. The state dump and the inbox snapshot must both be captured before any
64    /// block-level mutation taints the chain state, so we collect them up front and
65    /// hand them off through this tracker.
66    #[debug(skip_if = Option::is_none)]
67    prepared_checkpoint: Option<PreparedCheckpoint>,
68}
69
70/// Pre-block-computed inputs for a `SystemOperation::Checkpoint` transaction.
71#[derive(Clone, Debug)]
72pub struct PreparedCheckpoint {
73    /// The execution-state dump split into blobs at the current epoch's
74    /// `maximum_blob_size`.
75    pub blobs: Vec<Blob>,
76    /// For each chain we've received messages from since our last own checkpoint, the
77    /// position past the last bundle we've consumed. Used to emit a
78    /// `SystemMessage::CheckpointAck` to each origin so the origin can later trim its
79    /// outbox dump. This is the delta over the previous checkpoint, filtered by
80    /// `pending_checkpoint_ack_targets` to break the notification ping-pong.
81    pub origin_cursors: Vec<(ChainId, Cursor)>,
82    /// For *every* inbox with a non-default `next_cursor_to_remove`, the cursor itself.
83    /// A bootstrapping node uses these to seed each inbox's `restored_cursor`, so a
84    /// sender that hasn't seen the matching `CheckpointAck` yet and re-pushes an
85    /// already-consumed bundle is a silent no-op rather than a duplicate consumption.
86    pub inbox_cursors: Vec<(ChainId, Cursor)>,
87    /// Hashes of every block on this chain that the chain's outboxes still reference,
88    /// taken before the block runs. Included in the oracle response so the checkpoint
89    /// block's certificate transitively certifies those older blocks.
90    pub outbox_block_hashes: Vec<CryptoHash>,
91}
92
93/// The [`TransactionTracker`] contents after a transaction has finished.
94#[derive(Debug, Default)]
95pub struct TransactionOutcome {
96    /// The recorded oracle responses.
97    #[debug(skip_if = Vec::is_empty)]
98    pub oracle_responses: Vec<OracleResponse>,
99    /// The messages to be sent to other chains.
100    #[debug(skip_if = Vec::is_empty)]
101    pub outgoing_messages: Vec<OutgoingMessage>,
102    /// The index to be assigned to the next created application.
103    pub next_application_index: u32,
104    /// The index to be assigned to the next created chain.
105    pub next_chain_index: u32,
106    /// Events recorded by contracts' `emit` calls.
107    pub events: Vec<Event>,
108    /// Blobs created by contracts.
109    pub blobs: Vec<Blob>,
110    /// Operation result.
111    pub operation_result: Vec<u8>,
112    /// Blobs published by this transaction.
113    pub blobs_published: BTreeSet<BlobId>,
114    /// Blob IDs created or published by free apps (fees waived).
115    pub free_blob_ids: BTreeSet<BlobId>,
116}
117
118impl TransactionTracker {
119    /// Creates a new [`TransactionTracker`].
120    pub fn new(
121        local_time: Timestamp,
122        transaction_index: u32,
123        next_application_index: u32,
124        next_chain_index: u32,
125        oracle_responses: Option<Vec<OracleResponse>>,
126        blobs: &[Vec<Blob>],
127    ) -> Self {
128        let mut previously_created_blobs = BTreeMap::new();
129        for tx_blobs in blobs {
130            for blob in tx_blobs {
131                previously_created_blobs.insert(blob.id(), blob.content().clone());
132            }
133        }
134        TransactionTracker {
135            local_time,
136            transaction_index,
137            next_application_index,
138            next_chain_index,
139            replaying_oracle_responses: oracle_responses.map(Vec::into_iter),
140            previously_created_blobs,
141            ..Self::default()
142        }
143    }
144
145    /// Sets the blobs known to the tracker and returns the updated tracker.
146    pub fn with_blobs(mut self, blobs: BTreeMap<BlobId, BlobContent>) -> Self {
147        self.blobs = blobs;
148        self
149    }
150
151    /// Stashes pre-block-computed checkpoint inputs on the tracker. The matching
152    /// `SystemOperation::Checkpoint` operation handler will retrieve them via
153    /// [`Self::take_prepared_checkpoint`] when the operation runs.
154    pub fn set_prepared_checkpoint(&mut self, prepared: PreparedCheckpoint) {
155        self.prepared_checkpoint = Some(prepared);
156    }
157
158    /// Takes the pre-block-computed checkpoint inputs, if any were stashed.
159    pub fn take_prepared_checkpoint(&mut self) -> Option<PreparedCheckpoint> {
160        self.prepared_checkpoint.take()
161    }
162
163    /// Returns the local time recorded by the tracker.
164    pub fn local_time(&self) -> Timestamp {
165        self.local_time
166    }
167
168    /// Sets the local time recorded by the tracker.
169    pub fn set_local_time(&mut self, local_time: Timestamp) {
170        self.local_time = local_time;
171    }
172
173    /// Returns the index of the current transaction in the block.
174    pub fn transaction_index(&self) -> u32 {
175        self.transaction_index
176    }
177
178    /// Returns the index that would be assigned to the next created application, without consuming it.
179    pub fn peek_application_index(&self) -> u32 {
180        self.next_application_index
181    }
182
183    /// Returns the index to be assigned to the next created application and increments the counter.
184    pub fn next_application_index(&mut self) -> u32 {
185        let index = self.next_application_index;
186        self.next_application_index += 1;
187        index
188    }
189
190    /// Returns the index to be assigned to the next created chain and increments the counter.
191    pub fn next_chain_index(&mut self) -> u32 {
192        let index = self.next_chain_index;
193        self.next_chain_index += 1;
194        index
195    }
196
197    /// Records an outgoing message.
198    pub fn add_outgoing_message(&mut self, message: OutgoingMessage) {
199        self.outgoing_messages.push(message);
200    }
201
202    /// Records multiple outgoing messages.
203    pub fn add_outgoing_messages(&mut self, messages: impl IntoIterator<Item = OutgoingMessage>) {
204        for message in messages {
205            self.add_outgoing_message(message);
206        }
207    }
208
209    /// Records an event emitted on the given stream.
210    pub fn add_event(&mut self, stream_id: StreamId, index: u32, value: Vec<u8>) {
211        self.events.push(Event {
212            stream_id,
213            index,
214            value,
215        });
216    }
217
218    /// Returns the content of the blob with the given ID, if known to the tracker.
219    pub fn get_blob_content(&self, blob_id: &BlobId) -> Option<&BlobContent> {
220        if let Some(content) = self.blobs.get(blob_id) {
221            return Some(content);
222        }
223        self.previously_created_blobs.get(blob_id)
224    }
225
226    /// Records a blob created by this transaction.
227    pub fn add_created_blob(&mut self, blob: Blob) {
228        self.blobs.insert(blob.id(), blob.into_content());
229    }
230
231    /// Records a blob published by this transaction.
232    pub fn add_published_blob(&mut self, blob_id: BlobId) {
233        self.blobs_published.insert(blob_id);
234    }
235
236    /// Marks a blob as created/published by a free app, so its fees will be waived.
237    pub fn mark_blob_free(&mut self, blob_id: BlobId) {
238        self.free_blob_ids.insert(blob_id);
239    }
240
241    /// Returns the blobs created by this transaction.
242    pub fn created_blobs(&self) -> &BTreeMap<BlobId, BlobContent> {
243        &self.blobs
244    }
245
246    /// Records the result of the operation.
247    pub fn add_operation_result(&mut self, result: Option<Vec<u8>>) {
248        self.operation_result = result
249    }
250
251    /// In replay mode, returns the next recorded oracle response. Otherwise executes `f` and
252    /// records and returns the result. `f` is the implementation of the actual oracle and is
253    /// only called in validation mode, so it does not have to be fully deterministic.
254    pub async fn oracle<F, G>(&mut self, f: F) -> Result<&OracleResponse, ExecutionError>
255    where
256        F: FnOnce() -> G,
257        G: Future<Output = Result<OracleResponse, ExecutionError>>,
258    {
259        let response = match self.next_replayed_oracle_response()? {
260            Some(response) => response,
261            None => f().await?,
262        };
263        self.oracle_responses.push(response);
264        Ok(self.oracle_responses.last().unwrap())
265    }
266
267    /// Records that the given range of events on a stream must be processed by the application.
268    pub fn add_stream_to_process(
269        &mut self,
270        application_id: ApplicationId,
271        chain_id: ChainId,
272        stream_id: StreamId,
273        previous_index: u32,
274        first_index: u32,
275        next_index: u32,
276    ) {
277        if next_index == previous_index {
278            return; // No new events in the stream.
279        }
280        self.streams_to_process
281            .entry(application_id)
282            .or_default()
283            .entry((chain_id, stream_id))
284            .and_modify(|(pi, fi, ni)| {
285                *pi = (*pi).min(previous_index);
286                // The strongest floor wins: a later checkpoint prunes more.
287                *fi = (*fi).max(first_index);
288                *ni = (*ni).max(next_index);
289            })
290            .or_insert_with(|| (previous_index, first_index, next_index));
291    }
292
293    /// Removes a stream from the set of streams to be processed by the application.
294    pub fn remove_stream_to_process(
295        &mut self,
296        application_id: ApplicationId,
297        chain_id: ChainId,
298        stream_id: StreamId,
299    ) {
300        let Some(streams) = self.streams_to_process.get_mut(&application_id) else {
301            return;
302        };
303        if streams.remove(&(chain_id, stream_id)).is_some() && streams.is_empty() {
304            self.streams_to_process.remove(&application_id);
305        }
306    }
307
308    /// Takes the streams to be processed, grouped by application.
309    pub fn take_streams_to_process(&mut self) -> BTreeMap<ApplicationId, Vec<StreamUpdate>> {
310        mem::take(&mut self.streams_to_process)
311            .into_iter()
312            .map(|(app_id, streams)| {
313                let updates = streams
314                    .into_iter()
315                    .map(
316                        |((chain_id, stream_id), (previous_index, first_index, next_index))| {
317                            StreamUpdate {
318                                chain_id,
319                                stream_id,
320                                previous_index,
321                                first_index,
322                                next_index,
323                            }
324                        },
325                    )
326                    .collect();
327                (app_id, updates)
328            })
329            .collect()
330    }
331
332    /// Adds the oracle response to the record.
333    /// If replaying, it also checks that it matches the next replayed one and returns `true`.
334    pub fn replay_oracle_response(
335        &mut self,
336        oracle_response: OracleResponse,
337    ) -> Result<bool, ExecutionError> {
338        let replaying = if let Some(recorded_response) = self.next_replayed_oracle_response()? {
339            ensure!(
340                recorded_response == oracle_response,
341                ExecutionError::OracleResponseMismatch
342            );
343            true
344        } else {
345            false
346        };
347        self.oracle_responses.push(oracle_response);
348        Ok(replaying)
349    }
350
351    /// If in replay mode, returns the next oracle response, or an error if it is missing.
352    ///
353    /// If not in replay mode, `None` is returned, and the caller must execute the actual oracle
354    /// to obtain the value.
355    ///
356    /// In both cases, the value (returned or obtained from the oracle) must be recorded using
357    /// `add_oracle_response`.
358    fn next_replayed_oracle_response(&mut self) -> Result<Option<OracleResponse>, ExecutionError> {
359        let Some(responses) = &mut self.replaying_oracle_responses else {
360            return Ok(None); // Not in replay mode.
361        };
362        let response = responses
363            .next()
364            .ok_or_else(|| ExecutionError::MissingOracleResponse)?;
365        Ok(Some(response))
366    }
367
368    /// Consumes the tracker and returns the resulting [`TransactionOutcome`].
369    pub fn into_outcome(self) -> Result<TransactionOutcome, ExecutionError> {
370        let TransactionTracker {
371            replaying_oracle_responses,
372            oracle_responses,
373            outgoing_messages,
374            local_time: _,
375            transaction_index: _,
376            next_application_index,
377            next_chain_index,
378            events,
379            blobs,
380            previously_created_blobs: _,
381            operation_result,
382            streams_to_process,
383            blobs_published,
384            free_blob_ids,
385            prepared_checkpoint: _,
386        } = self;
387        ensure!(
388            streams_to_process.is_empty(),
389            ExecutionError::UnprocessedStreams
390        );
391        if let Some(mut responses) = replaying_oracle_responses {
392            ensure!(
393                responses.next().is_none(),
394                ExecutionError::UnexpectedOracleResponse
395            );
396        }
397        let blobs = blobs
398            .into_iter()
399            .map(|(blob_id, content)| Blob::new_with_hash_unchecked(blob_id, content))
400            .collect::<Vec<_>>();
401        Ok(TransactionOutcome {
402            outgoing_messages,
403            oracle_responses,
404            next_application_index,
405            next_chain_index,
406            events,
407            blobs,
408            operation_result: operation_result.unwrap_or_default(),
409            blobs_published,
410            free_blob_ids,
411        })
412    }
413}
414
415#[cfg(with_testing)]
416impl TransactionTracker {
417    /// Creates a new [`TransactionTracker`] for testing, with default values and the given
418    /// oracle responses.
419    pub fn new_replaying(oracle_responses: Vec<OracleResponse>) -> Self {
420        TransactionTracker::new(Timestamp::from(0), 0, 0, 0, Some(oracle_responses), &[])
421    }
422
423    /// Creates a new [`TransactionTracker`] for testing, with default values and oracle responses
424    /// for the given blobs.
425    pub fn new_replaying_blobs<T>(blob_ids: T) -> Self
426    where
427        T: IntoIterator,
428        T::Item: std::borrow::Borrow<BlobId>,
429    {
430        use std::borrow::Borrow;
431
432        let oracle_responses = blob_ids
433            .into_iter()
434            .map(|blob_id| OracleResponse::Blob(*blob_id.borrow()))
435            .collect();
436        TransactionTracker::new_replaying(oracle_responses)
437    }
438}
linera_execution/transaction_tracker.rs

linera_execution/
transaction_tracker.rs
