Views: base references, reclaim and base-builder admission

benchmarks/aft-search/slice-descriptors/land-views-c-base-references.json

@@ -0,0 +1,8 @@
+{
+  "targeted_mechanism": "none",
+  "kind": "harness",
+  "slice_class": "non_ranking",
+  "fixtures": [
+    "harness-goldens"
+  ]
+}

crates/aft/src/blob_store/v2.rs

@@ -97,7 +97,7 @@ impl FamilyPlane {
         }
     }
 
-    const fn payload_schema(self) -> u32 {
+    pub(crate) const fn payload_schema(self) -> u32 {
         match self {
             Self::Trigram => TRIGRAM_PAYLOAD_SCHEMA,
             Self::Semantic => crate::blob_store::SEMANTIC_PAYLOAD_SCHEMA,

crates/aft/src/blob_store/v3/tests.rs

@@ -179,6 +179,7 @@ fn publication_observes_floor_before_files_and_flush_before_commit() {
                 | DurabilityStep::ManifestParentSynced
                 | DurabilityStep::PointerCas
                 | DurabilityStep::PointerSynced => panic!("v2 boundary in v3 publication"),
+                other => panic!("base-reference boundary {other:?} in a copy publication"),
             }
         }
     }
@@ -685,6 +686,7 @@ fn killed_v3_publishers_recover_at_both_new_boundaries_without_losing_orphans()
             | DurabilityStep::ManifestParentSynced
             | DurabilityStep::PointerCas
             | DurabilityStep::PointerSynced => panic!("V3_ONLY contains a v2 boundary"),
+            other => panic!("V3_ONLY contains the base-path boundary {other:?}"),
         }
     }
 }
@@ -859,11 +861,13 @@ fn newer_format_keeps_artifact_refusal_and_malformed_file_reports_found_zero() {
     let path = db(root.path(), A);
     Connection::open(&path)
         .unwrap()
-        .execute("UPDATE format_meta SET format_version=2", [])
+        // Registry 2 is the base-reference protocol this build reads; the
+        // first newer version is 3.
+        .execute("UPDATE format_meta SET format_version=3", [])
         .unwrap();
     match LedgerReader::open_existing(root.path(), A).err().unwrap() {
         Error::Incompatible(error) => {
-            assert_eq!(error.found, 2);
+            assert_eq!(error.found, 3);
             assert_eq!(error.source, persisted_format::RefusalSource::Artifact);
         }
         error => panic!("{error}"),

crates/aft/src/persisted_format.rs

@@ -111,7 +111,8 @@ impl PersistedStore {
             PersistedStore::ArtifactOwner => crate::artifact_owner::SCHEMA_VERSION,
             PersistedStore::BackupMeta => crate::backup::SCHEMA_VERSION,
             PersistedStore::BashTask => crate::bash_background::persistence::SCHEMA_VERSION,
-            PersistedStore::ViewsV3Registry | PersistedStore::ViewsV3Store => 1,
+            PersistedStore::ViewsV3Registry => crate::blob_store::v3::REGISTRY_BASE_REFS_VERSION,
+            PersistedStore::ViewsV3Store => 1,
         }
     }
 

crates/aft/src/pins/live.rs

@@ -51,6 +51,35 @@ impl LivePin {
         })
     }
 
+    /// Creates an empty live pin in the `pins/` of the v3 member `scope`,
+    /// inside the v3 registry barrier: the member must still be registered and
+    /// not `evicting`, so the retention collector's final protection check
+    /// either sees this pin or has already removed the member.
+    pub fn create_v3(storage: &Path, family: &str, scope: &str) -> Result<Self, PinError> {
+        crate::views::base_refs::with_base_barrier(storage, family, |tx| {
+            let state: Option<String> = rusqlite::OptionalExtension::optional(tx.query_row(
+                "SELECT state FROM members WHERE scope=?1",
+                [scope],
+                |r| r.get(0),
+            ))?;
+            if state.is_none() || state.as_deref() == Some(crate::views::registry::STATE_EVICTING) {
+                return Err(crate::views::base_refs::BaseRefError::Unavailable(format!(
+                    "v3 member {scope} is not registered"
+                )));
+            }
+            let view_dir = crate::blob_store::v3::view_dir(storage, family, scope)?;
+            Self::create_in(&view_dir, family.to_owned(), scope.to_owned()).map_err(|error| {
+                crate::views::base_refs::BaseRefError::Io(std::io::Error::other(error.to_string()))
+            })
+        })
+        .map_err(|error| match error {
+            crate::views::base_refs::BaseRefError::Unavailable(_) => {
+                PinError::NotRegistered(scope.to_owned())
+            }
+            other => PinError::Registry(other.to_string()),
+        })
+    }
+
     fn create_in(view_dir: &Path, family: String, view: String) -> Result<Self, PinError> {
         let pid = std::process::id();
         let now = now_ms();

crates/aft/src/reader_floor.rs

@@ -785,12 +785,14 @@ pub(crate) mod tests {
     #[test]
     fn v3_registration_is_advertised_but_not_prepared() {
         let formats = persisted_format::formats_json();
-        for name in ["views_v3_registry", "views_v3_store"] {
-            assert_eq!(formats["stores"][name], 1, "{name}");
+        // The registry reads the base-reference protocol (2/2); plane stores
+        // keep format 1. Neither appears in the startup baseline floor.
+        for (name, version) in [("views_v3_registry", 2), ("views_v3_store", 1)] {
+            assert_eq!(formats["stores"][name], version, "{name}");
             let store = PersistedStore::from_name(name).expect("registered store");
             assert!(PersistedStore::ALL.contains(&store));
-            assert_eq!(store.supported(), 1);
-            assert_eq!(store.written(), 1);
+            assert_eq!(store.supported(), version);
+            assert_eq!(store.written(), version);
         }
         let root = tempfile::tempdir().unwrap();
         assert!(prepare_uncached(root.path()).is_empty());

crates/aft/src/root_cache.rs

@@ -761,6 +761,10 @@ pub struct ReadMarkerMetadata {
     pub pid: u32,
     pub hostname: String,
     pub created_at_ms: u64,
+    /// Per-acquisition identity of a base-reference marker. Ordinary readers
+    /// leave it out, so their files are unchanged.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub nonce: Option<String>,
 }
 
 #[derive(Debug)]
@@ -774,10 +778,45 @@ pub struct ReadMarker {
 
 impl ReadMarker {
     pub fn create(cache_dir: &Path, generation_label: &str) -> io::Result<Self> {
+        Self::create_inner(cache_dir, generation_label, None)
+    }
+
+    /// A marker whose file also records `nonce`, so its holder can later
+    /// prove the file at its unique path is still its own acquisition.
+    pub fn create_with_nonce(
+        cache_dir: &Path,
+        generation_label: &str,
+        nonce: &str,
+    ) -> io::Result<Self> {
+        Self::create_inner(cache_dir, generation_label, Some(nonce.to_owned()))
+    }
+
+    /// True when the marker file at this marker's own path still exists and
+    /// carries this marker's metadata, nonce included. A sweep that judged the
+    /// owner dead removes the file, and this then reports `false`.
+    pub fn verify_held(&self) -> io::Result<bool> {
+        use std::io::Read;
+        let mut file = match self.generation.repair_read_file(&self.name) {
+            Ok(file) => file,
+            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
+            Err(error) => return Err(error),
+        };
+        let mut bytes = Vec::new();
+        (&mut file).take(32 * 1024).read_to_end(&mut bytes)?;
+        Ok(serde_json::from_slice::<ReadMarkerMetadata>(&bytes)
+            .is_ok_and(|stored| stored == self.metadata))
+    }
+
+    fn create_inner(
+        cache_dir: &Path,
+        generation_label: &str,
+        nonce: Option<String>,
+    ) -> io::Result<Self> {
         let metadata = ReadMarkerMetadata {
             pid: std::process::id(),
             hostname: current_hostname(),
             created_at_ms: now_ms(),
+            nonce,
         };
         let dir = read_marker_dir(cache_dir, generation_label);
         crate::private_storage::open_dir(cache_dir, &dir)?;
@@ -1559,6 +1598,7 @@ mod tests {
             pid: 0,
             hostname: current_hostname(),
             created_at_ms: now_ms(),
+            nonce: None,
         };
         fs::create_dir_all(marker_path.parent().unwrap()).unwrap();
         write_marker_file(&marker_path, &metadata).unwrap();
@@ -1579,6 +1619,7 @@ mod tests {
             pid,
             hostname: current_hostname(),
             created_at_ms: 1_000,
+            nonce: None,
         };
         fs::create_dir_all(marker_path.parent().unwrap()).unwrap();
         write_marker_file(&marker_path, &metadata).unwrap();
@@ -1600,6 +1641,7 @@ mod tests {
             pid: 123,
             hostname: format!("other-{}", current_hostname()),
             created_at_ms: now_ms(),
+            nonce: None,
         };
         fs::create_dir_all(marker_path.parent().unwrap()).unwrap();
         write_marker_file(&marker_path, &metadata).unwrap();

crates/aft/src/storage_retention.rs

@@ -40,7 +40,7 @@ const INTERVAL: Duration = Duration::from_secs(10 * 60);
 const STARTUP_GRACE: Duration = Duration::from_secs(5 * 60);
 const PASS_BUDGET: Duration = Duration::from_secs(5);
 const PASS_ENTRIES: usize = 64;
-const TREE_ENTRIES: usize = 4096;
+pub(crate) const TREE_ENTRIES: usize = 4096;
 const TRASH_SCAN_HANDLES: usize = 64;
 const HISTORY_ENTRIES: usize = 8192;
 static REPORTED_BINDING_READ_ERRORS: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
@@ -1614,7 +1614,7 @@ fn read_json<T: serde::de::DeserializeOwned>(path: &Path) -> io::Result<T> {
     serde_json::from_slice(&fs::read(path)?).map_err(io::Error::other)
 }
 
-fn bounded_dirs(directory: &Path, cap: usize) -> io::Result<Vec<fs::DirEntry>> {
+pub(crate) fn bounded_dirs(directory: &Path, cap: usize) -> io::Result<Vec<fs::DirEntry>> {
     let entries = match fs::read_dir(directory) {
         Ok(entries) => entries,
         Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
@@ -4660,7 +4660,7 @@ enum V3Step {
 /// way, so they never keep a member. `Err` is uncertainty (an unreadable pin,
 /// or a `pins/` or `readers/` listing longer than the bound) and keeps the
 /// member without resetting its count, as the v2 family sweep does.
-fn v3_member_protected(
+pub(crate) fn v3_member_protected(
     tx: &rusqlite::Transaction<'_>,
     scope: &str,
     view_dir: &Path,
@@ -4887,10 +4887,16 @@ fn v3_finish_eviction(
     crate::views::registry::remove_view_dir(view_dir)?;
     registry
         .with_barrier(|tx| {
-            tx.execute(
+            let removed = tx.execute(
                 "DELETE FROM members WHERE scope=?1 AND state=?2",
                 rusqlite::params![scope, crate::views::registry::STATE_EVICTING],
             )?;
+            // A deregistered worker's referrer and strict-rebuild rows go in
+            // the same transaction as its member row, after its directory.
+            if removed > 0 && crate::views::base_refs::has_base_refs(tx)? {
+                tx.execute("DELETE FROM base_refs WHERE worker_scope=?1", [scope])?;
+                tx.execute("DELETE FROM strict_rebuild WHERE worker_scope=?1", [scope])?;
+            }
             Ok(())
         })
         .map_err(|error| io::Error::other(error.to_string()))?;
@@ -5203,6 +5209,44 @@ fn collect_v3_family(
             },
         )?;
     }
+    // The base-reference sweep: referrer reconciliation, the two-sweep base
+    // reclaim and the plane-row sweep. It needs the registry's base-reference
+    // tables and runs under the reservation lock this visit already holds.
+    if !cancelled()
+        && Instant::now() < deadline
+        && registry.registry_version().map_err(v3)?
+            >= crate::blob_store::v3::REGISTRY_BASE_REFS_VERSION
+    {
+        let plane_cursor = cursor_root(paths)?
+            .join("v3-plane-rows")
+            .join(format!("{family}.json"));
+        let bounds = crate::views::base_sweep::SweepBounds {
+            deadline: Some(deadline),
+            ..Default::default()
+        };
+        match crate::views::base_sweep::sweep_family_locked(
+            storage,
+            family,
+            &lock,
+            None,
+            bounds,
+            Some(&plane_cursor),
+        ) {
+            Ok(swept) => {
+                report.removed_roots += swept.reclaimed_bases.len();
+                report.removed_blobs += swept.plane_rows_deleted as usize;
+                report.examined_blobs += swept.plane_rows_examined as usize;
+                report.stopped_early |= !swept.plane_rows_complete && swept.skipped.is_none();
+            }
+            Err(error) => {
+                if report.errors.len() < 16 {
+                    report
+                        .errors
+                        .push(format!("v3 base sweep of family {family}: {error}"));
+                }
+            }
+        }
+    }
     // Every collector pass ends with a covering measurement, so committed
     // usage also reflects files removed by anything else since the last one.
     // It is skipped once one is already owed: the deadline has passed.
@@ -12827,3 +12871,77 @@ mod v3_collector_tests {
         assert!(member.join("data.bin").exists());
     }
 }
+
+#[cfg(test)]
+mod v3_base_reference_collector_tests {
+    use super::*;
+    use crate::views::base_refs::tests::{
+        bind, plane_has, publish_test_base, published, put_rows, refs, worker_delta, FAMILY,
+    };
+    use crate::views::base_refs::with_base_barrier;
+
+    fn pass(storage: &Path) -> SweepReport {
+        let mut report = SweepReport::default();
+        collect_one_v3_family(
+            &StoragePaths::same(storage),
+            &|| false,
+            &mut report,
+            V3Bounds::PASS,
+        )
+        .unwrap();
+        assert!(report.errors.is_empty(), "{report:?}");
+        report
+    }
+
+    fn registered(storage: &Path, scope: &str) -> bool {
+        crate::blob_store::v3::LedgerReader::open_existing(storage, FAMILY)
+            .unwrap()
+            .member(scope)
+            .unwrap()
+            .is_some()
+    }
+
+    #[test]
+    fn the_collector_keeps_a_referenced_base_and_reclaims_it_two_passes_after_its_worker_goes() {
+        let _serial = crate::blob_store::v3::serial_for_test();
+        let storage = tempfile::tempdir().unwrap();
+        let (identity, generation) = publish_test_base(storage.path(), &[("src/a.rs", b"alpha")]);
+        put_rows(storage.path(), &[b"new"]);
+        let (_, composed) = published(
+            bind(
+                storage.path(),
+                "worker-a",
+                &identity,
+                &generation,
+                worker_delta(&identity, &generation, b"new"),
+                None,
+            )
+            .unwrap(),
+        );
+        drop(composed);
+        for _ in 0..3 {
+            pass(storage.path());
+            assert!(registered(storage.path(), &identity.scope));
+            assert!(plane_has(storage.path(), b"alpha"));
+        }
+        // The worker is deregistered: its rows go with its member row.
+        with_base_barrier(storage.path(), FAMILY, |tx| {
+            tx.execute(
+                "UPDATE members SET state='evicting' WHERE scope='worker-a'",
+                [],
+            )?;
+            Ok(())
+        })
+        .unwrap();
+        pass(storage.path());
+        assert!(!registered(storage.path(), "worker-a"));
+        assert!(refs(storage.path()).is_empty());
+        // That same pass counted the base once; the next one reclaims it.
+        assert!(registered(storage.path(), &identity.scope));
+        pass(storage.path());
+        assert!(!registered(storage.path(), &identity.scope));
+        pass(storage.path());
+        assert!(!plane_has(storage.path(), b"alpha"));
+        assert!(!storage.path().join("views/v3").exists());
+    }
+}