diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 718c5c7d..aaefb012 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -117,3 +117,64 @@ jobs: - name: cargo test run: cargo test --workspace --all-features + + qdrant-integration: + # §7.0 Phase 2: the durable/hybrid Qdrant memory backend needs a live Qdrant, + # which the gauntlet lacks — so those scenarios are `#[ignore]`d there (they + # report as `ignored`, never a masked `passed` — see #358). This job boots a + # real Qdrant and runs them with `--ignored`, so the recall re-entry (#348) + # and async-drop (#349) fixes are exercised and cannot silently regress. + name: qdrant integration (--ignored) + runs-on: ubuntu-latest + services: + qdrant: + # Pin a server whose minor version is within one of the qdrant-client in + # Cargo.lock (the client refuses a wider skew). + image: qdrant/qdrant:v1.18.0 + ports: + - 6333:6333 + - 6334:6334 + steps: + - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 + + - name: Free disk space (ubuntu) + run: | + sudo rm -rf /usr/share/dotnet /usr/local/lib/android /opt/ghc \ + /usr/local/share/powershell /usr/share/swift /usr/local/.ghcup \ + /usr/lib/jvm /usr/lib/mono 2>/dev/null || true + df -h / + + - name: Install pinned toolchain + run: rustup toolchain install 1.96.1 --profile minimal + + - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2 + + - name: Wait for Qdrant readiness + run: | + for i in $(seq 1 60); do + if curl -sf http://localhost:6333/readyz >/dev/null; then + echo "qdrant ready after ${i}s"; exit 0 + fi + sleep 1 + done + echo "qdrant did not become ready" >&2; exit 1 + + - name: Gated Qdrant integration tests (--ignored) + env: + QDRANT_INTEGRATION_TEST: "1" + QDRANT_URL: http://localhost:6334 + run: | + # Crate-level: durable backend (incl. the async-drop restart path) and + # hybrid dense+sparse recall over Mock embeddings. The `--skip`ped tests + # need extras out of this job's scope: `snapshot_*` a longer snapshot + # deadline (tracked separately), the two semantic tests the BGE model. + cargo test -p ardur-memory-qdrant --test integration \ + -- --ignored --skip snapshot_into_receipt_records_event + cargo test -p ardur-memory-qdrant --test hybrid_integration \ + -- --ignored --skip semantic_hit_gated --skip hybrid_beats_either + # End-to-end through the fused runtime: recall re-entry (#348) and the + # async-context drop (#349) on the real durable + hybrid backends. + cargo test -p ardur-e2e-tests \ + --test scenario_qdrant_memory_persistence \ + --test scenario_hybrid_memory_full_pipeline -- --ignored + diff --git a/crates/e2e-tests/tests/scenario_hybrid_memory_full_pipeline.rs b/crates/e2e-tests/tests/scenario_hybrid_memory_full_pipeline.rs index 38dd2f0d..956726d0 100644 --- a/crates/e2e-tests/tests/scenario_hybrid_memory_full_pipeline.rs +++ b/crates/e2e-tests/tests/scenario_hybrid_memory_full_pipeline.rs @@ -4,12 +4,15 @@ //! live Qdrant: chat turn → receipt-chained memory store → hybrid dense+sparse //! recall → memory context display in the next provider request. //! -//! Gated on `QDRANT_INTEGRATION_TEST=1` (CI has no Qdrant). To run locally: +//! `#[ignore]`d because it needs a live Qdrant (CI has none by default); the +//! default suite reports it as `ignored`, never a silent `passed` (#358). Run it +//! explicitly against a Qdrant — the dedicated CI job does exactly this: //! //! ```text //! docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant -//! QDRANT_INTEGRATION_TEST=1 \ -//! cargo test -p ardur-e2e-tests --test scenario_hybrid_memory_full_pipeline +//! QDRANT_INTEGRATION_TEST=1 QDRANT_URL=http://localhost:6334 \ +//! cargo test -p ardur-e2e-tests --test scenario_hybrid_memory_full_pipeline \ +//! -- --ignored //! ``` use std::collections::VecDeque; @@ -32,12 +35,8 @@ use async_trait::async_trait; const COLLECTION: &str = "ardur_e2e_hybrid_full_pipeline"; -fn gate() -> Option { - if std::env::var("QDRANT_INTEGRATION_TEST").as_deref() != Ok("1") { - eprintln!("skipping scenario_hybrid_memory_full_pipeline: set QDRANT_INTEGRATION_TEST=1"); - return None; - } - Some(QdrantMemoryConfig::from_env().with_collection_name(COLLECTION)) +fn config() -> QdrantMemoryConfig { + QdrantMemoryConfig::from_env().with_collection_name(COLLECTION) } struct CapturingProvider { @@ -107,12 +106,9 @@ fn submit_request(prompt: &str, session_id: SessionId) -> SubmitRequest { } #[test] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] fn chat_store_recall_and_display_through_real_hybrid_memory() { - let Some(cfg) = gate() else { - return; - }; - - let qdrant = QdrantMemoryRuntime::connect(cfg).expect("connect qdrant"); + let qdrant = QdrantMemoryRuntime::connect(config()).expect("connect qdrant"); let bm25 = Bm25Index::new(None).expect("in-memory bm25"); let hybrid = Arc::new(HybridMemoryRetriever::new( qdrant, diff --git a/crates/e2e-tests/tests/scenario_qdrant_memory_persistence.rs b/crates/e2e-tests/tests/scenario_qdrant_memory_persistence.rs index c99e8f50..592c0300 100644 --- a/crates/e2e-tests/tests/scenario_qdrant_memory_persistence.rs +++ b/crates/e2e-tests/tests/scenario_qdrant_memory_persistence.rs @@ -9,12 +9,15 @@ //! Phase 1 in-process store would lose the fact here; the Qdrant store recovers //! it. //! -//! Gated on `QDRANT_INTEGRATION_TEST=1` (CI has no Qdrant). To run locally: +//! `#[ignore]`d because it needs a live Qdrant (CI has none by default); the +//! default suite reports it as `ignored`, never a silent `passed` (#358). Run it +//! explicitly against a Qdrant — the dedicated CI job does exactly this: //! //! ```text //! docker run -p 6334:6334 qdrant/qdrant -//! QDRANT_INTEGRATION_TEST=1 \ -//! cargo test -p ardur-e2e-tests --test scenario_qdrant_memory_persistence +//! QDRANT_INTEGRATION_TEST=1 QDRANT_URL=http://localhost:6334 \ +//! cargo test -p ardur-e2e-tests --test scenario_qdrant_memory_persistence \ +//! -- --ignored //! ``` use std::sync::Arc; @@ -28,23 +31,21 @@ use ardur_runtime::{CapTokenRef, ChatMessage, ChatRuntime, SessionId, SubmitRequ const PROMPT: &str = "remember this across a restart"; const COLLECTION: &str = "ardur_e2e_qdrant_persistence"; -/// The Qdrant config for this scenario, or `None` when the gate var is unset. -fn gate() -> Option { - if std::env::var("QDRANT_INTEGRATION_TEST").as_deref() != Ok("1") { - eprintln!("skipping scenario_qdrant_memory_persistence: set QDRANT_INTEGRATION_TEST=1"); - return None; - } - Some(QdrantMemoryConfig::from_env().with_collection_name(COLLECTION)) +/// The Qdrant config for this scenario. The endpoint comes from `QDRANT_URL` +/// (default `http://localhost:6334`); `#[ignore]` — not an env early-return — is +/// what keeps this off the default suite, so a skip can never masquerade as a +/// pass (#358). +fn config() -> QdrantMemoryConfig { + QdrantMemoryConfig::from_env().with_collection_name(COLLECTION) } /// The multi-thread flavor matters: the fused turn calls the synchronous /// `MemoryRuntime::record` from inside this runtime, and the Qdrant backend /// bridges it with `block_in_place`, which requires a multi-threaded runtime. #[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] async fn fused_turn_memory_survives_restart() { - let Some(cfg) = gate() else { - return; - }; + let cfg = config(); let subject = HolderId::from(TEST_HOLDER); diff --git a/crates/memory-qdrant/src/hybrid.rs b/crates/memory-qdrant/src/hybrid.rs index 55a159d4..5a3c02e7 100644 --- a/crates/memory-qdrant/src/hybrid.rs +++ b/crates/memory-qdrant/src/hybrid.rs @@ -103,7 +103,7 @@ impl HybridMemoryRetriever { pub async fn record(&self, rec: MemoryRecord) -> Result { let doc_id = rec.record_id.to_string(); let text = searchable_text(&rec); - let record_id = self.qdrant.record(rec)?; + let record_id = self.qdrant.record_async(rec).await?; self.bm25 .lock() .await @@ -156,12 +156,18 @@ impl HybridMemoryRetriever { // ARD-477: exclude any chain that has been tombstoned so a forgotten // memory is never re-injected. One scroll of the relevant records. - let dead = self.qdrant.dead_chains(subject)?; + // Awaited directly (not the sync bridge) so this recall runs on a single + // `block_on` pass — a nested bridge here re-enters the owned runtime and + // panics the turn (#348). + let dead = self.qdrant.dead_chains(subject).await?; // ---- dense: embed the query, ANN-search, drop tombstones, and keep the // hydrated records (vector hits carry their full record_json). let query_vec = self.embed_query(query).await?; - let vector_hits = self.qdrant.search_vectors(query_vec, candidate_k as u64)?; + let vector_hits = self + .qdrant + .search_vectors_async(query_vec, candidate_k as u64) + .await?; let mut hydrated: HashMap = HashMap::new(); let mut vector_list: Vec = Vec::with_capacity(vector_hits.len()); for (rec, score) in vector_hits { @@ -201,7 +207,7 @@ impl HybridMemoryRetriever { } let rec = match hydrated.remove(&doc.doc_id) { Some(rec) => rec, - None => match self.fetch_live(&doc.doc_id, subject, &dead)? { + None => match self.fetch_live(&doc.doc_id, subject, &dead).await? { Some(rec) => rec, None => continue, }, @@ -225,7 +231,7 @@ impl HybridMemoryRetriever { /// Hydrate a fused `doc_id` from the durable store, returning it only if it is /// a live (non-tombstone) record. An unparseable id or a missing point yields /// `None`. - fn fetch_live( + async fn fetch_live( &self, doc_id: &str, subject: Option<&HolderId>, @@ -236,7 +242,8 @@ impl HybridMemoryRetriever { }; Ok(self .qdrant - .fetch_record(RecordId(uuid))? + .fetch_record_async(RecordId(uuid)) + .await? .filter(|rec| rec.invalidation_time.is_none()) .filter(|rec| !dead.contains(&rec.correction_chain_root)) .filter(|rec| subject.is_none_or(|s| &rec.subject == s))) @@ -252,11 +259,14 @@ impl HybridMemoryRetriever { /// straight to the durable [`QdrantMemoryRuntime`]. The write/recall methods — /// [`record`](MemoryRuntime::record) and [`search`](MemoryRuntime::search) — are /// asynchronous on the inherent API (dual-write to Qdrant **and** the BM25 index; -/// fused recall over both), so the synchronous trait methods bridge onto the -/// runtime's own Tokio executor via its `block_on` (the same `block_in_place` -/// path the durable runtime uses for its sync trait methods). `self.record(..)` -/// and `self.search(..)` below resolve to the *inherent* async methods (inherent -/// methods shadow trait methods of the same name), so there is no recursion. +/// fused recall over both), so each synchronous trait method bridges onto the +/// runtime's own Tokio executor with a **single** outer `block_on` over a fully +/// async body. That body `await`s the durable store's `*_async` cores directly +/// (`record_async`, `search_vectors_async`, `dead_chains`, `fetch_record_async`) +/// rather than the sync methods, so it never re-enters the owned runtime — the +/// nested-bridge recall panic (#348). `self.record(..)` and `self.search(..)` +/// below resolve to the *inherent* async methods (inherent methods shadow trait +/// methods of the same name), so there is no recursion. impl MemoryRuntime for HybridMemoryRetriever { fn record(&self, rec: MemoryRecord) -> Result { self.qdrant.block_on(self.record(rec)) diff --git a/crates/memory-qdrant/src/runtime.rs b/crates/memory-qdrant/src/runtime.rs index f1bc1889..d652dab7 100644 --- a/crates/memory-qdrant/src/runtime.rs +++ b/crates/memory-qdrant/src/runtime.rs @@ -21,6 +21,29 @@ //! turn, or the server boot under `#[tokio::main]`), it uses //! [`tokio::task::block_in_place`] so it does not deadlock the caller's runtime; //! otherwise it blocks on its own runtime directly. +//! +//! Two invariants keep that bridge sound: +//! +//! * **Never nest a bridge inside itself.** Each inherent async method (e.g. +//! [`record_async`](QdrantMemoryRuntime::record_async), +//! [`search_vectors_async`](QdrantMemoryRuntime::search_vectors_async)) holds +//! the real logic and `await`s the client directly; every *synchronous* method +//! is a thin `block_on(..async..)` wrapper over one of them. A sync wrapper +//! therefore never calls another sync wrapper *inside* an awaited future, so a +//! single logical call bridges exactly once. This is what stops the +//! recall-time re-entry panic (#348): the [`HybridMemoryRetriever`] serves its +//! sync trait methods with a single outer `block_on` over a fully-async body. +//! As a backstop, [`block_on_runtime`] also detects re-entry into its *own* +//! runtime by handle identity (not merely by flavor) and falls through to a +//! dedicated thread rather than panicking. +//! * **Never drop the owned runtime inline.** Dropping a +//! [`tokio::runtime::Runtime`] blocks to join its workers, which panics when +//! the `Arc` is released from inside an async context +//! (server shutdown, end of a turn). The [`Drop`] impl hands teardown to +//! [`shutdown_background`](tokio::runtime::Runtime::shutdown_background), which +//! returns immediately (#349). +//! +//! [`HybridMemoryRetriever`]: crate::HybridMemoryRetriever use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -55,7 +78,12 @@ const SCROLL_LIMIT: u32 = 16_384; pub struct QdrantMemoryRuntime { client: Qdrant, config: QdrantMemoryConfig, - rt: tokio::runtime::Runtime, + /// The owned bridge runtime. `Option` so [`Drop`] can `take` it and hand it + /// to [`shutdown_background`](tokio::runtime::Runtime::shutdown_background) + /// (dropping it inline would panic from an async context — #349). It is + /// `Some` for the entire lifetime except during drop; [`rt`](Self::rt) + /// unwraps it. + rt: Option, /// The model that turns a record's [`searchable_text`] into its stored /// vector. `None` keeps the legacy placeholder embedding (reads still work, /// since they scroll by payload filter; only vector *search* is meaningless). @@ -89,7 +117,7 @@ impl QdrantMemoryRuntime { Ok(Self { client, config, - rt, + rt: Some(rt), embedder: None, }) } @@ -239,34 +267,49 @@ impl QdrantMemoryRuntime { // ---- internals ------------------------------------------------------- + /// Record `rec` durably: embed its [`searchable_text`], build the point, and + /// upsert it — all on one async pass so a caller already inside the bridge + /// runtime never re-enters it. The sync [`MemoryRuntime::record`] is a + /// `block_on` wrapper over this; the [`HybridMemoryRetriever`](crate::HybridMemoryRetriever) + /// awaits it directly from its own async `record`. + /// + /// # Errors + /// [`MemoryError::Backend`] if the embed, payload projection, or upsert fails. + pub async fn record_async(&self, rec: MemoryRecord) -> Result { + let id = rec.record_id; + let point = self.point_for(&rec).await?; + self.upsert(point).await?; + Ok(RecordId(id)) + } + /// Build the Qdrant point for a record: id = the record UUID, vector = the /// real embedding of its [`searchable_text`] (or the placeholder when no /// embedder is attached), payload = the projected [`QdrantPayload`]. - fn point_for(&self, rec: &MemoryRecord) -> Result { + async fn point_for(&self, rec: &MemoryRecord) -> Result { let payload = QdrantPayload::from_record(rec)?; let value = serde_json::to_value(&payload) .map_err(|e| MemoryError::Backend(format!("serialize payload: {e}")))?; let payload: Payload = Payload::try_from(value) .map_err(|e| MemoryError::Backend(format!("payload to qdrant: {e}")))?; - let vector = self.embed_record(rec)?; + let vector = self.embed_record(rec).await?; Ok(PointStruct::new(rec.record_id.to_string(), vector, payload)) } /// The stored vector for a record: the embedding of its /// [`searchable_text`](crate::searchable_text) when an [`Embedder`] is /// attached, else the legacy placeholder (a unit vector). - fn embed_record(&self, rec: &MemoryRecord) -> Result> { + async fn embed_record(&self, rec: &MemoryRecord) -> Result> { match &self.embedder { - Some(embedder) => self.embed_text(embedder, searchable_text(rec)), + Some(embedder) => self.embed_text(embedder, searchable_text(rec)).await, None => Ok(placeholder_embedding(self.config.vector_dim)), } } - /// Embed a single text through `embedder`, bridging its async surface onto the - /// runtime's blocking client. - fn embed_text(&self, embedder: &Arc, text: String) -> Result> { - let mut out = self - .block_on(embedder.embed(vec![text])) + /// Embed a single text through `embedder` (awaited directly — no bridge). + async fn embed_text(&self, embedder: &Arc, text: String) -> Result> { + let mut out = embedder + .embed(vec![text]) + .await .map_err(|e| MemoryError::Backend(format!("embed: {e}")))?; out.pop() .ok_or_else(|| MemoryError::Backend("embedder returned no vector".to_string())) @@ -286,19 +329,33 @@ impl QdrantMemoryRuntime { &self, query_vector: Vec, top_k: u64, + ) -> Result> { + self.block_on(self.search_vectors_async(query_vector, top_k)) + } + + /// The async core of [`search_vectors`](Self::search_vectors): the dense + /// half of hybrid recall, awaited directly by the + /// [`HybridMemoryRetriever`](crate::HybridMemoryRetriever) so its recall runs + /// on a single bridge pass (no nested `block_on` — #348). + /// + /// # Errors + /// [`MemoryError::Backend`] on a Qdrant transport or search error. + pub async fn search_vectors_async( + &self, + query_vector: Vec, + top_k: u64, ) -> Result> { if top_k == 0 { return Ok(Vec::new()); } - let resp = self.block_on(async { - self.client - .search_points( - SearchPointsBuilder::new(&self.config.collection_name, query_vector, top_k) - .with_payload(true), - ) - .await - .map_err(|e| MemoryError::Backend(format!("search_points: {e}"))) - })?; + let resp = self + .client + .search_points( + SearchPointsBuilder::new(&self.config.collection_name, query_vector, top_k) + .with_payload(true), + ) + .await + .map_err(|e| MemoryError::Backend(format!("search_points: {e}")))?; Ok(resp .result .into_iter() @@ -313,11 +370,24 @@ impl QdrantMemoryRuntime { /// [`MemoryError::Backend`] if no embedder is attached, the embed fails, or the /// search fails. pub fn search_text(&self, query: &str, top_k: u64) -> Result> { + self.block_on(self.search_text_async(query, top_k)) + } + + /// The async core of [`search_text`](Self::search_text). + /// + /// # Errors + /// [`MemoryError::Backend`] if no embedder is attached, the embed fails, or the + /// search fails. + pub async fn search_text_async( + &self, + query: &str, + top_k: u64, + ) -> Result> { let embedder = self.embedder.as_ref().ok_or_else(|| { MemoryError::Backend("search_text requires an attached embedder".to_string()) })?; - let vector = self.embed_text(embedder, query.to_string())?; - self.search_vectors(vector, top_k) + let vector = self.embed_text(embedder, query.to_string()).await?; + self.search_vectors_async(vector, top_k).await } /// Fetch a single record by its [`RecordId`] (the point id), if present — the @@ -326,7 +396,17 @@ impl QdrantMemoryRuntime { /// # Errors /// [`MemoryError::Backend`] on a Qdrant transport error. pub fn fetch_record(&self, id: RecordId) -> Result> { - self.get_record(id.0) + self.block_on(self.fetch_record_async(id)) + } + + /// The async core of [`fetch_record`](Self::fetch_record) — the hydration + /// hook the [`HybridMemoryRetriever`](crate::HybridMemoryRetriever) awaits + /// directly when resolving a fused id (no nested `block_on` — #348). + /// + /// # Errors + /// [`MemoryError::Backend`] on a Qdrant transport error. + pub async fn fetch_record_async(&self, id: RecordId) -> Result> { + self.get_record_async(id.0).await } /// Whether a real embedder is attached (vs. the placeholder vector). @@ -335,35 +415,37 @@ impl QdrantMemoryRuntime { self.embedder.is_some() } - /// Upsert one point, blocking on the bridge runtime. - fn upsert(&self, point: PointStruct) -> Result<()> { - self.block_on(async { - self.client - .upsert_points( - UpsertPointsBuilder::new(&self.config.collection_name, vec![point]).wait(true), - ) - .await - .map(|_| ()) - .map_err(|e| MemoryError::Backend(format!("upsert_points: {e}"))) - }) + /// Upsert one point (awaited directly on the bridge runtime). + async fn upsert(&self, point: PointStruct) -> Result<()> { + self.client + .upsert_points( + UpsertPointsBuilder::new(&self.config.collection_name, vec![point]).wait(true), + ) + .await + .map(|_| ()) + .map_err(|e| MemoryError::Backend(format!("upsert_points: {e}"))) } /// Scroll every point matching `filter` and reconstruct the records from the /// carried `record_json`. A point whose payload cannot be reconstructed is /// skipped (logged) rather than failing the whole read. fn scroll_records(&self, filter: Filter) -> Result> { - let points = self.block_on(async { - self.client - .scroll( - ScrollPointsBuilder::new(&self.config.collection_name) - .filter(filter) - .limit(SCROLL_LIMIT) - .with_payload(true) - .with_vectors(false), - ) - .await - .map_err(|e| MemoryError::Backend(format!("scroll: {e}"))) - })?; + self.block_on(self.scroll_records_async(filter)) + } + + /// The async core of [`scroll_records`](Self::scroll_records). + async fn scroll_records_async(&self, filter: Filter) -> Result> { + let points = self + .client + .scroll( + ScrollPointsBuilder::new(&self.config.collection_name) + .filter(filter) + .limit(SCROLL_LIMIT) + .with_payload(true) + .with_vectors(false), + ) + .await + .map_err(|e| MemoryError::Backend(format!("scroll: {e}")))?; Ok(points .result @@ -382,30 +464,31 @@ impl QdrantMemoryRuntime { /// /// # Errors /// [`MemoryError::Backend`] if the scroll fails. - pub(crate) fn dead_chains(&self, subject: Option<&HolderId>) -> Result> { + pub(crate) async fn dead_chains(&self, subject: Option<&HolderId>) -> Result> { let filter = match subject { Some(s) => Filter::must([Condition::matches("subject", s.0.clone())]), None => Filter::default(), }; - let records = self.scroll_records(filter)?; + let records = self.scroll_records_async(filter).await?; Ok(chain_cutoff_map(&records).into_keys().collect()) } /// Fetch a single record by its UUID (the point id), if present. fn get_record(&self, id: Uuid) -> Result> { - let points = self.block_on(async { - self.client - .get_points( - GetPointsBuilder::new( - &self.config.collection_name, - vec![id.to_string().into()], - ) + self.block_on(self.get_record_async(id)) + } + + /// The async core of [`get_record`](Self::get_record). + async fn get_record_async(&self, id: Uuid) -> Result> { + let points = self + .client + .get_points( + GetPointsBuilder::new(&self.config.collection_name, vec![id.to_string().into()]) .with_payload(true) .with_vectors(false), - ) - .await - .map_err(|e| MemoryError::Backend(format!("get_points: {e}"))) - })?; + ) + .await + .map_err(|e| MemoryError::Backend(format!("get_points: {e}")))?; Ok(points .result .into_iter() @@ -422,21 +505,63 @@ impl QdrantMemoryRuntime { F: std::future::Future + Send, F::Output: Send, { - block_on_runtime(&self.rt, fut) + block_on_runtime(self.rt(), fut) + } + + /// The owned bridge runtime. Present for the whole lifetime; only [`Drop`] + /// (which never calls [`block_on`](Self::block_on)) leaves it `None`. + fn rt(&self) -> &tokio::runtime::Runtime { + self.rt + .as_ref() + .expect("qdrant memory bridge runtime is present until drop") + } +} + +impl Drop for QdrantMemoryRuntime { + /// Tear the owned bridge runtime down *without blocking*. Dropping a + /// [`tokio::runtime::Runtime`] inline joins its worker threads, which panics + /// when the drop happens inside an async context — "Cannot drop a runtime in + /// a context where blocking is not allowed" (#349), exactly how the fused + /// runtime/server release the memory `Arc` at end-of-turn or on shutdown. + /// [`shutdown_background`](tokio::runtime::Runtime::shutdown_background) + /// signals the workers and returns immediately, so the drop is safe from any + /// context. + fn drop(&mut self) { + if let Some(rt) = self.rt.take() { + rt.shutdown_background(); + } } } /// Drive `fut` to completion on the owned `rt`, cooperating with whatever /// ambient Tokio runtime the caller is on. /// -/// The subtlety is `block_in_place`: it is only legal under a **multi-thread** -/// runtime and **panics** under a current-thread one (M0c). The previous code -/// always used it when an ambient runtime was present, so a caller on -/// `#[tokio::main(flavor = "current_thread")]` turned every sync memory op into -/// a panic. Dispatch on the flavor: block-in-place under multi-thread, and under -/// a current-thread ambient runtime fall back to a dedicated scoped thread -/// (calling `self.rt.block_on` on the ambient thread would itself panic — -/// "cannot start a runtime from within a runtime"). +/// Two subtleties drive the dispatch: +/// +/// * `block_in_place` is only legal under a **multi-thread** runtime and +/// **panics** under a current-thread one (M0c). An older revision always used +/// it when an ambient runtime was present, so a caller on +/// `#[tokio::main(flavor = "current_thread")]` turned every sync memory op into +/// a panic. +/// * Re-entering `rt` — calling `rt.block_on` from a thread already *inside* +/// `rt` — panics with "Cannot start a runtime from within a runtime". Checking +/// the ambient runtime's **flavor** alone does not catch this: when the ambient +/// runtime *is* our own multi-thread `rt` (a sync memory method reached from +/// inside an awaited memory future), the flavor is `MultiThread`, so the +/// flavor-only guard took the `block_in_place(|| rt.block_on(..))` arm and +/// re-entered `rt` → the recall-time panic (#348). The primary fix keeps the +/// inherent async methods from ever nesting a bridge, but this guard also +/// compares runtime **identity** ([`Handle::id`]) so any residual re-entry +/// falls through to a dedicated thread instead of panicking. +/// +/// Dispatch, in order: +/// 1. ambient runtime **is** `rt` → dedicated thread (a fresh std thread owns no +/// ambient runtime, so `rt.block_on` there is legal even while `rt` runs +/// elsewhere); +/// 2. a *different* multi-thread ambient runtime → `block_in_place` (cheap, the +/// production fused-turn path); +/// 3. any other (current-thread) ambient runtime → dedicated thread; +/// 4. no ambient runtime → drive `rt` directly. fn block_on_runtime(rt: &tokio::runtime::Runtime, fut: F) -> F::Output where F: std::future::Future + Send, @@ -445,24 +570,39 @@ where use tokio::runtime::RuntimeFlavor; match tokio::runtime::Handle::try_current() { + // The ambient runtime is our own `rt`: re-entering it would panic, so + // hand off to a dedicated thread (identity, not just flavor — #348). + Ok(handle) if handle.id() == rt.handle().id() => run_on_scoped_thread(rt, fut), + // A different multi-thread ambient runtime (the production fused turn): + // block-in-place cooperates without spawning a thread. Ok(handle) if handle.runtime_flavor() == RuntimeFlavor::MultiThread => { tokio::task::block_in_place(|| rt.block_on(fut)) } - // Current-thread (or any non-multi-thread) ambient runtime: run on a - // dedicated thread that owns no ambient runtime, so `rt.block_on` is - // legal there. `scope` lets the thread borrow `rt` and `fut` without a - // `'static` bound. - Ok(_) => std::thread::scope(|scope| { - scope - .spawn(|| rt.block_on(fut)) - .join() - .expect("memory block_on worker thread panicked") - }), + // A different current-thread ambient runtime: `block_in_place` is illegal + // there, so use the dedicated thread. + Ok(_) => run_on_scoped_thread(rt, fut), // No ambient runtime: drive directly. Err(_) => rt.block_on(fut), } } +/// Drive `fut` on `rt` from a fresh std thread that owns no ambient runtime, so +/// `rt.block_on` is legal there even when `rt` is concurrently running elsewhere +/// (the re-entry and current-thread-ambient paths). `scope` lets the thread +/// borrow `rt` and `fut` without a `'static` bound. +fn run_on_scoped_thread(rt: &tokio::runtime::Runtime, fut: F) -> F::Output +where + F: std::future::Future + Send, + F::Output: Send, +{ + std::thread::scope(|scope| { + scope + .spawn(|| rt.block_on(fut)) + .join() + .expect("memory block_on worker thread panicked") + }) +} + fn payload_indexes() -> Vec<(&'static str, FieldType)> { vec![ ("subject", FieldType::Keyword), @@ -485,10 +625,7 @@ fn qdrant_index_already_exists(message: &str) -> bool { impl MemoryRuntime for QdrantMemoryRuntime { fn record(&self, rec: MemoryRecord) -> Result { - let id = rec.record_id; - let point = self.point_for(&rec)?; - self.upsert(point)?; - Ok(RecordId(id)) + self.block_on(self.record_async(rec)) } fn at_time(&self, subject: &HolderId, as_of: UnixTsMillis) -> Vec { @@ -691,6 +828,19 @@ mod tests { assert_eq!(block_on_runtime(&owned, async { 40 + 2 }), 42); } + /// #348: bridging back onto the *same* runtime we are already inside must not + /// panic. The old flavor-only guard saw `MultiThread` and took + /// `block_in_place(|| rt.block_on(..))`, re-entering `rt` → "Cannot start a + /// runtime from within a runtime". The identity guard routes this re-entry to + /// a dedicated thread instead. (The production code no longer nests bridges, + /// but this proves the backstop holds.) + #[test] + fn block_on_runtime_survives_reentry_on_its_own_runtime() { + let owned = owned_multi_thread_runtime(); + let out = owned.block_on(async { block_on_runtime(&owned, async { 6 * 7 }) }); + assert_eq!(out, 42); + } + fn fact( subject: &str, payload: serde_json::Value, diff --git a/crates/memory-qdrant/tests/hybrid_integration.rs b/crates/memory-qdrant/tests/hybrid_integration.rs index a2d35e8b..e4c83c30 100644 --- a/crates/memory-qdrant/tests/hybrid_integration.rs +++ b/crates/memory-qdrant/tests/hybrid_integration.rs @@ -1,25 +1,22 @@ //! Live integration tests for [`HybridMemoryRetriever`] — dense + sparse recall //! over a real Qdrant collection. //! -//! Two gates: -//! - `QDRANT_INTEGRATION_TEST=1` — needed by every test (the dense half is a real -//! Qdrant ANN search). Tests that only exercise the wiring use the deterministic -//! [`MockEmbedder`], so this gate alone runs them. -//! - `EMBEDDINGS_LIVE_TEST=1` — additionally required by the *semantic* tests, -//! which download and run the real BGE-small model so recall matches on meaning. -//! -//! The multi-thread flavor matters: the hybrid retriever calls the synchronous -//! `QdrantMemoryRuntime` (which bridges its async client with `block_in_place`) -//! from inside this async test, and `block_in_place` requires a multi-threaded -//! runtime. +//! Every test is `#[ignore]`d: it needs a live Qdrant, which CI lacks by +//! default. `#[ignore]` (not a silent env early-return) keeps them off the +//! default suite, so a skip reports as `ignored`, never a masked `passed` +//! (#358). Run them with `-- --ignored`: //! //! ```text //! docker run -p 6334:6334 qdrant/qdrant -//! QDRANT_INTEGRATION_TEST=1 cargo test -p ardur-memory-qdrant --test hybrid_integration -//! # plus the semantic tests: -//! QDRANT_INTEGRATION_TEST=1 EMBEDDINGS_LIVE_TEST=1 \ -//! cargo test -p ardur-memory-qdrant --test hybrid_integration +//! QDRANT_INTEGRATION_TEST=1 QDRANT_URL=http://localhost:6334 \ +//! cargo test -p ardur-memory-qdrant --test hybrid_integration -- --ignored //! ``` +//! +//! The two *semantic* tests additionally download and run the real BGE-small +//! model (so recall matches on meaning); set `EMBED_MODEL` and run them by name. +//! The multi-thread flavor matters: the hybrid retriever bridges its async +//! Qdrant client with `block_in_place` from inside the per-test async runtime, +//! and `block_in_place` requires a multi-threaded runtime. use std::sync::Arc; @@ -31,22 +28,11 @@ use ardur_memory_qdrant::{ QdrantMemoryConfig, QdrantMemoryRuntime, }; -/// `Some(config)` only when the Qdrant gate is set; otherwise the caller returns. -fn qdrant_gate(collection: &str) -> Option { - if std::env::var("QDRANT_INTEGRATION_TEST").as_deref() != Ok("1") { - eprintln!("skipping {collection}: set QDRANT_INTEGRATION_TEST=1 to run"); - return None; - } - Some(QdrantMemoryConfig::from_env().with_collection_name(collection)) -} - -/// Whether the live-embedder gate is set (the semantic tests need it). -fn embeddings_live() -> bool { - if std::env::var("EMBEDDINGS_LIVE_TEST").as_deref() != Ok("1") { - eprintln!("skipping semantic assertion: set EMBEDDINGS_LIVE_TEST=1 to run"); - return false; - } - true +/// The Qdrant config for `collection`. Endpoint from `QDRANT_URL` (default +/// `http://localhost:6334`); `#[ignore]` gates these tests, not an env +/// early-return, so a skip can never masquerade as a pass (#358). +fn config(collection: &str) -> QdrantMemoryConfig { + QdrantMemoryConfig::from_env().with_collection_name(collection) } fn fact(subject: &str, predicate: &str, object: &str, t: u64) -> MemoryRecord { @@ -83,10 +69,9 @@ fn async_rt() -> tokio::runtime::Runtime { /// `record` writes to **both** backends: the durable Qdrant store (bi-temporal /// read finds it) and the BM25 lexical index (a term-only query surfaces it). #[test] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] fn record_writes_to_both() { - let Some(cfg) = qdrant_gate("ardur_hyb_both") else { - return; - }; + let cfg = config("ardur_hyb_both"); let async_rt = async_rt(); let hybrid = retriever(cfg, Arc::new(MockEmbedder::new(384))); @@ -130,10 +115,9 @@ fn record_writes_to_both() { /// `search` returns at most `top_k` records end-to-end. #[test] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] fn search_respects_top_k() { - let Some(cfg) = qdrant_gate("ardur_hyb_topk") else { - return; - }; + let cfg = config("ardur_hyb_topk"); let async_rt = async_rt(); let hybrid = retriever(cfg, Arc::new(MockEmbedder::new(384))); @@ -160,13 +144,9 @@ fn search_respects_top_k() { /// semantically close still surfaces it (the dense half), while the dense half /// also keeps an unrelated fact away. Gated on the live embedder. #[test] +#[ignore = "requires a live Qdrant and the BGE-small model (EMBED_MODEL); run by name with `-- --ignored`"] fn semantic_hit_gated() { - let Some(cfg) = qdrant_gate("ardur_hyb_semantic") else { - return; - }; - if !embeddings_live() { - return; - } + let cfg = config("ardur_hyb_semantic"); let async_rt = async_rt(); let embedder = Arc::new(FastEmbedEmbedder::from_env().expect("load embedder")); let hybrid = retriever(cfg, embedder); @@ -203,13 +183,9 @@ fn semantic_hit_gated() { /// and the semantic axis outranks records strong on only one. Gated on the live /// embedder (so the semantic axis is real). #[test] +#[ignore = "requires a live Qdrant and the BGE-small model (EMBED_MODEL); run by name with `-- --ignored`"] fn hybrid_beats_either() { - let Some(cfg) = qdrant_gate("ardur_hyb_beats") else { - return; - }; - if !embeddings_live() { - return; - } + let cfg = config("ardur_hyb_beats"); let async_rt = async_rt(); let embedder = Arc::new(FastEmbedEmbedder::from_env().expect("load embedder")); let hybrid = retriever(cfg, embedder); @@ -255,10 +231,9 @@ fn hybrid_beats_either() { /// surfaces it — while a still-live memory in the same subject stays recallable. /// Gated on the Qdrant integration gate (the dense half is a real ANN search). #[test] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] fn recall_excludes_invalidated_memory() { - let Some(cfg) = qdrant_gate("ardur_hyb_inval") else { - return; - }; + let cfg = config("ardur_hyb_inval"); let async_rt = async_rt(); let hybrid = retriever(cfg, Arc::new(MockEmbedder::new(384))); let subject = HolderId::from("user:inval"); diff --git a/crates/memory-qdrant/tests/integration.rs b/crates/memory-qdrant/tests/integration.rs index b8f76673..09fdc5c2 100644 --- a/crates/memory-qdrant/tests/integration.rs +++ b/crates/memory-qdrant/tests/integration.rs @@ -1,11 +1,14 @@ //! Live Qdrant integration tests for [`QdrantMemoryRuntime`]. //! -//! Gated on `QDRANT_INTEGRATION_TEST=1` so CI (which has no Qdrant) skips them. -//! To run locally: +//! Each is `#[ignore]`d: it needs a live Qdrant, which CI lacks by default. +//! `#[ignore]` (not a silent env early-return) keeps them off the default suite, +//! so a skip reports as `ignored`, never a masked `passed` (#358). Run with +//! `-- --ignored`: //! //! ```text //! docker run -p 6333:6333 -p 6334:6334 qdrant/qdrant -//! QDRANT_INTEGRATION_TEST=1 cargo test -p ardur-memory-qdrant --test integration +//! QDRANT_INTEGRATION_TEST=1 QDRANT_URL=http://localhost:6334 \ +//! cargo test -p ardur-memory-qdrant --test integration -- --ignored //! ``` //! //! Each test uses its own collection name so they do not collide when run in @@ -16,14 +19,11 @@ use ardur_memory::{ }; use ardur_memory_qdrant::{MemorySnapshot, QdrantMemoryConfig, QdrantMemoryRuntime}; -/// Skip-or-config: returns `None` (and the caller returns early) unless the gate -/// var is set. When enabled, builds a config pointed at the given collection. -fn gate(collection: &str) -> Option { - if std::env::var("QDRANT_INTEGRATION_TEST").as_deref() != Ok("1") { - eprintln!("skipping {collection}: set QDRANT_INTEGRATION_TEST=1 to run"); - return None; - } - Some(QdrantMemoryConfig::from_env().with_collection_name(collection)) +/// The Qdrant config for `collection`. Endpoint from `QDRANT_URL` (default +/// `http://localhost:6334`); `#[ignore]` gates these tests, not an env +/// early-return, so a skip can never masquerade as a pass (#358). +fn config(collection: &str) -> QdrantMemoryConfig { + QdrantMemoryConfig::from_env().with_collection_name(collection) } fn fact(subject: &str, payload: serde_json::Value, t: u64) -> MemoryRecord { @@ -40,10 +40,9 @@ fn fact(subject: &str, payload: serde_json::Value, t: u64) -> MemoryRecord { /// Insert a record, then read it back via the bi-temporal "as-of" view. #[test] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] fn insert_then_query() { - let Some(cfg) = gate("ardur_it_insert_query") else { - return; - }; + let cfg = config("ardur_it_insert_query"); let rt = QdrantMemoryRuntime::connect(cfg).expect("connect"); rt.delete_collection().ok(); rt.init().expect("init"); @@ -65,10 +64,9 @@ fn insert_then_query() { /// Invalidation cuts off the chain from the cutoff forward, but history is /// retained and the pre-cutoff past is still readable. #[test] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] fn invalidate_preserves_history_and_past() { - let Some(cfg) = gate("ardur_it_invalidate") else { - return; - }; + let cfg = config("ardur_it_invalidate"); let rt = QdrantMemoryRuntime::connect(cfg).expect("connect"); rt.delete_collection().ok(); rt.init().expect("init"); @@ -121,10 +119,9 @@ fn invalidate_preserves_history_and_past() { /// The snapshot hook creates a Qdrant snapshot and records a `MemorySnapshot` /// event on the receipt chain. #[test] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] fn snapshot_into_receipt_records_event() { - let Some(cfg) = gate("ardur_it_snapshot") else { - return; - }; + let cfg = config("ardur_it_snapshot"); let rt = QdrantMemoryRuntime::connect(cfg).expect("connect"); rt.delete_collection().ok(); rt.init().expect("init"); @@ -145,10 +142,9 @@ fn snapshot_into_receipt_records_event() { /// readable after the instance is dropped and a fresh one reconnects to the same /// collection — a simulated process restart. #[test] +#[ignore = "requires a live Qdrant; run with `-- --ignored` (see module docs)"] fn survives_simulated_restart() { - let Some(cfg) = gate("ardur_it_restart") else { - return; - }; + let cfg = config("ardur_it_restart"); let user = HolderId::from("user:it-restart"); // First "process": write, then drop the whole backend (its client + runtime).