From fdb8f51e824a89656349212c9462cdf83b5849cc Mon Sep 17 00:00:00 2001 From: g4titanx Date: Mon, 3 Aug 2026 22:25:15 +0100 Subject: [PATCH 1/5] feat: add creation and runtime contract opcode queries --- Cargo.lock | 7 + Cargo.toml | 1 + src/cli.rs | 2 +- src/db/mod.rs | 93 ++++- src/db/sql.rs | 42 ++- src/db/views.rs | 181 +++++++++- src/decode/bytecode_meta.rs | 21 +- src/decode/mod.rs | 656 +++++++++++++++++++++++++++++++++++- src/decode/opcodes.rs | 177 ++++++++++ src/extract/tail.rs | 20 +- 10 files changed, 1188 insertions(+), 12 deletions(-) create mode 100644 src/decode/opcodes.rs diff --git a/Cargo.lock b/Cargo.lock index e80a8df..3cc669f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1688,6 +1688,7 @@ dependencies = [ "chrono", "clap", "duckdb", + "eot", "futures", "hex", "indicatif", @@ -2467,6 +2468,12 @@ dependencies = [ "syn 2.0.114", ] +[[package]] +name = "eot" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26c685edb5c949f9e07e31fc36a8f02b15654eb725c29ec29b4fcb25990e2552" + [[package]] name = "equivalent" version = "1.0.2" diff --git a/Cargo.toml b/Cargo.toml index 095b949..c9422e5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,7 @@ reqwest = { version = "0.12", features = ["json", "gzip"] } axum = "0.8.9" tower-http = { version = "0.6.10", features = ["fs", "cors", "trace", "compression-gzip"] } duckdb = { version = "1.10", features = ["bundled"] } +eot = "0.2" tracing = "0.1.44" tracing-subscriber = { version = "0.3.23", features = ["env-filter", "chrono"] } hex = "0.4.3" diff --git a/src/cli.rs b/src/cli.rs index f0a9e98..ed05361 100644 --- a/src/cli.rs +++ b/src/cli.rs @@ -39,7 +39,7 @@ pub enum Commands { Contracts(ContractsArgs), /// Load local contract datasets into Blink Load(LoadArgs), - /// Decode bytecode locally: compiler version, language, ERC standards, proxy detection + /// Decode bytecode locally: metadata plus creation/runtime opcode sets Decode(DecodeArgs), /// Build persistent block-time checkpoints from configured RPCs Checkpoints(CheckpointsArgs), diff --git a/src/db/mod.rs b/src/db/mod.rs index 315011d..35f7288 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -67,6 +67,15 @@ pub struct Db { read_only: bool, } +pub(crate) struct LiveCreationBytecode { + pub chain_id: u64, + pub block_number: u32, + pub create_index: u32, + pub contract_address: Vec, + pub bytecode_hash: Vec, + pub code: Vec, +} + impl Db { pub fn open_with_mode(data_dir: &Path, contracts_glob: &str, read_only: bool) -> Result { Self::open( @@ -308,11 +317,91 @@ impl Db { }) .map(|(code_hash, code)| { let metadata = crate::decode::bytecode_meta::analyze(&code); - (code_hash, metadata) + let opcodes = crate::decode::opcodes::opcode_names( + &code, + crate::decode::opcodes::CodeKind::Runtime, + ); + (code_hash, metadata, opcodes) }) .collect::>(); let mut conn = writer.blocking_lock(); - crate::decode::flush_hash_batch(&mut conn, &analyzed) + let metadata = analyzed + .iter() + .map(|(hash, metadata, _)| (hash.clone(), metadata.clone())) + .collect::>(); + let opcode_sets = analyzed + .into_iter() + .map(|(bytecode_hash, _, opcodes)| crate::decode::OpcodeSetRow { + bytecode_hash, + code_kind: crate::decode::opcodes::CodeKind::Runtime, + opcodes, + }) + .collect::>(); + let metadata_inserted = crate::decode::flush_hash_batch(&mut conn, &metadata)?; + let opcode_sets_inserted = crate::decode::flush_opcode_sets(&mut conn, &opcode_sets)?; + Ok(metadata_inserted.max(opcode_sets_inserted)) + }) + .await + .map_err(|error| anyhow!("join error: {error}"))? + } + + pub(crate) async fn decode_live_creation_bytecodes( + &self, + bytecodes: Vec, + source_file: String, + ) -> 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(|bytecode| { + bytecode.bytecode_hash.len() == 32 + && bytecode.contract_address.len() == 20 + && !bytecode.code.is_empty() + && bytecode.code.len() <= 65_536 + }) + .map(|bytecode| { + let opcodes = crate::decode::opcodes::opcode_names( + &bytecode.code, + crate::decode::opcodes::CodeKind::Creation, + ); + ( + crate::decode::OpcodeSetRow { + bytecode_hash: bytecode.bytecode_hash.clone(), + code_kind: crate::decode::opcodes::CodeKind::Creation, + opcodes, + }, + crate::decode::CreationBytecodeLink { + chain_id: bytecode.chain_id, + block_number: bytecode.block_number, + create_index: bytecode.create_index, + contract_address: bytecode.contract_address, + bytecode_hash: bytecode.bytecode_hash, + }, + ) + }) + .collect::>(); + let mut seen = std::collections::HashSet::with_capacity(analyzed.len()); + let opcode_sets = analyzed + .iter() + .filter(|(set, _)| seen.insert(set.bytecode_hash.as_slice())) + .map(|(set, _)| crate::decode::OpcodeSetRow { + bytecode_hash: set.bytecode_hash.clone(), + code_kind: set.code_kind, + opcodes: set.opcodes.clone(), + }) + .collect::>(); + let links = analyzed + .into_iter() + .map(|(_, link)| link) + .collect::>(); + let mut conn = writer.blocking_lock(); + let inserted = crate::decode::flush_opcode_sets(&mut conn, &opcode_sets)?; + crate::decode::flush_creation_links(&mut conn, &source_file, &links)?; + Ok(inserted) }) .await .map_err(|error| anyhow!("join error: {error}"))? diff --git a/src/db/sql.rs b/src/db/sql.rs index 16704c5..c5bc824 100644 --- a/src/db/sql.rs +++ b/src/db/sql.rs @@ -1,7 +1,7 @@ //! Read-only SQL guard rails and JSON conversion for `POST /api/query`. use anyhow::{anyhow, Result}; -use duckdb::types::ValueRef; +use duckdb::types::{Value as DuckValue, ValueRef}; use serde_json::{Number, Value}; #[derive(Debug, Clone, serde::Serialize, utoipa::ToSchema)] @@ -119,6 +119,11 @@ pub(crate) fn wrap_dashboard_query(sql: &str, limit: u32, chain_id: Option) SELECT * FROM contract_metadata_all WHERE chain_id = {chain_id} + ), + contract_opcodes AS ( + SELECT * + FROM contract_opcodes_all + WHERE chain_id = {chain_id} ) SELECT * FROM ({sql}) AS _blink_dashboard_query @@ -166,6 +171,41 @@ pub(crate) fn value_ref_to_json(value: ValueRef<'_>) -> Value { days, nanos, } => Value::String(format!("{months} months {days} days {nanos} ns")), + ValueRef::List(..) | ValueRef::Array(..) => duckdb_value_to_json(value.to_owned()), + other => Value::String(format!("{other:?}")), + } +} + +fn duckdb_value_to_json(value: DuckValue) -> Value { + match value { + DuckValue::Null => Value::Null, + DuckValue::Boolean(value) => Value::Bool(value), + DuckValue::TinyInt(value) => Value::Number(Number::from(value)), + DuckValue::SmallInt(value) => Value::Number(Number::from(value)), + DuckValue::Int(value) => Value::Number(Number::from(value)), + DuckValue::BigInt(value) => Value::Number(Number::from(value)), + DuckValue::HugeInt(value) => i64::try_from(value) + .map(Number::from) + .map(Value::Number) + .unwrap_or_else(|_| Value::String(value.to_string())), + DuckValue::UTinyInt(value) => Value::Number(Number::from(value)), + DuckValue::USmallInt(value) => Value::Number(Number::from(value)), + DuckValue::UInt(value) => Value::Number(Number::from(value)), + DuckValue::UBigInt(value) => Value::Number(Number::from(value)), + DuckValue::Float(value) => Number::from_f64(value as f64) + .map(Value::Number) + .unwrap_or(Value::Null), + DuckValue::Double(value) => Number::from_f64(value) + .map(Value::Number) + .unwrap_or(Value::Null), + DuckValue::Text(value) | DuckValue::Enum(value) => Value::String(value), + DuckValue::Blob(value) => Value::String(format!("0x{}", hex::encode(value))), + DuckValue::List(values) | DuckValue::Array(values) => Value::Array( + values + .into_iter() + .map(duckdb_value_to_json) + .collect::>(), + ), other => Value::String(format!("{other:?}")), } } diff --git a/src/db/views.rs b/src/db/views.rs index eff6823..ca70744 100644 --- a/src/db/views.rs +++ b/src/db/views.rs @@ -83,6 +83,34 @@ pub(crate) fn ensure_schema(conn: &Connection) -> Result<()> { UPDATE bytecode_metadata_by_hash SET is_proxy_minimal = false WHERE is_proxy_minimal IS NULL; + + CREATE TABLE IF NOT EXISTS bytecode_opcode_sets ( + bytecode_hash BLOB NOT NULL, + code_kind VARCHAR NOT NULL, + opcodes VARCHAR[] NOT NULL, + parser_version USMALLINT NOT NULL, + decoded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + CREATE INDEX IF NOT EXISTS bytecode_opcode_sets_lookup_idx + ON bytecode_opcode_sets(code_kind, bytecode_hash, parser_version); + CREATE TABLE IF NOT EXISTS contract_creation_bytecodes ( + chain_id UBIGINT NOT NULL, + block_number UINTEGER NOT NULL, + create_index UINTEGER NOT NULL, + contract_address BLOB NOT NULL, + bytecode_hash BLOB NOT NULL, + source_file VARCHAR NOT NULL, + linked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ); + CREATE TABLE IF NOT EXISTS opcode_decode_runs ( + file_path VARCHAR NOT NULL, + parser_version USMALLINT NOT NULL, + file_size UBIGINT NOT NULL, + file_mtime_secs BIGINT NOT NULL, + rows_processed UBIGINT NOT NULL, + completed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (file_path, parser_version) + ); "#, ) .context("create blink schema")?; @@ -346,7 +374,7 @@ fn create_enrichment_current_view(conn: &Connection) -> Result<()> { } /// Views consumed by `POST /api/query` users: `bytecodes`, -/// `decoded_bytecodes`, `contract_metadata_all`. +/// `decoded_bytecodes`, `contract_metadata_all`, `contract_opcodes_all`. fn create_standard_query_views(conn: &Connection) -> Result<()> { let has_rollups = table_exists(conn, "rollup_code_counts")?; let has_zellic_bytecodes = table_exists(conn, "zellic_bytecodes")?; @@ -469,7 +497,156 @@ fn create_standard_query_views(conn: &Connection) -> Result<()> { conn.execute_batch(&decoded_sql) .context("create decoded bytecodes query view")?; - create_contract_metadata_view(conn) + create_contract_metadata_view(conn)?; + create_opcode_query_views(conn) +} + +fn create_opcode_query_views(conn: &Connection) -> Result<()> { + let has_opcode_sets = table_exists(conn, "bytecode_opcode_sets")?; + let opcode_sets_sql = if has_opcode_sets { + r#" + CREATE OR REPLACE TEMP VIEW bytecode_opcode_sets_current AS + SELECT + bytecode_hash, + lower('0x' || hex(bytecode_hash)) AS bytecode_hash_hex, + code_kind, + opcodes, + parser_version, + decoded_at + FROM bytecode_opcode_sets + QUALIFY row_number() OVER ( + PARTITION BY bytecode_hash, code_kind + ORDER BY parser_version DESC, decoded_at DESC NULLS LAST + ) = 1; + "# + } else { + r#" + CREATE OR REPLACE TEMP VIEW bytecode_opcode_sets_current AS + SELECT + CAST(NULL AS BLOB) AS bytecode_hash, + CAST(NULL AS VARCHAR) AS bytecode_hash_hex, + CAST(NULL AS VARCHAR) AS code_kind, + CAST(NULL AS VARCHAR[]) AS opcodes, + CAST(NULL AS USMALLINT) AS parser_version, + CAST(NULL AS TIMESTAMP) AS decoded_at + WHERE FALSE; + "# + }; + conn.execute_batch(opcode_sets_sql) + .context("create current opcode-set view")?; + + let has_creation_links = table_exists(conn, "contract_creation_bytecodes")?; + let creation_links_sql = if has_creation_links { + r#" + CREATE OR REPLACE TEMP VIEW contract_creation_bytecodes_current AS + SELECT + chain_id, + block_number, + create_index, + contract_address, + any_value(bytecode_hash) AS bytecode_hash + FROM contract_creation_bytecodes + GROUP BY chain_id, block_number, create_index, contract_address; + "# + } else { + r#" + CREATE OR REPLACE TEMP VIEW contract_creation_bytecodes_current AS + SELECT + CAST(NULL AS UBIGINT) AS chain_id, + CAST(NULL AS UINTEGER) AS block_number, + CAST(NULL AS UINTEGER) AS create_index, + CAST(NULL AS BLOB) AS contract_address, + CAST(NULL AS BLOB) AS bytecode_hash + WHERE FALSE; + "# + }; + conn.execute_batch(creation_links_sql) + .context("create current creation-bytecode link view")?; + + let has_deployments = table_exists(conn, "contract_deployments_native")?; + let runtime_select = if has_deployments { + Some( + r#" + SELECT + c.chain_id, + c.block_number, + c.create_index, + c.contract_address, + lower('0x' || hex(c.contract_address)) AS address, + 'runtime'::VARCHAR AS code_kind, + c.code_hash AS bytecode_hash, + lower('0x' || hex(c.code_hash)) AS bytecode_hash_hex, + o.opcodes, + o.parser_version, + o.decoded_at + FROM contract_deployments_native c + JOIN bytecode_opcode_sets_current o + ON o.bytecode_hash = c.code_hash + AND o.code_kind = 'runtime' + "#, + ) + } else { + None + }; + let creation_select = if has_creation_links { + Some( + r#" + SELECT + c.chain_id, + c.block_number, + c.create_index, + c.contract_address, + lower('0x' || hex(c.contract_address)) AS address, + 'creation'::VARCHAR AS code_kind, + c.bytecode_hash, + lower('0x' || hex(c.bytecode_hash)) AS bytecode_hash_hex, + o.opcodes, + o.parser_version, + o.decoded_at + FROM contract_creation_bytecodes_current c + JOIN bytecode_opcode_sets_current o + ON o.bytecode_hash = c.bytecode_hash + AND o.code_kind = 'creation' + "#, + ) + } else { + None + }; + let body = [runtime_select, creation_select] + .into_iter() + .flatten() + .collect::>() + .join("\nUNION ALL\n"); + let body = if body.is_empty() { + r#" + SELECT + CAST(NULL AS UBIGINT) AS chain_id, + CAST(NULL AS UINTEGER) AS block_number, + CAST(NULL AS UINTEGER) AS create_index, + CAST(NULL AS BLOB) AS contract_address, + CAST(NULL AS VARCHAR) AS address, + CAST(NULL AS VARCHAR) AS code_kind, + CAST(NULL AS BLOB) AS bytecode_hash, + CAST(NULL AS VARCHAR) AS bytecode_hash_hex, + CAST(NULL AS VARCHAR[]) AS opcodes, + CAST(NULL AS USMALLINT) AS parser_version, + CAST(NULL AS TIMESTAMP) AS decoded_at + WHERE FALSE + "# + .to_string() + } else { + body + }; + conn.execute_batch(&format!( + r#" + CREATE OR REPLACE TEMP VIEW contract_opcodes_all AS + {body}; + + CREATE OR REPLACE TEMP VIEW contract_opcodes AS + SELECT * FROM contract_opcodes_all; + "# + )) + .context("create contract opcode query views") } /// The SQL-explorer surface: `contract_metadata_all` / `contract_metadata`. diff --git a/src/decode/bytecode_meta.rs b/src/decode/bytecode_meta.rs index 3bee14b..3e2dbaf 100644 --- a/src/decode/bytecode_meta.rs +++ b/src/decode/bytecode_meta.rs @@ -39,7 +39,7 @@ impl Language { pub fn analyze(code: &[u8]) -> BytecodeMetadata { let mut meta = BytecodeMetadata::default(); decode_cbor_tail(code, &mut meta); - scan_opcodes(code, &mut meta); + scan_opcodes(&code[..runtime_code_end(code)], &mut meta); detect_minimal_proxy(code, &mut meta); meta } @@ -98,6 +98,25 @@ fn decode_cbor_tail(code: &[u8], meta: &mut BytecodeMetadata) { } } +/// End of executable runtime bytecode, excluding a validated CBOR compiler +/// metadata trailer and its two-byte length suffix when present. +pub(crate) fn runtime_code_end(code: &[u8]) -> usize { + if code.len() < 4 { + return code.len(); + } + let n = code.len(); + let metadata_len = u16::from_be_bytes([code[n - 2], code[n - 1]]) as usize; + if metadata_len == 0 || metadata_len + 2 > n { + return n; + } + let start = n - 2 - metadata_len; + if parse_cbor_map(&code[start..n - 2]).is_ok() { + start + } else { + n + } +} + fn format_solc_version(bytes: &[u8]) -> Option { if bytes.len() == 3 { Some(format!("{}.{}.{}", bytes[0], bytes[1], bytes[2])) diff --git a/src/decode/mod.rs b/src/decode/mod.rs index 38a7163..86d216d 100644 --- a/src/decode/mod.rs +++ b/src/decode/mod.rs @@ -7,6 +7,7 @@ //! (unless `--overwrite`). //! pub mod bytecode_meta; +pub mod opcodes; use std::{ fs::File, @@ -20,7 +21,10 @@ use duckdb::{params, Connection}; use indicatif::{ProgressBar, ProgressStyle}; use rayon::prelude::*; -use self::bytecode_meta::{analyze, BytecodeMetadata}; +use self::{ + bytecode_meta::{analyze, BytecodeMetadata}, + opcodes::{opcode_names, CodeKind, OPCODE_PARSER_VERSION}, +}; use crate::{ cli::DecodeArgs, util::{format_count, match_simple_glob, print_header, print_kv, print_kv_accent}, @@ -28,6 +32,33 @@ use crate::{ const MAX_DECODE_CODE_BYTES: u64 = 65_536; +#[derive(Debug)] +pub(crate) struct OpcodeSetRow { + pub bytecode_hash: Vec, + pub code_kind: CodeKind, + pub opcodes: Vec, +} + +#[derive(Debug, Clone)] +pub(crate) struct CreationBytecodeLink { + pub chain_id: u64, + pub block_number: u32, + pub create_index: u32, + pub contract_address: Vec, + pub bytecode_hash: Vec, +} + +struct OpcodeContractInput { + chain_id: u64, + block_number: u32, + create_index: u32, + contract_address: Vec, + init_code_hash: Option>, + init_code: Option>, + code_hash: Vec, + code: Vec, +} + const SCHEMA: &str = r#" -- Scalable address-level decode storage. The legacy `bytecode_metadata` table -- had a BLOB primary key and became memory-heavy at tens of millions of rows. @@ -83,6 +114,43 @@ CREATE TABLE IF NOT EXISTS decode_runs ( rows_processed UBIGINT NOT NULL, completed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); + +-- Compact presence sets keyed by bytecode hash and code kind. Keeping one +-- VARCHAR[] per unique bytecode avoids multiplying every instruction across +-- every deployment while remaining directly queryable with list_contains(). +CREATE TABLE IF NOT EXISTS bytecode_opcode_sets ( + bytecode_hash BLOB NOT NULL, + code_kind VARCHAR NOT NULL, + opcodes VARCHAR[] NOT NULL, + parser_version USMALLINT NOT NULL, + decoded_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP +); +CREATE INDEX IF NOT EXISTS bytecode_opcode_sets_lookup_idx + ON bytecode_opcode_sets(code_kind, bytecode_hash, parser_version); + +-- Runtime hashes already live on contract_deployments_native. Creation +-- hashes do not, so retain this narrow address -> init-code-hash mapping. +CREATE TABLE IF NOT EXISTS contract_creation_bytecodes ( + chain_id UBIGINT NOT NULL, + block_number UINTEGER NOT NULL, + create_index UINTEGER NOT NULL, + contract_address BLOB NOT NULL, + bytecode_hash BLOB NOT NULL, + source_file VARCHAR NOT NULL, + linked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP +); + +-- Independent from decode_runs: databases decoded before opcode sets were +-- introduced must backfill once even though their metadata pass is complete. +CREATE TABLE IF NOT EXISTS opcode_decode_runs ( + file_path VARCHAR NOT NULL, + parser_version USMALLINT NOT NULL, + file_size UBIGINT NOT NULL, + file_mtime_secs BIGINT NOT NULL, + rows_processed UBIGINT NOT NULL, + completed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (file_path, parser_version) +); "#; pub async fn run_decode(args: DecodeArgs) -> Result<()> { @@ -110,7 +178,7 @@ fn run_decode_blocking(args: DecodeArgs, data_dir: PathBuf) -> Result<()> { if args.overwrite { write_conn .execute_batch( - "DROP TABLE IF EXISTS bytecode_metadata; DELETE FROM bytecode_metadata_v2; DELETE FROM bytecode_metadata_by_hash; DELETE FROM decode_runs;", + "DROP TABLE IF EXISTS bytecode_metadata; DELETE FROM bytecode_metadata_v2; DELETE FROM bytecode_metadata_by_hash; DELETE FROM decode_runs; DELETE FROM bytecode_opcode_sets; DELETE FROM contract_creation_bytecodes; DELETE FROM opcode_decode_runs;", ) .context("clear bytecode_metadata for --overwrite")?; } @@ -367,7 +435,17 @@ fn run_decode_blocking(args: DecodeArgs, data_dir: PathBuf) -> Result<()> { let _ = write_conn.execute_batch("CHECKPOINT"); } + let (opcode_scanned, opcode_sets_written) = decode_opcode_datasets( + &db_path, + &mut write_conn, + &files, + args.batch_size, + has_zellic_bytecodes, + args.overwrite, + )?; + let elapsed = started.elapsed(); + let work_rows = total_rows.max(opcode_scanned); println!(); if total_skipped_files > 0 { print_kv( @@ -377,12 +455,20 @@ fn run_decode_blocking(args: DecodeArgs, data_dir: PathBuf) -> Result<()> { } print_kv_accent("scanned", &format_count(total_rows)); print_kv_accent("decoded", &format!("{} rows", format_count(total_decoded))); + print_kv_accent( + "opcodes", + &format!( + "{} contracts scanned · {} bytecode sets", + format_count(opcode_scanned), + format_count(opcode_sets_written) + ), + ); print_kv_accent( "speed", - &if total_rows > 0 { + &if work_rows > 0 { format!( "{:.0} rows/sec · total {:.1}s", - total_rows as f64 / elapsed.as_secs_f64().max(0.001), + work_rows as f64 / elapsed.as_secs_f64().max(0.001), elapsed.as_secs_f64() ) } else { @@ -608,6 +694,538 @@ fn copy_zellic_decode_rows_to_tsv(conn: &Connection, temp_path: &Path) -> Result .context("export Zellic decode rows") } +fn decode_opcode_datasets( + db_path: &Path, + conn: &mut Connection, + files: &[PathBuf], + batch_size: usize, + has_zellic_bytecodes: bool, + overwrite: bool, +) -> Result<(u64, u64)> { + println!(); + print_kv_accent( + "opcode parser", + &format!("eot · generation {}", OPCODE_PARSER_VERSION), + ); + + let mut scanned = 0u64; + let mut written = 0u64; + if has_zellic_bytecodes { + let (source_scanned, source_written) = + decode_zellic_opcode_sets(db_path, conn, batch_size)?; + scanned += source_scanned; + written += source_written; + } + + let completed: std::collections::HashMap = { + let mut stmt = conn.prepare( + "SELECT file_path, file_size, file_mtime_secs + FROM opcode_decode_runs WHERE parser_version = ?", + )?; + let rows = stmt.query_map(params![OPCODE_PARSER_VERSION], |row| { + Ok(( + row.get::<_, String>(0)?, + (row.get::<_, u64>(1)?, row.get::<_, i64>(2)?), + )) + })?; + rows.collect::>()? + }; + + for file in files { + let source_file = file.display().to_string(); + let (file_size, file_mtime_secs) = match std::fs::metadata(file) { + Ok(metadata) => ( + metadata.len(), + metadata + .modified() + .ok() + .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok()) + .map(|duration| duration.as_secs() as i64) + .unwrap_or(0), + ), + Err(_) => (0, 0), + }; + if !overwrite + && completed + .get(&source_file) + .is_some_and(|&(size, mtime)| size == file_size && mtime == file_mtime_secs) + { + continue; + } + + // A failed earlier attempt may have appended only part of this + // source's creation links. Opcode sets are hash-keyed and idempotent. + conn.execute( + "DELETE FROM contract_creation_bytecodes WHERE source_file = ?", + params![source_file.as_str()], + ) + .context("clear partial creation bytecode links")?; + + let (file_scanned, file_written) = decode_parquet_opcode_sets(conn, file, batch_size)?; + scanned += file_scanned; + written += file_written; + + conn.execute( + r#" + INSERT INTO opcode_decode_runs ( + file_path, parser_version, file_size, file_mtime_secs, + rows_processed, completed_at + ) VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP) + ON CONFLICT (file_path, parser_version) DO UPDATE SET + file_size = excluded.file_size, + file_mtime_secs = excluded.file_mtime_secs, + rows_processed = excluded.rows_processed, + completed_at = excluded.completed_at + "#, + params![ + source_file, + OPCODE_PARSER_VERSION, + file_size, + file_mtime_secs, + file_scanned + ], + ) + .context("record opcode_decode_runs marker")?; + let _ = conn.execute_batch("CHECKPOINT"); + } + + Ok((scanned, written)) +} + +fn decode_zellic_opcode_sets( + db_path: &Path, + conn: &mut Connection, + batch_size: usize, +) -> Result<(u64, u64)> { + let pending: u64 = conn + .query_row( + &format!( + r#" + SELECT COUNT(*)::UBIGINT + FROM zellic_bytecodes b + WHERE b.n_code_bytes BETWEEN 1 AND {MAX_DECODE_CODE_BYTES} + AND NOT EXISTS ( + SELECT 1 FROM bytecode_opcode_sets o + WHERE o.bytecode_hash = b.code_hash + AND o.code_kind = 'runtime' + AND o.parser_version = {OPCODE_PARSER_VERSION} + ) + "# + ), + [], + |row| row.get(0), + ) + .unwrap_or(0); + if pending == 0 { + return Ok((0, 0)); + } + + let temp_path = temp_zellic_opcode_path(db_path); + if let Some(parent) = temp_path.parent() { + std::fs::create_dir_all(parent).with_context(|| format!("create {}", parent.display()))?; + } + let _ = std::fs::remove_file(&temp_path); + let temp_path_sql = temp_path.display().to_string().replace('\'', "''"); + conn.execute_batch(&format!( + r#" + COPY ( + SELECT hex(b.code_hash), hex(b.code) + FROM zellic_bytecodes b + WHERE b.n_code_bytes BETWEEN 1 AND {MAX_DECODE_CODE_BYTES} + AND octet_length(b.code) = b.n_code_bytes + AND NOT EXISTS ( + SELECT 1 FROM bytecode_opcode_sets o + WHERE o.bytecode_hash = b.code_hash + AND o.code_kind = 'runtime' + AND o.parser_version = {OPCODE_PARSER_VERSION} + ) + ) TO '{temp_path_sql}' (FORMAT CSV, HEADER false, DELIMITER '\t'); + "# + )) + .context("export Zellic opcode rows")?; + + let file = File::open(&temp_path) + .with_context(|| format!("open Zellic opcode export {}", temp_path.display()))?; + let mut raw = Vec::with_capacity(batch_size.max(5_000)); + let mut scanned = 0u64; + let mut written = 0u64; + for line in BufReader::new(file).lines() { + let line = line?; + let Some((hash_hex, code_hex)) = line.split_once('\t') else { + continue; + }; + let (Ok(bytecode_hash), Ok(code)) = (hex::decode(hash_hex), hex::decode(code_hex)) else { + continue; + }; + if bytecode_hash.len() != 32 || code.is_empty() || code.len() as u64 > MAX_DECODE_CODE_BYTES + { + continue; + } + raw.push((bytecode_hash, code)); + scanned += 1; + if raw.len() >= batch_size.max(5_000) { + written += analyze_and_flush_runtime_opcode_batch(conn, std::mem::take(&mut raw))?; + raw = Vec::with_capacity(batch_size.max(5_000)); + } + } + if !raw.is_empty() { + written += analyze_and_flush_runtime_opcode_batch(conn, raw)?; + } + let _ = std::fs::remove_file(&temp_path); + Ok((scanned, written)) +} + +fn analyze_and_flush_runtime_opcode_batch( + conn: &mut Connection, + raw: Vec<(Vec, Vec)>, +) -> Result { + let rows = raw + .into_par_iter() + .map(|(bytecode_hash, code)| OpcodeSetRow { + bytecode_hash, + code_kind: CodeKind::Runtime, + opcodes: opcode_names(&code, CodeKind::Runtime), + }) + .collect::>(); + flush_opcode_sets(conn, &rows) +} + +fn decode_parquet_opcode_sets( + conn: &mut Connection, + parquet_file: &Path, + batch_size: usize, +) -> Result<(u64, u64)> { + let temp_path = temp_opcode_decode_path(parquet_file); + let _ = std::fs::remove_file(&temp_path); + copy_opcode_rows_to_tsv(parquet_file, &temp_path)?; + + let file = File::open(&temp_path) + .with_context(|| format!("open opcode export {}", temp_path.display()))?; + let mut raw = Vec::with_capacity(batch_size.max(5_000)); + let mut scanned = 0u64; + let mut written = 0u64; + let source_file = parquet_file.display().to_string(); + + for line in BufReader::new(file).lines() { + let line = line?; + let fields = line.split('\t').collect::>(); + if fields.len() != 8 { + continue; + } + let (Ok(chain_id), Ok(block_number), Ok(create_index)) = ( + fields[0].parse::(), + fields[1].parse::(), + fields[2].parse::(), + ) else { + continue; + }; + let Ok(contract_address) = hex::decode(fields[3]) else { + continue; + }; + let Ok(code) = hex::decode(fields[7]) else { + continue; + }; + if contract_address.len() != 20 + || code.is_empty() + || code.len() as u64 > MAX_DECODE_CODE_BYTES + { + continue; + } + let code_hash = hex::decode(fields[6]) + .ok() + .filter(|hash| hash.len() == 32) + .unwrap_or_else(|| alloy::primitives::keccak256(&code).to_vec()); + let init_code = hex::decode(fields[5]) + .ok() + .filter(|bytes| !bytes.is_empty() && bytes.len() as u64 <= MAX_DECODE_CODE_BYTES); + let init_code_hash = init_code.as_ref().map(|init_code| { + hex::decode(fields[4]) + .ok() + .filter(|hash| hash.len() == 32) + .unwrap_or_else(|| alloy::primitives::keccak256(init_code).to_vec()) + }); + + raw.push(OpcodeContractInput { + chain_id, + block_number, + create_index, + contract_address, + init_code_hash, + init_code, + code_hash, + code, + }); + scanned += 1; + + if raw.len() >= batch_size.max(5_000) { + written += flush_opcode_contract_batch(conn, &source_file, std::mem::take(&mut raw))?; + raw = Vec::with_capacity(batch_size.max(5_000)); + } + } + if !raw.is_empty() { + written += flush_opcode_contract_batch(conn, &source_file, raw)?; + } + let _ = std::fs::remove_file(&temp_path); + Ok((scanned, written)) +} + +fn flush_opcode_contract_batch( + conn: &mut Connection, + source_file: &str, + raw: Vec, +) -> Result { + let analyzed = raw + .into_par_iter() + .map(|input| { + let runtime = OpcodeSetRow { + bytecode_hash: input.code_hash, + code_kind: CodeKind::Runtime, + opcodes: opcode_names(&input.code, CodeKind::Runtime), + }; + let creation = + input + .init_code + .zip(input.init_code_hash) + .map(|(init_code, init_code_hash)| { + ( + OpcodeSetRow { + bytecode_hash: init_code_hash.clone(), + code_kind: CodeKind::Creation, + opcodes: opcode_names(&init_code, CodeKind::Creation), + }, + CreationBytecodeLink { + chain_id: input.chain_id, + block_number: input.block_number, + create_index: input.create_index, + contract_address: input.contract_address, + bytecode_hash: init_code_hash, + }, + ) + }); + (runtime, creation) + }) + .collect::>(); + + let mut seen = std::collections::HashSet::with_capacity(analyzed.len() * 2); + let mut sets = Vec::with_capacity(analyzed.len() * 2); + let mut links = Vec::with_capacity(analyzed.len()); + for (runtime, creation) in analyzed { + if seen.insert((runtime.code_kind, runtime.bytecode_hash.clone())) { + sets.push(runtime); + } + if let Some((creation, link)) = creation { + if seen.insert((creation.code_kind, creation.bytecode_hash.clone())) { + sets.push(creation); + } + links.push(link); + } + } + + let inserted = flush_opcode_sets(conn, &sets)?; + flush_creation_links(conn, source_file, &links)?; + Ok(inserted) +} + +fn copy_opcode_rows_to_tsv(parquet_file: &Path, temp_path: &Path) -> Result<()> { + let columns = parquet_columns(parquet_file)?; + if !columns.contains("contract_address") || !columns.contains("code") { + return Err(anyhow!("parquet lacks contract_address/code columns")); + } + + let expr = |column: &str, fallback: &str| { + if columns.contains(column) { + column.to_string() + } else { + fallback.to_string() + } + }; + let chain_id = expr("chain_id", "1"); + let block_number = expr("block_number", "0"); + let create_index = expr("create_index", "0"); + let init_code_hash = expr("init_code_hash", "NULL::BLOB"); + let code_hash = expr("code_hash", "NULL::BLOB"); + let init_code = if columns.contains("init_code") { + format!( + "CASE WHEN octet_length(init_code) BETWEEN 1 AND {MAX_DECODE_CODE_BYTES} \ + THEN init_code ELSE NULL::BLOB END" + ) + } else { + "NULL::BLOB".to_string() + }; + let runtime_filter = if columns.contains("n_code_bytes") { + format!( + "n_code_bytes BETWEEN 1 AND {MAX_DECODE_CODE_BYTES} AND \ + octet_length(code) = n_code_bytes" + ) + } else { + format!("octet_length(code) BETWEEN 1 AND {MAX_DECODE_CODE_BYTES}") + }; + + let parquet_path = parquet_file.display().to_string().replace('\'', "''"); + let temp_path = temp_path.display().to_string().replace('\'', "''"); + let sql = format!( + r#" + COPY ( + SELECT + COALESCE({chain_id}, 1)::VARCHAR, + COALESCE({block_number}, 0)::VARCHAR, + COALESCE({create_index}, 0)::VARCHAR, + hex(contract_address), + COALESCE(hex({init_code_hash}), ''), + COALESCE(hex({init_code}), ''), + COALESCE(hex({code_hash}), ''), + hex(code) + FROM read_parquet('{parquet_path}') + WHERE contract_address IS NOT NULL + AND octet_length(contract_address) = 20 + AND code IS NOT NULL + AND {runtime_filter} + ) TO '{temp_path}' (FORMAT CSV, HEADER false, DELIMITER '\t'); + "# + ); + let output = std::process::Command::new("duckdb") + .arg("-c") + .arg(&sql) + .output() + .context("run duckdb CLI for opcode row export")?; + if !output.status.success() { + return Err(anyhow!( + "duckdb CLI opcode export failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + )); + } + Ok(()) +} + +fn parquet_columns(parquet_file: &Path) -> Result> { + let parquet_path = parquet_file.display().to_string().replace('\'', "''"); + let sql = + format!("SELECT string_agg(DISTINCT name, ',') FROM parquet_schema('{parquet_path}');"); + let output = std::process::Command::new("duckdb") + .arg("-noheader") + .arg("-c") + .arg(&sql) + .output() + .context("run duckdb CLI for parquet column discovery")?; + if !output.status.success() { + return Err(anyhow!( + "duckdb CLI schema discovery failed: {}", + String::from_utf8_lossy(&output.stderr).trim() + )); + } + Ok(String::from_utf8_lossy(&output.stdout) + .trim() + .split(',') + .map(str::trim) + .filter(|name| !name.is_empty()) + .map(ToOwned::to_owned) + .collect()) +} + +pub(crate) fn flush_opcode_sets(conn: &mut Connection, rows: &[OpcodeSetRow]) -> Result { + if rows.is_empty() { + return Ok(0); + } + conn.execute_batch( + r#" + CREATE TEMP TABLE IF NOT EXISTS _opcode_set_stage ( + bytecode_hash BLOB, + code_kind VARCHAR, + opcodes_json VARCHAR + ); + DELETE FROM _opcode_set_stage; + "#, + )?; + { + let mut appender = conn.appender("_opcode_set_stage")?; + for row in rows { + let opcodes_json = serde_json::to_string(&row.opcodes)?; + appender.append_row(params![ + &row.bytecode_hash, + row.code_kind.as_str(), + opcodes_json + ])?; + } + appender.flush()?; + } + + let inserted = conn.execute( + &format!( + r#" + INSERT INTO bytecode_opcode_sets ( + bytecode_hash, code_kind, opcodes, parser_version, decoded_at + ) + SELECT + s.bytecode_hash, + s.code_kind, + from_json(s.opcodes_json, '["VARCHAR"]')::VARCHAR[], + {OPCODE_PARSER_VERSION}, + CURRENT_TIMESTAMP + FROM _opcode_set_stage s + WHERE NOT EXISTS ( + SELECT 1 FROM bytecode_opcode_sets existing + WHERE existing.bytecode_hash = s.bytecode_hash + AND existing.code_kind = s.code_kind + AND existing.parser_version = {OPCODE_PARSER_VERSION} + ) + "# + ), + [], + )?; + Ok(inserted as u64) +} + +pub(crate) fn flush_creation_links( + conn: &mut Connection, + source_file: &str, + links: &[CreationBytecodeLink], +) -> Result { + if links.is_empty() { + return Ok(0); + } + conn.execute_batch( + r#" + CREATE TEMP TABLE IF NOT EXISTS _creation_link_stage ( + chain_id UBIGINT, + block_number UINTEGER, + create_index UINTEGER, + contract_address BLOB, + bytecode_hash BLOB, + source_file VARCHAR + ); + DELETE FROM _creation_link_stage; + "#, + )?; + { + let mut appender = conn.appender("_creation_link_stage")?; + for link in links { + appender.append_row(params![ + link.chain_id, + link.block_number, + link.create_index, + &link.contract_address, + &link.bytecode_hash, + source_file + ])?; + } + appender.flush()?; + } + let inserted = conn.execute( + r#" + INSERT INTO contract_creation_bytecodes ( + chain_id, block_number, create_index, contract_address, + bytecode_hash, source_file + ) + SELECT + chain_id, block_number, create_index, contract_address, + bytecode_hash, source_file + FROM _creation_link_stage + "#, + [], + )?; + Ok(inserted as u64) +} + fn parquet_has_column(parquet_file: &Path, column: &str) -> Result { let parquet_path = parquet_file.display().to_string().replace('\'', "''"); let sql = @@ -651,6 +1269,36 @@ fn temp_zellic_decode_path(db_path: &Path) -> PathBuf { dir.join(format!("blink_zellic_decode_{}.tsv", std::process::id())) } +fn temp_zellic_opcode_path(db_path: &Path) -> PathBuf { + let dir = db_path + .parent() + .map(|path| path.join(".blink").join("tmp")) + .unwrap_or_else(std::env::temp_dir); + dir.join(format!("blink_zellic_opcodes_{}.tsv", std::process::id())) +} + +fn temp_opcode_decode_path(file: &Path) -> PathBuf { + let file_name = file + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("contracts"); + let safe_name = file_name + .chars() + .map(|character| { + if character.is_ascii_alphanumeric() { + character + } else { + '_' + } + }) + .collect::(); + std::env::temp_dir().join(format!( + "blink_opcode_decode_{}_{}.tsv", + std::process::id(), + safe_name + )) +} + fn flush_batch( conn: &mut Connection, source_file: &str, diff --git a/src/decode/opcodes.rs b/src/decode/opcodes.rs new file mode 100644 index 0000000..e28ef1a --- /dev/null +++ b/src/decode/opcodes.rs @@ -0,0 +1,177 @@ +//! EVM opcode-set extraction for creation and deployed runtime bytecode. +//! +//! The public dataset is presence-oriented, so this module returns each +//! mnemonic at most once. `PUSH1` through `PUSH32` payload bytes are always +//! skipped. Runtime bytecode is decoded linearly (minus recognized compiler +//! metadata); creation bytecode is decoded from PC 0 along statically +//! discoverable control-flow edges so embedded runtime code and constructor +//! arguments are not treated as constructor instructions merely because they +//! happen to contain opcode-shaped bytes. + +use super::bytecode_meta::runtime_code_end; +use eot::OpCode; + +/// Increment this whenever opcode parsing semantics change. Stored alongside +/// decoded rows so `blink decode` can backfill a new generation without +/// confusing it with older results. +pub const OPCODE_PARSER_VERSION: u16 = 1; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum CodeKind { + Creation, + Runtime, +} + +impl CodeKind { + pub const fn as_str(self) -> &'static str { + match self { + Self::Creation => "creation", + Self::Runtime => "runtime", + } + } +} + +/// Return the sorted, unique opcode mnemonics found in `code`. +pub fn opcode_names(code: &[u8], kind: CodeKind) -> Vec { + let mut present = [false; 256]; + match kind { + CodeKind::Runtime => scan_linear(&code[..runtime_code_end(code)], &mut present), + CodeKind::Creation => scan_reachable_creation(code, &mut present), + } + + present + .iter() + .enumerate() + .filter(|(_, found)| **found) + .map(|(opcode, _)| opcode_name(opcode as u8)) + .collect() +} + +fn scan_linear(code: &[u8], present: &mut [bool; 256]) { + let mut pc = 0usize; + while pc < code.len() { + let opcode = code[pc]; + present[opcode as usize] = true; + pc = next_pc(pc, opcode, code.len()); + } +} + +/// Best-effort control-flow traversal for constructor bytecode. +/// +/// Solidity and Vyper constructor branches use an immediate `PUSHn` target +/// directly before `JUMP`/`JUMPI`, which lets us follow their executable +/// basic blocks while stopping before the embedded runtime image. Dynamic +/// jump targets that cannot be resolved this way are conservatively not +/// followed: inventing instructions from the data tail would recreate the +/// raw-byte false positives this dataset exists to avoid. +fn scan_reachable_creation(code: &[u8], present: &mut [bool; 256]) { + if code.is_empty() { + return; + } + + let mut worklist = vec![0usize]; + let mut visited = vec![false; code.len()]; + + while let Some(mut pc) = worklist.pop() { + let mut previous_push: Option = None; + while pc < code.len() && !visited[pc] { + visited[pc] = true; + let opcode = code[pc]; + present[opcode as usize] = true; + let following_pc = next_pc(pc, opcode, code.len()); + + if opcode == 0x56 || opcode == 0x57 { + if let Some(destination) = previous_push { + if destination < code.len() && code[destination] == 0x5b { + worklist.push(destination); + } + } + if opcode == 0x56 { + break; + } + previous_push = None; + pc = following_pc; + continue; + } + + if is_halting(opcode) { + break; + } + + previous_push = push_value(code, pc, opcode); + pc = following_pc; + } + } +} + +#[inline] +fn next_pc(pc: usize, opcode: u8, code_len: usize) -> usize { + let immediate_len = OpCode::from_byte(opcode) + .info() + .map(|info| usize::from(info.immediate_size)) + .unwrap_or(0); + pc.saturating_add(1 + immediate_len).min(code_len) +} + +fn push_value(code: &[u8], pc: usize, opcode: u8) -> Option { + if !(0x60..=0x7f).contains(&opcode) { + return None; + } + let width = usize::from(opcode - 0x5f); + let immediate = code.get(pc + 1..pc + 1 + width)?; + let mut value = 0usize; + for &byte in immediate { + value = value.checked_mul(256)?.checked_add(usize::from(byte))?; + } + Some(value) +} + +#[inline] +fn is_halting(opcode: u8) -> bool { + let opcode = OpCode::from_byte(opcode); + opcode.terminates() || opcode == OpCode::SELFDESTRUCT +} + +fn opcode_name(opcode: u8) -> String { + let opcode = OpCode::from_byte(opcode); + match opcode.info() { + Some(info) => info.name.to_string(), + None => format!("UNKNOWN_0x{:02X}", opcode.byte()), + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + + use super::*; + + #[test] + fn push_payload_is_not_an_opcode() { + let names = opcode_names(&[0x61, 0xf5, 0xf4, 0xf5], CodeKind::Runtime); + assert_eq!(names, vec!["PUSH2", "CREATE2"]); + assert!(!names.iter().any(|name| name == "DELEGATECALL")); + } + + #[test] + fn creation_scan_stops_before_data_after_halt() { + let names = opcode_names(&[0x00, 0xf5], CodeKind::Creation); + assert_eq!(names, vec!["STOP"]); + } + + #[test] + fn creation_scan_follows_static_jump_target() { + let names = opcode_names( + &[0x60, 0x04, 0x56, 0xaa, 0x5b, 0xf5, 0x00], + CodeKind::Creation, + ); + assert!(names.iter().any(|name| name == "CREATE2")); + assert!(!names.iter().any(|name| name == "UNKNOWN_0xAA")); + } + + #[test] + fn every_opcode_byte_has_a_stable_name() { + let names = (0u8..=u8::MAX).map(opcode_name).collect::>(); + assert_eq!(names.len(), 256); + } +} diff --git a/src/extract/tail.rs b/src/extract/tail.rs index 5350106..7655d09 100644 --- a/src/extract/tail.rs +++ b/src/extract/tail.rs @@ -16,7 +16,10 @@ use chrono::Utc; use futures::{stream, StreamExt}; use super::{batch::BatchClient, parquet_io, traces::extract_contracts}; -use crate::{db::Db, types::ChunkReport}; +use crate::{ + db::{Db, LiveCreationBytecode}, + types::ChunkReport, +}; const TAIL_BATCH_BLOCK_LIMIT: u64 = 1_000; @@ -93,6 +96,7 @@ pub async fn tail_once( let mut writer = None; let mut rows_written = 0usize; let mut live_bytecodes = HashMap::, Vec>::new(); + let mut live_creation_bytecodes = Vec::new(); let mut pending: BTreeMap)>> = BTreeMap::new(); let mut next_index = 0usize; @@ -131,6 +135,14 @@ pub async fn tail_once( live_bytecodes .entry(row.code_hash.clone()) .or_insert_with(|| row.code.clone()); + live_creation_bytecodes.push(LiveCreationBytecode { + chain_id: row.chain_id, + block_number: row.block_number, + create_index: row.create_index, + contract_address: row.contract_address.clone(), + bytecode_hash: row.init_code_hash.clone(), + code: row.init_code.clone(), + }); } batch_rows.sort_unstable_by(|a, b| { a.block_number @@ -168,9 +180,15 @@ pub async fn tail_once( let decoded = db .decode_live_bytecodes(live_bytecodes.into_iter().collect()) .await?; + let creation_decoded = db + .decode_live_creation_bytecodes(live_creation_bytecodes, output_path.display().to_string()) + .await?; if decoded > 0 { tracing::info!("decoded {} new live bytecode(s)", decoded); } + if creation_decoded > 0 { + tracing::info!("decoded {} new live creation bytecode(s)", creation_decoded); + } let size_bytes = std::fs::metadata(&output_path).ok().map(|m| m.len()); Ok(Some(ChunkReport { From 053e0bd5c6ba05656b4530cbd4f5edcde8d1832e Mon Sep 17 00:00:00 2001 From: g4titanx Date: Mon, 3 Aug 2026 22:25:24 +0100 Subject: [PATCH 2/5] test: cover opcode parsing, backfill, and queries --- tests/db.rs | 71 ++++++++++++++++++++++++++++ tests/decode.rs | 123 ++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 194 insertions(+) diff --git a/tests/db.rs b/tests/db.rs index 435a687..7b9ba9a 100644 --- a/tests/db.rs +++ b/tests/db.rs @@ -471,6 +471,77 @@ async fn dashboard_sql_scopes_contract_metadata_by_chain() { assert_eq!(result.rows[0][1], serde_json::json!(300)); } +#[tokio::test] +async fn contract_opcodes_queries_creation_and_runtime_separately() { + let dir = TestDir::new("contract_opcodes"); + write_backfill_parquet(&dir.path); + let db = Db::open_with_mode(&dir.path, "*.parquet", false).unwrap(); + + db.execute_batch( + r#" + INSERT INTO bytecode_opcode_sets ( + bytecode_hash, code_kind, opcodes, parser_version + ) VALUES + (unhex(repeat('03', 32)), 'creation', ['PUSH1', 'CREATE2'], 1), + (unhex(repeat('09', 32)), 'runtime', ['PUSH1', 'DELEGATECALL'], 1), + (unhex(repeat('19', 32)), 'runtime', ['SELFDESTRUCT'], 1); + + INSERT INTO contract_creation_bytecodes ( + chain_id, block_number, create_index, contract_address, + bytecode_hash, source_file + ) VALUES ( + 1, 200, 0, unhex(repeat('05', 20)), + unhex(repeat('03', 32)), 'test' + ); + "# + .to_string(), + ) + .await + .unwrap(); + + let creation = db + .query_sql( + "SELECT address, code_kind, opcodes FROM contract_opcodes \ + WHERE code_kind = 'creation' AND list_contains(opcodes, 'CREATE2')" + .to_string(), + 10, + Some(1), + ) + .await + .unwrap(); + assert_eq!(creation.row_count, 1); + assert_eq!(creation.rows[0][1], serde_json::json!("creation")); + assert_eq!(creation.rows[0][2], serde_json::json!(["PUSH1", "CREATE2"])); + + let runtime = db + .query_sql( + "SELECT address FROM contract_opcodes \ + WHERE code_kind = 'runtime' AND list_contains(opcodes, 'DELEGATECALL')" + .to_string(), + 10, + Some(1), + ) + .await + .unwrap(); + assert_eq!(runtime.row_count, 1); + assert_eq!( + runtime.rows[0][0], + serde_json::json!(format!("0x{}", hex_string(0x05, 20))) + ); + + let wrong_chain = db + .query_sql( + "SELECT address FROM contract_opcodes \ + WHERE list_contains(opcodes, 'SELFDESTRUCT')" + .to_string(), + 10, + Some(1), + ) + .await + .unwrap(); + assert!(wrong_chain.rows.is_empty()); +} + fn hex_string(byte: u8, len: usize) -> String { hex::encode(vec![byte; len]) } diff --git a/tests/decode.rs b/tests/decode.rs index 59a482a..3eec296 100644 --- a/tests/decode.rs +++ b/tests/decode.rs @@ -1,6 +1,43 @@ //! Bytecode metadata decoding tests. +use std::{ + fs, + path::{Path, PathBuf}, + process::Command, + time::{SystemTime, UNIX_EPOCH}, +}; + use blink::decode::bytecode_meta::{analyze, Language}; +use duckdb::Connection; + +struct TestDir(PathBuf); + +impl TestDir { + fn new(name: &str) -> Self { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = std::env::temp_dir().join(format!( + "blink_decode_test_{}_{}_{}", + std::process::id(), + name, + unique + )); + fs::create_dir_all(&path).unwrap(); + Self(path) + } + + fn path(&self) -> &Path { + &self.0 + } +} + +impl Drop for TestDir { + fn drop(&mut self) { + let _ = fs::remove_dir_all(&self.0); + } +} #[test] fn empty_bytecode() { @@ -82,3 +119,89 @@ fn ignores_absurd_cbor_map_count() { assert!(m.language.is_none()); assert!(m.compiler_version.is_none()); } + +#[test] +fn decode_backfills_creation_and_runtime_opcode_sets_idempotently() { + let dir = TestDir::new("opcode_sets"); + let parquet_path = dir.path().join("contracts__0000000100__0000000100.parquet"); + let parquet_path_sql = parquet_path.display().to_string().replace('\'', "''"); + Connection::open_in_memory() + .unwrap() + .execute_batch(&format!( + r#" + COPY ( + SELECT + 100::UINTEGER AS block_number, + unhex(repeat('01', 32)) AS block_hash, + 0::UINTEGER AS create_index, + unhex(repeat('02', 32)) AS transaction_hash, + unhex(repeat('03', 20)) AS contract_address, + unhex(repeat('04', 20)) AS deployer, + unhex(repeat('05', 20)) AS factory, + unhex('f500') AS init_code, + unhex('61f4f5f4') AS code, + unhex(repeat('06', 32)) AS init_code_hash, + 2::UINTEGER AS n_init_code_bytes, + 4::UINTEGER AS n_code_bytes, + unhex(repeat('07', 32)) AS code_hash, + 1::UBIGINT AS chain_id + ) TO '{parquet_path_sql}' (FORMAT PARQUET); + "# + )) + .unwrap(); + + for _ in 0..2 { + let output = Command::new(env!("CARGO_BIN_EXE_blink")) + .args([ + "decode", + "--data-dir", + dir.path().to_str().unwrap(), + "--batch-size", + "1", + ]) + .output() + .unwrap(); + assert!( + output.status.success(), + "blink decode failed:\n{}", + String::from_utf8_lossy(&output.stderr) + ); + } + + let conn = Connection::open(dir.path().join("blink.duckdb")).unwrap(); + let (sets, links, runs): (u64, u64, u64) = conn + .query_row( + r#" + SELECT + (SELECT COUNT(*) FROM bytecode_opcode_sets), + (SELECT COUNT(*) FROM contract_creation_bytecodes), + (SELECT COUNT(*) FROM opcode_decode_runs) + "#, + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .unwrap(); + assert_eq!((sets, links, runs), (2, 1, 1)); + + let (creation_has_create2, runtime_has_delegatecall, runtime_has_create2): (bool, bool, bool) = + conn.query_row( + r#" + SELECT + (SELECT list_contains(opcodes, 'CREATE2') + FROM bytecode_opcode_sets WHERE code_kind = 'creation'), + (SELECT list_contains(opcodes, 'DELEGATECALL') + FROM bytecode_opcode_sets WHERE code_kind = 'runtime'), + (SELECT list_contains(opcodes, 'CREATE2') + FROM bytecode_opcode_sets WHERE code_kind = 'runtime') + "#, + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .unwrap(); + assert!(creation_has_create2); + assert!(runtime_has_delegatecall); + assert!( + !runtime_has_create2, + "CREATE2 bytes inside PUSH2 data must not be indexed as an instruction" + ); +} From dc61b9e83bca1f2df0d41c40618035f69b8ca07b Mon Sep 17 00:00:00 2001 From: g4titanx Date: Mon, 3 Aug 2026 22:25:32 +0100 Subject: [PATCH 3/5] docs: document contract opcode queries --- README.md | 37 +++++++++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/README.md b/README.md index b429ede..ff34f6c 100644 --- a/README.md +++ b/README.md @@ -88,6 +88,43 @@ Each parquet file contains contract creations with this schema: | code_hash | binary | Keccak256 of deployed code | | chain_id | uint64 | Chain ID | +## Opcode queries + +Run the decoder after loading or extracting contract data. It stores compact, +deduplicated opcode sets for both constructor and deployed runtime bytecode: + +```bash +blink decode --data-dir ./data/blink +``` + +The SQL explorer exposes `contract_opcodes` with one row per contract and code +kind. Opcode names are canonical uppercase mnemonics supplied by `eot`. + +```sql +-- Constructor/creation bytecode only +SELECT address +FROM contract_opcodes +WHERE code_kind = 'creation' + AND list_contains(opcodes, 'CREATE2'); + +-- Deployed runtime bytecode only +SELECT address +FROM contract_opcodes +WHERE code_kind = 'runtime' + AND list_contains(opcodes, 'DELEGATECALL'); + +-- Either kind +SELECT DISTINCT address +FROM contract_opcodes +WHERE list_contains(opcodes, 'SELFDESTRUCT'); +``` + +`PUSH1` through `PUSH32` payload bytes are not interpreted as opcodes. Runtime +compiler metadata is excluded when recognized. Creation analysis follows +statically discoverable control-flow from program counter zero so embedded +runtime code and constructor arguments are not indexed as constructor +instructions merely because they contain opcode-shaped bytes. + ## Performance blink is designed for speed: From 4866d456ece427c9a85ed83c7a36ae5322423159 Mon Sep 17 00:00:00 2001 From: g4titanx Date: Mon, 3 Aug 2026 22:57:13 +0100 Subject: [PATCH 4/5] fix: remove duckdb cli dependency from decode --- src/decode/mod.rs | 74 ++++++++++++++++------------------------------- tests/decode.rs | 1 + 2 files changed, 26 insertions(+), 49 deletions(-) diff --git a/src/decode/mod.rs b/src/decode/mod.rs index 86d216d..4d49af0 100644 --- a/src/decode/mod.rs +++ b/src/decode/mod.rs @@ -480,6 +480,8 @@ fn run_decode_blocking(args: DecodeArgs, data_dir: PathBuf) -> Result<()> { } fn copy_decode_rows_to_tsv(parquet_file: &Path, temp_path: &Path) -> Result<()> { + let parquet_conn = + Connection::open_in_memory().context("open bundled DuckDB for decode row export")?; let parquet_path = parquet_file.display().to_string().replace('\'', "''"); let temp_path_str = temp_path.display().to_string().replace('\'', "''"); // Cryo's schema has a precomputed `n_code_bytes` (uint32). Filtering on @@ -488,7 +490,7 @@ fn copy_decode_rows_to_tsv(parquet_file: &Path, temp_path: &Path) -> Result<()> // that column, so we fall back to `octet_length(code)` for those files — // they're the official paradigm dataset and don't have the corruption // problem in practice. - let has_n_code_bytes = parquet_has_column(parquet_file, "n_code_bytes")?; + let has_n_code_bytes = parquet_has_column(&parquet_conn, parquet_file, "n_code_bytes")?; let length_filter = if has_n_code_bytes { format!( "AND n_code_bytes IS NOT NULL @@ -513,15 +515,9 @@ fn copy_decode_rows_to_tsv(parquet_file: &Path, temp_path: &Path) -> Result<()> {length_filter} ) TO '{temp_path_str}' (FORMAT CSV, HEADER false, DELIMITER '\\t');", ); - let output = std::process::Command::new("duckdb") - .arg("-c") - .arg(&sql) - .output() - .context("run duckdb CLI for decode row export")?; - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - return Err(anyhow!("duckdb CLI export failed: {}", stderr.trim())); - } + parquet_conn + .execute_batch(&sql) + .with_context(|| format!("export decode rows from {}", parquet_file.display()))?; Ok(()) } @@ -1027,7 +1023,9 @@ fn flush_opcode_contract_batch( } fn copy_opcode_rows_to_tsv(parquet_file: &Path, temp_path: &Path) -> Result<()> { - let columns = parquet_columns(parquet_file)?; + let parquet_conn = + Connection::open_in_memory().context("open bundled DuckDB for opcode row export")?; + let columns = parquet_columns(&parquet_conn, parquet_file)?; if !columns.contains("contract_address") || !columns.contains("code") { return Err(anyhow!("parquet lacks contract_address/code columns")); } @@ -1083,37 +1081,24 @@ fn copy_opcode_rows_to_tsv(parquet_file: &Path, temp_path: &Path) -> Result<()> ) TO '{temp_path}' (FORMAT CSV, HEADER false, DELIMITER '\t'); "# ); - let output = std::process::Command::new("duckdb") - .arg("-c") - .arg(&sql) - .output() - .context("run duckdb CLI for opcode row export")?; - if !output.status.success() { - return Err(anyhow!( - "duckdb CLI opcode export failed: {}", - String::from_utf8_lossy(&output.stderr).trim() - )); - } + parquet_conn + .execute_batch(&sql) + .with_context(|| format!("export opcode rows from {}", parquet_file.display()))?; Ok(()) } -fn parquet_columns(parquet_file: &Path) -> Result> { +fn parquet_columns( + conn: &Connection, + parquet_file: &Path, +) -> Result> { let parquet_path = parquet_file.display().to_string().replace('\'', "''"); let sql = format!("SELECT string_agg(DISTINCT name, ',') FROM parquet_schema('{parquet_path}');"); - let output = std::process::Command::new("duckdb") - .arg("-noheader") - .arg("-c") - .arg(&sql) - .output() - .context("run duckdb CLI for parquet column discovery")?; - if !output.status.success() { - return Err(anyhow!( - "duckdb CLI schema discovery failed: {}", - String::from_utf8_lossy(&output.stderr).trim() - )); - } - Ok(String::from_utf8_lossy(&output.stdout) + let names: Option = conn + .query_row(&sql, [], |row| row.get(0)) + .with_context(|| format!("read parquet schema for {}", parquet_file.display()))?; + Ok(names + .unwrap_or_default() .trim() .split(',') .map(str::trim) @@ -1226,22 +1211,13 @@ pub(crate) fn flush_creation_links( Ok(inserted as u64) } -fn parquet_has_column(parquet_file: &Path, column: &str) -> Result { +fn parquet_has_column(conn: &Connection, parquet_file: &Path, column: &str) -> Result { let parquet_path = parquet_file.display().to_string().replace('\'', "''"); let sql = format!("SELECT COUNT(*) FROM parquet_schema('{parquet_path}') WHERE name = '{column}';"); - let output = std::process::Command::new("duckdb") - .arg("-noheader") - .arg("-c") - .arg(&sql) - .output() - .context("run duckdb CLI for parquet schema check")?; - if !output.status.success() { - let stderr = String::from_utf8_lossy(&output.stderr); - return Err(anyhow!("duckdb CLI schema check failed: {}", stderr.trim())); - } - let stdout = String::from_utf8_lossy(&output.stdout); - let count: u64 = stdout.trim().parse().unwrap_or(0); + let count: i64 = conn + .query_row(&sql, [], |row| row.get(0)) + .with_context(|| format!("read parquet schema for {}", parquet_file.display()))?; Ok(count > 0) } diff --git a/tests/decode.rs b/tests/decode.rs index 3eec296..7027f6f 100644 --- a/tests/decode.rs +++ b/tests/decode.rs @@ -159,6 +159,7 @@ fn decode_backfills_creation_and_runtime_opcode_sets_idempotently() { "--batch-size", "1", ]) + .env("PATH", "") .output() .unwrap(); assert!( From eaf8c6af2255208145d8556298dc36b55bdc42c7 Mon Sep 17 00:00:00 2001 From: g4titanx Date: Tue, 4 Aug 2026 02:53:46 +0100 Subject: [PATCH 5/5] feat: paginate SQL explorer query results --- README.md | 5 +++++ src/db/queries.rs | 17 ++++++++++++++++- src/db/sql.rs | 14 ++++++++++++-- src/serve.rs | 7 +++++-- tests/db.rs | 32 ++++++++++++++++++++++++++++++++ 5 files changed, 70 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index ff34f6c..8982a2a 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,11 @@ FROM contract_opcodes WHERE list_contains(opcodes, 'SELFDESTRUCT'); ``` +The HTTP SQL explorer returns at most 1,000 rows per request. Its response +includes `offset` and `has_more`; pass the next `offset` to `POST /api/query` +to page through every match. Use a deterministic `ORDER BY` when paging, and +omit an SQL `LIMIT` if the full result set should remain available. + `PUSH1` through `PUSH32` payload bytes are not interpreted as opcodes. Runtime compiler metadata is excluded when recognized. Creation analysis follows statically discoverable control-flow from program counter zero so embedded diff --git a/src/db/queries.rs b/src/db/queries.rs index edfe1c8..4a02a14 100644 --- a/src/db/queries.rs +++ b/src/db/queries.rs @@ -220,12 +220,23 @@ impl Db { sql: String, limit: u32, chain_id: Option, + ) -> Result { + self.query_sql_page(sql, limit, 0, chain_id).await + } + + pub async fn query_sql_page( + &self, + sql: String, + limit: u32, + offset: u64, + chain_id: Option, ) -> Result { let normalized = sql::normalize_read_only_sql(&sql)?; let limit = limit.clamp(1, 1_000); + let fetch_limit = limit + 1; self.run_read(move |conn| { let started = Instant::now(); - let wrapped = sql::wrap_dashboard_query(&normalized, limit, chain_id); + let wrapped = sql::wrap_dashboard_query(&normalized, fetch_limit, offset, chain_id); let mut stmt = conn.prepare(&wrapped).context("prepare dashboard query")?; let mut rows = stmt.query([]).context("execute dashboard query")?; let (columns, column_count) = { @@ -242,6 +253,8 @@ impl Db { } out.push(values); } + let has_more = out.len() > limit as usize; + out.truncate(limit as usize); let elapsed_ms = started.elapsed().as_millis(); if elapsed_ms >= 1_000 { // The HTTP-level slow log can't see the request body; name @@ -258,6 +271,8 @@ impl Db { row_count: out.len(), rows: out, limit, + offset, + has_more, elapsed_ms, }) }) diff --git a/src/db/sql.rs b/src/db/sql.rs index c5bc824..2419c68 100644 --- a/src/db/sql.rs +++ b/src/db/sql.rs @@ -11,6 +11,8 @@ pub struct SqlQueryResult { pub rows: Vec>, pub row_count: usize, pub limit: u32, + pub offset: u64, + pub has_more: bool, pub elapsed_ms: u128, } @@ -111,7 +113,12 @@ pub(crate) fn normalize_read_only_sql(sql: &str) -> Result { Ok(without_trailing_semicolon.to_string()) } -pub(crate) fn wrap_dashboard_query(sql: &str, limit: u32, chain_id: Option) -> String { +pub(crate) fn wrap_dashboard_query( + sql: &str, + limit: u32, + offset: u64, + chain_id: Option, +) -> String { match chain_id { Some(chain_id) => format!( r#" @@ -128,9 +135,12 @@ pub(crate) fn wrap_dashboard_query(sql: &str, limit: u32, chain_id: Option) SELECT * FROM ({sql}) AS _blink_dashboard_query LIMIT {limit} + OFFSET {offset} "# ), - None => format!("SELECT * FROM ({sql}) AS _blink_dashboard_query LIMIT {limit}"), + None => { + format!("SELECT * FROM ({sql}) AS _blink_dashboard_query LIMIT {limit} OFFSET {offset}") + } } } diff --git a/src/serve.rs b/src/serve.rs index c4b7b90..5bf53a0 100644 --- a/src/serve.rs +++ b/src/serve.rs @@ -1398,8 +1398,10 @@ async fn recent_handler( struct SqlQueryRequest { /// Read-only SQL query over dashboard views. sql: String, - /// Maximum rows to return. + /// Maximum rows to return per page (capped at 1,000). limit: Option, + /// Number of matching rows to skip before returning this page. + offset: Option, /// Chain id used to scope the `contract_metadata` dashboard view. chain_id: Option, } @@ -1421,9 +1423,10 @@ async fn query_handler( ) -> Result, AppError> { state .db - .query_sql( + .query_sql_page( req.sql, req.limit.unwrap_or(100), + req.offset.unwrap_or(0), Some(selected_chain_id(req.chain_id)), ) .await diff --git a/tests/db.rs b/tests/db.rs index 7b9ba9a..77123a6 100644 --- a/tests/db.rs +++ b/tests/db.rs @@ -471,6 +471,38 @@ async fn dashboard_sql_scopes_contract_metadata_by_chain() { assert_eq!(result.rows[0][1], serde_json::json!(300)); } +#[tokio::test] +async fn dashboard_sql_pages_through_all_matching_rows() { + let dir = TestDir::new("query_pagination"); + let db = Db::open_with_mode(&dir.path, "*.parquet", false).unwrap(); + let sql = "SELECT range AS value FROM range(2505) ORDER BY value".to_string(); + + let first = db + .query_sql_page(sql.clone(), 1_000, 0, Some(1)) + .await + .unwrap(); + assert_eq!(first.row_count, 1_000); + assert_eq!(first.offset, 0); + assert!(first.has_more); + assert_eq!(first.rows[0][0], serde_json::json!(0)); + assert_eq!(first.rows[999][0], serde_json::json!(999)); + + let second = db + .query_sql_page(sql.clone(), 1_000, 1_000, Some(1)) + .await + .unwrap(); + assert_eq!(second.row_count, 1_000); + assert_eq!(second.offset, 1_000); + assert!(second.has_more); + assert_eq!(second.rows[0][0], serde_json::json!(1_000)); + + let last = db.query_sql_page(sql, 1_000, 2_000, Some(1)).await.unwrap(); + assert_eq!(last.row_count, 505); + assert_eq!(last.offset, 2_000); + assert!(!last.has_more); + assert_eq!(last.rows[504][0], serde_json::json!(2_504)); +} + #[tokio::test] async fn contract_opcodes_queries_creation_and_runtime_separately() { let dir = TestDir::new("contract_opcodes");