1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
//! PR-9 item 1 — engine-side embed serialization.
//!
//! The projection pool runs `PROJECTION_WORKERS` (2) workers. PR-9 serializes
//! the embed *call* engine-side so the shared `Arc<dyn Embedder>` is invoked
//! by at most one worker at a time, while commit/IO stays parallel.
//!
//! Rationale is SAFETY, not throughput: the engine accepts arbitrary
//! caller-supplied embedders (pyo3 / napi bridges) that are `Sync` only by
//! contract and may not be truly concurrency-safe; serializing engine-side
//! makes the projection robust to them. (Throughput is ~neutral — candle
//! fans every forward onto one process-wide rayon pool, so concurrent
//! forwards share it rather than getting 2x; the PR-9 pre-flight also
//! confirmed concurrent CandleBge embeds neither wedge nor corrupt.)
//!
//! This test proves serialization mechanically with a fast mock embedder
//! that records the maximum number of `embed()` calls in flight at once.
//!
//! RED (no guard): two workers embed concurrently → max in flight == 2.
//! GREEN (guard): embeds run one at a time → max in flight == 1.
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::Duration;
use fathomdb_embedder_api::{Embedder, EmbedderError, EmbedderIdentity, Vector};
use fathomdb_engine::lifecycle::ProjectionStatus;
use fathomdb_engine::{Engine, PreparedWrite};
use fathomdb_schema::SQLITE_SUFFIX;
use tempfile::TempDir;
const DIM: u32 = 8;
fn unit_vector() -> Vector {
let mut v = vec![0.0_f32; DIM as usize];
v[0] = 1.0;
v
}
/// Records peak concurrency of `embed()`. Each call bumps an in-flight
/// counter, holds it for `delay` (so overlapping calls are observable),
/// and updates the shared `max_in_flight` high-water mark.
#[derive(Debug)]
struct ConcurrencyProbeEmbedder {
identity: EmbedderIdentity,
in_flight: AtomicUsize,
max_in_flight: Arc<AtomicUsize>,
delay: Duration,
}
impl ConcurrencyProbeEmbedder {
fn new(max_in_flight: Arc<AtomicUsize>, delay: Duration) -> Self {
Self {
identity: EmbedderIdentity::new("probe", "rev-a", DIM),
in_flight: AtomicUsize::new(0),
max_in_flight,
delay,
}
}
}
impl Embedder for ConcurrencyProbeEmbedder {
fn identity(&self) -> EmbedderIdentity {
self.identity.clone()
}
fn embed(&self, _text: &str) -> Result<Vector, EmbedderError> {
let cur = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1;
self.max_in_flight.fetch_max(cur, Ordering::SeqCst);
thread::sleep(self.delay);
self.in_flight.fetch_sub(1, Ordering::SeqCst);
Ok(unit_vector())
}
}
fn fixture_path(name: &str) -> (TempDir, std::path::PathBuf) {
let dir = TempDir::new().expect("tempdir");
let path = dir.path().join(format!("{name}{SQLITE_SUFFIX}"));
(dir, path)
}
/// Item 1: the production projection path must embed one-at-a-time. With two
/// projection workers and a deliberately-slow embed, an unserialized engine
/// would show two concurrent `embed()` calls; the engine-side guard pins the
/// peak at exactly one.
#[test]
fn embeds_are_serialized_engine_side() {
let (_dir, path) = fixture_path("pr9_serialize");
let max_in_flight = Arc::new(AtomicUsize::new(0));
let embedder =
Arc::new(ConcurrencyProbeEmbedder::new(max_in_flight.clone(), Duration::from_millis(50)));
let opened = Engine::open_with_embedder_for_test(&path, embedder).expect("open");
let engine = opened.engine;
engine.configure_vector_kind_for_test("doc").expect("vector kind");
// A worker grabs a whole PROJECTION_COMMIT_BATCH (16) at once, and the
// dispatcher enqueues up to PROJECTION_INFLIGHT_LIMIT (32) per scan — so
// we write > one batch (40 docs) to ensure both workers pick up work and
// would embed concurrently if the engine permitted it.
let nodes: Vec<PreparedWrite> = (0..40)
.map(|i| PreparedWrite::Node {
kind: "doc".to_string(),
body: format!("serialize-doc-{i}"),
source_id: fathomdb_engine::SourceId::new("test:fixture").expect("test source id"),
logical_id: None,
state: fathomdb_engine::InitialState::Active,
reason: None,
valid_from: None,
valid_until: None,
})
.collect();
engine.write(&nodes).expect("write");
engine.drain(30_000).expect("drain");
assert_eq!(
engine.projection_status_for_test("doc").expect("status"),
ProjectionStatus::UpToDate,
"all docs must project successfully"
);
let peak = max_in_flight.load(Ordering::SeqCst);
assert_eq!(
peak, 1,
"embeds must run one at a time engine-side (PR-9 oversubscription fix); \
observed {peak} concurrent embed() calls"
);
}