From 901b157a29aa4bf0cc72810410b3e2bfee716e73 Mon Sep 17 00:00:00 2001 From: Stephen Chen <20940639+stephenctw@users.noreply.github.com> Date: Sun, 13 Sep 2026 23:27:21 +0800 Subject: [PATCH 1/2] feat(api): add GET /fee for frontend max_fee quotes Wallets need a number they can sign before POST /tx. Quote the open-frame fee and recommended_fee, plus suggested_max_fee (max of those with 1.5x slack) so inclusion can survive a frame rotation. --- AGENTS.md | 6 +- CLAUDE.md | 2 +- README.md | 22 +++- docs/threat-model/README.md | 2 +- sdk/rust-client/src/errors.rs | 10 ++ sdk/rust-client/src/lib.rs | 23 +++- sequencer-core/src/api.rs | 24 +++++ sequencer-core/src/fee.rs | 46 ++++++++ sequencer/src/commands/run/workers.rs | 2 +- sequencer/src/http.rs | 10 +- sequencer/src/ingress/api.rs | 102 ++++++++++++++++-- sequencer/src/ingress/mod.rs | 6 +- .../src/integration_tests/e2e_sequencer.rs | 31 ++++++ .../integration_tests/snapshot_endpoints.rs | 62 ++++++++++- sequencer/src/storage/ingress.rs | 11 ++ 15 files changed, 333 insertions(+), 26 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 0896353f..78c6c5db 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -158,8 +158,8 @@ Top-level layout follows the system's data flow. Each sequencer module correspon - `sequencer/src/harness.rs` — CLI harness: the `setup`/`run`/`flush-mempool` subcommand parser, `dispatch`, and the exit-code projection. An app's `main` is ~5 lines (`run_main` + a genesis-app closure). - `sequencer/src/http.rs` — shared HTTP error type, JSON `ErrorResponse`, `ApiConfig`, and `axum::serve` orchestration. - `sequencer/src/commands/` — the operator command brackets: `setup` (phase A — pin identity, initial sync, genesis snapshot, atomic `setup_complete` fact), `run` (phase B — recover, prepare, admit, and boot workers; its `workers` supervisor lives beside it), and `flush` (`flush-mempool`). `sequencer/src/commands/` also owns the command-scoped `config` and `error` taxonomy (incl. the exit-code projection); `sequencer/src/runtime/` is exactly the runtime authority capabilities — the process lock and `shutdown` (runtime scope and graceful notification) — consumed crate-wide. `L1Config` lives in `sequencer/src/l1/`; the crate-wide wall clock is `sequencer/src/clock.rs`. -- `sequencer/src/ingress/` — public write path. - - `api.rs` — `POST /tx` handler, JSON-rejection mapping. +- `sequencer/src/ingress/` — public-facing HTTP + inclusion lane. + - `api.rs` — `POST /tx` and `GET /fee` handlers, JSON-rejection mapping. - `inclusion_lane/` — single-lane hot-path loop (`mod.rs`), catch-up replay, config, error types. - `sequencer/src/egress/` — internal read path. - `api/` — `/ws/subscribe`, `/livez`, `/readyz`, `/healthz`. @@ -300,7 +300,7 @@ restate them here. ## HTTP Endpoints -- **Ingress** (public-facing): `POST /tx`. +- **Ingress** (public-facing): `POST /tx`, `GET /fee`. - **Egress** (internal indexers/watchdog): `GET /ws/subscribe`, `GET /finalized_state`, `GET /finalized_state/inclusion_block`, `GET /latest_snapshot`, `GET /livez`, `GET /readyz`, `GET /healthz`. The snapshot/state endpoints are **operator-only** (no auth) and must not be exposed publicly; the streaming routes hold a GC lease for the response lifetime ([`docs/snapshots/lifecycle.md`](docs/snapshots/lifecycle.md)). Today both sides serve from one listener; the planned API split puts each side on its own port (same binary) so internal probes and subscribers can be firewalled from public submit traffic. diff --git a/CLAUDE.md b/CLAUDE.md index df9e35a0..56824121 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -48,7 +48,7 @@ Rust edition 2024 / Axum API / SQLite (rusqlite, WAL) / EIP-712 signing / SSZ en `error` taxonomy (incl. exit-code projection). - `runtime/` — the runtime authority capabilities, consumed crate-wide: the exclusive process lock and the runtime scope/shutdown machinery. -- `ingress/` — public write path: `api.rs` (`POST /tx`) + `inclusion_lane/` (hot path). +- `ingress/` — public-facing: `api.rs` (`POST /tx`, `GET /fee`) + `inclusion_lane/` (hot path). - `egress/` — internal read path: `api/` (WS subscribe + health) + `l2_tx_feed/`. - `l1/` — reader, submitter, fee oracle, provider, partition helper. - `recovery/` — startup preemptive-recovery procedure, runtime danger detector, mempool flusher. diff --git a/README.md b/README.md index 27ae2c09..a571cc1d 100644 --- a/README.md +++ b/README.md @@ -77,7 +77,7 @@ The sequencer is designed to handle: ### User Operations -Users submit signed operations via `POST /tx` (JSON). Operations are signed with EIP-712 using the rollup's chain ID and app address. The sequencer validates the signature, executes the operation against the current app state, and returns a soft confirmation. +Users submit signed operations via `POST /tx` (JSON). Operations are signed with EIP-712 using the rollup's chain ID and app address. The sequencer validates the signature, executes the operation against the current app state, and returns a soft confirmation. `GET /fee` quotes the live frame fee, the next-frame recommendation, and a suggested `max_fee` a wallet can sign. ### Sequenced Transaction Feed @@ -166,7 +166,23 @@ Notes: - payload size is bounded at ingress; oversized requests are rejected before entering the hot path. - overload is enforced at queue admission: if the inclusion-lane queue is full, `POST /tx` returns HTTP `429` with code `OVERLOADED` and message `queue full`. - queue capacity is an internal runtime constant tuned alongside inclusion-lane chunking to absorb short bursts; if this starts triggering persistently, it is a signal to revisit runtime sizing or throughput rather than add another admission layer. -- Browser wallets can call `POST /tx` from any origin with any request headers; preflight permits POST and is cached for one hour. CORS is applied only to ingress. Egress routes remain operator-only and require network access controls. +- Browser wallets can call `POST /tx` and `GET /fee` from any origin with any request headers; preflight permits GET and POST and is cached for one hour. CORS is applied only to ingress. Egress routes remain operator-only and require network access controls. + +### `GET /fee` + +Fee quote for setting signed user-op `max_fee` before `POST /tx`. All three fields are log-space exponents (base 129/128), the same encoding as `max_fee`. Inclusion rejects any op with `max_fee` below the open-frame `fee`. + +```json +{ "fee": 1356, "recommended_fee": 1356, "suggested_max_fee": 1409 } +``` + +Notes: + +- `fee` is frozen for the lifetime of the open frame (the live inclusion check). +- `recommended_fee` is what the next frame will sample at rotation (currently after five newly-safe L1 blocks, best-effort). +- `suggested_max_fee` is `max(fee, recommended_fee)` plus 1.5× log-space slack. Wallets can copy this into signed `max_fee`; the user pays the frame fee, not this cap. Clients that want their own policy can ignore it and combine the two facts themselves. +- `200` while an open frame exists (the admitted runtime always has one). +- `503` with code `UNAVAILABLE` during shutdown, or if no open frame exists. ### `GET /ws/subscribe?from_offset=` @@ -237,7 +253,7 @@ released even on client disconnect. - `examples/wallet-sequencer/`: binary crate composing the sequencer library with the placeholder wallet app - `sequencer/src/http.rs`: shared HTTP error type, JSON error shape, and `axum::serve` orchestration - `sequencer/src/runtime/`: process lock and shutdown scope; command bootstrap and config live in `commands/`, the shared clock in `clock.rs`, and EIP-712 domain construction in `sequencer-core/` -- `sequencer/src/ingress/`: public write path — `POST /tx` (`api.rs`) and the inclusion lane (`inclusion_lane/`: hot-path loop, chunk/frame/batch rotation, catch-up, snapshot lifecycle) +- `sequencer/src/ingress/`: public-facing — `POST /tx` and `GET /fee` (`api.rs`) and the inclusion lane (`inclusion_lane/`: hot-path loop, chunk/frame/batch rotation, catch-up, snapshot lifecycle) - `sequencer/src/egress/`: internal read path — WS subscribe + health probes (`api/`) and the DB-backed ordered-L2Tx feed (`l2_tx_feed/`) - `sequencer/src/l1/`: L1 client surface — input reader, batch submitter, fee oracle, shared EIP-1559 estimation, provider, partition helper - `sequencer/src/recovery/`: preemptive recovery startup, runtime danger detector, mempool flusher diff --git a/docs/threat-model/README.md b/docs/threat-model/README.md index 50866a81..077b0912 100644 --- a/docs/threat-model/README.md +++ b/docs/threat-model/README.md @@ -25,7 +25,7 @@ What we are protecting: | Batch-submitter private key | Private | Held in operator infra. Not reachable by the network. | | Sequencer's own code | Trusted (bug-free is a precondition) | Bugs are prevented through tests/review and contained by fail-loud runtime invariant checks; they are not treated as adversarial behavior that the protocol can recover around. See "self-trust" below. | | **L1 mempool and block builders** | **Fully adversarial** | May reorder, delay, drop, or selectively include submitted transactions. Private mempools mean "dropped" is indistinguishable from "delayed indefinitely." | -| HTTP clients at `POST /tx` | Untrusted | Arbitrary public callers. May submit malformed, malicious, or replay payloads. | +| HTTP clients at `POST /tx` and `GET /fee` | Untrusted | Arbitrary public callers. May submit malformed, malicious, or replay payloads. `GET /fee` is an intentional public quote of the open-frame fee. | | WebSocket subscribers at `/ws/subscribe` | Internal, but untrusted for data-exposure | Intended for internal indexers. Treat as public for what is exposed. | | Direct-input senders on L1 | Untrusted | Arbitrary L1 accounts calling InputBox. May submit any calldata. | diff --git a/sdk/rust-client/src/errors.rs b/sdk/rust-client/src/errors.rs index 6ad68121..69c3ff08 100644 --- a/sdk/rust-client/src/errors.rs +++ b/sdk/rust-client/src/errors.rs @@ -57,6 +57,16 @@ pub enum SubmitRejected { Decode(String), } +#[derive(Debug, Error)] +pub enum GetFeeError { + #[error("fee request failed: {0}")] + Transport(#[from] SubmitTxError), + #[error("/fee rejected with status {status}: {body}")] + Http { status: u16, body: String }, + #[error("invalid /fee success body: {0}")] + Decode(String), +} + #[derive(Debug, Error)] pub enum SubscribeError { #[error("invalid endpoint: {0}")] diff --git a/sdk/rust-client/src/lib.rs b/sdk/rust-client/src/lib.rs index a35e1406..734a884e 100644 --- a/sdk/rust-client/src/lib.rs +++ b/sdk/rust-client/src/lib.rs @@ -3,9 +3,9 @@ mod errors; -pub use errors::{ClientBuildError, SubmitRejected, SubmitTxError, SubscribeError}; +pub use errors::{ClientBuildError, GetFeeError, SubmitRejected, SubmitTxError, SubscribeError}; -use sequencer_core::api::{TxRequest, TxResponse}; +use sequencer_core::api::{FeeResponse, TxRequest, TxResponse}; use std::time::Duration; use tokio::net::TcpStream; use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async}; @@ -102,6 +102,25 @@ impl SequencerClient { serde_json::from_str::(&body).map_err(|e| SubmitRejected::Decode(e.to_string())) } + pub async fn get_fee(&self) -> Result { + let url = format!("{}/fee", self.endpoint.trim_end_matches('/')); + let response = self + .http_client + .get(&url) + .send() + .await + .map_err(map_reqwest_error)?; + let status = response.status().as_u16(); + let body = response + .text() + .await + .map_err(|e| SubmitTxError::IoRead(e.to_string()))?; + if status != 200 { + return Err(GetFeeError::Http { status, body }); + } + serde_json::from_str::(&body).map_err(|e| GetFeeError::Decode(e.to_string())) + } + pub async fn subscribe(&self, from_offset: u64) -> Result { let url = self.ws_subscribe_url(from_offset); let (stream, _response) = connect_async(url.as_str()) diff --git a/sequencer-core/src/api.rs b/sequencer-core/src/api.rs index b7a2342b..7650648d 100644 --- a/sequencer-core/src/api.rs +++ b/sequencer-core/src/api.rs @@ -118,6 +118,30 @@ pub struct TxResponse { pub nonce: u32, } +/// Fee quote for wallets that need to set signed `max_fee` before `POST /tx`. +/// +/// `fee` is frozen on the open frame (the live inclusion check). +/// `recommended_fee` is what the next frame will sample. +/// `suggested_max_fee` is `max(fee, recommended_fee)` plus 1.5× log-space +/// slack — the value a wallet can copy into `max_fee` so the signature +/// survives a frame rotation. The user pays the frame fee, not this cap. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct FeeResponse { + pub fee: u16, + pub recommended_fee: u16, + pub suggested_max_fee: u16, +} + +impl FeeResponse { + pub fn quote(fee: u16, recommended_fee: u16) -> Self { + Self { + fee, + recommended_fee, + suggested_max_fee: crate::fee::suggested_signing_max_fee(fee, recommended_fee), + } + } +} + pub type WsTxMessage = BroadcastTxMessage; fn decode_hex_0x(value: &str) -> Result, String> { diff --git a/sequencer-core/src/fee.rs b/sequencer-core/src/fee.rs index ca3f70d8..81234fd9 100644 --- a/sequencer-core/src/fee.rs +++ b/sequencer-core/src/fee.rs @@ -218,6 +218,26 @@ pub fn log_fee_ratio(num: u64, denom: u64) -> i32 { } } +/// Smallest log-space exponent `n` such that `(129/128)^n >= 3/2`. +/// +/// Signing slack for [`suggested_signing_max_fee`]: wallets may over-estimate +/// `max_fee` because the user pays the frame fee, not the signed cap. 1.5× +/// sits on top of `max(open-frame fee, recommended_fee)` so a quote can +/// survive a frame rotation. Distinct from `batch_policy.log_slack` (the 10× +/// DA margin already baked into both inputs). +pub const SUGGESTED_MAX_FEE_SLACK: u16 = 53; + +/// Suggested user-op `max_fee`: `max(frame_fee, recommended_fee)` plus +/// [`SUGGESTED_MAX_FEE_SLACK`] (1.5× in log space), clamped to +/// [`MAX_EXPONENT`]. +pub fn suggested_signing_max_fee(frame_fee: u16, recommended_fee: u16) -> u16 { + let base = frame_fee.max(recommended_fee); + match base.checked_add(SUGGESTED_MAX_FEE_SLACK) { + Some(sum) if sum <= MAX_EXPONENT => sum, + _ => MAX_EXPONENT, + } +} + /// Fixed-point multiplication: `(a * b) >> FRAC_BITS`. /// /// Uses a 512-bit intermediate to avoid overflow. @@ -313,6 +333,32 @@ mod tests { assert_eq!(log_fee_ratio(10, 1), 296); } + #[test] + fn suggested_max_fee_slack_is_ceil_three_halves() { + assert_eq!(SUGGESTED_MAX_FEE_SLACK, 53); + // Nearest-rounding log(1.5) undercharges; the signing slack is ceil. + assert!(log_fee_ratio(3, 2) < i32::from(SUGGESTED_MAX_FEE_SLACK)); + let base = 1000_u16; + let suggested = suggested_signing_max_fee(base, 0); + assert_eq!(suggested, base + SUGGESTED_MAX_FEE_SLACK); + assert!( + fee_to_linear(suggested) * U256::from(2u64) >= fee_to_linear(base) * U256::from(3u64) + ); + } + + #[test] + fn suggested_signing_max_fee_uses_the_higher_input_and_clamps() { + assert_eq!( + suggested_signing_max_fee(100, 200), + 200 + SUGGESTED_MAX_FEE_SLACK + ); + assert_eq!(suggested_signing_max_fee(MAX_EXPONENT, 0), MAX_EXPONENT); + assert_eq!( + suggested_signing_max_fee(MAX_EXPONENT - 1, MAX_EXPONENT - 10), + MAX_EXPONENT + ); + } + #[test] fn fee_from_linear_zero_and_one() { assert_eq!(fee_from_linear(U256::ZERO), 0); diff --git a/sequencer/src/commands/run/workers.rs b/sequencer/src/commands/run/workers.rs index b42f3856..c73f9d32 100644 --- a/sequencer/src/commands/run/workers.rs +++ b/sequencer/src/commands/run/workers.rs @@ -299,7 +299,7 @@ impl PreparedRuntime { let submitter = submitter.start_preflighted(shutdown.clone()); let detector = detector.start_preflighted(shutdown.signal()); let fee_oracle = fee_oracle.map(|oracle| oracle.start(shutdown.signal())); - // HTTP server (ingress /tx + egress /ws/subscribe + /health, currently merged). + // HTTP server (ingress /tx + /fee + egress /ws/subscribe + /health, currently merged). let server = http::start_on_listener( listener, tx, diff --git a/sequencer/src/http.rs b/sequencer/src/http.rs index 9c8dffd4..b654eb4b 100644 --- a/sequencer/src/http.rs +++ b/sequencer/src/http.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 (see LICENSE) //! Shared HTTP surface: error type + JSON response shape used by both -//! ingress (`/tx`) and egress (`/ws/subscribe`, future routes), plus the +//! ingress (`/tx`, `/fee`) and egress (`/ws/subscribe`, future routes), plus the //! `axum::serve` orchestration that wires the two side routers together. //! //! Today both sides serve from one listener; the planned api split puts each @@ -26,7 +26,7 @@ use tower_http::trace::TraceLayer; pub use crate::egress::api::SnapshotState; use crate::egress::api::SubscribeState; use crate::egress::l2_tx_feed::L2TxFeed; -use crate::ingress::api::SubmitState; +use crate::ingress::api::{FeeState, SubmitState}; use crate::ingress::inclusion_lane::{PendingUserOp, SequencerError}; use crate::runtime::shutdown::{RuntimeScope, abort_terminal}; use crate::storage::ReleaseScheduler; @@ -266,13 +266,17 @@ pub(crate) fn start_on_listener( config.max_user_op_data_bytes, shutdown.clone(), )); + let fee_state = Arc::new(FeeState::new( + snapshot_state.db_path.clone(), + shutdown.clone(), + )); let subscribe_state = Arc::new(SubscribeState::new( shutdown.clone(), tx_feed, config.ws_max_subscribers, config.ws_max_catchup_events, )); - let app: Router = crate::ingress::api::router(submit_state) + let app: Router = crate::ingress::api::router(submit_state, fee_state) .merge(crate::egress::api::router( subscribe_state, health_state, diff --git a/sequencer/src/ingress/api.rs b/sequencer/src/ingress/api.rs index 35be161e..204e1b36 100644 --- a/sequencer/src/ingress/api.rs +++ b/sequencer/src/ingress/api.rs @@ -1,9 +1,13 @@ // (c) Cartesi and individual authors (see AUTHORS) // SPDX-License-Identifier: Apache-2.0 (see LICENSE) -//! `POST /tx` — validate a signed user op, enqueue it for the inclusion lane, -//! and wait for the lane's commit ack before responding. Synchronous from the -//! client's perspective: 200 means included. +//! Public ingress HTTP: +//! +//! - `POST /tx` — validate a signed user op, enqueue it for the inclusion +//! lane, and wait for the lane's commit ack. Synchronous from the client's +//! perspective: 200 means included. +//! - `GET /fee` — quote the open-frame fee, recommended fee, and a suggested +//! `max_fee` so a wallet can sign before submitting. use std::sync::Arc; use std::time::{Duration, SystemTime}; @@ -13,7 +17,7 @@ use axum::Router; use axum::extract::{Json, State}; use axum::http::{Method, StatusCode}; use axum::response::{IntoResponse, Response}; -use axum::routing::post; +use axum::routing::{get, post}; use tokio::sync::mpsc::{self, error::TrySendError}; use tokio::sync::oneshot; use tower_http::cors::{Any, CorsLayer}; @@ -22,7 +26,8 @@ use tracing::debug; use crate::http::ApiError; use crate::ingress::inclusion_lane::PendingUserOp; use crate::runtime::shutdown::RuntimeScope; -use sequencer_core::api::{TxRequest, TxResponse}; +use crate::storage::Storage; +use sequencer_core::api::{FeeResponse, TxRequest, TxResponse}; use sequencer_core::user_op::SignedUserOp; /// State for the submit endpoint. Kept narrow — only what `/tx` actually needs. @@ -58,15 +63,38 @@ impl SubmitState { } } +/// State for `GET /fee`. Reads the open-frame fee from SQLite; the inclusion +/// lane remains the sole writer of that fact. +#[derive(Clone)] +pub(crate) struct FeeState { + db_path: String, + shutdown: RuntimeScope, +} + +impl FeeState { + pub(crate) fn new(db_path: String, shutdown: RuntimeScope) -> Self { + Self { db_path, shutdown } + } + + fn reject_if_shutting_down(&self) -> Result<(), ApiError> { + if self.shutdown.is_shutdown_requested() { + Err(ApiError::unavailable("sequencer shutting down")) + } else { + Ok(()) + } + } +} + /// Build the ingress router. Caller wires it into an `axum::serve` listener. -pub(crate) fn router(state: Arc) -> Router { +pub(crate) fn router(submit: Arc, fee: Arc) -> Router { Router::new() .route("/tx", post(submit_tx)) - .with_state(state) + .with_state(submit) + .merge(Router::new().route("/fee", get(get_fee)).with_state(fee)) .layer( CorsLayer::new() .allow_origin(Any) - .allow_methods([Method::POST]) + .allow_methods([Method::GET, Method::POST]) .allow_headers(Any) .max_age(Duration::from_secs(3600)), ) @@ -99,6 +127,34 @@ async fn submit_tx( .into_response()) } +async fn get_fee(State(state): State>) -> Result { + state.reject_if_shutting_down()?; + let db_path = state.db_path.clone(); + let shutdown = state.shutdown.clone(); + let quote = tokio::task::spawn_blocking(move || { + let _runtime_lifetime = shutdown; + read_fee_quote(&db_path) + }) + .await + .map_err(|_| ApiError::internal_error("fee read task failed"))??; + Ok(Json(quote).into_response()) +} + +fn read_fee_quote(db_path: &str) -> Result { + let mut storage = Storage::open_read_only(db_path).map_err(|err| { + tracing::debug!(error = %err, "GET /fee storage open failed"); + ApiError::internal_error("fee unavailable") + })?; + match storage.current_fee_quote() { + Ok(Some((fee, recommended_fee))) => Ok(FeeResponse::quote(fee, recommended_fee)), + Ok(None) => Err(ApiError::unavailable("no open frame")), + Err(err) => { + tracing::debug!(error = %err, "GET /fee quote read failed"); + Err(ApiError::internal_error("fee unavailable")) + } + } +} + /// Normalize JSON-extractor failures into fixed client-facing messages. /// Keeps the public API contract stable across axum upgrades and avoids /// reflecting parser internals (serde line/column, token excerpts) to callers. @@ -227,6 +283,36 @@ mod tests { assert_eq!(err.code(), "UNAVAILABLE"); } + #[tokio::test(flavor = "current_thread")] + async fn get_fee_rejects_when_shutdown_has_started() { + let shutdown = RuntimeScope::default(); + shutdown.request_shutdown(); + let state = Arc::new(FeeState::new("unused.db".into(), shutdown)); + + let err = get_fee(State(state)) + .await + .expect_err("fee should be rejected during shutdown"); + assert_eq!(err.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(err.code(), "UNAVAILABLE"); + } + + #[tokio::test(flavor = "current_thread")] + async fn get_fee_is_unavailable_when_no_open_frame() { + let db = TempDir::new().expect("create temp dir"); + let db_path = db.path().join("sequencer.db"); + let _storage = Storage::open(&db_path.to_string_lossy()).expect("create db"); + let state = Arc::new(FeeState::new( + db_path.to_string_lossy().into_owned(), + RuntimeScope::default(), + )); + + let err = get_fee(State(state)) + .await + .expect_err("fee requires an open frame"); + assert_eq!(err.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(err.code(), "UNAVAILABLE"); + } + fn sign_user_op_hex( domain: &Eip712Domain, user_op: &UserOp, diff --git a/sequencer/src/ingress/mod.rs b/sequencer/src/ingress/mod.rs index 3795ac2f..3c64a322 100644 --- a/sequencer/src/ingress/mod.rs +++ b/sequencer/src/ingress/mod.rs @@ -1,9 +1,9 @@ // (c) Cartesi and individual authors (see AUTHORS) // SPDX-License-Identifier: Apache-2.0 (see LICENSE) -//! Inbound side: HTTP submit endpoint and the inclusion lane that consumes its -//! queue. The submit API is the public-facing port; the lane is the only writer -//! of open batch/frame state in storage. +//! Inbound side: public HTTP (`POST /tx`, `GET /fee`) and the inclusion lane +//! that consumes the submit queue. The lane is the only writer of open +//! batch/frame state in storage. pub mod api; pub mod inclusion_lane; diff --git a/sequencer/src/integration_tests/e2e_sequencer.rs b/sequencer/src/integration_tests/e2e_sequencer.rs index df2d2c27..13f22f19 100644 --- a/sequencer/src/integration_tests/e2e_sequencer.rs +++ b/sequencer/src/integration_tests/e2e_sequencer.rs @@ -861,6 +861,37 @@ async fn api_accepts_user_op_with_max_fee_equal_to_current_frame_fee() { shutdown_runtime(runtime).await; } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn api_quotes_open_frame_fee() { + let db = temp_db("fee-endpoint"); + let domain = test_domain(); + bootstrap_open_frame(db.path.as_str()); + + let Some(runtime) = start_full_server(db.path.as_str(), domain).await else { + return; + }; + + let endpoint = format!("http://{}", runtime.addr); + let client = SequencerClient::new_with_timeout(endpoint, Duration::from_secs(2)) + .expect("build sequencer client"); + let quoted = client.get_fee().await.expect("GET /fee"); + assert_eq!( + quoted.fee, 1356, + "quoted fee must match the bootstrapped open-frame fee" + ); + assert_eq!( + quoted.recommended_fee, 1356, + "bootstrapped recommended_fee matches the open-frame fee" + ); + assert_eq!( + quoted.suggested_max_fee, + sequencer_core::fee::suggested_signing_max_fee(quoted.fee, quoted.recommended_fee), + "suggested_max_fee is max(fee, recommended_fee) plus signing slack" + ); + + shutdown_runtime(runtime).await; +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn api_rejects_user_op_when_balance_below_fee_cost() { // if sender's balance < `fee_to_linear(current_frame_fee)` the diff --git a/sequencer/src/integration_tests/snapshot_endpoints.rs b/sequencer/src/integration_tests/snapshot_endpoints.rs index cb9b686a..54aba477 100644 --- a/sequencer/src/integration_tests/snapshot_endpoints.rs +++ b/sequencer/src/integration_tests/snapshot_endpoints.rs @@ -611,11 +611,63 @@ async fn cors_permits_browser_preflight_on_tx() { .to_str() .expect("header utf8"); assert_eq!(allow_origin, "*"); - assert_eq!(resp.headers()["access-control-allow-methods"], "POST"); + let allow_methods = resp + .headers() + .get("access-control-allow-methods") + .expect("Access-Control-Allow-Methods") + .to_str() + .expect("header utf8"); + assert!( + allow_methods + .split(',') + .any(|method| method.trim() == "POST"), + "preflight must allow POST, got {allow_methods}" + ); assert_eq!(resp.headers()["access-control-allow-headers"], "*"); assert_eq!(resp.headers()["access-control-max-age"], "3600"); } +#[tokio::test] +async fn cors_permits_browser_preflight_on_fee() { + let db = temp_db("cors-fee-preflight"); + let Some(server) = start_server(db.path.as_str()).await else { + return; + }; + + let resp = reqwest::Client::new() + .request(reqwest::Method::OPTIONS, server.url("/fee")) + .header("Origin", "https://wallet.example") + .header("Access-Control-Request-Method", "GET") + .send() + .await + .expect("OPTIONS /fee"); + + assert!( + resp.status().is_success(), + "preflight status: {}", + resp.status() + ); + let allow_origin = resp + .headers() + .get("access-control-allow-origin") + .expect("Access-Control-Allow-Origin") + .to_str() + .expect("header utf8"); + assert_eq!(allow_origin, "*"); + let allow_methods = resp + .headers() + .get("access-control-allow-methods") + .expect("Access-Control-Allow-Methods") + .to_str() + .expect("header utf8"); + assert!( + allow_methods + .split(',') + .any(|method| method.trim() == "GET"), + "preflight must allow GET, got {allow_methods}" + ); +} + #[tokio::test] async fn cors_is_limited_to_ingress_and_covers_rejections() { let db = temp_db("cors-route-scope"); @@ -636,6 +688,14 @@ async fn cors_is_limited_to_ingress_and_covers_rejections() { assert_eq!(rejected.status().as_u16(), 400); assert_eq!(rejected.headers()["access-control-allow-origin"], "*"); + let fee = client + .get(server.url("/fee")) + .header("Origin", "https://wallet.example") + .send() + .await + .expect("GET /fee"); + assert_eq!(fee.headers()["access-control-allow-origin"], "*"); + for route in ["/livez", "/finalized_state"] { let response = client .get(server.url(route)) diff --git a/sequencer/src/storage/ingress.rs b/sequencer/src/storage/ingress.rs index 05e957a0..ecf1cdf8 100644 --- a/sequencer/src/storage/ingress.rs +++ b/sequencer/src/storage/ingress.rs @@ -56,6 +56,17 @@ impl Storage { self.read(load_current_write_head) } + /// Open-frame fee and current `recommended_fee`. `None` if there is no Tip. + pub fn current_fee_quote(&mut self) -> Result> { + self.read(|tx| { + let Some(head) = load_current_write_head(tx)? else { + return Ok(None); + }; + let policy = query_batch_policy(tx)?; + Ok(Some((head.frame_fee, policy.recommended_fee))) + }) + } + /// Bootstrap the very first batch + frame with explicit values, returning /// its loaded [`WriteHead`]. Asserts no open state exists. /// From 9cd9599cbc18ba1cadaffad506b1c31ebecd1d1d Mon Sep 17 00:00:00 2001 From: Stephen Chen <20940639+stephenctw@users.noreply.github.com> Date: Wed, 16 Sep 2026 11:48:07 +0800 Subject: [PATCH 2/2] fix(http): fail loud on GET /fee storage faults Share snapshot storage_task across ingress and egress so a panic or corrupt DB aborts instead of returning 500 while POST /tx keeps running. --- docs/watchdog/operator-deployment.md | 6 +- sequencer/src/egress/api/snapshot.rs | 120 +++++++++------------------ sequencer/src/http.rs | 56 +++++++++++++ sequencer/src/ingress/api.rs | 63 ++++++++++---- sequencer/src/storage/ingress.rs | 10 +++ 5 files changed, 151 insertions(+), 104 deletions(-) diff --git a/docs/watchdog/operator-deployment.md b/docs/watchdog/operator-deployment.md index cbeedd07..1202813d 100644 --- a/docs/watchdog/operator-deployment.md +++ b/docs/watchdog/operator-deployment.md @@ -9,9 +9,9 @@ For **local development only** (Anvil + `sequencer-devnet`, CI smoke tests), use ## Two deployment tiers ```text - ┌─────────────────────────────────────┐ - Internet / users │ Public ingress (POST /tx, WS) │ ← benchmarks, wallets - └─────────────────┬───────────────────┘ + ┌──────────────────────────────────────────────┐ + Internet / users │ Public ingress (POST /tx, GET /fee, WS) │ ← benchmarks, wallets + └──────────────────────┬───────────────────────┘ │ ┌─────────────────▼───────────────────┐ Operator network │ Sequencer process │ diff --git a/sequencer/src/egress/api/snapshot.rs b/sequencer/src/egress/api/snapshot.rs index 081d56b5..58d9e930 100644 --- a/sequencer/src/egress/api/snapshot.rs +++ b/sequencer/src/egress/api/snapshot.rs @@ -38,10 +38,11 @@ use tokio::fs::File; use tokio::io::{AsyncRead, ReadBuf}; use tokio_util::io::ReaderStream; +use crate::http::{StorageTaskError, storage_task}; use crate::runtime::shutdown::{RuntimeScope, abort_terminal}; use crate::storage::{FinalizedLease, LeaseGuard, LeasedDump, ReleaseScheduler, Storage}; -type BoxError = Box; +type BoxError = StorageTaskError; /// Wiring for the snapshot endpoints: where the DB is, and how to find the /// canonical state file inside a dump. `state_file_in_dump` is threaded as a @@ -89,9 +90,11 @@ struct InclusionBlockResponse { /// opened). 404 if no finalized snapshot exists. async fn finalized_inclusion_block(State(state): State>) -> Response { let db_path = state.snapshot.db_path.clone(); - let result = storage_task(&state, "read finalized inclusion block", move |_scope| { - Ok(Storage::open_read_only(&db_path)?.finalized_dump()?) - }) + let result = storage_task( + state.shutdown.clone(), + "read finalized inclusion block", + move |_scope| Ok(Storage::open_read_only(&db_path)?.finalized_dump()?), + ) .await; match result { Ok(Some(finalized)) => Json(InclusionBlockResponse { @@ -188,69 +191,44 @@ fn stream_body(file: File, guard: LeaseGuard) -> Body { })) } -// ── Blocking storage tasks ───────────────────────────────────────────────── - -/// Classify inside the blocking task: cancellation of the HTTP request must -/// not discard a persistent fault discovered by work that already started. -async fn storage_task( - state: &SnapshotApiState, - operation: &'static str, - work: F, -) -> Result -where - T: Send + 'static, - F: FnOnce(RuntimeScope) -> Result + Send + 'static, -{ - let scope = state.shutdown.clone(); - match tokio::task::spawn_blocking(move || { - // The independent clone outlives both work and its SQLite connection, - // even when work consumes its scope argument before returning. - let _runtime_lifetime = scope.clone(); - match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| work(scope))) { - Ok(Err(error)) if persistent_storage_error(error.as_ref()) => { - abort_terminal(format_args!("{operation}: {error}")); - } - Ok(result) => result, - Err(_) => abort_terminal(format_args!("{operation}: storage task panicked")), - } - }) - .await - { - Ok(result) => result, - Err(join) if join.is_panic() => abort_terminal(format_args!("{operation}: {join}")), - Err(join) => Err(Box::new(join)), - } -} - // ── Lease acquisition (storage returns the dump bundled with its release) ── async fn acquire_finalized(state: &SnapshotApiState) -> Result, BoxError> { let db_path = state.snapshot.db_path.clone(); let release_scheduler = state.release_scheduler.clone(); - storage_task(state, "acquire finalized snapshot lease", move |scope| { - let report_persistent_failure: crate::storage::PersistentReleaseFailureReporter = - Arc::new(move |cause: &str| { - let _runtime_lifetime = &scope; - abort_terminal(cause) - }); - let mut storage = Storage::open_writer(&db_path)?; - Ok(storage.acquire_finalized_lease(release_scheduler, report_persistent_failure)?) - }) + storage_task( + state.shutdown.clone(), + "acquire finalized snapshot lease", + move |scope| { + let report_persistent_failure: crate::storage::PersistentReleaseFailureReporter = + Arc::new(move |cause: &str| { + let _runtime_lifetime = &scope; + abort_terminal(cause) + }); + let mut storage = Storage::open_writer(&db_path)?; + Ok(storage.acquire_finalized_lease(release_scheduler, report_persistent_failure)?) + }, + ) .await } async fn acquire_latest(state: &SnapshotApiState) -> Result, BoxError> { let db_path = state.snapshot.db_path.clone(); let release_scheduler = state.release_scheduler.clone(); - storage_task(state, "acquire latest snapshot lease", move |scope| { - let report_persistent_failure: crate::storage::PersistentReleaseFailureReporter = - Arc::new(move |cause: &str| { - let _runtime_lifetime = &scope; - abort_terminal(cause) - }); - let mut storage = Storage::open_writer(&db_path)?; - Ok(storage.acquire_latest_snapshot_lease(release_scheduler, report_persistent_failure)?) - }) + storage_task( + state.shutdown.clone(), + "acquire latest snapshot lease", + move |scope| { + let report_persistent_failure: crate::storage::PersistentReleaseFailureReporter = + Arc::new(move |cause: &str| { + let _runtime_lifetime = &scope; + abort_terminal(cause) + }); + let mut storage = Storage::open_writer(&db_path)?; + Ok(storage + .acquire_latest_snapshot_lease(release_scheduler, report_persistent_failure)?) + }, + ) .await } @@ -289,23 +267,6 @@ fn internal_error(context: &str, err: impl std::fmt::Display) -> Response { tracing::warn!(error = %err, context, "snapshot endpoint failed"); StatusCode::INTERNAL_SERVER_ERROR.into_response() } -fn persistent_storage_error(mut error: &(dyn std::error::Error + 'static)) -> bool { - loop { - let persistent = error - .downcast_ref::() - .is_some_and(crate::storage::is_persistent_storage_error) - || error - .downcast_ref::() - .is_some_and(crate::storage::is_persistent_storage_open_error); - if persistent { - return true; - } - let Some(source) = error.source() else { - return false; - }; - error = source; - } -} #[cfg(test)] mod tests { @@ -430,7 +391,7 @@ mod tests { None, )); - assert!(!persistent_storage_error(&error)); + assert!(!crate::http::persistent_storage_error(&error)); } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] @@ -477,19 +438,12 @@ mod tests { ) { return; } - let state = SnapshotApiState { - snapshot: SnapshotState { - db_path: String::new(), - state_file_in_dump: |prefix| prefix.join("state"), - }, - shutdown: RuntimeScope::default(), - release_scheduler: Arc::new(|release| release()), - }; + let shutdown = RuntimeScope::default(); let (started_tx, started_rx) = tokio::sync::oneshot::channel(); let (release_tx, release_rx) = std::sync::mpsc::channel(); let request = tokio::spawn(async move { storage_task::<(), _>( - &state, + shutdown, "cancelled request corruption probe", move |_scope| { started_tx.send(()).expect("started storage task"); diff --git a/sequencer/src/http.rs b/sequencer/src/http.rs index b654eb4b..eb84b5e7 100644 --- a/sequencer/src/http.rs +++ b/sequencer/src/http.rs @@ -216,6 +216,62 @@ async fn run_snapshot_release_supervisor( } } +// ── Blocking storage tasks ───────────────────────────────────────────────── +// +// Shared by ingress (`GET /fee`) and egress snapshot handlers. Classify +// inside the blocking task: cancellation of the HTTP request must not +// discard a persistent fault discovered by work that already started. + +pub(crate) type StorageTaskError = Box; + +/// Run `work` on a blocking thread with a runtime-scope clone that outlives +/// the SQLite connection. Panics and persistent storage errors abort the +/// process; other errors return to the handler. +pub(crate) async fn storage_task( + shutdown: RuntimeScope, + operation: &'static str, + work: F, +) -> Result +where + T: Send + 'static, + F: FnOnce(RuntimeScope) -> Result + Send + 'static, +{ + match tokio::task::spawn_blocking(move || { + let _runtime_lifetime = shutdown.clone(); + match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| work(shutdown))) { + Ok(Err(error)) if persistent_storage_error(error.as_ref()) => { + abort_terminal(format_args!("{operation}: {error}")); + } + Ok(result) => result, + Err(_) => abort_terminal(format_args!("{operation}: storage task panicked")), + } + }) + .await + { + Ok(result) => result, + Err(join) if join.is_panic() => abort_terminal(format_args!("{operation}: {join}")), + Err(join) => Err(Box::new(join)), + } +} + +pub(crate) fn persistent_storage_error(mut error: &(dyn std::error::Error + 'static)) -> bool { + loop { + let persistent = error + .downcast_ref::() + .is_some_and(crate::storage::is_persistent_storage_error) + || error + .downcast_ref::() + .is_some_and(crate::storage::is_persistent_storage_open_error); + if persistent { + return true; + } + let Some(source) = error.source() else { + return false; + }; + error = source; + } +} + /// The API's per-deployment configuration: the two ingress values that vary /// (the EIP-712 verification domain and the app's payload bound) plus three /// service limits. The limits are module constants by design — not diff --git a/sequencer/src/ingress/api.rs b/sequencer/src/ingress/api.rs index 204e1b36..15753349 100644 --- a/sequencer/src/ingress/api.rs +++ b/sequencer/src/ingress/api.rs @@ -23,7 +23,7 @@ use tokio::sync::oneshot; use tower_http::cors::{Any, CorsLayer}; use tracing::debug; -use crate::http::ApiError; +use crate::http::{ApiError, storage_task}; use crate::ingress::inclusion_lane::PendingUserOp; use crate::runtime::shutdown::RuntimeScope; use crate::storage::Storage; @@ -130,26 +130,18 @@ async fn submit_tx( async fn get_fee(State(state): State>) -> Result { state.reject_if_shutting_down()?; let db_path = state.db_path.clone(); - let shutdown = state.shutdown.clone(); - let quote = tokio::task::spawn_blocking(move || { - let _runtime_lifetime = shutdown; - read_fee_quote(&db_path) + let result = storage_task(state.shutdown.clone(), "read fee quote", move |_scope| { + let mut storage = Storage::open_read_only(&db_path)?; + Ok(storage.current_fee_quote()?) }) - .await - .map_err(|_| ApiError::internal_error("fee read task failed"))??; - Ok(Json(quote).into_response()) -} - -fn read_fee_quote(db_path: &str) -> Result { - let mut storage = Storage::open_read_only(db_path).map_err(|err| { - tracing::debug!(error = %err, "GET /fee storage open failed"); - ApiError::internal_error("fee unavailable") - })?; - match storage.current_fee_quote() { - Ok(Some((fee, recommended_fee))) => Ok(FeeResponse::quote(fee, recommended_fee)), + .await; + match result { + Ok(Some((fee, recommended_fee))) => { + Ok(Json(FeeResponse::quote(fee, recommended_fee)).into_response()) + } Ok(None) => Err(ApiError::unavailable("no open frame")), Err(err) => { - tracing::debug!(error = %err, "GET /fee quote read failed"); + tracing::warn!(error = %err, "GET /fee failed"); Err(ApiError::internal_error("fee unavailable")) } } @@ -313,6 +305,41 @@ mod tests { assert_eq!(err.code(), "UNAVAILABLE"); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[cfg(unix)] + async fn corrupt_fee_policy_trips_terminal_storage_fault() { + if !crate::runtime::shutdown::abort_test_child( + "ingress::api::tests::corrupt_fee_policy_trips_terminal_storage_fault", + ) { + return; + } + let db = TempDir::new().expect("create temp dir"); + let db_path = db.path().join("sequencer.db"); + let mut storage = Storage::open(&db_path.to_string_lossy()).expect("create db"); + storage + .initialize_open_state(0, crate::storage::SafeInputRange::empty_at(0)) + .expect("open tip"); + drop(storage); + + let conn = Storage::open_connection(&db_path.to_string_lossy()).expect("raw connection"); + conn.execute_batch( + "PRAGMA ignore_check_constraints = ON; + UPDATE batch_policy SET log_delta = -10000 WHERE singleton_id = 0;", + ) + .expect("inject impossible policy row"); + drop(conn); + + let state = Arc::new(FeeState::new( + db_path.to_string_lossy().into_owned(), + RuntimeScope::default(), + )); + let result = get_fee(State(state)).await; + panic!( + "terminal fee fault returned instead of aborting: {:?}", + result.as_ref().err().map(ApiError::status) + ); + } + fn sign_user_op_hex( domain: &Eip712Domain, user_op: &UserOp, diff --git a/sequencer/src/storage/ingress.rs b/sequencer/src/storage/ingress.rs index ecf1cdf8..3b73c9e5 100644 --- a/sequencer/src/storage/ingress.rs +++ b/sequencer/src/storage/ingress.rs @@ -1527,6 +1527,11 @@ mod tests { head.frame_fee, 1356, "WriteHead.frame_fee must stay stable until advance_frame runs", ); + assert_eq!( + storage.current_fee_quote().expect("quote"), + Some((1356, 1456)), + "quote splits the frozen open-frame fee from the advanced recommended_fee", + ); // Closing the frame picks up the new policy — the *next* frame opens // at 1456. This is the expected policy-flow boundary. @@ -1538,6 +1543,11 @@ mod tests { head.frame_fee, 1456, "the next frame must use the updated policy's fee (policy flows in at close)", ); + assert_eq!( + storage.current_fee_quote().expect("quote"), + Some((1456, 1456)), + "after rotation the quote's two fees agree again", + ); } #[test]