From a7efc8e434cbd9a913d83c5958b89040fa59c69d Mon Sep 17 00:00:00 2001 From: Evgeny Formanenko Date: Sat, 24 Oct 2026 02:13:41 +0300 Subject: [PATCH 1/2] chore(hotblocks): bump sqd-query-engine to add value sets and a 16 MiB request bound Co-Authored-By: Claude Opus 5.5 (1M context) --- Cargo.lock | 5 +++-- crates/hotblocks/Cargo.toml | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 062577cc..c76adc79 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5041,7 +5041,7 @@ dependencies = [ [[package]] name = "sqd-metadata" version = "0.1.0" -source = "git+https://github.com/subsquid/sqd-query-engine?rev=155460e0e74f6699c9bc856d4526bab027cc5591#155460e0e74f6699c9bc856d4526bab027cc5591" +source = "git+https://github.com/subsquid/sqd-query-engine?rev=fee829e5a260204c4b27a65698395b47a6f3ac2f#fee829e5a260204c4b27a65698395b47a6f3ac2f" dependencies = [ "anyhow", "indexmap", @@ -5105,10 +5105,11 @@ dependencies = [ [[package]] name = "sqd-query-engine" version = "0.1.0" -source = "git+https://github.com/subsquid/sqd-query-engine?rev=155460e0e74f6699c9bc856d4526bab027cc5591#155460e0e74f6699c9bc856d4526bab027cc5591" +source = "git+https://github.com/subsquid/sqd-query-engine?rev=fee829e5a260204c4b27a65698395b47a6f3ac2f#fee829e5a260204c4b27a65698395b47a6f3ac2f" dependencies = [ "anyhow", "arrow", + "base64 0.22.1", "bytes", "faster-hex", "indexmap", diff --git a/crates/hotblocks/Cargo.toml b/crates/hotblocks/Cargo.toml index 28df0472..818102ad 100644 --- a/crates/hotblocks/Cargo.toml +++ b/crates/hotblocks/Cargo.toml @@ -32,7 +32,7 @@ sqd-dataset = { path = "../dataset" } sqd-polars = { path = "../polars" } sqd-primitives = { path = "../primitives", features = ["serde", "valuable"] } sqd-query = { path = "../query", features = ["storage"] } -sqd-query-engine = { git = "https://github.com/subsquid/sqd-query-engine", rev = "155460e0e74f6699c9bc856d4526bab027cc5591" } +sqd-query-engine = { git = "https://github.com/subsquid/sqd-query-engine", rev = "fee829e5a260204c4b27a65698395b47a6f3ac2f" } sqd-storage = { path = "../storage" } tikv-jemallocator = "0.6.0" tokio = { workspace = true, features = ["full"] } From 7b45c4ee1e91617aebf4fb29afabb792ea2724bb Mon Sep 17 00:00:00 2001 From: Evgeny Formanenko Date: Sat, 24 Oct 2026 02:33:50 +0300 Subject: [PATCH 2/2] feat(hotblocks): let a stream request ask for the new query engine The shadow compares the new engine's answers but never serves them. With `x-sqd-query-engine: dynamic` the new engine alone parses and answers a stream request, with no shadow; without the header nothing changes. The service reads only the block bounds from the body, so a request may use what the legacy parser refuses, such as value sets. The plan compiles in an admitted task before the query waits for the head, so a request the engine refuses is refused at once. Co-Authored-By: Claude Opus 5.5 (1M context) --- Cargo.lock | 12 + crates/hotblocks/Cargo.toml | 4 + crates/hotblocks/spec/13-interface-binding.md | 7 + crates/hotblocks/src/api.rs | 174 +++++++++- crates/hotblocks/src/cli.rs | 9 +- crates/hotblocks/src/metrics.rs | 3 + crates/hotblocks/src/query/mod.rs | 3 + crates/hotblocks/src/query/range.rs | 99 ++++++ crates/hotblocks/src/query/response.rs | 11 +- crates/hotblocks/src/query/running.rs | 236 +++++++++----- crates/hotblocks/src/query/service.rs | 62 ++-- crates/hotblocks/src/query_engine/mod.rs | 26 +- crates/hotblocks/tests/query_engine_header.rs | 305 ++++++++++++++++++ 13 files changed, 830 insertions(+), 121 deletions(-) create mode 100644 crates/hotblocks/src/query/range.rs create mode 100644 crates/hotblocks/tests/query_engine_header.rs diff --git a/Cargo.lock b/Cargo.lock index c76adc79..c34e8c2b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4971,6 +4971,7 @@ dependencies = [ "arrow", "async-stream", "axum", + "base64 0.22.1", "bytes", "chrono", "clap", @@ -5004,6 +5005,8 @@ dependencies = [ "tracing", "tracing-subscriber", "url", + "xorf", + "xxhash-rust", "zstd", ] @@ -6301,6 +6304,15 @@ version = "0.13.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "66fee0b777b0f5ac1c69bb06d361268faafa61cd4682ae064a171c16c433e9e4" +[[package]] +name = "xorf" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f13846c9009c0e550697ddde851e934b5dfafbce757a5ef5edfdb27ea20803e" +dependencies = [ + "libm", +] + [[package]] name = "xxhash-rust" version = "0.8.15" diff --git a/crates/hotblocks/Cargo.toml b/crates/hotblocks/Cargo.toml index 818102ad..586349a5 100644 --- a/crates/hotblocks/Cargo.toml +++ b/crates/hotblocks/Cargo.toml @@ -45,6 +45,10 @@ url = { workspace = true, features = ["serde"] } [dev-dependencies] anyhow = { workspace = true } arrow = { workspace = true } +# Build the value sets a client sends to the new engine. +base64 = "0.22" +xorf = { version = "0.13", default-features = false, features = ["binary-fuse"] } +xxhash-rust = { version = "0.8.15", features = ["xxh3"] } reqwest = { workspace = true } sqd-hotblocks-harness = { path = "../hotblocks-harness" } tempfile = { workspace = true } diff --git a/crates/hotblocks/spec/13-interface-binding.md b/crates/hotblocks/spec/13-interface-binding.md index 05b79352..fdbb8620 100644 --- a/crates/hotblocks/spec/13-interface-binding.md +++ b/crates/hotblocks/spec/13-interface-binding.md @@ -18,6 +18,13 @@ annotated with the relevant GAP. `Content-Encoding` and `Vary: Accept-Encoding` set. - **IB-3** Correlation: responses carry a request-id header; clients MAY send a client identity header used for bounded-cardinality attribution (OB-10). +- **IB-3a** Engine choice: on the two query routes a client MAY send + `x-sqd-query-engine: dynamic` to have the new query engine answer alone, with no + shadow; `legacy`, or no header, leaves the answer to the legacy engine. Another value + is `MALFORMED_REQUEST` (400), and a dataset kind with no loaded catalog is + `UNSUPPORTED_QUERY` (400). With `dynamic` only the new engine parses the body, so a + request may use what the legacy parser refuses, such as value sets; a `type` other + than the dataset's kind is `MALFORMED_REQUEST` (400). ## 2. Operations → routes diff --git a/crates/hotblocks/src/api.rs b/crates/hotblocks/src/api.rs index b869a18a..ccbe56fc 100644 --- a/crates/hotblocks/src/api.rs +++ b/crates/hotblocks/src/api.rs @@ -33,12 +33,44 @@ use crate::{ BlockItemIsNotAvailable, BlockRangeMissing, Busy, QueryIsAboveTheHead, QueryKindMismatch, QueryTaskPanicked, UnknownDataset, UnsupportedQuery }, - query::QueryResponse, + query::{Engine, QueryRange, QueryResponse}, types::{ClientId, RetentionStrategy} }; const DEFAULT_CLIENT_ID: &str = "unknown"; +/// `dynamic` makes the new query engine answer a stream request on its own, with no +/// shadow; `legacy`, or no header, leaves it to the legacy engine. +const QUERY_ENGINE_HEADER: &str = "x-sqd-query-engine"; + +#[derive(Clone, Copy, Debug, PartialEq)] +enum RequestedEngine { + Legacy, + Dynamic +} + +impl RequestedEngine { + fn from_headers(headers: &HeaderMap) -> Result { + let Some(value) = headers.get(QUERY_ENGINE_HEADER) else { + return Ok(Self::Legacy); + }; + match value.as_bytes() { + b"legacy" => Ok(Self::Legacy), + b"dynamic" => Ok(Self::Dynamic), + _ => Err(format!("{QUERY_ENGINE_HEADER} must be 'legacy' or 'dynamic'")) + } + } + + /// The `engine` label of the response metrics. + fn label(requested: &Result) -> &'static str { + match requested { + Ok(Self::Legacy) => "legacy", + Ok(Self::Dynamic) => "dynamic", + Err(_) => "invalid" + } + } +} + #[derive(Clone, Copy, Debug)] enum ErrorCode { MalformedRequest, @@ -282,6 +314,11 @@ impl ResponseWithMetadata { self } + fn with_engine(mut self, engine: &'static str) -> Self { + self.labels.0.push(("engine", engine.to_owned())); + self + } + pub fn with_response(mut self, clause: F) -> Self where F: FnOnce() -> Response @@ -307,11 +344,14 @@ async fn stream( body: Bytes ) -> impl IntoResponse { let encoding = ContentEncoding::from_headers(&headers); - let response = stream_internal(app, dataset_id, body, false, client_id.clone(), encoding).await; + let engine = RequestedEngine::from_headers(&headers); + let engine_label = RequestedEngine::label(&engine); + let response = stream_internal(app, dataset_id, body, false, client_id.clone(), encoding, engine).await; ResponseWithMetadata::new() .with_client_id(&client_id) .with_dataset_id(dataset_id) .with_endpoint("/stream") + .with_engine(engine_label) .with_response(|| response) } @@ -323,11 +363,14 @@ async fn finalized_stream( body: Bytes ) -> impl IntoResponse { let encoding = ContentEncoding::from_headers(&headers); - let response = stream_internal(app, dataset_id, body, true, client_id.clone(), encoding).await; + let engine = RequestedEngine::from_headers(&headers); + let engine_label = RequestedEngine::label(&engine); + let response = stream_internal(app, dataset_id, body, true, client_id.clone(), encoding, engine).await; ResponseWithMetadata::new() .with_client_id(&client_id) .with_dataset_id(dataset_id) .with_endpoint("/finalized_stream") + .with_engine(engine_label) .with_response(|| response) } @@ -337,28 +380,67 @@ async fn stream_internal( body: Bytes, finalized: bool, client_id: ClientId, - encoding: ContentEncoding + encoding: ContentEncoding, + engine: Result ) -> Response { let dataset = get_dataset!(app, dataset_id); - let query: Query = match Json::::from_bytes(&body) { - Ok(Json(q)) => q, - Err(rejection) => return error_response(rejection.status(), ErrorCode::MalformedRequest, rejection.body_text()) + let engine = match engine { + Ok(engine) => engine, + Err(message) => return error_response(StatusCode::BAD_REQUEST, ErrorCode::MalformedRequest, message) }; - if let Err(err) = query.validate() { - return error_response(StatusCode::BAD_REQUEST, ErrorCode::MalformedRequest, err.to_string()); - } + let (range, engine) = match engine { + RequestedEngine::Legacy => { + let query: Query = match Json::::from_bytes(&body) { + Ok(Json(q)) => q, + Err(rejection) => { + return error_response(rejection.status(), ErrorCode::MalformedRequest, rejection.body_text()); + } + }; + + if let Err(err) = query.validate() { + return error_response(StatusCode::BAD_REQUEST, ErrorCode::MalformedRequest, err.to_string()); + } + + let range = match QueryRange::of(&query) { + Ok(range) => range, + Err(err) => return error_to_response(err.into(), &body) + }; + let shadow = app.shadow.sample(dataset_id, &query, &body); + (range, Engine::Legacy { query, shadow }) + } + // The legacy parser refuses what only the new engine knows, so the new + // engine parses and checks the body itself. + RequestedEngine::Dynamic => { + let kind = dataset.dataset_kind(); + let Some(catalog) = app.catalogs.get(kind) else { + return error_response( + StatusCode::BAD_REQUEST, + ErrorCode::UnsupportedQuery, + format!("the new query engine does not answer {} queries", kind.as_str()) + ); + }; - let shadow = app.shadow.sample(dataset_id, &query, &body); + let range = match QueryRange::from_body(&body, kind) { + Ok(range) => range, + Err(message) => return error_response(StatusCode::BAD_REQUEST, ErrorCode::MalformedRequest, message) + }; + let engine = Engine::New { + catalog, + body: body.clone() + }; + (range, engine) + } + }; let query_result = if finalized { app.query_service - .query_finalized(&dataset, query, body.clone(), client_id, encoding, shadow) + .query_finalized(&dataset, range, body.clone(), client_id, encoding, engine) .await } else { app.query_service - .query(&dataset, query, body.clone(), client_id, encoding, shadow) + .query(&dataset, range, body.clone(), client_id, encoding, engine) .await }; @@ -550,6 +632,8 @@ fn error_to_response(err: anyhow::Error, body: &Bytes) -> Response { (StatusCode::BAD_REQUEST, ErrorCode::RangeUnavailable) } else if err.is::() { (StatusCode::BAD_REQUEST, ErrorCode::ItemUnavailable) + } else if let Some(code) = new_engine_error_code(&err) { + (StatusCode::BAD_REQUEST, code) } else if err.is::() { (StatusCode::SERVICE_UNAVAILABLE, ErrorCode::Overloaded) } else { @@ -568,6 +652,18 @@ fn error_to_response(err: anyhow::Error, body: &Bytes) -> Response { error_response(status_code, error_code, err.to_string()) } +/// A new-engine error the client caused; `None` for one the service caused, +/// which is a 500. +fn new_engine_error_code(err: &anyhow::Error) -> Option { + use sqd_query_engine::error::{ErrorKind, error_kind}; + + match error_kind(err)? { + ErrorKind::TableNotFound | ErrorKind::ColumnNotFound => Some(ErrorCode::ItemUnavailable), + kind if kind.is_validation() => Some(ErrorCode::MalformedRequest), + _ => None + } +} + fn error_response(status: StatusCode, code: ErrorCode, message: impl Into) -> Response { // FIXME(GAP-36): expose `code` on the HTTP binding once a backwards-compatible // representation is chosen. sqd-portal forwards this body verbatim, so keep the @@ -605,6 +701,58 @@ mod error_response_tests { let body = axum::body::to_bytes(response.into_body(), usize::MAX).await.unwrap(); assert_eq!(body, "substrate queries are not supported"); } + + fn engine_error(kind: sqd_query_engine::error::ErrorKind) -> anyhow::Error { + anyhow::Error::new(sqd_query_engine::error::EngineError::new(kind, "refused")) + } + + fn classified(err: anyhow::Error) -> (StatusCode, Option<&'static str>) { + let response = error_to_response(err, &Bytes::new()); + let code = response.extensions().get::().map(|code| code.as_str()); + (response.status(), code) + } + + #[test] + fn new_engine_errors_are_classified_by_their_kind() { + use sqd_query_engine::error::ErrorKind; + + assert_eq!( + classified(engine_error(ErrorKind::TableNotFound)), + (StatusCode::BAD_REQUEST, Some("ITEM_UNAVAILABLE")) + ); + assert_eq!( + classified(engine_error(ErrorKind::UnknownField).context("compiling the query")), + (StatusCode::BAD_REQUEST, Some("MALFORMED_REQUEST")) + ); + assert_eq!( + classified(engine_error(ErrorKind::MalformedChunkData)), + (StatusCode::INTERNAL_SERVER_ERROR, Some("INTERNAL")) + ); + } + + #[test] + fn the_engine_header_names_one_of_two_engines() { + let headers = |value: &str| { + let mut headers = HeaderMap::new(); + headers.insert(QUERY_ENGINE_HEADER, HeaderValue::from_str(value).unwrap()); + headers + }; + + assert_eq!( + RequestedEngine::from_headers(&HeaderMap::new()), + Ok(RequestedEngine::Legacy) + ); + assert_eq!( + RequestedEngine::from_headers(&headers("legacy")), + Ok(RequestedEngine::Legacy) + ); + assert_eq!( + RequestedEngine::from_headers(&headers("dynamic")), + Ok(RequestedEngine::Dynamic) + ); + assert!(RequestedEngine::from_headers(&headers("new")).is_err()); + assert!(RequestedEngine::from_headers(&headers("Dynamic")).is_err()); + } } #[derive(Serialize)] diff --git a/crates/hotblocks/src/cli.rs b/crates/hotblocks/src/cli.rs index 51bd0efe..ec77631b 100644 --- a/crates/hotblocks/src/cli.rs +++ b/crates/hotblocks/src/cli.rs @@ -222,7 +222,8 @@ pub struct App { pub metrics_registry: prometheus_client::registry::Registry, pub known_clients: HashSet, pub settings: SettingsRef, - pub shadow: ShadowRef + pub shadow: ShadowRef, + pub catalogs: Arc } impl CLI { @@ -314,8 +315,9 @@ impl CLI { None => MetadataCatalogs::default() }; info!(kinds = ?catalogs.kinds(), "loaded the metadata catalogs"); + let catalogs = Arc::new(catalogs); let shadow = - Shadow::new(self.shadow_max_in_flight, runtime_settings.clone(), Arc::new(catalogs)).map(Arc::new)?; + Shadow::new(self.shadow_max_in_flight, runtime_settings.clone(), catalogs.clone()).map(Arc::new)?; Ok(App { db, @@ -325,7 +327,8 @@ impl CLI { metrics_registry, known_clients, settings: runtime_settings, - shadow + shadow, + catalogs }) } } diff --git a/crates/hotblocks/src/metrics.rs b/crates/hotblocks/src/metrics.rs index cd3cb411..a4962f30 100644 --- a/crates/hotblocks/src/metrics.rs +++ b/crates/hotblocks/src/metrics.rs @@ -312,6 +312,8 @@ pub(crate) fn report_shadow_run_finished() { /// The `QuerySlot::run` call site of a query task. #[derive(Copy, Clone, Hash, Debug, Eq, PartialEq)] pub(crate) enum QueryStep { + /// Compiles a new-engine query before it waits for the head. + Compile, /// Opens the response stream. First, /// Fills the buffer for the next data pack of an open stream. @@ -321,6 +323,7 @@ pub(crate) enum QueryStep { impl QueryStep { pub(crate) const fn as_str(self) -> &'static str { match self { + Self::Compile => "compile", Self::First => "first", Self::Next => "next" } diff --git a/crates/hotblocks/src/query/mod.rs b/crates/hotblocks/src/query/mod.rs index bd7e0aa1..52ff2b1a 100644 --- a/crates/hotblocks/src/query/mod.rs +++ b/crates/hotblocks/src/query/mod.rs @@ -1,4 +1,5 @@ mod executor; +mod range; mod response; mod running; mod service; @@ -7,6 +8,8 @@ mod static_snapshot; pub use executor::QueryExecutorCollector; pub(crate) use executor::TaskQueue; +pub use range::QueryRange; pub use response::*; +pub use running::Engine; pub use service::*; pub use static_snapshot::StaticChunkReader; diff --git a/crates/hotblocks/src/query/range.rs b/crates/hotblocks/src/query/range.rs new file mode 100644 index 00000000..de3ea39a --- /dev/null +++ b/crates/hotblocks/src/query/range.rs @@ -0,0 +1,99 @@ +use serde::Deserialize; +use sqd_primitives::BlockNumber; +use sqd_query::Query; + +use crate::{errors::UnsupportedQuery, types::DatasetKind}; + +/// What the service reads from a query before an engine runs it: the dataset +/// kind and the blocks the answer covers. +#[derive(Clone, Debug, PartialEq)] +pub struct QueryRange { + pub kind: DatasetKind, + pub first_block: BlockNumber, + pub last_block: Option, + pub parent_block_hash: Option +} + +impl QueryRange { + pub fn of(query: &Query) -> Result { + Ok(Self { + kind: DatasetKind::from_query(query)?, + first_block: query.first_block(), + last_block: query.last_block(), + parent_block_hash: query.parent_block_hash().map(str::to_owned) + }) + } + + /// Read only the block keys of a request body to a dataset of `kind` and + /// leave the rest, `type` too, to the new engine, which accepts requests the + /// legacy parser refuses. The error is the message for the client. + pub fn from_body(body: &[u8], kind: DatasetKind) -> Result { + #[derive(Deserialize)] + #[serde(rename_all = "camelCase")] + struct Keys { + #[serde(default)] + from_block: BlockNumber, + to_block: Option, + parent_block_hash: Option + } + + let keys: Keys = serde_json::from_slice(body).map_err(|err| format!("invalid query: {err}"))?; + + let reversed = keys.to_block.is_some_and(|to_block| to_block < keys.from_block); + if reversed { + return Err("got \"toBlock\" < \"fromBlock\"".to_string()); + } + + Ok(Self { + kind, + first_block: keys.from_block, + last_block: keys.to_block, + parent_block_hash: keys.parent_block_hash + }) + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + fn from_body(body: serde_json::Value) -> Result { + QueryRange::from_body(body.to_string().as_bytes(), DatasetKind::Evm) + } + + #[test] + fn the_range_ignores_every_other_key() { + let range = from_body(json!({ + "type": "evm", + "fromBlock": 10, + "toBlock": 20, + "parentBlockHash": "0xab", + "sets": {"s": {"fuse8": "AAAA", "canaries": []}}, + "logs": [{"address": [{"set": "s"}]}] + })) + .unwrap(); + + assert_eq!( + range, + QueryRange { + kind: DatasetKind::Evm, + first_block: 10, + last_block: Some(20), + parent_block_hash: Some("0xab".to_string()) + } + ); + } + + #[test] + fn the_range_reads_the_keys_as_the_legacy_parser_does() { + let body = json!({"type": "evm", "toBlock": 5}); + let legacy = QueryRange::of(&Query::from_json_value(body.clone()).unwrap()).unwrap(); + + assert_eq!(from_body(body), Ok(legacy)); + assert!(from_body(json!({"type": "evm", "fromBlock": 6, "toBlock": 5})).is_err()); + assert!(from_body(json!({"type": "evm", "fromBlock": -1})).is_err()); + assert!(from_body(json!({"type": "evm", "parentBlockHash": 1})).is_err()); + } +} diff --git a/crates/hotblocks/src/query/response.rs b/crates/hotblocks/src/query/response.rs index b9b6d436..1ff59d5c 100644 --- a/crates/hotblocks/src/query/response.rs +++ b/crates/hotblocks/src/query/response.rs @@ -6,12 +6,12 @@ use std::{ use anyhow::bail; use bytes::Bytes; use sqd_primitives::BlockRef; -use sqd_query::Query; use sqd_storage::db::DatasetId; use super::{ executor::{QueryExecutor, QuerySlot}, - running::{RunningQuery, RunningQueryStats}, + range::QueryRange, + running::{ReadyEngine, RunningQuery, RunningQueryStats}, slow_tasks::{SlowTasksRef, TaskReporter} }; use crate::{ @@ -21,7 +21,6 @@ use crate::{ QueryStep, STREAM_BLOCKS, STREAM_BLOCKS_PER_SECOND, STREAM_BYTES, STREAM_BYTES_PER_SECOND, STREAM_CHUNKS, STREAM_DURATIONS }, - shadow::ShadowQuery, types::{ClientId, DBRef} }; @@ -91,13 +90,13 @@ impl QueryResponse { slow_tasks: SlowTasksRef, db: DBRef, dataset_id: DatasetId, - query: Query, + range: QueryRange, body: Bytes, only_finalized: bool, time_limit: Option, client_id: ClientId, encoding: ContentEncoding, - shadow: Option + engine: ReadyEngine ) -> anyhow::Result { let Some(slot) = executor.get_slot() else { bail!(Busy) }; @@ -109,7 +108,7 @@ impl QueryResponse { QueryStep::First, move |slot| -> anyhow::Result<_> { let mut runner = - RunningQuery::new(db, dataset_id, &query, only_finalized, encoding, shadow).map(Box::new)?; + RunningQuery::new(db, dataset_id, &range, only_finalized, encoding, engine).map(Box::new)?; next_run(&mut runner, slot)?; Ok(runner) }, diff --git a/crates/hotblocks/src/query/running.rs b/crates/hotblocks/src/query/running.rs index 1eed561d..a3b8ec43 100644 --- a/crates/hotblocks/src/query/running.rs +++ b/crates/hotblocks/src/query/running.rs @@ -1,10 +1,11 @@ -use std::io::Write; +use std::{io::Write, sync::Arc}; use anyhow::{anyhow, bail, ensure}; use bytes::{BufMut, Bytes, BytesMut}; use flate2::{Compression, write::GzEncoder}; use sqd_primitives::{BlockNumber, BlockRef}; use sqd_query::{JsonLinesWriter, Plan, Query}; +use sqd_query_engine::metadata::DatasetDescription; use sqd_storage::db::{Chunk as StorageChunk, DatasetId}; use zstd::stream::write::Encoder as ZstdEncoder; @@ -12,16 +13,50 @@ use crate::{ encoding::ContentEncoding, errors::{BlockItemIsNotAvailable, BlockRangeMissing, QueryIsAboveTheHead, QueryKindMismatch}, metrics::{QUERIED_BLOCKS, QUERIED_CHUNKS}, - query::static_snapshot::{StaticChunkIterator, StaticChunkReader, StaticSnapshot}, + query::{ + QueryRange, + static_snapshot::{StaticChunkIterator, StaticChunkReader, StaticSnapshot} + }, + query_engine, shadow::{ChunkBounds, ShadowQuery}, - types::{ClientId, DBRef, DatasetKind} + types::{ClientId, DBRef} }; +/// The engine that answers a query. +pub enum Engine { + /// The legacy engine, shadowed by the new one when the query is sampled. + Legacy { query: Query, shadow: Option }, + /// The new engine alone, with the request body to compile against the catalog. + New { + catalog: Arc, + body: Bytes + } +} + +/// An [`Engine`] once the service has compiled the new engine's plan, which it +/// does before the query waits for the head. +pub enum ReadyEngine { + Legacy { query: Query, shadow: Option }, + New(query_engine::Plan) +} + +/// An [`Engine`] with its plan compiled. +enum ChunkEngine { + Legacy { plan: Plan, shadow: Option }, + New(query_engine::Plan) +} + struct LeftOver { chunk: StaticChunkReader, next_block: BlockNumber } +/// What one chunk wrote to the response. +struct Written { + blocks: usize, + last_block: BlockNumber +} + #[derive(Clone, Copy)] pub struct RunningQueryStats { pub chunks_read: u64, @@ -106,41 +141,46 @@ impl Compressor { } pub struct RunningQuery { - plan: Plan, + engine: ChunkEngine, + /// Where the next chunk is read from, when not from its first block. + first_block: Option, + /// The hash the next chunk's first block must have as its parent, when the + /// chunk is to check it. + parent_block_hash: Option, last_block: Option, left_over: Option, next_chunk: Option>, chunk_iterator: StaticChunkIterator, finalized_head: Option, buf: Compressor, - stats: RunningQueryStats, - shadow: Option + stats: RunningQueryStats } -fn finalized_query_last_block(query: &Query, finalized_head: Option<&BlockRef>) -> anyhow::Result { +fn finalized_query_last_block( + last_block: Option, + finalized_head: Option<&BlockRef> +) -> anyhow::Result { let Some(finalized_head) = finalized_head else { bail!(QueryIsAboveTheHead { finalized_head: None }) }; - Ok(query - .last_block() - .map_or(finalized_head.number, |end| end.min(finalized_head.number))) + Ok(last_block.map_or(finalized_head.number, |end| end.min(finalized_head.number))) } impl RunningQuery { pub fn new( db: DBRef, dataset_id: DatasetId, - query: &Query, + range: &QueryRange, only_finalized: bool, encoding: ContentEncoding, - shadow: Option + engine: ReadyEngine ) -> anyhow::Result { let snapshot = StaticSnapshot::new(db); let finalized_head = match snapshot.get_label(dataset_id)? { None => bail!("dataset {} does not exist", dataset_id), Some(label) => { - let kind = DatasetKind::from_query(query)?; + let kind = range.kind; ensure!( kind.storage_kind() == label.kind(), QueryKindMismatch { @@ -152,7 +192,7 @@ impl RunningQuery { } }; - let mut chunk_iterator = StaticChunkIterator::new(snapshot, dataset_id, query.first_block(), None); + let mut chunk_iterator = StaticChunkIterator::new(snapshot, dataset_id, range.first_block, None); let mut stats = RunningQueryStats::new(); let Some(first_chunk) = chunk_iterator.next().transpose()? else { @@ -162,15 +202,15 @@ impl RunningQuery { stats.blocks_read += first_chunk.last_block() - first_chunk.first_block() + 1; ensure!( - first_chunk.first_block() <= query.first_block(), + first_chunk.first_block() <= range.first_block, BlockRangeMissing { - first_block: query.first_block(), + first_block: range.first_block, last_block: first_chunk.first_block() - 1 } ); - let plan = if query.first_block() == first_chunk.first_block() { - if let Some(parent_hash) = query.parent_block_hash() { + let (first_block, parent_block_hash) = if range.first_block == first_chunk.first_block() { + if let Some(parent_hash) = range.parent_block_hash.as_deref() { ensure!( parent_hash == first_chunk.parent_block_hash(), sqd_query::UnexpectedBaseBlock { @@ -182,30 +222,36 @@ impl RunningQuery { } ); } - let mut plan = query.compile(); - plan.set_first_block(None); - plan.set_parent_block_hash(None); - plan + (None, None) } else { - query.compile() + (Some(range.first_block), range.parent_block_hash.clone()) + }; + + let engine = match engine { + ReadyEngine::Legacy { query, shadow } => ChunkEngine::Legacy { + plan: query.compile(), + shadow + }, + ReadyEngine::New(plan) => ChunkEngine::New(plan) }; let last_block = if only_finalized { - Some(finalized_query_last_block(query, finalized_head.as_ref())?) + Some(finalized_query_last_block(range.last_block, finalized_head.as_ref())?) } else { - query.last_block() + range.last_block }; Ok(Self { - plan, + engine, + first_block, + parent_block_hash, last_block, left_over: None, next_chunk: Some(Ok(first_chunk)), chunk_iterator, finalized_head, buf: Compressor::new(encoding)?, - stats, - shadow + stats }) } @@ -238,8 +284,7 @@ impl RunningQuery { /// Everything written to the buffer is always well-formed. pub fn write_next_chunk(&mut self) -> anyhow::Result<()> { let (chunk, is_new_chunk) = if let Some(left_over) = self.left_over.take() { - let first_block = left_over.next_block; - self.plan.set_first_block(first_block); + self.first_block = Some(left_over.next_block); (left_over.chunk, false) } else { let storage_chunk = self.next_chunk()?; @@ -247,67 +292,50 @@ impl RunningQuery { (chunk, true) }; - if self.last_block.map_or(false, |end| end < chunk.last_block()) { - let last_block = self.last_block; - self.plan.set_last_block(last_block); - } else { - self.plan.set_last_block(None); - } - - if let Some(shadow) = &self.shadow { - shadow.submit( - &chunk, - ChunkBounds { - first_block: self.plan.first_block(), - last_block: self.plan.last_block(), - parent_block_hash: self.plan.parent_block_hash().map(str::to_owned) + // Taken: no matter what, we are moving to the next chunk. + let bounds = ChunkBounds { + first_block: self.first_block.take(), + last_block: self.last_block.filter(|&end| end < chunk.last_block()), + parent_block_hash: self.parent_block_hash.take() + }; + let written = match &mut self.engine { + ChunkEngine::Legacy { plan, shadow } => { + if let Some(shadow) = shadow { + shadow.submit(&chunk, bounds.clone()); } - ); - } - - let query_result = chunk.with_reader(|reader| self.plan.execute(reader)).map_err(|err| { - if let Some(err) = err.downcast_ref::() { - return anyhow!(BlockItemIsNotAvailable { - item_name: err.table_name, - first_block: chunk.first_block(), - last_block: chunk.last_block() - }); + plan.set_first_block(bounds.first_block); + plan.set_last_block(bounds.last_block); + plan.set_parent_block_hash(bounds.parent_block_hash); + write_legacy(plan, &chunk, &mut self.buf) } - err - }); - - // no matter what, we are moving to the next chunk - self.plan.set_first_block(None); - self.plan.set_parent_block_hash(None); + ChunkEngine::New(plan) => { + plan.set_range( + bounds.first_block.unwrap_or(chunk.first_block()), + bounds.last_block, + bounds.parent_block_hash + ); + write_new(plan, &chunk, &mut self.buf) + } + }; - let Some(mut block_writer) = query_result? else { + let Some(written) = written? else { return Ok(()); }; if is_new_chunk { self.stats.chunks_returned += 1; } - self.stats.blocks_returned += block_writer.num_blocks() as u64; + self.stats.blocks_returned += written.blocks as u64; - if chunk.last_block() > block_writer.last_block() - && self.last_block.map_or(true, |end| end > block_writer.last_block()) - { + let chunk_has_more = chunk.last_block() > written.last_block; + let query_wants_more = self.last_block.map_or(true, |end| end > written.last_block); + if chunk_has_more && query_wants_more { self.left_over = Some(LeftOver { chunk, - next_block: block_writer.last_block() + 1 + next_block: written.last_block + 1 }) } - let mut json_lines_writer = JsonLinesWriter::new(&mut self.buf); - - json_lines_writer - .write_blocks(&mut block_writer) - .expect("IO errors are not possible"); - - json_lines_writer.finish().expect("IO errors are not possible"); - - self.buf.flush().expect("IO errors are not possible"); - Ok(()) } @@ -352,6 +380,60 @@ impl RunningQuery { } } +fn write_legacy(plan: &Plan, chunk: &StaticChunkReader, buf: &mut Compressor) -> anyhow::Result> { + let query_result = chunk.with_reader(|reader| plan.execute(reader)).map_err(|err| { + if let Some(err) = err.downcast_ref::() { + return anyhow!(BlockItemIsNotAvailable { + item_name: err.table_name, + first_block: chunk.first_block(), + last_block: chunk.last_block() + }); + } + err + }); + + let Some(mut block_writer) = query_result? else { + return Ok(None); + }; + let written = Written { + blocks: block_writer.num_blocks(), + last_block: block_writer.last_block() + }; + + let mut json_lines_writer = JsonLinesWriter::new(&mut *buf); + + json_lines_writer + .write_blocks(&mut block_writer) + .expect("IO errors are not possible"); + + json_lines_writer.finish().expect("IO errors are not possible"); + + buf.flush().expect("IO errors are not possible"); + + Ok(Some(written)) +} + +fn write_new( + plan: &query_engine::Plan, + chunk: &StaticChunkReader, + buf: &mut Compressor +) -> anyhow::Result> { + let Some(output) = chunk.with_reader(|reader| plan.execute(reader))? else { + return Ok(None); + }; + let written = Written { + blocks: output.num_blocks(), + last_block: output.last_block() + }; + + buf.write_all(&output.into_json_lines()) + .expect("IO errors are not possible"); + + buf.flush().expect("IO errors are not possible"); + + Ok(Some(written)) +} + #[cfg(test)] mod tests { use super::*; @@ -364,7 +446,7 @@ mod tests { })) .unwrap(); - let err = finalized_query_last_block(&query, None).unwrap_err(); + let err = finalized_query_last_block(query.last_block(), None).unwrap_err(); assert!(err.is::()); } } diff --git a/crates/hotblocks/src/query/service.rs b/crates/hotblocks/src/query/service.rs index e13d17c8..8a3a8a69 100644 --- a/crates/hotblocks/src/query/service.rs +++ b/crates/hotblocks/src/query/service.rs @@ -8,21 +8,24 @@ use std::{ use anyhow::{bail, ensure}; use bytes::Bytes; -use sqd_query::Query; +use sqd_query_engine::metadata::DatasetDescription; use super::{ executor::QueryExecutor, + range::QueryRange, response::QueryResponse, + running::{Engine, ReadyEngine}, slow_tasks::{SlowTasks, SlowTasksRef} }; use crate::{ dataset_controller::DatasetController, encoding::ContentEncoding, errors::{Busy, QueryIsAboveTheHead, QueryKindMismatch}, + metrics::QueryStep, query::QueryExecutorCollector, + query_engine, settings::SettingsRef, - shadow::ShadowQuery, - types::{ClientId, DBRef, DatasetKind} + types::{ClientId, DBRef} }; pub type QueryServiceRef = Arc; @@ -95,40 +98,40 @@ impl QueryService { pub async fn query( &self, dataset: &DatasetController, - query: Query, + range: QueryRange, body: Bytes, client_id: ClientId, encoding: ContentEncoding, - shadow: Option + engine: Engine ) -> anyhow::Result { - self.query_internal(dataset, query, body, false, client_id, encoding, shadow) + self.query_internal(dataset, range, body, false, client_id, encoding, engine) .await } pub async fn query_finalized( &self, dataset: &DatasetController, - query: Query, + range: QueryRange, body: Bytes, client_id: ClientId, encoding: ContentEncoding, - shadow: Option + engine: Engine ) -> anyhow::Result { - self.query_internal(dataset, query, body, true, client_id, encoding, shadow) + self.query_internal(dataset, range, body, true, client_id, encoding, engine) .await } async fn query_internal( &self, dataset: &DatasetController, - query: Query, + range: QueryRange, body: Bytes, finalized: bool, client_id: ClientId, encoding: ContentEncoding, - shadow: Option + engine: Engine ) -> anyhow::Result { - let query_kind = DatasetKind::from_query(&query)?; + let query_kind = range.kind; ensure!( dataset.dataset_kind() == query_kind, QueryKindMismatch { @@ -137,6 +140,11 @@ impl QueryService { } ); + let engine = match engine { + Engine::Legacy { query, shadow } => ReadyEngine::Legacy { query, shadow }, + Engine::New { catalog, body: request } => ReadyEngine::New(self.compile(catalog, request).await?) + }; + let target_head = if finalized { dataset.get_finalized_head() } else { @@ -144,9 +152,9 @@ impl QueryService { }; let should_wait = match target_head { - Some(head) if head.number >= query.first_block() => false, - Some(head) if head.number + 1 == query.first_block() => { - if let Some(parent_hash) = query.parent_block_hash() { + Some(head) if head.number >= range.first_block => false, + Some(head) if head.number + 1 == range.first_block => { + if let Some(parent_hash) = range.parent_block_hash.as_deref() { ensure!( head.hash == parent_hash, sqd_query::UnexpectedBaseBlock { @@ -166,9 +174,9 @@ impl QueryService { }; tokio::time::timeout(Duration::from_secs(5), async { if finalized { - dataset.wait_for_finalized_block(query.first_block()).await + dataset.wait_for_finalized_block(range.first_block).await } else { - dataset.wait_for_block(query.first_block()).await + dataset.wait_for_block(range.first_block).await } }) .await @@ -180,17 +188,33 @@ impl QueryService { self.slow_tasks.clone(), self.db.clone(), dataset.dataset_id(), - query, + range, body, finalized, None, client_id, encoding, - shadow + engine ) .await } + /// Compile in an admitted task before the query waits for the head: a large + /// body never holds a runtime thread, a busy service refuses it before any + /// parsing, and a request the engine refuses is refused at once rather than + /// answered as no data when the wait runs out. + async fn compile(&self, catalog: Arc, body: Bytes) -> anyhow::Result { + let Some(slot) = self.executor.get_slot() else { + bail!(Busy) + }; + slot.run( + QueryStep::Compile, + move |_| query_engine::Plan::compile(&body, catalog), + |_, _| {} + ) + .await? + } + pub fn metrics_collector(&self) -> QueryExecutorCollector { self.executor.metrics_collector() } diff --git a/crates/hotblocks/src/query_engine/mod.rs b/crates/hotblocks/src/query_engine/mod.rs index 3a389002..6c7d7a10 100644 --- a/crates/hotblocks/src/query_engine/mod.rs +++ b/crates/hotblocks/src/query_engine/mod.rs @@ -5,10 +5,10 @@ mod chunk; use std::{path::Path, sync::Arc}; use anyhow::Context; -use sqd_primitives::BlockNumber; +use sqd_primitives::{BlockNumber, BlockRef}; use sqd_query_engine::{ metadata::{DatasetDescription, load_dataset_description_strict}, - output::{ExecOptions, QueryOutput, execute_chunk_with}, + output::{ExecOptions, QueryOutput, UnexpectedBaseBlock, execute_chunk_with}, query::{compile, parse_query}, scan::ColumnarChunkReader }; @@ -105,6 +105,26 @@ impl Plan { /// block is selected. pub fn execute(&self, chunk: &ChunkReader<'_>) -> anyhow::Result> { let chunk = ColumnarChunkReader::new(HotChunk { reader: chunk }); - execute_chunk_with(&self.inner, &self.catalog, &chunk, self.options) + execute_chunk_with(&self.inner, &self.catalog, &chunk, self.options).map_err(as_legacy_fork) } } + +/// The engine's fork error as the legacy engine's, which the API answers with the +/// blocks a client rewinds to. Both say the same text. +fn as_legacy_fork(err: anyhow::Error) -> anyhow::Error { + let fork = match err.downcast::() { + Ok(fork) => fork, + Err(err) => return err + }; + anyhow::Error::new(sqd_query::UnexpectedBaseBlock { + prev_blocks: fork + .prev_blocks + .into_iter() + .map(|block| BlockRef { + number: block.number, + hash: block.hash + }) + .collect(), + expected_hash: fork.expected_hash + }) +} diff --git a/crates/hotblocks/tests/query_engine_header.rs b/crates/hotblocks/tests/query_engine_header.rs new file mode 100644 index 00000000..1936efd5 --- /dev/null +++ b/crates/hotblocks/tests/query_engine_header.rs @@ -0,0 +1,305 @@ +//! A stream request with `x-sqd-query-engine: dynamic` is answered by the new query +//! engine alone: with the bytes the legacy engine sends, and with no shadow. + +use std::{path::Path, sync::Arc}; + +use anyhow::{Result, ensure}; +use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; +use serde_json::{Value, json}; +use sqd_hotblocks_harness::{ + Harness, HarnessConfig, + chain::{Chain, Evm}, + types::{BlockNumber, block_hash} +}; +use xorf::BinaryFuse8; +use xxhash_rust::xxh3::xxh3_64; + +const START: BlockNumber = 1_000; +/// Copies of the pinned engine revision's catalogs; refresh them when the pin +/// moves. +const METADATA_DIR: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/metadata"); +const CONTRACT: &str = "0x3333333333333333333333333333333333333333"; +const SENDER: &str = "0x2222222222222222222222222222222222222222"; +const TRANSFER_TOPIC: &str = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"; +/// The catalog edit after which the new engine no longer knows the logs request, +/// which the legacy one answers. +const NO_LOGS: (&str, &str) = ( + " request:\n name: logs\n", + " request:\n name: no_logs\n" +); + +#[tokio::test(flavor = "multi_thread")] +async fn the_new_engine_answers_with_the_legacy_bytes() -> Result<()> { + let mut h = start(Some(Path::new(METADATA_DIR)), &[]).await?; + h.finalize(START + 40)?; + h.settle().await?; + + let fields = json!({ + "block": {"number": true, "hash": true, "parentHash": true, "timestamp": true}, + "transaction": {"transactionIndex": true, "hash": true, "from": true, "nonce": true}, + "log": {"logIndex": true, "transactionIndex": true, "address": true, "topics": true, "data": true} + }); + let queries = [ + Evm.scan_query(START, None, None), + Evm.scan_query(START + 7, Some(START + 23), None), + Evm.scan_query(START + 5, None, Some(&block_hash(START + 4, 0))), + json!({ + "type": "evm", "fromBlock": START, "fields": fields, + "logs": [{"address": [CONTRACT], "topic0": [TRANSFER_TOPIC], "transaction": true}] + }), + json!({ + "type": "evm", "fromBlock": START + 3, "fields": fields, + "transactions": [{"from": [SENDER], "logs": true}] + }), + json!({ + "type": "evm", "fromBlock": START, "toBlock": START + 30, "includeAllBlocks": true, + "fields": fields, "logs": [{"address": [CONTRACT]}] + }), + json!({ + "type": "evm", "fromBlock": START, "fields": fields, + "logs": [{"address": ["0x4444444444444444444444444444444444444444"]}] + }) + ]; + for query in &queries { + expect_same_answer(&h, "stream", query, 200).await?; + } + expect_same_answer(&h, "finalized-stream", &queries[0], 200).await?; + + // Only the requests without the header were shadowed. + let metrics = h.client.metrics().await?; + let shadowed = metrics.get("hotblocks_shadow_queries_total", Some(("dataset", &h.dataset))); + assert_eq!(shadowed, Some(queries.len() as f64 + 1.0)); + ensure!( + metrics.get("hotblocks_http_status_total", Some(("engine", "dynamic"))) > Some(0.0), + "no response was counted for the new engine" + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_new_engine_is_the_one_that_answers() -> Result<()> { + // Rolled in another order, every log's topics differ in the new engine's answer. + let catalogs = edited_catalogs( + "columns: [ topic0, topic1, topic2, topic3 ]", + "columns: [ topic1, topic0, topic2, topic3 ]" + )?; + let h = start(Some(catalogs.path()), &[]).await?; + + let query = Evm.scan_query(START, None, None); + let legacy = ask(&h, "stream", &query, None).await?; + let new = ask(&h, "stream", &query, Some("dynamic")).await?; + ensure!(legacy.status == 200 && new.status == 200, "{legacy:?}\n{new:?}"); + ensure!( + legacy.body != new.body, + "the legacy engine answered the request for the new one" + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_answer_the_new_engine_cuts_goes_on_from_the_cut() -> Result<()> { + // Every block outweighs the new engine's budget, so it cuts each chunk's + // answer after one block. + let catalogs = edited_catalogs( + " hash:\n type: string\n encoding: hex_bytes\n parent_hash:\n", + " hash:\n type: string\n encoding: hex_bytes\n weight: 100000000\n parent_hash:\n" + )?; + let h = start(Some(catalogs.path()), &[]).await?; + + for query in [ + Evm.scan_query(START, None, None), + Evm.scan_query(START + 3, Some(START + 15), None) + ] { + expect_same_answer(&h, "stream", &query, 200).await?; + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_new_engine_reports_a_fork_as_the_legacy_one_does() -> Result<()> { + let h = start(Some(Path::new(METADATA_DIR)), &[]).await?; + + // Some of these blocks start a chunk and some are inside one, where the + // engine checks the parent hash itself. + for first in [START + 5, START + 17, START + 33] { + let query = Evm.scan_query(first, None, Some(&block_hash(first - 1, 1))); + expect_same_answer(&h, "stream", &query, 409).await?; + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn an_unknown_engine_is_refused() -> Result<()> { + let h = start(Some(Path::new(METADATA_DIR)), &[]).await?; + + let answer = ask(&h, "stream", &Evm.scan_query(START, None, None), Some("newest")).await?; + ensure!(answer.status == 400, "{answer:?}"); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn the_new_engine_is_refused_without_a_catalog() -> Result<()> { + let h = start(None, &[]).await?; + + let answer = ask(&h, "stream", &Evm.scan_query(START, None, None), Some("dynamic")).await?; + ensure!(answer.status == 400, "{answer:?}"); + ensure!(answer.body.contains("new query engine"), "{answer:?}"); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_set_query_is_answered_only_with_the_header() -> Result<()> { + let h = start(Some(Path::new(METADATA_DIR)), &[]).await?; + + let fields = json!({"log": {"logIndex": true, "transactionIndex": true, "address": true}}); + let with_list = json!({ + "type": "evm", "fromBlock": START, "fields": fields, + "logs": [{"address": [CONTRACT]}] + }); + let with_set = json!({ + "type": "evm", "fromBlock": START, "fields": fields, + "sets": {"contracts": value_set(&[CONTRACT])}, + "logs": [{"address": [{"set": "contracts"}]}] + }); + + let list = ask(&h, "stream", &with_list, None).await?; + ensure!(list.status == 200 && list.body.contains(CONTRACT), "{list:?}"); + + let legacy = ask(&h, "stream", &with_set, None).await?; + ensure!(legacy.status == 422, "the legacy engine took a set: {legacy:?}"); + + let dynamic = ask(&h, "stream", &with_set, Some("dynamic")).await?; + ensure!( + dynamic == list, + "the set answered unlike the list:\nset: {dynamic:?}\nlist: {list:?}" + ); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_query_the_new_engine_refuses_is_a_client_error() -> Result<()> { + let catalogs = edited_catalogs(NO_LOGS.0, NO_LOGS.1)?; + let h = start(Some(catalogs.path()), &[]).await?; + + let query = Evm.scan_query(START, None, None); + let legacy = ask(&h, "stream", &query, None).await?; + ensure!(legacy.status == 200, "{legacy:?}"); + let refused = ask(&h, "stream", &query, Some("dynamic")).await?; + ensure!(refused.status == 400, "{refused:?}"); + + let other_kind = json!({"type": "solana", "fromBlock": START}); + let refused = ask(&h, "stream", &other_kind, Some("dynamic")).await?; + ensure!(refused.status == 400, "{refused:?}"); + + // Above the head both are refused before the wait for it, not answered as + // no data when the wait runs out. + let above_head = START + 1_000; + for query in [ + Evm.scan_query(above_head, None, None), + json!({"type": "solana", "fromBlock": above_head}) + ] { + let refused = ask(&h, "stream", &query, Some("dynamic")).await?; + ensure!(refused.status == 400, "{query} answered {refused:?}"); + } + Ok(()) +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_dynamic_query_is_compiled_only_once_admitted() -> Result<()> { + // With no query capacity, a query the new engine cannot compile is turned + // away as busy: compiling it first would answer 400. + let catalogs = edited_catalogs(NO_LOGS.0, NO_LOGS.1)?; + let h = start(Some(catalogs.path()), &["--query-task-queue", "0"]).await?; + + let answer = ask(&h, "stream", &Evm.scan_query(START, None, None), Some("dynamic")).await?; + ensure!(answer.status == 503, "{answer:?}"); + Ok(()) +} + +/// EVM blocks written and settled, with every request without the header +/// shadowed. +async fn start(metadata_dir: Option<&Path>, args: &[&str]) -> Result { + let mut cfg = HarnessConfig::from_block(env!("CARGO_BIN_EXE_sqd-hotblocks"), Arc::new(Evm), START); + cfg.sut_args.extend( + ["--shadow-percent", "100", "--shadow-max-in-flight", "64"] + .into_iter() + .map(String::from) + ); + if let Some(dir) = metadata_dir { + cfg.sut_args.push("--metadata-dir".to_string()); + cfg.sut_args.push(dir.display().to_string()); + } + cfg.sut_args.extend(args.iter().map(|arg| arg.to_string())); + + let mut h = Harness::start(cfg).await?; + h.produce(60)?; + h.settle().await?; + Ok(h) +} + +/// The test catalogs with one edit to `evm.yaml`, which makes the new engine +/// answer differently from the legacy one. +fn edited_catalogs(from: &str, to: &str) -> Result { + let evm = std::fs::read_to_string(Path::new(METADATA_DIR).join("evm.yaml"))?; + ensure!( + evm.matches(from).count() == 1, + "evm.yaml holds `{from}` other than once" + ); + + let dir = tempfile::tempdir()?; + std::fs::write(dir.path().join("evm.yaml"), evm.replace(from, to))?; + Ok(dir) +} + +/// A set as a client declares it: a binary fuse filter over the XXH3 of each +/// value's text, with the values as its canaries. +fn value_set(values: &[&str]) -> Value { + let mut keys: Vec = values.iter().map(|value| xxh3_64(value.as_bytes())).collect(); + keys.sort_unstable(); + keys.dedup(); + let filter = BinaryFuse8::try_from(&keys).expect("the keys are distinct"); + + let descriptor = &filter.descriptor; + let mut bytes = descriptor.seed.to_le_bytes().to_vec(); + bytes.extend(descriptor.segment_length.to_le_bytes()); + bytes.extend(descriptor.segment_length_mask.to_le_bytes()); + bytes.extend(descriptor.segment_count_length.to_le_bytes()); + bytes.extend(filter.fingerprints.iter()); + + json!({"fuse8": BASE64.encode(bytes), "canaries": values}) +} + +#[derive(Debug, PartialEq)] +struct Answer { + status: u16, + body: String +} + +async fn ask(h: &Harness, endpoint: &str, query: &Value, engine: Option<&str>) -> Result { + let client = reqwest::Client::builder().gzip(true).build()?; + let mut request = client + .post(format!("{}/datasets/{}/{endpoint}", h.sut.base_url(), h.dataset)) + .json(query); + if let Some(engine) = engine { + request = request.header("x-sqd-query-engine", engine); + } + + let response = request.send().await?; + Ok(Answer { + status: response.status().as_u16(), + body: response.text().await? + }) +} + +async fn expect_same_answer(h: &Harness, endpoint: &str, query: &Value, status: u16) -> Result<()> { + let legacy = ask(h, endpoint, query, None).await?; + ensure!(legacy.status == status, "{query} answered {legacy:?}"); + ensure!(!legacy.body.is_empty(), "{query} answered with no body"); + + let new = ask(h, endpoint, query, Some("dynamic")).await?; + ensure!( + new == legacy, + "{query} answered differently\nlegacy: {legacy:?}\nnew: {new:?}" + ); + Ok(()) +}