diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 55a7157..9fb4490 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -8,7 +8,7 @@ jobs: test: strategy: matrix: - os: [ubuntu-latest, macos-latest] + os: [ubuntu-latest, macos-latest, windows-latest] runs-on: ${{ matrix.os }} timeout-minutes: 20 steps: diff --git a/Cargo.lock b/Cargo.lock index e7a44ca..91316d1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -46,6 +46,7 @@ dependencies = [ name = "hnsw-plane" version = "0.0.1" dependencies = [ + "libc", "memmap2", "napi", "napi-build", diff --git a/DESIGN.md b/DESIGN.md index cda417d..66dff99 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -96,18 +96,24 @@ One file per index (per slice, once C2 lands): `.hnsw`. | freelist_head | u64 atomic | CAS push/pop; ABA-guarded with a 32-bit tag | | txn_watermark | u64 | last durably indexed transaction; advanced by msync cadence | | clean_shutdown flag | u8 | torn-state detection on open | +| invalidated latch | u8 | one-way (v7): watermark reads 0 on every handle, open refuses | +| write_epoch | u64 atomic | bumped by every node write; re-arms the read-side repair probe | **Main region — layer-0 slots**, addressed `4096 + id × slot_size`: -| Field | Size (768-d int8, cap 64) | -| ------------------------------- | ---------------------------------- | -| seq (seqlock) | 4 B | -| flags (valid/deleted) + level | 2 B | -| scale (f32) + invMag (f32) | 8 B | -| degree | 2 B | -| vector (int8 × 768) | 768 B | -| neighbor ids (u32 × layer0_cap) | 256 B | -| **total, padded** | **1,040 B → 1 KB-aligned 1,088 B** | +| Field | Size (768-d int8, cap 64) | +| ------------------------------- | ----------------------------------- | +| seq (seqlock) | 4 B | +| flags (valid/deleted) + level | 2 B | +| scale (f32) + invMag (f32) | 8 B | +| degree | 2 B | +| vector (int8 × 768) | 768 B (padded to a 4-byte boundary) | +| neighbor ids (u32 × layer0_cap) | 256 B | +| **total, padded** | **1,040 B → 1 KB-aligned 1,088 B** | + +The vector's trailing pad keeps the neighbor array 4-aligned for every `dims`, so the search +hot path reads each neighbor id as one aligned volatile `u32`. Upper-layer id lists are padded +the same way (`degree u16 + pad u16 + ids`). At 100M nodes: ~109 GB (int8). A binary-code v2 slot (96 B codes + ids) is ~384 B → ~38 GB. For comparison, today's encoding averages 1,425 B/node _plus_ RocksDB overhead — so v1 is @@ -167,6 +173,24 @@ bounded-lag durability with deterministic catch-up. For an approximate index who truth (records + pk→nodeId) remains fully transactional, bounded lag is the right trade — it buys the entire performance model. +**Invalidation (a plane the host cannot delete).** Disabling a plane deletes its file; when the +unlink fails (Windows sharing violation while another process maps it) the file must not be +adopted later at its nonzero watermark, or it silently serves searches missing every mutation +made while mirroring was off. `invalidate_plane(path)` / `invalidate_file(&handle)` leave two +markers, always attempting both: in band — `PlaneFile::invalidate` sets a one-way header latch, +zeroes the watermark, and msyncs the header page alone (a whole-mapping flush cannot run inline +on a multi-GB plane, and lowering the watermark is the safe direction) — then a `.stale` +sidecar, created with create-new semantics (a planted symlink is never followed) and fsync'd +together with its directory entry (the directory fsync is skipped on Windows, where `std` has no +directory handle and `FlushFileBuffers` on the marker covers its creation). The package enforces +both markers: `open` refuses a file carrying either, `create` refuses a path with a leftover +sidecar, and `watermark()` reads 0 on every handle while the latch is set — so a flush already +in flight on another handle, which still stamps the word, cannot revive the plane. In band +first: the sidecar is what a process that cannot map the file checks, the latch is what covers a +plane whose sidecar a crash lost. A temporary handle opened for the in-band mark is unmapped +and closed before the sidecar step — its own mapping would keep the file undeletable — and the +call fails only when neither marker is durable, leaving the file exactly as found. + **Backup/copy-db/reseed:** the file is node-local derived state. Backup either includes it (consistent-enough after an msync barrier) or marks the index rebuild-on-restore. Replica reseed = rebuild from records (C5 bulk construction makes this fast; until then, the existing @@ -272,6 +296,14 @@ Decided (Kris, 2026-08-31): Open: +- **Atomic slot payloads.** Fields a concurrent reader acts on (flags, level, degree, scale, + invMag, neighbor and upper ids) are read through aligned `read_volatile`, which forbids the + reload/split/sink across the seqlock's validating fence that `lto = true, codegen-units = 1` + otherwise licenses. That is not the same as being race-free under Rust's memory model: only + making those fields `AtomicU8`/`AtomicU16`/`AtomicU32` in the slot layout would be, and that + is a format change deferred past phase 1. The stored vector stays an ordinary load on + purpose — `cosine_int8_raw` must keep autovectorizing, and a torn vector only perturbs a + distance the generation check discards. - **msync cadence default** — bounded-lag durability window vs write amplification; needs a workload measurement, not a guess. - **f32 (quantization:"none") slot variant** — 3,072 B vectors → 3.4 KB slots; supported by the diff --git a/README.md b/README.md index 7e15246..72ab1cd 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,13 @@ const predicated = await plane.searchWithPredicate(queryVector, 10, 512, (ids) = ); ``` +A plane is derived state; when the host must stop maintaining one and cannot delete the file +(Windows sharing violations while another process maps it), `invalidatePlane(path)` — or +`plane.invalidateFile()` through a handle the host already holds — durably marks it +unadoptable: a one-way in-band latch (watermark reads 0, `Plane.open` refuses) plus a fsync'd +`.stale` sidecar (`stalePathFor(path)`, which `open` also refuses). It throws only when +neither marker lands. Hosts delete both files and rebuild. + Full API in [index.d.ts](index.d.ts). ## Benchmarks @@ -81,8 +88,8 @@ equal recall. ## Status -Extracted from the Harper vector-index engine; the format (v2) and API are young and may -change with a major version + reindex. Roadmap: prebuilds, binary-quantized slot format +Extracted from the Harper vector-index engine; the format (v7) and API are young and may +change with a version bump + reindex (an older format version fails to open; rebuild). Roadmap: prebuilds, binary-quantized slot format (~4× smaller traversal plane), Matryoshka dimension truncation, mremap growth, index slicing with native top-k merge. diff --git a/index.d.ts b/index.d.ts index b42ed11..5180b85 100644 --- a/index.d.ts +++ b/index.d.ts @@ -12,7 +12,10 @@ export interface SearchHit { export declare class Plane { /** Create a new plane file. `maxNodes` is a sparse reservation — pages materialize on write. */ static create(path: string, dims: number, layer0Cap: number, maxNodes: number): Plane; - /** Open an existing plane file (format-version mismatch throws: rebuild the index). */ + /** + * Open an existing plane file. Throws on a format-version mismatch and on an invalidated + * plane (header latch or `.stale` sidecar): delete the file and its sidecar, rebuild. + */ static open(path: string): Plane; /** @@ -101,4 +104,42 @@ export declare class Plane { flush(watermark?: number): void; /** flush() on the libuv thread pool — a whole-map msync can stall its calling thread. */ flushAsync(watermark?: number): Promise; + /** + * In-band half of invalidateFile() only — no sidecar, so a process that cannot map the + * file sees nothing; prefer invalidateFile(). Sets the one-way header latch, zeroes the + * watermark, msyncs the header page (a 4 KB barrier, not a whole-mapping flush). From then + * on every handle reads watermark 0, whatever a racing flush stamps, and open() throws. + */ + invalidate(): void; + /** + * invalidatePlane() through this handle: the in-band mark via this mapping (no second open, + * no second registry slot — on Windows this mapping is why the unlink failed) and the + * `.stale` sidecar next to the path it opened. The path must not have been replaced since. + */ + invalidateFile(): InvalidationOutcome; + /** Whether the plane was invalidated, by any handle, since this one opened. */ + invalidated(): boolean; } + +export interface InvalidationOutcome { + /** The watermark was zeroed and its header page msync'd. */ + inBand: boolean; + /** `.stale` exists and is fsync'd (on POSIX, so is its directory entry). */ + sidecar: boolean; + inBandError?: string; + sidecarError?: string; +} + +/** + * Make a plane file that could not be deleted unadoptable, durably, through a temporary + * handle that is unmapped and closed before this returns. Both markers are always attempted: + * the in-band latch and the fsync'd `.stale` sidecar; open() refuses a file carrying either. + * Throws only when neither marker became durable; nothing is deleted or renamed, and an + * in-band mark whose msync failed may still have landed in the shared mapping (the safe + * direction: it reads as incomplete). Idempotent. Synchronous (three small fsyncs on a cold path). + */ +export declare function invalidatePlane(path: string): InvalidationOutcome; +/** invalidatePlane() on the libuv thread pool. */ +export declare function invalidatePlaneAsync(path: string): Promise; +/** The sidecar convention: `.stale`. */ +export declare function stalePathFor(path: string): string; diff --git a/package.json b/package.json index 09ccee8..455d581 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@harperfast/hnsw", - "version": "0.1.0", + "version": "0.2.0", "description": "Persistent, incrementally-maintained, concurrently-searchable native HNSW for Node.js: a memory-mapped fixed-slot graph file with off-event-loop search, seqlock concurrency, int8 asymmetric distance, and bitset/predicate filtering.", "license": "Apache-2.0", "repository": { @@ -41,9 +41,9 @@ "rust" ], "optionalDependencies": { - "@harperfast/hnsw-darwin-arm64": "0.1.0", - "@harperfast/hnsw-linux-arm64-glibc": "0.1.0", - "@harperfast/hnsw-linux-x64-glibc": "0.1.0", - "@harperfast/hnsw-win32-x64": "0.1.0" + "@harperfast/hnsw-darwin-arm64": "0.2.0", + "@harperfast/hnsw-linux-arm64-glibc": "0.2.0", + "@harperfast/hnsw-linux-x64-glibc": "0.2.0", + "@harperfast/hnsw-win32-x64": "0.2.0" } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml new file mode 100644 index 0000000..9b60ae1 --- /dev/null +++ b/pnpm-lock.yaml @@ -0,0 +1,9 @@ +lockfileVersion: '9.0' + +settings: + autoInstallPeers: true + excludeLinksFromLockfile: false + +importers: + + .: {} diff --git a/smoke.mjs b/smoke.mjs index cf7aa13..49f0bff 100644 --- a/smoke.mjs +++ b/smoke.mjs @@ -1,7 +1,7 @@ // End-to-end smoke test: `npm run build && node smoke.mjs` (also the CI path). import { createRequire } from 'module'; const require = createRequire(import.meta.url); -const { Plane } = require('./index.js'); +const { Plane, invalidatePlane, invalidatePlaneAsync, stalePathFor } = require('./index.js'); const dims = 64; const { tmpdir } = await import('node:os'); @@ -81,4 +81,43 @@ plane.flush(); const reopened = Plane.open(path); const hits2 = reopened.searchSync(vec(42), 5, 128); if (hits2[0].distance > 1e-3) throw new Error('reopened self-query failed'); -console.log('reopen + sidecar OK. smoke PASSED'); +console.log('reopen OK'); + +// invalidation through the caller's own handle: both markers land, the latch survives a +// later flush, and every later open is refused +const { existsSync, mkdirSync, rmSync } = await import('node:fs'); +reopened.setWatermark(4096); +reopened.flush(); +const outcome = reopened.invalidateFile(); +if (!outcome.inBand || !outcome.sidecar) throw new Error(`invalidation incomplete: ${JSON.stringify(outcome)}`); +if (stalePathFor(path) !== `${path}.stale` || !existsSync(stalePathFor(path))) throw new Error('no .stale sidecar'); +reopened.flush(900); +if (reopened.getWatermark() !== 0 || !reopened.invalidated()) throw new Error('a later flush revived the plane'); +rmSync(stalePathFor(path)); +let refused; +try { + Plane.open(path); +} catch (error) { + refused = error; +} +if (!refused || !/invalidated/.test(refused.message)) throw new Error(`open must refuse an invalidated plane, got ${refused}`); +// by path: a temporary open that must not survive the call (idempotent on a latched plane) +const byPath = invalidatePlane(path); +if (!byPath.inBand || !byPath.sidecar) throw new Error(`path invalidation incomplete: ${JSON.stringify(byPath)}`); +const byPathAsync = await invalidatePlaneAsync(path); +if (!byPathAsync.inBand || !byPathAsync.sidecar) throw new Error(`async path invalidation incomplete: ${JSON.stringify(byPathAsync)}`); +rmSync(stalePathFor(path)); +// neither marker possible: not a plane, and a directory squatting the sidecar path +const bogus = join(tmpdir(), `smoke-bogus-${process.pid}.hnsw`); +const { writeFileSync } = await import('node:fs'); +writeFileSync(bogus, 'not a plane'); +mkdirSync(stalePathFor(bogus)); +let threw; +try { + invalidatePlane(bogus); +} catch (error) { + threw = error; +} +if (!threw || !/in-band:.*sidecar:/.test(threw.message)) throw new Error(`double failure must throw naming both causes, got ${threw}`); +rmSync(stalePathFor(bogus), { recursive: true }); +console.log('invalidatePlane OK. smoke PASSED'); diff --git a/src/bin/bench.rs b/src/bin/bench.rs index 86fd7c4..f3fe4d2 100644 --- a/src/bin/bench.rs +++ b/src/bin/bench.rs @@ -122,7 +122,7 @@ fn main() { let build_start = Instant::now(); for i in 0..n { let v = corpus.row(&mut rng); - insert(&graph, &v, ¶ms, &mut scratch); + insert(&graph, &v, ¶ms, &mut scratch).expect("build insert"); if (i + 1) % 50_000 == 0 { let rate = (i + 1) as f64 / build_start.elapsed().as_secs_f64(); println!(" built {} ({:.0} inserts/s)", i + 1, rate); diff --git a/src/format.rs b/src/format.rs index 2457946..7101657 100644 --- a/src/format.rs +++ b/src/format.rs @@ -1,14 +1,14 @@ //! On-disk format: 4 KB header + fixed-size layer-0 slot array + upper-layer region. -//! See ../../../hnsw-native-plane.md §4. Format changes bump VERSION and require reindex. +//! See ../DESIGN.md §4. Format changes bump VERSION and require reindex. use memmap2::MmapMut; use std::fs::OpenOptions; use std::io; -use std::path::Path; -use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU32, AtomicU64, AtomicU8, Ordering}; pub const MAGIC: u32 = 0x484e_5357; // "HNSW" -pub const VERSION: u32 = 5; // v5: opener registry + aligned freelist pointer (older files: reindex) +pub const VERSION: u32 = 7; // v7: sticky invalidation latch; v6: 4-aligned neighbor + upper id arrays (older files: reindex) pub const HEADER_SIZE: usize = 4096; // Header field byte offsets. @@ -23,6 +23,12 @@ const H_ID_HIGH_WATER: usize = 32; // u64 atomic const H_FREELIST_HEAD: usize = 40; // u64 atomic: (tag << 32) | id; id u32::MAX = empty const H_TXN_WATERMARK: usize = 48; // u64 const H_CLEAN_SHUTDOWN: usize = 56; // u8 +// One-way: set by invalidate(), cleared by nothing. While set the watermark reads 0 on every +// handle whatever a racing flush stamps into it, and open() refuses the file. +const H_INVALIDATED: usize = 57; // u8 +// Bumped by every node write through any handle: the search-side repair probe's evidence +// that a fully dead graph may have gained a live node since it last came back empty. +const H_WRITE_EPOCH: usize = 96; // u64 atomic const H_MAX_NODES: usize = 64; // u64 const H_UPPER_HIGH_WATER: usize = 72; // u64 atomic: upper-entry allocator const H_UPPER_FREELIST: usize = 80; // u64 atomic: (tag<<32)|idx; NO_UPPER = empty @@ -39,11 +45,15 @@ pub const REGISTRY_SLOTS: usize = 64; /// 1/8 of max_nodes (2x headroom). P(level >= 9) at mL = 1/ln16 is ~e^-25 — unreachable. pub const MAX_UPPER_LEVELS: usize = 8; pub const UPPER_CAP: usize = 64; // matches the JS graph's upper cap (M<<2 under optimizeRouting) -// entry: seq u32 | levels u8 | pad | per-level (degree u16 + ids u32*UPPER_CAP) +// entry: seq u32 | levels u8 | pad | per-level (degree u16 + pad u16 + ids u32*UPPER_CAP) pub const U_SEQ: usize = 0; pub const U_LEVELS: usize = 4; pub const U_LISTS: usize = 8; -pub const UPPER_LEVEL_STRIDE: usize = 2 + UPPER_CAP * 4 + 2; // degree + ids + pad -> 132 +/// The pad follows the degree rather than the ids so every id array starts 4-aligned; the +/// stride (and so the entry size) is unchanged either way. +pub const UL_DEGREE: usize = 0; +pub const UL_IDS: usize = 4; +pub const UPPER_LEVEL_STRIDE: usize = UL_IDS + UPPER_CAP * 4; pub const NO_UPPER: u32 = u32::MAX; // Slot layout offsets (within a slot). @@ -55,8 +65,15 @@ pub const S_SCALE: usize = 8; // f32 pub const S_INV_MAG: usize = 12; // f32 pub const S_UPPER_IDX: usize = 16; // u32 index into the upper region; NO_UPPER = none pub const S_VECTOR: usize = 20; // dims bytes (int8) or dims*4 (f32) - // neighbors: u32 * layer0_cap, follows vector - // deleted slots reuse the first neighbor word as freelist next-pointer + // neighbors: u32 * layer0_cap, follows the 4-padded vector + +/// Byte offset of a slot's neighbor array. The vector is padded to a 4-byte boundary so this +/// is 4-aligned for every dims: the search hot path then reads each neighbor as one aligned +/// volatile u32 instead of four byte loads plus shifts. +#[inline] +pub const fn neighbor_offset(dims: usize) -> usize { + S_VECTOR + (dims + 3) / 4 * 4 +} pub const FLAG_VALID: u8 = 1; pub const FLAG_DELETED: u8 = 2; @@ -65,6 +82,9 @@ pub const NO_ID: u32 = u32::MAX; pub struct PlaneFile { /// Kept open for the lifetime of the mapping: the opener-registry OFD lock lives on it. file: std::fs::File, + /// The path this handle opened or created, as given; the sidecar of `invalidate_file` is + /// placed next to it. + pub path: PathBuf, /// This handle's registry tag (low bits encode its registry slot). 0 = unregistered /// (registry full or platform without OFD locks): this handle's own dead locks cannot be /// reclaimed by others, and it never reclaims. @@ -77,8 +97,9 @@ pub struct PlaneFile { upper_offset: usize, pub upper_capacity: u64, /// Whether the file recorded a clean shutdown when opened (create() reports true). - /// An unclean open has had its torn seqlocks scrubbed, but individual slots may hold - /// unflushed/partial states — hosts should rebuild rather than trust completeness. + /// Advisory only: open() performs no repair — torn seqlocks are taken over lazily at + /// their slot (seqlock.rs) — and slots may hold unflushed states; hosts rebuild rather + /// than trust completeness. pub opened_clean: bool, /// Slots per 4 KB page under page-grouped addressing; 0 = packed (slots may straddle /// pages). Grouped is chosen at create when the per-page waste is small (e.g. 1,344 B @@ -101,8 +122,21 @@ fn advise_random(map: &MmapMut) { let _ = map; } +fn stale_sidecar_present(path: &Path) -> bool { + // any entry counts, a directory or dangling link included, and so does any stat failure + // other than absence: a marker that fails closed cannot be defeated by a transient EIO + match std::fs::symlink_metadata(crate::invalidate::stale_path_for(path)) { + Ok(_) => true, + Err(e) => e.kind() != io::ErrorKind::NotFound, + } +} + +fn invalidated_error(path: &Path) -> io::Error { + io::Error::new(io::ErrorKind::InvalidData, format!("{} was invalidated: delete it and its .stale sidecar, then rebuild the index", path.display())) +} + fn slot_size_for(dims: usize, layer0_cap: usize) -> usize { - let raw = S_VECTOR + dims + layer0_cap * 4; + let raw = neighbor_offset(dims) + layer0_cap * 4; raw.next_multiple_of(64) // cache-line align } @@ -134,6 +168,11 @@ impl PlaneFile { if max_nodes >= NO_ID as u64 { return Err(io::Error::new(io::ErrorKind::InvalidInput, "maxNodes must be below 2^32-1")); } + if stale_sidecar_present(path) { + // a leftover sidecar would make the new file unopenable forever; the host must + // clear it deliberately + return Err(io::Error::other(format!("{} has a stale sidecar: remove {} before creating", path.display(), crate::invalidate::stale_path_for(path).display()))); + } let slot_size = slot_size_for(dims, layer0_cap); let slots_per_page = slots_per_page_for(slot_size); let data_len = slot_region_len(max_nodes, slot_size, slots_per_page); @@ -156,12 +195,16 @@ impl PlaneFile { .copy_from_slice(&((NO_ID as u64) | 0u64 << 32).to_le_bytes()); map[H_MAX_NODES..H_MAX_NODES + 8].copy_from_slice(&max_nodes.to_le_bytes()); map[H_UPPER_FREELIST..H_UPPER_FREELIST + 8].copy_from_slice(&(NO_UPPER as u64).to_le_bytes()); + // zero would read as "node 0 was the previous entry point" and hand every re-election + // and search-side repair a candidate that was never an entry point + map[H_ENTRY_PREV..H_ENTRY_PREV + 8].copy_from_slice(&(NO_ID as u64).to_le_bytes()); map[H_VERSION..H_VERSION + 4].copy_from_slice(&VERSION.to_le_bytes()); std::sync::atomic::fence(Ordering::Release); map[H_MAGIC..H_MAGIC + 4].copy_from_slice(&MAGIC.to_le_bytes()); let upper_offset = HEADER_SIZE + slot_region_len(max_nodes, slot_size, slots_per_page) as usize; let mut plane = PlaneFile { file, + path: path.to_path_buf(), self_tag: 0, map, dims, @@ -174,10 +217,36 @@ impl PlaneFile { opened_clean: true, }; plane.register_opener(); + if stale_sidecar_present(path) { + // best-effort: an invalidation that raced the create (its in-band leg found no + // header yet, or was overwritten by ours) left only the sidecar. Latch the finished + // file so losing that sidecar cannot make this failed create adoptable. A sidecar + // landing after this check is caught by the next open, not by this handle. + let _ = plane.invalidate(); + return Err(io::Error::other(format!("{} gained a stale sidecar during create: remove {} and rebuild", path.display(), crate::invalidate::stale_path_for(path).display()))); + } Ok(plane) } + /// Open an existing plane. Refuses one that was invalidated — by its header latch or by a + /// `.stale` sidecar — so a stale mirror is never adopted by any package consumer; + /// the host deletes both files and rebuilds. pub fn open(path: &Path) -> io::Result { + if stale_sidecar_present(path) { + return Err(invalidated_error(path)); + } + let plane = Self::open_for_invalidation(path)?; + // the pre-map check is only half the refusal: a sidecar landed by an invalidation + // whose in-band leg failed (no latch to see) can appear between the check and the map + if plane.invalidated() || stale_sidecar_present(path) { + return Err(invalidated_error(path)); + } + Ok(plane) + } + + /// `open` without the invalidation refusals: the handle `invalidate_plane` marks through, + /// which must reach an already-invalidated file so a repeated invalidation is idempotent. + pub(crate) fn open_for_invalidation(path: &Path) -> io::Result { let file = OpenOptions::new().read(true).write(true).open(path)?; let file_len = file.metadata()?.len(); if file_len < HEADER_SIZE as u64 { @@ -220,6 +289,7 @@ impl PlaneFile { let opened_clean = map[H_CLEAN_SHUTDOWN] == 1; let mut plane = PlaneFile { file, + path: path.to_path_buf(), self_tag: 0, map, dims, @@ -243,8 +313,6 @@ impl PlaneFile { Ok(plane) } - /// Force any persisted-odd seqlocks (slot + upper regions) back to even after an unclean - /// shutdown. Safe because open() runs before any concurrent access exists. #[inline] pub fn slot_ptr(&self, id: u32) -> *const u8 { let off = if self.slots_per_page > 0 { @@ -295,8 +363,9 @@ impl PlaneFile { let _ = head.compare_exchange(cur, NO_ID as u64, Ordering::AcqRel, Ordering::Acquire); continue; } - // next-pointer lives in the dead slot's scale field: offset 8, aligned for any - // dims (the first neighbor word at S_VECTOR+dims is 4-aligned only when dims%4==0) + // next-pointer lives in the dead slot's scale field rather than its first neighbor + // word: the neighbor array is a live reader's aligned volatile load target, and a + // freelist pointer parked there would be decoded as a neighbor id let next = unsafe { (*(self.slot_ptr(id).add(S_SCALE) as *const AtomicU32)).load(Ordering::Acquire) }; let tag = (cur >> 32).wrapping_add(1); let new = (next as u64) | (tag << 32); @@ -353,25 +422,112 @@ impl PlaneFile { pub fn set_entry_point(&self, id: u32, level: u32) { let prev = self.header_atomic_u64(H_ENTRY).swap((id as u64) | ((level as u64) << 32), Ordering::AcqRel); - if (prev & 0xffff_ffff) as u32 != NO_ID && (prev & 0xffff_ffff) as u32 != id { - self.header_atomic_u64(H_ENTRY_PREV).store(prev, Ordering::Release); + self.record_previous_entry(prev, id); + } + + /// Remember the entry point a PROMOTION displaced. Only promotions are recorded: the node + /// they displace was live and high-level, which is what makes it a usable hint. Recording + /// a re-election's replacement instead would fill the hint with the dead node that forced + /// the re-election. + #[inline] + fn record_previous_entry(&self, prev_packed: u64, new_id: u32) { + let prev_id = (prev_packed & 0xffff_ffff) as u32; + if prev_id == NO_ID || prev_id == new_id || (prev_id as u64) >= self.max_nodes { + return; + } + // a hint is only worth keeping while its node is live: the host mirrors a post-delete + // re-election through this same call, and storing the node that died would evict a + // usable hint with one the repair path can never follow + // volatile like every other read of a field a concurrent writer mutates (graph.rs's + // `vread`): this one is outside the slot seqlock, so the retry cannot even catch a tear + if unsafe { self.slot_ptr(prev_id).add(S_FLAGS).read_volatile() } != FLAG_VALID { + return; } + self.header_atomic_u64(H_ENTRY_PREV).store(prev_packed, Ordering::Release); + } + + /// Claim the entry point of an EMPTY graph: a strict compare-exchange from the empty + /// encoding, so exactly one racer wins. `set_entry_point_if_not_better` cannot serve here — + /// it is a not-worse install, so a second first-inserter would replace the winner with its + /// own edgeless node and orphan everything already rooted at the winner. A loser must join + /// the winner's graph instead of returning an unlinked node. + pub fn claim_entry_if_empty(&self, id: u32, level: u32) -> bool { + self.header_atomic_u64(H_ENTRY) + .compare_exchange(NO_ID as u64, (id as u64) | ((level as u64) << 32), Ordering::AcqRel, Ordering::Acquire) + .is_ok() } /// Entry-point CAS for re-election: install (id, level) only while the current entry is /// still `expected_id` or is of a lower level — a concurrent insert that just promoted a /// higher-level entry must not be clobbered by a delete's level-0 survivor. pub fn set_entry_point_if_not_better(&self, id: u32, level: u32, expected_id: u32) { + self.cas_entry_if_not_better(id, level, expected_id, false); + } + + /// The same CAS for an insert that PROMOTED itself above the entry it observed: the + /// displaced entry is live, so it is recorded as the previous-entry hint that re-election + /// and the search-side repair both consult before any O(high-water) scan. + pub fn promote_entry_point(&self, id: u32, level: u32, expected_id: u32) { + self.cas_entry_if_not_better(id, level, expected_id, true); + } + + fn cas_entry_if_not_better(&self, id: u32, level: u32, expected_id: u32, record_prev: bool) { let cell = self.header_atomic_u64(H_ENTRY); let new = (id as u64) | ((level as u64) << 32); let mut cur = cell.load(Ordering::Acquire); loop { let cur_id = (cur & 0xffff_ffff) as u32; let cur_level = (cur >> 32) as u32; - if cur_id != expected_id && cur_id != NO_ID && cur_level > level { - return; // someone installed a better entry meanwhile + // `>=`, not `>`: an equal-level entry installed meanwhile may be a fresh + // `claim_entry_if_empty` winner with no in-edges yet; displacing it orphans that + // node, and an equal-level swap gains nothing + if cur_id != expected_id && cur_id != NO_ID && cur_level >= level { + return; // someone installed a not-worse entry meanwhile } match cell.compare_exchange(cur, new, Ordering::AcqRel, Ordering::Acquire) { + Ok(_) => { + if record_prev { + self.record_previous_entry(cur, id); + } + return; + } + Err(now) => cur = now, + } + } + } + + /// Install `(id, level)` ONLY while the entry still names `expected_id`. The read-side + /// repair publishes through this rather than `set_entry_point_if_not_better`: the entry it + /// is replacing is dead, so "not worse" is the wrong test — a level-0 root installed while + /// the repair ran would lose to a higher-level candidate and be orphaned. + /// + /// It compares the id, not the incarnation, so under the crate's own freelist reuse it can + /// match a different node that took the same slot. That is a routing-quality window, not a + /// lost node: the value it could displace is a live edged node, never the edgeless claimer + /// (`claim_entry_if_empty` fires only from NO_ID, which no reuse can produce). Harper's host + /// ids are monotonic and never reused, so this cannot arise there at all. + pub fn replace_entry_if(&self, expected_id: u32, id: u32, level: u32) -> bool { + let cell = self.header_atomic_u64(H_ENTRY); + let new = (id as u64) | ((level as u64) << 32); + let mut cur = cell.load(Ordering::Acquire); + while (cur & 0xffff_ffff) as u32 == expected_id { + match cell.compare_exchange(cur, new, Ordering::AcqRel, Ordering::Acquire) { + Ok(_) => return true, + Err(now) => cur = now, + } + } + false + } + + /// Clear the entry point, but only while it still names `expected_id`. A re-election that + /// found no candidate must not erase an entry a concurrent insert installed meanwhile — + /// `set_entry_point_if_not_better(NO_ID, 0, ..)` would, because a level-0 live entry is not + /// "better" than the level-0 clear. + pub fn clear_entry_point_if(&self, expected_id: u32) { + let cell = self.header_atomic_u64(H_ENTRY); + let mut cur = cell.load(Ordering::Acquire); + while (cur & 0xffff_ffff) as u32 == expected_id { + match cell.compare_exchange(cur, NO_ID as u64, Ordering::AcqRel, Ordering::Acquire) { Ok(_) => return, Err(now) => cur = now, } @@ -382,10 +538,34 @@ impl PlaneFile { self.header_atomic_u64(H_TXN_WATERMARK).store(txn, Ordering::Release); } + /// The completion stamp — 0, "incomplete mirror", once the plane is invalidated, whatever + /// a flush racing the invalidation wrote into the word afterwards. pub fn watermark(&self) -> u64 { + if self.invalidated() { + return 0; + } self.header_atomic_u64(H_TXN_WATERMARK).load(Ordering::Acquire) } + #[inline] + fn invalidated_cell(&self) -> &AtomicU8 { + unsafe { &*(self.map.as_ptr().add(H_INVALIDATED) as *const AtomicU8) } + } + + pub fn write_epoch(&self) -> u64 { + self.header_atomic_u64(H_WRITE_EPOCH).load(Ordering::Acquire) + } + + /// Release-ordered so a probe that acquires the new epoch also sees the slot it publishes. + pub fn bump_write_epoch(&self) { + self.header_atomic_u64(H_WRITE_EPOCH).fetch_add(1, Ordering::Release); + } + + /// Whether the one-way invalidation latch is set (by this or any other handle). + pub fn invalidated(&self) -> bool { + self.invalidated_cell().load(Ordering::Acquire) != 0 + } + #[inline] pub fn upper_ptr(&self, idx: u32) -> *const u8 { debug_assert!((idx as u64) < self.upper_capacity); @@ -491,6 +671,12 @@ impl PlaneFile { } } + /// Every nonzero registry tag, live or not (liveness is `tag_is_dead`). + #[cfg(test)] + pub(crate) fn registered_tags(&self) -> Vec { + (0..REGISTRY_SLOTS).map(|slot| self.registry_tag_cell(slot).load(Ordering::Acquire)).filter(|&t| t != 0).collect() + } + /// Try to take the OFD write lock on a registry slot's byte range. `probe` releases it /// immediately (liveness check); otherwise it is held for this handle's lifetime. #[cfg(target_os = "linux")] @@ -570,4 +756,23 @@ impl PlaneFile { unsafe { *(self.map.as_ptr().add(H_CLEAN_SHUTDOWN) as *mut u8) = 1 }; self.map.flush_range(0, HEADER_SIZE) } + + /// Mark the plane invalidated, durably, and nothing else: set the one-way latch, zero the + /// watermark, and msync the header page alone. Every handle then reads watermark 0 and + /// every later `open` refuses the file. The latch is what makes this stick against a + /// `flush_with_watermark` already in flight on this or another handle: that flush still + /// stamps the word, but nothing reads the word past the latch. + /// + /// Deliberately NOT `flush_with_watermark(Some(0))`: that writes the whole mapping back + /// first, and the caller invalidating a multi-GB plane cannot pay a full msync inline. + /// Skipping the data flush is sound because the data is being discarded, and because + /// lowering the watermark is the safe direction: the ordering hazard + /// `flush_with_watermark` exists to prevent is a NEW watermark over missing data, never + /// an old one over durable data. The stores precede the msync, so on an msync failure + /// the mark may still reach disk through ordinary writeback — also the safe direction. + pub fn invalidate(&self) -> io::Result<()> { + self.invalidated_cell().store(1, Ordering::Release); + self.set_watermark(0); + self.map.flush_range(0, HEADER_SIZE) + } } diff --git a/src/graph.rs b/src/graph.rs index 081ea74..c768b7a 100644 --- a/src/graph.rs +++ b/src/graph.rs @@ -6,14 +6,43 @@ use crate::distance::{cosine_i8_i8_raw, cosine_int8_raw, Query}; use crate::format::{ - PlaneFile, FLAG_DELETED, FLAG_VALID, MAX_UPPER_LEVELS, NO_UPPER, S_DEGREE, S_FLAGS, S_INV_MAG, S_LEVEL, S_SCALE, - S_UPPER_IDX, S_VECTOR, UPPER_CAP, UPPER_LEVEL_STRIDE, U_LEVELS, U_LISTS, + neighbor_offset, PlaneFile, FLAG_DELETED, FLAG_VALID, MAX_UPPER_LEVELS, NO_UPPER, S_DEGREE, S_FLAGS, S_INV_MAG, + S_LEVEL, S_SCALE, S_UPPER_IDX, S_VECTOR, UPPER_CAP, UPPER_LEVEL_STRIDE, UL_DEGREE, UL_IDS, U_LEVELS, U_LISTS, }; use crate::seqlock; use crate::seqlock::Wedged; +/// Aligned volatile load of a slot/upper-entry field another process may be mutating. +/// +/// This forbids the optimizer from duplicating, splitting, or sinking the load across the +/// seqlock's validating fence, which would let a reader act on bytes the generation check +/// never covered. It does NOT make the access race-free under Rust's memory model — only +/// atomics would, and that is the format change DESIGN.md §10 records as +/// follow-up. The vector is deliberately not read this way: `cosine_int8_raw` must stay +/// autovectorized, and a torn vector only perturbs a distance the generation check discards. +/// Every field read here is naturally aligned (slots are 64-aligned; the neighbor and upper +/// id arrays are 4-padded by format.rs), so these compile to single loads. +#[inline(always)] +unsafe fn vread(p: *const T) -> T { + p.read_volatile() +} + pub struct Graph { pub file: PlaneFile, + /// Rotates `probe_for_entry`'s starting offset so this plane's consecutive repairs sample + /// different ids. Per handle, not per process: a shared counter is advanced by every other + /// plane's repairs too, so one plane's calls can land on a single residue indefinitely — + /// which is the coverage the rotation exists to provide. + probe_rotation: std::sync::atomic::AtomicU32, + /// (high-water, write epoch) at which this handle's last `stride` consecutive probes all + /// came back empty, so a fully dead graph stops paying the probe; any handle's node write + /// bumps the header epoch and re-arms it. + probe_futile_hw: std::sync::atomic::AtomicU64, + probe_futile_epoch: std::sync::atomic::AtomicU64, + probe_futile_runs: std::sync::atomic::AtomicU32, + /// One repair probe at a time per handle: concurrent searches on the pool would each pay + /// the full walk before one of them publishes. + probe_in_flight: std::sync::atomic::AtomicBool, } /// A consistent full copy of one node (construction paths only; search uses zero-copy). @@ -27,7 +56,19 @@ pub struct NodeRead { impl Graph { pub fn new(file: PlaneFile) -> Self { - Graph { file } + Graph { + file, + probe_rotation: std::sync::atomic::AtomicU32::new(0), + probe_futile_hw: std::sync::atomic::AtomicU64::new(u64::MAX), + probe_futile_epoch: std::sync::atomic::AtomicU64::new(u64::MAX), + probe_futile_runs: std::sync::atomic::AtomicU32::new(0), + probe_in_flight: std::sync::atomic::AtomicBool::new(false), + } + } + + #[inline] + fn node_written(&self) { + self.file.bump_write_epoch(); } #[inline] @@ -67,12 +108,12 @@ impl Graph { seqlock::read_consistent(seq, self.file.self_tag, || { let p = self.file.slot_ptr(id); unsafe { - let flags = *p.add(S_FLAGS); + let flags = vread(p.add(S_FLAGS)); if flags & FLAG_VALID == 0 || flags & FLAG_DELETED != 0 { return None; } - let scale = (p.add(S_SCALE) as *const f32).read_unaligned(); - let inv_mag = (p.add(S_INV_MAG) as *const f32).read_unaligned(); + let scale = vread(p.add(S_SCALE) as *const f32); + let inv_mag = vread(p.add(S_INV_MAG) as *const f32); Some(cosine_int8_raw(query, p.add(S_VECTOR) as *const i8, scale, inv_mag)) } }, self.slot_sanitizer(id), || None, self.owner_dead()) @@ -119,20 +160,20 @@ impl Graph { } let seq = self.file.seq_atomic(id); let cap = self.file.layer0_cap; - let dims = self.file.dims; + let nbase = neighbor_offset(self.file.dims); seqlock::read_consistent(seq, self.file.self_tag, || { out.clear(); let p = self.file.slot_ptr(id); unsafe { - let flags = *p.add(S_FLAGS); + let flags = vread(p.add(S_FLAGS)); if flags & FLAG_VALID == 0 || flags & FLAG_DELETED != 0 { return None; } - let level = *p.add(S_LEVEL); - let degree = u16::from_le((p.add(S_DEGREE) as *const u16).read_unaligned()) as usize; - let base = p.add(S_VECTOR + dims) as *const u32; + let level = vread(p.add(S_LEVEL)); + let degree = u16::from_le(vread(p.add(S_DEGREE) as *const u16)) as usize; + let base = p.add(nbase) as *const u32; for i in 0..degree.min(cap) { - out.push(u32::from_le(base.add(i).read_unaligned())); + out.push(u32::from_le(vread(base.add(i)))); } Some(level) } @@ -149,11 +190,11 @@ impl Graph { seqlock::read_consistent(seq, self.file.self_tag, || { let p = self.file.slot_ptr(id); unsafe { - let flags = *p.add(S_FLAGS); + let flags = vread(p.add(S_FLAGS)); if flags & FLAG_VALID == 0 || flags & FLAG_DELETED != 0 { return NO_UPPER; } - (p.add(S_UPPER_IDX) as *const u32).read_unaligned() + vread(p.add(S_UPPER_IDX) as *const u32) } }, self.slot_sanitizer(id), || NO_UPPER, self.owner_dead()) } @@ -172,15 +213,15 @@ impl Graph { out.clear(); let p = self.file.upper_ptr(idx); unsafe { - let levels = *p.add(U_LEVELS); + let levels = vread(p.add(U_LEVELS)); if level > levels { return false; } let lp = p.add(U_LISTS + (level as usize - 1) * UPPER_LEVEL_STRIDE); - let degree = u16::from_le((lp as *const u16).read_unaligned()) as usize; - let base = lp.add(2) as *const u32; + let degree = u16::from_le(vread(lp.add(UL_DEGREE) as *const u16)) as usize; + let base = lp.add(UL_IDS) as *const u32; for i in 0..degree.min(UPPER_CAP) { - out.push(u32::from_le(base.add(i).read_unaligned())); + out.push(u32::from_le(vread(base.add(i)))); } true } @@ -206,8 +247,8 @@ impl Graph { for (l, list) in levels.iter().take(n).enumerate() { let lp = p.add(U_LISTS + l * UPPER_LEVEL_STRIDE); let deg = list.len().min(UPPER_CAP); - (lp as *mut u16).write_unaligned((deg as u16).to_le()); - let base = lp.add(2) as *mut u32; + (lp.add(UL_DEGREE) as *mut u16).write_unaligned((deg as u16).to_le()); + let base = lp.add(UL_IDS) as *mut u32; for (i, id) in list.iter().take(deg).enumerate() { base.add(i).write_unaligned(id.to_le()); } @@ -229,8 +270,8 @@ impl Graph { for (l, list) in levels.iter().take(n).enumerate() { let lp = p.add(U_LISTS + l * UPPER_LEVEL_STRIDE); let deg = list.len().min(UPPER_CAP); - (lp as *mut u16).write_unaligned((deg as u16).to_le()); - let base = lp.add(2) as *mut u32; + (lp.add(UL_DEGREE) as *mut u16).write_unaligned((deg as u16).to_le()); + let base = lp.add(UL_IDS) as *mut u32; for (i, id) in list.iter().take(deg).enumerate() { base.add(i).write_unaligned(id.to_le()); } @@ -246,7 +287,7 @@ impl Graph { return false; } let seq = self.file.seq_atomic(id); - seqlock::read_consistent(seq, self.file.self_tag, || unsafe { *self.file.slot_ptr(id).add(S_FLAGS) != 0 }, self.slot_sanitizer(id), || true, self.owner_dead()) + seqlock::read_consistent(seq, self.file.self_tag, || unsafe { vread(self.file.slot_ptr(id).add(S_FLAGS)) != 0 }, self.slot_sanitizer(id), || true, self.owner_dead()) } /// The slot's stored upper idx regardless of valid/deleted flags. Taken under the slot @@ -294,7 +335,7 @@ impl Graph { // readable. Emptied in place rather than freed: the freelist hand-off is not atomic // with publishing the slot below, so a mirror that read this index first could // republish a slot pointing at an entry already given to another node. One idle - // entry per id is the bounded retention hnsw-native-plane.md §10 accepts. + // entry per id is the bounded retention DESIGN.md §10 accepts. if existing != NO_UPPER { self.rewrite_upper(existing, &[])?; } @@ -356,12 +397,12 @@ impl Graph { return Ok(false); } let lp = p.add(U_LISTS + (level as usize - 1) * UPPER_LEVEL_STRIDE); - let degree = u16::from_le((lp as *const u16).read_unaligned()) as usize; - let base = lp.add(2) as *mut u32; + let degree = u16::from_le((lp.add(UL_DEGREE) as *const u16).read_unaligned()) as usize; + let base = lp.add(UL_IDS) as *mut u32; let mut list: Vec = (0..degree.min(UPPER_CAP)).map(|i| u32::from_le(base.add(i).read_unaligned())).collect(); f(&mut list); list.truncate(UPPER_CAP); - (lp as *mut u16).write_unaligned((list.len() as u16).to_le()); + (lp.add(UL_DEGREE) as *mut u16).write_unaligned((list.len() as u16).to_le()); for (i, id) in list.iter().enumerate() { base.add(i).write_unaligned(id.to_le()); } @@ -377,20 +418,21 @@ impl Graph { let seq = self.file.seq_atomic(id); let dims = self.file.dims; let cap = self.file.layer0_cap; + let nbase_off = neighbor_offset(dims); seqlock::read_consistent(seq, self.file.self_tag, || { let p = self.file.slot_ptr(id); unsafe { - let flags = *p.add(S_FLAGS); + let flags = vread(p.add(S_FLAGS)); if flags & FLAG_VALID == 0 || flags & FLAG_DELETED != 0 { return None; } - let level = *p.add(S_LEVEL); - let degree = u16::from_le((p.add(S_DEGREE) as *const u16).read_unaligned()) as usize; - let scale = (p.add(S_SCALE) as *const f32).read_unaligned(); - let inv_mag = (p.add(S_INV_MAG) as *const f32).read_unaligned(); + let level = vread(p.add(S_LEVEL)); + let degree = u16::from_le(vread(p.add(S_DEGREE) as *const u16)) as usize; + let scale = vread(p.add(S_SCALE) as *const f32); + let inv_mag = vread(p.add(S_INV_MAG) as *const f32); let vector = std::slice::from_raw_parts(p.add(S_VECTOR) as *const i8, dims).to_vec(); - let nbase = p.add(S_VECTOR + dims) as *const u32; - let neighbors = (0..degree.min(cap)).map(|i| u32::from_le(nbase.add(i).read_unaligned())).collect(); + let nbase = p.add(nbase_off) as *const u32; + let neighbors = (0..degree.min(cap)).map(|i| u32::from_le(vread(nbase.add(i)))).collect(); Some(NodeRead { level, scale, inv_mag, vector, neighbors }) } }, self.slot_sanitizer(id), || None, self.owner_dead()) @@ -413,11 +455,15 @@ impl Graph { (p.add(S_UPPER_IDX) as *mut u32).write_unaligned(upper_idx); std::ptr::copy_nonoverlapping(vector.as_ptr() as *const u8, p.add(S_VECTOR), dims); for (i, n) in neighbors.iter().enumerate() { - (p.add(S_VECTOR + dims + i * 4) as *mut u32).write_unaligned(n.to_le()); + (p.add(neighbor_offset(dims) + i * 4) as *mut u32).write_unaligned(n.to_le()); } // valid last within the locked section; the seqlock release publishes it *p.add(S_FLAGS) = FLAG_VALID; } + drop(_guard); + // after the release: a probe that consumed a bump while the slot was still invalid + // would otherwise latch on a graph that holds a live node + self.node_written(); Ok(()) } @@ -440,7 +486,7 @@ impl Graph { return Ok(false); } let degree = u16::from_le((p.add(S_DEGREE) as *const u16).read_unaligned()) as usize; - let base = p.add(S_VECTOR + dims) as *mut u32; + let base = p.add(neighbor_offset(dims)) as *mut u32; let mut list: Vec = (0..degree.min(cap)).map(|i| u32::from_le(base.add(i).read_unaligned())).collect(); f(&mut list); list.truncate(cap); @@ -473,7 +519,7 @@ impl Graph { if degree != expected.len() { return Ok(false); } - let base = p.add(S_VECTOR + dims) as *mut u32; + let base = p.add(neighbor_offset(dims)) as *mut u32; for (i, want) in expected.iter().enumerate() { if u32::from_le(base.add(i).read_unaligned()) != *want { return Ok(false); @@ -497,7 +543,7 @@ impl Graph { unsafe { (p.add(S_DEGREE) as *mut u16).write_unaligned((neighbors.len() as u16).to_le()); for (i, n) in neighbors.iter().enumerate() { - (p.add(S_VECTOR + dims + i * 4) as *mut u32).write_unaligned(n.to_le()); + (p.add(neighbor_offset(dims) + i * 4) as *mut u32).write_unaligned(n.to_le()); } } Ok(()) @@ -517,6 +563,13 @@ impl Graph { if entry_id == id { self.neighbors_into(id, &mut candidates); } + // Re-elect before the tombstone, not after: between marking the slot deleted and + // installing a replacement, every concurrent search routes through a node that reads + // as absent and returns nothing. The node is still live here, so a crash inside the + // window leaves the header naming a live entry either way. + if entry_id == id { + self.reelect_entry_point_replacing(&candidates, id); + } let upper_idx; { let seq = self.file.seq_atomic(id); @@ -541,9 +594,6 @@ impl Graph { self.rewrite_upper(upper_idx, &[])?; } self.file.free_upper(upper_idx); - if entry_id == id { - self.reelect_entry_point_replacing(&candidates, id); - } self.file.free_id(id); Ok(()) } @@ -552,7 +602,7 @@ impl Graph { /// first live node found scanning the id range (rare path: only when the entry's whole /// neighborhood is gone). An empty graph clears the entry. /// A node's level without copying its vector or edges (cheap re-election scans). - fn node_level(&self, id: u32) -> Option { + pub(crate) fn node_level(&self, id: u32) -> Option { if !self.in_range(id) { return None; } @@ -560,25 +610,82 @@ impl Graph { seqlock::read_consistent(seq, self.file.self_tag, || { let p = self.file.slot_ptr(id); unsafe { - if *p.add(S_FLAGS) != FLAG_VALID { + if vread(p.add(S_FLAGS)) != FLAG_VALID { return None; } - Some(*p.add(S_LEVEL)) + Some(vread(p.add(S_LEVEL))) } }, self.slot_sanitizer(id), || None, self.owner_dead()) } + /// Highest-level live node among at most `limit` probes, skipping `skip`. The read-side + /// repair's last resort, bounded because `reelect_entry_point_replacing`'s scan runs to the + /// high-water mark and a search on the shared pool thread cannot afford it. + /// + /// Walks down from the newest id with a stride spanning the whole range, so it assumes + /// nothing about where the live nodes sit: Harper allocates ids monotonically and never + /// reuses them, so a churned table's low prefix is all tombstones, while the crate's own + /// freelist reuses ids and keeps live nodes low. + /// + /// The start rotates per handle, so `stride` consecutive repairs of this plane cover every id + /// while each stays capped at `limit`; a fixed start would probe one residue class forever + /// and leave a graph lying between its samples invisible permanently, not for one search. + /// That coverage rests on `stride * limit >= hw`, which is why the stride is a ceiling + /// division: below it a walk stops short of id 0 and no offset ever reaches the tail. + /// + /// Best-level rather than first-live: a level-0 entry degrades every later search to a + /// layer-0-only beam. + pub(crate) fn probe_for_entry(&self, limit: u32, skip: u32) -> Option<(u32, u8)> { + let hw = self.file.id_high_water().min(self.file.max_nodes) as u32; + if hw == 0 || limit == 0 { + return None; + } + let stride = hw.div_ceil(limit); + use std::sync::atomic::Ordering::Relaxed; + let epoch = self.file.write_epoch(); + let unchanged = self.probe_futile_hw.load(Relaxed) == hw as u64 && self.probe_futile_epoch.load(Relaxed) == epoch; + if unchanged && self.probe_futile_runs.load(Relaxed) >= stride { + return None; // every residue probed since the last write anywhere: nothing to find + } + if self.probe_in_flight.swap(true, std::sync::atomic::Ordering::AcqRel) { + return None; // another search on this handle is repairing; it publishes for both + } + let offset = self.probe_rotation.fetch_add(1, Relaxed) % stride; + let mut best: Option<(u32, u8)> = None; + let mut cand = hw - 1 - offset; + for _ in 0..limit { + if cand != skip { + if let Some(level) = self.node_level(cand) { + if best.map(|(_, l)| level > l).unwrap_or(true) { + best = Some((cand, level)); + } + } + } + if cand < stride { + break; + } + cand -= stride; + } + if best.is_some() { + self.probe_futile_runs.store(0, Relaxed); + } else if unchanged { + self.probe_futile_runs.fetch_add(1, Relaxed); + } else { + self.probe_futile_hw.store(hw as u64, Relaxed); + self.probe_futile_epoch.store(epoch, Relaxed); + self.probe_futile_runs.store(1, Relaxed); + } + self.probe_in_flight.store(false, std::sync::atomic::Ordering::Release); + best + } + /// Pick a new entry point: the highest-level live node among `preferred`, else the /// highest-level live node found scanning the id range (level reads only — no per-node /// vector copies; still O(high-water), which only runs when an entry point vanished /// with no live neighborhood). Preferring level keeps the hierarchy navigable — a /// level-0 entry degrades every search to a layer-0-only beam. An empty graph clears /// the entry. - pub(crate) fn reelect_entry_point(&self, preferred: &[u32]) { - self.reelect_entry_point_replacing(preferred, crate::format::NO_ID) - } - - fn reelect_entry_point_replacing(&self, preferred: &[u32], replacing: u32) { + pub(crate) fn reelect_entry_point_replacing(&self, preferred: &[u32], replacing: u32) { let mut best: Option<(u32, u8)> = None; // the most recently replaced entry point is the best cheap candidate: usually alive, // usually high-level — and it makes the full fallback scan a last resort @@ -589,6 +696,9 @@ impl Graph { } } for &cand in preferred { + if cand == replacing { + continue; // the node on its way out is never its own replacement + } if let Some(level) = self.node_level(cand) { if best.map(|(_, l)| level > l).unwrap_or(true) { best = Some((cand, level)); @@ -598,6 +708,9 @@ impl Graph { if best.is_none() { let hw = self.file.id_high_water().min(self.file.max_nodes) as u32; for cand in 0..hw { + if cand == replacing { + continue; + } if let Some(level) = self.node_level(cand) { if best.map(|(_, l)| level > l).unwrap_or(true) { best = Some((cand, level)); @@ -610,7 +723,7 @@ impl Graph { } match best { Some((cand, level)) => self.file.set_entry_point_if_not_better(cand, level as u32, replacing), - None => self.file.set_entry_point_if_not_better(crate::format::NO_ID, 0, replacing), + None => self.file.clear_entry_point_if(replacing), } } @@ -659,16 +772,73 @@ impl Graph { (p.add(S_UPPER_IDX) as *mut u32).write_unaligned(upper_idx); std::ptr::copy_nonoverlapping(vector.as_ptr() as *const u8, p.add(S_VECTOR), dims); for (i, n) in neighbors.iter().enumerate() { - (p.add(S_VECTOR + dims + i * 4) as *mut u32).write_unaligned(n.to_le()); + (p.add(neighbor_offset(dims) + i * 4) as *mut u32).write_unaligned(n.to_le()); } *p.add(S_FLAGS) = FLAG_VALID; true } } }; - if !written { + if written { + self.node_written(); + } else { self.file.free_upper(upper_idx); } Ok(written) } } + +#[cfg(test)] +mod probe_tests { + use super::*; + use crate::distance::Query; + use crate::insert::{insert, InsertParams}; + use crate::search::{search, SearchScratch}; + use std::sync::atomic::Ordering::Relaxed; + + fn vector_for(i: u32, dims: usize) -> Vec { + (0..dims).map(|d| ((i as f32 * 0.31 + d as f32) * 0.7).sin()).collect() + } + + /// A graph whose every node is dead must stop paying the repair probe once a full rotation + /// has come back empty, and must resume it after ANY handle writes a node — the revival + /// here comes through a second handle on the same file, as another process's would. + #[test] + fn a_fully_dead_graph_stops_probing_until_a_node_is_written() { + let dims = 32; + let path = std::env::temp_dir().join(format!("hnsw-probefutile-{}.hnsw", std::process::id())); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 4_096).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..2_100 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + let hw = graph.file.id_high_water() as u32; + let stride = hw.div_ceil(1_024); // REPAIR_PROBE_LIMIT + assert!(stride > 1, "precondition: a rotation the guard has to wait out"); + let (entry, _) = graph.file.entry_point(); + for id in 0..hw { + let _ = graph.clear_node(id); + } + graph.file.clear_entry_point_if(entry); + + let query = Query::new(vector_for(7, dims)); + for _ in 0..stride { + assert!(search(&graph, &query, 5, 64, &mut scratch).0.is_empty()); + } + let rotation = graph.probe_rotation.load(Relaxed); + assert!(search(&graph, &query, 5, 64, &mut scratch).0.is_empty()); + assert_eq!(graph.probe_rotation.load(Relaxed), rotation, "a probed-out plane must not probe again"); + + // a node written through another handle, with no entry-point update (mirroring hosts do + // not always re-elect), must be findable again within one rotation + let revived = 3u32; + let q = crate::distance::quantize_int8(&vector_for(revived, dims)); + let other = Graph::new(PlaneFile::open(&path).expect("a second handle")); + other.write_node_raw(revived, 0, &q.0, q.1, q.2, &[], &[]).expect("revive"); + let found = (0..stride).any(|_| !search(&graph, &Query::new(vector_for(revived, dims)), 5, 64, &mut scratch).0.is_empty()); + assert!(found, "a write must re-arm the probe"); + let _ = std::fs::remove_file(&path); + } +} diff --git a/src/insert.rs b/src/insert.rs index bc54ded..f20f84e 100644 --- a/src/insert.rs +++ b/src/insert.rs @@ -34,7 +34,7 @@ fn level_for(id: u32, ml: f64) -> u8 { /// Remove `to` from `from`'s adjacency at `level` (edge-replacement maintenance). fn remove_edge(graph: &Graph, from: u32, to: u32, level: u8) { if level == 0 { - graph.update_neighbors(from, |list| { + let _ = graph.update_neighbors(from, |list| { if let Some(pos) = list.iter().position(|&x| x == to) { list.remove(pos); } @@ -93,6 +93,25 @@ fn prune_with_coverage(graph: &Graph, base: u32, list: &mut Vec, cap: usize *list = scored.into_iter().map(|(cand, _)| cand).collect(); } +/// The contended fallback's merge: add `new_id` under the slot lock, displacing the tail once the +/// list is at `cap`. Which neighbor that is is arbitrary — appends push at the tail, so a list is +/// distance-ordered only immediately after a prune — but it must not be `new_id` itself, which is +/// what a push followed by `truncate(cap)` drops. That loss is the systematic one: the edge being +/// added is the in-edge keeping a freshly inserted node reachable from `nid`, and it disappears +/// every time the list is full and the CAS path is contended. Picking a better victim needs +/// distances, which this path deliberately keeps outside the lock. +fn merge_neighbor_capped(graph: &Graph, nid: u32, new_id: u32, cap: usize) { + let _ = graph.update_neighbors(nid, |list| { + if list.contains(&new_id) { + return; + } + if list.len() >= cap { + list.truncate(cap.saturating_sub(1)); + } + list.push(new_id); + }); +} + /// Add `new_id` to `nid`'s adjacency at `level`, coverage-pruning to `cap` when over. The /// prune's distance computations (which can major-fault on a cold mapping) run OUTSIDE the /// slot lock: the list is snapshotted, pruned, and applied with a compare-and-set; after a @@ -117,12 +136,7 @@ fn add_reverse_edge(graph: &Graph, nid: u32, new_id: u32, level: u8, cap: usize) } } // contended twice: merge cheaply under the lock (bounded critical section) - let _ = graph.update_neighbors(nid, |list| { - if !list.contains(&new_id) { - list.push(new_id); - list.truncate(cap); - } - }); + merge_neighbor_capped(graph, nid, new_id, cap); } else { let _ = graph.update_upper_level(nid, level, |list| { if list.contains(&new_id) { @@ -161,35 +175,58 @@ pub fn insert( let layer0_cap = graph.file.layer0_cap; let m = params.m; - let (entry_id, entry_level) = graph.file.entry_point(); - if entry_id == NO_ID { - let upper_idx = if level > 0 { graph.write_upper(&vec![Vec::new(); level as usize]).unwrap_or(NO_UPPER) } else { NO_UPPER }; - graph.write_node(id, level, &bytes, scale, inv_mag, &[], upper_idx).map_err(|_| InsertError::Wedged)?; - // CAS: a concurrent first insert may have installed an entry already — never clobber - graph.file.set_entry_point_if_not_better(id, level as u32, NO_ID); - return Ok(id); - } - let mut stats = SearchStats { visits: 0 }; - let (entry_id, entry_level, entry_dist) = match graph.distance_to(entry_id, &query) { - Some(d) => (entry_id, entry_level, d), - None => { - // The stored entry point is gone (e.g. a mirroring host cleared it without - // re-electing). Self-promoting an edgeless new node here would orphan the whole - // existing graph behind an unreachable root — re-elect from the live graph and - // continue; only a truly empty graph makes this node the first entry. - graph.reelect_entry_point(&[]); - let (re_id, re_level) = graph.file.entry_point(); - match (re_id != NO_ID).then(|| graph.distance_to(re_id, &query)).flatten() { - Some(d) => (re_id, re_level, d), - None => { - let upper_idx = if level > 0 { graph.write_upper(&vec![Vec::new(); level as usize]).unwrap_or(NO_UPPER) } else { NO_UPPER }; - graph.write_node(id, level, &bytes, scale, inv_mag, &[], upper_idx).map_err(|_| InsertError::Wedged)?; - graph.file.set_entry_point_if_not_better(id, level as u32, NO_ID); - return Ok(id); - } + // Upper entry a first-entry claim attempt already published for `id`. Its slot names the + // index, so the join path must rewrite it in place — freeing an index a live slot names + // would let another node adopt it mid-traversal. + let mut published_upper = NO_UPPER; + let mut published = false; + let publish_edgeless = |published: &mut bool, published_upper: &mut u32| -> Result<(), InsertError> { + if *published { + return Ok(()); + } + *published_upper = + if level > 0 { graph.write_upper(&vec![Vec::new(); level as usize]).unwrap_or(NO_UPPER) } else { NO_UPPER }; + graph.write_node(id, level, &bytes, scale, inv_mag, &[], *published_upper).map_err(|_| InsertError::Wedged)?; + *published = true; + Ok(()) + }; + + // Resolve an entry point to grow from. Every turn makes progress — it claims an empty + // graph, joins a live entry, or replaces one that is provably gone — so the cap only + // guards an insert/delete interleaving that keeps clearing the entry under us. + let mut joined = None; + for _ in 0..16 { + let (entry_id, entry_level) = graph.file.entry_point(); + if entry_id == NO_ID { + publish_edgeless(&mut published, &mut published_upper)?; + // Claim only from EMPTY, and only the winner returns: a not-worse install would put + // this edgeless node over a live equal-or-lower-level entry and orphan the graph + // behind it, and it cannot report losing, which a loser must know to join instead. + if graph.file.claim_entry_if_empty(id, level as u32) { + return Ok(id); } + continue; // a racer rooted the graph — join it rather than stand alone } + if let Some(d) = graph.distance_to(entry_id, &query) { + joined = Some((entry_id, entry_level, d)); + break; + } + // The stored entry point is gone (e.g. a mirroring host cleared it without + // re-electing). Self-promoting an edgeless new node here would orphan the whole + // existing graph behind an unreachable root — re-elect from the live graph and + // continue; only a truly empty graph makes this node the first entry. + graph.reelect_entry_point_replacing(&[], entry_id); + } + // An unresolvable entry point is an error the host retries: Ok here would report success + // for a node no search can reach. + let Some((entry_id, entry_level, entry_dist)) = joined else { + if published { + // the edgeless node a failed claim left behind is a live-reading slot with no + // in-edges: a later re-election or repair probe could root the graph at it + let _ = graph.delete_node(id); + } + return Err(InsertError::Wedged); }; let top = level.min(entry_level as u8); let (mut ep, mut ep_dist) = @@ -261,7 +298,12 @@ pub fn insert( .unwrap_or_default() }) .collect(); - graph.write_upper(&levels).unwrap_or(NO_UPPER) + if published_upper != NO_UPPER { + graph.rewrite_upper(published_upper, &levels).map_err(|_| InsertError::Wedged)?; + published_upper + } else { + graph.write_upper(&levels).unwrap_or(NO_UPPER) + } } else { NO_UPPER }; @@ -279,7 +321,7 @@ pub fn insert( if (level as u32) > entry_level { // CAS against the observed entry: a concurrent higher-level promotion wins - graph.file.set_entry_point_if_not_better(id, level as u32, entry_id); + graph.file.promote_entry_point(id, level as u32, entry_id); } Ok(id) } @@ -290,3 +332,41 @@ fn scratch_begin(graph: &Graph, scratch: &mut SearchScratch) { // via this helper to keep the public surface small. scratch.begin_public(graph.file.id_high_water()); } + +#[cfg(test)] +mod reverse_edge_tests { + use super::*; + use crate::PlaneFile; + + /// The contended fallback must still add the edge when the neighbor list is already full — + /// the one case where a push-then-`truncate(cap)` discards `new_id` rather than a neighbor, + /// losing the in-edge exactly in the contended-and-full case the fallback exists to serve. + #[test] + fn a_contended_merge_into_a_full_neighbor_list_keeps_the_edge_it_adds() { + let dims = 8; + let cap = 8usize; + let path = std::env::temp_dir().join(format!("hnsw-revedge-{}.hnsw", std::process::id())); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, cap, 4_096).expect("create")); + + let vector = vec![0i8; dims]; + let full: Vec = (1..=cap as u32).collect(); + graph.write_node_raw(0, 0, &vector, 1.0, 1.0, &full, &[]).expect("seed the full list"); + for &nid in &full { + graph.write_node_raw(nid, 0, &vector, 1.0, 1.0, &[], &[]).expect("seed a neighbor"); + } + let newcomer = cap as u32 + 1; + graph.write_node_raw(newcomer, 0, &vector, 1.0, 1.0, &[], &[]).expect("seed the newcomer"); + + merge_neighbor_capped(&graph, 0, newcomer, cap); + + let mut neighbors: Vec = Vec::new(); + graph.neighbors_into(0, &mut neighbors).expect("node 0 is live"); + assert!( + neighbors.contains(&newcomer), + "the contended merge dropped the edge it was adding: {neighbors:?}" + ); + assert_eq!(neighbors.len(), cap, "the merge must stay within the layer-0 cap"); + let _ = std::fs::remove_file(&path); + } +} diff --git a/src/invalidate.rs b/src/invalidate.rs new file mode 100644 index 0000000..7af4787 --- /dev/null +++ b/src/invalidate.rs @@ -0,0 +1,336 @@ +//! Path-level invalidation: make a plane file that could not be deleted unadoptable, durably. +//! +//! Two markers, both attempted every call: the in-band latch (`PlaneFile::invalidate` — the +//! sticky header byte plus a zeroed watermark, msync'd) and a `.stale` sidecar, fsync'd +//! along with its directory entry. `PlaneFile::open` refuses a file carrying either, so the +//! markers are enforced by the package, not by each host's attach path. In band first: the +//! sidecar is what a process that cannot map the file checks, the latch is what covers a +//! plane whose sidecar a crash lost. A temporary handle opened here is dropped before the +//! sidecar is written and before returning — its mapping is the kind of thing that keeps a +//! file undeletable in the first place, and its registry slot must not wait on a finalizer. + +use crate::format::PlaneFile; +use std::fs::{File, OpenOptions}; +use std::io; +use std::path::{Path, PathBuf}; + +/// Which markers landed. `Ok` for a marker means it is durable, not merely written. +#[derive(Debug)] +pub struct Invalidation { + pub in_band: io::Result<()>, + pub sidecar: io::Result<()>, +} + +/// The sidecar convention: `.stale`. Its presence means the plane file is stale +/// and must never be opened; hosts delete both and rebuild. +pub fn stale_path_for(path: &Path) -> PathBuf { + let mut name = path.as_os_str().to_owned(); + name.push(".stale"); + PathBuf::from(name) +} + +/// Invalidate the plane at `path` through a temporary handle. Returns `Err` only when NEITHER +/// marker became durable. Nothing here deletes or renames, so the caller keeps whatever +/// recovery state it had; an in-band mark whose msync failed may still have landed in the +/// shared mapping, which is the safe direction (every handle reads it as incomplete). +/// Idempotent: an already invalidated plane reports both markers again. +pub fn invalidate_plane(path: &Path) -> io::Result { + invalidate_at(path, None) +} + +/// Invalidate the file `plane` maps, in band through the handle itself and with the sidecar +/// next to the path it was opened at. The caller's own handle is the right one when it +/// exists: on Windows that mapping is why the unlink failed, and a second open would claim a +/// second registry slot for nothing. The path must not have been replaced underneath the +/// handle since it opened; a host that unlinked and recreated it has nothing to invalidate. +pub fn invalidate_file(plane: &PlaneFile) -> io::Result { + invalidate_at(&plane.path, Some(plane)) +} + +fn invalidate_at(path: &Path, attached: Option<&PlaneFile>) -> io::Result { + invalidate_with(path, attached, write_sidecar) +} + +/// The order is the contract: the in-band mark is durable before the sidecar exists, so a +/// crash between the two cannot leave a sidecar-less file with its old watermark. +fn invalidate_with( + path: &Path, + attached: Option<&PlaneFile>, + sidecar: impl FnOnce(&Path) -> io::Result<()>, +) -> io::Result { + let in_band = match attached { + Some(plane) => plane.invalidate(), + None => PlaneFile::open_for_invalidation(path).and_then(|plane| plane.invalidate()), + }; + let sidecar = sidecar(&stale_path_for(path)); + if let (Err(in_band), Err(sidecar)) = (&in_band, &sidecar) { + return Err(io::Error::other(format!( + "neither invalidation marker is durable for {}: in-band: {in_band}; sidecar: {sidecar}", + path.display() + ))); + } + Ok(Invalidation { in_band, sidecar }) +} + +/// Create-new rather than create: the plane directory may be writable by another principal, +/// and a planted symlink at the sidecar path would otherwise be followed and its target +/// truncated. An existing marker is re-synced through a no-follow open checked on the open +/// handle, so a swap between the two calls cannot redirect the sync either. +fn write_sidecar(stale: &Path) -> io::Result<()> { + let marker = match OpenOptions::new().write(true).create_new(true).open(stale) { + Ok(file) => file, + Err(e) if e.kind() == io::ErrorKind::AlreadyExists => open_existing_marker(stale)?, + Err(e) => return Err(e), + }; + marker.sync_all()?; + sync_dir(parent_dir(stale)) +} + +fn open_existing_marker(stale: &Path) -> io::Result { + let mut options = OpenOptions::new(); + options.write(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + // O_NONBLOCK: a FIFO planted here must fail (ENXIO) rather than block the open + options.custom_flags(libc::O_NOFOLLOW | libc::O_NONBLOCK); + } + #[cfg(windows)] + { + use std::os::windows::fs::OpenOptionsExt; + const FILE_FLAG_OPEN_REPARSE_POINT: u32 = 0x0020_0000; + options.custom_flags(FILE_FLAG_OPEN_REPARSE_POINT); + } + let file = options.open(stale)?; + if !file.metadata()?.file_type().is_file() { + return Err(io::Error::other(format!("{} exists and is not a regular file", stale.display()))); + } + Ok(file) +} + +fn parent_dir(stale: &Path) -> &Path { + stale.parent().filter(|p| !p.as_os_str().is_empty()).unwrap_or(Path::new(".")) +} + +/// Make the directory entry durable. Windows has no directory fsync through `std` (a +/// directory handle needs backup semantics); there `FlushFileBuffers` on the marker itself +/// is documented to flush the metadata of its creation, so the marker's `sync_all` is the +/// durability point and this is a no-op. +#[cfg(unix)] +fn sync_dir(dir: &Path) -> io::Result<()> { + File::open(dir)?.sync_all() +} + +#[cfg(not(unix))] +fn sync_dir(_dir: &Path) -> io::Result<()> { + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn tmp(tag: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("hnsw-invalidate-{}-{tag}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + dir.join("plane.hnsw") + } + + fn complete_looking_plane(path: &Path) { + let plane = PlaneFile::create(path, 8, 8, 256).expect("create"); + plane.flush_with_watermark(Some(4_096)).expect("barrier"); + assert_eq!(plane.watermark(), 4_096); + } + + fn open_is_refused(path: &Path) { + let err = PlaneFile::open(path).err().expect("an invalidated plane must not open"); + assert!(err.to_string().contains("invalidated"), "{err}"); + } + + #[test] + fn both_markers_land_and_every_later_open_is_refused() { + let path = tmp("both"); + complete_looking_plane(&path); + let outcome = invalidate_plane(&path).expect("at least one marker"); + assert!(outcome.in_band.is_ok(), "{:?}", outcome.in_band); + assert!(outcome.sidecar.is_ok(), "{:?}", outcome.sidecar); + assert!(stale_path_for(&path).is_file()); + open_is_refused(&path); + } + + /// The in-band mark must be on the file before the sidecar is even attempted, through a + /// temporary handle and through the caller's own: the sidecar writer here observes the + /// latch (and no sidecar yet) at the moment it is called. + #[test] + fn the_in_band_mark_lands_before_the_sidecar_is_written() { + let path = tmp("order"); + complete_looking_plane(&path); + let observed = std::cell::Cell::new(false); + let outcome = invalidate_with(&path, None, |stale| { + assert!(!stale.exists(), "the sidecar must not exist before the in-band mark"); + observed.set(PlaneFile::open_for_invalidation(&path).expect("temp reopen").invalidated()); + write_sidecar(stale) + }) + .expect("invalidate"); + assert!(observed.get(), "the latch must be set before the sidecar step runs"); + assert!(outcome.in_band.is_ok() && outcome.sidecar.is_ok(), "{outcome:?}"); + + let path = tmp("orderattached"); + complete_looking_plane(&path); + let attached = PlaneFile::open(&path).expect("open"); + let observed = std::cell::Cell::new(false); + invalidate_with(&path, Some(&attached), |stale| { + observed.set(attached.invalidated() && !stale.exists()); + write_sidecar(stale) + }) + .expect("invalidate"); + assert!(observed.get(), "the attached handle must carry the latch before the sidecar step"); + } + + #[test] + fn invalidation_is_idempotent() { + let path = tmp("twice"); + complete_looking_plane(&path); + invalidate_plane(&path).expect("first"); + let again = invalidate_plane(&path).expect("second"); + assert!(again.in_band.is_ok() && again.sidecar.is_ok(), "{again:?}"); + } + + #[test] + fn the_sidecar_is_named_next_to_the_plane() { + assert_eq!(stale_path_for(Path::new("/data/t/a%2Fb.hnsw")), PathBuf::from("/data/t/a%2Fb.hnsw.stale")); + assert_eq!(parent_dir(Path::new("plane.hnsw.stale")), Path::new(".")); + assert_eq!(parent_dir(Path::new("/data/t/plane.hnsw.stale")), Path::new("/data/t")); + } + + /// A directory squatting the sidecar path makes its creation fail on every platform. + #[test] + fn in_band_alone_still_succeeds_when_the_sidecar_cannot_be_written() { + let path = tmp("inbandonly"); + complete_looking_plane(&path); + std::fs::create_dir(stale_path_for(&path)).unwrap(); + let outcome = invalidate_plane(&path).expect("the in-band mark is enough"); + assert!(outcome.in_band.is_ok()); + assert!(outcome.sidecar.is_err(), "a directory at the sidecar path must be reported"); + std::fs::remove_dir(stale_path_for(&path)).unwrap(); + open_is_refused(&path); + } + + /// A file that is not a plane cannot carry the in-band mark; the sidecar must still land, + /// and the temporary open that failed must not leave anything holding the file. + #[test] + fn the_sidecar_alone_still_succeeds_when_the_plane_cannot_be_opened() { + let path = tmp("sidecaronly"); + std::fs::write(&path, b"not a plane").unwrap(); + let outcome = invalidate_plane(&path).expect("the sidecar is enough"); + assert!(outcome.in_band.is_err()); + assert!(outcome.sidecar.is_ok(), "{:?}", outcome.sidecar); + assert!(stale_path_for(&path).is_file()); + std::fs::remove_file(&path).expect("nothing of ours may hold the file"); + } + + /// Neither marker: an error that names both causes, and nothing deleted or replaced — the + /// caller's recovery state (retry later, stay disabled in-process) is preserved. + #[test] + fn a_double_failure_is_an_error_and_deletes_nothing() { + let path = tmp("double"); + std::fs::write(&path, b"not a plane").unwrap(); + std::fs::create_dir(stale_path_for(&path)).unwrap(); + let err = invalidate_plane(&path).expect_err("no marker landed"); + let message = err.to_string(); + assert!(message.contains("in-band:") && message.contains("sidecar:"), "{message}"); + assert_eq!(std::fs::read(&path).unwrap(), b"not a plane"); + assert!(stale_path_for(&path).is_dir(), "nothing may be deleted or replaced"); + std::fs::remove_file(&path).expect("nothing of ours may hold the file"); + } + + /// The in-band mark must survive the writers that can still reach the header: a flush + /// already in flight on this handle, and another handle's own watermark stamps. Without + /// the sticky latch a `flushAsync(900)` racing the invalidation restored the old + /// completion stamp and the plane was adoptable again. + #[test] + fn a_later_flush_or_stamp_cannot_revive_an_invalidated_plane() { + let path = tmp("revive"); + complete_looking_plane(&path); + let ours = PlaneFile::open(&path).expect("our handle"); + let theirs = PlaneFile::open(&path).expect("another worker's handle"); + let outcome = invalidate_file(&ours).expect("invalidate"); + assert!(outcome.in_band.is_ok() && outcome.sidecar.is_ok(), "{outcome:?}"); + ours.flush_with_watermark(Some(900)).expect("the pending flush lands after the invalidation"); + theirs.set_watermark(4_096); + theirs.flush_with_watermark(None).expect("their cadence barrier"); + assert_eq!(ours.watermark(), 0, "an invalidated plane must read incomplete on every handle"); + assert_eq!(theirs.watermark(), 0); + std::fs::remove_file(stale_path_for(&path)).unwrap(); + open_is_refused(&path); + } + + /// A planted symlink at the sidecar path must not be followed, on the first invalidation + /// (create-new) and on a repeat that finds the marker swapped for a link: the victim keeps + /// its bytes and the sidecar is reported as not durable. + #[cfg(unix)] + #[test] + fn a_symlink_at_the_sidecar_path_is_never_followed() { + let path = tmp("symlink"); + complete_looking_plane(&path); + let victim = path.with_file_name("victim.txt"); + std::fs::write(&victim, b"precious").unwrap(); + std::os::unix::fs::symlink(&victim, stale_path_for(&path)).unwrap(); + let outcome = invalidate_plane(&path).expect("in band still lands"); + assert!(outcome.sidecar.is_err(), "a symlink at the sidecar path must be refused"); + assert_eq!(std::fs::read(&victim).unwrap(), b"precious"); + let err = open_existing_marker(&stale_path_for(&path)).err().expect("the no-follow reopen must refuse a link"); + assert!(err.raw_os_error().is_some() || err.to_string().contains("regular file"), "{err}"); + } + + /// The package enforces the markers at open: a sidecar alone (the plane's own header was + /// never reached) refuses the open, and a create over a leftover sidecar refuses too + /// rather than minting a plane that can never be opened again. + #[test] + fn open_and_create_refuse_a_path_with_a_sidecar() { + let path = tmp("sidecaropen"); + complete_looking_plane(&path); + std::fs::write(stale_path_for(&path), b"").unwrap(); + open_is_refused(&path); + let err = PlaneFile::create(&path, 8, 8, 256).err().expect("create must refuse"); + assert!(err.to_string().contains("stale"), "{err}"); + std::fs::remove_file(stale_path_for(&path)).unwrap(); + assert_eq!(PlaneFile::open(&path).expect("clean again").watermark(), 4_096); + } + + /// The temporary handle must be gone before the call returns: the file is deletable + /// (which its mapping would block on Windows) and its registry slot reads dead to a + /// concurrent opener (Linux, where the registry exists). + #[test] + fn the_temporary_handle_is_released_before_returning() { + let path = tmp("release"); + complete_looking_plane(&path); + let observer = PlaneFile::open(&path).expect("a concurrent opener"); + invalidate_plane(&path).expect("invalidate"); + #[cfg(target_os = "linux")] + { + let registered: Vec = observer.registered_tags().into_iter().filter(|&t| t != observer.self_tag).collect(); + assert!(!registered.is_empty(), "the temporary open must have registered itself"); + for tag in registered { + assert!(observer.tag_is_dead(tag), "registry tag {tag:#x} still reads alive after the call returned"); + } + } + drop(observer); + std::fs::remove_file(&path).expect("no mapping of ours may hold the file"); + } + + /// Through the caller's own handle nothing is opened here, so no registry slot is claimed. + #[test] + fn an_attached_handle_means_no_temporary_open() { + let path = tmp("attached"); + complete_looking_plane(&path); + let attached = PlaneFile::open(&path).expect("open"); + let before = attached.registered_tags(); + let outcome = invalidate_file(&attached).expect("invalidate"); + assert!(outcome.in_band.is_ok() && outcome.sidecar.is_ok(), "{outcome:?}"); + assert_eq!(attached.registered_tags(), before, "no second opener may appear"); + assert_eq!(attached.watermark(), 0); + assert!(stale_path_for(&path).is_file()); + } +} diff --git a/src/lib.rs b/src/lib.rs index 654bdd4..b7ae1ef 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,11 +1,12 @@ //! hnsw-plane: native HNSW traversal plane over a memory-mapped fixed-slot file. -//! Design: ../../hnsw-native-plane.md. NAPI bindings land behind the `napi` feature in +//! Design: ../DESIGN.md. NAPI bindings land behind the `napi` feature in //! phase-1 integration; the core is buildable and benchmarkable standalone. pub mod distance; pub mod format; pub mod graph; pub mod insert; +pub mod invalidate; #[cfg(feature = "napi")] mod napi; pub mod search; @@ -13,3 +14,4 @@ pub mod seqlock; pub use format::PlaneFile; pub use graph::Graph; +pub use invalidate::{invalidate_file, invalidate_plane, stale_path_for, Invalidation}; diff --git a/src/napi.rs b/src/napi.rs index 910dc2d..e8e8c27 100644 --- a/src/napi.rs +++ b/src/napi.rs @@ -28,6 +28,69 @@ impl ScratchPool { } } +#[napi(object)] +pub struct InvalidationOutcome { + pub in_band: bool, + pub sidecar: bool, + pub in_band_error: Option, + pub sidecar_error: Option, +} + +impl From for InvalidationOutcome { + fn from(outcome: crate::invalidate::Invalidation) -> Self { + InvalidationOutcome { + in_band: outcome.in_band.is_ok(), + sidecar: outcome.sidecar.is_ok(), + in_band_error: outcome.in_band.err().map(|e| e.to_string()), + sidecar_error: outcome.sidecar.err().map(|e| e.to_string()), + } + } +} + +/// Make the plane file at `path` unadoptable, durably, through a temporary handle released +/// before this returns: the in-band latch (watermark 0 on every handle, every later open +/// refused) and the fsync'd `.stale` sidecar. Both markers are attempted; throws only when +/// neither became durable, leaving the file exactly as found. Synchronous: three small +/// fsyncs on a cold path. Use invalidatePlaneAsync where the caller can await. +#[napi] +pub fn invalidate_plane(path: String) -> Result { + crate::invalidate::invalidate_plane(std::path::Path::new(&path)) + .map(InvalidationOutcome::from) + .map_err(|e| Error::from_reason(e.to_string())) +} + +pub struct InvalidateTask { + path: String, +} + +#[napi] +impl Task for InvalidateTask { + type Output = InvalidationOutcome; + type JsValue = InvalidationOutcome; + + fn compute(&mut self) -> Result { + crate::invalidate::invalidate_plane(std::path::Path::new(&self.path)) + .map(InvalidationOutcome::from) + .map_err(|e| Error::from_reason(e.to_string())) + } + + fn resolve(&mut self, _env: Env, output: Self::Output) -> Result { + Ok(output) + } +} + +/// invalidatePlane on the libuv thread pool, for callers that can await the fsyncs. +#[napi(ts_return_type = "Promise")] +pub fn invalidate_plane_async(path: String) -> AsyncTask { + AsyncTask::new(InvalidateTask { path }) +} + +/// The sidecar convention checked at attach: `.stale`. +#[napi] +pub fn stale_path_for(path: String) -> String { + crate::invalidate::stale_path_for(std::path::Path::new(&path)).to_string_lossy().into_owned() +} + #[napi(object)] pub struct SearchHit { pub id: u32, @@ -92,7 +155,7 @@ impl Task for PredicateSearchTask { dispatch: Box::new(move |ids: Vec| { let tx = tx.clone(); let ids_echo = ids.clone(); - tsfn.call_with_return_value( + let status = tsfn.call_with_return_value( ids, ThreadsafeFunctionCallMode::NonBlocking, move |ret: Uint8Array| { @@ -102,6 +165,10 @@ impl Task for PredicateSearchTask { Ok(()) }, ); + // a closing or saturated queue drops the callback without invoking it, so this + // batch will never answer; reporting it lets the drain finish on the batches + // that will, instead of holding teardown for the full deadline + status == Status::Ok }), rx, }; @@ -480,4 +547,28 @@ impl Plane { pub fn flush(&self, watermark: Option) -> Result<()> { self.graph.file.flush_with_watermark(watermark.map(|w| w as u64)).map_err(|e| Error::from_reason(e.to_string())) } + + /// Durably mark this plane invalidated in band: set the one-way latch, zero the watermark, + /// and msync the header page alone, so a host disabling a plane it cannot delete has the + /// mark on disk before it writes any out-of-band tombstone. Synchronous by design — it is + /// a 4 KB msync, not the whole-mapping writeback `flush` performs. + #[napi] + pub fn invalidate(&self) -> Result<()> { + self.graph.file.invalidate().map_err(|e| Error::from_reason(e.to_string())) + } + + /// invalidatePlane through this handle: the in-band mark via this mapping (no second + /// open, no second registry slot) and the `.stale` sidecar next to the path it opened. + #[napi] + pub fn invalidate_file(&self) -> Result { + crate::invalidate::invalidate_file(&self.graph.file) + .map(InvalidationOutcome::from) + .map_err(|e| Error::from_reason(e.to_string())) + } + + /// Whether the plane was invalidated (by any handle) since this one opened. + #[napi] + pub fn invalidated(&self) -> bool { + self.graph.file.invalidated() + } } diff --git a/src/search.rs b/src/search.rs index 2630879..9e8ca8b 100644 --- a/src/search.rs +++ b/src/search.rs @@ -219,6 +219,42 @@ pub fn greedy_descend( (current, current_dist) } +/// Slots a read-side repair may probe when the previous-entry hint is dead too. Bounded so a +/// search never pays the write path's O(high-water) re-election scan. +const REPAIR_PROBE_LIMIT: u32 = 1024; + +/// Resolve a live entry point for a read, repairing a dead one in place. +/// +/// A search that finds the header naming a deleted or sanitized node returns EMPTY, and on a +/// read-mostly table nothing ever repairs it: write-path re-election only runs on delete, and +/// a slot a reader sanitized after its writer died had no delete at all. +/// +/// The candidate is the O(1) previous-entry hint, then a probe capped at `REPAIR_PROBE_LIMIT` — +/// the hint is a single slot and can be dead itself. The cap is what keeps a read off the write +/// path's O(high-water) scan on the pool thread every search shares, and the repair publishes, +/// so only the first search after a wedge pays even the probe. +fn resolve_entry(graph: &Graph, query: &Query, stats: &mut SearchStats) -> Option<(u32, u32, f32)> { + let (entry_id, entry_level) = graph.file.entry_point(); + if entry_id != NO_ID { + if let Some(d) = graph.distance_to(entry_id, query) { + stats.visits += 1; + return Some((entry_id, entry_level, d)); + } + } + let hint = graph.file.previous_entry_point(); + let candidate = (hint != NO_ID && hint != entry_id) + .then(|| graph.node_level(hint).map(|level| (hint, level))) + .flatten() + .or_else(|| graph.probe_for_entry(REPAIR_PROBE_LIMIT, entry_id)); + let (id, level) = candidate?; + let d = graph.distance_to(id, query)?; + stats.visits += 1; + // Strict on the entry we observed dead, not a not-worse install: a live level-0 root claimed + // since the read above must win, or it is orphaned with nothing pointing at it. + graph.file.replace_entry_if(entry_id, id, level as u32); + Some((id, level as u32, d)) +} + /// Full search: greedy descent through upper layers, then beam at layer 0. pub fn search( graph: &Graph, @@ -228,19 +264,10 @@ pub fn search( scratch: &mut SearchScratch, ) -> (Vec<(u32, f32)>, SearchStats) { let mut stats = SearchStats { visits: 0 }; - let (entry_id, entry_level) = graph.file.entry_point(); - if entry_id == NO_ID { + let Some((entry_id, entry_level, entry_dist)) = resolve_entry(graph, query, &mut stats) else { return (Vec::new(), stats); - } - scratch.begin(graph.file.id_high_water()); - - let entry_dist = match graph.distance_to(entry_id, query) { - Some(d) => { - stats.visits += 1; - d - } - None => return (Vec::new(), stats), }; + scratch.begin(graph.file.id_high_water()); let (ep, ep_dist) = greedy_descend(graph, query, entry_id, entry_dist, entry_level, 0, &mut stats); let mut out = search_layer(graph, query, ep, ep_dist, ef, 0, scratch, &mut stats, None, u64::MAX); out.truncate(k); @@ -259,18 +286,10 @@ pub fn search_filtered( scratch: &mut SearchScratch, ) -> (Vec<(u32, f32)>, SearchStats) { let mut stats = SearchStats { visits: 0 }; - let (entry_id, entry_level) = graph.file.entry_point(); - if entry_id == NO_ID { + let Some((entry_id, entry_level, entry_dist)) = resolve_entry(graph, query, &mut stats) else { return (Vec::new(), stats); - } - scratch.begin_public(graph.file.id_high_water()); - let entry_dist = match graph.distance_to(entry_id, query) { - Some(d) => { - stats.visits += 1; - d - } - None => return (Vec::new(), stats), }; + scratch.begin_public(graph.file.id_high_water()); let (ep, ep_dist) = greedy_descend(graph, query, entry_id, entry_dist, entry_level, 0, &mut stats); let budget = if filter.is_some() { (ef * filter_expansion) as u64 } else { u64::MAX }; let mut out = search_layer(graph, query, ep, ep_dist, ef, 0, scratch, &mut stats, filter, budget); @@ -284,8 +303,11 @@ pub fn search_filtered( /// steer result admission only; routing uses pure distance order, bounded by the visit /// budget, so a slow or saturated JS loop degrades speculative overshoot, not correctness. pub struct PredicatePipe { - /// Sends one batch of ids for evaluation. Must not block. - pub dispatch: Box) + Send>, + /// Sends one batch of ids for evaluation. Must not block. Returns whether the batch was + /// actually handed off: a refused enqueue never produces a verdict, so counting it as + /// outstanding would make the tail drain wait out its whole deadline for an answer that + /// cannot arrive. + pub dispatch: Box) -> bool + Send>, /// Receives (ids, verdicts) pairs; verdicts[i] != 0 admits ids[i]. pub rx: std::sync::mpsc::Receiver<(Vec, Vec)>, } @@ -307,18 +329,10 @@ pub fn search_predicated( scratch: &mut SearchScratch, ) -> (Vec<(u32, f32)>, SearchStats) { let mut stats = SearchStats { visits: 0 }; - let (entry_id, entry_level) = graph.file.entry_point(); - if entry_id == NO_ID { + let Some((entry_id, entry_level, entry_dist)) = resolve_entry(graph, query, &mut stats) else { return (Vec::new(), stats); - } - scratch.begin_public(graph.file.id_high_water()); - let entry_dist = match graph.distance_to(entry_id, query) { - Some(d) => { - stats.visits += 1; - d - } - None => return (Vec::new(), stats), }; + scratch.begin_public(graph.file.id_high_water()); let (ep, ep_dist) = greedy_descend(graph, query, entry_id, entry_dist, entry_level, 0, &mut stats); use std::collections::HashMap; @@ -336,9 +350,13 @@ pub fn search_predicated( let mut nbuf = std::mem::take(&mut scratch.neighbors); + // guarded on `outstanding` rather than draining until the channel is empty: with a blocking + // receive the unguarded form pays another full timeout after the last verdict lands, on every + // filtered query macro_rules! drain { ($recv:expr) => { - while let Ok((ids, flags)) = $recv { + while outstanding > 0 { + let Ok((ids, flags)) = $recv else { break }; outstanding -= 1; for (i, id) in ids.iter().enumerate() { verdicts.insert(*id, flags.get(i).copied().unwrap_or(0) != 0); @@ -390,8 +408,7 @@ pub fn search_predicated( candidates.push(Candidate { distance: d, id: nid }); speculative.push((nid, d)); batch.push(nid); - if batch.len() >= PREDICATE_BATCH { - (pipe.dispatch)(std::mem::take(&mut batch)); + if batch.len() >= PREDICATE_BATCH && (pipe.dispatch)(std::mem::take(&mut batch)) { outstanding += 1; } } @@ -401,8 +418,7 @@ pub fn search_predicated( scratch.neighbors = nbuf; // flush the tail batch and block-drain what's still in flight - if !batch.is_empty() { - (pipe.dispatch)(std::mem::take(&mut batch)); + if !batch.is_empty() && (pipe.dispatch)(std::mem::take(&mut batch)) { outstanding += 1; } let deadline = std::time::Instant::now() + DRAIN_TIMEOUT; @@ -458,9 +474,7 @@ mod predicate_tests { }); let mut pipe = PredicatePipe { - dispatch: Box::new(move |ids| { - let _ = req_tx.send(ids); - }), + dispatch: Box::new(move |ids| req_tx.send(ids).is_ok()), rx: res_rx, }; let q: Vec = (0..dims).map(|d| ((41.0f32 * 0.31 + d as f32) * 0.7).sin()).collect(); @@ -474,4 +488,102 @@ mod predicate_tests { worker.join().unwrap(); let _ = std::fs::remove_file(&path); } + + /// The tail drain must stop receiving the moment the last verdict lands. Draining until the + /// channel reports empty sits out another full `recv_timeout` after `outstanding` reaches + /// zero — 50 ms added to every filtered query, against a sub-millisecond search. Measured + /// from the evaluator's last send so the search's own cost is not in the number, and over + /// the best of several queries so scheduler noise on one of them cannot pass for the extra + /// receive, which every query would pay. + #[test] + fn a_predicated_search_returns_as_soon_as_the_last_verdict_lands() { + let dims = 32; + let path = std::env::temp_dir().join(format!("hnsw-preddrain-{}.hnsw", std::process::id())); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 4_096).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..1_000u32 { + let v: Vec = (0..dims).map(|d| ((i as f32 * 0.31 + d as f32) * 0.7).sin()).collect(); + insert(&graph, &v, ¶ms, &mut scratch).unwrap(); + } + + let (req_tx, req_rx) = std::sync::mpsc::channel::>(); + let (res_tx, res_rx) = std::sync::mpsc::channel::<(Vec, Vec)>(); + // stamped before the send, so the search can never observe a verdict newer than the stamp + let last_send = std::sync::Arc::new(std::sync::Mutex::new(None::)); + let stamps = last_send.clone(); + let worker = std::thread::spawn(move || { + while let Ok(ids) = req_rx.recv() { + let verdicts = vec![1u8; ids.len()]; + *stamps.lock().unwrap() = Some(std::time::Instant::now()); + if res_tx.send((ids, verdicts)).is_err() { + break; + } + } + }); + + let mut pipe = PredicatePipe { + dispatch: Box::new(move |ids| req_tx.send(ids).is_ok()), + rx: res_rx, + }; + let q: Vec = (0..dims).map(|d| ((41.0f32 * 0.31 + d as f32) * 0.7).sin()).collect(); + let mut best = std::time::Duration::MAX; + for _ in 0..5 { + let (hits, _) = search_predicated( + &graph, + &Query::new(q.clone()), + 10, + 64, + &mut pipe, + 64 * 24, + &mut scratch, + ); + let tail = last_send.lock().unwrap().expect("the evaluator answered a batch").elapsed(); + assert!(!hits.is_empty(), "precondition: an admitting predicate returns results"); + best = best.min(tail); + } + assert!( + best < std::time::Duration::from_millis(25), + "the drain sat {best:?} past the last verdict on every query instead of returning on it" + ); + drop(pipe); + worker.join().unwrap(); + let _ = std::fs::remove_file(&path); + } + + /// A refused enqueue never answers. Counting it outstanding makes the tail drain wait out + /// its whole `DRAIN_TIMEOUT` for a verdict that cannot arrive — which is exactly the state + /// a closing environment puts every in-flight filtered query in, so teardown pays five + /// seconds per query instead of returning on the batches that did land. + #[test] + fn a_refused_predicate_enqueue_does_not_hold_the_drain() { + let dims = 32; + let path = std::env::temp_dir().join(format!("hnsw-refused-{}.hnsw", std::process::id())); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 4_096).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..1_000u32 { + let v: Vec = (0..dims).map(|d| ((i as f32 * 0.31 + d as f32) * 0.7).sin()).collect(); + insert(&graph, &v, ¶ms, &mut scratch).unwrap(); + } + + // the sender stays alive for the whole search, so a drain that believes a batch is + // outstanding blocks on the deadline rather than on a disconnected channel + let (tx, rx) = std::sync::mpsc::channel::<(Vec, Vec)>(); + let mut pipe = PredicatePipe { dispatch: Box::new(|_ids| false), rx }; + let q: Vec = (0..dims).map(|d| ((41.0f32 * 0.31 + d as f32) * 0.7).sin()).collect(); + let started = std::time::Instant::now(); + let (hits, _) = + search_predicated(&graph, &Query::new(q), 10, 64, &mut pipe, 64 * 24, &mut scratch); + let elapsed = started.elapsed(); + drop(tx); + assert!(hits.is_empty(), "no verdict can arrive for a refused batch, so nothing may be admitted"); + assert!( + elapsed < std::time::Duration::from_secs(1), + "the search waited {elapsed:?} on batches that were never enqueued (deadline is {DRAIN_TIMEOUT:?})" + ); + let _ = std::fs::remove_file(&path); + } } diff --git a/tests/concurrent.rs b/tests/concurrent.rs index 57122de..e3699e2 100644 --- a/tests/concurrent.rs +++ b/tests/concurrent.rs @@ -6,7 +6,7 @@ use hnsw_plane::insert::{insert, InsertParams}; use hnsw_plane::search::{search, SearchScratch}; use hnsw_plane::{Graph, PlaneFile}; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::Arc; +use std::sync::{Arc, Barrier}; fn vector_for(i: u32, dims: usize) -> Vec { // deterministic distinct unit-ish vectors on a few clusters @@ -16,6 +16,13 @@ fn vector_for(i: u32, dims: usize) -> Vec { let x = ((i as f32 * 0.37 + d as f32 * 1.13).sin() * 0.1) + if d % 7 == cluster { 1.0 } else { 0.0 }; v[d] = x; } + // Per-node signature (unique for i < dims^3). The cluster spike plus 0.1-amplitude noise + // alone leaves every member of a cluster inside int8 quantization noise of every other, so + // a self-query cannot tell "found this node" from "found some other node" — and a + // distance-only assertion over such a corpus passes even when the node is orphaned. + v[(i as usize) % dims] += 0.5; + v[(i as usize / dims) % dims] += 0.35; + v[(i as usize / (dims * dims)) % dims] += 0.22; v } @@ -31,18 +38,25 @@ fn concurrent_insert_search() { let per_writer = 2_000u32; let done = Arc::new(AtomicBool::new(false)); - std::thread::scope(|s| { - for w in 0..writers { - let graph = graph.clone(); - s.spawn(move || { - let params = InsertParams::default(); - let mut scratch = SearchScratch::new(); - for i in 0..per_writer { - let v = vector_for(w * per_writer + i, dims); - insert(&graph, &v, ¶ms, &mut scratch).expect("insert"); - } - }); - } + // (corpus index, node id): ids come from the plane's own allocator, so writers interleave + // them — a self-query must be checked against the id its insert actually returned + let inserted: Vec<(u32, u32)> = std::thread::scope(|s| { + let writers_done: Vec<_> = (0..writers) + .map(|w| { + let graph = graph.clone(); + s.spawn(move || { + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + let mut mine = Vec::with_capacity(per_writer as usize); + for i in 0..per_writer { + let index = w * per_writer + i; + let v = vector_for(index, dims); + mine.push((index, insert(&graph, &v, ¶ms, &mut scratch).expect("insert"))); + } + mine + }) + }) + .collect(); for _ in 0..4 { let graph = graph.clone(); let done = done.clone(); @@ -70,6 +84,7 @@ fn concurrent_insert_search() { } done_ref.store(true, Ordering::Relaxed); }); + writers_done.into_iter().flat_map(|h| h.join().expect("writer panicked")).collect() }); let total = writers * per_writer; @@ -78,11 +93,12 @@ fn concurrent_insert_search() { // Every stored vector must be found as its own nearest neighbor at generous ef. let mut scratch = SearchScratch::new(); let mut misses = 0; - for i in (0..total).step_by(97) { - let query = Query::new(vector_for(i, dims)); + for &(index, id) in inserted.iter().step_by(97) { + let query = Query::new(vector_for(index, dims)); let (results, _) = search(&graph, &query, 10, 256, &mut scratch); - // identical vectors exist across ids (clusters), so accept any zero-ish distance hit - if !results.iter().any(|&(_, d)| d < 1e-3) { + // by ID, not by distance: this corpus is clustered near-duplicates, so a hit at + // distance ~0 is routinely a DIFFERENT node and would mask an orphaned one + if !results.iter().any(|&(rid, _)| rid == id) { misses += 1; } } @@ -107,3 +123,52 @@ fn concurrent_insert_search() { let _ = std::fs::remove_file(&path); } + +/// Orthogonal per-writer vector: every writer's self-query has exactly one right answer, so a +/// node that lost the first-entry race is unmissable rather than covered by a near-duplicate. +fn axis_vector(writer: u32, dims: usize) -> Vec { + let mut v = vec![0.0f32; dims]; + v[writer as usize % dims] = 1.0; + v +} + +/// Many small fresh graphs, each racing its FIRST insert: that window is where the empty-graph +/// entry-point claim races, and a single barrier in a long build samples it about once. +#[test] +fn racing_first_inserts_all_stay_reachable() { + let dims = 32; + let writers = 4u32; + let rounds = 200; + for round in 0..rounds { + let path = std::env::temp_dir().join(format!("hnsw-first-{}-{round}.hnsw", std::process::id())); + let _ = std::fs::remove_file(&path); + let graph = Arc::new(Graph::new(PlaneFile::create(&path, dims, 16, 256).expect("create"))); + let barrier = Arc::new(Barrier::new(writers as usize)); + let ids: Vec<(u32, u32)> = std::thread::scope(|s| { + let handles: Vec<_> = (0..writers) + .map(|w| { + let graph = graph.clone(); + let barrier = barrier.clone(); + s.spawn(move || { + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + barrier.wait(); + (w, insert(&graph, &axis_vector(w, dims), ¶ms, &mut scratch).expect("insert")) + }) + }) + .collect(); + handles.into_iter().map(|h| h.join().expect("writer panicked")).collect() + }); + + let mut scratch = SearchScratch::new(); + for (w, id) in &ids { + let (results, _) = search(&graph, &Query::new(axis_vector(*w, dims)), 8, 64, &mut scratch); + assert!( + results.iter().any(|&(rid, _)| rid == *id), + "round {round}: writer {w}'s node {id} is unreachable from the entry point (found {results:?})" + ); + } + drop(graph); + let _ = std::fs::remove_file(&path); + } +} diff --git a/tests/reopen.rs b/tests/reopen.rs index 005c898..f80dba3 100644 --- a/tests/reopen.rs +++ b/tests/reopen.rs @@ -419,3 +419,390 @@ fn a_wedged_untouched_write_frees_its_upper_entry() { ); let _ = std::fs::remove_file(&path); } + +/// A wedged upper-entry cleanup must not leave the header naming a deleted entry point: the +/// cleanup is fallible, so an early return there strands every search on a dead entry. Asserts +/// the observable (searches still return hits), not the header word. +#[test] +fn a_wedged_upper_cleanup_still_reelects_the_entry_point() { + use std::sync::atomic::Ordering as O; + let dims = 32; + let path = tmp("wedgedelete"); + let _ = std::fs::remove_file(&path); + let graph = std::sync::Arc::new(Graph::new(PlaneFile::create(&path, dims, 16, 64).expect("create"))); + let raw = |id: u32, level: u8, neighbors: &[u32], upper: &[Vec]| { + let q = hnsw_plane::distance::quantize_int8(&vector_for(id, dims)); + graph.write_node_raw(id, level, &q.0, q.1, q.2, neighbors, upper).expect("mirror"); + }; + // node 0 is the entry point and the only node with a hierarchy, so it owns upper entry 0 + raw(0, 1, &[1], &[vec![1]]); + raw(1, 0, &[0], &[]); + graph.file.set_entry_point(0, 1); + + let upper_seq = graph.file.upper_seq_atomic(0) as *const _ as usize; + let g2 = graph.clone(); + let held = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let held2 = held.clone(); + let hold = std::thread::spawn(move || { + let seq = unsafe { &*(upper_seq as *const std::sync::atomic::AtomicU32) }; + let g2 = &g2; + let guard = hnsw_plane::seqlock::write_lock(seq, g2.file.self_tag, || panic!("live owner sanitized"), |tag| { + g2.file.tag_is_dead(tag) + }) + .expect("the holder must actually take the lock, or the test proves nothing"); + held2.store(true, O::Release); + std::thread::sleep(std::time::Duration::from_millis(6_500)); // past WRITE_WEDGE_AFTER + drop(guard); + }); + await_lock(&held); + assert_eq!(graph.delete_node(0), Err(hnsw_plane::seqlock::Wedged), "the held upper lock must wedge the cleanup"); + hold.join().unwrap(); + + assert_eq!(graph.file.entry_point().0, 1, "the entry point must be re-elected before the fallible cleanup"); + let mut scratch = SearchScratch::new(); + let (hits, _) = search(&graph, &Query::new(vector_for(1, dims)), 5, 64, &mut scratch); + assert!(!hits.is_empty(), "searches must keep working after a wedged delete of the entry point"); + let _ = std::fs::remove_file(&path); +} + +/// A search must repair an entry point that no writer will: a host that cleared the entry, or +/// a slot a reader sanitized after its writer died, leaves no delete to run the write-path +/// re-election, so on a read-mostly table every search returns empty indefinitely. +#[test] +fn search_repairs_an_entry_point_no_writer_will() { + let dims = 32; + let path = tmp("entryheal"); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 1_024).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..200 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + let prev = graph.file.previous_entry_point(); + assert_ne!(prev, hnsw_plane::format::NO_ID, "promotions must record a previous-entry hint to repair from"); + + // the entry's slot reads as gone with no delete having run (dead-writer sanitization, or a + // mirroring host clearing the node) — nothing on the write path will ever re-elect + let (entry, _) = graph.file.entry_point(); + graph.clear_node(entry).expect("tombstone the entry slot"); + + let (hits, _) = search(&graph, &Query::new(vector_for(7, dims)), 5, 64, &mut scratch); + assert!(!hits.is_empty(), "a search must self-heal past a dead entry point instead of returning empty"); + assert_ne!(graph.file.entry_point().0, entry, "the repair must be published, not repeated per search"); + let _ = std::fs::remove_file(&path); +} + +/// `invalidate` demotes a plane that already looks like a complete mirror back to "incomplete, +/// rebuild me", and reports barrier failure to its caller instead of into a dropped promise — +/// which is what lets the host order it before creating a `.stale` sidecar. (Durability itself +/// is not observable in-process: the mapping is MAP_SHARED, so every store is already visible to +/// a reopen and to `read()` whether or not the msync ran. The ordering that a crash would expose +/// is asserted on the host side, in `vectorIndexPlane.test.js`.) +#[test] +fn invalidate_demotes_a_complete_looking_mirror_and_reports_failure() { + let dims = 32; + let path = tmp("invalidate"); + let _ = std::fs::remove_file(&path); + { + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 1_024).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..50 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + graph.file.flush_with_watermark(Some(4_096)).expect("barrier"); + assert_eq!(graph.file.watermark(), 4_096, "precondition: a complete-looking mirror"); + graph.file.invalidate().expect("invalidate must report its barrier, not swallow it"); + assert_eq!(graph.file.watermark(), 0, "invalidation must mark the mirror incomplete in band"); + } + let refused = PlaneFile::open(&path).err().expect("a fresh opener must refuse the invalidated plane, not adopt it"); + assert!(refused.to_string().contains("invalidated"), "{refused}"); + let _ = std::fs::remove_file(&path); +} + +/// The hint is one slot and can die too: promote over a node, then lose BOTH that node and the +/// entry it was promoted over. Without the bounded probe the repair has nowhere left to look and +/// every later search returns empty although most of the graph is live. +#[test] +fn search_repairs_an_entry_point_whose_hint_is_dead_too() { + let dims = 32; + let path = tmp("entryhealdeadhint"); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 1_024).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..200 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + let hint = graph.file.previous_entry_point(); + assert_ne!(hint, hnsw_plane::format::NO_ID, "precondition: a hint to invalidate"); + let (entry, _) = graph.file.entry_point(); + + // both sanitized with no delete having run, so no write-path re-election ever happens and + // the hint the repair would follow names a node that reads as gone + graph.clear_node(hint).expect("tombstone the hint slot"); + graph.clear_node(entry).expect("tombstone the entry slot"); + + let (hits, _) = search(&graph, &Query::new(vector_for(7, dims)), 5, 64, &mut scratch); + assert!(!hits.is_empty(), "a dead hint must fall back to the bounded probe, not return empty forever"); + let repaired = graph.file.entry_point().0; + assert_ne!(repaired, entry, "the repair must be published"); + assert_ne!(repaired, hint, "the repair must not publish the dead hint"); + let _ = std::fs::remove_file(&path); +} + +/// Harper allocates node ids monotonically and never reuses them, so a table that has churned +/// has its whole low prefix tombstoned and only the newest ids live. A repair that probed a +/// fixed prefix would find nothing there and every search would return empty forever. +#[test] +fn search_repairs_an_entry_point_in_a_churned_graph_whose_low_ids_are_all_dead() { + let dims = 32; + let path = tmp("entryhealchurn"); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 4_096).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..1_200 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + // every id a prefix probe would reach is gone, as it is for any long-lived churned table + for id in 0..1_100u32 { + let _ = graph.clear_node(id); + } + let (entry, _) = graph.file.entry_point(); + let hint = graph.file.previous_entry_point(); + let _ = graph.clear_node(entry); + if hint != hnsw_plane::format::NO_ID { + let _ = graph.clear_node(hint); + } + + let (hits, _) = search(&graph, &Query::new(vector_for(1_150, dims)), 5, 64, &mut scratch); + assert!(!hits.is_empty(), "the probe must reach the live tail, not only a dead low prefix"); + let repaired = graph.file.entry_point().0; + assert!(graph.read_node(repaired).is_some(), "the repair must publish a live node"); + let _ = std::fs::remove_file(&path); +} + +/// With a stride above 1 a fixed start probes one residue class forever, so a live graph lying +/// entirely between its probes would never be found. The rotation makes `stride` consecutive +/// repairs cover every id; here the sole survivor is deliberately in the residue the unrotated +/// walk skips. +#[test] +fn a_repair_probe_rotates_so_no_live_node_stays_between_its_samples() { + let dims = 32; + let path = tmp("entryhealrotate"); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 4_096).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..2_100 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + let hw = graph.file.id_high_water() as u32; + let stride = hw.div_ceil(1_024); // REPAIR_PROBE_LIMIT + assert!(stride > 1, "precondition: a stride the rotation actually has to cover, got {stride}"); + // an unrotated walk starts at hw-1 and steps by `stride`, so it only ever sees that residue; + // keep exactly one node alive in a different one + let survivor = (0..hw).rev().find(|id| (hw - 1 - id) % stride != 0).expect("a skipped residue"); + for id in 0..hw { + if id != survivor { + let _ = graph.clear_node(id); + } + } + assert!(graph.read_node(survivor).is_some(), "precondition: the survivor is live"); + + let mut found = false; + for _ in 0..stride { + let (hits, _) = search(&graph, &Query::new(vector_for(survivor, dims)), 5, 64, &mut scratch); + if !hits.is_empty() { + found = true; + break; + } + } + assert!(found, "a rotating probe must reach every residue within `stride` repairs"); + assert_eq!(graph.file.entry_point().0, survivor, "the only live node must be the repaired entry"); + let _ = std::fs::remove_file(&path); +} + +/// The stride must be a ceiling division. Flooring it leaves `stride * limit < hw` whenever `hw` +/// is not a multiple of `limit`, so every rotated walk stops above the lowest `hw % limit` ids — +/// a permanent blind spot, not a one-search one, since no offset ever reaches it. A graph whose +/// only survivors sit in that prefix would return empty from every later search; this one's does. +#[test] +fn a_repair_probe_reaches_the_low_ids_a_floored_stride_would_never_sample() { + let dims = 32; + let limit = 1_024u32; // REPAIR_PROBE_LIMIT + let path = tmp("entryheallowprefix"); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 4_096).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..2_100 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + let hw = graph.file.id_high_water() as u32; + // a floored walk bottoms out at `hw % limit` whatever its rotation offset, so ids below that + // are exactly what the ceiling buys + let floored_reach = hw % limit; + assert!(hw > limit && floored_reach > 1, "precondition: a low prefix a floored stride skips, hw {hw}"); + let survivor = floored_reach / 2; + for id in 0..hw { + if id != survivor { + let _ = graph.clear_node(id); + } + } + assert!(graph.read_node(survivor).is_some(), "precondition: the survivor is live"); + assert_ne!(graph.file.previous_entry_point(), survivor, "precondition: the probe must be what finds it"); + + let mut found = false; + for _ in 0..hw.div_ceil(limit) { + let (hits, _) = search(&graph, &Query::new(vector_for(survivor, dims)), 5, 64, &mut scratch); + if !hits.is_empty() { + found = true; + break; + } + } + assert!(found, "a full rotation must cover every id, the lowest included"); + assert_eq!(graph.file.entry_point().0, survivor, "the only live node must be the repaired entry"); + let _ = std::fs::remove_file(&path); +} + +/// Rotation has to be per handle. With one process-wide counter, every other plane's repairs +/// advance it too, so two planes repairing in turn each see offsets stepping by 2 — one residue +/// class apiece, indefinitely, which is exactly what rotating was meant to prevent. Both planes +/// here hide their survivor in the same residue, so a shared counter must strand one of them +/// whichever offset it starts on. +#[test] +fn repair_probe_rotation_is_per_plane_not_per_process() { + let dims = 32; + let mut graphs = Vec::new(); + let mut survivors = Vec::new(); + let mut stride = 0u32; + for which in 0..2 { + let path = tmp(&format!("entryhealperplane{which}")); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 4_096).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..2_100 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + let hw = graph.file.id_high_water() as u32; + stride = hw.div_ceil(1_024); + assert!(stride > 1, "precondition: a stride the rotation has to cover"); + // the same skipped residue on both planes, so a shared counter cannot serve both + let survivor = (0..hw).rev().find(|id| (hw - 1 - id) % stride != 0).expect("a skipped residue"); + for id in 0..hw { + if id != survivor { + let _ = graph.clear_node(id); + } + } + graphs.push((graph, path)); + survivors.push(survivor); + } + + let mut scratch = SearchScratch::new(); + let mut found = [false; 2]; + for _ in 0..stride { + for (which, (graph, _)) in graphs.iter().enumerate() { + let (hits, _) = search(graph, &Query::new(vector_for(survivors[which], dims)), 5, 64, &mut scratch); + if !hits.is_empty() { + found[which] = true; + } + } + } + assert!(found[0] && found[1], "each plane must cover its own residues: {found:?}"); + for (_, path) in &graphs { + let _ = std::fs::remove_file(path); + } +} + +/// A repair publishes with a strict CAS on the entry it observed dead. A first insert that +/// claims the header in between owns the graph, and a higher-level repair candidate must lose to +/// it — installing the candidate would leave that insert's node with nothing pointing at it. +#[test] +fn a_repair_never_displaces_a_root_installed_while_it_ran() { + let dims = 32; + let path = tmp("entryhealrace"); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 1_024).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..64 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + let (observed, _) = graph.file.entry_point(); + let candidate = (0..64u32) + .find(|&id| id != observed && id != 7 && graph.read_node(id).is_some()) + .expect("a live repair candidate"); + let candidate_level = graph.read_node(candidate).expect("live").level; + assert!(graph.read_node(7).is_some(), "precondition: the racing root is a live node"); + + // the interleaving a repair races: the header no longer names the entry it read + graph.file.set_entry_point(7, 0); + assert!( + !graph.file.replace_entry_if(observed, candidate, candidate_level as u32), + "a repair must not publish over an entry installed after it read the dead one" + ); + assert_eq!(graph.file.entry_point().0, 7, "the root installed meanwhile stays"); + + // and it does publish when nothing raced it + let (current, _) = graph.file.entry_point(); + assert!(graph.file.replace_entry_if(current, candidate, candidate_level as u32)); + assert_eq!(graph.file.entry_point().0, candidate); + let _ = std::fs::remove_file(&path); +} + +/// The path-level entry point a host disables a plane through when it cannot delete the file: +/// afterwards a fresh opener is refused, the sidecar sits next to the plane, and the file is +/// deletable (no mapping of the helper's survives the call). +#[test] +fn invalidating_by_path_marks_the_plane_in_band_and_with_a_sidecar() { + let dims = 32; + let path = tmp("invalidatepath"); + let _ = std::fs::remove_file(&path); + { + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 1_024).expect("create")); + let params = InsertParams::default(); + let mut scratch = SearchScratch::new(); + for i in 0..50 { + insert(&graph, &vector_for(i, dims), ¶ms, &mut scratch).unwrap(); + } + graph.file.flush_with_watermark(Some(4_096)).expect("barrier"); + } + let outcome = hnsw_plane::invalidate_plane(&path).expect("at least one marker"); + assert!(outcome.in_band.is_ok() && outcome.sidecar.is_ok(), "{outcome:?}"); + let stale = hnsw_plane::stale_path_for(&path); + assert!(stale.is_file(), "the sidecar must exist next to the plane"); + assert!(PlaneFile::open(&path).is_err(), "a fresh opener must refuse the invalidated plane"); + std::fs::remove_file(&path).expect("the helper's temporary mapping must not outlive the call"); + let _ = std::fs::remove_file(&stale); +} + +/// A re-election that read a dead entry and stalled must not land over an EQUAL-level entry +/// installed meanwhile: that entry may be a fresh `claim_entry_if_empty` winner with no +/// in-edges yet, and displacing it orphans the node its insert already reported as landed. +/// A strictly higher-level install still wins. +#[test] +fn a_stale_reelection_never_displaces_an_equal_level_entry_installed_meanwhile() { + let dims = 32; + let path = tmp("staleelect"); + let _ = std::fs::remove_file(&path); + let graph = Graph::new(PlaneFile::create(&path, dims, 16, 64).expect("create")); + let raw = |id: u32, level: u8, upper: &[Vec]| { + let q = hnsw_plane::distance::quantize_int8(&vector_for(id, dims)); + graph.write_node_raw(id, level, &q.0, q.1, q.2, &[], upper).expect("mirror"); + }; + raw(1, 0, &[]); // the claimer, edgeless + raw(2, 0, &[]); // the stalled re-election's level-0 candidate + raw(3, 1, &[vec![]]); + let dead_entry = 9u32; + assert!(graph.file.claim_entry_if_empty(1, 0), "precondition: the claim wins on an empty header"); + + graph.file.set_entry_point_if_not_better(2, 0, dead_entry); + assert_eq!(graph.file.entry_point().0, 1, "an equal-level re-election must not displace the claimer"); + graph.file.set_entry_point_if_not_better(3, 1, dead_entry); + assert_eq!(graph.file.entry_point().0, 3, "a higher-level install still wins"); + let _ = std::fs::remove_file(&path); +}