feat(index): add the rebuildable record index

apps/origin-src/benches/store.rs

@@ -1,13 +1,22 @@
 //! Store throughput: durable writes with and without batching, verified
-//! reads, and target appends. Run on a real disk: syncs cost nothing on a
-//! RAM-backed file system.
+//! reads, target appends, and the record index. Run on a real disk: syncs
+//! cost nothing on a RAM-backed file system.
 
+use std::fs::File;
+use std::io;
+use std::path::Path;
 use std::slice;
 
 use divan::counter::{BytesCount, ItemsCount};
 use divan::{Bencher, black_box};
-use origin_platform::Native;
-use origin_shared::{AcceptanceBody, EncodedObject, NodeKey, Record, TargetName, Tree, TreeId};
+use origin_platform::{
+    CloneFile, Durability, Error, FileLock, FileSystem, FsKind, Native, Platform, ReadAt,
+};
+use origin_shared::{
+    AcceptanceBody, Base, ChangeId, EncodedObject, Kind, NodeKey, Record, RecordId, Revision,
+    Statement, TargetName, Tree, TreeId, Verdict,
+};
+use origin_src::index::Index;
 use origin_src::store::Store;
 
 fn main() {
@@ -26,8 +35,13 @@ fn content(len: usize, seed: usize) -> Vec<u8> {
 
 /// An initialized store in a new temporary directory.
 fn store() -> (tempfile::TempDir, Store) {
+    store_on(Native::default())
+}
+
+/// An initialized store on `platform` in a new temporary directory.
+fn store_on<P: Platform>(platform: P) -> (tempfile::TempDir, Store<P>) {
     let dir = tempfile::tempdir().unwrap();
-    let store = Store::new(dir.path().join(".origin"), Native::default());
+    let store = Store::new(dir.path().join(".origin"), platform);
     store.init().unwrap();
     (dir, store)
 }
@@ -120,3 +134,130 @@ fn append(bencher: Bencher<'_, '_>) {
         store.append(&main, &acceptance).unwrap()
     });
 }
+
+/// `count` records, as a busy repository receives them: a revision of one
+/// of a hundred changes, evolving from that change's previous one, then nine
+/// statements about it.
+fn records(count: usize) -> Vec<EncodedObject> {
+    let snapshot = TreeId::from(Tree::default().encode().id());
+    let mut heads = vec![None; 100];
+    let mut revision = None;
+    (0..count)
+        .map(|i| {
+            if i % 10 == 0 {
+                let change = i / 10 % heads.len();
+                let encoded = Revision {
+                    change: ChangeId::from_bytes([u8::try_from(change).unwrap(); 16]),
+                    predecessors: heads[change].into_iter().collect(),
+                    base: Base::Acceptance(EncodedObject::blob(b"main").id().into()),
+                    deps: Vec::new(),
+                    snapshot,
+                    description: EncodedObject::blob(&content(64, i)).id().into(),
+                    author: "user:bench".parse().unwrap(),
+                }
+                .encode();
+                heads[change] = Some(encoded.id().into());
+                revision = Some(encoded.id().into());
+                encoded
+            } else {
+                Statement {
+                    subject: RecordId::Revision(revision.unwrap()),
+                    kind: Kind::new(format!("ci.check-{i}")).unwrap(),
+                    verdict: Some(Verdict::Pass),
+                    issuer: "node:ci-1".parse().unwrap(),
+                    inputs: None,
+                    payload: None,
+                    supersedes: None,
+                }
+                .encode()
+            }
+        })
+        .collect()
+}
+
+/// The native platform without syncs, to measure what syncs hide.
+#[derive(Clone, Copy, Debug, Default)]
+struct NoSync;
+
+impl Durability for NoSync {
+    fn sync_file(&self, _: &File) -> io::Result<()> {
+        Ok(())
+    }
+
+    fn sync_dir(&self, _: &Path) -> io::Result<()> {
+        Ok(())
+    }
+
+    fn sync_batch<'f>(&self, _: impl IntoIterator<Item = &'f Path>) -> Result<(), Error> {
+        Ok(())
+    }
+}
+
+impl FileLock for NoSync {}
+
+impl ReadAt for NoSync {}
+
+impl CloneFile for NoSync {
+    fn cannot_clone(&self, error: &io::Error) -> bool {
+        Native::default().cannot_clone(error)
+    }
+}
+
+impl FileSystem for NoSync {
+    fn file_system(&self, path: &Path) -> io::Result<FsKind> {
+        Native::default().file_system(path)
+    }
+}
+
+/// Writing new records in one batch, which indexes them, against blobs of
+/// the same content, which aren't indexed: the difference is what indexing
+/// costs, which [`NoSync`] shows apart from the syncs. One record is what
+/// issuing a statement costs.
+#[divan::bench(
+    types = [Native, NoSync],
+    consts = [true, false],
+    args = [1, 1_000, 20_000],
+    sample_count = 10,
+    sample_size = 1,
+)]
+fn write_records<P: Platform + Default, const INDEXED: bool>(
+    bencher: Bencher<'_, '_>,
+    count: usize,
+) {
+    let objects: Vec<_> = records(count)
+        .into_iter()
+        .map(|record| {
+            if INDEXED {
+                record
+            } else {
+                EncodedObject::blob(record.bytes())
+            }
+        })
+        .collect();
+    bencher
+        .counter(ItemsCount::new(count))
+        .with_inputs(|| store_on(P::default()))
+        .bench_local_refs(|(_, store)| store.write_all(&objects).unwrap());
+}
+
+/// Opening the index of 20k records written in batches of a hundred.
+#[divan::bench(sample_count = 20)]
+fn open_index(bencher: Bencher<'_, '_>) {
+    let (_dir, store) = store();
+    for batch in records(20_000).chunks(100) {
+        store.write_all(batch).unwrap();
+    }
+    bencher
+        .counter(ItemsCount::new(20_000_usize))
+        .bench(|| Index::open(&store).unwrap());
+}
+
+/// Rebuilding the index of 20k records from the store.
+#[divan::bench(sample_count = 10)]
+fn rebuild_index(bencher: Bencher<'_, '_>) {
+    let (_dir, store) = store();
+    store.write_all(&records(20_000)).unwrap();
+    bencher
+        .counter(ItemsCount::new(20_000_usize))
+        .bench(|| Index::rebuild(&store).unwrap());
+}