Added a parent iterator by object_id or (objectid,version) (#166)

Co-authored-by: George Danezis <george@danez.is>

fastpay_core/src/authority.rs

@@ -497,4 +497,16 @@ impl AuthorityState {
     ) -> Result<Vec<Option<Object>>, FastPayError> {
         self._database.get_objects(_objects)
     }
+
+    /// Returns all parents (object_ref and transaction digests) that match an object_id, at
+    /// any object version, or optionally at a specific version.
+    pub async fn get_parent_iterator(
+        &self,
+        object_id: ObjectID,
+        seq: Option<SequenceNumber>,
+    ) -> Result<Vec<(ObjectRef, TransactionDigest)>, FastPayError> {
+        {
+            self._database.get_parent_iterator(object_id, seq)
+        }
+    }
 }

fastpay_core/src/authority/authority_store.rs

@@ -142,6 +142,32 @@ impl AuthorityStore {
             .map_err(|_| FastPayError::StorageError)
     }
 
+    /// 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(
+        &self,
+        object_id: ObjectID,
+        seq: Option<SequenceNumber>,
+    ) -> Result<Vec<(ObjectRef, TransactionDigest)>, FastPayError> {
+        let seq_inner = seq.unwrap_or_else(|| SequenceNumber::from(0));
+        let obj_dig_inner = ObjectDigest::new([0; 32]);
+
+        Ok(self
+            .parent_sync
+            .iter()
+            // The object id [0; 16] is the smallest possible
+            .skip_to(&(object_id, seq_inner, obj_dig_inner))
+            .map_err(|_| FastPayError::StorageError)?
+            .take_while(|((id, iseq, _digest), _txd)| {
+                let mut flag = id == &object_id;
+                if seq.is_some() {
+                    flag &= seq_inner == *iseq;
+                }
+                flag
+            })
+            .collect())
+    }
+
     // Methods to mutate the store
 
     /// Insert an object

fastpay_core/src/unit_tests/authority_tests.rs

@@ -767,7 +767,7 @@ async fn test_handle_confirmation_order_ok() {
                 .unwrap();
             authority_state.read_certificate(&refx).await.unwrap()
         },
-        Some(certified_transfer_order)
+        Some(certified_transfer_order.clone())
     );
 
     // Check locks are set and archived correctly
@@ -780,6 +780,15 @@ async fn test_handle_confirmation_order_ok() {
         .await
         .expect("Exists")
         .is_none());
+
+    // Check that all the parents are returned.
+    assert!(
+        authority_state.get_parent_iterator(object_id, None).await
+            == Ok(vec![(
+                (object_id, 1.into(), new_account.digest()),
+                certified_transfer_order.order.digest()
+            )])
+    );
 }
 
 #[tokio::test]