diff --git a/Cargo.lock b/Cargo.lock index fba46947..f13b6167 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11711,6 +11711,28 @@ dependencies = [ "sp-runtime", ] +[[package]] +name = "pallet-prover-db-indexer" +version = "0.1.0" +dependencies = [ + "log", + "native-api", + "pallet-commitments", + "pallet-indexing", + "pallet-permissions", + "pallet-system-tables", + "pallet-tables", + "pallet-zkpay", + "parity-scale-codec", + "parking_lot 0.12.3", + "polkadot-sdk 0.9.0", + "proof-of-sql-commitment-map", + "prost 0.13.5", + "prost-build 0.13.5", + "scale-info", + "sxt-core", +] + [[package]] name = "pallet-proxy" version = "38.0.0" @@ -15502,6 +15524,16 @@ dependencies = [ "prost-derive 0.12.6", ] +[[package]] +name = "prost" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2796faa41db3ec313a31f7624d9286acf277b52de526150b7e69f3debf891ee5" +dependencies = [ + "bytes", + "prost-derive 0.13.5", +] + [[package]] name = "prost-build" version = "0.11.9" @@ -15545,6 +15577,26 @@ dependencies = [ "tempfile", ] +[[package]] +name = "prost-build" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" +dependencies = [ + "heck 0.5.0", + "itertools 0.14.0", + "log", + "multimap 0.10.0", + "once_cell", + "petgraph", + "prettyplease 0.2.30", + "prost 0.13.5", + "prost-types 0.13.5", + "regex", + "syn 2.0.106", + "tempfile", +] + [[package]] name = "prost-derive" version = "0.11.9" @@ -15571,6 +15623,19 @@ dependencies = [ "syn 2.0.106", ] +[[package]] +name = "prost-derive" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" +dependencies = [ + "anyhow", + "itertools 0.14.0", + "proc-macro2", + "quote", + "syn 2.0.106", +] + [[package]] name = "prost-types" version = "0.11.9" @@ -15589,6 +15654,15 @@ dependencies = [ "prost 0.12.6", ] +[[package]] +name = "prost-types" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52c2c1bf36ddb1a1c396b3601a3cec27c2462e45f07c386894ec3ccf5332bd16" +dependencies = [ + "prost 0.13.5", +] + [[package]] name = "protobuf" version = "2.28.0" @@ -21131,6 +21205,7 @@ dependencies = [ "substrate-build-script-utils", "sxt-core", "sxt-runtime", + "url", ] [[package]] @@ -21144,6 +21219,7 @@ dependencies = [ "pallet-indexing", "pallet-keystore", "pallet-permissions", + "pallet-prover-db-indexer", "pallet-rewards", "pallet-smartcontracts", "pallet-system-contracts", diff --git a/Cargo.toml b/Cargo.toml index 515cc286..aa653c72 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,7 +31,7 @@ members = [ "pallets/rewards", "pallets/keystore", "pallets/smartcontracts", - "event-forwarder", + "event-forwarder", "chain-utils", "pallets/system-contracts", "canaries", @@ -39,6 +39,7 @@ members = [ "proof-of-sql/commitment-column-mapping", "eth-ecdsa", "arrow-ipc-no-std", + "pallets/prover_db_indexer", ] exclude = [ "utoipa", @@ -47,7 +48,7 @@ resolver = "2" [workspace.dependencies] polkadot-sdk = { version = "0.9.0", default-features = false } -sc-cli = { version = "0.47.0", default-features = false } +sc-cli = { version = "0.47.0", default-features = false } sc-client-db = { version = "0.44.1", default-features = false } # Required for the runtime_interface macro in our native modules @@ -79,6 +80,7 @@ pallet-smartcontracts = { path = "pallets/smartcontracts", default-features = fa sxt-core = { path = "sxt-core", default-features = false } pallet-tables = { path = "pallets/tables", default-features = false } pallet-indexing = { path = "pallets/indexing", default-features = false } +pallet-prover-db-indexer = { path = "pallets/prover_db_indexer", default-features = false } native = { path = "native", default-features = false } native-api = { path = "native-api", default-features = false } num-bigint = { version = "0.4", default-features = false } diff --git a/node/Cargo.toml b/node/Cargo.toml index 228dba5e..0f482e04 100644 --- a/node/Cargo.toml +++ b/node/Cargo.toml @@ -46,6 +46,7 @@ rand.workspace = true node-rpc.workspace = true serde.workspace = true codec = { workspace = true, default-features = true } +url = { workspace = true } [build-dependencies] substrate-build-script-utils.workspace = true diff --git a/node/src/cli.rs b/node/src/cli.rs index 922f8bf4..2d1b7d4c 100644 --- a/node/src/cli.rs +++ b/node/src/cli.rs @@ -19,6 +19,17 @@ pub struct Cli { #[clap(long)] pub event_forwarder_rpc: Option, + /// If set, writes the URL into OCW persistent local storage under + /// `prover_db_indexer::prover_db_url` at startup, telling the + /// prover-db-indexer OCW where to POST forwarded events. Seeds the + /// storage before the first block is authored, so no events are + /// missed between node-up and URL-configured. + /// + /// Parsed as a `url::Url` so an invalid value rejects at startup + /// rather than silently failing on the first OCW HTTP request. + #[clap(long)] + pub prover_db_url: Option, + #[allow(missing_docs)] #[clap(flatten)] pub storage_monitor: sc_storage_monitor::StorageMonitorParams, diff --git a/node/src/service.rs b/node/src/service.rs index f24972b7..677b370a 100644 --- a/node/src/service.rs +++ b/node/src/service.rs @@ -14,6 +14,7 @@ use polkadot_sdk::sc_service::error::Error as ServiceError; use polkadot_sdk::sc_service::TaskManager; use polkadot_sdk::sc_telemetry::{Telemetry, TelemetryWorker}; use polkadot_sdk::sc_transaction_pool_api::OffchainTransactionPoolFactory; +use polkadot_sdk::sp_core::offchain::{OffchainStorage, STORAGE_PREFIX}; use polkadot_sdk::sp_runtime::traits::Block as BlockT; use polkadot_sdk::{ sc_authority_discovery, @@ -75,6 +76,32 @@ pub type TransactionPool = sc_transaction_pool::FullPool; /// imported and generated. const GRANDPA_JUSTIFICATION_PERIOD: u32 = 512; +/// Seed the prover-db-indexer OCW's persistent local storage with the +/// prover-db URL given on the command line, before the first block is +/// authored. +/// +/// The storage key is +/// `pallet_prover_db_indexer::PROVER_DB_URL_KEY = "prover_db_indexer::prover_db_url"`; +/// the value is a SCALE-encoded `Vec` of the URL bytes. +#[expect( + clippy::result_large_err, + reason = "ServiceError is from substrate and cannot be modified" +)] +fn configure_prover_db_url(backend: &FullBackend, url: &url::Url) -> Result<(), ServiceError> { + use codec::Encode; + let Some(mut storage) = backend.offchain_storage() else { + return Err(ServiceError::Other( + "backend did not expose an offchain storage handle; \ + cannot apply --prover-db-url" + .into(), + )); + }; + let key = sxt_runtime::pallet_prover_db_indexer::PROVER_DB_URL_KEY; + let encoded = url.as_str().as_bytes().to_vec().encode(); + storage.set(STORAGE_PREFIX, key, &encoded); + Ok(()) +} + #[allow(clippy::type_complexity)] #[expect( clippy::result_large_err, @@ -290,6 +317,7 @@ pub struct NewFullBase { )] pub fn new_full_base::Hash>>( config: Configuration, + prover_db_url: Option, ) -> Result { let role = config.role; let force_authoring = config.force_authoring; @@ -322,6 +350,34 @@ pub fn new_full_base::Hash>>( other: (rpc_builder, import_setup, rpc_setup, mut telemetry, statement_store), } = new_partial(&config)?; + // The prover-db-indexer pallet's `on_finalize` calls + // `sp_io::offchain_index::set` to hand events to the OCW. That call is a + // silent no-op unless `--enable-offchain-indexing=true` was passed. We + // surface that loudly rather than letting operators discover it by + // watching zero rows arrive at their indexer. + if let Some(url) = prover_db_url.as_ref() { + if !config.offchain_worker.indexing_enabled { + return Err(ServiceError::Other( + "--prover-db-url was set but --enable-offchain-indexing is not \ + true: prover-db-indexer's on_finalize writes would be silently \ + dropped. Restart with --enable-offchain-indexing=true, or \ + omit --prover-db-url." + .into(), + )); + } + configure_prover_db_url(backend.as_ref(), url)?; + } else if !config.offchain_worker.indexing_enabled { + // Prover-db-indexer is in the runtime but offchain indexing is off. + // Forwarding can't work even if someone later writes the URL via RPC. + // Loud eprintln! so it shows up even without log filter setup. + eprintln!( + "warning: --enable-offchain-indexing is not true; the \ + prover-db-indexer OCW cannot forward events even if \ + prover_db_indexer::prover_db_url is set via RPC. Pass \ + --enable-offchain-indexing=true to enable forwarding." + ); + } + let metrics = N::register_notification_metrics( config.prometheus_config.as_ref().map(|cfg| &cfg.registry), ); @@ -603,13 +659,14 @@ pub fn new_full(config: Configuration, cli: Cli) -> Result { - new_full_base::>(config) + new_full_base::>(config, prover_db_url) .map(|NewFullBase { task_manager, .. }| task_manager)? } sc_network::config::NetworkBackendType::Litep2p => { - new_full_base::(config) + new_full_base::(config, prover_db_url) .map(|NewFullBase { task_manager, .. }| task_manager)? } }; diff --git a/pallets/prover_db_indexer/Cargo.toml b/pallets/prover_db_indexer/Cargo.toml new file mode 100644 index 00000000..39a4c4ee --- /dev/null +++ b/pallets/prover_db_indexer/Cargo.toml @@ -0,0 +1,67 @@ +[package] +name = "pallet-prover-db-indexer" +description = "Offchain worker that forwards table lifecycle and data events to an external prover-db indexer (HTTP)." +version = "0.1.0" +license = "Unlicense" +authors.workspace = true +homepage.workspace = true +repository.workspace = true +edition.workspace = true +publish = false + +[package.metadata.docs.rs] +targets = ["x86_64-unknown-linux-gnu"] + +[dependencies] +codec = { features = ["derive"], workspace = true } +scale-info = { features = ["derive"], workspace = true } + +polkadot-sdk = { workspace = true, features = [ + "frame-support", "frame-system", "sp-core", "sp-io", "sp-runtime", +]} + +sxt-core.workspace = true +pallet-tables.workspace = true +pallet-indexing.workspace = true +native-api.workspace = true +log = { workspace = true, default-features = false } +prost = { version = "0.13", default-features = false, features = ["prost-derive"] } + +[build-dependencies] +prost-build = "0.13" + +[dev-dependencies] +polkadot-sdk = { workspace = true, features = [ + "sp-core", "sp-io", "sp-runtime", "sp-staking", + "pallet-balances", "pallet-staking", "pallet-staking-reward-curve", + "pallet-session", "pallet-timestamp", + "frame-election-provider-support", +]} +pallet-permissions = { workspace = true, features = ["std"] } +pallet-commitments = { workspace = true, features = ["std"] } +pallet-tables = { workspace = true, features = ["std"] } +pallet-indexing = { workspace = true, features = ["std"] } +pallet-system-tables = { workspace = true, features = ["std"] } +pallet-zkpay = { workspace = true, features = ["std"] } +proof-of-sql-commitment-map.workspace = true +parking_lot = "0.12" + +[features] +default = ["std"] +std = [ + "polkadot-sdk/std", + "codec/std", + "scale-info/std", + "sxt-core/std", + "pallet-tables/std", + "pallet-indexing/std", + "native-api/std", + "log/std", + "prost/std", +] +try-runtime = [ + "polkadot-sdk/try-runtime", +] + +[lints] +workspace = true diff --git a/pallets/prover_db_indexer/README.md b/pallets/prover_db_indexer/README.md new file mode 100644 index 00000000..5722c555 --- /dev/null +++ b/pallets/prover_db_indexer/README.md @@ -0,0 +1,121 @@ +# pallet-prover-db-indexer + +Offchain worker that forwards table lifecycle and data events to an +external prover-db indexer (HTTP) using protobuf-over-HTTP. + +Watches for these events from other pallets and relays them: +- `pallet_tables::SchemaUpdated` +- `pallet_tables::TablesCreatedWithCommitments` +- `pallet_tables::TableDropped` +- `pallet_indexing::QuorumReached` + +Wire contract is under `proto/prover-db.proto`; five POST endpoints at +`/v1/{create_table,drop_table,put_batches,checkpoint,get_last_checkpoint}`. + +## Operational requirements + +The pallet has **two** node-side flags that must be correctly set for +forwarding to work. Getting either wrong produces a silent no-op on the +producer side — events look like they'd be forwarded but nothing arrives +at the indexer. + +### 1. `--enable-offchain-indexing=true` — **required** + +The producer (`on_finalize` hook) writes events into offchain local +storage via `sp_io::offchain_index::set`. That host function is a +**silent no-op unless `--enable-offchain-indexing=true` is passed on the +command line**. Substrate defaults the flag to `false`. + +Symptom if missing: `on_finalize(N): writing K events to offchain DB` +logs appear normally, but the OCW's `offchain_index::read(N)` returns +`None` and no forwarding happens. + +As of commit `dd873ed`, the node boots with a loud `warning:` to stderr +if this flag is not set; running with `--prover-db-url` *without* the flag +is a hard startup error. + +### 2. `--offchain-worker=always` — required on dev nodes + +Substrate defaults offchain-worker scheduling to `when-authority` — +OCWs only fire on validator/collator nodes. For a dev `--dev --tmp` +node, that means zero OCW invocations. Pass `--offchain-worker=always` +to force them to run regardless of role. + +### 3. `--prover-db-url ` — optional (but usually what you want) + +Tells the OCW where to POST forwarded events. The node writes the URL +into the OCW's persistent local storage before the first block is +authored, so no events are missed between node-up and URL-configured. + +If omitted, the OCW stays dormant until the URL is written some other +way — e.g. the `offchain_localStorageSet` RPC (requires +`--rpc-methods=unsafe`) or a persistent base-path node carrying the URL +over from a previous run. + +## Minimal one-command dev setup + +``` +./target/release/sxt-node \ + --dev --tmp \ + --rpc-cors=all \ + --offchain-worker=always \ + --enable-offchain-indexing=true \ + --prover-db-url http://127.0.0.1:9999 +``` + +## Runtime wiring + +Three Config associated types; the runtime provides them: + +```rust +impl pallet_prover_db_indexer::Config for Runtime { + type RuntimeEvent = RuntimeEvent; + + // Used to resolve pallet_indexing's pallet index dynamically via + // PalletInfoAccess::index(), so reordering construct_runtime! never + // breaks the filter silently. + type IndexingPallet = Indexing; + + // Variant index of `QuorumReached` within pallet_indexing::Event. + // The runtime supplies a resolver that looks the variant up by name + // at startup via scale_info::TypeInfo. See + // `DynamicQuorumReachedIndex` in runtime/src/lib.rs. + type QuorumReachedVariantIndex = DynamicQuorumReachedIndex; +} +``` + +## Data format + +- `CreateTable.arrow_schema`: raw `CREATE TABLE …` DDL bytes (UTF-8). + The HTTP server parses them via `sqlparser` to derive the Arrow schema. +- `PutBatches.record_batch`: postcard-encoded `OnChainTable` bytes — + what `pallet_indexing::QuorumReached.data` carries. The chain builds + this in `finalize_quorum`: it converts the indexer's Arrow IPC + submission to an `OnChainTable`, appends commitment meta columns, + and postcard-serializes the result. The prover-db-indexer never + decodes this payload — it's an opaque pass-through to the server. + +The pallet does **not** introduce any new host functions; everything +goes through already-in-stable `sp_io` APIs (`offchain_index::set`, +`offchain::http::Request`, etc.). + +## Dedup key contract + +Every forwarded table is registered with the dedup key column +`META_ROW_NUMBER`. The HTTP server rejects `CreateTable` requests whose +Arrow schema lacks a `META_ROW_NUMBER BIGINT NOT NULL` column. If your +on-chain table DDLs don't include this column, the first forward +attempt will fail at the server. Either add the column to the DDL or +patch the forwarder's `DEDUP_KEY_COLUMN` constant. + +## Testing + +- **Unit / integration.** `cargo test -p pallet-prover-db-indexer` — 7 + tests using `sp_core::offchain::testing::TestOffchainExt` to exercise + skip/forward/delete/multi-block/resume paths without a real HTTP + server. +- **End-to-end against a real indexer.** Build and run `prb-service` + with `--features indexer` (see the sxtdb repo), then launch + `sxt-node` with `--prover-db-url http://:`. + The harness at `spaceandtimefdn/sxt-int-harness` (standalone crate) + drives table/data actions against the node via a TOML file. diff --git a/pallets/prover_db_indexer/build.rs b/pallets/prover_db_indexer/build.rs new file mode 100644 index 00000000..58595a6c --- /dev/null +++ b/pallets/prover_db_indexer/build.rs @@ -0,0 +1,8 @@ +//! Compiles `proto/prover-db.proto` into Rust types via `prost_build`. + +fn main() { + prost_build::Config::new() + .compile_protos(&["proto/prover-db.proto"], &["proto"]) + .expect("failed to compile proto/prover-db.proto"); + println!("cargo:rerun-if-changed=proto/prover-db.proto"); +} diff --git a/pallets/prover_db_indexer/proto/prover-db.proto b/pallets/prover_db_indexer/proto/prover-db.proto new file mode 100644 index 00000000..2a93e18e --- /dev/null +++ b/pallets/prover_db_indexer/proto/prover-db.proto @@ -0,0 +1,42 @@ +syntax = "proto3"; + +package io.spaceandtime.indexer; + +// Stripped-down version of the indexer proto for HTTP+protobuf transport. +// The OCW encodes these messages with prost and POSTs them as +// Content-Type: application/x-protobuf to the indexer HTTP endpoint. +// google.protobuf.Empty responses are represented by empty HTTP 200 bodies. + +message CreateTableRequest { + uint64 sequence_number = 1; + string table_name = 2; + bytes arrow_schema = 3; + string key = 4; +} + +message TableBatch { + string table_name = 1; + bytes record_batch = 2; +} + +message PutBatchesRequest { + uint64 sequence_number = 1; + repeated TableBatch batches = 2; +} + +// Control-plane messages use JSON-in-string on the gRPC side, but for HTTP +// transport we use proper protobuf fields to avoid double-encoding. + +message DropTableRequest { + uint64 sequence_number = 1; + string table_name = 2; +} + +message CheckpointRequest { + uint64 sequence_number = 1; +} + +message GetLastCheckpointResponse { + uint64 sequence_number = 1; + bool has_checkpoint = 2; +} diff --git a/pallets/prover_db_indexer/src/http_client.rs b/pallets/prover_db_indexer/src/http_client.rs new file mode 100644 index 00000000..7a89d848 --- /dev/null +++ b/pallets/prover_db_indexer/src/http_client.rs @@ -0,0 +1,152 @@ +//! HTTP+protobuf client for the indexer service. +//! +//! Uses `polkadot_sdk::sp_io::offchain::http` to POST protobuf-encoded request bodies to +//! the indexer's HTTP endpoints. Works in both std and wasm (no_std) runtimes. + +use alloc::format; +use alloc::string::String; +use alloc::vec::Vec; + +use polkadot_sdk::sp_runtime::offchain::http::Request; +use polkadot_sdk::sp_runtime::offchain::Duration; +use prost::Message; + +use crate::proto; + +/// Default HTTP request deadline (30 seconds). +const HTTP_TIMEOUT_MS: u64 = 30_000; + +/// Errors from the HTTP client. +#[derive(Debug)] +#[allow(clippy::enum_variant_names)] +pub enum Error { + /// Failed to send the HTTP request. + SendFailed, + /// HTTP request timed out. + DeadlineReached, + /// IO or unknown error during HTTP request. + IoError, + /// Server returned a non-200 status code. + Status(u16), + /// Failed to decode the response body. + Decode(prost::DecodeError), +} + +impl From for Error { + fn from(e: prost::DecodeError) -> Self { + Error::Decode(e) + } +} + +impl core::fmt::Display for Error { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Error::SendFailed => write!(f, "failed to send HTTP request"), + Error::DeadlineReached => write!(f, "HTTP request deadline reached"), + Error::IoError => write!(f, "HTTP IO error"), + Error::Status(code) => write!(f, "server returned status {}", code), + Error::Decode(e) => write!(f, "protobuf decode error: {}", e), + } + } +} + +/// Fetch the last checkpoint from the server. +pub fn get_last_checkpoint(base_url: &str) -> Result, Error> { + let url = format!("{}/v1/get_last_checkpoint", base_url); + let body = post(&url, &[])?; + let resp = proto::GetLastCheckpointResponse::decode(body.as_slice())?; + Ok(resp.has_checkpoint.then_some(resp.sequence_number)) +} + +/// Register a new table. v1 semantics: `arrow_schema` carries raw DDL bytes. +pub fn create_table( + base_url: &str, + sequence_number: u64, + table_name: String, + arrow_schema: Vec, + key: String, +) -> Result<(), Error> { + let req = proto::CreateTableRequest { + sequence_number, + table_name, + arrow_schema, + key, + }; + let url = format!("{}/v1/create_table", base_url); + post(&url, &req.encode_to_vec())?; + Ok(()) +} + +/// Soft-delete a table. +pub fn drop_table(base_url: &str, sequence_number: u64, table_name: String) -> Result<(), Error> { + let req = proto::DropTableRequest { + sequence_number, + table_name, + }; + let url = format!("{}/v1/drop_table", base_url); + post(&url, &req.encode_to_vec())?; + Ok(()) +} + +/// Ingest one or more table batches. +pub fn put_batches( + base_url: &str, + sequence_number: u64, + batches: Vec, +) -> Result<(), Error> { + let req = proto::PutBatchesRequest { + sequence_number, + batches, + }; + let url = format!("{}/v1/put_batches", base_url); + post(&url, &req.encode_to_vec())?; + Ok(()) +} + +/// Record a checkpoint for the given sequence number (block number). +pub fn checkpoint(base_url: &str, sequence_number: u64) -> Result<(), Error> { + let req = proto::CheckpointRequest { sequence_number }; + let url = format!("{}/v1/checkpoint", base_url); + post(&url, &req.encode_to_vec())?; + Ok(()) +} + +/// Low-level POST helper. Sends `body` to `url` with +/// `Content-Type: application/x-protobuf` and returns the response body bytes. +fn post(url: &str, body: &[u8]) -> Result, Error> { + let deadline = + polkadot_sdk::sp_io::offchain::timestamp().add(Duration::from_millis(HTTP_TIMEOUT_MS)); + + // Pass body chunks as `Vec>`. Empty body → no chunks (otherwise + // `Request::send` would write an empty chunk and then try to finalize + // again, which fails with "body sending already finished"). + let chunks: Vec> = if body.is_empty() { + Vec::new() + } else { + alloc::vec![body.to_vec()] + }; + + let pending = Request::post(url, chunks) + .add_header("Content-Type", "application/x-protobuf") + .deadline(deadline) + .send() + .map_err(|_| Error::SendFailed)?; + + let response = pending + .try_wait(deadline) + .map_err(|_| Error::DeadlineReached)? + .map_err(|e| match e { + polkadot_sdk::sp_runtime::offchain::http::Error::DeadlineReached => { + Error::DeadlineReached + } + polkadot_sdk::sp_runtime::offchain::http::Error::IoError => Error::IoError, + polkadot_sdk::sp_runtime::offchain::http::Error::Unknown => Error::IoError, + })?; + + let status = response.code; + if status != 200 { + return Err(Error::Status(status)); + } + + Ok(response.body().collect()) +} diff --git a/pallets/prover_db_indexer/src/lib.rs b/pallets/prover_db_indexer/src/lib.rs new file mode 100644 index 00000000..8dc78901 --- /dev/null +++ b/pallets/prover_db_indexer/src/lib.rs @@ -0,0 +1,351 @@ +//! # Prover-Db Indexer Pallet +//! +//! Forwards table lifecycle and data events to an external prover-db +//! indexer (HTTP) using protobuf-over-HTTP. +//! +//! ## Architecture +//! +//! **Producer** (`on_finalize`, runs during block execution including sync): +//! Reads `polkadot_sdk::frame_system::Events`, extracts relevant variants, writes a +//! SCALE-encoded `BlockIndex` to the offchain DB keyed by block number. +//! +//! **Consumer** (`offchain_worker`, fires at chain tip only): +//! Walks the offchain DB from `cursor+1` to `current_block`, forwards each +//! entry via HTTP+protobuf, checkpoints on the server, deletes consumed +//! entries, advances cursor. +//! +//! ## Why two hooks? +//! +//! `on_finalize` can read events (they exist during block execution) but +//! cannot do HTTP. `offchain_worker` can do HTTP but cannot read past events +//! (cleared at next block). The offchain DB bridges them. +//! +//! ## No new host functions +//! +//! Raw DDL bytes (for schemas) and postcard-encoded `OnChainTable` bytes +//! (for row data — what `pallet_indexing::QuorumReached.data` carries, +//! after the chain has converted the indexer's Arrow IPC submission to +//! an augmented `OnChainTable` and postcard-serialized it in +//! `finalize_quorum`) are shipped as-is. The HTTP server decodes both. + +#![cfg_attr(not(feature = "std"), no_std)] + +extern crate alloc; + +#[cfg(test)] +mod mock; + +#[cfg(test)] +mod tests; + +mod http_client; +mod offchain_index; + +pub mod native_pallet; + +/// Generated protobuf types for the indexer HTTP adapter wire format. +#[allow(missing_docs, clippy::missing_docs_in_private_items)] +mod proto { + include!(concat!(env!("OUT_DIR"), "/io.spaceandtime.indexer.rs")); +} + +pub use pallet::*; + +/// Offchain local-storage key for the prover-db indexer HTTP endpoint URL. +pub const PROVER_DB_URL_KEY: &[u8] = b"prover_db_indexer::prover_db_url"; + +/// Dedup key column name sent with every `CreateTable` request. +const DEDUP_KEY_COLUMN: &str = "META_ROW_NUMBER"; + +/// Maximum blocks to process per OCW invocation. Controls catch-up speed. +/// At 100 blocks/invocation and 6s block time, catch-up is ~100x realtime. +const MAX_BLOCKS_PER_INVOCATION: u64 = 100; + +#[polkadot_sdk::frame_support::pallet] +#[allow(missing_docs, clippy::missing_docs_in_private_items, dead_code)] +pub mod pallet { + use alloc::format; + use alloc::string::String; + use alloc::vec::Vec; + + use polkadot_sdk::frame_support::pallet_prelude::*; + use polkadot_sdk::frame_system::pallet_prelude::*; + use sxt_core::tables::TableIdentifier; + + use crate::offchain_index::{BlockEvent, BlockIndex, CreateEntry, DataEntry}; + + #[pallet::pallet] + pub struct Pallet(_); + + #[pallet::config] + pub trait Config: + polkadot_sdk::frame_system::Config + pallet_tables::Config + pallet_indexing::Config + { + /// The runtime's overarching event type. Bounds let us downcast + /// it back to the per-pallet typed events without re-encoding, + /// using `TryInto` impls that `construct_runtime!` generates. + type RuntimeEvent: From> + + IsType<::RuntimeEvent> + + TryInto> + + TryInto>; + } + + #[pallet::event] + #[pallet::generate_deposit(pub(super) fn deposit_event)] + pub enum Event, I: 'static = ()> { + /// The offchain indexer successfully forwarded a block. + BlockForwarded { + /// Block number that was successfully forwarded. + block_number: u64, + }, + /// The offchain indexer encountered an error. + ForwardingError { + /// Block number that the forwarder failed on. + block_number: u64, + }, + } + + #[pallet::hooks] + impl, I: 'static> Hooks> for Pallet { + // ─── PRODUCER ─────────────────────────────────────────────────── + // Runs during block execution (including sync). Captures events + // and persists them to the offchain DB for the OCW to consume. + fn on_finalize(n: BlockNumberFor) { + let block_number: u64 = n.try_into().unwrap_or(0); + let index = Self::extract_block_index(); + + if index.is_empty() { + return; + } + + log::debug!( + target: "prover_db_indexer", + "on_finalize({}): writing {} events to offchain DB", + block_number, + index.events.len(), + ); + + crate::offchain_index::write(block_number, &index); + } + + // ─── CONSUMER ─────────────────────────────────────────────────── + // Fires at chain tip only (not during sync). Drains the offchain + // DB queue, forwards to the HTTP server, deletes consumed entries. + fn offchain_worker(_block_number: BlockNumberFor) { + if let Err(e) = Self::run_consumer() { + log::error!( + target: "prover_db_indexer", + "offchain indexer error: {:?}", + e, + ); + } + } + } + + impl, I: 'static> Pallet { + // ═══════════════════════════════════════════════════════════════ + // PRODUCER: extract events → BlockIndex (called from on_finalize) + // ═══════════════════════════════════════════════════════════════ + + fn extract_block_index() -> BlockIndex { + // Each frame-system event may produce zero or more `BlockEvent`s + // (a `pallet_tables::SchemaUpdated` carries N updates, an + // indexing quorum yields one, anything else yields nothing). + // The extractors are pure: they take an event and return what + // would be appended, leaving accumulation to the caller. + let events = polkadot_sdk::frame_system::Pallet::::read_events_no_consensus() + .flat_map(|record| { + let from_tables = Self::try_extract_table_event(record.event.clone()); + let from_indexing = Self::try_extract_indexing_event(record.event); + from_tables.into_iter().chain(from_indexing) + }) + .collect(); + BlockIndex { events } + } + + fn try_extract_table_event( + event: ::RuntimeEvent, + ) -> Vec { + // `event` is the frame-system RuntimeEvent. Bounce through our + // Config's RuntimeEvent (same type at runtime, bridged by `IsType`) + // so the `TryInto` bound applies. + let our_event = <>::RuntimeEvent as From<_>>::from(event); + let Ok(table_event): Result, _> = our_event.try_into() else { + return Vec::new(); + }; + match table_event { + pallet_tables::Event::SchemaUpdated(_who, update_list) => update_list + .iter() + .map(|update| { + BlockEvent::Create(CreateEntry { + ident: update.ident.clone(), + ddl: update.create_statement.to_vec(), + }) + }) + .collect(), + pallet_tables::Event::TablesCreatedWithCommitments { table_list, .. } => table_list + .iter() + .map(|req| { + BlockEvent::Create(CreateEntry { + ident: req.table_name.clone(), + ddl: req.ddl.to_vec(), + }) + }) + .collect(), + pallet_tables::Event::TableDropped(_who, _table_type, ident, _source) => { + alloc::vec![BlockEvent::Drop(ident)] + } + _ => Vec::new(), + } + } + + fn try_extract_indexing_event( + event: ::RuntimeEvent, + ) -> Option { + let our_event = <>::RuntimeEvent as From<_>>::from(event); + let indexing_event: pallet_indexing::Event = our_event.try_into().ok()?; + let pallet_indexing::Event::QuorumReached { quorum, data } = indexing_event else { + return None; + }; + Some(BlockEvent::Data(DataEntry { + table: quorum.table, + data: data.to_vec(), + })) + } + + // ═══════════════════════════════════════════════════════════════ + // CONSUMER: drain offchain DB → HTTP → delete (called from OCW) + // ═══════════════════════════════════════════════════════════════ + + fn run_consumer() -> Result<(), &'static str> { + use polkadot_sdk::sp_runtime::offchain::storage::StorageValueRef; + + // 1. Check if this node is configured as an indexer. + let url_ref = StorageValueRef::persistent(crate::PROVER_DB_URL_KEY); + let url_bytes: Option> = url_ref.get::>().ok().flatten(); + + let Some(url_bytes) = url_bytes else { + return Ok(()); + }; + + let url = + core::str::from_utf8(&url_bytes).map_err(|_| "invalid UTF-8 in indexer URL")?; + + // 2. Current block number. + let current_block: u64 = polkadot_sdk::frame_system::Pallet::::block_number() + .try_into() + .map_err(|_| "block number conversion failed")?; + + // 3. Ask the server for its last checkpoint. The server is the + // sole source of truth — no local cursor. This costs one HTTP + // round-trip per OCW invocation but guarantees we never + // diverge from what the server has actually committed. + let cursor: u64 = match crate::http_client::get_last_checkpoint(url) { + Ok(Some(seq)) => seq, + Ok(None) => 0, + Err(e) => { + log::warn!( + target: "prover_db_indexer", + "get_last_checkpoint failed: {}; skipping this round", + e, + ); + return Ok(()); + } + }; + + // 4. Walk blocks from cursor+1 to tip, capped. + let start = cursor + 1; + let end = core::cmp::min(current_block, start + crate::MAX_BLOCKS_PER_INVOCATION - 1); + + if start > current_block { + return Ok(()); + } + + log::debug!( + target: "prover_db_indexer", + "processing blocks {}..={} (server_checkpoint={}, tip={})", + start, end, cursor, current_block, + ); + + for block_num in start..=end { + // Read from offchain DB. + let entry = crate::offchain_index::read(block_num); + + match &entry { + Some(idx) if !idx.is_empty() => { + log::info!( + target: "prover_db_indexer", + "block {} — {} events to forward", + block_num, + idx.events.len(), + ); + Self::forward_block(url, block_num, idx)?; + } + _ => {} + } + + // Checkpoint on the server (always, even for empty blocks). + crate::http_client::checkpoint(url, block_num).map_err(|_| "checkpoint failed")?; + + // Delete consumed entry from offchain DB. + if entry.is_some() { + crate::offchain_index::clear(block_num); + } + } + + Ok(()) + } + + /// Forward a single block's events in deposit order. + fn forward_block( + url: &str, + block_num: u64, + index: &BlockIndex, + ) -> Result<(), &'static str> { + use crate::offchain_index::BlockEvent; + + let seq = block_num; + + for event in &index.events { + match event { + BlockEvent::Drop(ident) => { + let name = Self::fq_name(ident); + crate::http_client::drop_table(url, seq, name) + .map_err(|_| "drop_table failed")?; + } + BlockEvent::Create(entry) => { + let name = Self::fq_name(&entry.ident); + crate::http_client::create_table( + url, + seq, + name, + entry.ddl.clone(), + crate::DEDUP_KEY_COLUMN.into(), + ) + .map_err(|_| "create_table failed")?; + } + BlockEvent::Data(entry) => { + let name = Self::fq_name(&entry.table); + crate::http_client::put_batches( + url, + seq, + alloc::vec![crate::proto::TableBatch { + table_name: name, + record_batch: entry.data.clone(), + }], + ) + .map_err(|_| "put_batches failed")?; + } + } + } + + Ok(()) + } + + fn fq_name(id: &TableIdentifier) -> String { + let ns = core::str::from_utf8(&id.namespace).unwrap_or("?"); + let name = core::str::from_utf8(&id.name).unwrap_or("?"); + format!("{}.{}", ns, name) + } + } +} diff --git a/pallets/prover_db_indexer/src/mock.rs b/pallets/prover_db_indexer/src/mock.rs new file mode 100644 index 00000000..b9139ce0 --- /dev/null +++ b/pallets/prover_db_indexer/src/mock.rs @@ -0,0 +1,224 @@ +//! Mock runtime for testing the prover-db indexer pallet. +//! +//! Mirrors `pallet_indexing`'s mock because our `Config: pallet_indexing::Config` +//! supertrait pulls in the same transitive dependency chain (system_tables +//! → session/staking/balances/zkpay/timestamp). The mock exists purely to +//! satisfy those trait bounds at compile time — no test actually exercises +//! the producer side that consumes events from these pallets. + +use native_api::Api; +use polkadot_sdk::frame_election_provider_support::bounds::{ + ElectionBounds, ElectionBoundsBuilder, +}; +use polkadot_sdk::frame_election_provider_support::{onchain, SequentialPhragmen}; +use polkadot_sdk::frame_support::pallet_prelude::ConstU32; +use polkadot_sdk::frame_support::traits::ConstU128; +use polkadot_sdk::frame_support::{derive_impl, parameter_types}; +use polkadot_sdk::sp_core::{ConstU64, H256}; +use polkadot_sdk::sp_runtime::traits::{IdentityLookup, OpaqueKeys}; +use polkadot_sdk::sp_runtime::{BuildStorage, KeyTypeId}; +use polkadot_sdk::{ + frame_support, frame_system, pallet_balances, pallet_session, pallet_staking, + pallet_staking_reward_curve, pallet_timestamp, sp_core, sp_io, sp_runtime, sp_staking, +}; +use proof_of_sql_commitment_map::generic_over_commitment::ConcreteType; +use proof_of_sql_commitment_map::PerCommitmentScheme; + +use crate as pallet_prover_db_indexer; + +type Block = frame_system::mocking::MockBlock; + +frame_support::construct_runtime!( + pub enum Test + { + System: frame_system, + Indexing: pallet_indexing::native_pallet, + Permissions: pallet_permissions, + Commitments: pallet_commitments, + ZkPay: pallet_zkpay, + Tables: pallet_tables, + Session: pallet_session, + SystemTables: pallet_system_tables, + Balances: pallet_balances, + Staking: pallet_staking, + ProverDbIndexer: pallet_prover_db_indexer::native_pallet, + } +); + +type AccountId = sp_runtime::AccountId32; +type Nonce = u32; +type Balance = u128; + +#[derive_impl(frame_system::config_preludes::TestDefaultConfig)] +impl frame_system::Config for Test { + type Nonce = Nonce; + type AccountId = AccountId; + type AccountData = pallet_balances::AccountData; + type Block = Block; + type Lookup = IdentityLookup; + type Hash = H256; +} + +impl pallet_balances::Config for Test { + type AccountStore = System; + type Balance = Balance; + type RuntimeEvent = RuntimeEvent; + type RuntimeHoldReason = (); + type RuntimeFreezeReason = (); + type WeightInfo = (); + type DustRemoval = (); + type ExistentialDeposit = ConstU128<1>; + type ReserveIdentifier = (); + type FreezeIdentifier = (); + type MaxLocks = (); + type MaxReserves = (); + type MaxFreezes = (); +} + +impl pallet_timestamp::Config for Test { + type Moment = u64; + type OnTimestampSet = (); + type MinimumPeriod = ConstU64<5>; + type WeightInfo = (); +} + +pallet_staking_reward_curve::build! { + const I_NPOS: sp_runtime::curve::PiecewiseLinear<'static> = curve!( + min_inflation: 0_025_000, + max_inflation: 0_100_000, + ideal_stake: 0_500_000, + falloff: 0_050_000, + max_piece_count: 40, + test_precision: 0_005_000, + ); +} +parameter_types! { + pub const RewardCurve: &'static sp_runtime::curve::PiecewiseLinear<'static> = &I_NPOS; + pub static ElectionsBounds: ElectionBounds = ElectionBoundsBuilder::default().build(); +} + +pub struct OnChainSeqPhragmen; +impl onchain::Config for OnChainSeqPhragmen { + type System = Test; + type Solver = SequentialPhragmen; + type DataProvider = Staking; + type WeightInfo = (); + type MaxWinners = ConstU32<100>; + type Bounds = ElectionsBounds; +} + +#[derive_impl(pallet_staking::config_preludes::TestDefaultConfig)] +impl pallet_staking::Config for Test { + type Currency = Balances; + type CurrencyBalance = ::Balance; + type UnixTime = pallet_timestamp::Pallet; + type AdminOrigin = frame_system::EnsureRoot; + type SessionInterface = Self; + type EraPayout = pallet_staking::ConvertCurve; + type NextNewSession = Session; + type ElectionProvider = onchain::OnChainExecution; + type GenesisElectionProvider = Self::ElectionProvider; + type VoterList = pallet_staking::UseNominatorsAndValidatorsMap; + type TargetList = pallet_staking::UseValidatorsMap; + type CurrencyToVote = sp_staking::currency_to_vote::U128CurrencyToVote; + type NominationsQuota = pallet_staking::FixedNominationsQuota<16>; + type RewardRemainder = (); + type RuntimeEvent = RuntimeEvent; + type Slash = (); + type Reward = (); + type MaxControllersInDeprecationBatch = (); + type EventListeners = (); + type DisablingStrategy = pallet_staking::UpToLimitDisablingStrategy; + type WeightInfo = pallet_staking::weights::SubstrateWeight; +} + +impl pallet_indexing::pallet::Config for Test { + type RuntimeEvent = RuntimeEvent; + type WeightInfo = pallet_indexing::weights::SubstrateWeight; +} + +impl pallet_prover_db_indexer::pallet::Config for Test { + type RuntimeEvent = RuntimeEvent; +} + +pub type BlockNumber = u64; + +parameter_types! { + pub const Period: BlockNumber = 1; + pub const Offset: BlockNumber = 0; +} + +pub struct TestSessionHandler; +impl pallet_session::SessionHandler for TestSessionHandler { + const KEY_TYPE_IDS: &'static [KeyTypeId] = &[sp_core::crypto::key_types::DUMMY]; + + fn on_new_session( + _changed: bool, + _validators: &[(AccountId, Ks)], + _queued_validators: &[(AccountId, Ks)], + ) { + } + + fn on_disabled(_validator_index: u32) {} + + fn on_genesis_session(_validators: &[(AccountId, Ks)]) {} +} + +impl pallet_session::Config for Test { + type SessionManager = (); + type Keys = sp_runtime::testing::UintAuthorityId; + type ShouldEndSession = pallet_session::PeriodicSessions; + type SessionHandler = TestSessionHandler; + type RuntimeEvent = RuntimeEvent; + type ValidatorId = AccountId; + type ValidatorIdOf = pallet_staking::StashOf; + type NextSessionRotation = pallet_session::PeriodicSessions; + type WeightInfo = (); +} +impl pallet_session::historical::Config for Test { + type FullIdentification = pallet_staking::Exposure; + type FullIdentificationOf = pallet_staking::ExposureOf; +} + +sp_runtime::impl_opaque_keys! { + pub struct SessionKeys { + pub foo: sp_runtime::testing::UintAuthorityId, + } +} + +impl pallet_zkpay::Config for Test { + type RuntimeEvent = RuntimeEvent; +} + +impl pallet_system_tables::Config for Test { + type RuntimeEvent = RuntimeEvent; +} + +impl pallet_tables::Config for Test { + type RuntimeEvent = RuntimeEvent; + type WeightInfo = (); +} + +impl pallet_permissions::Config for Test { + type RuntimeEvent = RuntimeEvent; + type WeightInfo = (); +} + +impl pallet_commitments::Config for Test { + const END_ROW_LIMITS_PER_SCHEME: PerCommitmentScheme> = PerCommitmentScheme { + hyper_kzg: 4, + dynamic_dory: 4, + }; +} + +pub fn new_test_ext() -> sp_io::TestExternalities { + let mut storage = frame_system::GenesisConfig::::default() + .build_storage() + .unwrap(); + + pallet_commitments::GenesisConfig::::default() + .assimilate_storage(&mut storage) + .unwrap(); + + storage.into() +} diff --git a/pallets/prover_db_indexer/src/native_pallet.rs b/pallets/prover_db_indexer/src/native_pallet.rs new file mode 100644 index 00000000..609dc9a2 --- /dev/null +++ b/pallets/prover_db_indexer/src/native_pallet.rs @@ -0,0 +1,16 @@ +//! Type aliases that pin our pallet's instance parameter to +//! `native_api::Api`, so `construct_runtime!` in the runtime can refer +//! to a single-type-parameter `Pallet` form. + +use native_api::Api; + +/// Wrap the pallet type to use the `Api` instance. +pub type Pallet = crate::pallet::Pallet; + +/// Wrap the event type to use the `Api` instance. +pub type Event = crate::pallet::Event; + +pub use crate::pallet::{ + __substrate_call_check, __substrate_event_check, tt_default_parts, tt_default_parts_v2, + tt_error_token, +}; diff --git a/pallets/prover_db_indexer/src/offchain_index.rs b/pallets/prover_db_indexer/src/offchain_index.rs new file mode 100644 index 00000000..990de342 --- /dev/null +++ b/pallets/prover_db_indexer/src/offchain_index.rs @@ -0,0 +1,83 @@ +//! Block-indexed offchain storage for durable event forwarding. +//! +//! `on_finalize` writes a `BlockIndex` entry per block (if non-empty). +//! The OCW reads, forwards, and deletes consumed entries. +//! Events are stored in deposit order to preserve intra-block semantics. + +use alloc::vec::Vec; + +use codec::{Decode, Encode}; +use sxt_core::tables::TableIdentifier; + +/// Key prefix for block-indexed entries in the offchain DB. +const PREFIX: &[u8] = b"prover_db_indexer::block::"; + +/// Compute the offchain DB key for a given block number. +pub fn key_for_block(block: u64) -> Vec { + let mut k = PREFIX.to_vec(); + k.extend_from_slice(&block.to_be_bytes()); + k +} + +/// A table-creation event. +#[derive(Encode, Decode, Debug, Clone)] +pub struct CreateEntry { + pub ident: TableIdentifier, + pub ddl: Vec, +} + +/// A data-quorum event. +#[derive(Encode, Decode, Debug, Clone)] +pub struct DataEntry { + pub table: TableIdentifier, + pub data: Vec, +} + +/// A single event captured during block execution. Stored in the order +/// events were deposited so the OCW replays them in the correct sequence. +#[derive(Encode, Decode, Debug, Clone)] +pub enum BlockEvent { + /// Table created or schema updated. + Create(CreateEntry), + /// Table dropped. + Drop(TableIdentifier), + /// Data quorum reached (finalized row data). + Data(DataEntry), +} + +/// All relevant events from a single block, in deposit order. +#[derive(Encode, Decode, Debug, Clone, Default)] +pub struct BlockIndex { + pub events: Vec, +} + +impl BlockIndex { + pub fn is_empty(&self) -> bool { + self.events.is_empty() + } +} + +/// Write a BlockIndex to the offchain DB. Called from `on_finalize`. +pub fn write(block: u64, index: &BlockIndex) { + let key = key_for_block(block); + polkadot_sdk::sp_io::offchain_index::set(&key, &index.encode()); +} + +/// Read a BlockIndex from the offchain DB. Called from the OCW. +pub fn read(block: u64) -> Option { + let key = key_for_block(block); + let raw = polkadot_sdk::sp_io::offchain::local_storage_get( + polkadot_sdk::sp_core::offchain::StorageKind::PERSISTENT, + &key, + )?; + BlockIndex::decode(&mut &raw[..]).ok() +} + +/// Delete a consumed entry. Called from the OCW after forwarding. +pub fn clear(block: u64) { + let key = key_for_block(block); + polkadot_sdk::sp_io::offchain::local_storage_clear( + polkadot_sdk::sp_core::offchain::StorageKind::PERSISTENT, + &key, + ); +} diff --git a/pallets/prover_db_indexer/src/tests.rs b/pallets/prover_db_indexer/src/tests.rs new file mode 100644 index 00000000..8933d3ff --- /dev/null +++ b/pallets/prover_db_indexer/src/tests.rs @@ -0,0 +1,306 @@ +//! Integration tests for the block forwarder pallet. +//! +//! The OCW always asks the server for its last checkpoint (no local cursor). +//! Tests pre-populate the offchain DB as if on_finalize had written entries, +//! then verify the OCW reads, forwards in order, and deletes consumed entries. + +use codec::Encode; +use polkadot_sdk::frame_support::traits::Hooks; +use polkadot_sdk::sp_core::offchain::testing::{PendingRequest, TestOffchainExt}; +use polkadot_sdk::sp_core::offchain::{OffchainDbExt, OffchainStorage, OffchainWorkerExt}; +use polkadot_sdk::sp_runtime::BoundedVec; +use prost::Message; +use sxt_core::tables::TableIdentifier; + +use crate::mock::*; +use crate::{offchain_index, proto, PROVER_DB_URL_KEY}; + +type StateArc = + std::sync::Arc>; + +const MOCK_URL: &str = "http://127.0.0.1:9999"; + +fn encode_url() -> Vec { + codec::Encode::encode(&MOCK_URL.as_bytes().to_vec()) +} + +fn table_id(namespace: &str, name: &str) -> TableIdentifier { + TableIdentifier { + namespace: BoundedVec::try_from(namespace.as_bytes().to_vec()).unwrap(), + name: BoundedVec::try_from(name.as_bytes().to_vec()).unwrap(), + } +} + +fn expected_request(path: &str, body: Vec, response: Vec) -> PendingRequest { + PendingRequest { + method: "POST".into(), + uri: format!("{}{}", MOCK_URL, path), + headers: vec![("Content-Type".into(), "application/x-protobuf".into())], + body, + sent: true, + response: Some(response), + ..Default::default() + } +} + +fn checkpoint_at_response(seq: u64) -> Vec { + proto::GetLastCheckpointResponse { + sequence_number: seq, + has_checkpoint: true, + } + .encode_to_vec() +} + +fn no_checkpoint_response() -> Vec { + proto::GetLastCheckpointResponse { + sequence_number: 0, + has_checkpoint: false, + } + .encode_to_vec() +} + +fn setup_with_url() -> (polkadot_sdk::sp_io::TestExternalities, StateArc) { + let mut ext = new_test_ext(); + let (offchain, state) = TestOffchainExt::new(); + ext.register_extension(OffchainWorkerExt::new(offchain.clone())); + ext.register_extension(OffchainDbExt::new(offchain)); + state + .write() + .persistent_storage + .set(b"", PROVER_DB_URL_KEY, &encode_url()); + (ext, state) +} + +// ─── Tests ────────────────────────────────────────────────────────────── + +#[test] +fn ocw_skips_when_not_configured() { + let mut ext = new_test_ext(); + let (offchain, _) = TestOffchainExt::new(); + ext.register_extension(OffchainWorkerExt::new(offchain.clone())); + ext.register_extension(OffchainDbExt::new(offchain)); + ext.execute_with(|| { + System::set_block_number(1); + ProverDbIndexer::offchain_worker(1); + }); +} + +#[test] +fn ocw_forwards_and_deletes_offchain_entry() { + let (mut ext, state) = setup_with_url(); + + let ddl = b"CREATE TABLE PUBLIC.USERS (ID BIGINT NOT NULL)"; + let index = offchain_index::BlockIndex { + events: vec![offchain_index::BlockEvent::Create( + offchain_index::CreateEntry { + ident: table_id("PUBLIC", "USERS"), + ddl: ddl.to_vec(), + }, + )], + }; + let key = offchain_index::key_for_block(1); + state + .write() + .persistent_storage + .set(b"", &key, &index.encode()); + + { + let mut s = state.write(); + // 1. get_last_checkpoint → server at block 0 + s.expect_request(expected_request( + "/v1/get_last_checkpoint", + vec![], + no_checkpoint_response(), + )); + // 2. create_table for block 1 + s.expect_request(expected_request( + "/v1/create_table", + proto::CreateTableRequest { + sequence_number: 1, + table_name: "PUBLIC.USERS".into(), + arrow_schema: ddl.to_vec(), + key: "META_ROW_NUMBER".into(), + } + .encode_to_vec(), + vec![], + )); + // 3. checkpoint(1) + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(), + vec![], + )); + } + + ext.execute_with(|| { + System::set_block_number(1); + ProverDbIndexer::offchain_worker(1); + }); + + // Verify offchain entry was deleted. + let s = state.read(); + assert!( + s.persistent_storage.get(&key).is_none(), + "consumed entry should be deleted" + ); +} + +#[test] +fn ocw_checkpoints_empty_blocks() { + let (mut ext, state) = setup_with_url(); + + { + let mut s = state.write(); + // Server at block 0. + s.expect_request(expected_request( + "/v1/get_last_checkpoint", + vec![], + no_checkpoint_response(), + )); + // Empty block 1 → just checkpoint. + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(), + vec![], + )); + } + + ext.execute_with(|| { + System::set_block_number(1); + ProverDbIndexer::offchain_worker(1); + }); +} + +#[test] +fn ocw_resumes_from_server_checkpoint() { + let (mut ext, state) = setup_with_url(); + + // Server already at block 5. Block 6 has a drop event. + let index = offchain_index::BlockIndex { + events: vec![offchain_index::BlockEvent::Drop(table_id("NS", "OLD"))], + }; + state + .write() + .persistent_storage + .set(b"", &offchain_index::key_for_block(6), &index.encode()); + + { + let mut s = state.write(); + s.expect_request(expected_request( + "/v1/get_last_checkpoint", + vec![], + checkpoint_at_response(5), + )); + s.expect_request(expected_request( + "/v1/drop_table", + proto::DropTableRequest { + sequence_number: 6, + table_name: "NS.OLD".into(), + } + .encode_to_vec(), + vec![], + )); + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 6 }.encode_to_vec(), + vec![], + )); + } + + ext.execute_with(|| { + System::set_block_number(6); + ProverDbIndexer::offchain_worker(6); + }); +} + +#[test] +fn ocw_processes_multiple_blocks_in_order() { + let (mut ext, state) = setup_with_url(); + + // Block 1: drop T1. Block 2: create T2. Block 3: empty. + state.write().persistent_storage.set( + b"", + &offchain_index::key_for_block(1), + &offchain_index::BlockIndex { + events: vec![offchain_index::BlockEvent::Drop(table_id("NS", "T1"))], + } + .encode(), + ); + state.write().persistent_storage.set( + b"", + &offchain_index::key_for_block(2), + &offchain_index::BlockIndex { + events: vec![offchain_index::BlockEvent::Create( + offchain_index::CreateEntry { + ident: table_id("NS", "T2"), + ddl: b"CREATE TABLE NS.T2 (X INT NOT NULL)".to_vec(), + }, + )], + } + .encode(), + ); + + { + let mut s = state.write(); + // get_last_checkpoint → server at 0 + s.expect_request(expected_request( + "/v1/get_last_checkpoint", + vec![], + no_checkpoint_response(), + )); + // Block 1: drop + checkpoint + s.expect_request(expected_request( + "/v1/drop_table", + proto::DropTableRequest { + sequence_number: 1, + table_name: "NS.T1".into(), + } + .encode_to_vec(), + vec![], + )); + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 1 }.encode_to_vec(), + vec![], + )); + // Block 2: create + checkpoint + s.expect_request(expected_request( + "/v1/create_table", + proto::CreateTableRequest { + sequence_number: 2, + table_name: "NS.T2".into(), + arrow_schema: b"CREATE TABLE NS.T2 (X INT NOT NULL)".to_vec(), + key: "META_ROW_NUMBER".into(), + } + .encode_to_vec(), + vec![], + )); + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 2 }.encode_to_vec(), + vec![], + )); + // Block 3: empty, just checkpoint + s.expect_request(expected_request( + "/v1/checkpoint", + proto::CheckpointRequest { sequence_number: 3 }.encode_to_vec(), + vec![], + )); + } + + ext.execute_with(|| { + System::set_block_number(3); + ProverDbIndexer::offchain_worker(3); + }); + + // Both consumed entries should be deleted. + let s = state.read(); + assert!(s + .persistent_storage + .get(&offchain_index::key_for_block(1)) + .is_none()); + assert!(s + .persistent_storage + .get(&offchain_index::key_for_block(2)) + .is_none()); +} diff --git a/runtime/Cargo.toml b/runtime/Cargo.toml index 9da65324..00d23e6f 100644 --- a/runtime/Cargo.toml +++ b/runtime/Cargo.toml @@ -85,6 +85,7 @@ pallet-zkpay.workspace = true pallet-rewards.workspace = true pallet-system-tables.workspace = true pallet-system-contracts.workspace = true +pallet-prover-db-indexer.workspace = true proof-of-sql-commitment-map.workspace = true sxt-core.workspace = true @@ -101,6 +102,7 @@ std = [ "pallet-keystore/std", "pallet-attestation/std", "pallet-indexing/std", + "pallet-prover-db-indexer/std", "pallet-tables/std", "pallet-permissions/std", "pallet-system-tables/std", diff --git a/runtime/src/lib.rs b/runtime/src/lib.rs index cbca130d..08ea2d20 100644 --- a/runtime/src/lib.rs +++ b/runtime/src/lib.rs @@ -15,37 +15,22 @@ use alloc::vec; use alloc::vec::Vec; use polkadot_sdk::frame_election_provider_support::{ - generate_solution_type, - onchain, - SequentialPhragmen, + generate_solution_type, onchain, SequentialPhragmen, }; use polkadot_sdk::frame_support::dispatch::DispatchClass; use polkadot_sdk::frame_support::genesis_builder_helper::{build_state, get_preset}; use polkadot_sdk::frame_support::traits::VariantCountOf; pub use polkadot_sdk::frame_support::traits::{ - ConstBool, - ConstU128, - ConstU32, - ConstU64, - ConstU8, - Currency, - KeyOwnerProofSystem, - Randomness, + ConstBool, ConstU128, ConstU32, ConstU64, ConstU8, Currency, KeyOwnerProofSystem, Randomness, StorageInfo, }; pub use polkadot_sdk::frame_support::weights::constants::{ - BlockExecutionWeight, - ExtrinsicBaseWeight, - RocksDbWeight, - WEIGHT_REF_TIME_PER_MILLIS, + BlockExecutionWeight, ExtrinsicBaseWeight, RocksDbWeight, WEIGHT_REF_TIME_PER_MILLIS, }; use polkadot_sdk::frame_support::weights::ConstantMultiplier; pub use polkadot_sdk::frame_support::weights::{IdentityFee, Weight}; pub use polkadot_sdk::frame_support::{ - construct_runtime, - derive_impl, - parameter_types, - StorageValue, + construct_runtime, derive_impl, parameter_types, StorageValue, }; pub use polkadot_sdk::frame_system::Call as SystemCall; use polkadot_sdk::frame_system::EnsureRoot; @@ -63,31 +48,17 @@ use polkadot_sdk::sp_consensus_babe::AuthorityId as BabeId; use polkadot_sdk::sp_core::crypto::KeyTypeId; use polkadot_sdk::sp_core::OpaqueMetadata; use polkadot_sdk::sp_runtime::traits::{ - AccountIdLookup, - BlakeTwo256, - Block as BlockT, - Bounded, - IdentifyAccount, - NumberFor, - OpaqueKeys, - Verify, - Zero, + AccountIdLookup, BlakeTwo256, Block as BlockT, Bounded, IdentifyAccount, NumberFor, OpaqueKeys, + Verify, Zero, }; use polkadot_sdk::sp_runtime::transaction_validity::{ - TransactionPriority, - TransactionSource, - TransactionValidity, + TransactionPriority, TransactionSource, TransactionValidity, }; #[cfg(any(feature = "std", test))] pub use polkadot_sdk::sp_runtime::BuildStorage; use polkadot_sdk::sp_runtime::{ - create_runtime_str, - generic, - impl_opaque_keys, - ApplyExtrinsicResult, - FixedPointNumber, - FixedU128, - Perquintill, + create_runtime_str, generic, impl_opaque_keys, ApplyExtrinsicResult, FixedPointNumber, + FixedU128, Perquintill, }; pub use polkadot_sdk::sp_runtime::{Perbill, Percent, Permill}; use polkadot_sdk::sp_staking::SessionIndex; @@ -95,59 +66,23 @@ use polkadot_sdk::sp_staking::SessionIndex; use polkadot_sdk::sp_version::NativeVersion; use polkadot_sdk::sp_version::RuntimeVersion; use polkadot_sdk::{ - frame_support, - frame_system, - frame_system_rpc_runtime_api, - pallet_authority_discovery, - pallet_authorship, - pallet_babe, - pallet_bags_list, - pallet_balances, - pallet_election_provider_multi_phase, - pallet_grandpa, - pallet_im_online, - pallet_migrations, - pallet_multisig, - pallet_offences, - pallet_session, - pallet_staking, - pallet_staking_runtime_api, - pallet_statement, - pallet_sudo, - pallet_timestamp, - pallet_transaction_payment, - pallet_transaction_payment_rpc_runtime_api, - pallet_utility, - sp_api, - sp_authority_discovery, - sp_block_builder, - sp_consensus_babe, - sp_consensus_grandpa, - sp_core, - sp_genesis_builder, - sp_inherents, - sp_offchain, - sp_runtime, - sp_session, - sp_staking, - sp_statement_store, - sp_transaction_pool, - sp_version, + frame_support, frame_system, frame_system_rpc_runtime_api, pallet_authority_discovery, + pallet_authorship, pallet_babe, pallet_bags_list, pallet_balances, + pallet_election_provider_multi_phase, pallet_grandpa, pallet_im_online, pallet_migrations, + pallet_multisig, pallet_offences, pallet_session, pallet_staking, pallet_staking_runtime_api, + pallet_statement, pallet_sudo, pallet_timestamp, pallet_transaction_payment, + pallet_transaction_payment_rpc_runtime_api, pallet_utility, sp_api, sp_authority_discovery, + sp_block_builder, sp_consensus_babe, sp_consensus_grandpa, sp_core, sp_genesis_builder, + sp_inherents, sp_offchain, sp_runtime, sp_session, sp_staking, sp_statement_store, + sp_transaction_pool, sp_version, }; use proof_of_sql_commitment_map::generic_over_commitment::ConcreteType; use proof_of_sql_commitment_map::PerCommitmentScheme; use sxt_core::system_tables::ClaimedUnstake; pub use { - pallet_attestation, - pallet_commitments, - pallet_indexing, - pallet_keystore, - pallet_permissions, - pallet_rewards, - pallet_smartcontracts, - pallet_system_contracts, - pallet_system_tables, - pallet_tables, + pallet_attestation, pallet_commitments, pallet_indexing, pallet_keystore, pallet_permissions, + pallet_prover_db_indexer, pallet_rewards, pallet_smartcontracts, pallet_system_contracts, + pallet_system_tables, pallet_tables, }; /// An index to a block. @@ -891,6 +826,10 @@ impl pallet_rewards::Config for Runtime { type MaxPayoutsPerBlock = ConstU32<3>; } +impl pallet_prover_db_indexer::Config for Runtime { + type RuntimeEvent = RuntimeEvent; +} + #[cfg(feature = "runtime-benchmarks")] impl frame_system_benchmarking::Config for Runtime {} @@ -999,6 +938,8 @@ mod runtime { pub type Rewards = pallet_rewards; #[runtime::pallet_index(110)] pub type ZkPay = pallet_zkpay; + #[runtime::pallet_index(111)] + pub type ProverDbIndexer = pallet_prover_db_indexer::native_pallet::Pallet; } /// The address format for describing accounts.