perf(#190): the pooled reader's scope, measured on all three rows — plus the save-path attribution and the stride1 scaling curve

core/crates/layerfs-storage/src/cas/lifecycle.rs

@@ -80,6 +80,7 @@ impl MutationOwner {
             published_candidates,
             baseline_pack_id,
             next_pack_id: 0,
+            committed_pack_id: 0,
             ceiling: baseline_pack_id,
             placement: std::array::from_fn(|_| LanePlacement::new()),
             groups: std::array::from_fn(|_| PendingGroup::default()),
@@ -130,10 +131,31 @@ impl MutationOwner {
         self.transaction_open = true;
         self.transaction = TransactionState::default();
         self.next_pack_id = ownership::next_pack(&self.connection)?;
+        // The value this transaction starts from is the value the row holds, so a
+        // step that allocates no pack has nothing to write back.
+        self.committed_pack_id = self.next_pack_id;
         self.counters.transactions += 1;
         Ok(())
     }
 
+    /// Writes the pack watermark back only when this transaction moved it.
+    ///
+    /// The watermark is what stops a second writer handing out a pack id this save
+    /// already used, so it must be correct at every step boundary, not merely at
+    /// publication: deferring it to publication is exactly the change that would
+    /// let a second writer collide, and `tests/pack_watermark.rs` fails on it. It
+    /// moves only when `LanePlacement` starts a new pack, and this transaction
+    /// re-read the row when it began, so writing the unchanged value back is a
+    /// statement whose result the row already holds - one statement, and one
+    /// dirtied page on a commit that has nothing to do with pack allocation.
+    fn advance_pack_if_moved(&mut self) -> StorageResult<()> {
+        if self.next_pack_id != self.committed_pack_id {
+            ownership::advance_pack(&self.connection, self.next_pack_id)?;
+            self.committed_pack_id = self.next_pack_id;
+        }
+        Ok(())
+    }
+
     /// Acknowledges one physical group; no transaction survives preparation.
     pub fn maybe_commit(&mut self) -> StorageResult<()> {
         if self.transaction_open {
@@ -143,7 +165,7 @@ impl MutationOwner {
             // transaction therefore never outlives the step that opened it under the
             // arbitration lock, so every step commits before that lock is released.
             // Batching stays inside a step; it cannot span steps.
-            ownership::advance_pack(&self.connection, self.next_pack_id)?;
+            self.advance_pack_if_moved()?;
             write::commit(&self.connection)?;
             SaveProfile::charge(&mut self.profile.commit_ns, started);
             // Clear the flag before the next step re-acquires: if the lock is lost
@@ -229,7 +251,7 @@ impl MutationOwner {
         // writer can hand out a pack id this save already used. Either the save's
         // data, its index and its publication all become visible, or none does.
         ownership::publish(&self.connection, self.save_id)?;
-        ownership::advance_pack(&self.connection, self.next_pack_id)?;
+        self.advance_pack_if_moved()?;
         let started = Instant::now();
         write::commit(&self.connection)?;
         SaveProfile::charge(&mut self.profile.commit_ns, started);

core/crates/layerfs-storage/src/cas/owner.rs

@@ -272,6 +272,13 @@ pub struct MutationOwner {
     pub(super) capacities: StorageCapacities,
     pub(super) baseline_pack_id: i64,
     pub(super) next_pack_id: i64,
+    /// `store_policy.next_pack_id` as this connection last read or wrote it.
+    ///
+    /// The watermark only has to be written back when it moved: `begin_write`
+    /// re-reads it under the write lock, and only `LanePlacement` allocating a
+    /// pack moves it. Writing the unchanged value back is a statement whose
+    /// result the row already holds, on every step, for the whole operation.
+    pub(super) committed_pack_id: i64,
     /// Highest pack id this save created; contributes to the retained range on
     /// publication. Visibility is determined by the save row, not this ceiling.
     pub(super) ceiling: i64,

core/crates/layerfs-storage/src/cas/placement.rs

@@ -229,10 +229,16 @@ impl MutationOwner {
             self.counters.pack_appends += 1;
         }
         self.ceiling = self.ceiling.max(write.pack_id);
-        self.connection.execute(
-            "UPDATE saves SET pack_ceiling = MAX(pack_ceiling, ?2) WHERE save_id = ?1 AND active_slot IS NOT NULL",
-            [self.save_id, write.pack_id],
-        )?;
+        // The save's pack ceiling is read in exactly one place - `publish`, which
+        // folds it into the retained range in the transaction that publishes the
+        // save - and only a write that creates a pack can raise it. Re-asserting
+        // it on every append writes a value the row already holds.
+        if write.created {
+            self.connection.execute(
+                "UPDATE saves SET pack_ceiling = MAX(pack_ceiling, ?2) WHERE save_id = ?1 AND active_slot IS NOT NULL",
+                [self.save_id, write.pack_id],
+            )?;
+        }
         self.transaction.rows += 1;
         self.transaction.bytes += write.bytes.len() as u64;
         Ok(())

core/crates/layerfs-storage/src/sqlite/lookup.rs

@@ -77,14 +77,11 @@ pub(crate) fn candidates(
             "SELECT o.object_id,o.object_role,o.canonical_length,o.pack_id,o.group_number,o.record_number,o.save_id,\
              (o.save_id = r.save_id OR s.publication <= r.publication) \
              FROM objects o JOIN saves s USING(save_id),temp.layerfs_read_scope r \
-             WHERE o.object_id IN ({}) AND o.pack_id <= ?{} ORDER BY o.object_id,o.save_id LIMIT ?{}",
-            placeholders(page.len(),1),page.len()+1,page.len()+2,
+             WHERE o.object_id IN ({}) AND o.pack_id <= ?{}",
+            placeholders(page.len(),1),page.len()+1,
         );
         let mut parameters: Vec<Value> = page.iter().copied().map(id_value).collect();
         parameters.push(Value::Integer(ceiling));
-        parameters.push(Value::Integer(
-            (page.len() * super::ownership::SAVE_SLOTS + 1) as i64,
-        ));
         let mut statement = connection.prepare_cached(&sql)?;
         let mut rows = statement.query(rusqlite::params_from_iter(parameters))?;
         let mut counts = std::collections::BTreeMap::new();

core/crates/layerfs-storage/tests/pack_watermark.rs

@@ -0,0 +1,116 @@
+//! The pack-allocation watermark at a step boundary, and the save's own ceiling.
+//!
+//! The watermark is what stops two writers over one Store from being handed the
+//! same pack identifier, and it is read at `begin_write` under the write lock. A
+//! step that allocates no pack therefore has nothing to add to it, and a step that
+//! does must publish it before it releases the lock - deferring it to publication
+//! is the change that lets the second writer collide. These cases pin both halves
+//! from outside the crate: the row is read through an independent connection while
+//! the save is open, and the identifiers two interleaved writers actually wrote
+//! are compared.
+mod support;
+use layerfs_storage::Store;
+use support::{construct_file, create_store, disabled, noise, TempDir};
+
+/// The policy row through a connection that is not the save's.
+fn watermark(path: &std::path::Path) -> i64 {
+    let connection = rusqlite::Connection::open(path).expect("independent connection");
+    connection
+        .query_row(
+            "SELECT next_pack_id FROM store_policy WHERE id = 1",
+            [],
+            |row| row.get(0),
+        )
+        .expect("watermark")
+}
+
+/// Every pack id in the Store, with the save that wrote it.
+fn packs(path: &std::path::Path) -> Vec<(i64, i64)> {
+    let connection = rusqlite::Connection::open(path).expect("independent connection");
+    let mut statement = connection
+        .prepare("SELECT pack_id, save_id FROM object_packs ORDER BY pack_id")
+        .expect("pack query");
+    let rows: Vec<(i64, i64)> = statement
+        .query_map([], |row| Ok((row.get(0)?, row.get(1)?)))
+        .expect("rows")
+        .map(|row| row.expect("pack row"))
+        .collect();
+    rows
+}
+
+#[test]
+fn a_step_never_leaves_the_watermark_behind_a_pack_it_wrote() {
+    let temp = TempDir::new("watermark");
+    let path = temp.store_path("shared");
+    let store = create_store(&path);
+    let bytes = noise(900_000);
+    let (objects, root, _) = construct_file(&bytes);
+
+    assert_eq!(watermark(&path), 1, "a fresh Store hands out pack 1 first");
+    let mut save = disabled(|s| store.begin_save(s.child("save"))).unwrap();
+    for object in objects.finalized() {
+        save.accept(object).unwrap();
+        // Between steps the save holds no transaction, so the row is readable.
+        let mark = watermark(&path);
+        let highest = packs(&path)
+            .last()
+            .map(|(pack_id, _)| *pack_id)
+            .unwrap_or(0);
+        assert!(
+            mark > highest,
+            "watermark {mark} must be ahead of every committed pack id {highest}"
+        );
+    }
+    disabled(|s| save.finish(s.child("finish"))).unwrap();
+    let written = packs(&path);
+    assert!(!written.is_empty(), "the save wrote at least one pack");
+    assert!(
+        watermark(&path) > written.last().unwrap().0,
+        "the published watermark is ahead of every pack row"
+    );
+    assert_eq!(support::read_logical(&store, root), bytes);
+}
+
+#[test]
+fn two_writers_interleaved_between_steps_never_share_a_pack_identifier() {
+    let temp = TempDir::new("watermark-pair");
+    let path = temp.store_path("shared");
+    let store = create_store(&path);
+    let opened = disabled(|s| Store::open(&path, s.child("open"))).unwrap();
+    let bytes = noise(900_000);
+    let (objects, _, _) = construct_file(&bytes);
+
+    let mut a = disabled(|s| store.begin_save(s.child("a"))).unwrap();
+    let mut b = disabled(|s| opened.begin_save(s.child("b"))).unwrap();
+    let payload = objects.finalized();
+
+    // A steps first and allocates packs; B then begins between A's steps and
+    // allocates its own; A steps again and must not be handed B's identifiers.
+    for object in payload.iter().take(payload.len() / 3).cloned() {
+        a.accept(object).unwrap();
+    }
+    for object in payload.iter().cloned() {
+        b.accept(object).unwrap();
+    }
+    for object in payload.iter().skip(payload.len() / 3).cloned() {
+        a.accept(object).unwrap();
+    }
+    disabled(|s| b.finish(s.child("finish-b"))).unwrap();
+    disabled(|s| a.finish(s.child("finish-a"))).unwrap();
+
+    let written = packs(&path);
+    let mut seen = std::collections::BTreeSet::new();
+    for (pack_id, _) in &written {
+        assert!(seen.insert(*pack_id), "pack {pack_id} was written twice");
+    }
+    let a_saves: std::collections::BTreeSet<i64> = written
+        .iter()
+        .filter(|(pack_id, _)| *pack_id >= 1)
+        .map(|(_, save_id)| *save_id)
+        .collect();
+    assert_eq!(a_saves.len(), 2, "two saves wrote packs: {written:?}");
+    assert!(
+        watermark(&path) > written.last().unwrap().0,
+        "the watermark is ahead of every pack row"
+    );
+}