1use super::super::builder::BatchBuilder;
2use super::super::cid::Cid;
3use super::super::config::Config;
4use super::super::error::{Error, Mutation as TreeMutation};
5use super::super::read::{OwnedValueLease, ReadSession, ScanOutcome};
6use super::super::splice::{splice_with_origin, SpliceStats};
7use super::super::store::{NodePublication, PublicationOrigin, Store};
8use super::super::Prolly;
9use super::builder::{build_hierarchy, build_hierarchy_parallel, IndexedRecord};
10use super::cache::{ContentCache, DEFAULT_PROXIMITY_CACHE_NODES};
11use super::distance::{prepare_vector, query_score, score};
12use super::mutation::{mutate_hierarchy, LogicalEdit};
13use super::search::{
14 adaptive_should_stop, insert_reranked_top_k, insert_top_k, plan_search,
15 retained_candidate_bytes, retained_search_candidate_bytes, AdaptiveContext, FrontierEntry,
16 PreparedFilter, RerankCandidate, SearchCandidate, SearchPlan,
17};
18use super::storage::quantized::ScalarQuantized;
19use super::storage::vector::ExternalVector;
20use super::storage::{
21 Descriptor, PhysicalNodeKind, ProximityNode, StoredRecord, StoredRecordRef, VectorRef,
22};
23use super::vector::promotion_level;
24use super::{
25 AcceleratorSet, BuildParallelism, DistanceMetric, ExactProximityRecord, Neighbor,
26 ProximityBuildStats, ProximityConfig, ProximityMutation, ProximityMutationStats,
27 ProximityRecord, ProximityRecordRef, ProximitySearchStats, ProximityTree, ProximityVectorRef,
28 ProximityVerification, SearchBackend, SearchBudget, SearchCompletion, SearchIo, SearchPolicy,
29 SearchRequest, SearchResult,
30};
31use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet};
32use std::ops::ControlFlow;
33use std::sync::{Arc, Mutex};
34
35pub struct ProximityMap<S: Store> {
37 store: S,
38 directory: Prolly<S>,
39 tree: ProximityTree,
40 node_cache: Mutex<ContentCache<ProximityNode>>,
41}
42
43pub struct ProximityReadSession<'map, S: Store> {
45 directory: ReadSession<'map, 'map, S>,
46 dimensions: u32,
47}
48
49impl<S: Store> ProximityReadSession<'_, S> {
50 pub fn get_with<R>(
51 &mut self,
52 key: &[u8],
53 read: impl for<'record> FnOnce(ProximityRecordRef<'record>) -> R,
54 ) -> Result<Option<R>, Error> {
55 self.directory
56 .get_with(key, |bytes| {
57 let stored = StoredRecordRef::decode(bytes, self.dimensions)?;
58 Ok(read(ProximityRecordRef {
59 vector: ProximityVectorRef::from_encoded(stored.vector),
60 value: stored.value,
61 }))
62 })?
63 .transpose()
64 }
65
66 pub fn get_lease(&mut self, key: &[u8]) -> Result<Option<OwnedValueLease>, Error> {
70 let lease = self.directory.get_lease(key)?;
71 if let Some(lease) = &lease {
72 StoredRecordRef::decode(lease.as_bytes()?, self.dimensions)?;
73 }
74 Ok(lease)
75 }
76
77 pub fn contains_key(&mut self, key: &[u8]) -> Result<bool, Error> {
78 Ok(self.get_with(key, |_| ())?.is_some())
79 }
80
81 pub fn scan_records(
82 &mut self,
83 mut visit: impl for<'record> FnMut(&[u8], ProximityRecordRef<'record>),
84 ) -> Result<u64, Error> {
85 Ok(self
86 .scan_records_until(|key, record| {
87 visit(key, record);
88 ControlFlow::<()>::Continue(())
89 })?
90 .visited)
91 }
92
93 pub fn scan_records_until<B>(
94 &mut self,
95 visit: impl for<'record> FnMut(&[u8], ProximityRecordRef<'record>) -> ControlFlow<B>,
96 ) -> Result<ScanOutcome<B>, Error> {
97 self.scan_records_range_until(&[], None, visit)
98 }
99
100 pub fn scan_records_range_until<B>(
104 &mut self,
105 start: &[u8],
106 end: Option<&[u8]>,
107 mut visit: impl for<'record> FnMut(&[u8], ProximityRecordRef<'record>) -> ControlFlow<B>,
108 ) -> Result<ScanOutcome<B>, Error> {
109 let dimensions = self.dimensions;
110 let outcome = self.directory.scan_range_until(start, end, |entry| {
111 let stored = match StoredRecordRef::decode(entry.value(), dimensions) {
112 Ok(stored) => stored,
113 Err(error) => return ControlFlow::Break(Err(error)),
114 };
115 let record = ProximityRecordRef {
116 vector: ProximityVectorRef::from_encoded(stored.vector),
117 value: stored.value,
118 };
119 match visit(entry.key(), record) {
120 ControlFlow::Continue(()) => ControlFlow::Continue(()),
121 ControlFlow::Break(value) => ControlFlow::Break(Ok(value)),
122 }
123 })?;
124 match outcome.break_value {
125 Some(Ok(value)) => Ok(ScanOutcome::stopped(outcome.visited, value)),
126 Some(Err(error)) => Err(error),
127 None => Ok(ScanOutcome::complete(outcome.visited)),
128 }
129 }
130}
131
132impl<S> ProximityMap<S>
133where
134 S: Store + Clone + Send + Sync,
135 S::Error: Send + Sync,
136{
137 pub fn build(
139 store: S,
140 config: ProximityConfig,
141 records: impl IntoIterator<Item = ProximityRecord>,
142 ) -> Result<Self, Error> {
143 Self::build_with_parallelism(store, config, records, BuildParallelism::default())
144 .map(|(map, _)| map)
145 }
146
147 pub fn build_with_parallelism(
149 store: S,
150 config: ProximityConfig,
151 records: impl IntoIterator<Item = ProximityRecord>,
152 parallelism: BuildParallelism,
153 ) -> Result<(Self, ProximityBuildStats), Error> {
154 config.validate()?;
155 let mut records: Vec<_> = records.into_iter().collect();
156 records.sort_by(|left, right| left.key.cmp(&right.key));
157 for pair in records.windows(2) {
158 if pair[0].key == pair[1].key {
159 return Err(Error::DuplicateProximityKey {
160 key: pair[0].key.clone(),
161 });
162 }
163 }
164
165 let directory_config = Config::default();
166 let mut directory_builder = BatchBuilder::new_with_origin(
167 store.clone(),
168 directory_config.clone(),
169 PublicationOrigin::Maintenance,
170 );
171 let mut indexed = Vec::with_capacity(records.len());
172 for record in records {
173 let stored = StoredRecord::new(
174 &record.vector,
175 record.value,
176 config.metric,
177 config.dimensions,
178 )?;
179 indexed.push(IndexedRecord {
180 key: record.key.clone(),
181 vector: stored.vector.clone(),
182 });
183 directory_builder.add(record.key, stored.encode());
184 }
185 let directory_tree = directory_builder.build()?;
186 let hierarchy = build_hierarchy_parallel(&indexed, &config, parallelism.threads())?;
187 let objects_written = put_missing_nodes(&store, &hierarchy.nodes)?;
188
189 let descriptor = Descriptor {
190 config: config.clone(),
191 count: indexed.len() as u64,
192 directory: directory_tree.clone(),
193 proximity_root: hierarchy.root.clone(),
194 };
195 let descriptor_bytes = descriptor.encode();
196 let descriptor_cid = Cid::from_bytes(&descriptor_bytes);
197 publish_maintenance_content(&store, &descriptor_cid, &descriptor_bytes)?;
198
199 let stats = ProximityBuildStats {
200 distance_evaluations: hierarchy.distance_evaluations,
201 proximity_objects: hierarchy.nodes.len(),
202 proximity_objects_written: objects_written,
203 };
204 Ok((
205 Self {
206 directory: Prolly::new(store.clone(), directory_config),
207 store,
208 tree: ProximityTree {
209 directory: directory_tree,
210 proximity_root: hierarchy.root,
211 descriptor: descriptor_cid,
212 count: indexed.len() as u64,
213 config,
214 },
215 node_cache: Mutex::new(ContentCache::new(DEFAULT_PROXIMITY_CACHE_NODES)),
216 },
217 stats,
218 ))
219 }
220
221 pub fn load(store: S, descriptor_cid: Cid) -> Result<Self, Error> {
223 let descriptor_bytes = load_content(&store, &descriptor_cid)?;
224 let descriptor = Descriptor::decode(&descriptor_bytes)?;
225 let root_bytes = load_content(&store, &descriptor.proximity_root)?;
226 if root_bytes.len() > descriptor.config.overflow.max_page_bytes as usize {
227 return Err(Error::InvalidProximityObject {
228 kind: "node",
229 reason: "root exceeds descriptor max_node_bytes".to_owned(),
230 });
231 }
232 let root = ProximityNode::decode(&root_bytes, descriptor.config.dimensions)?;
233 if root.subtree_count != descriptor.count {
234 return Err(Error::InvalidProximityObject {
235 kind: "descriptor",
236 reason: "record count disagrees with proximity root".to_owned(),
237 });
238 }
239 let directory_config = descriptor.directory.config.clone();
240 Ok(Self {
241 directory: Prolly::new(store.clone(), directory_config),
242 store,
243 tree: ProximityTree {
244 directory: descriptor.directory,
245 proximity_root: descriptor.proximity_root,
246 descriptor: descriptor_cid,
247 count: descriptor.count,
248 config: descriptor.config,
249 },
250 node_cache: Mutex::new(ContentCache::new(DEFAULT_PROXIMITY_CACHE_NODES)),
251 })
252 }
253
254 pub fn tree(&self) -> &ProximityTree {
256 &self.tree
257 }
258
259 pub fn clear_content_cache(&self) -> Result<(), Error> {
262 self.node_cache
263 .lock()
264 .map_err(|_| Error::InvalidProximityObject {
265 kind: "cache",
266 reason: "node cache lock poisoned".to_owned(),
267 })?
268 .clear();
269 Ok(())
270 }
271
272 pub fn get(&self, key: &[u8]) -> Result<Option<ExactProximityRecord>, Error> {
274 self.get_with(key, |record| record.to_owned())
275 }
276
277 pub fn read(&self) -> Result<ProximityReadSession<'_, S>, Error> {
279 Ok(ProximityReadSession {
280 directory: self.directory.read(&self.tree.directory)?,
281 dimensions: self.tree.config.dimensions,
282 })
283 }
284
285 pub fn get_with<R>(
287 &self,
288 key: &[u8],
289 read: impl for<'record> FnOnce(ProximityRecordRef<'record>) -> R,
290 ) -> Result<Option<R>, Error> {
291 self.read()?.get_with(key, read)
292 }
293
294 pub fn scan_records(
296 &self,
297 visit: impl for<'record> FnMut(&[u8], ProximityRecordRef<'record>),
298 ) -> Result<u64, Error> {
299 self.read()?.scan_records(visit)
300 }
301
302 pub fn scan_records_until<B>(
304 &self,
305 visit: impl for<'record> FnMut(&[u8], ProximityRecordRef<'record>) -> ControlFlow<B>,
306 ) -> Result<ScanOutcome<B>, Error> {
307 self.read()?.scan_records_until(visit)
308 }
309
310 pub fn scan_records_range_until<B>(
313 &self,
314 start: &[u8],
315 end: Option<&[u8]>,
316 visit: impl for<'record> FnMut(&[u8], ProximityRecordRef<'record>) -> ControlFlow<B>,
317 ) -> Result<ScanOutcome<B>, Error> {
318 self.read()?.scan_records_range_until(start, end, visit)
319 }
320
321 pub fn contains_key(&self, key: &[u8]) -> Result<bool, Error> {
323 self.read()?.contains_key(key)
324 }
325
326 pub fn rebuild_batch(
328 &self,
329 mutations: impl IntoIterator<Item = ProximityMutation>,
330 ) -> Result<Self, Error> {
331 let mutations = validate_mutations(mutations)?;
332 let mut records = self.collect_records()?;
333 apply_mutations(&mut records, &mutations, &self.tree.config)?;
334 Self::build(
335 self.store.clone(),
336 self.tree.config.clone(),
337 records.into_values(),
338 )
339 }
340
341 pub fn mutate_batch(
343 &self,
344 mutations: impl IntoIterator<Item = ProximityMutation>,
345 ) -> Result<(Self, ProximityMutationStats), Error> {
346 let mutations = validate_mutations(mutations)?;
347 if mutations.is_empty() {
348 return Ok((
349 Self::load(self.store.clone(), self.tree.descriptor.clone())?,
350 Default::default(),
351 ));
352 }
353 let keys: Vec<_> = mutations
354 .iter()
355 .map(|mutation| mutation.key.clone())
356 .collect();
357 let old_values = self.directory.get_many(&self.tree.directory, &keys)?;
358 let mut logical_edits = Vec::new();
359 let mut directory_mutations = Vec::with_capacity(mutations.len());
360 let mut count = self.tree.count;
361 for (mutation, old_bytes) in mutations.iter().zip(old_values) {
362 let old = old_bytes
363 .as_deref()
364 .map(|bytes| StoredRecord::decode(bytes, self.tree.config.dimensions))
365 .transpose()?;
366 let new = mutation
367 .value
368 .as_ref()
369 .map(|(vector, value)| {
370 StoredRecord::new(
371 vector,
372 value.clone(),
373 self.tree.config.metric,
374 self.tree.config.dimensions,
375 )
376 })
377 .transpose()?;
378 match (&old, &new) {
379 (None, Some(_)) => {
380 count = count
381 .checked_add(1)
382 .ok_or_else(|| Error::InvalidProximityObject {
383 kind: "mutation",
384 reason: "record count overflow".to_owned(),
385 })?
386 }
387 (Some(_), None) => count -= 1,
388 _ => {}
389 }
390 let old_vector = old.as_ref().map(|record| record.vector.clone());
391 let new_vector = new.as_ref().map(|record| record.vector.clone());
392 if old_vector != new_vector {
393 logical_edits.push(LogicalEdit {
394 key: mutation.key.clone(),
395 old: old_vector,
396 new: new_vector,
397 level: promotion_level(
398 &mutation.key,
399 self.tree.config.hierarchy.log_chunk_size,
400 self.tree.config.hierarchy.level_hash_seed,
401 ),
402 });
403 }
404 directory_mutations.push(match new {
405 Some(record) => TreeMutation::Upsert {
406 key: mutation.key.clone(),
407 val: record.encode(),
408 },
409 None => TreeMutation::Delete {
410 key: mutation.key.clone(),
411 },
412 });
413 }
414 let (directory_tree, directory_stats) = splice_with_origin(
415 &self.directory,
416 &self.tree.directory,
417 directory_mutations,
418 PublicationOrigin::Maintenance,
419 )?;
420
421 if logical_edits.is_empty() {
422 let descriptor = Descriptor {
423 config: self.tree.config.clone(),
424 count,
425 directory: directory_tree.clone(),
426 proximity_root: self.tree.proximity_root.clone(),
427 };
428 let bytes = descriptor.encode();
429 let descriptor_cid = Cid::from_bytes(&bytes);
430 publish_maintenance_content(&self.store, &descriptor_cid, &bytes)?;
431 let map = Self {
432 directory: Prolly::new(self.store.clone(), directory_tree.config.clone()),
433 store: self.store.clone(),
434 tree: ProximityTree {
435 directory: directory_tree,
436 proximity_root: self.tree.proximity_root.clone(),
437 descriptor: descriptor_cid,
438 count,
439 config: self.tree.config.clone(),
440 },
441 node_cache: Mutex::new(ContentCache::new(DEFAULT_PROXIMITY_CACHE_NODES)),
442 };
443 return Ok((
444 map,
445 ProximityMutationStats {
446 nodes_reused: 1,
447 directory_entries_scanned: directory_stats.entries_scanned,
448 directory_nodes_read: directory_stats.nodes_read,
449 directory_nodes_rebuilt: directory_stats.nodes_rebuilt,
450 directory_nodes_written: directory_stats.nodes_written,
451 directory_nodes_reused: directory_stats.nodes_reused,
452 directory_levels_rebuilt: directory_stats.levels_rebuilt,
453 directory_right_edge_rebuilt: directory_stats.right_edge_rebuilt,
454 ..Default::default()
455 },
456 ));
457 }
458
459 let (old_root, _) = self.load_node(&self.tree.proximity_root)?;
460 let max_edit_level = logical_edits
461 .iter()
462 .map(|edit| edit.level)
463 .max()
464 .unwrap_or(0);
465 let (proximity_root, nodes, mut stats) =
466 if old_root.entries.is_empty() || max_edit_level >= old_root.level {
467 let records = self.collect_records_from(&directory_tree)?;
468 let indexed: Vec<_> = records
469 .values()
470 .map(|record| IndexedRecord {
471 key: record.key.clone(),
472 vector: record.vector.clone(),
473 })
474 .collect();
475 let built = build_hierarchy(&indexed, &self.tree.config)?;
476 let stats = ProximityMutationStats {
477 records_rebuilt: indexed.len(),
478 distance_evaluations: built.distance_evaluations,
479 full_proximity_rebuild: true,
480 ..Default::default()
481 };
482 (built.root, built.nodes, stats)
483 } else {
484 let local = mutate_hierarchy(
485 &self.store,
486 &self.tree.proximity_root,
487 &self.tree.config,
488 &logical_edits,
489 )?;
490 (local.root, local.nodes, local.stats)
491 };
492 let pending_count = nodes.len();
493 let nodes_written = put_missing_nodes(&self.store, &nodes)?;
494 stats.nodes_written = nodes_written;
495 stats.nodes_reused += pending_count.saturating_sub(nodes_written);
496 apply_directory_stats(&mut stats, directory_stats);
497
498 let descriptor = Descriptor {
499 config: self.tree.config.clone(),
500 count,
501 directory: directory_tree.clone(),
502 proximity_root: proximity_root.clone(),
503 };
504 let descriptor_bytes = descriptor.encode();
505 let descriptor_cid = Cid::from_bytes(&descriptor_bytes);
506 publish_maintenance_content(&self.store, &descriptor_cid, &descriptor_bytes)?;
507 Ok((
508 Self {
509 directory: Prolly::new(self.store.clone(), directory_tree.config.clone()),
510 store: self.store.clone(),
511 tree: ProximityTree {
512 directory: directory_tree,
513 proximity_root,
514 descriptor: descriptor_cid,
515 count,
516 config: self.tree.config.clone(),
517 },
518 node_cache: Mutex::new(ContentCache::new(DEFAULT_PROXIMITY_CACHE_NODES)),
519 },
520 stats,
521 ))
522 }
523
524 pub fn search(&self, request: SearchRequest<'_>) -> Result<SearchResult, Error> {
526 self.search_with_trace(request, None)
527 }
528
529 pub fn search_with(
532 &self,
533 accelerators: &AcceleratorSet<S>,
534 search_io: &SearchIo<S>,
535 request: SearchRequest<'_>,
536 ) -> Result<SearchResult, Error> {
537 request.validate()?;
538 let physical_bytes_before = search_io.physical_bytes_read();
539 let eligibility = PreparedFilter::new(request.filter.clone(), &self.tree.directory)?;
540 let plan = plan_search(&self.tree, accelerators, &request, &eligibility)?;
541 search_io.runtime().load(
542 search_io,
543 super::super::content_graph::ContentObjectKind::ProximityDescriptor,
544 &self.tree.descriptor,
545 2,
546 |bytes| Descriptor::decode(bytes).map(|_| ()),
547 )?;
548 match &plan {
549 SearchPlan::Hnsw { .. } => {
550 let index = accelerators.hnsw().expect("planner validated HNSW");
551 search_io.runtime().load(
552 search_io,
553 super::super::content_graph::ContentObjectKind::HnswManifest,
554 index.manifest_cid(),
555 2,
556 |bytes| super::accelerator::hnsw::storage::Manifest::decode(bytes).map(|_| ()),
557 )?;
558 }
559 SearchPlan::ProductQuantized { .. } => {
560 let index = accelerators.pq().expect("planner validated PQ");
561 search_io.runtime().load(
562 search_io,
563 super::super::content_graph::ContentObjectKind::ProductQuantization,
564 index.manifest_cid(),
565 2,
566 |bytes| super::accelerator::pq::Manifest::decode(bytes).map(|_| ()),
567 )?;
568 }
569 SearchPlan::Composite { .. } => {
570 let accelerator = accelerators
571 .composite()
572 .expect("planner validated composite");
573 search_io.runtime().load(
574 search_io,
575 super::super::content_graph::ContentObjectKind::CompositeAccelerator,
576 accelerator.manifest_cid(),
577 1,
578 |bytes| super::accelerator::composite::Manifest::decode(bytes).map(|_| ()),
579 )?;
580 }
581 SearchPlan::Native | SearchPlan::EligibleExact { .. } => {}
582 }
583 let bound_map = ProximityMap::load(
584 search_io.for_kind_with_dimensions(
585 super::super::content_graph::ContentObjectKind::ProximityNode,
586 self.tree.config.dimensions,
587 ),
588 self.tree.descriptor.clone(),
589 )?;
590 let mut result = match &plan {
591 SearchPlan::Native => {
592 let mut native = request;
593 native.options.backend = SearchBackend::Native;
594 bound_map.search(native)
595 }
596 SearchPlan::EligibleExact {
597 key_count,
598 source_bound,
599 } => bound_map.search_eligible_exact(
600 &request,
601 &eligibility,
602 *key_count,
603 *source_bound,
604 plan,
605 ),
606 SearchPlan::Hnsw { .. } => {
607 let index = accelerators
608 .hnsw()
609 .ok_or_else(|| Error::InvalidProximitySearch {
610 reason: "planned HNSW accelerator is unavailable".to_owned(),
611 })?;
612 let index = index.rebind(
613 search_io.for_kind(super::super::content_graph::ContentObjectKind::HnswPage),
614 );
615 super::accelerator::hnsw::search::search_planned(&index, &bound_map, request, &plan)
616 }
617 SearchPlan::ProductQuantized { .. } => {
618 let index = accelerators
619 .pq()
620 .ok_or_else(|| Error::InvalidProximitySearch {
621 reason: "planned product-quantized accelerator is unavailable".to_owned(),
622 })?;
623 let index =
624 index.rebind(search_io.for_kind(
625 super::super::content_graph::ContentObjectKind::ProductQuantization,
626 ));
627 index.search_planned(&bound_map, request, &plan)
628 }
629 SearchPlan::Composite { .. } => self.search_composite(
630 accelerators
631 .composite()
632 .ok_or_else(|| Error::InvalidProximitySearch {
633 reason: "planned composite accelerator is unavailable".to_owned(),
634 })?,
635 search_io,
636 &bound_map,
637 request,
638 &eligibility,
639 &plan,
640 ),
641 }?;
642 result.stats.physical_bytes_read = search_io
643 .physical_bytes_read()
644 .saturating_sub(physical_bytes_before);
645 Ok(result)
646 }
647
648 pub(crate) fn search_composite(
649 &self,
650 composite: &super::accelerator::composite::CompositeAccelerator<S>,
651 search_io: &SearchIo<S>,
652 current: &ProximityMap<SearchIo<S>>,
653 request: SearchRequest<'_>,
654 eligibility: &PreparedFilter<'_>,
655 plan: &SearchPlan,
656 ) -> Result<SearchResult, Error> {
657 let SearchPlan::Composite {
658 base,
659 delta_records,
660 shadow_records,
661 merge_target,
662 } = plan
663 else {
664 return Err(Error::InvalidProximitySearch {
665 reason: "composite executor requires a composite plan".to_owned(),
666 });
667 };
668 if *delta_records != composite.delta_count as usize
669 || *shadow_records != composite.shadow_count as usize
670 || composite.current_source != self.tree.descriptor
671 {
672 return Err(Error::InvalidProximityObject {
673 kind: "composite accelerator",
674 reason: "plan or source binding disagrees with manifest".to_owned(),
675 });
676 }
677 let ordered_store =
678 search_io.for_kind(super::super::content_graph::ContentObjectKind::OrderedNode);
679 let shadow_manager =
680 Prolly::new(ordered_store.clone(), composite.shadow_tree.config.clone());
681 let delta_manager = Prolly::new(ordered_store, composite.delta_tree.config.clone());
682 let mut stats = ProximitySearchStats::default();
683 let mut shadow = BTreeSet::new();
684 for entry in shadow_manager.range(&composite.shadow_tree, &[], None)? {
685 let (key, value) = entry?;
686 if !value.is_empty() || !shadow.insert(key.clone()) {
687 return Err(Error::InvalidProximityObject {
688 kind: "composite shadow",
689 reason: "shadow tree contains a value or duplicate key".to_owned(),
690 });
691 }
692 if request
693 .budget
694 .max_nodes
695 .is_some_and(|limit| stats.nodes_read >= limit)
696 || request
697 .budget
698 .max_committed_bytes
699 .is_some_and(|limit| stats.committed_bytes.saturating_add(key.len()) > limit)
700 {
701 return Ok(SearchResult {
702 neighbors: Vec::new(),
703 stats,
704 completion: SearchCompletion::BudgetExhausted,
705 plan: plan.summary(),
706 });
707 }
708 stats.nodes_read += 1;
709 stats.bytes_read += key.len();
710 stats.committed_bytes += key.len();
711 }
712 if shadow.len() != *shadow_records {
713 return Err(Error::InvalidProximityObject {
714 kind: "composite shadow",
715 reason: "shadow tree cardinality disagrees with manifest".to_owned(),
716 });
717 }
718 if search_budget_exhausted(&request.budget, &stats) {
719 return Ok(SearchResult {
720 neighbors: Vec::new(),
721 stats,
722 completion: SearchCompletion::BudgetExhausted,
723 plan: plan.summary(),
724 });
725 }
726 let mut base_request = request.clone();
727 base_request.budget = remaining_budget(&request.budget, &stats);
728 let mut base_result = match &composite.base {
729 super::accelerator::composite::CompositeBase::Hnsw(index) => {
730 let index = index.rebind(
731 search_io.for_kind(super::super::content_graph::ContentObjectKind::HnswPage),
732 );
733 super::accelerator::hnsw::search::search_planned_with_exclusion(
734 &index,
735 current,
736 &composite.base_source,
737 base_request,
738 base,
739 |key| Ok(shadow.contains(key)),
740 )
741 }
742 super::accelerator::composite::CompositeBase::ProductQuantized(index) => {
743 let index =
744 index.rebind(search_io.for_kind(
745 super::super::content_graph::ContentObjectKind::ProductQuantization,
746 ));
747 index.search_planned_with_exclusion(
748 current,
749 &composite.base_source,
750 base_request,
751 base,
752 |key| Ok(shadow.contains(key)),
753 )
754 }
755 }?;
756 add_search_stats(&mut stats, &base_result.stats);
757 let mut completion = base_result.completion;
758 let query = prepare_vector(
759 self.tree.config.metric,
760 request.query,
761 self.tree.config.dimensions,
762 )?;
763 enum CompositeValue {
764 Owned { value: Vec<u8>, distance: f64 },
765 Retained(RerankCandidate),
766 }
767 impl CompositeValue {
768 fn distance(&self) -> f64 {
769 match self {
770 Self::Owned { distance, .. } => *distance,
771 Self::Retained(candidate) => candidate.distance,
772 }
773 }
774 }
775 let mut delta_seen = 0usize;
776 let mut merged = BTreeMap::<Vec<u8>, CompositeValue>::new();
777 for neighbor in base_result.neighbors.drain(..) {
778 if merged
779 .insert(
780 neighbor.key,
781 CompositeValue::Owned {
782 value: neighbor.value,
783 distance: neighbor.distance,
784 },
785 )
786 .is_some()
787 {
788 return Err(Error::InvalidProximityObject {
789 kind: "composite base",
790 reason: "base executor returned a duplicate key".to_owned(),
791 });
792 }
793 }
794 let mut current_directory = current
795 .directory_manager()
796 .read(¤t.tree().directory)?;
797 let mut retained_backings = HashSet::new();
798 let mut retained_bytes = 0usize;
799 for entry in delta_manager.range(&composite.delta_tree, &[], None)? {
800 let (key, bytes) = entry?;
801 delta_seen += 1;
802 if request
803 .budget
804 .max_nodes
805 .is_some_and(|limit| stats.nodes_read.saturating_add(1) > limit)
806 || request
807 .budget
808 .max_committed_bytes
809 .is_some_and(|limit| stats.committed_bytes.saturating_add(bytes.len()) > limit)
810 {
811 completion = SearchCompletion::BudgetExhausted;
812 break;
813 }
814 let record = StoredRecordRef::decode(&bytes, self.tree.config.dimensions)?;
815 stats.nodes_read += 1;
816 stats.bytes_read += bytes.len();
817 stats.committed_bytes += bytes.len();
818 if !eligibility.contains(&key) {
819 continue;
820 }
821 if request
822 .budget
823 .max_nodes
824 .is_some_and(|limit| stats.nodes_read.saturating_add(1) > limit)
825 || request
826 .budget
827 .max_distance_evaluations
828 .is_some_and(|limit| {
829 stats
830 .distance_evaluations
831 .saturating_add(stats.quantized_distance_evaluations)
832 .saturating_add(1)
833 > limit
834 })
835 {
836 completion = SearchCompletion::BudgetExhausted;
837 break;
838 }
839 let Some(handle) = current_directory.get_handle(&key)? else {
840 return Err(Error::InvalidProximityObject {
841 kind: "composite delta",
842 reason: "delta key is absent from current source".to_owned(),
843 });
844 };
845 let authoritative_bytes = handle.value()?.len();
846 if request.budget.max_committed_bytes.is_some_and(|limit| {
847 stats.committed_bytes.saturating_add(authoritative_bytes) > limit
848 }) {
849 completion = SearchCompletion::BudgetExhausted;
850 break;
851 }
852 let authoritative =
853 StoredRecordRef::decode(handle.value()?, self.tree.config.dimensions)?;
854 if !encoded_vectors_equal(authoritative.vector, record.vector) {
855 return Err(Error::InvalidProximityObject {
856 kind: "composite delta",
857 reason: "delta vector disagrees with current source".to_owned(),
858 });
859 }
860 let distance = record
861 .vector
862 .score(request.kernel, self.tree.config.metric, &query);
863 stats.nodes_read += 1;
864 stats.bytes_read += authoritative_bytes;
865 stats.committed_bytes += authoritative_bytes;
866 stats.distance_evaluations += 1;
867 stats.reranked_candidates += 1;
868 let candidate = RerankCandidate::new(handle, &key, distance)?;
869 if retained_backings.insert(candidate.backing_id()) {
870 retained_bytes = retained_bytes.saturating_add(candidate.retained_bytes());
871 }
872 if merged
873 .insert(key, CompositeValue::Retained(candidate))
874 .is_some()
875 {
876 return Err(Error::InvalidProximityObject {
877 kind: "composite accelerator",
878 reason: "delta key was not shadowed from the base".to_owned(),
879 });
880 }
881 stats.candidate_handles_peak =
882 stats.candidate_handles_peak.max(retained_backings.len());
883 stats.candidate_retained_bytes_peak =
884 stats.candidate_retained_bytes_peak.max(retained_bytes);
885 }
886 if completion != SearchCompletion::BudgetExhausted && delta_seen != *delta_records {
887 return Err(Error::InvalidProximityObject {
888 kind: "composite delta",
889 reason: "delta tree cardinality disagrees with manifest".to_owned(),
890 });
891 }
892 let mut candidates = merged.into_iter().collect::<Vec<_>>();
893 candidates.sort_by(|(left_key, left), (right_key, right)| {
894 left.distance()
895 .total_cmp(&right.distance())
896 .then_with(|| left_key.cmp(right_key))
897 });
898 candidates.truncate((*merge_target).min(request.k));
899 let neighbors = candidates
900 .into_iter()
901 .map(|(key, candidate)| match candidate {
902 CompositeValue::Owned { value, distance } => Ok(Neighbor {
903 key,
904 value,
905 distance,
906 }),
907 CompositeValue::Retained(candidate) => {
908 let record = candidate.record(self.tree.config.dimensions)?;
909 Ok(Neighbor {
910 key,
911 value: record.value.to_vec(),
912 distance: candidate.distance,
913 })
914 }
915 })
916 .collect::<Result<Vec<_>, Error>>()?;
917 Ok(SearchResult {
918 neighbors,
919 stats,
920 completion,
921 plan: plan.summary(),
922 })
923 }
924
925 fn search_eligible_exact(
926 &self,
927 request: &SearchRequest<'_>,
928 eligibility: &PreparedFilter<'_>,
929 key_count: u64,
930 source_bound: bool,
931 plan: SearchPlan,
932 ) -> Result<SearchResult, Error> {
933 let Some((keys, prepared_source_bound)) = eligibility.sorted_keys() else {
934 return Err(Error::InvalidProximitySearch {
935 reason: "eligible-exact plan requires sorted eligible keys".to_owned(),
936 });
937 };
938 if key_count != keys.len() as u64 || source_bound != prepared_source_bound {
939 return Err(Error::InvalidProximitySearch {
940 reason: "eligible-exact plan disagrees with prepared eligibility".to_owned(),
941 });
942 }
943 let query = prepare_vector(
944 self.tree.config.metric,
945 request.query,
946 self.tree.config.dimensions,
947 )?;
948 let mut stats = ProximitySearchStats::default();
949 let candidate_limit = request
950 .budget
951 .max_frontier_entries
952 .unwrap_or(request.k)
953 .min(request.k);
954 let mut completion = if candidate_limit < request.k.min(keys.len()) {
955 SearchCompletion::BudgetExhausted
956 } else {
957 SearchCompletion::Exact
958 };
959 let mut candidates = Vec::<RerankCandidate>::with_capacity(candidate_limit);
960 let mut directory = self.directory.read(&self.tree.directory)?;
961 for key in keys {
962 if request
963 .budget
964 .max_nodes
965 .is_some_and(|limit| stats.nodes_read >= limit)
966 || request
967 .budget
968 .max_distance_evaluations
969 .is_some_and(|limit| stats.distance_evaluations >= limit)
970 {
971 completion = SearchCompletion::BudgetExhausted;
972 break;
973 }
974 stats.nodes_read += 1;
975 let Some(handle) = directory.get_handle(key)? else {
976 if source_bound {
977 return Err(Error::InvalidProximityObject {
978 kind: "eligible keys",
979 reason:
980 "source-bound eligible key is absent from the authoritative directory"
981 .to_owned(),
982 });
983 }
984 continue;
985 };
986 let bytes = handle.value()?.len();
987 if request
988 .budget
989 .max_committed_bytes
990 .is_some_and(|limit| stats.committed_bytes.saturating_add(bytes) > limit)
991 {
992 completion = SearchCompletion::BudgetExhausted;
993 break;
994 }
995 let record = StoredRecordRef::decode(handle.value()?, self.tree.config.dimensions)?;
996 stats.bytes_read = stats.bytes_read.saturating_add(bytes);
997 stats.committed_bytes = stats.committed_bytes.saturating_add(bytes);
998 stats.distance_evaluations += 1;
999 let distance = record
1000 .vector
1001 .score(request.kernel, self.tree.config.metric, &query);
1002 insert_reranked_top_k(
1003 &mut candidates,
1004 RerankCandidate::new(handle, key, distance)?,
1005 candidate_limit,
1006 );
1007 stats.frontier_peak = stats.frontier_peak.max(candidates.len());
1008 stats.candidate_handles_peak = stats.candidate_handles_peak.max(candidates.len());
1009 stats.candidate_retained_bytes_peak = stats
1010 .candidate_retained_bytes_peak
1011 .max(retained_candidate_bytes(&candidates));
1012 }
1013 stats.reranked_candidates = stats.distance_evaluations;
1014 let neighbors = candidates
1015 .into_iter()
1016 .map(|candidate| candidate.into_neighbor(self.tree.config.dimensions))
1017 .collect::<Result<Vec<_>, Error>>()?;
1018 Ok(SearchResult {
1019 neighbors,
1020 stats,
1021 completion,
1022 plan: plan.summary(),
1023 })
1024 }
1025
1026 pub(super) fn search_with_trace(
1027 &self,
1028 request: SearchRequest<'_>,
1029 mut trace: Option<&mut Vec<super::proof::ProximitySearchEvent>>,
1030 ) -> Result<SearchResult, Error> {
1031 request.validate()?;
1032 if matches!(
1033 request.options.backend,
1034 SearchBackend::ProductQuantized | SearchBackend::Hnsw
1035 ) {
1036 return Err(Error::InvalidProximitySearch {
1037 reason: "requested backend requires a validated accelerator sidecar".to_owned(),
1038 });
1039 }
1040 let filter = PreparedFilter::new(request.filter.clone(), &self.tree.directory)?;
1041 let query = prepare_vector(
1042 self.tree.config.metric,
1043 request.query,
1044 self.tree.config.dimensions,
1045 )?;
1046 let use_scalar_quantization =
1047 super::accelerator::sq8::enabled(&self.tree.config, request.policy);
1048 let mut stats = ProximitySearchStats::default();
1049 let mut frontier = BinaryHeap::new();
1050 frontier.push(FrontierEntry {
1051 bound: 0.0,
1052 score: 0.0,
1053 key: Vec::new(),
1054 cid: self.tree.proximity_root.clone(),
1055 expected_level: None,
1056 });
1057 if let Some(trace) = trace.as_deref_mut() {
1058 trace.push(super::proof::ProximitySearchEvent::FrontierPushed {
1059 cid: self.tree.proximity_root.clone(),
1060 bound_bits: 0.0f64.to_bits(),
1061 });
1062 }
1063 let mut candidates = Vec::<SearchCandidate>::new();
1064 let mut score_cache = HashMap::<Vec<u8>, f64>::new();
1067 let mut visited = HashSet::new();
1068 let mut levels = HashSet::new();
1069 let mut last_fanout = 0usize;
1070 let mut completion = SearchCompletion::Exact;
1071
1072 while let Some(next) = frontier.peek() {
1073 if !use_scalar_quantization
1074 && self.tree.config.metric == DistanceMetric::L2Squared
1075 && candidates.len() == request.k
1076 && next.bound > candidates.last().expect("full top-k").score
1077 {
1078 break;
1079 }
1080 if let SearchPolicy::Adaptive(quality) = request.policy {
1081 if candidates.last().is_some_and(|worst| {
1082 let overlapping = frontier
1083 .iter()
1084 .filter(|entry| entry.bound <= worst.score)
1085 .count();
1086 adaptive_should_stop(
1087 quality,
1088 AdaptiveContext {
1089 results: candidates.len(),
1090 k: request.k,
1091 frontier_bound: next.bound,
1092 worst_score: worst.score,
1093 overlapping_clusters: overlapping,
1094 logical_level: next.expected_level.unwrap_or(u8::MAX),
1095 last_fanout,
1096 cluster_count: frontier.len(),
1097 },
1098 )
1099 }) {
1100 completion = SearchCompletion::ApproximatePolicySatisfied;
1101 break;
1102 }
1103 }
1104 if request
1105 .budget
1106 .max_nodes
1107 .is_some_and(|maximum| stats.nodes_read >= maximum)
1108 {
1109 completion = SearchCompletion::BudgetExhausted;
1110 break;
1111 }
1112 let next = frontier.pop().expect("peeked frontier");
1113 if let Some(trace) = trace.as_deref_mut() {
1114 trace.push(super::proof::ProximitySearchEvent::FrontierPopped {
1115 cid: next.cid.clone(),
1116 bound_bits: next.bound.to_bits(),
1117 });
1118 }
1119 if !visited.insert(next.cid.clone()) {
1120 return Err(Error::InvalidProximityObject {
1121 kind: "node",
1122 reason: "cycle or repeated child ownership".to_owned(),
1123 });
1124 }
1125 let (node, mut bytes) = self.load_node(&next.cid)?;
1126 if let Some(trace) = trace.as_deref_mut() {
1127 trace.push(super::proof::ProximitySearchEvent::VisitedObject(
1128 next.cid.clone(),
1129 ));
1130 }
1131 let quantizer = if use_scalar_quantization && node.kind.has_children(node.level) {
1132 let (quantizer, quantizer_bytes) = self.load_scalar_quantizer(&node)?;
1133 bytes = bytes.saturating_add(quantizer_bytes);
1134 Some(quantizer)
1135 } else {
1136 None
1137 };
1138 if next
1139 .expected_level
1140 .is_some_and(|expected| node.level != expected)
1141 {
1142 return Err(Error::InvalidProximityObject {
1143 kind: "node",
1144 reason: "child has an unexpected logical level".to_owned(),
1145 });
1146 }
1147 stats.bytes_read = stats.bytes_read.saturating_add(bytes);
1148 if request
1149 .budget
1150 .max_committed_bytes
1151 .is_some_and(|maximum| stats.committed_bytes.saturating_add(bytes) > maximum)
1152 {
1153 completion = SearchCompletion::BudgetExhausted;
1154 break;
1155 }
1156 stats.nodes_read += 1;
1157 stats.committed_bytes += bytes;
1158 last_fanout = node.entries.len();
1159 levels.insert(node.level);
1160 stats.levels_visited = levels.len();
1161
1162 for (entry_index, entry) in node.entries.iter().enumerate() {
1163 if node.kind.has_children(node.level) {
1164 if !filter.intersects(&entry.min_key, &entry.max_key) {
1165 continue;
1166 }
1167 let Some(child) = &entry.child else {
1168 return Err(Error::InvalidProximityObject {
1169 kind: "node",
1170 reason: "internal entry has no child".to_owned(),
1171 });
1172 };
1173 let representative_score = if let Some(quantizer) = &quantizer {
1174 if distance_budget_exhausted(&request, &stats) {
1175 completion = SearchCompletion::BudgetExhausted;
1176 break;
1177 }
1178 stats.quantized_distance_evaluations += 1;
1179 quantizer.approximate_score(self.tree.config.metric, &query, entry_index)?
1180 } else {
1181 match score_cache.get(&entry.key) {
1182 Some(score) => *score,
1183 None => {
1184 if distance_budget_exhausted(&request, &stats) {
1185 completion = SearchCompletion::BudgetExhausted;
1186 break;
1187 }
1188 stats.distance_evaluations += 1;
1189 let value = query_score(
1190 request.kernel,
1191 self.tree.config.metric,
1192 &query,
1193 entry.vector.inline()?,
1194 );
1195 score_cache.insert(entry.key.clone(), value);
1196 value
1197 }
1198 }
1199 };
1200 let bound = if quantizer.is_none()
1201 && self.tree.config.metric == DistanceMetric::L2Squared
1202 {
1203 super::distance::canonical::l2_lower_bound_down(
1204 representative_score,
1205 entry.covering_radius,
1206 )
1207 } else {
1208 representative_score
1209 };
1210 if request
1211 .budget
1212 .max_frontier_entries
1213 .is_some_and(|maximum| frontier.len() >= maximum)
1214 {
1215 completion = SearchCompletion::BudgetExhausted;
1216 break;
1217 }
1218 frontier.push(FrontierEntry {
1219 bound,
1220 score: representative_score,
1221 key: entry.key.clone(),
1222 cid: child.clone(),
1223 expected_level: Some(if node.kind == PhysicalNodeKind::OverflowDirectory {
1224 node.level
1225 } else {
1226 node.level - 1
1227 }),
1228 });
1229 if let Some(trace) = trace.as_deref_mut() {
1230 trace.push(super::proof::ProximitySearchEvent::FrontierPushed {
1231 cid: child.clone(),
1232 bound_bits: bound.to_bits(),
1233 });
1234 }
1235 stats.frontier_peak = stats.frontier_peak.max(frontier.len());
1236 } else if filter.contains(&entry.key) {
1237 let leaf_score = match score_cache.get(&entry.key) {
1238 Some(score) => *score,
1239 None => {
1240 if distance_budget_exhausted(&request, &stats) {
1241 completion = SearchCompletion::BudgetExhausted;
1242 break;
1243 }
1244 stats.distance_evaluations += 1;
1245 let value = query_score(
1246 request.kernel,
1247 self.tree.config.metric,
1248 &query,
1249 entry.vector.inline()?,
1250 );
1251 score_cache.insert(entry.key.clone(), value);
1252 value
1253 }
1254 };
1255 if let Some(trace) = trace.as_deref_mut() {
1256 trace.push(super::proof::ProximitySearchEvent::CandidateScored {
1257 key: entry.key.clone(),
1258 distance_bits: leaf_score.to_bits(),
1259 });
1260 }
1261 insert_top_k(
1262 &mut candidates,
1263 SearchCandidate::new(node.clone(), entry_index, leaf_score),
1264 request.k,
1265 );
1266 }
1267 }
1268 if completion == SearchCompletion::BudgetExhausted {
1269 break;
1270 }
1271 }
1272
1273 stats.candidate_handles_peak = stats.candidate_handles_peak.max(candidates.len());
1274 stats.candidate_retained_bytes_peak = stats
1275 .candidate_retained_bytes_peak
1276 .max(retained_search_candidate_bytes(&candidates));
1277 let keys = candidates
1278 .iter()
1279 .map(SearchCandidate::key)
1280 .collect::<Result<Vec<_>, Error>>()?;
1281 if use_scalar_quantization {
1282 stats.reranked_candidates = candidates.len();
1283 }
1284 let mut neighbors = Vec::with_capacity(candidates.len());
1285 let mut rerank_error = None;
1286 let mut directory = self.directory.read(&self.tree.directory)?;
1287 directory.get_many_with(&keys, |position, _, stored| {
1288 if rerank_error.is_some() {
1289 return;
1290 }
1291 let result = (|| {
1292 let candidate =
1293 candidates
1294 .get(position)
1295 .ok_or_else(|| Error::InvalidProximityObject {
1296 kind: "candidate",
1297 reason: "directory multi-get returned an invalid position".to_owned(),
1298 })?;
1299 let bytes = stored.ok_or_else(|| Error::InvalidProximityObject {
1300 kind: "node",
1301 reason: "leaf key is absent from exact directory".to_owned(),
1302 })?;
1303 let record = StoredRecordRef::decode(bytes, self.tree.config.dimensions)?;
1304 if !encoded_vector_matches(record.vector, candidate.vector()?) {
1305 return Err(Error::InvalidProximityObject {
1306 kind: "node",
1307 reason: "leaf vector disagrees with exact directory".to_owned(),
1308 });
1309 }
1310 neighbors.push(Neighbor {
1311 key: candidate.key()?.to_vec(),
1312 value: record.value.to_vec(),
1313 distance: candidate.score,
1314 });
1315 Ok(())
1316 })();
1317 if let Err(error) = result {
1318 rerank_error = Some(error);
1319 }
1320 })?;
1321 if let Some(error) = rerank_error {
1322 return Err(error);
1323 }
1324 if let Some(trace) = trace {
1325 trace.push(super::proof::ProximitySearchEvent::Completed(completion));
1326 }
1327 Ok(SearchResult {
1328 neighbors,
1329 stats,
1330 completion,
1331 plan: super::search::SearchPlan::Native.summary(),
1332 })
1333 }
1334
1335 pub fn verify(&self) -> Result<ProximityVerification, Error> {
1337 let records = self.collect_records()?;
1338 let root_bytes = load_content(&self.store, &self.tree.proximity_root)?;
1339 let root = ProximityNode::decode(&root_bytes, self.tree.config.dimensions)?;
1340 let mut state = VerificationState {
1341 records: &records,
1342 seen_nodes: HashSet::new(),
1343 seen_external_vectors: HashSet::new(),
1344 seen_scalar_quantizers: HashSet::new(),
1345 seen_leaf_keys: HashSet::new(),
1346 summary: ProximityVerification {
1347 record_count: self.tree.count,
1348 maximum_level: root.level,
1349 ..Default::default()
1350 },
1351 };
1352 let verified = self.verify_node(
1353 &self.tree.proximity_root,
1354 Some(root.level),
1355 None,
1356 &mut state,
1357 )?;
1358 if verified.count != self.tree.count || records.len() as u64 != self.tree.count {
1359 return Err(Error::InvalidProximityObject {
1360 kind: "descriptor",
1361 reason: "logical counts disagree".to_owned(),
1362 });
1363 }
1364 if state.seen_leaf_keys.len() != records.len()
1365 || records
1366 .keys()
1367 .any(|key| !state.seen_leaf_keys.contains(key))
1368 {
1369 return Err(Error::InvalidProximityObject {
1370 kind: "node",
1371 reason: "leaf identities do not match the exact directory".to_owned(),
1372 });
1373 }
1374 Ok(state.summary)
1375 }
1376
1377 fn load_node(&self, cid: &Cid) -> Result<(Arc<ProximityNode>, usize), Error> {
1378 if let Some((node, bytes)) = self
1379 .node_cache
1380 .lock()
1381 .map_err(|_| Error::InvalidProximityObject {
1382 kind: "cache",
1383 reason: "node cache lock poisoned".to_owned(),
1384 })?
1385 .get(cid)
1386 {
1387 return Ok((node, bytes));
1388 }
1389 let bytes = load_content(&self.store, cid)?;
1390 if bytes.len() > self.tree.config.overflow.max_page_bytes as usize {
1391 return Err(Error::InvalidProximityObject {
1392 kind: "node",
1393 reason: "node exceeds descriptor max_node_bytes".to_owned(),
1394 });
1395 }
1396 let len = bytes.len();
1397 let mut node = ProximityNode::decode(&bytes, self.tree.config.dimensions)?;
1398 let vector_bytes = self.resolve_external_vectors(&mut node)?;
1399 let node = Arc::new(node);
1400 self.node_cache
1401 .lock()
1402 .map_err(|_| Error::InvalidProximityObject {
1403 kind: "cache",
1404 reason: "node cache lock poisoned".to_owned(),
1405 })?
1406 .insert(cid.clone(), node.clone(), len + vector_bytes);
1407 Ok((node, len + vector_bytes))
1408 }
1409
1410 fn resolve_external_vectors(&self, node: &mut ProximityNode) -> Result<usize, Error> {
1411 let mut bytes_read = 0usize;
1412 for entry in &mut node.entries {
1413 let VectorRef::External(cid) = &entry.vector else {
1414 continue;
1415 };
1416 let bytes = load_content(&self.store, cid)?;
1417 let external = ExternalVector::decode(&bytes)?;
1418 if external.vector.len() != self.tree.config.dimensions as usize {
1419 return Err(Error::InvalidProximityObject {
1420 kind: "vector",
1421 reason: "external vector dimension mismatch".to_owned(),
1422 });
1423 }
1424 bytes_read += bytes.len();
1425 entry.vector = VectorRef::Inline(external.vector);
1426 }
1427 Ok(bytes_read)
1428 }
1429
1430 fn load_scalar_quantizer(
1431 &self,
1432 node: &ProximityNode,
1433 ) -> Result<(ScalarQuantized, usize), Error> {
1434 let config = self
1435 .tree
1436 .config
1437 .scalar_quantization
1438 .as_ref()
1439 .ok_or_else(|| Error::InvalidProximityObject {
1440 kind: "quantizer",
1441 reason: "quantized search requires descriptor configuration".to_owned(),
1442 })?;
1443 let cid = node
1444 .quantizer
1445 .as_ref()
1446 .ok_or_else(|| Error::InvalidProximityObject {
1447 kind: "quantizer",
1448 reason: "configured node has no scalar quantizer".to_owned(),
1449 })?;
1450 let bytes = load_content(&self.store, cid)?;
1451 let quantizer = ScalarQuantized::decode(&bytes)?;
1452 if quantizer.dimensions != self.tree.config.dimensions
1453 || quantizer.group_size != config.group_size
1454 {
1455 return Err(Error::InvalidProximityObject {
1456 kind: "quantizer",
1457 reason: "quantizer configuration disagrees with descriptor".to_owned(),
1458 });
1459 }
1460 if quantizer.entry_count != node.entries.len() as u64 {
1461 return Err(Error::InvalidProximityObject {
1462 kind: "quantizer",
1463 reason: "quantizer entry count disagrees with node".to_owned(),
1464 });
1465 }
1466 Ok((quantizer, bytes.len()))
1467 }
1468
1469 pub(crate) fn collect_records(&self) -> Result<BTreeMap<Vec<u8>, ProximityRecord>, Error> {
1470 self.collect_records_from(&self.tree.directory)
1471 }
1472
1473 pub(crate) fn store_clone(&self) -> S {
1474 self.store.clone()
1475 }
1476
1477 pub(super) fn directory_manager(&self) -> &Prolly<S> {
1478 &self.directory
1479 }
1480
1481 pub(super) fn load_descriptor_bytes(&self) -> Result<Vec<u8>, Error> {
1482 load_content(&self.store, &self.tree.descriptor)
1483 }
1484
1485 fn collect_records_from(
1486 &self,
1487 directory: &super::super::tree::Tree,
1488 ) -> Result<BTreeMap<Vec<u8>, ProximityRecord>, Error> {
1489 let mut records = BTreeMap::new();
1490 let mut decode_error = None;
1491 self.directory
1492 .scan_range_until(directory, &[], None, |entry| {
1493 let stored =
1494 match StoredRecordRef::decode(entry.value(), self.tree.config.dimensions) {
1495 Ok(stored) => stored,
1496 Err(error) => {
1497 decode_error = Some(error);
1498 return ControlFlow::Break(());
1499 }
1500 };
1501 let key = entry.key().to_vec();
1502 records.insert(
1503 key.clone(),
1504 ProximityRecord {
1505 key,
1506 vector: ProximityVectorRef::from_encoded(stored.vector).to_vec(),
1507 value: stored.value.to_vec(),
1508 },
1509 );
1510 ControlFlow::Continue(())
1511 })?;
1512 if let Some(error) = decode_error {
1513 return Err(error);
1514 }
1515 Ok(records)
1516 }
1517
1518 fn verify_node(
1519 &self,
1520 cid: &Cid,
1521 expected_level: Option<u8>,
1522 parent: Option<(
1523 &super::storage::ProximityEntry,
1524 &[super::storage::ProximityEntry],
1525 )>,
1526 state: &mut VerificationState<'_>,
1527 ) -> Result<VerifiedSubtree, Error> {
1528 if !state.seen_nodes.insert(cid.clone()) {
1529 return Err(Error::InvalidProximityObject {
1530 kind: "node",
1531 reason: "cycle or repeated child ownership".to_owned(),
1532 });
1533 }
1534 let bytes = load_content(&self.store, cid)?;
1535 if bytes.len() > self.tree.config.overflow.max_page_bytes as usize {
1536 return Err(Error::InvalidProximityObject {
1537 kind: "node",
1538 reason: "node exceeds descriptor max_node_bytes".to_owned(),
1539 });
1540 }
1541 let mut node = ProximityNode::decode(&bytes, self.tree.config.dimensions)?;
1542 for entry in &node.entries {
1543 if let VectorRef::External(vector) = &entry.vector {
1544 if state.seen_external_vectors.insert(vector.clone()) {
1545 state.summary.external_vector_count += 1;
1546 }
1547 }
1548 }
1549 self.resolve_external_vectors(&mut node)?;
1550 match (&self.tree.config.scalar_quantization, &node.quantizer) {
1551 (None, None) => {}
1552 (Some(config), Some(cid)) => {
1553 let quantizer_bytes = load_content(&self.store, cid)?;
1554 let quantizer = ScalarQuantized::decode(&quantizer_bytes)?;
1555 if quantizer.dimensions != self.tree.config.dimensions
1556 || quantizer.group_size != config.group_size
1557 {
1558 return Err(Error::InvalidProximityObject {
1559 kind: "quantizer",
1560 reason: "quantizer configuration disagrees with descriptor".to_owned(),
1561 });
1562 }
1563 let vectors = node
1564 .entries
1565 .iter()
1566 .map(|entry| entry.vector.inline())
1567 .collect::<Result<Vec<_>, _>>()?;
1568 quantizer.verify(&vectors)?;
1569 state.summary.quantized_node_count += 1;
1570 if state.seen_scalar_quantizers.insert(cid.clone()) {
1571 state.summary.scalar_quantizer_count += 1;
1572 }
1573 }
1574 _ => {
1575 return Err(Error::InvalidProximityObject {
1576 kind: "quantizer",
1577 reason: "node quantizer presence disagrees with descriptor".to_owned(),
1578 })
1579 }
1580 }
1581 if expected_level != Some(node.level) {
1582 return Err(Error::InvalidProximityObject {
1583 kind: "node",
1584 reason: "unexpected logical level".to_owned(),
1585 });
1586 }
1587 state.summary.proximity_node_count += 1;
1588 match node.kind {
1589 PhysicalNodeKind::OverflowPage => state.summary.overflow_page_count += 1,
1590 PhysicalNodeKind::OverflowDirectory => state.summary.overflow_directory_count += 1,
1591 PhysicalNodeKind::Leaf | PhysicalNodeKind::Route => {}
1592 }
1593 state.summary.maximum_node_bytes = state.summary.maximum_node_bytes.max(bytes.len());
1594
1595 if node.kind != PhysicalNodeKind::OverflowDirectory {
1596 if let Some((selected, candidates)) = parent {
1597 for entry in &node.entries {
1598 if entry.key == selected.key {
1602 continue;
1603 }
1604 let selected_distance = score(
1605 self.tree.config.metric,
1606 entry.vector.inline()?,
1607 selected.vector.inline()?,
1608 );
1609 for candidate in candidates {
1610 state.summary.distance_checks += 1;
1611 let candidate_distance = score(
1612 self.tree.config.metric,
1613 entry.vector.inline()?,
1614 candidate.vector.inline()?,
1615 );
1616 let candidate_is_better = candidate_distance
1617 .total_cmp(&selected_distance)
1618 .then_with(|| candidate.key.cmp(&selected.key))
1619 .is_lt();
1620 if candidate_is_better {
1621 return Err(Error::InvalidProximityObject {
1622 kind: "node",
1623 reason: "nearest-representative invariant violated".to_owned(),
1624 });
1625 }
1626 }
1627 }
1628 }
1629 }
1630
1631 if node.kind != PhysicalNodeKind::OverflowDirectory {
1632 for entry in &node.entries {
1633 if super::vector::promotion_level(
1634 &entry.key,
1635 self.tree.config.hierarchy.log_chunk_size,
1636 self.tree.config.hierarchy.level_hash_seed,
1637 ) < node.level
1638 {
1639 return Err(Error::InvalidProximityObject {
1640 kind: "node",
1641 reason: "entry appears above its deterministic promotion level".to_owned(),
1642 });
1643 }
1644 }
1645 }
1646
1647 let verified = if node.kind.is_logical_leaf(node.level) {
1648 let mut points = Vec::with_capacity(node.entries.len());
1649 for entry in &node.entries {
1650 if !state.seen_leaf_keys.insert(entry.key.clone()) {
1651 return Err(Error::InvalidProximityObject {
1652 kind: "node",
1653 reason: "duplicate leaf identity".to_owned(),
1654 });
1655 }
1656 let record =
1657 state
1658 .records
1659 .get(&entry.key)
1660 .ok_or_else(|| Error::InvalidProximityObject {
1661 kind: "node",
1662 reason: "leaf key is absent from exact directory".to_owned(),
1663 })?;
1664 if record.vector.as_slice() != entry.vector.inline()? {
1665 return Err(Error::InvalidProximityObject {
1666 kind: "node",
1667 reason: "leaf vector disagrees with exact directory".to_owned(),
1668 });
1669 }
1670 points.push((entry.key.clone(), entry.vector.inline()?.to_vec()));
1671 }
1672 VerifiedSubtree::from_points(node.entries.len() as u64, points)
1673 } else {
1674 let mut count = 0u64;
1675 let mut points = Vec::new();
1676 let mut minimum: Option<Vec<u8>> = None;
1677 let mut maximum: Option<Vec<u8>> = None;
1678 for entry in &node.entries {
1679 let child = entry
1680 .child
1681 .as_ref()
1682 .ok_or_else(|| Error::InvalidProximityObject {
1683 kind: "node",
1684 reason: "internal entry has no child".to_owned(),
1685 })?;
1686 let child_verified = self.verify_node(
1687 child,
1688 Some(if node.kind == PhysicalNodeKind::OverflowDirectory {
1689 node.level
1690 } else {
1691 node.level - 1
1692 }),
1693 if node.kind == PhysicalNodeKind::OverflowDirectory {
1694 parent
1695 } else {
1696 Some((entry, &node.entries))
1697 },
1698 state,
1699 )?;
1700 if child_verified.count != entry.child_count {
1701 return Err(Error::InvalidProximityObject {
1702 kind: "node",
1703 reason: "child count summary mismatch".to_owned(),
1704 });
1705 }
1706 if child_verified.minimum.as_deref() != Some(entry.min_key.as_slice())
1707 || child_verified.maximum.as_deref() != Some(entry.max_key.as_slice())
1708 {
1709 return Err(Error::InvalidProximityObject {
1710 kind: "node",
1711 reason: "child key-bound summary mismatch".to_owned(),
1712 });
1713 }
1714 for (_, vector) in &child_verified.points {
1715 let required = super::distance::euclidean_radius_up(
1716 score(
1717 super::DistanceMetric::L2Squared,
1718 entry.vector.inline()?,
1719 vector,
1720 ),
1721 0.0,
1722 );
1723 if required > entry.covering_radius {
1724 return Err(Error::InvalidProximityObject {
1725 kind: "node",
1726 reason: "covering-radius summary is not conservative".to_owned(),
1727 });
1728 }
1729 }
1730 count = count.checked_add(child_verified.count).ok_or_else(|| {
1731 Error::InvalidProximityObject {
1732 kind: "node",
1733 reason: "subtree count overflow".to_owned(),
1734 }
1735 })?;
1736 if minimum.as_ref().is_none_or(|key| entry.min_key < *key) {
1737 minimum = Some(entry.min_key.clone());
1738 }
1739 if maximum.as_ref().is_none_or(|key| entry.max_key > *key) {
1740 maximum = Some(entry.max_key.clone());
1741 }
1742 points.extend(child_verified.points);
1743 }
1744 VerifiedSubtree {
1745 count,
1746 minimum,
1747 maximum,
1748 points,
1749 }
1750 };
1751 if verified.count != node.subtree_count {
1752 return Err(Error::InvalidProximityObject {
1753 kind: "node",
1754 reason: "subtree count mismatch".to_owned(),
1755 });
1756 }
1757 Ok(verified)
1758 }
1759}
1760
1761struct VerificationState<'a> {
1762 records: &'a BTreeMap<Vec<u8>, ProximityRecord>,
1763 seen_nodes: HashSet<Cid>,
1764 seen_external_vectors: HashSet<Cid>,
1765 seen_scalar_quantizers: HashSet<Cid>,
1766 seen_leaf_keys: HashSet<Vec<u8>>,
1767 summary: ProximityVerification,
1768}
1769
1770struct VerifiedSubtree {
1771 count: u64,
1772 minimum: Option<Vec<u8>>,
1773 maximum: Option<Vec<u8>>,
1774 points: Vec<(Vec<u8>, Vec<f32>)>,
1775}
1776
1777impl VerifiedSubtree {
1778 fn from_points(count: u64, points: Vec<(Vec<u8>, Vec<f32>)>) -> Self {
1779 let minimum = points.iter().map(|(key, _)| key).min().cloned();
1780 let maximum = points.iter().map(|(key, _)| key).max().cloned();
1781 Self {
1782 count,
1783 minimum,
1784 maximum,
1785 points,
1786 }
1787 }
1788}
1789
1790fn remaining_budget(budget: &SearchBudget, used: &ProximitySearchStats) -> SearchBudget {
1791 SearchBudget {
1792 max_nodes: budget
1793 .max_nodes
1794 .map(|limit| limit.saturating_sub(used.nodes_read)),
1795 max_committed_bytes: budget
1796 .max_committed_bytes
1797 .map(|limit| limit.saturating_sub(used.committed_bytes)),
1798 max_distance_evaluations: budget.max_distance_evaluations.map(|limit| {
1799 limit.saturating_sub(
1800 used.distance_evaluations
1801 .saturating_add(used.quantized_distance_evaluations),
1802 )
1803 }),
1804 max_frontier_entries: budget.max_frontier_entries,
1805 }
1806}
1807
1808fn search_budget_exhausted(budget: &SearchBudget, used: &ProximitySearchStats) -> bool {
1809 budget
1810 .max_nodes
1811 .is_some_and(|limit| used.nodes_read >= limit)
1812 || budget
1813 .max_committed_bytes
1814 .is_some_and(|limit| used.committed_bytes >= limit)
1815 || budget.max_distance_evaluations.is_some_and(|limit| {
1816 used.distance_evaluations
1817 .saturating_add(used.quantized_distance_evaluations)
1818 >= limit
1819 })
1820}
1821
1822fn add_search_stats(total: &mut ProximitySearchStats, added: &ProximitySearchStats) {
1823 total.levels_visited = total.levels_visited.saturating_add(added.levels_visited);
1824 total.nodes_read = total.nodes_read.saturating_add(added.nodes_read);
1825 total.bytes_read = total.bytes_read.saturating_add(added.bytes_read);
1826 total.physical_bytes_read = total
1827 .physical_bytes_read
1828 .saturating_add(added.physical_bytes_read);
1829 total.committed_bytes = total.committed_bytes.saturating_add(added.committed_bytes);
1830 total.distance_evaluations = total
1831 .distance_evaluations
1832 .saturating_add(added.distance_evaluations);
1833 total.quantized_distance_evaluations = total
1834 .quantized_distance_evaluations
1835 .saturating_add(added.quantized_distance_evaluations);
1836 total.reranked_candidates = total
1837 .reranked_candidates
1838 .saturating_add(added.reranked_candidates);
1839 total.frontier_peak = total.frontier_peak.max(added.frontier_peak);
1840 total.candidate_handles_peak = total
1841 .candidate_handles_peak
1842 .max(added.candidate_handles_peak);
1843 total.candidate_retained_bytes_peak = total
1844 .candidate_retained_bytes_peak
1845 .max(added.candidate_retained_bytes_peak);
1846}
1847
1848pub(super) fn encoded_vector_matches(
1849 encoded: super::storage::EncodedVectorRef<'_>,
1850 expected: &[f32],
1851) -> bool {
1852 encoded.dimensions as usize == expected.len()
1853 && encoded
1854 .bytes
1855 .chunks_exact(4)
1856 .zip(expected)
1857 .all(|(bytes, expected)| {
1858 u32::from_le_bytes(bytes.try_into().expect("validated vector component"))
1859 == expected.to_bits()
1860 })
1861}
1862
1863pub(super) fn encoded_vectors_equal(
1864 left: super::storage::EncodedVectorRef<'_>,
1865 right: super::storage::EncodedVectorRef<'_>,
1866) -> bool {
1867 left.dimensions == right.dimensions && left.bytes == right.bytes
1868}
1869
1870fn load_content<S: Store>(store: &S, cid: &Cid) -> Result<Vec<u8>, Error> {
1871 let bytes = store
1872 .get(cid.as_bytes())
1873 .map_err(|error| Error::Store(Box::new(error)))?
1874 .ok_or_else(|| Error::NotFound(cid.clone()))?;
1875 let actual = Cid::from_bytes(&bytes);
1876 if actual != *cid {
1877 return Err(Error::CidMismatch {
1878 expected: cid.clone(),
1879 actual,
1880 });
1881 }
1882 Ok(bytes)
1883}
1884
1885fn publish_maintenance_content<S: Store>(store: &S, cid: &Cid, bytes: &[u8]) -> Result<(), Error> {
1886 let entries = [(cid.as_bytes(), bytes)];
1887 store
1888 .publish_nodes(NodePublication::new(
1889 &entries,
1890 PublicationOrigin::Maintenance,
1891 ))
1892 .map_err(|error| Error::Store(Box::new(error)))
1893}
1894
1895fn distance_budget_exhausted(request: &SearchRequest<'_>, stats: &ProximitySearchStats) -> bool {
1896 request
1897 .budget
1898 .max_distance_evaluations
1899 .is_some_and(|maximum| {
1900 stats
1901 .distance_evaluations
1902 .saturating_add(stats.quantized_distance_evaluations)
1903 >= maximum
1904 })
1905}
1906
1907fn put_missing_nodes<S: Store>(store: &S, nodes: &[(Cid, Vec<u8>)]) -> Result<usize, Error> {
1908 let keys: Vec<_> = nodes.iter().map(|(cid, _)| cid.as_bytes()).collect();
1909 let existing = store
1910 .batch_get_ordered_unique(&keys)
1911 .map_err(|error| Error::Store(Box::new(error)))?;
1912 for ((expected, _), value) in nodes.iter().zip(&existing) {
1913 if let Some(bytes) = value {
1914 let actual = Cid::from_bytes(bytes);
1915 if actual != *expected {
1916 return Err(Error::CidMismatch {
1917 expected: expected.clone(),
1918 actual,
1919 });
1920 }
1921 }
1922 }
1923 let missing: Vec<_> = nodes
1924 .iter()
1925 .zip(existing)
1926 .filter_map(|((cid, bytes), value)| {
1927 value
1928 .is_none()
1929 .then_some((cid.as_bytes(), bytes.as_slice()))
1930 })
1931 .collect();
1932 if !missing.is_empty() {
1933 store
1934 .publish_nodes(NodePublication::new(
1935 &missing,
1936 PublicationOrigin::Maintenance,
1937 ))
1938 .map_err(|error| Error::Store(Box::new(error)))?;
1939 }
1940 Ok(missing.len())
1941}
1942
1943fn apply_directory_stats(target: &mut ProximityMutationStats, source: SpliceStats) {
1944 target.directory_entries_scanned = source.entries_scanned;
1945 target.directory_nodes_read = source.nodes_read;
1946 target.directory_nodes_rebuilt = source.nodes_rebuilt;
1947 target.directory_nodes_written = source.nodes_written;
1948 target.directory_nodes_reused = source.nodes_reused;
1949 target.directory_levels_rebuilt = source.levels_rebuilt;
1950 target.directory_right_edge_rebuilt = source.right_edge_rebuilt;
1951}
1952
1953pub(super) fn validate_mutations(
1954 mutations: impl IntoIterator<Item = ProximityMutation>,
1955) -> Result<Vec<ProximityMutation>, Error> {
1956 let mut mutations: Vec<_> = mutations.into_iter().collect();
1957 mutations.sort_by(|left, right| left.key.cmp(&right.key));
1958 for pair in mutations.windows(2) {
1959 if pair[0].key == pair[1].key {
1960 return Err(Error::DuplicateProximityKey {
1961 key: pair[0].key.clone(),
1962 });
1963 }
1964 }
1965 Ok(mutations)
1966}
1967
1968pub(super) fn apply_mutations(
1969 records: &mut BTreeMap<Vec<u8>, ProximityRecord>,
1970 mutations: &[ProximityMutation],
1971 config: &ProximityConfig,
1972) -> Result<(), Error> {
1973 for mutation in mutations {
1974 match &mutation.value {
1975 Some((vector, value)) => {
1976 records.insert(
1977 mutation.key.clone(),
1978 ProximityRecord {
1979 key: mutation.key.clone(),
1980 vector: prepare_vector(config.metric, vector, config.dimensions)?,
1981 value: value.clone(),
1982 },
1983 );
1984 }
1985 None => {
1986 records.remove(&mutation.key);
1987 }
1988 }
1989 }
1990 Ok(())
1991}
1992
1993#[cfg(test)]
1994mod tests {
1995 use super::*;
1996 use crate::prolly::proximity::distance::{query_kernel_calls, reset_query_kernel_calls};
1997 use crate::prolly::store::MemStore;
1998
1999 fn config() -> ProximityConfig {
2000 let mut config = ProximityConfig::new(1);
2001 config.hierarchy.log_chunk_size = 1;
2002 config.hierarchy.level_hash_seed = 7;
2003 config.overflow.max_page_bytes = 256 * 1024;
2004 config
2005 }
2006
2007 fn two_representative_map() -> (Arc<MemStore>, ProximityMap<Arc<MemStore>>) {
2008 let keys: Vec<_> = (0..10_000)
2009 .map(|index| format!("candidate-{index}").into_bytes())
2010 .filter(|key| promotion_level(key, 1, 7) == 1)
2011 .take(2)
2012 .collect();
2013 assert_eq!(keys.len(), 2);
2014 let store = Arc::new(MemStore::new());
2015 let map = ProximityMap::build(
2016 store.clone(),
2017 config(),
2018 keys.into_iter()
2019 .enumerate()
2020 .map(|(index, key)| ProximityRecord {
2021 key,
2022 vector: vec![index as f32],
2023 value: Vec::new(),
2024 }),
2025 )
2026 .unwrap();
2027 (store, map)
2028 }
2029
2030 #[test]
2031 fn exact_read_lease_retains_and_validates_the_stored_record() {
2032 let store = Arc::new(MemStore::new());
2033 let map = ProximityMap::build(
2034 store,
2035 ProximityConfig::new(2),
2036 [ProximityRecord {
2037 key: b"key".to_vec(),
2038 vector: vec![1.0, 2.0],
2039 value: b"value".to_vec(),
2040 }],
2041 )
2042 .unwrap();
2043 let mut read = map.read().unwrap();
2044 let lease = read.get_lease(b"key").unwrap().unwrap();
2045 let stored = StoredRecordRef::decode(lease.as_bytes().unwrap(), 2).unwrap();
2046 assert_eq!(
2047 ProximityVectorRef::from_encoded(stored.vector).to_vec(),
2048 vec![1.0, 2.0]
2049 );
2050 assert_eq!(stored.value, b"value");
2051 assert!(read.get_lease(b"missing").unwrap().is_none());
2052 }
2053
2054 #[test]
2055 fn construction_and_mutation_never_enter_a_query_kernel() {
2056 reset_query_kernel_calls();
2057 let store = Arc::new(MemStore::new());
2058 let map = ProximityMap::build(
2059 store,
2060 config(),
2061 (0..64).map(|index| ProximityRecord {
2062 key: format!("key-{index:03}").into_bytes(),
2063 vector: vec![index as f32],
2064 value: Vec::new(),
2065 }),
2066 )
2067 .unwrap();
2068 let (map, _) = map
2069 .mutate_batch([ProximityMutation {
2070 key: b"key-017".to_vec(),
2071 value: Some((vec![17.25], b"updated".to_vec())),
2072 }])
2073 .unwrap();
2074 assert_eq!(query_kernel_calls(), 0);
2075
2076 let mut request = SearchRequest::exact(&[17.0], 3);
2077 request.kernel = super::super::QueryKernel::SimdDeterministic;
2078 map.search(request).unwrap();
2079 assert!(query_kernel_calls() > 0);
2080 }
2081
2082 fn publish_root_descriptor(
2083 store: &Arc<MemStore>,
2084 map: &ProximityMap<Arc<MemStore>>,
2085 root: ProximityNode,
2086 ) -> Cid {
2087 let root_bytes = root.encode().unwrap();
2088 let root_cid = Cid::from_bytes(&root_bytes);
2089 store.put(root_cid.as_bytes(), &root_bytes).unwrap();
2090
2091 let descriptor_bytes = store.get(map.tree.descriptor.as_bytes()).unwrap().unwrap();
2092 let mut descriptor = Descriptor::decode(&descriptor_bytes).unwrap();
2093 descriptor.proximity_root = root_cid;
2094 let descriptor_bytes = descriptor.encode();
2095 let descriptor_cid = Cid::from_bytes(&descriptor_bytes);
2096 store
2097 .put(descriptor_cid.as_bytes(), &descriptor_bytes)
2098 .unwrap();
2099 descriptor_cid
2100 }
2101
2102 fn publish_replacement_root(
2103 store: &Arc<MemStore>,
2104 map: &ProximityMap<Arc<MemStore>>,
2105 root: ProximityNode,
2106 ) -> ProximityMap<Arc<MemStore>> {
2107 let descriptor_cid = publish_root_descriptor(store, map, root);
2108 ProximityMap::load(store.clone(), descriptor_cid).unwrap()
2109 }
2110
2111 #[test]
2112 fn verify_rejects_a_leaf_vector_that_disagrees_with_the_exact_directory() {
2113 let store = Arc::new(MemStore::new());
2114 let mut leaf_config = config();
2115 leaf_config.hierarchy.log_chunk_size = 63;
2116 let map = ProximityMap::build(
2117 store.clone(),
2118 leaf_config,
2119 [ProximityRecord {
2120 key: b"key".to_vec(),
2121 vector: vec![1.0],
2122 value: Vec::new(),
2123 }],
2124 )
2125 .unwrap();
2126 let bytes = store
2127 .get(map.tree.proximity_root.as_bytes())
2128 .unwrap()
2129 .unwrap();
2130 let mut root = ProximityNode::decode(&bytes, 1).unwrap();
2131 root.entries[0].vector = super::super::storage::VectorRef::Inline(vec![2.0]);
2132 let corrupt = publish_replacement_root(&store, &map, root);
2133
2134 assert!(matches!(
2135 corrupt.verify(),
2136 Err(Error::InvalidProximityObject { reason, .. })
2137 if reason == "leaf vector disagrees with exact directory"
2138 ));
2139 }
2140
2141 #[test]
2142 fn verify_rejects_repeated_child_ownership() {
2143 let (store, map) = two_representative_map();
2144 let bytes = store
2145 .get(map.tree.proximity_root.as_bytes())
2146 .unwrap()
2147 .unwrap();
2148 let mut root = ProximityNode::decode(&bytes, 1).unwrap();
2149 assert_eq!(root.level, 1);
2150 assert_eq!(root.entries.len(), 2);
2151 root.entries[1].child = root.entries[0].child.clone();
2152 let corrupt = publish_replacement_root(&store, &map, root);
2153
2154 assert!(matches!(
2155 corrupt.verify(),
2156 Err(Error::InvalidProximityObject { reason, .. })
2157 if reason == "cycle or repeated child ownership"
2158 ));
2159 }
2160
2161 #[test]
2162 fn verify_rejects_an_invalid_child_level() {
2163 let (store, map) = two_representative_map();
2164 let bytes = store
2165 .get(map.tree.proximity_root.as_bytes())
2166 .unwrap()
2167 .unwrap();
2168 let mut root = ProximityNode::decode(&bytes, 1).unwrap();
2169 root.entries[0].child = Some(map.tree.proximity_root.clone());
2170 let corrupt = publish_replacement_root(&store, &map, root);
2171
2172 assert!(matches!(
2173 corrupt.verify(),
2174 Err(Error::InvalidProximityObject { reason, .. })
2175 if reason == "unexpected logical level"
2176 ));
2177 }
2178
2179 #[test]
2180 fn verify_rejects_a_representative_below_its_node_level() {
2181 let (store, map) = two_representative_map();
2182 let bytes = store
2183 .get(map.tree.proximity_root.as_bytes())
2184 .unwrap()
2185 .unwrap();
2186 let mut root = ProximityNode::decode(&bytes, 1).unwrap();
2187 let replacement = (0..10_000)
2188 .map(|index| format!("!invalid-{index}").into_bytes())
2189 .find(|key| promotion_level(key, 1, 7) == 0 && key < &root.entries[1].key)
2190 .unwrap();
2191 root.entries[0].key = replacement.clone();
2192 root.entries[0].min_key = replacement;
2193 let corrupt = publish_replacement_root(&store, &map, root);
2194
2195 assert!(matches!(
2196 corrupt.verify(),
2197 Err(Error::InvalidProximityObject { reason, .. })
2198 if reason == "entry appears above its deterministic promotion level"
2199 ));
2200 }
2201
2202 #[test]
2203 fn verify_rejects_a_non_nearest_parent_route() {
2204 let (store, map) = two_representative_map();
2205 let bytes = store
2206 .get(map.tree.proximity_root.as_bytes())
2207 .unwrap()
2208 .unwrap();
2209 let mut root = ProximityNode::decode(&bytes, 1).unwrap();
2210 let first = root.entries[0].child.clone();
2211 root.entries[0].child = root.entries[1].child.clone();
2212 root.entries[1].child = first;
2213 let corrupt = publish_replacement_root(&store, &map, root);
2214
2215 assert!(matches!(
2216 corrupt.verify(),
2217 Err(Error::InvalidProximityObject { reason, .. })
2218 if reason == "nearest-representative invariant violated"
2219 ));
2220 }
2221
2222 #[test]
2223 fn load_rejects_a_root_subtree_count_that_disagrees_with_the_descriptor() {
2224 let (store, map) = two_representative_map();
2225 let bytes = store
2226 .get(map.tree.proximity_root.as_bytes())
2227 .unwrap()
2228 .unwrap();
2229 let mut root = ProximityNode::decode(&bytes, 1).unwrap();
2230 root.subtree_count += 1;
2231 root.entries[0].child_count += 1;
2232 let descriptor = publish_root_descriptor(&store, &map, root);
2233
2234 assert!(matches!(
2235 ProximityMap::load(store, descriptor),
2236 Err(Error::InvalidProximityObject { reason, .. })
2237 if reason == "record count disagrees with proximity root"
2238 ));
2239 }
2240
2241 #[test]
2242 fn verify_rejects_a_non_conservative_covering_radius() {
2243 let store = Arc::new(MemStore::new());
2244 let map = ProximityMap::build(
2245 store.clone(),
2246 config(),
2247 (0..128).map(|index| ProximityRecord {
2248 key: format!("radius-{index:04}").into_bytes(),
2249 vector: vec![index as f32],
2250 value: Vec::new(),
2251 }),
2252 )
2253 .unwrap();
2254 let bytes = store
2255 .get(map.tree.proximity_root.as_bytes())
2256 .unwrap()
2257 .unwrap();
2258 let mut root = ProximityNode::decode(&bytes, 1).unwrap();
2259 let entry = root
2260 .entries
2261 .iter_mut()
2262 .find(|entry| entry.covering_radius > 0.0)
2263 .expect("test hierarchy has a nontrivial cluster");
2264 entry.covering_radius = 0.0;
2265 let corrupt = publish_replacement_root(&store, &map, root);
2266
2267 assert!(matches!(
2268 corrupt.verify(),
2269 Err(Error::InvalidProximityObject { reason, .. })
2270 if reason == "covering-radius summary is not conservative"
2271 ));
2272 }
2273
2274 #[test]
2275 fn verify_rejects_a_scalar_quantizer_that_disagrees_with_its_node() {
2276 let store = Arc::new(MemStore::new());
2277 let mut quantized_config = config();
2278 quantized_config.scalar_quantization =
2279 Some(super::super::ScalarQuantizationConfig { group_size: 1 });
2280 let map = ProximityMap::build(
2281 store.clone(),
2282 quantized_config,
2283 (0..64).map(|index| ProximityRecord {
2284 key: format!("quantized-{index:03}").into_bytes(),
2285 vector: vec![index as f32],
2286 value: Vec::new(),
2287 }),
2288 )
2289 .unwrap();
2290 let bytes = store
2291 .get(map.tree.proximity_root.as_bytes())
2292 .unwrap()
2293 .unwrap();
2294 let mut root = ProximityNode::decode(&bytes, 1).unwrap();
2295 let fake_vectors = vec![vec![999.0]; root.entries.len()];
2296 let fake_refs: Vec<_> = fake_vectors.iter().map(Vec::as_slice).collect();
2297 let fake = ScalarQuantized::build(&fake_refs, 1, 1).unwrap();
2298 let fake_bytes = fake.encode().unwrap();
2299 let fake_cid = Cid::from_bytes(&fake_bytes);
2300 store.put(fake_cid.as_bytes(), &fake_bytes).unwrap();
2301 root.quantizer = Some(fake_cid);
2302 let corrupt = publish_replacement_root(&store, &map, root);
2303
2304 assert!(matches!(
2305 corrupt.verify(),
2306 Err(Error::InvalidProximityObject { kind: "quantizer", reason })
2307 if reason.contains("disagree")
2308 ));
2309 }
2310}