feat: content hashes iterator for sync store (#1501)

## Description

Adds an iterator over all content hashes in the document store. Needed
for #1496.

## Notes & open questions

<!-- Any notes, remarks or open questions you have to make about the PR.
-->

## Change checklist

- [x] Self-review.
- [x] Documentation updates if relevant.
- [x] Tests if relevant.

Author: Frando

iroh-sync/src/store.rs

@@ -1,6 +1,7 @@
 //! Storage trait and implementation for iroh-sync documents
 
 use anyhow::Result;
+use iroh_bytes::Hash;
 use rand_core::CryptoRngCore;
 use serde::{Deserialize, Serialize};
 
@@ -26,6 +27,11 @@ pub trait Store: std::fmt::Debug + Clone + Send + Sync + 'static {
     where
         Self: 'a;
 
+    /// Iterator over all content hashes in the store, returned from [`Self::content_hashes`]
+    type ContentHashesIter<'a>: Iterator<Item = Result<Hash>>
+    where
+        Self: 'a;
+
     /// Iterator over replica namespaces in the store, returned from [`Self::list_namespaces`]
     type NamespaceIter<'a>: Iterator<Item = Result<NamespaceId>>
     where
@@ -78,6 +84,9 @@ pub trait Store: std::fmt::Debug + Clone + Send + Sync + 'static {
         author: AuthorId,
         key: impl AsRef<[u8]>,
     ) -> Result<Option<SignedEntry>>;
+
+    /// Get all content hashes of all replicas in the store.
+    fn content_hashes(&self) -> Result<Self::ContentHashesIter<'_>>;
 }
 
 /// Filter a get query onto a namespace

iroh-sync/src/store/fs.rs

@@ -5,6 +5,7 @@ use std::{cmp::Ordering, collections::HashMap, path::Path, sync::Arc};
 use anyhow::Result;
 use derive_more::From;
 use ed25519_dalek::{SignatureError, VerifyingKey};
+use iroh_bytes::Hash;
 use ouroboros::self_referencing;
 use parking_lot::RwLock;
 use redb::{
@@ -111,6 +112,7 @@ impl Store {
 impl super::Store for Store {
     type Instance = StoreInstance;
     type GetIter<'a> = RangeIterator<'a>;
+    type ContentHashesIter<'a> = ContentHashesIterator<'a>;
     type AuthorsIter<'a> = std::vec::IntoIter<Result<Author>>;
     type NamespaceIter<'a> = std::vec::IntoIter<Result<NamespaceId>>;
 
@@ -225,6 +227,10 @@ impl super::Store for Store {
 
         Ok(Some(signed_entry))
     }
+
+    fn content_hashes(&self) -> Result<Self::ContentHashesIter<'_>> {
+        ContentHashesIterator::create(&self.db)
+    }
 }
 
 impl Store {
@@ -503,6 +509,47 @@ impl crate::ranger::Store<SignedEntry> for StoreInstance {
     }
 }
 
+/// Iterator over all content hashes for the fs store.
+#[self_referencing]
+pub struct ContentHashesIterator<'a> {
+    read_tx: ReadTransaction<'a>,
+    #[borrows(read_tx)]
+    #[covariant]
+    record_table: RecordsTable<'this>,
+    #[covariant]
+    #[borrows(record_table)]
+    records: RecordsRange<'this>,
+}
+impl<'a> ContentHashesIterator<'a> {
+    fn create(db: &'a Arc<Database>) -> anyhow::Result<Self> {
+        let iter = Self::try_new(
+            db.begin_read()?,
+            |read_tx| {
+                read_tx
+                    .open_table(RECORDS_TABLE)
+                    .map_err(anyhow::Error::from)
+            },
+            |table| table.iter().map_err(anyhow::Error::from),
+        )?;
+        Ok(iter)
+    }
+}
+
+impl Iterator for ContentHashesIterator<'_> {
+    type Item = Result<Hash>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        self.with_mut(|fields| match fields.records.next() {
+            None => None,
+            Some(Err(err)) => Some(Err(err.into())),
+            Some(Ok((_key, value))) => {
+                let (_timestamp, _namespace_sig, _author_sig, _len, hash) = value.value();
+                Some(Ok(Hash::from(hash)))
+            }
+        })
+    }
+}
+
 #[self_referencing]
 pub struct RangeIterator<'a> {
     read_tx: ReadTransaction<'a>,

iroh-sync/src/store/memory.rs

@@ -8,6 +8,7 @@ use std::{
 
 use anyhow::Result;
 use ed25519_dalek::{SignatureError, VerifyingKey};
+use iroh_bytes::Hash;
 use parking_lot::{RwLock, RwLockReadGuard};
 
 use crate::{
@@ -31,11 +32,12 @@ pub struct Store {
 type Rid = (AuthorId, Vec<u8>);
 type Rvalue = SignedEntry;
 type RecordMap = BTreeMap<Rid, Rvalue>;
-type ReplicaRecordsOwned = HashMap<NamespaceId, RecordMap>;
+type ReplicaRecordsOwned = BTreeMap<NamespaceId, RecordMap>;
 
 impl super::Store for Store {
     type Instance = ReplicaStoreInstance;
     type GetIter<'a> = RangeIterator<'a>;
+    type ContentHashesIter<'a> = ContentHashesIterator<'a>;
     type AuthorsIter<'a> = std::vec::IntoIter<Result<Author>>;
     type NamespaceIter<'a> = std::vec::IntoIter<Result<NamespaceId>>;
 
@@ -117,6 +119,16 @@ impl super::Store for Store {
 
         Ok(value.cloned())
     }
+
+    /// Get all content hashes of all replicas in the store.
+    fn content_hashes(&self) -> Result<Self::ContentHashesIter<'_>> {
+        let records = self.replica_records.read();
+        Ok(ContentHashesIterator {
+            records,
+            namespace_i: 0,
+            record_i: 0,
+        })
+    }
 }
 
 impl Store {
@@ -234,6 +246,31 @@ impl GetFilter {
     }
 }
 
+/// Iterator over all content hashes in the memory store.
+pub struct ContentHashesIterator<'a> {
+    records: ReplicaRecords<'a>,
+    namespace_i: usize,
+    record_i: usize,
+}
+impl<'a> Iterator for ContentHashesIterator<'a> {
+    type Item = Result<Hash>;
+    fn next(&mut self) -> Option<Self::Item> {
+        loop {
+            let records = self.records.values().nth(self.namespace_i)?;
+            match records.values().nth(self.record_i) {
+                None => {
+                    self.namespace_i += 1;
+                    self.record_i = 0;
+                }
+                Some(record) => {
+                    self.record_i += 1;
+                    return Some(Ok(record.content_hash()));
+                }
+            }
+        }
+    }
+}
+
 /// Iterator over entries in the memory store
 #[derive(Debug)]
 pub struct RangeIterator<'a> {

iroh-sync/src/sync.rs

@@ -719,6 +719,8 @@ impl Record {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashSet;
+
     use anyhow::Result;
     use rand_core::SeedableRng;
 
@@ -887,6 +889,42 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_content_hashes_iterator_memory() -> Result<()> {
+        let store = store::memory::Store::default();
+        test_basics(store)
+    }
+
+    #[cfg(feature = "fs-store")]
+    #[test]
+    fn test_content_hashes_iterator_fs() -> Result<()> {
+        let dbfile = tempfile::NamedTempFile::new()?;
+        let store = store::fs::Store::new(dbfile.path())?;
+        test_content_hashes_iterator(store)
+    }
+
+    fn test_content_hashes_iterator<S: store::Store>(store: S) -> Result<()> {
+        let mut rng = rand::thread_rng();
+        let mut expected = HashSet::new();
+        let n_replicas = 3;
+        let n_entries = 4;
+        for i in 0..n_replicas {
+            let namespace = Namespace::new(&mut rng);
+            let author = store.new_author(&mut rng)?;
+            let replica = store.new_replica(namespace)?;
+            for j in 0..n_entries {
+                let key = format!("{j}");
+                let data = format!("{i}:{j}");
+                let hash = replica.hash_and_insert(key, &author, data)?;
+                expected.insert(hash);
+            }
+        }
+        assert_eq!(expected.len(), n_replicas * n_entries);
+        let actual = store.content_hashes()?.collect::<Result<HashSet<Hash>>>()?;
+        assert_eq!(actual, expected);
+        Ok(())
+    }
+
     #[test]
     fn test_multikey() {
         let mut rng = rand::thread_rng();