sui_core/
transaction_outputs.rs

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

sui_core/
transaction_outputs.rs
