fix: more robustness in the face of a trampling-herd of threads loadi…

…ng a single index.

The motivating example is here: praetorian-inc/noseyparker#179

gitoxide-core/src/repository/odb.rs

@@ -1,4 +1,5 @@
 use std::io;
+use std::sync::atomic::Ordering;
 
 use anyhow::bail;
 
@@ -50,6 +51,8 @@ pub mod statistics {
     pub struct Options {
         pub format: OutputFormat,
         pub thread_limit: Option<usize>,
+        /// A debug-flag that triggers looking up the headers of all objects again, but without indices preloaded
+        pub extra_header_lookup: bool,
     }
 }
 
@@ -59,7 +62,11 @@ pub fn statistics(
     mut progress: impl gix::Progress,
     out: impl io::Write,
     mut err: impl io::Write,
-    statistics::Options { format, thread_limit }: statistics::Options,
+    statistics::Options {
+        format,
+        thread_limit,
+        extra_header_lookup,
+    }: statistics::Options,
 ) -> anyhow::Result<()> {
     use bytesize::ByteSize;
     use gix::odb::{find, HeaderExt};
@@ -76,6 +83,10 @@ pub fn statistics(
     #[cfg_attr(feature = "serde", derive(serde::Serialize))]
     #[derive(Default)]
     struct Statistics {
+        /// All objects that were used to produce these statistics.
+        /// Only `Some` if we are doing an extra round of header queries on a repository without loaded indices.
+        #[cfg_attr(feature = "serde", serde(skip_serializing))]
+        ids: Option<Vec<gix::ObjectId>>,
         total_objects: usize,
         loose_objects: usize,
         packed_objects: usize,
@@ -135,14 +146,17 @@ pub fn statistics(
     }
 
     impl gix::parallel::Reduce for Reduce {
-        type Input = Result<Vec<gix::odb::find::Header>, anyhow::Error>;
+        type Input = Result<Vec<(gix::ObjectId, gix::odb::find::Header)>, anyhow::Error>;
         type FeedProduce = ();
         type Output = Statistics;
         type Error = anyhow::Error;
 
         fn feed(&mut self, items: Self::Input) -> Result<Self::FeedProduce, Self::Error> {
-            for item in items? {
+            for (id, item) in items? {
                 self.stats.consume(item);
+                if let Some(ids) = self.stats.ids.as_mut() {
+                    ids.push(id);
+                }
             }
             Ok(())
         }
@@ -154,9 +168,9 @@ pub fn statistics(
     }
 
     let cancelled = || anyhow::anyhow!("Cancelled by user");
-    let object_ids = repo.objects.store_ref().iter()?.filter_map(Result::ok);
+    let object_ids = repo.objects.iter()?.filter_map(Result::ok);
     let chunk_size = 1_000;
-    let stats = if gix::parallel::num_threads(thread_limit) > 1 {
+    let mut stats = if gix::parallel::num_threads(thread_limit) > 1 {
         gix::parallel::in_parallel(
             gix::interrupt::Iter::new(
                 gix::features::iter::Chunks {
@@ -166,19 +180,30 @@ pub fn statistics(
                 cancelled,
             ),
             thread_limit,
-            move |_| (repo.objects.clone().into_inner(), counter),
+            {
+                let objects = repo.objects.clone();
+                move |_| (objects.clone().into_inner(), counter)
+            },
             |ids, (handle, counter)| {
                 let ids = ids?;
-                counter.fetch_add(ids.len(), std::sync::atomic::Ordering::Relaxed);
+                counter.fetch_add(ids.len(), Ordering::Relaxed);
                 let out = ids
                     .into_iter()
-                    .map(|id| handle.header(id))
+                    .map(|id| handle.header(id).map(|hdr| (id, hdr)))
                     .collect::<Result<Vec<_>, _>>()?;
                 Ok(out)
             },
-            Reduce::default(),
+            Reduce {
+                stats: Statistics {
+                    ids: extra_header_lookup.then(Vec::new),
+                    ..Default::default()
+                },
+            },
         )?
     } else {
+        if extra_header_lookup {
+            bail!("extra-header-lookup is only meaningful in threaded mode");
+        }
         let mut stats = Statistics::default();
 
         for (count, id) in object_ids.enumerate() {
@@ -193,6 +218,40 @@ pub fn statistics(
 
     progress.show_throughput(start);
 
+    if let Some(mut ids) = stats.ids.take() {
+        // Critical to re-open the repo to assure we don't have any ODB state and start fresh.
+        let start = std::time::Instant::now();
+        let repo = gix::open_opts(repo.git_dir(), repo.open_options().to_owned())?;
+        progress.set_name("re-counting".into());
+        progress.init(Some(ids.len()), gix::progress::count("objects"));
+        let counter = progress.counter();
+        counter.store(0, Ordering::Relaxed);
+        dbg!("starting");
+        let errors = gix::parallel::in_parallel_with_slice(
+            &mut ids,
+            thread_limit,
+            {
+                let objects = repo.objects.clone();
+                move |_| (objects.clone().into_inner(), counter, false)
+            },
+            |id, (odb, counter, has_error), _threads_left, _stop_everything| -> anyhow::Result<()> {
+                counter.fetch_add(1, Ordering::Relaxed);
+                if let Err(_err) = odb.header(id) {
+                    *has_error = true;
+                    gix::trace::error!(err = ?_err, "Object that is known to be present wasn't found");
+                }
+                Ok(())
+            },
+            || Some(std::time::Duration::from_millis(100)),
+            |(_, _, has_error)| has_error,
+        )?;
+
+        progress.show_throughput(start);
+        if errors.contains(&true) {
+            bail!("At least one object couldn't be looked up even though it must exist");
+        }
+    }
+
     #[cfg(feature = "serde")]
     {
         serde_json::to_writer_pretty(out, &stats)?;

gix-odb/src/store_impls/dynamic/load_index.rs

@@ -86,7 +86,7 @@ impl super::Store {
             Ok(Some(self.collect_snapshot()))
         } else {
             // always compare to the latest state
-            // Nothing changed in the mean time, try to load another index…
+            // Nothing changed in the meantime, try to load another index…
             if self.load_next_index(index) {
                 Ok(Some(self.collect_snapshot()))
             } else {
@@ -119,7 +119,7 @@ impl super::Store {
                         let slot = &self.files[index.slot_indices[slot_map_index]];
                         let _lock = slot.write.lock();
                         if slot.generation.load(Ordering::SeqCst) > index.generation {
-                            // There is a disk consolidation in progress which just overwrote a slot that cold be disposed with some other
+                            // There is a disk consolidation in progress which just overwrote a slot that could be disposed with some other
                             // index, one we didn't intend to load.
                             // Continue with the next slot index in the hope there is something else we can do…
                             continue 'retry_with_next_slot_index;
@@ -134,7 +134,8 @@ impl super::Store {
                                     slot.files.store(bundle);
                                     break 'retry_with_next_slot_index;
                                 }
-                                Err(_) => {
+                                Err(_err) => {
+                                    gix_features::trace::error!(err=?_err, "Failed to load index file - some objects may seem to not exist");
                                     slot.files.store(bundle);
                                     continue 'retry_with_next_slot_index;
                                 }
@@ -183,27 +184,30 @@ impl super::Store {
         let previous_index_state = Arc::as_ptr(&index) as usize;
 
         // IMPORTANT: get a lock after we recorded the previous state.
+        eprintln!("needs_init: {needs_init}, load_new_index: {load_new_index}");
         let write = self.write.lock();
         let objects_directory = &self.path;
 
         // Now we know the index isn't going to change anymore, even though threads might still load indices in the meantime.
         let index = self.index.load();
         if previous_index_state != Arc::as_ptr(&index) as usize {
             // Someone else took the look before and changed the index. Return it without doing any additional work.
+            dbg!("somebody already did the work - arc pointer changed");
             return Ok(Some(self.collect_snapshot()));
         }
 
         let was_uninitialized = !index.is_initialized();
 
         // We might not be able to detect by pointer if the state changed, as this itself is racy. So we keep track of double-initialization
         // using a flag, which means that if `needs_init` was true we saw the index uninitialized once, but now that we are here it's
-        // initialized meaning that somebody was faster and we couldn't detect it by comparisons to the index.
+        // initialized meaning that somebody was faster, and we couldn't detect it by comparisons to the index.
         // If so, make sure we collect the snapshot instead of returning None in case nothing actually changed, which is likely with a
         // race like this.
         if !was_uninitialized && needs_init {
+            dbg!("somebody already did the work - isn't uninitialized anymore");
             return Ok(Some(self.collect_snapshot()));
         }
-        self.num_disk_state_consolidation.fetch_add(1, Ordering::Relaxed);
+        dbg!(self.num_disk_state_consolidation.fetch_add(1, Ordering::Relaxed));
 
         let db_paths: Vec<_> = std::iter::once(objects_directory.to_owned())
             .chain(crate::alternate::resolve(objects_directory.clone(), &self.current_dir)?)
@@ -382,6 +386,7 @@ impl super::Store {
                     .unwrap_or_else(|| Arc::new(num_loaded_indices.into())),
                 num_indices_currently_being_loaded: Default::default(),
             });
+            dbg!("storing new index");
             self.index.store(new_index);
         }
 
@@ -408,7 +413,14 @@ impl super::Store {
 
         let new_index = self.index.load();
         Ok(if index.state_id() == new_index.state_id() {
-            // there was no change, and nothing was loaded in the meantime, reflect that in the return value to not get into loops
+            dbg!("nothing changed after reloading");
+            // there was no change, and nothing was loaded in the meantime, reflect that in the return value to not get into loops.
+            // However, if we came here in desperation for something, we should collect what's there for a chance that
+            // it will be what was needed.
+            // The original problem here is that a trampling-herd of threads want to load just a single index. Most of these threads
+            // notice that something changed in the meantime and can avoid checking disk once again. However, one might not get it
+            // as their index is just too new. Maybe there is a way to detect this kind of race or to prevent it?
+            // load_new_index.then(|| self.collect_snapshot())
             None
         } else {
             if load_new_index {

gix-odb/src/store_impls/dynamic/types.rs

@@ -18,7 +18,7 @@ pub(crate) type AtomicGeneration = AtomicU32;
 
 /// A way to indicate which pack indices we have seen already and which of them are loaded, along with an idea
 /// of whether stored `PackId`s are still usable.
-#[derive(Default, Copy, Clone)]
+#[derive(Default, Copy, Clone, Debug)]
 pub struct SlotIndexMarker {
     /// The generation the `loaded_until_index` belongs to. Indices of different generations are completely incompatible.
     /// This value changes once the internal representation is compacted, something that may happen only if there is no handle

src/plumbing/main.rs

@@ -1153,7 +1153,7 @@ pub fn main() -> Result<()> {
             ),
         },
         Subcommands::Odb(cmd) => match cmd {
-            odb::Subcommands::Stats => prepare_and_run(
+            odb::Subcommands::Stats { extra_header_lookup } => prepare_and_run(
                 "odb-stats",
                 trace,
                 auto_verbose,
@@ -1166,7 +1166,11 @@ pub fn main() -> Result<()> {
                         progress,
                         out,
                         err,
-                        core::repository::odb::statistics::Options { format, thread_limit },
+                        core::repository::odb::statistics::Options {
+                            format,
+                            thread_limit,
+                            extra_header_lookup,
+                        },
                     )
                 },
             ),

src/plumbing/options/mod.rs

@@ -586,7 +586,11 @@ pub mod odb {
         Info,
         /// Count and obtain information on all, possibly duplicate, objects in the database.
         #[clap(visible_alias = "statistics")]
-        Stats,
+        Stats {
+            /// Lookup headers again, but without preloading indices.
+            #[clap(long)]
+            extra_header_lookup: bool,
+        },
     }
 }