Buffer multi-dense offsets store to prevent reload corruption

[agourlay]

Fixes #9451 (comment)

Problem

The appendable multi-dense vector storage keeps two independently-flushed ChunkedVectors stores: vectors (flattened rows) and offsets (point key -> {offset, count, capacity}).

Each ChunkedVectors flusher snapshots its status.len at creation time but msyncs chunk bytes at execution time. When a re-upsert grows a point past its reserved capacity, the append path relocates the point's rows and rewrites its offsets entry in place to point at the freshly-appended region. If that rewrite lands inside a flush's creation-to-execution window, the relocated offset entry becomes durable while the vectors store's recorded length still predates the rows it references.

The result is a durable forward reference: on reload that point is unreadable, and a subsequent WAL append reuses the same row region, clobbering the head rows of another point. This surfaced as multi-dense reload divergence in the collection model test, which is why the m vector was disabled there.

Fix

Wrap the offsets store in a write-back buffer (BufferedOffsets) so the durable offsets can never reference rows beyond the durable vectors length:

• Offset writes stage in a pending overlay (set) and only land in the durable store while a flush executes.
• The flusher snapshots the pending set at creation time, so any write after that stays buffered for the next flush.
• Both the vectors flusher and the offsets flusher snapshot at the same instant (the storage flusher() call), yielding a consistent durable cut. Rows written after the cut are unreferenced garbage the next append harmlessly overwrites.

This prevents the skew at the source rather than repairing an inconsistent state on reload, and as a side effect closes the residual offsets-length smear the storage had on its own.

Convention

BufferedOffsets follows the Gridstore flusher convention (lib/gridstore/src/gridstore/mod.rs, tracker/mod.rs):

• Pending writes live inside the single lock-guarded store, so reads consult the overlay then the durable bytes under one lock (mirrors Tracker::get).
• The flusher applies + reconciles pending and builds the backing flusher under the write lock, then runs the durable msync after releasing the lock, so the sync never stalls concurrent reads (mirrors flush_tracker).
• Batch reads resolve all offsets under a single read lock (resolve_rows, mirroring Tracker::iter).

One deliberate divergence: on a cancelled flush (storage dropped) we return Ok(()) rather than Gridstore's Err(FlushCancelled), since OperationError has no cancellation variant and a dropped storage has nothing to persist.

Testing

• New unit tests cover the corrupting in-window relocation at both the buffer level (post_snapshot_write_does_not_leak_into_durable_store, deferred_write_commits_on_next_flush) and the storage level (relocation_after_flush_start_is_deferred_not_corrupt).
• Re-enabled the m multivector in the collection model test (INITIAL_ACTIVE + the CreateVectorName op filter); model validation passed.

Notes for reviewers

Reads now go through a parking_lot::RwLock overlay instead of straight mmap. The hot batch paths resolve under a single lock per batch, but this is still one lock acquisition that did not exist before. The previous shared read_only::iter_vectors offsets-side prefetch is dropped for the appendable variant (rows-side prefetch is retained). This is the cost tradeoff versus a reload-time repair; worth a benchmark if the read path is sensitive.

🤖 Generated with Claude Code

[timvisee]

I was a little bit confused by the nesting. Hope this comment change is fine, as it clarified it better for me:

Suggested change

	// Apply + reconcile + build the backing flusher under the write lock,
	// then release the lock and run the durable sync — Gridstore's
	// `flush_tracker` likewise never holds the store lock across the msync,
	// so concurrent reads are not stalled by the sync.
	//
	// The backing flusher runs on every flush, even with an empty snapshot:
	// the underlying mmap sync only writes still-dirty pages, so a sync that
	// failed on an earlier flush is retried on the next one. This matches
	// Gridstore (its tracker store is synced unconditionally) and the sibling
	// `vectors` store; short-circuiting an empty snapshot would instead leave
	// a failed offsets sync stranded while `vectors` recovered, reopening the
	// very skew this buffer closes.
	let store_flusher = {
	let mut inner = inner.write();
	// Apply in ascending key order so the backing store grows its
	// chunks/length monotonically (appends sit at the current end).
	// Dense key allocation (`new_id = len`) guarantees no gaps: any
	// append key in the snapshot has every lower key already durable
	// or applied in this same pass.
	let mut items: Vec<(VectorOffsetType, MultivectorMmapOffset)> =
	snapshot.iter().map(|(key, entry)| (*key, *entry)).collect();
	items.sort_unstable_by_key(|(key, _)| *key);
	let hw_counter = HardwareCounterCell::disposable();
	for (key, entry) in &items {
	inner.store.insert(*key, &[*entry], &hw_counter)?;
	}
	// Drop every key we just persisted from the live pending set, unless
	// it was overwritten with a newer value after the snapshot was taken.
	inner.pending.retain(|key, value| {
	snapshot.get(key).is_none_or(|persisted| persisted != value)
	});
	// Built here so its `status.len` snapshot covers the entries just
	// applied; executed below, outside the lock.
	inner.store.flusher()
	};
	store_flusher()
	// Apply + reconcile + build the backing flusher under the write lock,
	// then release the lock and run the durable sync — Gridstore's
	// `flush_tracker` likewise never holds the store lock across the msync,
	// so concurrent reads are not stalled by the sync.
	//
	// The backing flusher runs on every flush, even with an empty snapshot:
	// the underlying mmap sync only writes still-dirty pages, so a sync that
	// failed on an earlier flush is retried on the next one. This matches
	// Gridstore (its tracker store is synced unconditionally) and the sibling
	// `vectors` store; short-circuiting an empty snapshot would instead leave
	// a failed offsets sync stranded while `vectors` recovered, reopening the
	// very skew this buffer closes.
	let store_flusher = {
	// Apply in ascending key order so the backing store grows its
	// chunks/length monotonically (appends sit at the current end).
	// Dense key allocation (`new_id = len`) guarantees no gaps: any
	// append key in the snapshot has every lower key already durable
	// or applied in this same pass.
	let mut items: Vec<(VectorOffsetType, MultivectorMmapOffset)> =
	snapshot.iter().map(|(key, entry)| (*key, *entry)).collect();
	items.sort_unstable_by_key(|(key, _)| *key);
	let mut inner = inner.write();
	let hw_counter = HardwareCounterCell::disposable();
	for (key, entry) in &items {
	inner.store.insert(*key, &[*entry], &hw_counter)?;
	}
	// Drop every key we just persisted from the live pending set, unless
	// it was overwritten with a newer value after the snapshot was taken.
	inner.pending.retain(|key, value| {
	snapshot.get(key).is_none_or(|persisted| persisted != value)
	});
	// Create store flusher and drop inner guard before we actually flush
	inner.store.flusher()
	};
	store_flusher()

I also moved the inner.write() lock one line down. It is fine to lock it just when we need it, not before.

[agourlay]

Applied in 1e473bf — moved the snapshot build/sort above the write lock and took your comment wording. Thanks, agreed it reads clearer and shortens the lock hold.