diff --git a/src/blocks.rs b/src/blocks.rs index b565039..1dc88ee 100644 --- a/src/blocks.rs +++ b/src/blocks.rs @@ -1,104 +1,124 @@ //! Approximate block-number → wall-clock-time conversion for supported chains. //! -//! The contract Parquet schema does not store block timestamps, so the -//! dashboard derives them from `block_number`. Ethereum uses piecewise-linear -//! interpolation before the Merge and exact slots after. Gnosis uses calibrated -//! checkpoints because its long-run average block cadence has drifted enough -//! that a plain 5-second-from-genesis estimate mislabels recent blocks by months. +//! Blink loads sparse, exact `(block_number, unix_timestamp)` checkpoints from DuckDB, +//! interpolates between them, and extrapolates from the latest measured rate. + +use std::{ + collections::HashMap, + sync::{OnceLock, RwLock}, +}; use chrono::{DateTime, TimeZone, Utc}; use crate::chains::GNOSIS_CHAIN_ID; -/// Hardcoded (block_number, unix_timestamp) checkpoints for Ethereum mainnet. -/// Used to interpolate timestamps without an extra timestamp lookup table. -const CHECKPOINTS: &[(u64, i64)] = &[ - (0, 1_438_269_988), // 2015-07-30 genesis - (200_000, 1_443_534_600), // 2015-09-29 - (1_000_000, 1_455_404_488), // 2016-02-13 - (2_000_000, 1_470_173_578), // 2016-08-02 - (3_000_000, 1_484_802_716), // 2017-01-19 - (4_000_000, 1_499_633_567), // 2017-07-09 - (4_370_000, 1_508_131_331), // 2017-10-16 byzantium - (5_000_000, 1_517_319_693), // 2018-01-30 - (6_000_000, 1_532_118_564), // 2018-07-21 - (7_000_000, 1_546_466_492), // 2019-01-02 - (7_280_000, 1_551_383_524), // 2019-02-28 constantinople - (8_000_000, 1_561_100_149), // 2019-06-21 - (9_000_000, 1_574_706_444), // 2019-11-25 - (10_000_000, 1_588_598_533), // 2020-05-04 - (11_000_000, 1_602_667_372), // 2020-10-14 - (12_000_000, 1_617_270_478), // 2021-04-01 - (12_244_000, 1_618_481_223), // 2021-04-15 berlin - (12_965_000, 1_628_166_822), // 2021-08-05 london - (13_000_000, 1_628_643_581), // 2021-08-12 - (14_000_000, 1_642_114_795), // 2022-01-13 - (15_000_000, 1_656_586_444), // 2022-06-30 - (15_537_393, 1_663_224_162), // 2022-09-15 merge -]; +pub type BlockCheckpoint = (u64, i64); +pub type ChainCheckpoints = HashMap>; -const POST_MERGE_BLOCK: u64 = 15_537_393; -const POST_MERGE_TIMESTAMP: i64 = 1_663_224_162; -const SECS_PER_BLOCK_POST_MERGE: i64 = 12; +static RUNTIME_CHECKPOINTS: OnceLock> = OnceLock::new(); -const GNOSIS_GENESIS_TIMESTAMP: i64 = 1_539_024_180; -const GNOSIS_SECS_PER_BLOCK: i64 = 5; +fn checkpoint_store() -> &'static RwLock { + RUNTIME_CHECKPOINTS.get_or_init(|| RwLock::new(HashMap::new())) +} + +#[doc(hidden)] +pub fn replace_runtime_checkpoints(checkpoints: ChainCheckpoints) { + if let Ok(mut store) = checkpoint_store().write() { + *store = checkpoints; + } +} -const GNOSIS_CHECKPOINTS: &[(u64, i64)] = &[ - (0, GNOSIS_GENESIS_TIMESTAMP), // 2018-10-08 xDai/Gnosis genesis - (44_212_810, 1_768_666_350), // 2026-01-17 16:12:30 - (46_762_380, 1_781_798_438), // 2026-06-18 16:00:38 -]; +pub(crate) fn upsert_runtime_checkpoint(chain_id: u64, block_number: u64, timestamp: i64) { + if let Ok(mut store) = checkpoint_store().write() { + let chain = store.entry(chain_id).or_default(); + match chain.binary_search_by_key(&block_number, |(block, _)| *block) { + Ok(index) => chain[index].1 = timestamp, + Err(index) => chain.insert(index, (block_number, timestamp)), + } + } +} + +/// Ideal slot times, used only to size chart buckets — the timestamp math +/// uses measured checkpoint rates instead. +const SECS_PER_BLOCK_POST_MERGE: i64 = 12; +const POST_MERGE_BLOCK: u64 = 15_537_393; +const GNOSIS_SECS_PER_BLOCK: i64 = 5; /// Approximate the block timestamp for a given chain and block number. pub fn block_timestamp(chain_id: u64, block_number: u64) -> DateTime { - let secs = match chain_id { - GNOSIS_CHAIN_ID => gnosis_block_timestamp_secs(block_number), - _ => ethereum_block_timestamp_secs(block_number), - }; + let checkpoints = checkpoint_store() + .read() + .ok() + .and_then(|store| store.get(&chain_id).cloned()) + .unwrap_or_default(); + let secs = checkpoint_timestamp_secs(chain_id, &checkpoints, block_number); Utc.timestamp_opt(secs, 0).single().unwrap_or_else(Utc::now) } /// Approximate the block number for a given chain and timestamp. pub fn block_number_at_time(chain_id: u64, timestamp: DateTime) -> u64 { - match chain_id { - GNOSIS_CHAIN_ID => gnosis_block_number_at_time(timestamp.timestamp()), - _ => ethereum_block_number_at_time(timestamp.timestamp()), - } + let checkpoints = checkpoint_store() + .read() + .ok() + .and_then(|store| store.get(&chain_id).cloned()) + .unwrap_or_default(); + checkpoint_block_at_time(chain_id, &checkpoints, timestamp.timestamp()) } -fn gnosis_block_timestamp_secs(block_number: u64) -> i64 { - let (last_block, last_timestamp) = GNOSIS_CHECKPOINTS[GNOSIS_CHECKPOINTS.len() - 1]; - if block_number >= last_block { - return last_timestamp + (block_number - last_block) as i64 * GNOSIS_SECS_PER_BLOCK; +/// Milliseconds per block over the final checkpoint span — the measured +/// recent rate, used to extrapolate past the newest checkpoint. +fn fallback_ms_per_block(chain_id: u64, block_number: u64) -> i128 { + match chain_id { + GNOSIS_CHAIN_ID => GNOSIS_SECS_PER_BLOCK as i128 * 1000, + _ if block_number >= POST_MERGE_BLOCK => SECS_PER_BLOCK_POST_MERGE as i128 * 1000, + _ => 14_000, } - interpolate_checkpoint_timestamp(GNOSIS_CHECKPOINTS, block_number) } -fn gnosis_block_number_at_time(timestamp: i64) -> u64 { - let (last_block, last_timestamp) = GNOSIS_CHECKPOINTS[GNOSIS_CHECKPOINTS.len() - 1]; - if timestamp >= last_timestamp { - let blocks = (timestamp - last_timestamp) / GNOSIS_SECS_PER_BLOCK; - return last_block + blocks.max(0) as u64; +fn trailing_ms_per_block( + chain_id: u64, + checkpoints: &[BlockCheckpoint], + block_number: u64, +) -> i128 { + if checkpoints.len() < 2 { + return fallback_ms_per_block(chain_id, block_number); } - interpolate_checkpoint_block(GNOSIS_CHECKPOINTS, timestamp) + let (b1, t1) = checkpoints[checkpoints.len() - 1]; + let (b0, t0) = checkpoints[checkpoints.len() - 2]; + if b1 <= b0 || t1 <= t0 { + return fallback_ms_per_block(chain_id, block_number); + } + ((t1 - t0) as i128 * 1000) / (b1 - b0) as i128 } -fn ethereum_block_timestamp_secs(block_number: u64) -> i64 { - if block_number >= POST_MERGE_BLOCK { - POST_MERGE_TIMESTAMP + (block_number - POST_MERGE_BLOCK) as i64 * SECS_PER_BLOCK_POST_MERGE - } else { - interpolate_checkpoint_timestamp(CHECKPOINTS, block_number) +fn checkpoint_timestamp_secs( + chain_id: u64, + checkpoints: &[BlockCheckpoint], + block_number: u64, +) -> i64 { + if checkpoints.is_empty() { + return 0; + } + let (last_block, last_timestamp) = checkpoints[checkpoints.len() - 1]; + if block_number >= last_block { + let ms = (block_number - last_block) as i128 + * trailing_ms_per_block(chain_id, checkpoints, block_number); + return last_timestamp + (ms / 1000) as i64; } + interpolate_checkpoint_timestamp(checkpoints, block_number) } -fn ethereum_block_number_at_time(timestamp: i64) -> u64 { - if timestamp >= POST_MERGE_TIMESTAMP { - let blocks = (timestamp - POST_MERGE_TIMESTAMP) / SECS_PER_BLOCK_POST_MERGE; - return POST_MERGE_BLOCK + blocks.max(0) as u64; +fn checkpoint_block_at_time(chain_id: u64, checkpoints: &[BlockCheckpoint], timestamp: i64) -> u64 { + if checkpoints.is_empty() { + return 0; } - - interpolate_checkpoint_block(CHECKPOINTS, timestamp) + let (last_block, last_timestamp) = checkpoints[checkpoints.len() - 1]; + if timestamp >= last_timestamp { + let blocks = (timestamp - last_timestamp) as i128 * 1000 + / trailing_ms_per_block(chain_id, checkpoints, last_block); + return last_block + blocks.max(0) as u64; + } + interpolate_checkpoint_block(checkpoints, timestamp) } fn interpolate_checkpoint_timestamp(checkpoints: &[(u64, i64)], block_number: u64) -> i64 { diff --git a/src/checkpoints.rs b/src/checkpoints.rs new file mode 100644 index 0000000..59faf12 --- /dev/null +++ b/src/checkpoints.rs @@ -0,0 +1,266 @@ +//! Persistent block-time checkpoints and the `blink checkpoints` bootstrap. + +use std::{ + collections::{HashMap, HashSet}, + fs, + path::Path, + time::Duration, +}; + +use alloy::providers::{Provider, ProviderBuilder}; +use anyhow::{anyhow, Context, Result}; +use duckdb::{params, Connection}; + +use crate::{ + blocks::{self, ChainCheckpoints}, + cli::CheckpointsArgs, + extract::batch::BatchClient, + util::{format_count, print_header, print_kv, print_kv_accent}, +}; + +pub(crate) fn ensure_schema(conn: &Connection) -> Result<()> { + conn.execute_batch( + r#" + CREATE TABLE IF NOT EXISTS chain_block_checkpoints ( + chain_id UBIGINT NOT NULL, + block_number UBIGINT NOT NULL, + block_timestamp BIGINT NOT NULL, + source VARCHAR NOT NULL, + observed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (chain_id, block_number) + ); + "#, + ) + .context("create block checkpoint schema") +} + +pub(crate) fn load(conn: &Connection) -> Result { + if !crate::db::table_exists(conn, "chain_block_checkpoints")? { + return Ok(HashMap::new()); + } + let mut stmt = conn.prepare( + r#" + SELECT chain_id, block_number, block_timestamp + FROM chain_block_checkpoints + ORDER BY chain_id, block_number + "#, + )?; + let rows = stmt.query_map([], |row| { + Ok(( + row.get::<_, u64>(0)?, + row.get::<_, u64>(1)?, + row.get::<_, i64>(2)?, + )) + })?; + let mut checkpoints: ChainCheckpoints = HashMap::new(); + for row in rows { + let (chain_id, block_number, timestamp) = row?; + checkpoints + .entry(chain_id) + .or_default() + .push((block_number, timestamp)); + } + Ok(checkpoints) +} + +pub(crate) fn load_runtime(conn: &Connection) -> Result { + let checkpoints = load(conn)?; + let count = checkpoints.values().map(Vec::len).sum(); + blocks::replace_runtime_checkpoints(checkpoints); + Ok(count) +} + +pub(crate) fn upsert( + conn: &Connection, + chain_id: u64, + block_number: u64, + timestamp: i64, +) -> Result<()> { + conn.execute( + r#" + INSERT INTO chain_block_checkpoints ( + chain_id, block_number, block_timestamp, source, observed_at + ) VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP) + ON CONFLICT (chain_id, block_number) DO UPDATE SET + block_timestamp = excluded.block_timestamp, + source = excluded.source, + observed_at = excluded.observed_at + "#, + params![chain_id, block_number, timestamp, "rpc"], + )?; + Ok(()) +} + +fn persist_batch(db_path: &Path, chain_id: u64, checkpoints: &[(u64, i64)]) -> Result<()> { + let conn = + Connection::open(db_path).with_context(|| format!("open duckdb {}", db_path.display()))?; + ensure_schema(&conn)?; + conn.execute_batch("BEGIN")?; + let result = checkpoints + .iter() + .try_for_each(|(block, timestamp)| upsert(&conn, chain_id, *block, *timestamp)); + match result { + Ok(()) => conn.execute_batch("COMMIT").map_err(Into::into), + Err(error) => { + let _ = conn.execute_batch("ROLLBACK"); + Err(error).context("persist block checkpoint batch") + } + } +} + +fn indexed_chain_ranges(conn: &Connection) -> Result> { + if !crate::db::table_exists(conn, "rollup_block_counts")? { + return Ok(HashMap::new()); + } + let mut stmt = conn.prepare( + r#" + SELECT chain_id, MIN(block_number), MAX(block_number) + FROM rollup_block_counts + GROUP BY chain_id + "#, + )?; + let rows = stmt.query_map([], |row| { + Ok(( + row.get::<_, u64>(0)?, + u64::from(row.get::<_, u32>(1)?), + u64::from(row.get::<_, u32>(2)?), + )) + })?; + rows.map(|row| row.map(|(chain, start, end)| (chain, (start, end)))) + .collect::, _>>() + .context("read indexed chain ranges") +} + +fn existing_checkpoint_blocks(conn: &Connection) -> Result>> { + let mut existing: HashMap> = HashMap::new(); + for (chain_id, checkpoints) in load(conn)? { + existing + .entry(chain_id) + .or_default() + .extend(checkpoints.into_iter().map(|(block, _)| block)); + } + Ok(existing) +} + +fn sample_blocks(start: u64, end: u64, interval: u64) -> Vec { + if start > end { + return Vec::new(); + } + let interval = interval.max(1); + let mut blocks = vec![start]; + let mut next = start + .checked_div(interval) + .and_then(|bucket| bucket.checked_add(1)) + .and_then(|bucket| bucket.checked_mul(interval)) + .unwrap_or(end); + while next < end { + blocks.push(next); + next = match next.checked_add(interval) { + Some(value) => value, + None => break, + }; + } + if blocks.last().copied() != Some(end) { + blocks.push(end); + } + blocks +} + +pub async fn run_checkpoints(args: CheckpointsArgs) -> Result<()> { + let rpcs = args + .rpc + .into_iter() + .map(|rpc| rpc.trim().to_string()) + .filter(|rpc| !rpc.is_empty()) + .collect::>(); + if rpcs.is_empty() { + return Err(anyhow!( + "no RPC URLs configured; set BLINK_SERVE_RPCS or pass --rpc" + )); + } + if args.batch_size == 0 { + return Err(anyhow!("--batch-size must be greater than zero")); + } + + fs::create_dir_all(&args.data_dir) + .with_context(|| format!("create data dir {}", args.data_dir.display()))?; + let db_path = args.data_dir.join("blink.duckdb"); + let (ranges, existing) = { + let conn = Connection::open(&db_path) + .with_context(|| format!("open duckdb {}", db_path.display()))?; + ensure_schema(&conn)?; + ( + indexed_chain_ranges(&conn)?, + existing_checkpoint_blocks(&conn)?, + ) + }; + + print_header("blink checkpoints"); + print_kv("data dir", &args.data_dir.display().to_string()); + print_kv_accent("interval", &format_count(args.interval_blocks.max(1))); + + let mut stored = 0usize; + for rpc in rpcs { + let provider = ProviderBuilder::new().connect_http( + rpc.parse() + .with_context(|| format!("invalid rpc url {rpc}"))?, + ); + let chain_id = provider + .get_chain_id() + .await + .with_context(|| format!("fetch chain id from {rpc}"))?; + let head = provider + .get_block_number() + .await + .with_context(|| format!("fetch head block for chain {chain_id}"))?; + let start = ranges.get(&chain_id).map(|range| range.0).unwrap_or(0); + let known = existing.get(&chain_id); + let blocks = sample_blocks(start, head, args.interval_blocks) + .into_iter() + .filter(|block| !known.is_some_and(|known| known.contains(block))) + .collect::>(); + + print_kv( + &format!("chain {chain_id}"), + &format!( + "{} → {} · {} missing checkpoint(s)", + format_count(start), + format_count(head), + format_count(blocks.len() as u64) + ), + ); + if blocks.is_empty() { + continue; + } + + let client = BatchClient::new(rpc, 1)?; + for batch in blocks.chunks(args.batch_size) { + let rows = client + .block_timestamps_batch( + batch, + 5, + Duration::from_millis(500), + Duration::from_secs(15), + ) + .await?; + persist_batch(&db_path, chain_id, &rows)?; + stored += rows.len(); + } + } + + print_kv_accent("stored", &format_count(stored as u64)); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::sample_blocks; + + #[test] + fn samples_indexed_start_interval_boundaries_and_head() { + assert_eq!( + sample_blocks(47_205, 250_001, 100_000), + vec![47_205, 100_000, 200_000, 250_001] + ); + } +} diff --git a/src/cli.rs b/src/cli.rs index 1135449..f0a9e98 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -41,6 +41,8 @@ pub enum Commands { Load(LoadArgs), /// Decode bytecode locally: compiler version, language, ERC standards, proxy detection Decode(DecodeArgs), + /// Build persistent block-time checkpoints from configured RPCs + Checkpoints(CheckpointsArgs), /// Serve the public monitoring dashboard Serve(ServeArgs), } @@ -57,6 +59,7 @@ fn command() -> Command { "contracts" => Some("blink contracts"), "load" => Some("blink load"), "decode" => Some("blink decode"), + "checkpoints" => Some("blink checkpoints"), "serve" => Some("blink serve"), _ => None, }; @@ -161,6 +164,23 @@ pub struct DecodeArgs { pub overwrite: bool, } +#[derive(Parser, Debug, Clone)] +pub struct CheckpointsArgs { + /// RPC URLs to sample. Repeat for multiple chains. + /// BLINK_SERVE_RPCS accepts a comma-separated list. + #[arg(long, env = "BLINK_SERVE_RPCS", value_delimiter = ',')] + pub rpc: Vec, + /// Blink data directory containing blink.duckdb (env: BLINK_DATA_DIR) + #[arg(long, env = "BLINK_DATA_DIR", default_value = "./data/blink")] + pub data_dir: PathBuf, + /// Distance between persistent historical checkpoints + #[arg(long, default_value_t = 100_000)] + pub interval_blocks: u64, + /// Block headers per JSON-RPC batch request + #[arg(long, default_value_t = 50)] + pub batch_size: usize, +} + #[derive(Parser, Debug, Clone)] pub struct ServeArgs { /// Bind address (env: BLINK_BIND) @@ -203,4 +223,11 @@ pub struct ServeArgs { /// Read connections serving dashboard queries (0 = auto-size from cores) #[arg(long, env = "BLINK_DB_READERS", default_value_t = 0)] pub db_readers: usize, + /// Enable automatic Verifier Alliance download/import from this directory. + /// The AWS CLI must be installed and available on PATH. + #[arg(long, env = "BLINK_VA_DIR")] + pub verifier_alliance_dir: Option, + /// Seconds between automatic Verifier Alliance sync attempts + #[arg(long, env = "BLINK_VA_SYNC_INTERVAL_SECS", default_value_t = 3_600)] + pub verifier_alliance_sync_interval_secs: u64, } diff --git a/src/db/explorer.rs b/src/db/explorer.rs index 361061f..01d6c02 100644 --- a/src/db/explorer.rs +++ b/src/db/explorer.rs @@ -7,7 +7,16 @@ use super::{table_exists, views, Db}; const EXPLORER_FINGERPRINT_KEY: &str = "explorer_fingerprint"; /// Rows per build slice — bounds the join/sort working set per transaction. -const EXPLORER_SLICE_TARGET_ROWS: u64 = 2_000_000; +const EXPLORER_SLICE_TARGET_ROWS: u64 = 250_000; + +pub(crate) fn cleanup_stale_build(conn: &Connection) -> Result<()> { + if !table_exists(conn, "contract_metadata_native_build")? { + return Ok(()); + } + tracing::warn!("removing incomplete sql explorer build from an earlier run"); + conn.execute_batch("DROP TABLE contract_metadata_native_build;") + .context("remove incomplete sql explorer build") +} pub(crate) fn ensure_explorer_schema(conn: &Connection) -> Result<()> { conn.execute_batch( @@ -61,10 +70,10 @@ fn input_fingerprint(conn: &Connection) -> Result { } else { 0 }; - // Bump the version prefix when the table schema changes so upgraded - // servers rebuild instead of querying a stale shape. + // Bump the version prefix when the table schema or physical layout + // changes so upgraded servers rebuild instead of querying a stale copy. Ok(format!( - "v2|meta:{meta_rows}@{}|enr:{enrichment_rows}@{}|reg:{registry_rows}", + "v3|meta:{meta_rows}@{}|enr:{enrichment_rows}@{}|reg:{registry_rows}", meta_latest.unwrap_or_default(), enrichment_latest.unwrap_or_default() )) @@ -140,45 +149,42 @@ struct ChainSlice { end_block: u64, } -fn build_slices(conn: &Connection) -> Result> { - let mut stmt = conn.prepare( +fn build_slices(conn: &Connection, target_rows: u64) -> Result> { + let target_rows = target_rows.max(1); + let mut stmt = conn.prepare(&format!( r#" - SELECT chain_id, MIN(block_number), MAX(block_number), SUM(contract_count) - FROM rollup_block_counts - GROUP BY chain_id - ORDER BY chain_id - "#, - )?; - let chains = stmt + WITH cumulative AS ( + SELECT + chain_id, + block_number, + SUM(contract_count) OVER ( + PARTITION BY chain_id + ORDER BY block_number DESC + ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW + )::UBIGINT AS cumulative_rows + FROM rollup_block_counts + ), assigned AS ( + SELECT + chain_id, + block_number, + ((cumulative_rows - 1) // {target_rows})::UBIGINT AS slice_id + FROM cumulative + ) + SELECT chain_id, MIN(block_number), MAX(block_number) + FROM assigned + GROUP BY chain_id, slice_id + ORDER BY chain_id, slice_id + "# + ))?; + let slices = stmt .query_map([], |row| { - Ok(( - row.get::<_, u64>(0)?, - row.get::<_, u32>(1)?, - row.get::<_, u32>(2)?, - row.get::<_, i64>(3)?, - )) + Ok(ChainSlice { + chain_id: row.get(0)?, + start_block: row.get::<_, u32>(1)?.into(), + end_block: row.get::<_, u32>(2)?.into(), + }) })? .collect::, _>>()?; - - let mut slices = Vec::new(); - for (chain_id, min_block, max_block, rows) in chains { - let (min_block, max_block) = (u64::from(min_block), u64::from(max_block)); - let span = max_block - min_block + 1; - let slice_count = (rows.max(0) as u64) - .div_ceil(EXPLORER_SLICE_TARGET_ROWS) - .max(1); - let slice_blocks = span.div_ceil(slice_count).max(1); - let mut start = min_block; - while start <= max_block { - let end = (start + slice_blocks - 1).min(max_block); - slices.push(ChainSlice { - chain_id, - start_block: start, - end_block: end, - }); - start = end + 1; - } - } Ok(slices) } @@ -187,6 +193,22 @@ impl Db { /// rebuild ran. Takes and releases the writer lock per slice so the tail /// loop keeps running during the minutes-long build. pub(crate) fn refresh_explorer_blocking(&self) -> Result { + let result = self.refresh_explorer_blocking_inner(); + if result.is_err() { + let conn = self.writer.blocking_lock(); + if let Err(error) = conn.execute_batch( + r#" + DROP TABLE IF EXISTS contract_metadata_native_build; + DROP TABLE IF EXISTS explorer_meta_build; + "#, + ) { + tracing::warn!("could not clean up failed sql explorer build: {error}"); + } + } + result + } + + fn refresh_explorer_blocking_inner(&self) -> Result { if self.read_only { return Ok(false); } @@ -269,7 +291,7 @@ impl Db { let slices = { let conn = self.writer.blocking_lock(); - build_slices(&conn)? + build_slices(&conn, EXPLORER_SLICE_TARGET_ROWS)? }; let total_slices = slices.len(); let is_verified_expr = { @@ -311,19 +333,17 @@ impl Db { m.code_hash IS NOT NULL FROM contract_deployments_native c LEFT JOIN explorer_meta_build m ON c.code_hash = m.code_hash - -- The block-range condition relies on - -- backfill_enrichment_blocks having run at open: every - -- enrichment row whose address exists in the deployments - -- table carries that deployment's block_number, so the join - -- hash-builds only this slice's enrichment rows instead of - -- the whole table. + -- Current VA imports persist each verification's deployment + -- position, so this join hash-builds only the enrichment rows + -- for the current slice instead of the whole table. Legacy + -- rows without positions are repaired by the VA importer. LEFT JOIN enrichment_current e ON c.contract_address = e.contract_address AND c.chain_id = e.chain_id AND e.block_number BETWEEN {start_block} AND {end_block} WHERE c.chain_id = {chain_id} AND c.block_number BETWEEN {start_block} AND {end_block} - ORDER BY c.block_number, c.create_index; + ORDER BY c.block_number DESC, c.create_index DESC; "# )) .with_context(|| { @@ -373,3 +393,50 @@ impl Db { Ok(true) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn explorer_slices_follow_row_density_newest_first() { + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch( + r#" + CREATE TABLE rollup_block_counts ( + chain_id UBIGINT, + block_number UINTEGER, + contract_count UBIGINT + ); + INSERT INTO rollup_block_counts VALUES + (1, 100, 1), + (1, 90, 9), + (1, 80, 6), + (1, 70, 6), + (100, 200, 3); + "#, + ) + .unwrap(); + + let slices = build_slices(&conn, 10).unwrap(); + let ranges = slices + .into_iter() + .map(|slice| (slice.chain_id, slice.start_block, slice.end_block)) + .collect::>(); + assert_eq!( + ranges, + vec![(1, 90, 100), (1, 80, 80), (1, 70, 70), (100, 200, 200)] + ); + } + + #[test] + fn stale_explorer_build_is_removed_at_open() { + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch("CREATE TABLE contract_metadata_native_build (id INTEGER);") + .unwrap(); + + cleanup_stale_build(&conn).unwrap(); + assert!(!table_exists(&conn, "contract_metadata_native_build").unwrap()); + cleanup_stale_build(&conn).unwrap(); + } +} diff --git a/src/db/mod.rs b/src/db/mod.rs index 4a8778e..315011d 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -22,10 +22,12 @@ use std::{ atomic::{AtomicUsize, Ordering}, Arc, }, + time::Instant, }; use anyhow::{anyhow, Context, Result}; use duckdb::{params, AccessMode, Config, Connection}; +use rayon::prelude::*; use tokio::sync::Mutex; mod explorer; @@ -78,12 +80,15 @@ impl Db { } pub fn open(data_dir: &Path, contracts_glob: &str, options: DbOptions) -> Result { + let started = Instant::now(); if !options.read_only { std::fs::create_dir_all(data_dir) .with_context(|| format!("create data dir {}", data_dir.display()))?; } let db_path = data_dir.join("blink.duckdb"); + tracing::info!("opening dashboard database {}", db_path.display()); + let open_started = Instant::now(); let writer = if options.read_only { // Read-only connection coexists with an active writer since DuckDB // only takes an exclusive lock on writers. @@ -96,6 +101,10 @@ impl Db { Connection::open(&db_path) .with_context(|| format!("open duckdb {}", db_path.display()))? }; + tracing::info!( + "dashboard database file opened in {:.1}s", + open_started.elapsed().as_secs_f64() + ); configure_connection(&writer, data_dir, &options)?; if options.read_only { @@ -108,13 +117,25 @@ impl Db { ); } } else { + explorer::cleanup_stale_build(&writer)?; + tracing::info!("checking deployment rollups"); views::ensure_schema(&writer)?; rollups::ensure_rollup_schema(&writer)?; rollups::sync_rollups(&writer, data_dir, contracts_glob)?; - rollups::backfill_enrichment_blocks(&writer)?; + } + + let checkpoint_count = crate::checkpoints::load_runtime(&writer)?; + if checkpoint_count == 0 { + tracing::warn!( + "no block-time checkpoints loaded; run `blink checkpoints` for accurate chart dates" + ); } let files = rollups::list_contract_parquet_files(data_dir, contracts_glob)?; + tracing::info!( + "preparing dashboard query views ({} parquet files)", + files.len() + ); views::rebuild_query_views(&writer, &files)?; let reader_count = if options.readers == 0 { @@ -133,6 +154,12 @@ impl Db { readers.push(Arc::new(Mutex::new(conn))); } + tracing::info!( + "dashboard database ready in {:.1}s ({} readers)", + started.elapsed().as_secs_f64(), + reader_count + ); + Ok(Self { writer: Arc::new(Mutex::new(writer)), readers: Arc::new(readers), @@ -196,6 +223,101 @@ impl Db { .map_err(|e| anyhow!("join error: {}", e))? } + /// Confirm the shared DuckDB instance is still usable. A DuckDB fatal + /// error invalidates every connection in the instance until restart. + pub async fn health_check(&self) -> Result<()> { + self.run_read(|conn| { + conn.query_row("SELECT 1", [], |row| row.get::<_, i32>(0)) + .context("query dashboard database health")?; + Ok(()) + }) + .await + } + + pub async fn record_block_checkpoint( + &self, + chain_id: u64, + block_number: u64, + timestamp: i64, + ) -> Result<()> { + if self.read_only { + return Err(anyhow!("cannot record block checkpoint in read-only mode")); + } + let writer = self.writer.clone(); + tokio::task::spawn_blocking(move || -> Result<()> { + let conn = writer.blocking_lock(); + crate::checkpoints::ensure_schema(&conn)?; + crate::checkpoints::upsert(&conn, chain_id, block_number, timestamp)?; + crate::blocks::upsert_runtime_checkpoint(chain_id, block_number, timestamp); + Ok(()) + }) + .await + .map_err(|e| anyhow!("join error: {}", e))? + } + + pub async fn latest_checkpoint_block(&self, chain_id: u64) -> Result> { + self.run_read(move |conn| { + if !table_exists(conn, "chain_block_checkpoints")? { + return Ok(None); + } + let block = conn + .query_row( + "SELECT MAX(block_number) FROM chain_block_checkpoints WHERE chain_id = ?", + params![chain_id], + |row| row.get::<_, Option>(0), + ) + .unwrap_or(None); + Ok(block) + }) + .await + } + + pub async fn import_verifier_alliance( + &self, + verifier_alliance_dir: PathBuf, + chain_id: u64, + ) -> Result { + if self.read_only { + return Err(anyhow!( + "cannot import Verifier Alliance data in read-only mode" + )); + } + let writer = self.writer.clone(); + tokio::task::spawn_blocking(move || -> Result { + let conn = writer.blocking_lock(); + crate::load::import_verifier_alliance_from_dir(&conn, &verifier_alliance_dir, chain_id) + }) + .await + .map_err(|error| anyhow!("join error: {error}"))? + } + + /// Analyze bytecodes captured by the live tail and persist metadata by + /// runtime-code hash. Tail batches are small, so this keeps today's + /// compiler and standards widgets current without a separate decoder + /// process competing for the DuckDB write lock. + pub async fn decode_live_bytecodes(&self, bytecodes: Vec<(Vec, Vec)>) -> Result { + if self.read_only || bytecodes.is_empty() { + return Ok(0); + } + let writer = self.writer.clone(); + tokio::task::spawn_blocking(move || -> Result { + let analyzed = bytecodes + .into_par_iter() + .filter(|(code_hash, code)| { + code_hash.len() == 32 && !code.is_empty() && code.len() <= 65_536 + }) + .map(|(code_hash, code)| { + let metadata = crate::decode::bytecode_meta::analyze(&code); + (code_hash, metadata) + }) + .collect::>(); + let mut conn = writer.blocking_lock(); + crate::decode::flush_hash_batch(&mut conn, &analyzed) + }) + .await + .map_err(|error| anyhow!("join error: {error}"))? + } + fn reader(&self) -> Arc> { if self.readers.is_empty() { return self.writer.clone(); diff --git a/src/db/queries.rs b/src/db/queries.rs index ce2b00b..edfe1c8 100644 --- a/src/db/queries.rs +++ b/src/db/queries.rs @@ -7,10 +7,10 @@ use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use duckdb::{params, Connection, Row}; -use super::{column_exists, rollups, sql, table_exists, Db}; +use super::{rollups, sql, table_exists, Db}; /// Fast path for the "recent deployments" page: scan only this many blocks -/// below the cursor/head first, and fall back to an unbounded scan when the +/// below the curso``r/head first, and fall back to an unbounded scan when the /// window doesn't fill the page (tiny datasets, sparse chains). const RECENT_SCAN_WINDOW_BLOCKS: u64 = 100_000; @@ -128,23 +128,13 @@ fn read_string_u64_pair(row: &Row<'_>) -> duckdb::Result<(String, u64)> { Ok((row.get::<_, String>(0)?, row.get::<_, u64>(1)?)) } -fn verification_registry_loaded(conn: &Connection, chain_id: u64) -> Result { - if !table_exists(conn, "verification_registry_imports")? { - return Ok(false); - } - let count: i64 = conn - .query_row( - "SELECT COUNT(*) FROM verification_registry_imports WHERE source = 'verifier_alliance' AND chain_id = ?", - params![chain_id], - |row| row.get(0), - ) - .unwrap_or(0); - Ok(count > 0) +fn range_filter(block_range: Option<(u64, u64)>) -> String { + range_filter_on("block_number", block_range) } -fn range_filter(block_range: Option<(u64, u64)>) -> String { +fn range_filter_on(column: &str, block_range: Option<(u64, u64)>) -> String { block_range - .map(|(start, end)| format!("AND block_number BETWEEN {start} AND {end}")) + .map(|(start, end)| format!("AND {column} BETWEEN {start} AND {end}")) .unwrap_or_default() } @@ -197,53 +187,6 @@ fn code_counts_source(chain_id: u64, block_range: Option<(u64, u64)>) -> String parts.join("\nUNION ALL\n") } -/// Whether the materialized explorer table can back the metadata aggregates -/// (compilers/standards/sizes/languages). It denormalizes decode metadata per -/// deployment, sorted by (chain, block), so ranged aggregates are pruned -/// scans with no join against the multi-million-row metadata table. -/// -/// The column check matters: while a schema-upgrading rebuild runs (or if one -/// failed), the previous-generation table is still what's on disk — fall back -/// to the join path rather than referencing columns it doesn't have yet. -fn explorer_backed(conn: &Connection) -> Result { - Ok(table_exists(conn, "contract_metadata_native")? - && table_exists(conn, "contract_metadata_bounds")? - && column_exists(conn, "contract_metadata_native", "is_decoded")?) -} - -/// Per-deployment metadata rows for aggregate endpoints: the materialized -/// table for everything up to its bounds, plus live (undecorated) rows for -/// newer blocks — fresh contracts have rarely been decoded yet, matching the -/// join-based semantics. -fn explorer_code_rows_source(chain_id: u64, block_range: Option<(u64, u64)>) -> String { - let filter = range_filter(block_range); - format!( - r#" - SELECT - n_code_bytes, language, compiler_version, has_source_hash, - is_erc20, is_erc721, is_erc1155, is_proxy_eip1967, - is_proxy_minimal, uses_push0, is_decoded - FROM contract_metadata_native - WHERE chain_id = {chain_id} - {filter} - UNION ALL - SELECT - c.n_code_bytes, - CAST(NULL AS VARCHAR), CAST(NULL AS VARCHAR), false, - false, false, false, false, - false, false, false - FROM contract_deployments_native c - LEFT JOIN contract_metadata_bounds b ON c.chain_id = b.chain_id - WHERE c.chain_id = {chain_id} - AND (b.chain_id IS NULL OR c.block_number > b.max_block) - {live_filter} - "#, - live_filter = block_range - .map(|(start, end)| format!("AND c.block_number BETWEEN {start} AND {end}")) - .unwrap_or_default() - ) -} - fn deployments_code_scan(chain_id: u64, start: u64, end: u64) -> String { format!( r#" @@ -339,46 +282,36 @@ impl Db { .unwrap_or((0, None, None)); let total = total.max(0) as u64; - let registry_loaded = verification_registry_loaded(conn, chain_id)?; - let (enriched, verified): (i64, i64) = if registry_loaded { - let verified: i64 = conn - .query_row( - r#" - SELECT COUNT(*) - FROM contract_deployments_native c - JOIN enrichment e - ON c.contract_address = e.contract_address - AND c.chain_id = e.chain_id - WHERE e.is_verified - AND c.chain_id = ? - "#, - params![chain_id], - |row| row.get(0), - ) - .unwrap_or(0); - (total as i64, verified) - } else { + let verified: i64 = if table_exists(conn, "enrichment")? { conn.query_row( - "SELECT COUNT(*), COUNT(*) FILTER (WHERE is_verified) FROM enrichment WHERE chain_id = ?", + r#" + SELECT COUNT(*) + FROM contract_deployments_native c + WHERE c.chain_id = ? + AND EXISTS ( + SELECT 1 + FROM enrichment e + WHERE e.chain_id = c.chain_id + AND e.contract_address = c.contract_address + AND e.is_verified + ) + "#, params![chain_id], - |row| Ok((row.get(0)?, row.get(1)?)), + |row| row.get(0), ) - .unwrap_or((0, 0)) - }; - - let verified_count = verified.max(0) as u64; - let enriched_count = enriched.max(0) as u64; - let unverified_count = enriched_count.saturating_sub(verified_count); - let verified_pct = if enriched_count == 0 { - 0.0 + .unwrap_or(0) } else { - 100.0 * verified_count as f64 / enriched_count as f64 + 0 }; - let enrichment_coverage_pct = if total == 0 { + + let verified_count = (verified.max(0) as u64).min(total); + let unverified_count = total.saturating_sub(verified_count); + let verified_pct = if total == 0 { 0.0 } else { - 100.0 * enriched_count as f64 / total as f64 + 100.0 * verified_count as f64 / total as f64 }; + let enrichment_coverage_pct = if total == 0 { 0.0 } else { 100.0 }; Ok(Stats { total_contracts: total, @@ -443,20 +376,25 @@ impl Db { ) -> Result> { let bucket_blocks = bucket_blocks.max(1); self.run_read(move |conn| { - let registry_loaded = verification_registry_loaded(conn, chain_id)?; let filter = range_filter(block_range); + let deployment_filter = range_filter_on("c.block_number", block_range); let has_enrichment = table_exists(conn, "enrichment")?; let checked_select = if has_enrichment { format!( r#" SELECT - (block_number // {bucket_blocks})::UBIGINT AS bucket_id, - COUNT(*) FILTER (WHERE is_verified)::UBIGINT AS verified, - COUNT(*) FILTER (WHERE is_verified IS FALSE)::UBIGINT AS unverified - FROM enrichment - WHERE block_number IS NOT NULL - AND chain_id = {chain_id} - {filter} + (c.block_number // {bucket_blocks})::UBIGINT AS bucket_id, + COUNT(*)::UBIGINT AS verified + FROM contract_deployments_native c + WHERE c.chain_id = {chain_id} + {deployment_filter} + AND EXISTS ( + SELECT 1 + FROM enrichment e + WHERE e.chain_id = c.chain_id + AND e.contract_address = c.contract_address + AND e.is_verified + ) GROUP BY bucket_id "# ) @@ -464,29 +402,14 @@ impl Db { r#" SELECT CAST(NULL AS UBIGINT) AS bucket_id, - 0::UBIGINT AS verified, - 0::UBIGINT AS unverified + 0::UBIGINT AS verified WHERE FALSE "# .to_string() }; - // With a registry loaded, every unchecked deployment is known - // unverified; without one, unchecked deployments are "unknown". let verified_value = "LEAST(COALESCE(checked.verified, 0), totals.total)"; - let (unverified_expr, unknown_expr) = if registry_loaded { - ( - format!("GREATEST(totals.total - {verified_value}, 0)::UBIGINT"), - "0::UBIGINT".to_string(), - ) - } else { - ( - "COALESCE(checked.unverified, 0)::UBIGINT".to_string(), - format!( - "GREATEST(totals.total - {verified_value} - COALESCE(checked.unverified, 0), 0)::UBIGINT" - ), - ) - }; + let unverified_expr = format!("GREATEST(totals.total - {verified_value}, 0)::UBIGINT"); let sql = format!( r#" WITH totals AS ( @@ -505,7 +428,7 @@ impl Db { totals.bucket_id, {verified_value}::UBIGINT AS verified, {unverified_expr} AS unverified, - {unknown_expr} AS unknown + 0::UBIGINT AS unknown FROM totals LEFT JOIN checked ON totals.bucket_id = checked.bucket_id ORDER BY totals.bucket_id @@ -572,43 +495,29 @@ impl Db { ELSE 11 END AS bin_id "#; - let sql = if explorer_backed(conn)? { - let rows_source = explorer_code_rows_source(chain_id, block_range); - format!( - r#" - SELECT {bin_case}, COUNT(*)::UBIGINT AS cnt - FROM ({rows_source}) rows - WHERE n_code_bytes IS NOT NULL - AND n_code_bytes <= 24576 - GROUP BY bin_id - ORDER BY bin_id - "# + let count_source = code_counts_source(chain_id, block_range); + let sql = format!( + r#" + WITH counts AS ( + {count_source} ) - } else { - let count_source = code_counts_source(chain_id, block_range); - format!( - r#" - WITH counts AS ( - {count_source} - ) + SELECT + {bin_case}, + SUM(counts.contract_count)::UBIGINT AS cnt + FROM ( SELECT - {bin_case}, - SUM(counts.contract_count)::UBIGINT AS cnt - FROM ( - SELECT - counts.n_code_bytes AS n_code_bytes, - m.is_proxy_minimal AS is_proxy_minimal, - counts.contract_count AS contract_count - FROM counts - LEFT JOIN bytecode_metadata_by_hash m ON counts.code_hash = m.code_hash - ) counts - WHERE counts.n_code_bytes IS NOT NULL - AND counts.n_code_bytes <= 24576 - GROUP BY bin_id - ORDER BY bin_id - "# - ) - }; + counts.n_code_bytes AS n_code_bytes, + m.is_proxy_minimal AS is_proxy_minimal, + counts.contract_count AS contract_count + FROM counts + LEFT JOIN bytecode_metadata_by_hash m ON counts.code_hash = m.code_hash + ) counts + WHERE counts.n_code_bytes IS NOT NULL + AND counts.n_code_bytes <= 24576 + GROUP BY bin_id + ORDER BY bin_id + "# + ); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map([], read_u64_pair)?; for r in rows { @@ -642,35 +551,21 @@ impl Db { // Compiler distribution is bytecode-derived. Verification sources // can confirm source publication, but local decode remains the // source of truth for compiler metadata in this dashboard. - let sql = if explorer_backed(conn)? { - let rows_source = explorer_code_rows_source(chain_id, block_range); - format!( - r#" - SELECT compiler_version, COUNT(*)::UBIGINT AS cnt - FROM ({rows_source}) rows - WHERE compiler_version IS NOT NULL - GROUP BY compiler_version - ORDER BY cnt DESC - LIMIT ? - "# - ) - } else { - let count_source = code_counts_source(chain_id, block_range); - format!( - r#" - WITH counts AS ( - {count_source} - ) - SELECT m.compiler_version, SUM(c.contract_count)::UBIGINT AS cnt - FROM counts c - JOIN bytecode_metadata_by_hash m ON c.code_hash = m.code_hash - WHERE m.compiler_version IS NOT NULL - GROUP BY m.compiler_version - ORDER BY cnt DESC - LIMIT ? - "# + let count_source = code_counts_source(chain_id, block_range); + let sql = format!( + r#" + WITH counts AS ( + {count_source} ) - }; + SELECT m.compiler_version, SUM(c.contract_count)::UBIGINT AS cnt + FROM counts c + JOIN bytecode_metadata_by_hash m ON c.code_hash = m.code_hash + WHERE m.compiler_version IS NOT NULL + GROUP BY m.compiler_version + ORDER BY cnt DESC + LIMIT ? + "# + ); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map(params![limit as i64], read_string_u64_pair)?; let mut out = Vec::new(); @@ -692,29 +587,18 @@ impl Db { block_range: Option<(u64, u64)>, ) -> Result { self.run_read(move |conn| { - let sql = if explorer_backed(conn)? { - let rows_source = explorer_code_rows_source(chain_id, block_range); - format!( - r#" - SELECT COUNT(*)::BIGINT - FROM ({rows_source}) rows - WHERE compiler_version IS NOT NULL - "# - ) - } else { - let count_source = code_counts_source(chain_id, block_range); - format!( - r#" - WITH counts AS ( - {count_source} - ) - SELECT COALESCE(SUM(c.contract_count), 0)::BIGINT - FROM counts c - JOIN bytecode_metadata_by_hash m ON c.code_hash = m.code_hash - WHERE m.compiler_version IS NOT NULL - "# + let count_source = code_counts_source(chain_id, block_range); + let sql = format!( + r#" + WITH counts AS ( + {count_source} ) - }; + SELECT COALESCE(SUM(c.contract_count), 0)::BIGINT + FROM counts c + JOIN bytecode_metadata_by_hash m ON c.code_hash = m.code_hash + WHERE m.compiler_version IS NOT NULL + "# + ); let count: i64 = conn.query_row(&sql, [], |row| row.get(0)).unwrap_or(0); Ok(count.max(0) as u64) }) @@ -723,30 +607,17 @@ impl Db { pub async fn language_distribution(&self, chain_id: u64) -> Result> { self.run_read(move |conn| { - let sql = if explorer_backed(conn)? { - let rows_source = explorer_code_rows_source(chain_id, None); - format!( - r#" - SELECT COALESCE(language, 'unknown') AS lang, - COUNT(*)::UBIGINT AS cnt - FROM ({rows_source}) rows - GROUP BY lang - ORDER BY cnt DESC - "# - ) - } else { - format!( - r#" - SELECT COALESCE(m.language, 'unknown') AS lang, - SUM(c.contract_count)::UBIGINT AS cnt - FROM rollup_code_counts c - LEFT JOIN bytecode_metadata_by_hash m ON c.code_hash = m.code_hash - WHERE c.chain_id = {chain_id} - GROUP BY lang - ORDER BY cnt DESC - "# - ) - }; + let sql = format!( + r#" + SELECT COALESCE(m.language, 'unknown') AS lang, + SUM(c.contract_count)::UBIGINT AS cnt + FROM rollup_code_counts c + LEFT JOIN bytecode_metadata_by_hash m ON c.code_hash = m.code_hash + WHERE c.chain_id = {chain_id} + GROUP BY lang + ORDER BY cnt DESC + "# + ); let mut stmt = conn.prepare(&sql)?; let rows = stmt.query_map([], read_string_u64_pair)?; let mut out = Vec::new(); @@ -765,43 +636,25 @@ impl Db { block_range: Option<(u64, u64)>, ) -> Result { self.run_read(move |conn| { - let sql = if explorer_backed(conn)? { - let rows_source = explorer_code_rows_source(chain_id, block_range); - format!( - r#" - SELECT - COUNT(*) FILTER (WHERE is_erc20)::UBIGINT, - COUNT(*) FILTER (WHERE is_erc721)::UBIGINT, - COUNT(*) FILTER (WHERE is_erc1155)::UBIGINT, - COUNT(*) FILTER (WHERE is_proxy_eip1967)::UBIGINT, - COUNT(*) FILTER (WHERE is_proxy_minimal)::UBIGINT, - COUNT(*) FILTER (WHERE uses_push0)::UBIGINT, - COUNT(*) FILTER (WHERE has_source_hash)::UBIGINT, - COUNT(*) FILTER (WHERE is_decoded)::UBIGINT - FROM ({rows_source}) rows - "# - ) - } else { - let count_source = code_counts_source(chain_id, block_range); - format!( - r#" - WITH counts AS ( - {count_source} - ) - SELECT - COALESCE(SUM(CASE WHEN m.is_erc20 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, - COALESCE(SUM(CASE WHEN m.is_erc721 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, - COALESCE(SUM(CASE WHEN m.is_erc1155 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, - COALESCE(SUM(CASE WHEN m.is_proxy_eip1967 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, - COALESCE(SUM(CASE WHEN m.is_proxy_minimal THEN c.contract_count ELSE 0 END), 0)::UBIGINT, - COALESCE(SUM(CASE WHEN m.uses_push0 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, - COALESCE(SUM(CASE WHEN m.has_source_hash THEN c.contract_count ELSE 0 END), 0)::UBIGINT, - COALESCE(SUM(CASE WHEN m.code_hash IS NOT NULL THEN c.contract_count ELSE 0 END), 0)::UBIGINT - FROM counts c - LEFT JOIN bytecode_metadata_by_hash m ON c.code_hash = m.code_hash - "# + let count_source = code_counts_source(chain_id, block_range); + let sql = format!( + r#" + WITH counts AS ( + {count_source} ) - }; + SELECT + COALESCE(SUM(CASE WHEN m.is_erc20 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, + COALESCE(SUM(CASE WHEN m.is_erc721 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, + COALESCE(SUM(CASE WHEN m.is_erc1155 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, + COALESCE(SUM(CASE WHEN m.is_proxy_eip1967 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, + COALESCE(SUM(CASE WHEN m.is_proxy_minimal THEN c.contract_count ELSE 0 END), 0)::UBIGINT, + COALESCE(SUM(CASE WHEN m.uses_push0 THEN c.contract_count ELSE 0 END), 0)::UBIGINT, + COALESCE(SUM(CASE WHEN m.has_source_hash THEN c.contract_count ELSE 0 END), 0)::UBIGINT, + COALESCE(SUM(CASE WHEN m.code_hash IS NOT NULL THEN c.contract_count ELSE 0 END), 0)::UBIGINT + FROM counts c + LEFT JOIN bytecode_metadata_by_hash m ON c.code_hash = m.code_hash + "# + ); conn.query_row(&sql, [], |row| { Ok(StandardsBreakdown { erc20: row.get(0)?, @@ -828,8 +681,6 @@ impl Db { let limit = limit.clamp(1, 200); self.run_read(move |conn| { let page_limit = limit as usize + 1; - let registry_loaded = verification_registry_loaded(conn, chain_id)?; - // Recent pages live near the head of the chain: try a bounded // window first so the top-N scan prunes to a handful of row // groups, then widen only if the page didn't fill. @@ -861,11 +712,7 @@ impl Db { .get(&row.address) .cloned() .unwrap_or((None, None)); - let is_verified = if registry_loaded { - Some(verified.unwrap_or(false)) - } else { - verified - }; + let is_verified = Some(verified.unwrap_or(false)); let compiler_version = row .code_hash .as_ref() diff --git a/src/db/rollups.rs b/src/db/rollups.rs index bebeaed..81c3ccb 100644 --- a/src/db/rollups.rs +++ b/src/db/rollups.rs @@ -238,12 +238,14 @@ pub(crate) fn sync_rollups( let mut changed = !removed.is_empty(); for file in &files { let key = file.display().to_string(); - if tracked.contains(&key) { + if tracked.contains(&key) || source_is_tracked(conn, &key)? { + tracked.insert(key); continue; } match ingest_parquet(conn, file) { Ok(rows) => { changed = true; + tracked.insert(key.clone()); tracing::info!( "rolled up {} ({} new deployments)", file.file_name() @@ -284,6 +286,15 @@ fn tracked_sources(conn: &Connection) -> Result> { .context("list tracked rollup sources") } +fn source_is_tracked(conn: &Connection, source_key: &str) -> Result { + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM rollup_sources WHERE source_path = ?", + params![source_key], + |row| row.get(0), + )?; + Ok(count > 0) +} + /// Column names of a parquet file, so ingest can adapt to the differing /// schemas of blink/cryo/paradigm sources instead of failing outright. fn parquet_columns(conn: &Connection, path_sql: &str) -> Result> { @@ -461,7 +472,11 @@ fn record_source( inserted: u64, ) -> Result<()> { conn.execute( - "INSERT INTO rollup_sources (source_path, start_block, end_block, row_count) VALUES (?, ?, ?, ?)", + r#" + INSERT INTO rollup_sources (source_path, start_block, end_block, row_count) + VALUES (?, ?, ?, ?) + ON CONFLICT (source_path) DO NOTHING + "#, params![ source_key, min_block.map(u64::from), @@ -654,29 +669,27 @@ pub fn invalidate_zellic_rollups(conn: &Connection) -> Result<()> { invalidate_source(conn, ZELLIC_SOURCE_KEY) } -/// Fill enrichment rows that lack a block position from the deployments -/// table, so the verified-ratio chart can bucket them without joins at -/// request time. -pub(crate) fn backfill_enrichment_blocks(conn: &Connection) -> Result<()> { - if !table_exists(conn, "enrichment")? { - return Ok(()); +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn recording_an_existing_rollup_source_is_idempotent() { + let conn = Connection::open_in_memory().expect("in-memory database"); + ensure_rollup_schema(&conn).expect("rollup schema"); + + record_source(&conn, "/tmp/tail.parquet", Some(10), Some(20), 7).expect("record source"); + record_source(&conn, "/tmp/tail.parquet", Some(10), Some(20), 0) + .expect("record source again"); + + let (count, rows): (i64, u64) = conn + .query_row( + "SELECT COUNT(*), MAX(row_count) FROM rollup_sources WHERE source_path = ?", + params!["/tmp/tail.parquet"], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .expect("read rollup source"); + assert_eq!(count, 1); + assert_eq!(rows, 7); } - if !column_exists(conn, "enrichment", "block_number")? - || !column_exists(conn, "enrichment", "create_index")? - { - return Ok(()); - } - conn.execute_batch( - r#" - UPDATE enrichment AS e - SET - block_number = c.block_number, - create_index = c.create_index - FROM contract_deployments_native AS c - WHERE e.contract_address = c.contract_address - AND e.chain_id = c.chain_id - AND (e.block_number IS NULL OR e.create_index IS NULL); - "#, - ) - .context("backfill enrichment block positions") } diff --git a/src/db/views.rs b/src/db/views.rs index 7ba03a2..eff6823 100644 --- a/src/db/views.rs +++ b/src/db/views.rs @@ -85,7 +85,8 @@ pub(crate) fn ensure_schema(conn: &Connection) -> Result<()> { WHERE is_proxy_minimal IS NULL; "#, ) - .context("create blink schema") + .context("create blink schema")?; + crate::checkpoints::ensure_schema(conn) } /// Rebuild every temp view on this connection: the parquet-backed raw views @@ -475,8 +476,8 @@ fn create_standard_query_views(conn: &Connection) -> Result<()> { /// /// Prefers the materialized `contract_metadata_native` table (see /// `db::explorer`) with deployments newer than its bounds unioned in live and -/// undecorated; falls back to a live join over the native tables until the -/// materialization has been built. `transaction_hash` / `block_hash` / +/// decorated from the hash metadata table; falls back to a live join over the +/// native tables until the materialization has been built. `transaction_hash` / `block_hash` / /// `factory` are always NULL here — query the raw `contracts` view when /// those are needed. pub(crate) fn create_contract_metadata_view(conn: &Connection) -> Result<()> { @@ -548,16 +549,16 @@ pub(crate) fn create_contract_metadata_view(conn: &Connection) -> Result<()> { c.code_hash, lower('0x' || hex(c.code_hash)) AS code_hash_hex, c.n_code_bytes, - CAST(NULL AS VARCHAR) AS language, - CAST(NULL AS VARCHAR) AS compiler_version, - CAST(false AS BOOLEAN) AS has_source_hash, - CAST(false AS BOOLEAN) AS is_erc20, - CAST(false AS BOOLEAN) AS is_erc721, - CAST(false AS BOOLEAN) AS is_erc1155, - CAST(false AS BOOLEAN) AS is_proxy_eip1967, - CAST(false AS BOOLEAN) AS is_proxy_minimal, - CAST(false AS BOOLEAN) AS uses_push0, - CAST(NULL AS TIMESTAMP) AS decoded_at, + m.language, + m.compiler_version, + COALESCE(m.has_source_hash, false) AS has_source_hash, + COALESCE(m.is_erc20, false) AS is_erc20, + COALESCE(m.is_erc721, false) AS is_erc721, + COALESCE(m.is_erc1155, false) AS is_erc1155, + COALESCE(m.is_proxy_eip1967, false) AS is_proxy_eip1967, + COALESCE(m.is_proxy_minimal, false) AS is_proxy_minimal, + COALESCE(m.uses_push0, false) AS uses_push0, + m.decoded_at, {live_verified} AS is_verified, CAST(NULL AS VARCHAR) AS contract_name, CAST(NULL AS VARCHAR) AS verification_source, @@ -565,6 +566,7 @@ pub(crate) fn create_contract_metadata_view(conn: &Connection) -> Result<()> { CAST(NULL AS TIMESTAMP) AS verification_checked_at FROM contract_deployments_native c LEFT JOIN contract_metadata_bounds b ON c.chain_id = b.chain_id + LEFT JOIN decoded_bytecodes m ON c.code_hash = m.code_hash WHERE b.chain_id IS NULL OR c.block_number > b.max_block "# ) diff --git a/src/decode/mod.rs b/src/decode/mod.rs index 1024ede..38a7163 100644 --- a/src/decode/mod.rs +++ b/src/decode/mod.rs @@ -298,9 +298,12 @@ fn run_decode_blocking(args: DecodeArgs, data_dir: PathBuf) -> Result<()> { in_file += 1; if raw.len() >= raw_batch { - let analyzed: Vec<(Vec, BytecodeMetadata)> = std::mem::take(&mut raw) + let analyzed: Vec<(Vec, Vec, BytecodeMetadata)> = std::mem::take(&mut raw) .into_par_iter() - .map(|(a, c)| (a, analyze(&c))) + .map(|(address, code)| { + let code_hash = alloy::primitives::keccak256(&code).to_vec(); + (address, code_hash, analyze(&code)) + }) .collect(); written += flush_batch(&mut write_conn, &file_path_str, &analyzed)?; raw = Vec::with_capacity(raw_batch); @@ -320,9 +323,12 @@ fn run_decode_blocking(args: DecodeArgs, data_dir: PathBuf) -> Result<()> { } let _ = std::fs::remove_file(&temp_path); if !raw.is_empty() { - let analyzed: Vec<(Vec, BytecodeMetadata)> = std::mem::take(&mut raw) + let analyzed: Vec<(Vec, Vec, BytecodeMetadata)> = std::mem::take(&mut raw) .into_par_iter() - .map(|(a, c)| (a, analyze(&c))) + .map(|(address, code)| { + let code_hash = alloy::primitives::keccak256(&code).to_vec(); + (address, code_hash, analyze(&code)) + }) .collect(); written += flush_batch(&mut write_conn, &file_path_str, &analyzed)?; } @@ -648,7 +654,7 @@ fn temp_zellic_decode_path(db_path: &Path) -> PathBuf { fn flush_batch( conn: &mut Connection, source_file: &str, - buffer: &[(Vec, BytecodeMetadata)], + buffer: &[(Vec, Vec, BytecodeMetadata)], ) -> Result { if buffer.is_empty() { return Ok(0); @@ -678,7 +684,7 @@ fn flush_batch( { let mut app = conn.appender("_bm_stage_v2").context("open v2 appender")?; - for (addr, meta) in buffer { + for (addr, _, meta) in buffer { app.append_row(params![ addr, meta.language.map(|l| l.as_str()), @@ -712,10 +718,22 @@ fn flush_batch( "#, ) .context("insert staged metadata v2 rows")?; + + let mut seen_hashes = std::collections::HashSet::with_capacity(buffer.len()); + let hash_metadata = buffer + .iter() + .filter(|(_, code_hash, _)| seen_hashes.insert(code_hash.as_slice())) + .map(|(_, code_hash, metadata)| (code_hash.clone(), metadata.clone())) + .collect::>(); + flush_hash_batch(conn, &hash_metadata).context("insert hash metadata for parquet decode")?; + Ok(buffer.len() as u64) } -fn flush_hash_batch(conn: &mut Connection, buffer: &[(Vec, BytecodeMetadata)]) -> Result { +pub(crate) fn flush_hash_batch( + conn: &mut Connection, + buffer: &[(Vec, BytecodeMetadata)], +) -> Result { if buffer.is_empty() { return Ok(0); } @@ -723,7 +741,8 @@ fn flush_hash_batch(conn: &mut Connection, buffer: &[(Vec, BytecodeMetadata) let tx = conn .transaction() .context("open hash metadata transaction")?; - { + let inserted = { + let mut inserted = 0u64; let mut stmt = tx .prepare( r#" @@ -732,29 +751,35 @@ fn flush_hash_batch(conn: &mut Connection, buffer: &[(Vec, BytecodeMetadata) is_erc20, is_erc721, is_erc1155, is_proxy_eip1967, is_proxy_minimal, uses_push0, decoded_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) + SELECT ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP + WHERE NOT EXISTS ( + SELECT 1 FROM bytecode_metadata_by_hash WHERE code_hash = ? + ) "#, ) .context("prepare hash metadata insert")?; for (code_hash, meta) in buffer { - stmt.execute(params![ - code_hash, - meta.language.map(|l| l.as_str()), - meta.compiler_version.as_deref(), - meta.has_source_hash, - meta.is_erc20, - meta.is_erc721, - meta.is_erc1155, - meta.is_proxy_eip1967, - meta.is_proxy_minimal, - meta.uses_push0, - ]) - .context("insert hash metadata row")?; + inserted += stmt + .execute(params![ + code_hash, + meta.language.map(|l| l.as_str()), + meta.compiler_version.as_deref(), + meta.has_source_hash, + meta.is_erc20, + meta.is_erc721, + meta.is_erc1155, + meta.is_proxy_eip1967, + meta.is_proxy_minimal, + meta.uses_push0, + code_hash, + ]) + .context("insert hash metadata row")? as u64; } - } + inserted + }; tx.commit().context("commit hash metadata batch")?; - Ok(buffer.len() as u64) + Ok(inserted) } fn list_parquet_files(data_dir: &std::path::Path, pattern: &str) -> Result> { diff --git a/src/extract/batch.rs b/src/extract/batch.rs index 72c6fb6..6ab36c7 100644 --- a/src/extract/batch.rs +++ b/src/extract/batch.rs @@ -7,7 +7,7 @@ use std::{sync::Arc, time::Duration}; use alloy::rpc::types::trace::parity::LocalizedTransactionTrace; -use anyhow::{anyhow, Result}; +use anyhow::{anyhow, Context, Result}; use reqwest::Client; use serde::{Deserialize, Serialize}; use tokio::sync::Semaphore; @@ -16,7 +16,7 @@ use tokio::sync::Semaphore; struct JsonRpcRequest { jsonrpc: &'static str, method: &'static str, - params: [serde_json::Value; 1], + params: Vec, id: u64, } @@ -67,7 +67,7 @@ impl BatchClient { .map(|(i, &block)| JsonRpcRequest { jsonrpc: "2.0", method: "trace_block", - params: [serde_json::Value::String(format!("0x{:x}", block))], + params: vec![serde_json::Value::String(format!("0x{:x}", block))], id: i as u64, }) .collect(); @@ -163,6 +163,135 @@ impl BatchClient { } } } + + pub async fn block_timestamps_batch( + &self, + blocks: &[u64], + max_retries: u32, + initial_backoff: Duration, + max_backoff: Duration, + ) -> Result> { + let requests = blocks + .iter() + .enumerate() + .map(|(index, block)| JsonRpcRequest { + jsonrpc: "2.0", + method: "eth_getBlockByNumber", + params: vec![ + serde_json::Value::String(format!("0x{block:x}")), + serde_json::Value::Bool(false), + ], + id: index as u64, + }) + .collect::>(); + + let mut attempts = 0u32; + let mut backoff = initial_backoff; + loop { + attempts += 1; + let permit = self.semaphore.acquire().await?; + let response = self.client.post(&self.rpc_url).json(&requests).send().await; + match response { + Ok(response) if response.status().is_success() => { + let responses: Vec = response.json().await?; + let mut timestamps = vec![None; blocks.len()]; + let mut rate_limited = false; + for response in responses { + let index = response.id as usize; + if index >= blocks.len() { + return Err(anyhow!( + "invalid response id {} for batch len {}", + response.id, + blocks.len() + )); + } + if let Some(error) = response.error { + if is_rate_limited(&error) { + rate_limited = true; + break; + } + return Err(anyhow!( + "RPC error for block {}: {} (code {})", + blocks[index], + error.message, + error.code + )); + } + let value = response + .result + .ok_or_else(|| anyhow!("missing block {}", blocks[index]))?; + let timestamp = value + .get("timestamp") + .and_then(serde_json::Value::as_str) + .ok_or_else(|| { + anyhow!("missing timestamp for block {}", blocks[index]) + })?; + let timestamp = parse_quantity(timestamp).with_context(|| { + format!("invalid timestamp for block {}", blocks[index]) + })?; + timestamps[index] = Some(timestamp as i64); + } + + if rate_limited { + if attempts <= max_retries { + drop(permit); + tokio::time::sleep(backoff).await; + backoff = (backoff.saturating_mul(2)).min(max_backoff); + continue; + } + return Err(anyhow!( + "RPC rate limited for block timestamps {:?} after {} retries", + blocks, + max_retries + )); + } + + return blocks + .iter() + .enumerate() + .map(|(index, block)| { + timestamps[index] + .map(|timestamp| (*block, timestamp)) + .ok_or_else(|| anyhow!("missing timestamp for block {block}")) + }) + .collect(); + } + Ok(response) if attempts <= max_retries => { + tracing::debug!( + "block timestamp RPC returned {}; retrying", + response.status() + ); + drop(permit); + tokio::time::sleep(backoff).await; + backoff = (backoff.saturating_mul(2)).min(max_backoff); + } + Ok(response) => { + return Err(anyhow!( + "HTTP error {} fetching block timestamps {:?}", + response.status(), + blocks + )); + } + Err(_) if attempts <= max_retries => { + drop(permit); + tokio::time::sleep(backoff).await; + backoff = (backoff.saturating_mul(2)).min(max_backoff); + } + Err(error) => { + return Err(anyhow!( + "request failed fetching block timestamps {:?}: {}", + blocks, + error + )); + } + } + } + } +} + +fn parse_quantity(value: &str) -> Result { + let value = value.strip_prefix("0x").unwrap_or(value); + u64::from_str_radix(value, 16).map_err(Into::into) } fn is_rate_limited(err: &JsonRpcError) -> bool { diff --git a/src/extract/tail.rs b/src/extract/tail.rs index f623e5a..5350106 100644 --- a/src/extract/tail.rs +++ b/src/extract/tail.rs @@ -1,6 +1,11 @@ //! Continuous tail extraction for the dashboard's `serve` mode. -use std::{collections::BTreeMap, path::Path, sync::Arc, time::Duration}; +use std::{ + collections::{BTreeMap, HashMap}, + path::Path, + sync::Arc, + time::Duration, +}; use alloy::{ providers::{Provider, ProviderBuilder}, @@ -41,6 +46,8 @@ pub async fn tail_once( .context("tail: get head block")?; let target = head.saturating_sub(confirmations); + sync_block_time_checkpoint(db, rpc_url, chain_id, target).await?; + let highest_indexed = db.highest_block(chain_id).await?.unwrap_or(0); let start_block = if highest_indexed == 0 { target @@ -85,6 +92,7 @@ pub async fn tail_once( let schema = parquet_io::schema(); let mut writer = None; let mut rows_written = 0usize; + let mut live_bytecodes = HashMap::, Vec>::new(); let mut pending: BTreeMap)>> = BTreeMap::new(); let mut next_index = 0usize; @@ -119,6 +127,11 @@ pub async fn tail_once( batch_rows.append(&mut block_rows); } if !batch_rows.is_empty() { + for row in &batch_rows { + live_bytecodes + .entry(row.code_hash.clone()) + .or_insert_with(|| row.code.clone()); + } batch_rows.sort_unstable_by(|a, b| { a.block_number .cmp(&b.block_number) @@ -152,6 +165,12 @@ pub async fn tail_once( } db.refresh().await?; + let decoded = db + .decode_live_bytecodes(live_bytecodes.into_iter().collect()) + .await?; + if decoded > 0 { + tracing::info!("decoded {} new live bytecode(s)", decoded); + } let size_bytes = std::fs::metadata(&output_path).ok().map(|m| m.len()); Ok(Some(ChunkReport { @@ -166,3 +185,37 @@ pub async fn tail_once( skipped: false, })) } + +async fn sync_block_time_checkpoint( + db: &Db, + rpc_url: &str, + chain_id: u64, + target: u64, +) -> Result<()> { + let latest = db.latest_checkpoint_block(chain_id).await?; + let interval = crate::blocks::blocks_per_day(chain_id, target).max(1); + if latest.is_some_and(|block| target.saturating_sub(block) < interval) { + return Ok(()); + } + + let client = BatchClient::new(rpc_url.to_string(), 1)?; + let checkpoint = client + .block_timestamps_batch( + &[target], + 5, + Duration::from_millis(500), + Duration::from_secs(15), + ) + .await? + .into_iter() + .next() + .context("tail: missing checkpoint block response")?; + db.record_block_checkpoint(chain_id, checkpoint.0, checkpoint.1) + .await?; + tracing::info!( + "recorded block-time checkpoint chain_id={} block={}", + chain_id, + checkpoint.0 + ); + Ok(()) +} diff --git a/src/lib.rs b/src/lib.rs index ff4b776..1e1912b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,6 @@ pub mod blocks; pub mod chains; +pub mod checkpoints; pub mod cli; pub mod db; pub mod decode; @@ -8,3 +9,4 @@ pub mod load; pub mod serve; pub mod types; pub mod util; +pub mod va_sync; diff --git a/src/load.rs b/src/load.rs index 563234a..62dbdae 100644 --- a/src/load.rs +++ b/src/load.rs @@ -475,8 +475,6 @@ fn ensure_verification_schema(conn: &Connection) -> Result<()> { ); CREATE INDEX IF NOT EXISTS verification_registry_file_addresses_idx ON verification_registry_file_addresses(source, chain_id, table_name, path); - CREATE INDEX IF NOT EXISTS verification_registry_file_addresses_addr_idx - ON verification_registry_file_addresses(chain_id, contract_address); "#, ) .context("create verification registry schema") @@ -490,6 +488,29 @@ fn load_verifier_alliance_registry( chain_id: u64, rebuild_va: bool, ) -> Result<()> { + let db_path = data_dir.join("blink.duckdb"); + let conn = + Connection::open(&db_path).with_context(|| format!("open duckdb {}", db_path.display()))?; + configure_duckdb(&conn, memory_limit, threads)?; + import_verifier_alliance_with_connection(&conn, inputs, chain_id, rebuild_va).map(|_| ()) +} + +pub(crate) fn import_verifier_alliance_from_dir( + conn: &Connection, + verifier_alliance_dir: &Path, + chain_id: u64, +) -> Result { + let inputs = detect_verifier_alliance_inputs(Some(verifier_alliance_dir))? + .ok_or_else(|| anyhow!("Verifier Alliance directory is required"))?; + import_verifier_alliance_with_connection(conn, &inputs, chain_id, false) +} + +fn import_verifier_alliance_with_connection( + conn: &Connection, + inputs: &VerifierAllianceInputs, + chain_id: u64, + rebuild_va: bool, +) -> Result { let started = Instant::now(); print_kv( "step", @@ -499,18 +520,13 @@ fn load_verifier_alliance_registry( "import Verifier Alliance registry incrementally" }, ); - - let db_path = data_dir.join("blink.duckdb"); - let conn = - Connection::open(&db_path).with_context(|| format!("open duckdb {}", db_path.display()))?; - configure_duckdb(&conn, memory_limit, threads)?; - ensure_verification_schema(&conn)?; + ensure_verification_schema(conn)?; let entries = verifier_alliance_file_entries(inputs)?; if rebuild_va { - rebuild_verifier_alliance_registry(&conn, inputs, &entries, chain_id, started) + rebuild_verifier_alliance_registry(conn, inputs, &entries, chain_id, started) } else { - import_verifier_alliance_registry_incremental(&conn, inputs, &entries, chain_id, started) + import_verifier_alliance_registry_incremental(conn, inputs, &entries, chain_id, started) } } @@ -520,28 +536,19 @@ fn rebuild_verifier_alliance_registry( entries: &[VaFileEntry], chain_id: u64, started: Instant, -) -> Result<()> { +) -> Result { let deployments_list = sql_path_list(&inputs.contract_deployments)?; let verifications_list = sql_path_list(&inputs.verified_contracts)?; - conn.execute_batch("BEGIN TRANSACTION;") - .context("begin Verifier Alliance import")?; - - let result = (|| -> Result<()> { - conn.execute_batch(&format!( - r#" - DELETE FROM verification_registry_imports - WHERE source = 'verifier_alliance' - AND chain_id = {chain_id}; - DELETE FROM enrichment - WHERE verification_source = 'verifier_alliance' - AND chain_id = {chain_id}; - "#, - )) - .context("prepare verification registry import")?; - let sql = format!( - r#" - CREATE OR REPLACE TEMP TABLE va_verified_contracts AS + // Stage the joined result sets in temp tables (spillable, no write + // transaction held), then write to the indexed base tables in bounded + // transactions — one giant transaction OOMs a 4GB host at COMMIT from + // buffered ART index maintenance. + conn.execute_batch(&format!( + r#" + CREATE OR REPLACE TEMP TABLE va_verified_contracts AS + SELECT *, row_number() OVER () AS rn + FROM ( SELECT cd.address AS contract_address, (max(cd.block_number) FILTER (WHERE cd.block_number >= 0))::UINTEGER @@ -558,77 +565,120 @@ fn rebuild_verifier_alliance_registry( ON cd.id = vc.deployment_id WHERE cd.chain_id = {chain_id} AND cd.address IS NOT NULL - GROUP BY cd.address; + GROUP BY cd.address + ); - CREATE OR REPLACE TEMP TABLE enrichment_next AS - SELECT - contract_address, - {chain_id}::UBIGINT AS chain_id, - true AS is_verified, - CAST(NULL AS VARCHAR) AS contract_name, - COALESCE(checked_at, CURRENT_TIMESTAMP) AS checked_at, - 'verifier_alliance' AS verification_source, - CASE - WHEN runtime_match AND creation_match THEN 'runtime+creation' - WHEN runtime_match THEN 'runtime' - WHEN creation_match THEN 'creation' - WHEN runtime_metadata_match OR creation_metadata_match THEN 'metadata' - ELSE 'verified' - END AS match_type, - block_number, - create_index - FROM va_verified_contracts; + CREATE OR REPLACE TEMP TABLE va_file_addresses_stage AS + SELECT path, contract_address, row_number() OVER () AS rn + FROM ( + SELECT DISTINCT + vc.filename AS path, + cd.address AS contract_address + FROM read_parquet({verifications_list}, filename=true) vc + JOIN read_parquet({deployments_list}) cd + ON cd.id = vc.deployment_id + WHERE cd.chain_id = {chain_id} + AND cd.address IS NOT NULL + ); + "# + )) + .context("stage Verifier Alliance rebuild")?; - INSERT INTO enrichment ( - contract_address, - chain_id, - is_verified, - contract_name, - checked_at, - verification_source, - match_type, - block_number, - create_index + run_in_txn( + conn, + &format!( + "DELETE FROM verification_registry_imports \ + WHERE source = 'verifier_alliance' AND chain_id = {chain_id};" + ), + "clear Verifier Alliance import record", + )?; + delete_in_slices( + conn, + "enrichment", + &format!("verification_source = 'verifier_alliance' AND chain_id = {chain_id}"), + "clear Verifier Alliance enrichment rows", + )?; + delete_in_slices( + conn, + "verification_registry_file_addresses", + &format!("source = 'verifier_alliance' AND chain_id = {chain_id}"), + "clear Verifier Alliance file addresses", + )?; + + insert_in_slices( + conn, + "va_verified_contracts", + "insert Verifier Alliance enrichment rows", + |lo, hi| { + format!( + r#" + INSERT INTO enrichment ( + contract_address, chain_id, is_verified, contract_name, + checked_at, verification_source, match_type, + block_number, create_index + ) + SELECT + contract_address, + {chain_id}::UBIGINT, + true, + CAST(NULL AS VARCHAR), + COALESCE(checked_at, CURRENT_TIMESTAMP), + 'verifier_alliance', + CASE + WHEN runtime_match AND creation_match THEN 'runtime+creation' + WHEN runtime_match THEN 'runtime' + WHEN creation_match THEN 'creation' + WHEN runtime_metadata_match OR creation_metadata_match THEN 'metadata' + ELSE 'verified' + END, + block_number, + create_index + FROM va_verified_contracts + WHERE rn > {lo} AND rn <= {hi}; + "# ) - SELECT - contract_address, - chain_id, - is_verified, - contract_name, - checked_at, - verification_source, - match_type, - block_number, - create_index - FROM enrichment_next; + }, + )?; + insert_in_slices( + conn, + "va_file_addresses_stage", + "insert Verifier Alliance file addresses", + |lo, hi| { + format!( + r#" + INSERT INTO verification_registry_file_addresses ( + source, chain_id, table_name, path, contract_address + ) + SELECT 'verifier_alliance', {chain_id}, 'verified_contracts', path, contract_address + FROM va_file_addresses_stage + WHERE rn > {lo} AND rn <= {hi}; + "# + ) + }, + )?; + // Bookkeeping last, so an interrupted rebuild re-runs in full. + run_in_txn( + conn, + &format!( + r#" INSERT INTO verification_registry_imports ( - source, - chain_id, - imported_at, - verified_count + source, chain_id, imported_at, verified_count ) - SELECT - 'verifier_alliance', - {chain_id}, - CURRENT_TIMESTAMP, - COUNT(*)::UBIGINT + SELECT 'verifier_alliance', {chain_id}, CURRENT_TIMESTAMP, COUNT(*)::UBIGINT FROM va_verified_contracts; "# - ); - conn.execute_batch(&sql) - .context("import Verifier Alliance verified addresses")?; - rebuild_va_file_addresses(conn, inputs, chain_id)?; - replace_va_file_manifest(conn, chain_id, entries)?; - Ok(()) - })(); - - if let Err(err) = result { - let _ = conn.execute_batch("ROLLBACK;"); - return Err(err); - } - conn.execute_batch("COMMIT;") - .context("commit Verifier Alliance import")?; + ), + "record Verifier Alliance import", + )?; + replace_va_file_manifest(conn, chain_id, entries)?; + conn.execute_batch( + r#" + DROP TABLE IF EXISTS va_verified_contracts; + DROP TABLE IF EXISTS va_file_addresses_stage; + "#, + ) + .context("drop Verifier Alliance staging tables")?; let verified = va_verified_count(conn, chain_id)?; print_kv_accent( @@ -639,7 +689,7 @@ fn rebuild_verifier_alliance_registry( started.elapsed().as_secs_f64() ), ); - Ok(()) + Ok(true) } fn import_verifier_alliance_registry_incremental( @@ -648,14 +698,28 @@ fn import_verifier_alliance_registry_incremental( entries: &[VaFileEntry], chain_id: u64, started: Instant, -) -> Result<()> { +) -> Result { let changed_entries = changed_va_file_entries(conn, chain_id, entries)?; + tracing::info!( + "Verifier Alliance manifest checked for chain_id={} ({} changed files)", + chain_id, + changed_entries.len() + ); let import_exists = verification_registry_import_exists(conn, chain_id)?; + let repair_inconsistent_state = import_exists + && changed_entries.is_empty() + && !verification_registry_state_consistent(conn, chain_id)?; + if repair_inconsistent_state { + tracing::warn!( + "Verifier Alliance registry state is inconsistent for chain_id={chain_id}; repairing from local files" + ); + } let deployment_changed = changed_entries .iter() .any(|entry| entry.table_name == "contract_deployments"); - let mut changed_verified = if deployment_changed || !import_exists { + let mut changed_verified = if deployment_changed || !import_exists || repair_inconsistent_state + { entries .iter() .filter(|entry| entry.table_name == "verified_contracts") @@ -671,7 +735,7 @@ fn import_verifier_alliance_registry_incremental( changed_verified.sort_by_key(|entry| entry.path_key.clone()); changed_verified.dedup_by(|a, b| a.path_key == b.path_key); - if changed_entries.is_empty() && import_exists { + if changed_entries.is_empty() && import_exists && !repair_inconsistent_state { let verified = va_verified_count(conn, chain_id)?; print_kv_accent( "verified", @@ -681,7 +745,7 @@ fn import_verifier_alliance_registry_incremental( started.elapsed().as_secs_f64() ), ); - return Ok(()); + return Ok(false); } if changed_verified.is_empty() { @@ -695,7 +759,7 @@ fn import_verifier_alliance_registry_incremental( started.elapsed().as_secs_f64() ), ); - return Ok(()); + return Ok(false); } let deployments_list = sql_path_list(&inputs.contract_deployments)?; @@ -705,25 +769,25 @@ fn import_verifier_alliance_registry_incremental( let changed_file_values = sql_string_values(changed_verified.iter().map(|entry| entry.path_key.as_str())); - conn.execute_batch("BEGIN TRANSACTION;") - .context("begin incremental Verifier Alliance import")?; - - let result = (|| -> Result<()> { - let sql = format!( - r#" - CREATE OR REPLACE TEMP TABLE va_changed_files AS - SELECT col0 AS path - FROM (VALUES {changed_file_values}); + // Stage every derived row set in temp tables first — the heavy parquet + // joins can spill to disk without holding a write transaction open. + conn.execute_batch(&format!( + r#" + CREATE OR REPLACE TEMP TABLE va_changed_files AS + SELECT col0 AS path + FROM (VALUES {changed_file_values}); - CREATE OR REPLACE TEMP TABLE va_previous_affected_addresses AS - SELECT DISTINCT contract_address - FROM verification_registry_file_addresses - WHERE source = 'verifier_alliance' - AND chain_id = {chain_id} - AND table_name = 'verified_contracts' - AND path IN (SELECT path FROM va_changed_files); + CREATE OR REPLACE TEMP TABLE va_previous_affected_addresses AS + SELECT DISTINCT contract_address + FROM verification_registry_file_addresses + WHERE source = 'verifier_alliance' + AND chain_id = {chain_id} + AND table_name = 'verified_contracts' + AND path IN (SELECT path FROM va_changed_files); - CREATE OR REPLACE TEMP TABLE va_changed_file_addresses AS + CREATE OR REPLACE TEMP TABLE va_changed_file_addresses AS + SELECT path, contract_address, row_number() OVER () AS rn + FROM ( SELECT DISTINCT vc.filename AS path, cd.address AS contract_address @@ -731,42 +795,20 @@ fn import_verifier_alliance_registry_incremental( JOIN read_parquet({deployments_list}) cd ON cd.id = vc.deployment_id WHERE cd.chain_id = {chain_id} - AND cd.address IS NOT NULL; + AND cd.address IS NOT NULL + ); - CREATE OR REPLACE TEMP TABLE va_affected_addresses AS + CREATE OR REPLACE TEMP TABLE va_affected_addresses AS + SELECT contract_address, row_number() OVER () AS rn + FROM ( SELECT contract_address FROM va_previous_affected_addresses UNION - SELECT contract_address FROM va_changed_file_addresses; - - DELETE FROM enrichment - WHERE verification_source = 'verifier_alliance' - AND chain_id = {chain_id} - AND contract_address IN ( - SELECT contract_address FROM va_affected_addresses - ); - - DELETE FROM verification_registry_file_addresses - WHERE source = 'verifier_alliance' - AND chain_id = {chain_id} - AND table_name = 'verified_contracts' - AND path IN (SELECT path FROM va_changed_files); - - INSERT INTO verification_registry_file_addresses ( - source, - chain_id, - table_name, - path, - contract_address - ) - SELECT - 'verifier_alliance', - {chain_id}, - 'verified_contracts', - path, - contract_address - FROM va_changed_file_addresses; + SELECT contract_address FROM va_changed_file_addresses + ); - CREATE OR REPLACE TEMP TABLE va_verified_contracts AS + CREATE OR REPLACE TEMP TABLE va_verified_contracts AS + SELECT *, row_number() OVER () AS rn + FROM ( SELECT cd.address AS contract_address, (max(cd.block_number) FILTER (WHERE cd.block_number >= 0))::UINTEGER @@ -785,69 +827,79 @@ fn import_verifier_alliance_registry_incremental( ON affected.contract_address = cd.address WHERE cd.chain_id = {chain_id} AND cd.address IS NOT NULL - GROUP BY cd.address; + GROUP BY cd.address + ); + "# + )) + .context("stage incremental Verifier Alliance import")?; + + // Replace each bounded address slice atomically. A failed or interrupted + // import therefore leaves either the previous labels or the replacement + // labels visible for every slice, never the delete-before-insert gap that + // previously produced zero verified contracts on the live dashboard. + replace_va_enrichment_in_slices(conn, chain_id)?; + + // File-address bookkeeping is updated after enrichment is already + // correct. The manifest advances last, so an interruption retries these + // files on the next run. + delete_in_slices( + conn, + "verification_registry_file_addresses", + &format!( + "source = 'verifier_alliance' AND chain_id = {chain_id} \ + AND table_name = 'verified_contracts' \ + AND path IN (SELECT path FROM va_changed_files)" + ), + "clear changed Verifier Alliance file addresses", + )?; - INSERT INTO enrichment ( - contract_address, - chain_id, - is_verified, - contract_name, - checked_at, - verification_source, - match_type, - block_number, - create_index + insert_in_slices( + conn, + "va_changed_file_addresses", + "insert Verifier Alliance file addresses", + |lo, hi| { + format!( + r#" + INSERT INTO verification_registry_file_addresses ( + source, chain_id, table_name, path, contract_address + ) + SELECT 'verifier_alliance', {chain_id}, 'verified_contracts', path, contract_address + FROM va_changed_file_addresses + WHERE rn > {lo} AND rn <= {hi}; + "# ) - SELECT - contract_address, - {chain_id}::UBIGINT AS chain_id, - true AS is_verified, - CAST(NULL AS VARCHAR) AS contract_name, - COALESCE(checked_at, CURRENT_TIMESTAMP) AS checked_at, - 'verifier_alliance' AS verification_source, - CASE - WHEN runtime_match AND creation_match THEN 'runtime+creation' - WHEN runtime_match THEN 'runtime' - WHEN creation_match THEN 'creation' - WHEN runtime_metadata_match OR creation_metadata_match THEN 'metadata' - ELSE 'verified' - END AS match_type, - block_number, - create_index - FROM va_verified_contracts; - + }, + )?; + // Bookkeeping last, so an interrupted import re-runs in full. + run_in_txn( + conn, + &format!( + r#" DELETE FROM verification_registry_imports WHERE source = 'verifier_alliance' AND chain_id = {chain_id}; - INSERT INTO verification_registry_imports ( - source, - chain_id, - imported_at, - verified_count + source, chain_id, imported_at, verified_count ) - SELECT - 'verifier_alliance', - {chain_id}, - CURRENT_TIMESTAMP, - COUNT(*)::UBIGINT + SELECT 'verifier_alliance', {chain_id}, CURRENT_TIMESTAMP, COUNT(*)::UBIGINT FROM enrichment WHERE verification_source = 'verifier_alliance' AND chain_id = {chain_id}; "# - ); - conn.execute_batch(&sql) - .context("import changed Verifier Alliance verified addresses")?; - upsert_va_file_manifest_entries(conn, chain_id, &changed_entries)?; - Ok(()) - })(); - - if let Err(err) = result { - let _ = conn.execute_batch("ROLLBACK;"); - return Err(err); - } - conn.execute_batch("COMMIT;") - .context("commit incremental Verifier Alliance import")?; + ), + "record Verifier Alliance import", + )?; + upsert_va_file_manifest_entries(conn, chain_id, &changed_entries)?; + conn.execute_batch( + r#" + DROP TABLE IF EXISTS va_changed_files; + DROP TABLE IF EXISTS va_previous_affected_addresses; + DROP TABLE IF EXISTS va_changed_file_addresses; + DROP TABLE IF EXISTS va_affected_addresses; + DROP TABLE IF EXISTS va_verified_contracts; + "#, + ) + .context("drop Verifier Alliance staging tables")?; let verified = va_verified_count(conn, chain_id)?; print_kv_accent( @@ -859,44 +911,7 @@ fn import_verifier_alliance_registry_incremental( started.elapsed().as_secs_f64() ), ); - Ok(()) -} - -fn rebuild_va_file_addresses( - conn: &Connection, - inputs: &VerifierAllianceInputs, - chain_id: u64, -) -> Result<()> { - let deployments_list = sql_path_list(&inputs.contract_deployments)?; - let verifications_list = sql_path_list(&inputs.verified_contracts)?; - let sql = format!( - r#" - DELETE FROM verification_registry_file_addresses - WHERE source = 'verifier_alliance' - AND chain_id = {chain_id}; - - INSERT INTO verification_registry_file_addresses ( - source, - chain_id, - table_name, - path, - contract_address - ) - SELECT DISTINCT - 'verifier_alliance', - {chain_id}, - 'verified_contracts', - vc.filename, - cd.address - FROM read_parquet({verifications_list}, filename=true) vc - JOIN read_parquet({deployments_list}) cd - ON cd.id = vc.deployment_id - WHERE cd.chain_id = {chain_id} - AND cd.address IS NOT NULL; - "# - ); - conn.execute_batch(&sql) - .context("rebuild Verifier Alliance file address map") + Ok(true) } fn verification_registry_import_exists(conn: &Connection, chain_id: u64) -> Result { @@ -908,6 +923,34 @@ fn verification_registry_import_exists(conn: &Connection, chain_id: u64) -> Resu Ok(count > 0) } +fn verification_registry_state_consistent(conn: &Connection, chain_id: u64) -> Result { + let (recorded, enriched, missing_positions): (Option, i64, i64) = conn.query_row( + r#" + SELECT + ( + SELECT verified_count + FROM verification_registry_imports + WHERE source = 'verifier_alliance' AND chain_id = ? + ), + ( + SELECT COUNT(*) + FROM enrichment + WHERE verification_source = 'verifier_alliance' AND chain_id = ? + ), + ( + SELECT COUNT(*) + FROM enrichment + WHERE verification_source = 'verifier_alliance' + AND chain_id = ? + AND (block_number IS NULL OR create_index IS NULL) + ) + "#, + params![chain_id, chain_id, chain_id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + )?; + Ok(recorded == Some(enriched.max(0) as u64) && missing_positions == 0) +} + fn va_verified_count(conn: &Connection, chain_id: u64) -> Result { let count: i64 = conn.query_row( "SELECT COUNT(*) FROM enrichment WHERE verification_source = 'verifier_alliance' AND chain_id = ?", @@ -985,16 +1028,157 @@ fn changed_va_file_entries( Ok(changed) } +/// Rows per bounded write transaction against the enrichment/file-address +/// tables. They carry several ART indexes whose maintenance is buffered in +/// memory until COMMIT — landing millions of rows in one transaction OOMs a +/// 4GB host at the finish line, so writes are sliced. +const VA_WRITE_SLICE_ROWS: usize = 250_000; + +fn replace_va_enrichment_in_slices(conn: &Connection, chain_id: u64) -> Result<()> { + let total = staged_row_count(conn, "va_affected_addresses")?; + let mut lo = 0i64; + while lo < total { + let hi = (lo + VA_WRITE_SLICE_ROWS as i64).min(total); + run_in_txn( + conn, + &format!( + r#" + DELETE FROM enrichment + WHERE verification_source = 'verifier_alliance' + AND chain_id = {chain_id} + AND contract_address IN ( + SELECT contract_address + FROM va_affected_addresses + WHERE rn > {lo} AND rn <= {hi} + ); + + INSERT INTO enrichment ( + contract_address, chain_id, is_verified, contract_name, + checked_at, verification_source, match_type, + block_number, create_index + ) + SELECT + verified.contract_address, + {chain_id}::UBIGINT, + true, + CAST(NULL AS VARCHAR), + COALESCE(verified.checked_at, CURRENT_TIMESTAMP), + 'verifier_alliance', + CASE + WHEN verified.runtime_match AND verified.creation_match + THEN 'runtime+creation' + WHEN verified.runtime_match THEN 'runtime' + WHEN verified.creation_match THEN 'creation' + WHEN verified.runtime_metadata_match + OR verified.creation_metadata_match THEN 'metadata' + ELSE 'verified' + END, + verified.block_number, + verified.create_index + FROM va_verified_contracts verified + JOIN va_affected_addresses affected + ON affected.contract_address = verified.contract_address + WHERE affected.rn > {lo} AND affected.rn <= {hi}; + "# + ), + "replace Verifier Alliance enrichment slice", + )?; + lo = hi; + } + Ok(()) +} + +fn run_in_txn(conn: &Connection, sql: &str, what: &str) -> Result<()> { + let result = conn + .execute_batch(&format!("BEGIN;\n{sql}\nCOMMIT;")) + .with_context(|| what.to_string()); + if result.is_err() { + let _ = conn.execute_batch("ROLLBACK;"); + } + result +} + +fn staged_row_count(conn: &Connection, table: &str) -> Result { + conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| { + row.get(0) + }) + .with_context(|| format!("count staged rows in {table}")) +} + +/// Run `make_sql(lo, hi)` (an INSERT selecting staging rows with +/// `rn > lo AND rn <= hi`) once per slice, each in its own transaction. +fn insert_in_slices(conn: &Connection, staging: &str, what: &str, make_sql: F) -> Result<()> +where + F: Fn(i64, i64) -> String, +{ + let total = staged_row_count(conn, staging)?; + let mut lo = 0i64; + while lo < total { + let hi = (lo + VA_WRITE_SLICE_ROWS as i64).min(total); + run_in_txn(conn, &make_sql(lo, hi), what)?; + lo = hi; + } + Ok(()) +} + +/// Delete rows matching `predicate` from `table` in bounded batches, one +/// autocommitted transaction each. +fn delete_in_slices(conn: &Connection, table: &str, predicate: &str, what: &str) -> Result<()> { + loop { + let deleted = conn + .execute( + &format!( + "DELETE FROM {table} WHERE rowid IN \ + (SELECT rowid FROM {table} WHERE {predicate} LIMIT {VA_WRITE_SLICE_ROWS})" + ), + [], + ) + .with_context(|| what.to_string())?; + if deleted == 0 { + return Ok(()); + } + } +} + fn replace_va_file_manifest( conn: &Connection, chain_id: u64, entries: &[VaFileEntry], ) -> Result<()> { + // Deleting a key and re-inserting it inside the same transaction trips + // DuckDB's ART unique-index check, so remove only paths that vanished + // from the export and upsert the rest in place. + if entries.is_empty() { + conn.execute( + "DELETE FROM verification_registry_files WHERE source = 'verifier_alliance' AND chain_id = ?", + params![chain_id], + ) + .context("clear Verifier Alliance file manifest")?; + return Ok(()); + } + let keep = entries + .iter() + .map(|entry| { + format!( + "('{}', '{}')", + entry.table_name.replace('\'', "''"), + entry.path_key.replace('\'', "''") + ) + }) + .collect::>() + .join(", "); conn.execute( - "DELETE FROM verification_registry_files WHERE source = 'verifier_alliance' AND chain_id = ?", + &format!( + r#" + DELETE FROM verification_registry_files + WHERE source = 'verifier_alliance' + AND chain_id = ? + AND (table_name, path) NOT IN (VALUES {keep}) + "# + ), params![chain_id], ) - .context("clear Verifier Alliance file manifest")?; + .context("prune Verifier Alliance file manifest")?; upsert_va_file_manifest_entries(conn, chain_id, entries) } @@ -1003,18 +1187,11 @@ fn upsert_va_file_manifest_entries( chain_id: u64, entries: &[VaFileEntry], ) -> Result<()> { + let mut entries = entries.to_vec(); + entries.sort_by_key(|entry| (entry.table_name, entry.path_key.clone())); + entries.dedup_by(|a, b| a.table_name == b.table_name && a.path_key == b.path_key); + for entry in entries { - conn.execute( - r#" - DELETE FROM verification_registry_files - WHERE source = 'verifier_alliance' - AND chain_id = ? - AND table_name = ? - AND path = ? - "#, - params![chain_id, entry.table_name, entry.path_key], - ) - .context("delete Verifier Alliance file manifest row")?; conn.execute( r#" INSERT INTO verification_registry_files ( @@ -1027,6 +1204,10 @@ fn upsert_va_file_manifest_entries( imported_at ) VALUES ('verifier_alliance', ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) + ON CONFLICT (source, chain_id, table_name, path) DO UPDATE SET + size_bytes = excluded.size_bytes, + modified_unix_ns = excluded.modified_unix_ns, + imported_at = excluded.imported_at "#, params![ chain_id, @@ -1036,7 +1217,7 @@ fn upsert_va_file_manifest_entries( entry.modified_unix_ns ], ) - .context("insert Verifier Alliance file manifest row")?; + .context("upsert Verifier Alliance file manifest row")?; } Ok(()) } diff --git a/src/main.rs b/src/main.rs index efcc2b2..81697a4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -3,7 +3,7 @@ use anyhow::Result; use tracing_subscriber::{fmt::time::ChronoLocal, EnvFilter}; -use blink::{cli, decode, extract::run_contracts, load, serve}; +use blink::{checkpoints, cli, decode, extract::run_contracts, load, serve}; #[tokio::main] async fn main() -> Result<()> { @@ -13,6 +13,7 @@ async fn main() -> Result<()> { cli::Commands::Contracts(args) => run_contracts(args).await, cli::Commands::Load(args) => load::run_load(args).await, cli::Commands::Decode(args) => decode::run_decode(args).await, + cli::Commands::Checkpoints(args) => checkpoints::run_checkpoints(args).await, cli::Commands::Serve(args) => serve::run_serve(args).await, } } diff --git a/src/serve.rs b/src/serve.rs index 35814d9..c4b7b90 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -4,12 +4,13 @@ //! Optional background tasks: //! - repeated `--rpc URL` flags poll one or more chain heads and extract //! newly produced blocks into separate `tail__chain_*` parquet files. +//! - `--verifier-alliance-dir` periodically downloads and incrementally +//! imports Verifier Alliance labels without restarting the server. //! -//! Serving model: every cacheable endpoint is stale-while-revalidate. A -//! cached entry is returned immediately no matter its age; entries past -//! their TTL trigger a background refresh (deduplicated per key). Combined -//! with the rollup tables in `db`, cold misses are native-table queries, so -//! the server starts serving instantly and warms itself in the background. +//! Serving model: the default dashboard is warmed before the listener binds, +//! then every cacheable endpoint is stale-while-revalidate. A cached entry is +//! returned immediately no matter its age; entries past their TTL trigger a +//! background refresh (deduplicated per key). //! //! Endpoints (all return JSON): //! - `GET /api/stats` — totals, verified pct, last block, verification coverage. @@ -175,6 +176,8 @@ async fn track_latency(State(state): State, req: Request, next: Next) const API_CACHE_TTL: Duration = Duration::from_secs(600); const HIGHEST_BLOCK_TTL: Duration = Duration::from_secs(5); const TAIL_START_DELAY: Duration = Duration::from_secs(15); +const VA_SYNC_START_DELAY: Duration = Duration::from_secs(30); +const VA_SYNC_MIN_INTERVAL: Duration = Duration::from_secs(900); const DEFAULT_COMPILER_LIMIT: u32 = 12; const DEFAULT_RECENT_LIMIT: u32 = 20; const INITIAL_DEPLOYS_RANGE: &str = "day"; @@ -342,6 +345,57 @@ where self.lookup(key).map(|(_, value)| value) } + /// Refresh a key without discarding its current value. If another task is + /// already refreshing the key, that work wins and this call is a no-op. + async fn refresh_if_idle(&self, key: K, make: F) -> Result + where + F: FnOnce() -> Fut, + Fut: Future> + Send, + { + let Ok(slot) = self.claim(key) else { + return Ok(false); + }; + let result = make().await; + if let Ok(value) = &result { + self.insert(key, value.clone()); + } + self.release(&key); + drop(slot); + result.map(|_| true) + } + + fn expire_where(&self, predicate: impl Fn(K) -> bool) { + let expired_at = Instant::now() + .checked_sub(API_CACHE_TTL + Duration::from_secs(1)) + .unwrap_or_else(Instant::now); + for (key, (stored_at, _)) in self + .inner + .values + .lock() + .expect("cache map poisoned") + .iter_mut() + { + if predicate(*key) { + *stored_at = expired_at; + } + } + } + + fn touch_where(&self, predicate: impl Fn(K) -> bool) { + let now = Instant::now(); + for (key, (stored_at, _)) in self + .inner + .values + .lock() + .expect("cache map poisoned") + .iter_mut() + { + if predicate(*key) { + *stored_at = now; + } + } + } + /// Claim the compute slot for `key`: the holder gets the sender (waiters /// wake when it drops); if already claimed, the receiver to wait on. fn claim(&self, key: K) -> Result, watch::Receiver<()>> { @@ -363,6 +417,61 @@ where } } +#[cfg(test)] +mod cache_tests { + use super::*; + + #[tokio::test] + async fn explicit_refresh_keeps_stale_value_available() { + let cache = CacheMap::::default(); + cache.insert(1, 10); + cache.expire_where(|key| key == 1); + + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (finish_tx, finish_rx) = tokio::sync::oneshot::channel(); + let refresh_cache = cache.clone(); + let refresh = tokio::spawn(async move { + refresh_cache + .refresh_if_idle(1, || async move { + let _ = started_tx.send(()); + let _ = finish_rx.await; + Ok(20) + }) + .await + }); + + started_rx.await.expect("refresh started"); + let stale = tokio::time::timeout( + Duration::from_millis(100), + cache.get_or_refresh(1, API_CACHE_TTL, || async { + Err(anyhow::anyhow!("must not start a duplicate refresh")) + }), + ) + .await + .expect("stale cache read must not block") + .expect("stale cache value"); + assert_eq!(stale, 10); + + finish_tx.send(()).expect("finish refresh"); + assert!(refresh + .await + .expect("refresh task") + .expect("refresh result")); + assert_eq!(cache.get(&1), Some(20)); + } + + #[test] + fn recognizes_errors_that_require_a_database_restart() { + assert!(is_fatal_database_error( + "FATAL Error: Corrupted ART index - likely the same row id was inserted twice" + )); + assert!(is_fatal_database_error( + "database has been invalidated because of a previous fatal error" + )); + assert!(!is_fatal_database_error("HTTP error 429 Too Many Requests")); + } +} + #[derive(Debug)] pub struct RuntimeState { read_only: bool, @@ -586,28 +695,17 @@ pub async fn run_serve(args: ServeArgs) -> Result<()> { .map(str::to_string) .collect::>(); let tail_enabled = !rpcs.is_empty() && !args.read_only; + let verifier_alliance_dir = args.verifier_alliance_dir.clone(); let runtime = Arc::new(RuntimeState::new( args.read_only, tail_enabled, args.tail_interval_secs.max(15), )); let cache = Arc::new(ApiCache::default()); + seed_runtime_snapshot(&db, &runtime).await; - // Rollup-backed queries are fast enough to serve cold, so warming happens - // off the startup path — the listener binds immediately. - tokio::spawn(prewarm_initial_dashboard_cache(db.clone(), cache.clone())); - if !args.read_only { - // Materialized SQL-explorer table; queries fall back to a live join - // until it's ready. - let db_explorer = db.clone(); - tokio::spawn(async move { - match db_explorer.refresh_explorer().await { - Ok(true) => tracing::info!("sql explorer table ready"), - Ok(false) => {} - Err(err) => tracing::warn!("sql explorer table rebuild failed: {:#}", err), - } - }); - } + prewarm_initial_dashboard_cache(db.clone(), cache.clone()).await?; + let state = AppState { db: db.clone(), runtime: runtime.clone(), @@ -650,6 +748,17 @@ pub async fn run_serve(args: ServeArgs) -> Result<()> { tracing::info!("serving blink dashboard on http://{}", addr); tracing::info!(" data dir: {}", args.data_dir.display()); if !args.read_only { + // The materialized SQL-explorer table is maintenance work. Start it + // only after the dashboard cache is ready so it cannot contend with + // cold aggregate queries during startup. + let db_explorer = db.clone(); + tokio::spawn(async move { + match db_explorer.refresh_explorer().await { + Ok(true) => tracing::info!("sql explorer table ready"), + Ok(false) => {} + Err(err) => tracing::warn!("sql explorer table rebuild failed: {:#}", err), + } + }); spawn_tail_loops( db.clone(), runtime.clone(), @@ -663,12 +772,88 @@ pub async fn run_serve(args: ServeArgs) -> Result<()> { data_dir: args.data_dir.clone(), }, ); + if let Some(verifier_alliance_dir) = verifier_alliance_dir { + spawn_verifier_alliance_sync_loop( + db.clone(), + cache.clone(), + verifier_alliance_dir, + Duration::from_secs(args.verifier_alliance_sync_interval_secs), + ); + } + } else if verifier_alliance_dir.is_some() { + tracing::warn!("--read-only is set; automatic Verifier Alliance sync is disabled"); } axum::serve(listener, app) .await .context("axum server failed") } +fn spawn_verifier_alliance_sync_loop( + db: Db, + cache: Arc, + verifier_alliance_dir: PathBuf, + interval: Duration, +) { + let interval = interval.max(VA_SYNC_MIN_INTERVAL); + tracing::info!( + "automatic Verifier Alliance sync enabled (dir={}, interval={}s)", + verifier_alliance_dir.display(), + interval.as_secs() + ); + tokio::spawn(async move { + tokio::time::sleep(VA_SYNC_START_DELAY).await; + loop { + let started = Instant::now(); + defer_dashboard_cache_refreshes(&cache); + tracing::info!("syncing Verifier Alliance dataset from object storage"); + match crate::va_sync::sync_verifier_alliance_files(&verifier_alliance_dir).await { + Ok(()) => { + for chain in chains::supported_chains() { + match db + .import_verifier_alliance(verifier_alliance_dir.clone(), chain.chain_id) + .await + { + Ok(changed) => { + if changed { + refresh_verification_cache(&db, &cache, chain.chain_id).await; + } + tracing::info!( + "Verifier Alliance data current for chain_id={}{}", + chain.chain_id, + if changed { " (updated)" } else { "" } + ); + } + Err(error) => tracing::warn!( + "Verifier Alliance import failed for chain_id={}: {:#}", + chain.chain_id, + error + ), + } + } + tracing::info!( + "Verifier Alliance sync completed in {:.1}s", + started.elapsed().as_secs_f64() + ); + } + Err(error) => tracing::warn!("Verifier Alliance download failed: {:#}", error), + } + tokio::time::sleep(interval).await; + } + }); +} + +fn defer_dashboard_cache_refreshes(cache: &ApiCache) { + cache.stats.touch_where(|_| true); + cache.deploys.touch_where(|_| true); + cache.verified.touch_where(|_| true); + cache.bytecode_sizes.touch_where(|_| true); + cache.compilers.touch_where(|_| true); + cache.recent.touch_where(|_| true); + cache.languages.touch_where(|_| true); + cache.standards.touch_where(|_| true); + cache.highest_blocks.touch_where(|_| true); +} + fn dashboard_cors_layer() -> CorsLayer { CorsLayer::new() .allow_origin(HeaderValue::from_static("https://blink.mirageprivacy.com")) @@ -855,10 +1040,11 @@ fn normalized_cache_range( ) -> (Option, Option, Option) { if uses_relative_preset_window(q) { let range_code = q.range.as_deref().map(relative_range_code); - let end_bucket = window - .block_range - .map(|(_, end)| end / bucket_blocks.max(1)); - return (Some(bucket_blocks.max(1)), range_code, end_bucket); + // A relative range is one logical cache entry as the chain advances. + // Keeping the key stable lets stale-while-revalidate serve the last + // result immediately instead of creating a cold miss at each bucket + // boundary. + return (Some(bucket_blocks.max(1)), range_code, None); } ( None, @@ -1342,7 +1528,7 @@ struct LanguagesResponse { languages: Vec, } -async fn prewarm_initial_dashboard_cache(db: Db, cache: Arc) { +async fn prewarm_initial_dashboard_cache(db: Db, cache: Arc) -> Result<()> { let started = Instant::now(); tracing::info!("warming dashboard cache in background"); for chain in chains::supported_chains() { @@ -1355,10 +1541,14 @@ async fn prewarm_initial_dashboard_cache(db: Db, cache: Arc) { ) .await; } + db.health_check() + .await + .context("dashboard database was invalidated during cache warm")?; tracing::info!( "dashboard cache warmed in {:.1}s", started.elapsed().as_secs_f64() ); + Ok(()) } fn spawn_tail_loops( @@ -1395,15 +1585,23 @@ async fn prewarm_chain_dashboard_cache( include_widgets: bool, ) { let started = Instant::now(); - let highest = match db.highest_contract_block(chain_id).await { - Ok(Some(block)) => block, - Ok(None) => 0, + let db_for_highest = db.clone(); + let highest = match cache + .highest_blocks + .get_or_refresh(chain_id, HIGHEST_BLOCK_TTL, move || async move { + Ok(db_for_highest + .highest_contract_block(chain_id) + .await? + .unwrap_or(0)) + }) + .await + { + Ok(block) => block, Err(err) => { log_prewarm_error(chain_id, "highest block", err); return; } }; - cache.highest_blocks.insert(chain_id, highest); for range in chart_ranges { prewarm_chart_range(db, cache, chain_id, highest, range).await; @@ -1418,16 +1616,23 @@ async fn prewarm_chain_dashboard_cache( let aggregate_window = parse_time_series_window(&aggregate_query, chain_id, highest); let aggregate_key = range_cache_key_for_query(chain_id, &aggregate_query, aggregate_window); - match db.stats(chain_id).await { - Ok(stats) => cache.stats.insert(chain_id, stats), + match cache + .stats + .refresh_if_idle(chain_id, || db.stats(chain_id)) + .await + { + Ok(_) => {} Err(err) => log_prewarm_error(chain_id, "stats", err), } - match db - .bytecode_size_distribution(chain_id, aggregate_window.block_range) + match cache + .bytecode_sizes + .refresh_if_idle(aggregate_key, || { + db.bytecode_size_distribution(chain_id, aggregate_window.block_range) + }) .await { - Ok(bins) => cache.bytecode_sizes.insert(aggregate_key, bins), + Ok(_) => {} Err(err) => log_prewarm_error(chain_id, "bytecode sizes", err), } @@ -1443,34 +1648,43 @@ async fn prewarm_chain_dashboard_cache( start_block: compiler_start_block, end_block: compiler_end_block, }; - match async { - Ok::<_, anyhow::Error>(( - db.top_compilers( - chain_id, - DEFAULT_COMPILER_LIMIT, - aggregate_window.block_range, - ) - .await?, - db.compiler_version_total(chain_id, aggregate_window.block_range) + match cache + .compilers + .refresh_if_idle(compiler_key, || async { + Ok(( + db.top_compilers( + chain_id, + DEFAULT_COMPILER_LIMIT, + aggregate_window.block_range, + ) .await?, - )) - } - .await + db.compiler_version_total(chain_id, aggregate_window.block_range) + .await?, + )) + }) + .await { - Ok(compilers) => cache.compilers.insert(compiler_key, compilers), + Ok(_) => {} Err(err) => log_prewarm_error(chain_id, "compilers", err), } - match db.language_distribution(chain_id).await { - Ok(languages) => cache.languages.insert(chain_id, languages), + match cache + .languages + .refresh_if_idle(chain_id, || db.language_distribution(chain_id)) + .await + { + Ok(_) => {} Err(err) => log_prewarm_error(chain_id, "languages", err), } - match db - .standards_breakdown(chain_id, aggregate_window.block_range) + match cache + .standards + .refresh_if_idle(aggregate_key, || { + db.standards_breakdown(chain_id, aggregate_window.block_range) + }) .await { - Ok(standards) => cache.standards.insert(aggregate_key, standards), + Ok(_) => {} Err(err) => log_prewarm_error(chain_id, "standards", err), } @@ -1480,11 +1694,14 @@ async fn prewarm_chain_dashboard_cache( before_block: None, before_create_index: None, }; - match db - .recent_contracts(chain_id, DEFAULT_RECENT_LIMIT, None) + match cache + .recent + .refresh_if_idle(recent_key, || { + db.recent_contracts(chain_id, DEFAULT_RECENT_LIMIT, None) + }) .await { - Ok(recent) => cache.recent.insert(recent_key, recent), + Ok(_) => {} Err(err) => log_prewarm_error(chain_id, "recent deployments", err), } } @@ -1505,23 +1722,96 @@ async fn prewarm_chart_range(db: &Db, cache: &ApiCache, chain_id: u64, highest: let window = parse_time_series_window(&query, chain_id, highest); let cache_key = bucket_cache_key(chain_id, &query, window); - match db - .deploys_over_time(chain_id, window.bucket_blocks, window.block_range) + match cache + .deploys + .refresh_if_idle(cache_key, || { + db.deploys_over_time(chain_id, window.bucket_blocks, window.block_range) + }) .await { - Ok(buckets) => cache.deploys.insert(cache_key, buckets), + Ok(_) => {} Err(err) => log_prewarm_error(chain_id, &format!("deployments {range}"), err), } - match db - .verified_ratio_over_time(chain_id, window.bucket_blocks, window.block_range) + match cache + .verified + .refresh_if_idle(cache_key, || { + db.verified_ratio_over_time(chain_id, window.bucket_blocks, window.block_range) + }) .await { - Ok(buckets) => cache.verified.insert(cache_key, buckets), + Ok(_) => {} Err(err) => log_prewarm_error(chain_id, &format!("verification {range}"), err), } } +async fn refresh_verification_cache(db: &Db, cache: &ApiCache, chain_id: u64) { + cache.stats.expire_where(|key| key == chain_id); + cache.verified.expire_where(|key| key.chain_id == chain_id); + cache.recent.expire_where(|key| key.chain_id == chain_id); + + match cache + .stats + .refresh_if_idle(chain_id, || db.stats(chain_id)) + .await + { + Ok(_) => {} + Err(error) => log_prewarm_error(chain_id, "stats after VA sync", error), + } + + let db_for_highest = db.clone(); + let highest = match cache + .highest_blocks + .get_or_refresh(chain_id, HIGHEST_BLOCK_TTL, move || async move { + Ok(db_for_highest + .highest_contract_block(chain_id) + .await? + .unwrap_or(0)) + }) + .await + { + Ok(block) => block, + Err(error) => { + log_prewarm_error(chain_id, "highest block after VA sync", error); + return; + } + }; + let query = BucketQuery { + chain_id: Some(chain_id), + range: Some(INITIAL_VERIFIED_RANGE.to_string()), + ..BucketQuery::default() + }; + let window = parse_time_series_window(&query, chain_id, highest); + let cache_key = bucket_cache_key(chain_id, &query, window); + match cache + .verified + .refresh_if_idle(cache_key, || { + db.verified_ratio_over_time(chain_id, window.bucket_blocks, window.block_range) + }) + .await + { + Ok(_) => {} + Err(error) => log_prewarm_error(chain_id, "verification after VA sync", error), + } + + let recent_key = RecentCacheKey { + chain_id, + limit: DEFAULT_RECENT_LIMIT, + before_block: None, + before_create_index: None, + }; + match cache + .recent + .refresh_if_idle(recent_key, || { + db.recent_contracts(chain_id, DEFAULT_RECENT_LIMIT, None) + }) + .await + { + Ok(_) => {} + Err(error) => log_prewarm_error(chain_id, "recent deployments after VA sync", error), + } +} + fn log_prewarm_error(chain_id: u64, label: &str, err: anyhow::Error) { tracing::warn!( "dashboard cache prewarm failed (chain_id={}, {}): {:#}", @@ -1531,6 +1821,80 @@ fn log_prewarm_error(chain_id: u64, label: &str, err: anyhow::Error) { ); } +fn is_relative_range(start_block: Option, end_block: Option) -> bool { + end_block.is_none() && matches!(start_block, Some(1..=5)) +} + +fn is_live_relative_range(start_block: Option, end_block: Option) -> bool { + end_block.is_none() && matches!(start_block, Some(1 | 2)) +} + +async fn refresh_tail_dashboard_cache(db: &Db, cache: &ApiCache, chain_id: u64) { + // Keep existing values available while refreshes run. The time-series + // rollups are cheap enough to refresh each tail; metadata widgets only + // expire their 1H/1D views and refresh on demand because their joins are + // much heavier on the production dataset. + cache.stats.expire_where(|key| key == chain_id); + cache.deploys.expire_where(|key| { + key.chain_id == chain_id && is_relative_range(key.start_block, key.end_block) + }); + cache.verified.expire_where(|key| { + key.chain_id == chain_id && is_relative_range(key.start_block, key.end_block) + }); + cache.bytecode_sizes.expire_where(|key| { + key.chain_id == chain_id && is_live_relative_range(key.start_block, key.end_block) + }); + cache.compilers.expire_where(|key| { + key.chain_id == chain_id && is_live_relative_range(key.start_block, key.end_block) + }); + cache.standards.expire_where(|key| { + key.chain_id == chain_id && is_live_relative_range(key.start_block, key.end_block) + }); + cache.recent.expire_where(|key| { + key.chain_id == chain_id && key.before_block.is_none() && key.before_create_index.is_none() + }); + + let highest = match db.highest_contract_block(chain_id).await { + Ok(Some(block)) => block, + Ok(None) => 0, + Err(error) => { + log_prewarm_error(chain_id, "highest block after tail", error); + return; + } + }; + cache.highest_blocks.insert(chain_id, highest); + + match cache + .stats + .refresh_if_idle(chain_id, || db.stats(chain_id)) + .await + { + Ok(_) => {} + Err(error) => log_prewarm_error(chain_id, "stats after tail", error), + } + + for range in [INITIAL_DEPLOYS_RANGE, INITIAL_VERIFIED_RANGE] { + prewarm_chart_range(db, cache, chain_id, highest, range).await; + } + + let recent_key = RecentCacheKey { + chain_id, + limit: DEFAULT_RECENT_LIMIT, + before_block: None, + before_create_index: None, + }; + match cache + .recent + .refresh_if_idle(recent_key, || { + db.recent_contracts(chain_id, DEFAULT_RECENT_LIMIT, None) + }) + .await + { + Ok(_) => {} + Err(error) => log_prewarm_error(chain_id, "recent deployments after tail", error), + } +} + async fn background_tail_loop( db: Db, config: TailLoopConfig, @@ -1579,21 +1943,11 @@ async fn background_tail_loop( report.end_block, report.rows ); - // New blocks just landed in the rollups: overwrite this - // chain's cached dashboard entries right away instead of - // letting the top-right block number and the recent table - // trail by up to a cache TTL. These are rollup queries — - // milliseconds — so doing it every tick is cheap. + // Refresh the tail-sensitive entries without deleting stale + // values that active dashboard requests can serve. if let Some(chain_id) = chain_id { if report.rows > 0 { - prewarm_chain_dashboard_cache( - &db, - &cache, - chain_id, - &[INITIAL_DEPLOYS_RANGE, INITIAL_VERIFIED_RANGE], - true, - ) - .await; + refresh_tail_dashboard_cache(&db, &cache, chain_id).await; } } } @@ -1605,8 +1959,23 @@ async fn background_tail_loop( let msg = format!("{:#}", err); runtime.mark_tail_error(chain_id, msg.clone()).await; tracing::warn!("tail failed: {}", msg); + if is_fatal_database_error(&msg) { + tracing::error!( + "tail loop stopped for chain_id={}: DuckDB was invalidated; restart blink serve", + chain_id + .map(|chain_id| chain_id.to_string()) + .unwrap_or_else(|| "unknown".to_string()) + ); + break; + } } } tokio::time::sleep(config.interval).await; } } + +fn is_fatal_database_error(message: &str) -> bool { + message.contains("database has been invalidated") + || message.contains("Corrupted ART index") + || message.contains("FATAL Error") +} diff --git a/src/va_sync.rs b/src/va_sync.rs new file mode 100644 index 0000000..4209899 --- /dev/null +++ b/src/va_sync.rs @@ -0,0 +1,61 @@ +//! Verifier Alliance parquet synchronization used by the serve background job. + +use std::path::Path; + +use anyhow::{anyhow, Context, Result}; +use tokio::process::Command; + +pub async fn sync_verifier_alliance_files(root: &Path) -> Result<()> { + tokio::fs::create_dir_all(root.join("contract_deployments")) + .await + .with_context(|| format!("create Verifier Alliance dir {}", root.display()))?; + tokio::fs::create_dir_all(root.join("verified_contracts")) + .await + .with_context(|| format!("create Verifier Alliance dir {}", root.display()))?; + + sync_prefix( + "s3://verifier-alliance-parquet-export/v2/contract_deployments/", + &root.join("contract_deployments"), + ) + .await?; + sync_prefix( + "s3://verifier-alliance-parquet-export/v2/verified_contracts/", + &root.join("verified_contracts"), + ) + .await +} + +async fn sync_prefix(source: &str, destination: &Path) -> Result<()> { + let mut command = Command::new("aws"); + command.kill_on_drop(true).args([ + "s3", + "sync", + source, + destination + .to_str() + .ok_or_else(|| anyhow!("non-UTF-8 VA path {}", destination.display()))?, + "--endpoint-url", + "https://storage.googleapis.com", + "--no-sign-request", + "--only-show-errors", + ]); + let output = tokio::time::timeout(std::time::Duration::from_secs(30 * 60), command.output()) + .await + .context("aws s3 sync timed out after 30 minutes")??; + if output.status.success() { + return Ok(()); + } + + let stderr = String::from_utf8_lossy(&output.stderr); + let stdout = String::from_utf8_lossy(&output.stdout); + let details = if stderr.trim().is_empty() { + stdout.trim() + } else { + stderr.trim() + }; + Err(anyhow!( + "aws s3 sync failed for {source} ({}): {}", + output.status, + details + )) +} diff --git a/tests/blocks.rs b/tests/blocks.rs index 1ebdf7c..21624b3 100644 --- a/tests/blocks.rs +++ b/tests/blocks.rs @@ -1,11 +1,63 @@ //! Block <-> timestamp mapping tests. -use blink::blocks::{block_number_at_time, block_timestamp}; -use blink::chains::GNOSIS_CHAIN_ID; +use std::{collections::HashMap, sync::Once}; + +use blink::blocks::{block_number_at_time, block_timestamp, replace_runtime_checkpoints}; +use blink::chains::{ETHEREUM_CHAIN_ID, GNOSIS_CHAIN_ID}; use chrono::{Datelike, TimeZone}; +fn install_checkpoints() { + static INSTALL: Once = Once::new(); + INSTALL.call_once(|| { + replace_runtime_checkpoints(HashMap::from([ + ( + ETHEREUM_CHAIN_ID, + vec![(24_000_000, 1_765_584_371), (25_497_188, 1_783_626_143)], + ), + ( + GNOSIS_CHAIN_ID, + vec![(46_762_380, 1_781_798_438), (47_119_233, 1_783_626_150)], + ), + ])); + }); +} + +#[test] +fn ethereum_recent_blocks_map_to_current_calendar_dates() { + install_checkpoints(); + // Real anchor: block 25,497,188 was mined 2026-07-09 19:42:23 UTC. The + // pre-fix 12s-flat extrapolation from the merge put this ten days early. + assert_eq!( + block_timestamp(ETHEREUM_CHAIN_ID, 25_497_188), + chrono::Utc + .with_ymd_and_hms(2026, 7, 9, 19, 42, 23) + .unwrap() + ); + // Blocks between checkpoints interpolate to the right part of the + // calendar: block 24.5M sits a third of the way through the + // 2025-12-13 → 2026-07-09 span, i.e. late February 2026. + let ts = block_timestamp(ETHEREUM_CHAIN_ID, 24_500_000); + assert_eq!(ts.date_naive().year(), 2026); + assert_eq!(ts.date_naive().month(), 2); +} + +#[test] +fn ethereum_extrapolation_past_newest_checkpoint_uses_measured_rate() { + install_checkpoints(); + // A day's worth of blocks past the newest checkpoint should land ~a day + // later, not drift with the ideal 12s slot time. + let later = block_timestamp(ETHEREUM_CHAIN_ID, 25_497_188 + 7_146); // ≈1 day at 12.09s + let anchor = block_timestamp(ETHEREUM_CHAIN_ID, 25_497_188); + let delta = (later - anchor).num_seconds(); + assert!((86_000..87_000).contains(&delta), "delta {delta}"); + + let round_trip = block_number_at_time(ETHEREUM_CHAIN_ID, later); + assert!((round_trip as i64 - (25_497_188 + 7_146) as i64).abs() < 5); +} + #[test] fn gnosis_recent_blocks_map_to_current_calendar_dates() { + install_checkpoints(); assert_eq!( block_timestamp(GNOSIS_CHAIN_ID, 46_762_380), chrono::Utc @@ -16,6 +68,7 @@ fn gnosis_recent_blocks_map_to_current_calendar_dates() { #[test] fn gnosis_recent_time_ranges_convert_back_to_blocks() { + install_checkpoints(); let block = block_number_at_time( GNOSIS_CHAIN_ID, chrono::Utc.with_ymd_and_hms(2026, 6, 19, 0, 0, 0).unwrap(), diff --git a/tests/checkpoints.rs b/tests/checkpoints.rs new file mode 100644 index 0000000..6852773 --- /dev/null +++ b/tests/checkpoints.rs @@ -0,0 +1,59 @@ +//! Persistent block-time checkpoint tests. + +use std::{ + fs, + path::PathBuf, + time::{SystemTime, UNIX_EPOCH}, +}; + +use blink::{blocks, db::Db}; + +struct TestDir(PathBuf); + +impl TestDir { + fn new() -> Self { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = std::env::temp_dir().join(format!( + "blink_checkpoint_test_{}_{}", + std::process::id(), + unique + )); + fs::create_dir_all(&path).unwrap(); + Self(path) + } +} + +impl Drop for TestDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +#[tokio::test] +async fn checkpoints_survive_reopen_and_drive_interpolation() { + const CHAIN_ID: u64 = 9_999; + let dir = TestDir::new(); + { + let db = Db::open_with_mode(&dir.0, "*.parquet", false).unwrap(); + db.record_block_checkpoint(CHAIN_ID, 100, 1_000) + .await + .unwrap(); + db.record_block_checkpoint(CHAIN_ID, 200, 2_000) + .await + .unwrap(); + } + + blocks::replace_runtime_checkpoints(Default::default()); + let _db = Db::open_with_mode(&dir.0, "*.parquet", false).unwrap(); + assert_eq!(blocks::block_timestamp(CHAIN_ID, 150).timestamp(), 1_500); + assert_eq!( + blocks::block_number_at_time( + CHAIN_ID, + chrono::DateTime::from_timestamp(1_750, 0).unwrap() + ), + 175 + ); +} diff --git a/tests/db_dashboard.rs b/tests/db.rs similarity index 81% rename from tests/db_dashboard.rs rename to tests/db.rs index a03a2a9..435a687 100644 --- a/tests/db_dashboard.rs +++ b/tests/db.rs @@ -211,6 +211,59 @@ async fn stats_and_recent_include_parquet_rows_newer_than_zellic() { assert!(!recent.has_more); } +#[tokio::test] +async fn verification_counts_partition_deduplicated_deployments_without_registry_marker() { + let dir = TestDir::new("verification_stats_partition_total"); + write_contract_parquet( + &dir.path.join("contracts__0000000100__0000000100.parquet"), + 100, + 0x01, + ETHEREUM_CHAIN_ID, + ); + write_contract_parquet( + &dir.path.join("contracts__0000000200__0000000200.parquet"), + 200, + 0x02, + ETHEREUM_CHAIN_ID, + ); + + let db = Db::open_with_mode(&dir.path, "*.parquet", false).unwrap(); + db.execute_batch( + r#" + INSERT INTO enrichment ( + contract_address, chain_id, is_verified, checked_at, block_number, create_index + ) VALUES + (unhex(repeat('03', 20)), 1, true, CURRENT_TIMESTAMP, 100, 0), + (unhex(repeat('03', 20)), 1, true, CURRENT_TIMESTAMP, 100, 0), + (unhex(repeat('04', 20)), 1, false, CURRENT_TIMESTAMP, 200, 0), + (unhex(repeat('ff', 20)), 1, true, CURRENT_TIMESTAMP, 200, 0) + "# + .to_string(), + ) + .await + .unwrap(); + + let stats = db.stats(ETHEREUM_CHAIN_ID).await.unwrap(); + assert_eq!(stats.total_contracts, 2); + assert_eq!(stats.verified_count, 1); + assert_eq!(stats.unverified_count, 1); + assert_eq!(stats.verified_count + stats.unverified_count, 2); + assert_eq!(stats.verified_pct, 50.0); + assert_eq!(stats.enrichment_coverage_pct, 100.0); + + let buckets = db + .verified_ratio_over_time(ETHEREUM_CHAIN_ID, 100, Some((100, 200))) + .await + .unwrap(); + assert_eq!(buckets.len(), 2); + assert_eq!((buckets[0].verified, buckets[0].unverified), (1, 0)); + assert_eq!((buckets[1].verified, buckets[1].unverified), (0, 1)); + assert!(buckets.iter().all(|bucket| bucket.unknown == 0)); + assert!(buckets + .iter() + .all(|bucket| bucket.verified + bucket.unverified == 1)); +} + #[tokio::test] async fn rollups_are_idempotent_across_reopens() { let dir = TestDir::new("rollup_idempotent_reopen"); @@ -282,8 +335,8 @@ async fn chart_queries_deduplicate_overlapping_parquet_deployments() { .unwrap(); assert_eq!(verified.len(), 1); assert_eq!(verified[0].verified, 0); - assert_eq!(verified[0].unverified, 0); - assert_eq!(verified[0].unknown, 1); + assert_eq!(verified[0].unverified, 1); + assert_eq!(verified[0].unknown, 0); db.execute_batch( r#" @@ -499,6 +552,110 @@ async fn explorer_materialization_stays_correct_and_fresh() { ); } +#[tokio::test] +async fn live_decode_updates_ranged_aggregates_beyond_explorer_snapshot() { + let dir = TestDir::new("live_decode_after_explorer_snapshot"); + write_contract_parquet( + &dir.path + .join("contracts__chain_0000000001__0000000200__0000000200.parquet"), + 200, + 0x03, + 1, + ); + + let db = Db::open_with_mode(&dir.path, "*.parquet", false).unwrap(); + assert!(db.refresh_explorer().await.unwrap()); + + write_contract_parquet( + &dir.path + .join("tail__chain_0000000001__0000000400__0000000400.parquet"), + 400, + 0x33, + 1, + ); + db.refresh().await.unwrap(); + + // CBOR { "solc": h'000814' } followed by its two-byte length suffix. + let solc_0_8_20 = vec![ + 0xa1, 0x64, 0x73, 0x6f, 0x6c, 0x63, 0x43, 0x00, 0x08, 0x14, 0x00, 0x0a, + ]; + let live_hash = make_bytes(0x39, 32); + assert_eq!( + db.decode_live_bytecodes(vec![(live_hash.clone(), solc_0_8_20.clone())]) + .await + .unwrap(), + 1 + ); + assert_eq!( + db.decode_live_bytecodes(vec![(live_hash, solc_0_8_20)]) + .await + .unwrap(), + 0, + "replaying a tail bytecode must not duplicate hash metadata" + ); + + let compilers = db.top_compilers(1, 12, Some((400, 400))).await.unwrap(); + assert_eq!(compilers.len(), 1); + assert_eq!(compilers[0].compiler_version, "0.8.20"); + assert_eq!(compilers[0].count, 1); + assert_eq!( + db.compiler_version_total(1, Some((400, 400))) + .await + .unwrap(), + 1 + ); + + let standards = db.standards_breakdown(1, Some((400, 400))).await.unwrap(); + assert_eq!(standards.total_decoded, 1); + + let explorer = db + .query_sql( + "SELECT compiler_version FROM contract_metadata WHERE block_number = 400".to_string(), + 10, + Some(1), + ) + .await + .unwrap(); + assert_eq!(explorer.rows, vec![vec![serde_json::json!("0.8.20")]]); +} + +/// The default explorer query asks for newest deployments first. Keep the +/// materialized table in that physical order so DuckDB's Top-N scan can prune +/// old row groups instead of walking the chain's full history. +#[tokio::test] +async fn explorer_materialization_is_clustered_newest_first() { + let dir = TestDir::new("explorer_newest_first"); + write_multi_block_parquet( + &dir.path + .join("contracts__chain_0000000001__0000000100__0000000300.parquet"), + &[100, 300, 200], + ETHEREUM_CHAIN_ID, + ); + + let db = Db::open_with_mode(&dir.path, "*.parquet", false).unwrap(); + assert!(db.refresh_explorer().await.unwrap()); + + let rows = db + .query_sql( + "SELECT block_number FROM contract_metadata_native WHERE chain_id = 1".to_string(), + 10, + None, + ) + .await + .unwrap(); + assert_eq!( + rows.rows + .iter() + .map(|row| row[0].clone()) + .collect::>(), + vec![ + serde_json::json!(300), + serde_json::json!(200), + serde_json::json!(100), + ] + ); +} + /// While a schema-upgrading explorer rebuild runs (or after a failed one), /// the on-disk table is the previous generation without newer columns — /// aggregates must fall back to the join path instead of binder-erroring. @@ -664,9 +821,21 @@ async fn ranged_code_aggregates_are_exact_across_bucket_boundaries() { .unwrap(); assert_eq!(sizes.iter().map(|bin| bin.count).sum::(), 12); - // Once the materialized explorer table exists, the same endpoints switch - // to denormalized per-deployment scans — results must be identical. + // The SQL Explorer materialization must not become the aggregate source: + // it has one row per deployment and is far too large for dashboard scans. assert!(db.refresh_explorer().await.unwrap()); + db.execute_batch( + r#" + UPDATE contract_metadata_native + SET compiler_version = 'explorer-only', + language = 'explorer-only', + uses_push0 = false, + has_source_hash = false + "# + .to_string(), + ) + .await + .unwrap(); assert_eq!( db.compiler_version_total(1, Some((5_000, 25_000))) .await @@ -690,6 +859,7 @@ async fn ranged_code_aggregates_are_exact_across_bucket_boundaries() { .await .unwrap(); assert_eq!(compilers.len(), 1); + assert_eq!(compilers[0].compiler_version, "0.8.24"); assert_eq!(compilers[0].count, 12); let standards = db .standards_breakdown(1, Some((5_000, 25_000))) diff --git a/tests/serve_windows.rs b/tests/serve_windows.rs index 388b094..b7c22e8 100644 --- a/tests/serve_windows.rs +++ b/tests/serve_windows.rs @@ -85,10 +85,10 @@ fn range_end_block_moves_visible_window() { } #[test] -fn relative_preset_cache_key_survives_tail_moves_inside_bucket() { +fn relative_preset_cache_key_survives_tail_moves_across_bucket_boundary() { let q = query(Some("day"), None); let first = parse_time_series_window(&q, ETHEREUM_CHAIN_ID, 20_000_001); - let second = parse_time_series_window(&q, ETHEREUM_CHAIN_ID, 20_000_099); + let second = parse_time_series_window(&q, ETHEREUM_CHAIN_ID, 20_000_401); assert_ne!(first.block_range, second.block_range); assert_eq!( diff --git a/tests/va_sync.rs b/tests/va_sync.rs new file mode 100644 index 0000000..1cfa190 --- /dev/null +++ b/tests/va_sync.rs @@ -0,0 +1,225 @@ +//! In-process Verifier Alliance import tests. + +use std::{ + fs, + path::{Path, PathBuf}, + time::{SystemTime, UNIX_EPOCH}, +}; + +use blink::db::Db; +use duckdb::Connection; + +struct TestDir(PathBuf); + +impl TestDir { + fn new() -> Self { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = std::env::temp_dir().join(format!( + "blink_va_sync_test_{}_{}", + std::process::id(), + unique + )); + fs::create_dir_all(&path).unwrap(); + Self(path) + } +} + +impl Drop for TestDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} + +fn sql_path(path: &Path) -> String { + path.display().to_string().replace('\'', "''") +} + +fn write_va_fixture(root: &Path) { + let deployments = root.join("contract_deployments"); + let verified = root.join("verified_contracts"); + fs::create_dir_all(&deployments).unwrap(); + fs::create_dir_all(&verified).unwrap(); + + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch(&format!( + r#" + COPY ( + SELECT + 1::UBIGINT AS id, + 1::UBIGINT AS chain_id, + unhex(repeat('11', 20)) AS address, + 100::BIGINT AS block_number, + 0::BIGINT AS transaction_index + ) TO '{}' (FORMAT PARQUET); + + COPY ( + SELECT + 1::UBIGINT AS deployment_id, + TIMESTAMP '2026-01-01 00:00:00' AS created_at, + true AS runtime_match, + false AS creation_match, + false AS runtime_metadata_match, + false AS creation_metadata_match + ) TO '{}' (FORMAT PARQUET); + "#, + sql_path(&deployments.join("contract_deployments_0_1000.parquet")), + sql_path(&verified.join("verified_contracts_0_1000.parquet")), + )) + .unwrap(); +} + +fn remove_va_verification(root: &Path) { + let path = root + .join("verified_contracts") + .join("verified_contracts_0_1000.parquet"); + fs::remove_file(&path).unwrap(); + + let conn = Connection::open_in_memory().unwrap(); + conn.execute_batch(&format!( + r#" + COPY ( + SELECT + 1::UBIGINT AS deployment_id, + TIMESTAMP '2026-01-01 00:00:00' AS created_at, + true AS runtime_match, + false AS creation_match, + false AS runtime_metadata_match, + false AS creation_metadata_match + WHERE false + ) TO '{}' (FORMAT PARQUET); + "#, + sql_path(&path), + )) + .unwrap(); +} + +#[tokio::test] +async fn running_server_writer_imports_va_incrementally_without_reopen() { + let data = TestDir::new(); + let va = TestDir::new(); + write_va_fixture(&va.0); + + let db = Db::open_with_mode(&data.0, "*.parquet", false).unwrap(); + assert!(db.import_verifier_alliance(va.0.clone(), 1).await.unwrap()); + + let result = db + .query_sql( + "SELECT COUNT(*) FROM enrichment WHERE chain_id = 1 AND is_verified".to_string(), + 10, + None, + ) + .await + .unwrap(); + assert_eq!(result.rows, vec![vec![serde_json::json!(1)]]); + + assert!(!db.import_verifier_alliance(va.0.clone(), 1).await.unwrap()); +} + +#[tokio::test] +async fn changed_va_partition_replaces_previous_enrichment_rows() { + let data = TestDir::new(); + let va = TestDir::new(); + write_va_fixture(&va.0); + + let db = Db::open_with_mode(&data.0, "*.parquet", false).unwrap(); + assert!(db.import_verifier_alliance(va.0.clone(), 1).await.unwrap()); + + remove_va_verification(&va.0); + assert!(db.import_verifier_alliance(va.0.clone(), 1).await.unwrap()); + + let result = db + .query_sql( + "SELECT COUNT(*) FROM enrichment WHERE chain_id = 1 AND is_verified".to_string(), + 10, + None, + ) + .await + .unwrap(); + assert_eq!(result.rows, vec![vec![serde_json::json!(0)]]); + + let imported = db + .query_sql( + "SELECT verified_count FROM verification_registry_imports WHERE source = 'verifier_alliance' AND chain_id = 1".to_string(), + 10, + None, + ) + .await + .unwrap(); + assert_eq!(imported.rows, vec![vec![serde_json::json!(0)]]); +} + +#[tokio::test] +async fn unchanged_va_files_repair_missing_enrichment_rows() { + let data = TestDir::new(); + let va = TestDir::new(); + write_va_fixture(&va.0); + + let db = Db::open_with_mode(&data.0, "*.parquet", false).unwrap(); + assert!(db.import_verifier_alliance(va.0.clone(), 1).await.unwrap()); + + // Reproduce an interrupted legacy import: file manifests and tracked + // addresses committed, but the replacement enrichment rows did not. + db.execute_batch( + "DELETE FROM enrichment WHERE verification_source = 'verifier_alliance' AND chain_id = 1" + .to_string(), + ) + .await + .unwrap(); + + assert!(db.import_verifier_alliance(va.0.clone(), 1).await.unwrap()); + let result = db + .query_sql( + "SELECT COUNT(*) FROM enrichment WHERE chain_id = 1 AND is_verified".to_string(), + 10, + None, + ) + .await + .unwrap(); + assert_eq!(result.rows, vec![vec![serde_json::json!(1)]]); +} + +#[tokio::test] +async fn unchanged_va_files_repair_legacy_rows_without_positions() { + let data = TestDir::new(); + let va = TestDir::new(); + write_va_fixture(&va.0); + + let db = Db::open_with_mode(&data.0, "*.parquet", false).unwrap(); + assert!(db.import_verifier_alliance(va.0.clone(), 1).await.unwrap()); + + // Keep the recorded/enriched counts equal while reproducing a row from + // before block positions were persisted. This must still trigger repair. + db.execute_batch( + r#" + DELETE FROM enrichment + WHERE verification_source = 'verifier_alliance' AND chain_id = 1; + INSERT INTO enrichment ( + contract_address, chain_id, is_verified, contract_name, checked_at, + verification_source, match_type, block_number, create_index + ) VALUES ( + unhex(repeat('11', 20)), 1, true, NULL, CURRENT_TIMESTAMP, + 'verifier_alliance', 'runtime', NULL, NULL + ); + "# + .to_string(), + ) + .await + .unwrap(); + + assert!(db.import_verifier_alliance(va.0.clone(), 1).await.unwrap()); + let result = db + .query_sql( + "SELECT block_number, create_index FROM enrichment WHERE chain_id = 1".to_string(), + 10, + None, + ) + .await + .unwrap(); + assert_eq!( + result.rows, + vec![vec![serde_json::json!(100), serde_json::json!(0)]] + ); +}