sui_core/
transaction_outputs.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use mysten_common::debug_fatal;
5use parking_lot::Mutex;
6use std::collections::{BTreeMap, HashSet};
7use std::sync::Arc;
8use sui_types::accumulator_event::AccumulatorEvent;
9use sui_types::base_types::{FullObjectID, ObjectRef};
10use sui_types::effects::{TransactionEffects, TransactionEffectsAPI, TransactionEvents};
11use sui_types::full_checkpoint_content::ObjectSet;
12use sui_types::inner_temporary_store::{InnerTemporaryStore, WrittenObjects};
13use sui_types::storage::{FullObjectKey, MarkerValue, ObjectKey};
14use sui_types::transaction::{TransactionData, TransactionDataAPI, VerifiedTransaction};
15
16/// TransactionOutputs
17#[derive(Debug)]
18pub struct TransactionOutputs {
19    pub transaction: Arc<VerifiedTransaction>,
20    pub effects: TransactionEffects,
21    pub events: TransactionEvents,
22    pub unchanged_loaded_runtime_objects: Vec<ObjectKey>,
23    pub accumulator_events: Mutex<Option<Vec<AccumulatorEvent>>>,
24
25    pub markers: Vec<(FullObjectKey, MarkerValue)>,
26    pub wrapped: Vec<ObjectKey>,
27    pub deleted: Vec<ObjectKey>,
28    pub locks_to_delete: Vec<ObjectRef>,
29    pub new_locks_to_init: Vec<ObjectRef>,
30    pub written: WrittenObjects,
31}
32
33impl TransactionOutputs {
34    // Convert InnerTemporaryStore + Effects into the exact set of updates to the store
35    pub fn build_transaction_outputs(
36        transaction: VerifiedTransaction,
37        effects: TransactionEffects,
38        inner_temporary_store: InnerTemporaryStore,
39        unchanged_loaded_runtime_objects: Vec<ObjectKey>,
40    ) -> TransactionOutputs {
41        let InnerTemporaryStore {
42            input_objects,
43            stream_ended_consensus_objects,
44            mutable_inputs,
45            written,
46            events,
47            accumulator_events,
48            loaded_runtime_objects: _,
49            binary_config: _,
50            runtime_packages_loaded_from_db: _,
51            lamport_version,
52            accumulator_running_max_withdraws: _,
53            retry_request,
54        } = inner_temporary_store;
55
56        // A transaction that requested a retry is never committed: the authority discards its
57        // effects and re-enqueues it, so it must not reach the commit path.
58        if let Some(retry_request) = &retry_request {
59            debug_fatal!("a transaction requesting a retry must not be committed: {retry_request}");
60        }
61
62        let tx_digest = *transaction.digest();
63
64        // Get the actual set of objects that have been received -- any received
65        // object will show up in the modified-at set.
66        let modified_at: HashSet<_> = effects.modified_at_versions().into_iter().collect();
67        let possible_to_receive = transaction.transaction_data().receiving_objects();
68        let received_objects = possible_to_receive
69            .into_iter()
70            .filter(|obj_ref| modified_at.contains(&(obj_ref.0, obj_ref.1)));
71
72        // We record any received or deleted objects since they could be pruned, and smear object
73        // removals from consensus in the marker table. For deleted entries in the marker table we
74        // need to make sure we don't accidentally overwrite entries.
75        let markers: Vec<_> = {
76            let received = received_objects.clone().map(|objref| {
77                (
78                    // TODO: Add support for receiving consensus objects. For now this assumes fastpath.
79                    FullObjectKey::new(FullObjectID::new(objref.0, None), objref.1),
80                    MarkerValue::Received,
81                )
82            });
83
84            let tombstones = effects
85                .all_tombstones()
86                .into_iter()
87                .map(|(object_id, version)| {
88                    let consensus_key = input_objects
89                        .get(&object_id)
90                        .filter(|o| o.is_consensus())
91                        .map(|o| FullObjectKey::new(o.full_id(), version));
92                    if let Some(consensus_key) = consensus_key {
93                        (consensus_key, MarkerValue::ConsensusStreamEnded(tx_digest))
94                    } else {
95                        (
96                            FullObjectKey::new(FullObjectID::new(object_id, None), version),
97                            MarkerValue::FastpathStreamEnded,
98                        )
99                    }
100                });
101
102            let fastpath_stream_ended =
103                effects
104                    .transferred_to_consensus()
105                    .into_iter()
106                    .map(|(object_id, version, _)| {
107                        // Note: it's a bit of a misnomer to mark an object as `FastpathStreamEnded`
108                        // when it could have been transferred to consensus from `ObjectOwner`, as
109                        // its root owner may not have been a fastpath object. However, whether or
110                        // not it was technically in the fastpath at the version the marker is
111                        // written, it certainly is not in the fastpath *anymore*. This is needed
112                        // to produce the required behavior in `ObjectCacheRead::multi_input_objects_available`
113                        // when checking whether receiving objects are available.
114                        (
115                            FullObjectKey::new(FullObjectID::new(object_id, None), version),
116                            MarkerValue::FastpathStreamEnded,
117                        )
118                    });
119
120            let consensus_stream_ended = effects
121                .transferred_from_consensus()
122                .into_iter()
123                .chain(effects.consensus_owner_changed())
124                .map(|(object_id, version, _)| {
125                    let object = input_objects
126                        .get(&object_id)
127                        .expect("stream-ended object must be in input_objects");
128                    (
129                        FullObjectKey::new(object.full_id(), version),
130                        MarkerValue::ConsensusStreamEnded(tx_digest),
131                    )
132                });
133
134            // We "smear" removed consensus objects in the marker table to allow for proper
135            // sequencing of transactions that are submitted after the consensus stream ends.
136            // This means writing duplicate copies of the `ConsensusStreamEnded` marker for
137            // every output version that was scheduled to be created.
138            // NB: that we do _not_ smear objects that were taken immutably in the transaction
139            // (because these are not assigned output versions).
140            let smeared_objects = effects.stream_ended_mutably_accessed_consensus_objects();
141            let consensus_smears = smeared_objects.into_iter().map(|object_id| {
142                let id = input_objects
143                    .get(&object_id)
144                    .map(|obj| obj.full_id())
145                    .unwrap_or_else(|| {
146                        let start_version = stream_ended_consensus_objects.get(&object_id)
147                            .expect("stream-ended object must be in either input_objects or stream_ended_consensus_objects");
148                        FullObjectID::new(object_id, Some(*start_version))
149                    });
150                (
151                    FullObjectKey::new(id, lamport_version),
152                    MarkerValue::ConsensusStreamEnded(tx_digest),
153                )
154            });
155
156            received
157                .chain(tombstones)
158                .chain(fastpath_stream_ended)
159                .chain(consensus_stream_ended)
160                .chain(consensus_smears)
161                .collect()
162        };
163
164        let locks_to_delete: Vec<_> = mutable_inputs
165            .into_iter()
166            .filter_map(|(id, ((version, digest), owner))| {
167                owner.is_address_owned().then_some((id, version, digest))
168            })
169            .chain(received_objects)
170            .collect();
171
172        let new_locks_to_init: Vec<_> = written
173            .values()
174            .filter_map(|new_object| {
175                if new_object.is_address_owned() {
176                    Some(new_object.compute_object_reference())
177                } else {
178                    None
179                }
180            })
181            .collect();
182
183        let deleted = effects
184            .deleted()
185            .into_iter()
186            .chain(effects.unwrapped_then_deleted())
187            .map(ObjectKey::from)
188            .collect();
189
190        let wrapped = effects.wrapped().into_iter().map(ObjectKey::from).collect();
191
192        TransactionOutputs {
193            transaction: Arc::new(transaction),
194            effects,
195            events,
196            unchanged_loaded_runtime_objects,
197            accumulator_events: Mutex::new(Some(accumulator_events)),
198            markers,
199            wrapped,
200            deleted,
201            locks_to_delete,
202            new_locks_to_init,
203            written,
204        }
205    }
206
207    pub fn take_accumulator_events(&self) -> Vec<AccumulatorEvent> {
208        self.accumulator_events
209            .lock()
210            .take()
211            .expect("take_accumulator_events called twice")
212    }
213
214    #[cfg(test)]
215    pub fn new_for_testing(transaction: VerifiedTransaction, effects: TransactionEffects) -> Self {
216        Self {
217            transaction: Arc::new(transaction),
218            effects,
219            events: TransactionEvents { data: vec![] },
220            unchanged_loaded_runtime_objects: vec![],
221            accumulator_events: Mutex::new(Some(vec![])),
222            markers: vec![],
223            wrapped: vec![],
224            deleted: vec![],
225            locks_to_delete: vec![],
226            new_locks_to_init: vec![],
227            written: WrittenObjects::new(),
228        }
229    }
230}
231
232pub fn unchanged_loaded_runtime_objects(
233    _transaction: &TransactionData,
234    effects: &TransactionEffects,
235    loaded_runtime_objects: &ObjectSet,
236) -> Vec<ObjectKey> {
237    let mut unchanged_loaded_runtime_objects: BTreeMap<_, _> = loaded_runtime_objects
238        .iter()
239        // Don't include loaded packages (which are used for doing UID tracking inside the VM)
240        .filter(|o| !o.is_package())
241        .map(|o| (o.id(), o.version()))
242        .collect();
243
244    // Remove any object that is referenced in the changed objects effects set since it would be
245    // redundant to include it again.
246    for change in effects.object_changes() {
247        unchanged_loaded_runtime_objects.remove(&change.id);
248    }
249
250    unchanged_loaded_runtime_objects
251        .into_iter()
252        .map(|(id, v)| ObjectKey(id, v))
253        .collect()
254}
sui_core/transaction_outputs.rs

sui_core/
transaction_outputs.rs
