Extend query profiles and improve timeout histogram behavior#181
Conversation
Enable fixture-dependent query profile plumbing across generation and run commands, add vendor-specific fixture readiness/index setup for Neo4j and Memgraph, and align small/medium/large scripts so the same fixture workload can be reproduced consistently while investigating rust-engine memory issues. Co-Authored-By: Oz <oz-agent@warp.dev>
Co-Authored-By: Oz <oz-agent@warp.dev>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
|
Warning Review limit reached
Next review available in: 51 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughChangesQuery coverage and observability
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant BenchmarkCLI
participant QueryRepository
participant VendorClient
participant VendorDatabase
BenchmarkCLI->>QueryRepository: select query coverage profile
QueryRepository->>BenchmarkCLI: return profile-specific queries
BenchmarkCLI->>VendorClient: detect capabilities and prepare fixtures
VendorClient->>VendorDatabase: create indexes and seed fixture data
VendorDatabase-->>VendorClient: report query readiness
VendorClient-->>BenchmarkCLI: start benchmark execution
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Co-Authored-By: Oz <oz-agent@warp.dev>
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (3)
src/main.rs (1)
1549-1661: 🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy liftApply fixture setup when a cached local backup already exists.
The early returns at Lines 1576 and 2192 bypass the new fixture capability checks and setup. Consequently, repeated
load --query-profile fixture-dependentcalls leave older Neo4j/Memgraph backups unmodified despite the documented profile-aware loading behavior.Restore the cached backup, apply fixture/index setup, and persist the upgraded backup instead of returning immediately for fixture-dependent profiles.
Also applies to: 2160-2275
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main.rs` around lines 1549 - 1661, Update the cached-backup early-return paths in both Neo4j and Memgraph loading flows to handle fixture-dependent query profiles before returning. Restore the cached backup, run the required fixture capability checks and fixture/index setup using the existing profile-aware helpers, then persist the upgraded backup; retain the current skip behavior for profiles that do not require fixtures.src/aggregator.rs (1)
855-915: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winPreserve timeout metric presence separately from its value.
Line 915 appends
0.0when no timeout sample exists, making missing data indistinguishable from a valid 0% timeout rate. Consequently, the dashboard either hides legitimate zero values or shows the column for unsupported vendors.Only append the slot when a sample exists, then let the dashboard use slot presence.
Proposed producer/consumer fix
- arr.push(timeout_rates.get(&query).copied().unwrap_or(0.0)); + if let Some(rate) = timeout_rates.get(&query) { + arr.push(*rate); + }- histogram.length > 11 && - Number.isFinite(histogram[11]) && - histogram[11] > 0 + histogram.length > 11 && + Number.isFinite(histogram[11])🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/aggregator.rs` around lines 855 - 915, Update the timeout-rate handling in the aggregation flow around timeout_rates and the final arr.push so missing timeout samples remain distinguishable from valid 0.0 values. Track whether each query has an actual timeout metric sample, append the timeout slot only for those queries, and omit it otherwise; preserve valid zero-rate values and existing percentile slots.src/memgraph_client.rs (1)
181-223: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winPropagate result-stream errors instead of treating them as success.
while let Ok(Some(row))stops onstream.next()errors, but the outer path still returnsOk(()), so failed queries are recorded as successful and the latency/success metrics become wrong.Proposed fix
- while let Ok(Some(row)) = stream.next().await { - trace!("Row: {:?}", row); - black_box(row); + loop { + match stream.next().await { + Ok(Some(row)) => { + trace!("Row: {:?}", row); + black_box(row); + } + Ok(None) => break, + Err(e) => return Err(Neo4rsError(e)), + } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/memgraph_client.rs` around lines 181 - 223, Update the result-consumption loop in the memgraph_query closure to propagate errors returned by stream.next() instead of terminating successfully. Preserve row tracing and black_box handling for successful rows, and ensure the propagated error reaches the existing memgraph_result match so failure metrics and return values are recorded correctly.
🧹 Nitpick comments (1)
src/queries_repository.rs (1)
987-1093: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd profile-by-flavour catalog tests.
Cover that baseline excludes these queries, extended-core adds the temporal query only for supported flavours, and fixture-dependent adds all three exact smoke-query names. Execution-path capability detection depends on these names.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/queries_repository.rs` around lines 987 - 1093, Add catalog-level tests for the query registration flow around includes_extended_core() and includes_fixture_dependent(). Verify baseline profiles omit the three smoke queries and temporal_spatial_roundtrip, extended-core registers temporal_spatial_roundtrip only for supported flavours, and fixture-dependent registers all three exact smoke-query names; preserve the existing flavour-specific query text and names.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/main.rs`:
- Around line 421-427: Update is_timeout_error to recognize timeout text in both
OtherError and Neo4rsError BenchmarkError variants, using the error message
exposed by each variant and preserving case-insensitive matching. Ensure
server-side Neo4j timeout errors are classified as timeouts rather than
failures.
In `@src/memgraph_client.rs`:
- Around line 361-551: Update ensure_post_phase1_fixtures_ready to accept the
validated required fixture set from run_memgraph, and create, seed,
readiness-check, and probe only the requested fixture families (vector, node
full-text, and relationship full-text). Preserve retries and readiness
requirements for selected capabilities, while avoiding queries against
unavailable unrelated modules.
In `@src/neo4j_client.rs`:
- Around line 342-533: Update ensure_post_phase1_fixtures_ready to accept the
validated FixtureQueryPresence requirements and conditionally create, seed,
probe, and require readiness only for fixture query types actually present. Keep
unrelated fixture branches disabled, and preserve successful completion when the
supplied requirements contain only a subset of features.
---
Outside diff comments:
In `@src/aggregator.rs`:
- Around line 855-915: Update the timeout-rate handling in the aggregation flow
around timeout_rates and the final arr.push so missing timeout samples remain
distinguishable from valid 0.0 values. Track whether each query has an actual
timeout metric sample, append the timeout slot only for those queries, and omit
it otherwise; preserve valid zero-rate values and existing percentile slots.
In `@src/main.rs`:
- Around line 1549-1661: Update the cached-backup early-return paths in both
Neo4j and Memgraph loading flows to handle fixture-dependent query profiles
before returning. Restore the cached backup, run the required fixture capability
checks and fixture/index setup using the existing profile-aware helpers, then
persist the upgraded backup; retain the current skip behavior for profiles that
do not require fixtures.
In `@src/memgraph_client.rs`:
- Around line 181-223: Update the result-consumption loop in the memgraph_query
closure to propagate errors returned by stream.next() instead of terminating
successfully. Preserve row tracing and black_box handling for successful rows,
and ensure the propagated error reaches the existing memgraph_result match so
failure metrics and return values are recorded correctly.
---
Nitpick comments:
In `@src/queries_repository.rs`:
- Around line 987-1093: Add catalog-level tests for the query registration flow
around includes_extended_core() and includes_fixture_dependent(). Verify
baseline profiles omit the three smoke queries and temporal_spatial_roundtrip,
extended-core registers temporal_spatial_roundtrip only for supported flavours,
and fixture-dependent registers all three exact smoke-query names; preserve the
existing flavour-specific query text and names.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: ddd83484-5891-413b-acbf-bc75d9d12d6c
📒 Files selected for processing (16)
QUERY_EXPLANATIONS_AND_SAMPLES.mdreadme.mdscripts/run_large_benchmark.shscripts/run_medium_benchmark.shscripts/run_small_benchmark.shsrc/aggregator.rssrc/cli.rssrc/falkor/falkor_driver.rssrc/lib.rssrc/main.rssrc/memgraph_client.rssrc/neo4j_client.rssrc/queries_repository.rsui/app/components/VerticalBarChart.tsxui/app/components/dashboard.tsxui/components/ui/SidebarNavigation.tsx
| fn is_timeout_error(err: &benchmark::error::BenchmarkError) -> bool { | ||
| matches!( | ||
| err, | ||
| benchmark::error::BenchmarkError::OtherError(message) | ||
| if message.to_ascii_lowercase().contains("timeout") | ||
| ) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Locate the relevant function and error type definitions.
rg -n "fn is_timeout_error|enum BenchmarkError|Neo4rsError|OtherError" src/main.rs src -S
# Show the surrounding code around is_timeout_error.
sed -n '390,455p' src/main.rs
# Find the BenchmarkError definition.
rg -n "pub enum BenchmarkError|enum BenchmarkError" src -SRepository: FalkorDB/benchmark
Length of output: 14171
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,120p' src/error.rs
printf '\n---\n'
sed -n '260,380p' src/main.rs
printf '\n---\n'
rg -n "is_timeout_error\\(" src/main.rs src -SRepository: FalkorDB/benchmark
Length of output: 5747
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '2070,2125p' src/main.rsRepository: FalkorDB/benchmark
Length of output: 2596
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "TokioElapsed|timeout" src/memgraph_client.rs src/neo4j_client.rs src/main.rs src/error.rs -S
printf '\n---\n'
sed -n '1,260p' src/memgraph_client.rsRepository: FalkorDB/benchmark
Length of output: 10416
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,180p' src/neo4j_client.rs
printf '\n---\n'
sed -n '180,260p' src/neo4j_client.rs
printf '\n---\n'
sed -n '1,120p' src/error.rsRepository: FalkorDB/benchmark
Length of output: 11477
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '70,135p' src/neo4j_client.rsRepository: FalkorDB/benchmark
Length of output: 2581
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n "map_err\\(Neo4rsError\\)|tokio::time::timeout|TokioElapsed|timeout\\(" src/neo4j_client.rs src/memgraph_client.rs src/main.rs src/error.rs -SRepository: FalkorDB/benchmark
Length of output: 2874
🌐 Web query:
neo4rs error timeout message server-side timeout Neo4rsError Display
💡 Result:
In the neo4rs library, errors are handled via the Error enum, which differentiates between internal library errors and errors returned by the Neo4j database [1][2]. When a Neo4j server-side error occurs, it is encapsulated within the Error::Neo4j(Neo4jError) variant [1]. The Neo4jError struct contains both a code and a message [2][3]. When a transaction exceeds a configured timeout on the server side, Neo4j typically returns a specific status code, such as Neo.ClientError.Transaction.TransactionTimedOut [4]. The neo4rs library displays these as "Neo4j error [code]: [message]" [2]. Regarding timeouts specifically: 1. Server-Side Transaction Timeouts: If a query exceeds a transaction timeout defined on the server or via transaction configuration, the database returns a Neo4j error with a code indicating the timeout [4][5]. These are surfaced in neo4rs as Error::Neo4j(Neo4jError), and you can inspect the code and message methods on the Neo4jError struct to verify the specific cause [2][3]. 2. Client-Side/Connection Timeouts: Recent updates to neo4rs (e.g., PR #283) have introduced explicit support for connection and read timeouts [6]. If a connection attempt or read operation exceeds these configured limits, the library may return an Error::ConnectionTimedOut (or similar, depending on your specific version) rather than a server-side Neo4j error [6]. To debug timeout issues, it is recommended to: - Check the code value of the Neo4jError to see if it matches known Neo4j transaction timeout codes [4][2]. - Inspect the message for details, as server-side messages often provide context on whether the transaction was terminated due to a timeout [5][2]. - Ensure your neo4rs version is up-to-date to utilize native connection timeout features if you suspect network or pool-level latency [6].
Citations:
- 1: https://docs.rs/neo4rs/latest/neo4rs/enum.Error.html
- 2: https://docs.rs/neo4rs/latest/src/neo4rs/errors.rs.html
- 3: https://docs.rs/neo4rs/latest/neo4rs/struct.Neo4jError.html
- 4: https://errors.fyi/ns/neo4j/
- 5: https://stackoverflow.com/questions/57569999/neo4j-driver-with-transaction-timeout-how-to-detect-the-timeout-event
- 6: feat: add connection timeout, TCP keepalive, and pool health features neo4j-labs/neo4rs#283
Count timeout-like errors across BenchmarkError variants. is_timeout_error only checks OtherError, so a server-side Neo4j timeout surfaced as BenchmarkError::Neo4rsError(...) is recorded as a failure and the timeout rate is understated. Matching the error text across variants would cover both paths.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/main.rs` around lines 421 - 427, Update is_timeout_error to recognize
timeout text in both OtherError and Neo4rsError BenchmarkError variants, using
the error message exposed by each variant and preserving case-insensitive
matching. Ensure server-side Neo4j timeout errors are classified as timeouts
rather than failures.
| pub async fn ensure_post_phase1_fixtures_ready(&self) -> BenchmarkResult<()> { | ||
| // Required fixture indexes (Memgraph currently has no IF NOT EXISTS for these forms). | ||
| self.run_query_allowing_already_exists( | ||
| "create_text_index_user_ft", | ||
| "CREATE TEXT INDEX bench_user_ft_idx ON :User(ft_text)", | ||
| ) | ||
| .await?; | ||
| self.run_query_allowing_already_exists( | ||
| "create_text_edge_index_friend_ft", | ||
| "CREATE TEXT EDGE INDEX bench_friend_ft_idx ON :Friend(ft_text)", | ||
| ) | ||
| .await?; | ||
| self.run_query_allowing_already_exists( | ||
| "create_vector_index_user_embedding", | ||
| "CREATE VECTOR INDEX bench_user_embedding_idx ON :User(embedding) WITH CONFIG {\"dimension\": 3, \"capacity\": 1000000, \"metric\": \"cos\"}", | ||
| ) | ||
| .await?; | ||
|
|
||
| // Deterministic fixture seeding (idempotent SET updates). | ||
| self.run_query_no_results( | ||
| "MATCH (u:User) \ | ||
| WHERE toInteger(u.id) % 97 = 0 \ | ||
| SET \ | ||
| u.ft_text = 'fixture_alice user_' + toString(u.id), \ | ||
| u.embedding = [ \ | ||
| toFloat((toInteger(u.id) % 10) + 1) / 10.0, \ | ||
| toFloat(((toInteger(u.id) + 3) % 10) + 1) / 10.0, \ | ||
| toFloat(((toInteger(u.id) + 6) % 10) + 1) / 10.0 \ | ||
| ]", | ||
| ) | ||
| .await?; | ||
| self.run_query_no_results( | ||
| "MATCH (s:User)-[r:Friend]->(d:User) \ | ||
| WHERE toInteger(s.id) % 97 = 0 \ | ||
| SET r.ft_text = 'fixture_blue edge_' + toString(s.id) + '_' + toString(d.id)", | ||
| ) | ||
| .await?; | ||
|
|
||
| const MAX_ATTEMPTS: u32 = 18; | ||
| const DELAY_SECS: u64 = 5; | ||
|
|
||
| for attempt in 1..=MAX_ATTEMPTS { | ||
| let user_ft_count = match self | ||
| .query_single_u64( | ||
| "MATCH (u:User) \ | ||
| WHERE u.ft_text CONTAINS 'fixture_alice' \ | ||
| RETURN count(u) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while checking Memgraph user fulltext fixture readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
| let rel_ft_count = match self | ||
| .query_single_u64( | ||
| "MATCH ()-[r:Friend]->() \ | ||
| WHERE r.ft_text CONTAINS 'fixture_blue' \ | ||
| RETURN count(r) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while checking Memgraph relationship fulltext fixture readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
| let embedding_count = match self | ||
| .query_single_u64( | ||
| "MATCH (u:User) \ | ||
| WHERE u.embedding IS NOT NULL \ | ||
| RETURN count(u) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while checking Memgraph vector fixture readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
|
|
||
| let vector_hits = match self | ||
| .query_single_u64( | ||
| "CALL vector_search.search('bench_user_embedding_idx', 1, [0.1, 0.2, 0.3]) \ | ||
| YIELD node, similarity \ | ||
| RETURN count(node) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while probing Memgraph vector index readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
| let fulltext_node_hits = match self | ||
| .query_single_u64( | ||
| "CALL text_search.search('bench_user_ft_idx', 'data.ft_text:fixture_alice') \ | ||
| YIELD node, score \ | ||
| RETURN count(node) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while probing Memgraph fulltext node index readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
| let fulltext_rel_hits = match self | ||
| .query_single_u64( | ||
| "CALL text_search.search_edges('bench_friend_ft_idx', 'data.ft_text:fixture_blue') \ | ||
| YIELD edge, score \ | ||
| RETURN count(edge) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while probing Memgraph fulltext relationship index readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
|
|
||
| if user_ft_count > 0 | ||
| && rel_ft_count > 0 | ||
| && embedding_count > 0 | ||
| && vector_hits > 0 | ||
| && fulltext_node_hits > 0 | ||
| && fulltext_rel_hits > 0 | ||
| { | ||
| info!( | ||
| "Memgraph post-phase fixtures ready after {} attempt(s): users_ft={}, rel_ft={}, embedding={}, vector_hits={}, fulltext_node_hits={}, fulltext_rel_hits={}", | ||
| attempt, | ||
| user_ft_count, | ||
| rel_ft_count, | ||
| embedding_count, | ||
| vector_hits, | ||
| fulltext_node_hits, | ||
| fulltext_rel_hits | ||
| ); | ||
| return Ok(()); | ||
| } | ||
|
|
||
| info!( | ||
| "Memgraph post-phase fixtures not fully ready yet (attempt {}/{}): users_ft={}, rel_ft={}, embedding={}, vector_hits={}, fulltext_node_hits={}, fulltext_rel_hits={}", | ||
| attempt, | ||
| MAX_ATTEMPTS, | ||
| user_ft_count, | ||
| rel_ft_count, | ||
| embedding_count, | ||
| vector_hits, | ||
| fulltext_node_hits, | ||
| fulltext_rel_hits | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| } | ||
|
|
||
| Err(OtherError( | ||
| "Timed out waiting for Memgraph post-phase fixture indexes/procedures to become ready" | ||
| .to_string(), | ||
| )) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Prepare only the fixture capabilities required by the query file.
run_memgraph selectively validates FixtureQueryPresence, but this method always creates and probes vector, node-text, and edge-text fixtures. Partial fixture query files can pass validation and subsequently fail because an unrelated module is unavailable.
Thread the required fixture set into this method and conditionally seed/probe each family.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/memgraph_client.rs` around lines 361 - 551, Update
ensure_post_phase1_fixtures_ready to accept the validated required fixture set
from run_memgraph, and create, seed, readiness-check, and probe only the
requested fixture families (vector, node full-text, and relationship full-text).
Preserve retries and readiness requirements for selected capabilities, while
avoiding queries against unavailable unrelated modules.
| pub async fn ensure_post_phase1_fixtures_ready(&self) -> BenchmarkResult<()> { | ||
| // Required fixture indexes (idempotent creation). | ||
| self.run_query_no_results( | ||
| "CREATE FULLTEXT INDEX bench_user_ft_idx IF NOT EXISTS \ | ||
| FOR (u:User) ON EACH [u.ft_text]", | ||
| ) | ||
| .await?; | ||
| self.run_query_no_results( | ||
| "CREATE FULLTEXT INDEX bench_friend_ft_idx IF NOT EXISTS \ | ||
| FOR ()-[r:Friend]-() ON EACH [r.ft_text]", | ||
| ) | ||
| .await?; | ||
| self.run_query_no_results( | ||
| "CREATE VECTOR INDEX bench_user_embedding_idx IF NOT EXISTS \ | ||
| FOR (u:User) ON (u.embedding) \ | ||
| OPTIONS {indexConfig: {`vector.dimensions`: 3, `vector.similarity_function`: 'cosine'}}", | ||
| ) | ||
| .await?; | ||
|
|
||
| // Deterministic fixture seeding (idempotent SET updates). | ||
| self.run_query_no_results( | ||
| "MATCH (u:User) \ | ||
| WHERE toInteger(u.id) % 97 = 0 \ | ||
| SET \ | ||
| u.ft_text = 'fixture_alice user_' + toString(u.id), \ | ||
| u.embedding = [ \ | ||
| toFloat((toInteger(u.id) % 10) + 1) / 10.0, \ | ||
| toFloat(((toInteger(u.id) + 3) % 10) + 1) / 10.0, \ | ||
| toFloat(((toInteger(u.id) + 6) % 10) + 1) / 10.0 \ | ||
| ]", | ||
| ) | ||
| .await?; | ||
| self.run_query_no_results( | ||
| "MATCH (s:User)-[r:Friend]->(d:User) \ | ||
| WHERE toInteger(s.id) % 97 = 0 \ | ||
| SET r.ft_text = 'fixture_blue edge_' + toString(s.id) + '_' + toString(d.id)", | ||
| ) | ||
| .await?; | ||
|
|
||
| const MAX_ATTEMPTS: u32 = 18; | ||
| const DELAY_SECS: u64 = 5; | ||
|
|
||
| for attempt in 1..=MAX_ATTEMPTS { | ||
| let user_ft_count = match self | ||
| .query_single_u64( | ||
| "MATCH (u:User) \ | ||
| WHERE u.ft_text CONTAINS 'fixture_alice' \ | ||
| RETURN count(u) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while checking Neo4j user fulltext fixture readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
| let rel_ft_count = match self | ||
| .query_single_u64( | ||
| "MATCH ()-[r:Friend]->() \ | ||
| WHERE r.ft_text CONTAINS 'fixture_blue' \ | ||
| RETURN count(r) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while checking Neo4j relationship fulltext fixture readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
| let embedding_count = match self | ||
| .query_single_u64( | ||
| "MATCH (u:User) \ | ||
| WHERE u.embedding IS NOT NULL \ | ||
| RETURN count(u) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while checking Neo4j vector fixture readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
|
|
||
| let vector_hits = match self | ||
| .query_single_u64( | ||
| "CALL db.index.vector.queryNodes('bench_user_embedding_idx', 1, [0.1, 0.2, 0.3]) \ | ||
| YIELD node, score \ | ||
| RETURN count(node) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while probing Neo4j vector index readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
| let fulltext_node_hits = match self | ||
| .query_single_u64( | ||
| "CALL db.index.fulltext.queryNodes('bench_user_ft_idx', 'fixture_alice') \ | ||
| YIELD node, score \ | ||
| RETURN count(node) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while probing Neo4j fulltext node index readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
| let fulltext_rel_hits = match self | ||
| .query_single_u64( | ||
| "CALL db.index.fulltext.queryRelationships('bench_friend_ft_idx', 'fixture_blue') \ | ||
| YIELD relationship, score \ | ||
| RETURN count(relationship) AS count", | ||
| ) | ||
| .await | ||
| { | ||
| Ok(v) => v, | ||
| Err(e) => { | ||
| info!( | ||
| "Error while probing Neo4j fulltext relationship index readiness (attempt {}/{}): {}", | ||
| attempt, MAX_ATTEMPTS, e | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| continue; | ||
| } | ||
| }; | ||
|
|
||
| if user_ft_count > 0 | ||
| && rel_ft_count > 0 | ||
| && embedding_count > 0 | ||
| && vector_hits > 0 | ||
| && fulltext_node_hits > 0 | ||
| && fulltext_rel_hits > 0 | ||
| { | ||
| info!( | ||
| "Neo4j post-phase fixtures ready after {} attempt(s): users_ft={}, rel_ft={}, embedding={}, vector_hits={}, fulltext_node_hits={}, fulltext_rel_hits={}", | ||
| attempt, | ||
| user_ft_count, | ||
| rel_ft_count, | ||
| embedding_count, | ||
| vector_hits, | ||
| fulltext_node_hits, | ||
| fulltext_rel_hits | ||
| ); | ||
| return Ok(()); | ||
| } | ||
|
|
||
| info!( | ||
| "Neo4j post-phase fixtures not fully ready yet (attempt {}/{}): users_ft={}, rel_ft={}, embedding={}, vector_hits={}, fulltext_node_hits={}, fulltext_rel_hits={}", | ||
| attempt, | ||
| MAX_ATTEMPTS, | ||
| user_ft_count, | ||
| rel_ft_count, | ||
| embedding_count, | ||
| vector_hits, | ||
| fulltext_node_hits, | ||
| fulltext_rel_hits | ||
| ); | ||
| tokio::time::sleep(Duration::from_secs(DELAY_SECS)).await; | ||
| } | ||
|
|
||
| Err(OtherError( | ||
| "Timed out waiting for Neo4j post-phase fixture indexes/procedures to become ready" | ||
| .to_string(), | ||
| )) | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | 🏗️ Heavy lift
Scope fixture preparation to the queries actually present.
run_neo4j validates a selective FixtureQueryPresence, but this method always creates and probes vector, node-fulltext, and relationship-fulltext fixtures. A file containing only one fixture query can therefore pass validation and then fail on an unrelated feature.
Pass fixture requirements into this method and guard each setup/readiness branch accordingly.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/neo4j_client.rs` around lines 342 - 533, Update
ensure_post_phase1_fixtures_ready to accept the validated FixtureQueryPresence
requirements and conditionally create, seed, probe, and require readiness only
for fixture query types actually present. Keep unrelated fixture branches
disabled, and preserve successful completion when the supplied requirements
contain only a subset of features.
1 participant
Summary
Validation
Co-Authored-By: Oz oz-agent@warp.dev
Summary by CodeRabbit
New Features
Bug Fixes
Documentation