diff --git a/crates/core/README.md b/crates/core/README.md index b00d037..1dabf15 100644 --- a/crates/core/README.md +++ b/crates/core/README.md @@ -53,6 +53,9 @@ This is one of four language bindings published from the same Rust core. See the - [Sets](#sets) - [Lists](#lists) - [SQL Client](#sql-client) + - [`list_clusters`](#list_clusters) + - [`query_with_session`](#query_with_session) + - [`query_with_mpp_session`](#query_with_mpp_session) - [RPC & Tooling Access](#rpc--tooling-access) - [Crypto-micropayment lane (`rpc.call`)](#crypto-micropayment-lane-rpccall) - [Wallet generation](#wallet-generation) @@ -1756,11 +1759,26 @@ qn.kvstore.delete_list("my-list").await?; ### SQL Client -Accessed as `qn.sql`. Runs SQL queries against indexed blockchain data and fetches the database schema. Backed by `https://api.quicknode.com/sql/rest/v1/`. +Accessed as `qn.sql`. Runs SQL queries against indexed blockchain data and fetches the database schema. + +- Account host (API key): `https://api.quicknode.com/sql/rest/v1/` — `query` and `get_schema`. +- Public catalog / x402 drawdown: `https://x402.quicknode.com/sql/rest/v1/` — `list_clusters` and `query_with_session`. Use a keyless client so no `x-api-key` is sent. +- MPP session: `POST https://mpp.quicknode.com/session/sql/rest/v1/query` — `query_with_mpp_session`. + +##### `list_clusters` + +Lists clusters from the public catalog (`GET clusters`). Unauthenticated. + +**Returns**: `Vec` — each with `id` and `display_name`. + +```rust +// Rust +let clusters = qn.sql.list_clusters().await?; +``` ##### `query` -Executes a SQL query against a cluster and returns the result set. Paginate by writing `LIMIT`/`OFFSET` into the SQL. +Executes a SQL query against a cluster on the account host and returns the result set. Paginate by writing `LIMIT`/`OFFSET` into the SQL. **Parameters**: `QueryParams` with `query` (String, required) and `cluster_id` (String, required). @@ -1778,9 +1796,35 @@ let resp = qn println!("{} rows, {:?}", resp.rows, resp.data.first()); ``` +##### `query_with_session` + +Executes a SQL query on the x402 drawdown host with a SIWX `GatewaySession` JWT. Single attempt. A 402 `requires_payment` is `SdkError::Api` — this method never signs a per-request payment. Requires the `payments` feature. + +```rust +// Rust +let session = qn.rpc.gateway_authenticate().await?; +let resp = qn.sql.query_with_session(¶ms, &session).await?; +``` + +##### `query_with_mpp_session` + +Executes a SQL query on the MPP session route with a cumulative voucher. The increment is the SQL challenge `amount` (not `ChannelState.per_call`). A 402 insufficient-balance is terminal. + +Takes `&mut ChannelState` and advances `cumulative_spent` whenever the voucher reached the gateway — including on a non-2xx body and on a lost response. Persist the channel on every outcome, not only after a 200: the gateway refuses a re-signed stale cumulative, so a channel left behind the gateway cannot be used again. Requires the `payments-tempo` feature. + +```rust +// Rust +let result = qn + .sql + .query_with_mpp_session(¶ms, &payment, &mut channel) + .await; +save_channel(&channel)?; // persist first — the voucher may have settled +let result = result?; +``` + ##### `get_schema` -Fetches the database schema for a cluster: table names, columns, types, sort keys, and partition strategies. +Fetches the database schema for a cluster: table names, columns, types, sort keys, and partition strategies. Reads the configured SQL base URL and sends the API key. **Parameters**: `cluster_id` (`&str`, required). diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index e9f6182..478531c 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -26,9 +26,11 @@ pub use rpc::{ }; #[cfg(feature = "payments-tempo")] pub use rpc::{ChannelState, ChannelStatus}; +#[cfg(feature = "payments-tempo")] +pub use sql::MppQueryResult; pub use sql::{ ChainSchema, ColumnMeta, ColumnSchema, QueryParams, QueryResponse, QueryStatistics, - SqlApiClient, TableSchema, + SqlApiClient, SqlCluster, TableSchema, X402_SQL_BASE_URL, }; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; diff --git a/crates/core/src/rpc/payment/session.rs b/crates/core/src/rpc/payment/session.rs index b97ccb5..c43b034 100644 --- a/crates/core/src/rpc/payment/session.rs +++ b/crates/core/src/rpc/payment/session.rs @@ -375,6 +375,147 @@ pub async fn voucher_call( Ok(text) } +/// Session route for SQL Explorer. Distinct from [`SESSION_ROUTE_NETWORK`]: +/// the SQL 402 challenge prices a query at its own `amount`, not the RPC +/// `per_call` unit. +const SQL_SESSION_ROUTE: &str = "sql/rest/v1/query"; + +/// Makes one MPP-session SQL query. The voucher increment is the SQL +/// challenge `amount`, not [`ChannelState::per_call`]. Returns the response +/// body and `acceptedCumulative` from the receipt (or the signed cumulative +/// if the receipt omits it). A 402 insufficient-balance is terminal. +/// +/// Advances `channel.cumulative_spent` whenever the voucher reached the +/// gateway, errors included: a re-signed stale cumulative is refused, so +/// trailing the gateway strands the channel while leading it recovers. +pub async fn sql_voucher_call( + client: &reqwest::Client, + payment: &ResolvedPayment, + channel: &mut ChannelState, + body: &Value, +) -> Result<(String, u128), SdkError> { + let challenge = probe_sql_session_challenge(client, payment, body).await?; + let increment = require_amount(&challenge.request)?; + let new_cumulative = channel.cumulative_spent.saturating_add(increment); + if new_cumulative > channel.deposit { + return Err(SdkError::PaymentUnsupported { + offered: format!( + "voucher cumulative {new_cumulative} exceeds channel deposit {}; top up first", + channel.deposit + ), + }); + } + + let payer = payment.signer.address()?; + let signature = payment.signer.sign_session_voucher( + &channel.channel_id, + new_cumulative, + channel.chain_id, + &channel.escrow_contract, + )?; + let payload = serde_json::json!({ + "action": "voucher", + "channelId": channel.channel_id, + "cumulativeAmount": new_cumulative.to_string(), + "signature": signature, + }); + let credential = build_credential(&challenge, &payer, channel.chain_id, &payload)?; + + let url = session_base(payment, SQL_SESSION_ROUTE); + let paid = match client + .post(&url) + .header("Authorization", format!("Payment {credential}")) + .json(body) + .send() + .await + { + Ok(resp) => resp, + Err(e) => { + let err = SdkError::Http(e); + return Err(match err.http_kind() { + // A connect failure never put the voucher on the wire. + Some(HttpKind::Connect) => err, + // Otherwise it may have landed; assume it did. + _ => { + channel.cumulative_spent = new_cumulative; + SdkError::PaymentIndeterminate + } + }); + } + }; + + // Parse before `text()` consumes the response; a failed body can still bank + // the voucher. + let receipt_cumulative = paid + .headers() + .get("payment-receipt") + .and_then(|v| v.to_str().ok()) + .and_then(|h| super::decode_b64url_json(h).ok()) + .and_then(|v| { + v.get("acceptedCumulative") + .and_then(Value::as_str) + .map(str::to_string) + }) + .and_then(|s| s.parse::().ok()); + + let accepted = receipt_cumulative.unwrap_or(new_cumulative); + channel.cumulative_spent = accepted; + + let paid_status = paid.status(); + let text = paid.text().await.map_err(SdkError::Http)?; + if !paid_status.is_success() { + return Err(SdkError::Api { + status: paid_status, + body: text, + }); + } + Ok((text, accepted)) +} + +// Probe the SQL session route for the 402 challenge. The SQL amount is +// not the RPC lifecycle amount, so this must not reuse the pinned +// [`SESSION_ROUTE_NETWORK`] probe. +async fn probe_sql_session_challenge( + client: &reqwest::Client, + payment: &ResolvedPayment, + body: &Value, +) -> Result { + let url = session_base(payment, SQL_SESSION_ROUTE); + let resp = client + .post(&url) + .json(body) + .send() + .await + .map_err(SdkError::Http)?; + if resp.status().as_u16() == 404 { + return Err(SdkError::PaymentUnsupported { + offered: "the gateway does not serve the SQL session route \ + (/session/sql/rest/v1/query returned 404)" + .into(), + }); + } + if resp.status().as_u16() != 402 { + return Err(SdkError::PaymentUnsupported { + offered: format!( + "the SQL session endpoint did not return a 402 challenge (status {})", + resp.status().as_u16() + ), + }); + } + let header = resp + .headers() + .get("www-authenticate") + .and_then(|v| v.to_str().ok()) + .map(String::from) + .ok_or_else(|| SdkError::PaymentUnsupported { + offered: "SQL session 402 without a WWW-Authenticate header".into(), + })?; + parse_session_challenge( + &header, + super::caip2_or_bare_chain_id(&payment.pay_network)?, + ) +} + // ── HTTP + credential helpers ──────────────────────────────────────────────── fn session_base(payment: &ResolvedPayment, query_network: &str) -> String { diff --git a/crates/core/src/sql/mod.rs b/crates/core/src/sql/mod.rs index 56fe690..acefeb9 100644 --- a/crates/core/src/sql/mod.rs +++ b/crates/core/src/sql/mod.rs @@ -11,6 +11,9 @@ use serde::{Deserialize, Serialize}; use crate::{config::SqlConfig, errors::SdkError, SdkConfig}; const SQL_BASE_URL: &str = "https://api.quicknode.com/sql/rest/v1/"; +/// Public catalog + x402 drawdown SQL host. Callers that want this prefix set +/// it on [`SqlConfig::base_url`]; the account host above is the default. +pub const X402_SQL_BASE_URL: &str = "https://x402.quicknode.com/sql/rest/v1/"; // ── Resolved config ──────────────────────────────────────────────────────── @@ -32,6 +35,40 @@ impl ResolvedSqlConfig { } } +// ── Catalog types ────────────────────────────────────────────────────────── + +/// One cluster from `GET /sql/rest/v1/clusters`. +#[cfg_attr(feature = "python", gen_stub_pyclass)] +#[cfg_attr(feature = "python", pyclass(get_all, set_all))] +#[cfg_attr(feature = "node", napi(object))] +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct SqlCluster { + /// Cluster identifier (e.g. `"hyperliquid-core-mainnet"`). + pub id: String, + /// Human-readable name (e.g. `"Hyperliquid (HyperCore)"`). + pub display_name: String, +} + +#[cfg(feature = "python")] +#[gen_stub_pymethods] +#[pymethods] +impl SqlCluster { + #[new] + pub fn new(id: String, display_name: String) -> Self { + Self { id, display_name } + } +} + +/// A successful MPP-session SQL query plus the receipt's accepted cumulative +/// spend. The caller persists `accepted_cumulative` on the channel; do not +/// advance the channel by `query.credits` (SQL credits ≠ voucher increment). +#[cfg(feature = "payments-tempo")] +#[derive(Debug, Clone)] +pub struct MppQueryResult { + pub query: QueryResponse, + pub accepted_cumulative: u128, +} + // ── Request types ────────────────────────────────────────────────────────── /// Parameters for `query`. @@ -264,6 +301,91 @@ impl SqlApiClient { serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body }) } + /// Lists clusters from the public SQL catalog (`GET clusters`). Always + /// unauthenticated: uses the keyless client so an `x-api-key` is never + /// sent, even when the SDK was built with one. + pub async fn list_clusters(&self) -> Result, SdkError> { + let url = self.config.sql().base_url.join("clusters")?; + let resp = self + .config + .rpc_http_client() + .get(url) + .send() + .await + .map_err(SdkError::Http)?; + let status = resp.status(); + let body = resp.text().await.map_err(SdkError::Http)?; + if !status.is_success() { + return Err(SdkError::Api { status, body }); + } + serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body }) + } + + /// Executes a SQL query against the x402 drawdown host with a SIWX session + /// JWT (`Authorization: Bearer`). Single attempt; 401/403 stay + /// [`SdkError::Api`] so the caller can re-auth on `token_expired`. A 402 + /// `requires_payment` is also [`SdkError::Api`] — this lane never signs a + /// per-request payment. + #[cfg(feature = "payments")] + pub async fn query_with_session( + &self, + params: &QueryParams, + session: &crate::rpc::payment::drawdown::GatewaySession, + ) -> Result { + let url = self.config.sql().base_url.join("query")?; + let resp = self + .config + .rpc_http_client() + .post(url) + .bearer_auth(&session.token) + .json(params) + .send() + .await + .map_err(SdkError::Http)?; + let status = resp.status(); + let body = resp.text().await.map_err(SdkError::Http)?; + if !status.is_success() { + return Err(SdkError::Api { status, body }); + } + serde_json::from_str(&body).map_err(|source| SdkError::Decode { source, body }) + } + + /// Executes a SQL query on the MPP session route + /// (`POST {mpp}/session/sql/rest/v1/query`) with a cumulative voucher. + /// The increment is the SQL challenge `amount`, not + /// [`crate::rpc::payment::session::ChannelState::per_call`]. A 402 + /// insufficient-balance is terminal — this method does not sign a smaller + /// increment. + /// + /// Advances `channel.cumulative_spent` whenever the voucher reached the + /// gateway, errors included. Persist `channel` on every outcome, not only + /// after a 200, or a later query re-signs a cumulative the gateway refuses. + #[cfg(feature = "payments-tempo")] + pub async fn query_with_mpp_session( + &self, + params: &QueryParams, + payment: &crate::config::PaymentConfig, + channel: &mut crate::rpc::payment::session::ChannelState, + ) -> Result { + let resolved = crate::rpc::payment::ResolvedPayment::from_config(payment)?; + let body = serde_json::to_value(params).map_err(|e| { + SdkError::Config(format!("could not serialize the SQL query body: {e}")) + })?; + let (text, accepted_cumulative) = crate::rpc::payment::session::sql_voucher_call( + self.config.rpc_http_client(), + &resolved, + channel, + &body, + ) + .await?; + let query = serde_json::from_str(&text) + .map_err(|source| SdkError::Decode { source, body: text })?; + Ok(MppQueryResult { + query, + accepted_cumulative, + }) + } + /// Fetches the database schema for a cluster, including table names, /// columns, types, sort keys, and partition strategies. pub async fn get_schema(&self, cluster_id: &str) -> Result { @@ -464,4 +586,360 @@ mod tests { let err = sdk.sql.get_schema("bad-cluster").await.unwrap_err(); assert!(matches!(err, SdkError::Api { .. })); } + + // ── list_clusters ──────────────────────────────────────────────────────── + + #[tokio::test] + async fn list_clusters_decodes_id_and_display_name() { + let server = MockServer::start().await; + Mock::given(method("GET")) + .and(path("/clusters")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([ + {"id": "hyperliquid-core-mainnet", "display_name": "Hyperliquid (HyperCore)"}, + {"id": "solana-mainnet", "display_name": "Solana"} + ]))) + .mount(&server) + .await; + let sdk = make_sdk(format!("{}/", server.uri())); + let clusters = sdk.sql.list_clusters().await.unwrap(); + assert_eq!(clusters.len(), 2); + assert_eq!(clusters[0].id, "hyperliquid-core-mainnet"); + assert_eq!(clusters[0].display_name, "Hyperliquid (HyperCore)"); + assert_eq!(clusters[1].id, "solana-mainnet"); + assert_eq!(clusters[1].display_name, "Solana"); + } + + // ── query_with_session ─────────────────────────────────────────────────── + + #[cfg(feature = "payments")] + fn session() -> crate::rpc::payment::drawdown::GatewaySession { + crate::rpc::payment::drawdown::GatewaySession { + token: "jwt-abc".into(), + exp_unix: 4_102_444_800, + account_id: "a".into(), + } + } + + #[cfg(feature = "payments")] + fn ok_query_body() -> serde_json::Value { + serde_json::json!({ + "meta": [{"name": "1", "type": "UInt8"}], + "data": [{"1": 1}], + "rows": 1, + "rows_before_limit_at_least": 1, + "statistics": {"elapsed": 0.001, "rows_read": 1, "bytes_read": 1}, + "credits": 117 + }) + } + + #[cfg(feature = "payments")] + #[tokio::test] + async fn query_with_session_attaches_bearer_and_no_api_key() { + use wiremock::matchers::header; + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/query")) + .and(header("authorization", "Bearer jwt-abc")) + .and(|req: &wiremock::Request| !req.headers.contains_key("x-api-key")) + .respond_with(ResponseTemplate::new(200).set_body_json(ok_query_body())) + .expect(1) + .mount(&server) + .await; + let sdk = make_sdk(format!("{}/", server.uri())); + let resp = sdk + .sql + .query_with_session(&query_params(), &session()) + .await + .unwrap(); + assert_eq!(resp.credits, 117); + assert_eq!(resp.rows, 1); + } + + #[cfg(feature = "payments")] + #[tokio::test] + async fn query_with_session_402_requires_payment_does_not_sign() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/query")) + .respond_with(ResponseTemplate::new(402).set_body_json(serde_json::json!({ + "error": "requires_payment", + "message": "SIWX drawdown required" + }))) + .expect(1) + .mount(&server) + .await; + let sdk = make_sdk(format!("{}/", server.uri())); + let err = sdk + .sql + .query_with_session(&query_params(), &session()) + .await + .unwrap_err(); + assert!( + matches!(&err, SdkError::Api { status, body } + if status.as_u16() == 402 && body.contains("requires_payment")), + "unexpected error: {err:?}" + ); + } + + // ── query_with_mpp_session ─────────────────────────────────────────────── + + #[cfg(feature = "payments-tempo")] + fn tempo_payment(base: &str) -> crate::config::PaymentConfig { + crate::config::PaymentConfig { + scheme: "mpp".into(), + key: "ac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80".into(), + pay_network: "eip155:42431".into(), + asset: "0x20c0000000000000000000000000000000000000".into(), + max_amount: "1000000".into(), + svm_rpc_url: None, + base_url_override: Some(base.to_string()), + } + } + + #[cfg(feature = "payments-tempo")] + fn sample_channel() -> crate::rpc::payment::session::ChannelState { + crate::rpc::payment::session::ChannelState { + channel_id: format!("0x{}", "11".repeat(32)), + token: "0x20c0000000000000000000000000000000000000".into(), + payee: "0xfd24114c3981aba78ae2441991b1bdb89329c556".into(), + salt: format!("0x{}", "22".repeat(32)), + authorized_signer: "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266".into(), + escrow_contract: "0x33b901018174DDabE4841042ab76ba85D4e24f25".into(), + deposit: 1_000_000, + cumulative_spent: 10, + per_call: 10, + chain_id: 42431, + } + } + + #[cfg(feature = "payments-tempo")] + fn sql_session_offer(amount: &str) -> String { + let request = { + use base64::Engine; + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode( + serde_json::to_vec(&serde_json::json!({ + "amount": amount, + "currency": "0x20c0000000000000000000000000000000000000", + "recipient": "0xfd24114c3981aba78ae2441991b1bdb89329c556", + "methodDetails": { + "chainId": 42431, + "escrowContract": "0x33b901018174DDabE4841042ab76ba85D4e24f25" + } + })) + .unwrap(), + ) + }; + format!( + "Payment id=\"sql1\", realm=\"mpp.quicknode.com\", method=\"tempo\", \ + intent=\"session\", description=\"d\", expires=\"2099-01-01T00:00:00Z\", \ + request=\"{request}\"" + ) + } + + #[cfg(feature = "payments-tempo")] + fn receipt_header(accepted: &str) -> String { + use base64::Engine; + base64::engine::general_purpose::URL_SAFE_NO_PAD.encode( + serde_json::to_vec(&serde_json::json!({ + "acceptedCumulative": accepted, + "spent": accepted, + "status": "success", + "intent": "session", + "method": "tempo", + })) + .unwrap(), + ) + } + + #[cfg(feature = "payments-tempo")] + #[tokio::test] + async fn query_with_mpp_session_uses_challenge_amount_not_per_call() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use wiremock::{Request, Respond}; + + struct SqlSeq { + calls: AtomicUsize, + } + impl Respond for SqlSeq { + fn respond(&self, req: &Request) -> ResponseTemplate { + let n = self.calls.fetch_add(1, Ordering::SeqCst); + if n == 0 && !req.headers.contains_key("authorization") { + return ResponseTemplate::new(402) + .insert_header("www-authenticate", sql_session_offer("100")); + } + // Decode the voucher increment. increment 10 (per_call) is + // insufficient; increment 100 (challenge amount) is accepted. + let auth = req + .headers + .get("authorization") + .and_then(|v| v.to_str().ok()) + .unwrap_or(""); + let b64 = auth.strip_prefix("Payment ").unwrap_or(auth); + let cred: serde_json::Value = { + use base64::Engine; + let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(b64.trim_end_matches('=')) + .unwrap(); + serde_json::from_slice(&bytes).unwrap() + }; + let cumulative = cred["payload"]["cumulativeAmount"] + .as_str() + .unwrap() + .parse::() + .unwrap(); + // channel.cumulative_spent is 10; per_call would yield 20. + if cumulative == 20 { + return ResponseTemplate::new(402).set_body_json(serde_json::json!({ + "title": "Insufficient Balance", + "detail": "Insufficient balance: requested 100, available 10." + })); + } + assert_eq!(cumulative, 110, "expected challenge increment 100"); + ResponseTemplate::new(200) + .insert_header("payment-receipt", receipt_header("110").as_str()) + .set_body_json(ok_query_body()) + } + } + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/session/sql/rest/v1/query")) + .respond_with(SqlSeq { + calls: AtomicUsize::new(0), + }) + .expect(2) + .mount(&server) + .await; + + let sdk = make_sdk(format!("{}/", server.uri())); + let mut channel = sample_channel(); + let result = sdk + .sql + .query_with_mpp_session(&query_params(), &tempo_payment(&server.uri()), &mut channel) + .await + .unwrap(); + assert_eq!(result.query.credits, 117); + assert_eq!(result.accepted_cumulative, 110); + assert_eq!(channel.cumulative_spent, 110); + } + + #[cfg(feature = "payments-tempo")] + #[tokio::test] + async fn query_with_mpp_session_insufficient_balance_does_not_resign() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use wiremock::{Request, Respond}; + + struct OnceThenRefuse { + calls: AtomicUsize, + } + impl Respond for OnceThenRefuse { + fn respond(&self, req: &Request) -> ResponseTemplate { + let n = self.calls.fetch_add(1, Ordering::SeqCst); + if n == 0 && !req.headers.contains_key("authorization") { + return ResponseTemplate::new(402) + .insert_header("www-authenticate", sql_session_offer("100")); + } + ResponseTemplate::new(402).set_body_json(serde_json::json!({ + "title": "Insufficient Balance", + "detail": "Insufficient balance: requested 100, available 10." + })) + } + } + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/session/sql/rest/v1/query")) + .respond_with(OnceThenRefuse { + calls: AtomicUsize::new(0), + }) + .expect(2) + .mount(&server) + .await; + + let sdk = make_sdk(format!("{}/", server.uri())); + let mut channel = sample_channel(); + let err = sdk + .sql + .query_with_mpp_session(&query_params(), &tempo_payment(&server.uri()), &mut channel) + .await + .unwrap_err(); + assert!( + matches!(&err, SdkError::Api { status, body } + if status.as_u16() == 402 && body.contains("Insufficient")), + "unexpected error: {err:?}" + ); + // Advances despite the failure: no receipt, so the signed value stands. + assert_eq!(channel.cumulative_spent, 110); + } + + #[cfg(feature = "payments-tempo")] + #[tokio::test] + async fn query_with_mpp_session_advances_channel_from_receipt_on_error() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use wiremock::{Request, Respond}; + + struct ChallengeThenServerError { + calls: AtomicUsize, + } + impl Respond for ChallengeThenServerError { + fn respond(&self, req: &Request) -> ResponseTemplate { + let n = self.calls.fetch_add(1, Ordering::SeqCst); + if n == 0 && !req.headers.contains_key("authorization") { + return ResponseTemplate::new(402) + .insert_header("www-authenticate", sql_session_offer("100")); + } + // Voucher banked, query body failed. + ResponseTemplate::new(500) + .insert_header("payment-receipt", receipt_header("110").as_str()) + .set_body_string("query engine unavailable") + } + } + + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/session/sql/rest/v1/query")) + .respond_with(ChallengeThenServerError { + calls: AtomicUsize::new(0), + }) + .expect(2) + .mount(&server) + .await; + + let sdk = make_sdk(format!("{}/", server.uri())); + let mut channel = sample_channel(); + let err = sdk + .sql + .query_with_mpp_session(&query_params(), &tempo_payment(&server.uri()), &mut channel) + .await + .unwrap_err(); + assert!( + matches!(&err, SdkError::Api { status, .. } if status.as_u16() == 500), + "unexpected error: {err:?}" + ); + assert_eq!(channel.cumulative_spent, 110); + } + + #[cfg(feature = "payments-tempo")] + #[tokio::test] + async fn query_with_mpp_session_leaves_channel_alone_when_never_sent() { + // Nothing listening: the probe cannot connect, so no voucher is signed. + let sdk = make_sdk("http://127.0.0.1:1/".to_string()); + let mut channel = sample_channel(); + let before = channel.cumulative_spent; + let err = sdk + .sql + .query_with_mpp_session( + &query_params(), + &tempo_payment("http://127.0.0.1:1"), + &mut channel, + ) + .await + .unwrap_err(); + assert!( + !matches!(err, SdkError::PaymentIndeterminate), + "a failed probe must not read as an indeterminate payment: {err:?}" + ); + assert_eq!(channel.cumulative_spent, before); + } }