Make sure unwrapped object shadow deleted object (MystenLabs#387)

fastpay_core/src/authority.rs

@@ -274,12 +274,13 @@ impl AuthorityState {
         let mut tx_ctx = TxContext::new(order.sender(), transaction_digest);
 
         let gas_object_id = *order.gas_payment_object_id();
-        let (temporary_store, status) = self.execute_order(order, inputs, &mut tx_ctx)?;
+        let (mut temporary_store, status) = self.execute_order(order, inputs, &mut tx_ctx)?;
 
         // Remove from dependencies the generic hash
         transaction_dependencies.remove(&TransactionDigest::genesis());
 
-        // Update the database in an atomic manner
+        let unwrapped_object_ids = self.get_unwrapped_object_ids(temporary_store.written())?;
+        temporary_store.patch_unwrapped_object_version(unwrapped_object_ids);
         let to_signed_effects = temporary_store.to_signed_effects(
             &self.name,
             &*self.secret,
@@ -288,6 +289,7 @@ impl AuthorityState {
             status,
             &gas_object_id,
         );
+        // Update the database in an atomic manner
         self.update_state(temporary_store, certificate, to_signed_effects)
             .await // Returns the OrderInfoResponse
     }
@@ -614,4 +616,28 @@ impl AuthorityState {
     ) -> Result<Option<(ObjectRef, TransactionDigest)>, FastPayError> {
         self._database.get_latest_parent_entry(object_id)
     }
+
+    /// Given all mutated objects during a transaction, return the list of objects
+    /// that were unwrapped (i.e. re-appeared after being deleted).
+    fn get_unwrapped_object_ids(
+        &self,
+        written: &BTreeMap<ObjectID, Object>,
+    ) -> Result<Vec<ObjectID>, FastPayError> {
+        // For each mutated object, we first find out whether there was a transaction
+        // that deleted this object in the past.
+        let parents = self._database.multi_get_parents(
+            &written
+                .iter()
+                .map(|(object_id, object)| (*object_id, object.version(), OBJECT_DIGEST_DELETED))
+                .collect::<Vec<_>>(),
+        )?;
+        // Filter the list of mutated objects based on whether they were deleted in the past.
+        // These objects are the unwrapped ones.
+        let filtered = written
+            .iter()
+            .zip(parents.iter())
+            .filter_map(|((object_id, _object), d)| d.map(|_| *object_id))
+            .collect();
+        Ok(filtered)
+    }
 }

fastpay_core/src/authority/authority_store.rs

@@ -181,6 +181,16 @@ impl AuthorityStore {
         self.parent_sync.get(object_ref).map_err(|e| e.into())
     }
 
+    /// Batch version of `parent` function.
+    pub fn multi_get_parents(
+        &self,
+        object_refs: &[ObjectRef],
+    ) -> Result<Vec<Option<TransactionDigest>>, FastPayError> {
+        self.parent_sync
+            .multi_get(object_refs)
+            .map_err(|e| e.into())
+    }
+
     /// Returns all parents (object_ref and transaction digests) that match an object_id, at
     /// any object version, or optionally at a specific version.
     pub fn get_parent_iterator(
@@ -346,7 +356,7 @@ impl AuthorityStore {
         // Delete the old owner index entries
         write_batch = write_batch.delete_batch(&self.owner_index, old_object_owners)?;
 
-        // Index the certificate by the objects created
+        // Index the certificate by the objects mutated
         write_batch = write_batch.insert_batch(
             &self.parent_sync,
             written

fastpay_core/src/authority/temporary_store.rs

@@ -90,6 +90,23 @@ impl AuthorityTemporaryStore {
         }
     }
 
+    /// We need to special handle objects that was deleted in the past but re-appeared.
+    /// These objects must have been wrapped into another object, and then got unwrapped.
+    /// When an object was deleted at version `v`, we added an record into `parent_sync`
+    /// with version `v+1` along with OBJECT_DIGEST_MAX. Now when the object re-appeared,
+    /// it will also have version `v+1`, leading to a violation of the invariant that any
+    /// object_id and version pair must be unique. Hence for any object that re-appear
+    /// after deletion, we force incrementing its version number again to make it `v+2`
+    /// before writing to the store.
+    pub fn patch_unwrapped_object_version(&mut self, unwrapped_object_ids: Vec<ObjectID>) {
+        for id in unwrapped_object_ids {
+            // The way we constructed `unwrapped_object_ids` guarantees it must exists
+            // in written and is a mutable move object. Safe to unwrap.
+            let object = self.written.get_mut(&id).unwrap();
+            object.data.try_as_move_mut().unwrap().increment_version();
+        }
+    }
+
     pub fn to_signed_effects(
         &self,
         authority_name: &AuthorityName,