From ccedaa3227ff314c63d179204bae6039d5675e04 Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Fri, 27 Feb 2026 08:55:13 +0300 Subject: [PATCH] add audit command, implement fjall only for now --- src/bin/allegedly.rs | 9 ++ src/bin/audit.rs | 77 ++++++++++ src/bin/backfill.rs | 25 +--- src/lib.rs | 38 ++++- src/plc_fjall.rs | 337 ++++++++++++++++++++++++++++++++++--------- 5 files changed, 400 insertions(+), 86 deletions(-) create mode 100644 src/bin/audit.rs diff --git a/src/bin/allegedly.rs b/src/bin/allegedly.rs index 7989231..bbac072 100644 --- a/src/bin/allegedly.rs +++ b/src/bin/allegedly.rs @@ -5,6 +5,7 @@ use std::{path::PathBuf, time::Duration, time::Instant}; use tokio::fs::create_dir_all; use tokio::sync::mpsc; +mod audit; mod backfill; mod mirror; @@ -59,6 +60,13 @@ enum Commands { #[command(flatten)] instrumentation: InstrumentationArgs, }, + /// Audit a plc database for correctness + Audit { + #[command(flatten)] + args: audit::Args, + #[command(flatten)] + instrumentation: InstrumentationArgs, + }, /// Poll an upstream PLC server and log new ops to stdout Tail { /// Begin tailing from a specific timestamp for replay or wait-until @@ -118,6 +126,7 @@ async fn main() -> anyhow::Result<()> { } Commands::Mirror { args, .. } => mirror::run(globals, args, true).await?, Commands::Wrap { args, .. } => mirror::run(globals, args, false).await?, + Commands::Audit { args, .. } => audit::run(args).await?, Commands::Tail { after } => { let mut url = globals.upstream; url.set_path("/export"); diff --git a/src/bin/audit.rs b/src/bin/audit.rs new file mode 100644 index 0000000..78d1fc7 --- /dev/null +++ b/src/bin/audit.rs @@ -0,0 +1,77 @@ +use allegedly::{ + FjallDb, audit_fjall, + bin::{InstrumentationArgs, bin_init}, + drop_invalid_ops_fjall, file_to_invalid_ops, invalid_ops_to_stdout, logo, +}; +use clap::Parser; +use std::path::PathBuf; +use tokio::task::JoinSet; + +#[derive(Debug, clap::Args)] +pub struct Args { + /// path to a local fjall database directory + #[arg(long, env = "ALLEGEDLY_FJALL")] + fjall: Option, + /// path to a file containing invalid ops to fix + #[arg(long, env = "ALLEGEDLY_FIX")] + fix: Option, +} + +pub async fn run(Args { fjall, fix }: Args) -> anyhow::Result<()> { + let mut tasks = JoinSet::new(); + + if let Some(fjall) = fjall { + let (invalid_ops_tx, invalid_ops_rx) = tokio::sync::mpsc::channel(128); + let db = FjallDb::open(&fjall)?; + + if let Some(fix) = fix { + tasks.spawn(file_to_invalid_ops(fix, invalid_ops_tx)); + tasks.spawn(drop_invalid_ops_fjall(db, invalid_ops_rx)); + } else { + tasks.spawn(audit_fjall(db, invalid_ops_tx)); + tasks.spawn(invalid_ops_to_stdout(invalid_ops_rx)); + } + } else { + anyhow::bail!("no audit target provided"); + } + + while let Some(next) = tasks.join_next().await { + match next { + Err(e) if e.is_panic() => { + log::error!("a joinset task panicked: {e}. bailing now. (should we panic?)"); + return Err(e.into()); + } + Err(e) => { + log::error!("a joinset task failed to join: {e}"); + return Err(e.into()); + } + Ok(Err(e)) => { + log::error!("a joinset task completed with error: {e}"); + return Err(e); + } + Ok(Ok(name)) => { + log::trace!("a task completed: {name:?}. {} left", tasks.len()); + } + } + } + + Ok(()) +} + +#[derive(Debug, Parser)] +struct CliArgs { + #[command(flatten)] + instrumentation: InstrumentationArgs, + #[command(flatten)] + args: Args, +} + +#[allow(dead_code)] +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let args = CliArgs::parse(); + bin_init(args.instrumentation.enable_opentelemetry); + log::info!("{}", logo("audit")); + run(args.args).await?; + Ok(()) +} diff --git a/src/bin/backfill.rs b/src/bin/backfill.rs index 3b90425..3477544 100644 --- a/src/bin/backfill.rs +++ b/src/bin/backfill.rs @@ -44,21 +44,16 @@ pub struct Args { /// Cert for postgres (if needed) #[arg(long)] postgres_cert: Option, - /// Delete all operations from the postgres db before starting + /// Delete all operations from the db before starting /// - /// only used if `--to-postgres` is present + /// only used if `--to-postgres` or `--to-fjall` is present #[arg(long, action)] - postgres_reset: bool, + reset: bool, /// Bulk load into a local fjall embedded database /// /// Pass a directory path for the fjall database - #[arg(long, conflicts_with_all = ["to_postgres", "postgres_cert", "postgres_reset"])] + #[arg(long, conflicts_with_all = ["to_postgres", "postgres_cert"])] to_fjall: Option, - /// Delete all operations from the fjall db before starting - /// - /// only used if `--to-fjall` is present - #[arg(long, action, requires = "to_fjall")] - fjall_reset: bool, /// Stop at the week ending before this date #[arg(long)] until: Option
, @@ -80,9 +75,8 @@ pub async fn run( source_workers, to_postgres, postgres_cert, - postgres_reset, + reset, to_fjall, - fjall_reset, until, catch_up, }: Args, @@ -180,7 +174,7 @@ pub async fn run( tasks.spawn(backfill_to_fjall( db.clone(), - fjall_reset, + reset, bulk_out, found_last_tx, )); @@ -192,12 +186,7 @@ pub async fn run( let db = Db::new(pg_url.as_str(), postgres_cert).await?; log::trace!("connected to postgres"); - tasks.spawn(backfill_to_pg( - db.clone(), - postgres_reset, - bulk_out, - found_last_tx, - )); + tasks.spawn(backfill_to_pg(db.clone(), reset, bulk_out, found_last_tx)); if catch_up { tasks.spawn(pages_to_pg(db, full_out)); } diff --git a/src/lib.rs b/src/lib.rs index 29f5bc6..8a63cf6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,7 +20,7 @@ pub use backfill::backfill; pub use cached_value::{CachedValue, Fetcher}; pub use client::{CLIENT, UA}; pub use mirror::{ExperimentalConf, ListenConf, serve, serve_fjall}; -pub use plc_fjall::{FjallDb, backfill_to_fjall, pages_to_fjall}; +pub use plc_fjall::{FjallDb, audit as audit_fjall, backfill_to_fjall, pages_to_fjall, drop_invalid_ops as drop_invalid_ops_fjall}; pub use plc_pg::{Db, backfill_to_pg, pages_to_pg}; pub use poll::{PageBoundaryState, get_page, poll_upstream}; pub use ratelimit::{CreatePlcOpLimiter, GovernorMiddleware, IpLimiters}; @@ -138,6 +138,42 @@ pub async fn pages_to_stdout( Ok("pages_to_stdout") } +pub async fn invalid_ops_to_stdout( + mut rx: mpsc::Receiver<(String, Dt, String)>, +) -> anyhow::Result<&'static str> { + while let Some((did, at, cid)) = rx.recv().await { + let val = serde_json::json!({ + "did": did, + "at": at, + "cid": cid, + }); + println!("{val}"); + } + Ok("invalid_ops_to_stdout") +} + +pub async fn file_to_invalid_ops( + path: impl AsRef, + tx: mpsc::Sender<(String, Dt, String)>, +) -> anyhow::Result<&'static str> { + let file = tokio::fs::File::open(path).await?; + + use tokio::io::AsyncBufReadExt; + let mut lines = tokio::io::BufReader::new(file).lines(); + while let Some(line) = lines.next_line().await? { + #[derive(serde::Deserialize)] + struct Op { + did: String, + at: Dt, + cid: String, + } + let op: Op = serde_json::from_str(&line)?; + tx.send((op.did, op.at, op.cid)).await?; + } + + Ok("invalid_ops_to_stdout") +} + pub fn logo(name: &str) -> String { format!( r" diff --git a/src/plc_fjall.rs b/src/plc_fjall.rs index 58119a2..4aeaa47 100644 --- a/src/plc_fjall.rs +++ b/src/plc_fjall.rs @@ -1,6 +1,6 @@ use crate::{ BundleSource, Dt, ExportPage, Op as CommonOp, PageBoundaryState, Week, - crypto::{DidKey, Signature, assure_valid_sig}, + crypto::{AssuranceResults, DidKey, Signature, assure_valid_sig}, }; use anyhow::Context; use data_encoding::BASE32_NOPAD; @@ -797,6 +797,32 @@ impl StoredOp { } } +fn verify_op_sig(op: &StoredOp, prev: Option<&StoredOp>) -> anyhow::Result { + let keys: Vec<&DidKey> = match &op.prev { + None => op.get_keys(), + Some(_) => match prev { + None => anyhow::bail!("prev cid exists but the op for that cid is missing"), + Some(p) => p.get_keys(), + }, + }; + + if keys.is_empty() { + anyhow::bail!("no keys found for genesis op or prev op"); + } + + let data = { + let serde_json::Value::Object(mut data) = op.to_json_value() else { + unreachable!("we know op is valid, because it comes from StoredOp") + }; + data.remove("sig"); + serde_json::Value::Object(data) + }; + + let results = assure_valid_sig(keys, &op.sig, &data) + .expect("that our op is an object and we removed sig field"); + Ok(results) +} + // this is basically Op, but without the cid and created_at fields // since we have them in the key already #[derive(Debug, Deserialize, Serialize)] @@ -941,40 +967,27 @@ impl FjallDb { .transpose()? .flatten(); - let keys: Vec<&DidKey> = match &operation.prev { - None => operation.get_keys(), - Some(_) => match &prev_op { - None => { - log::error!( - "op {} {} has prev but the prev op is not found", - op.did, - op.cid - ); + let prev_stored = prev_op.as_ref().map(|(_, _, p)| &p.operation); + + match verify_op_sig(&operation, prev_stored) { + Ok(results) => { + if !results.valid { + let msg = results + .errors + .iter() + .map(|e| e.to_string()) + .collect::>() + .join("\n"); + log::warn!("invalid op {} {}:\n{msg}", op.did, op.cid); return Ok(0); } - Some((_, _, prev)) => prev.operation.get_keys(), - }, - }; - - if keys.is_empty() { - log::warn!("no keys for op {} {}", op.did, op.cid); - return Ok(0); - } - - let data = { - let serde_json::Value::Object(mut data) = operation.to_json_value() else { - unreachable!("we checked if operation is valid already") - }; - data.remove("sig"); - serde_json::Value::Object(data) - }; - let results = assure_valid_sig(keys, &operation.sig, &data)?; - if !results.valid { - for err in results.errors { - log::warn!("invalid signature for op {} {}: {err}", op.did, op.cid); } - return Ok(0); + Err(e) => { + log::warn!("invalid op {} {}: {e}", op.did, op.cid); + return Ok(0); + } } + log::debug!("verified op {} {}", op.did, op.cid); } let db_op = DbOp { @@ -1004,6 +1017,38 @@ impl FjallDb { Ok(1) } + fn decode_by_did_entry( + &self, + by_did_key: &[u8], + prefix_len: usize, + ) -> anyhow::Result<(Dt, PlcCid, DbOp)> { + let key_rest = by_did_key + .get(prefix_len..) + .ok_or_else(|| anyhow::anyhow!("invalid by_did key {by_did_key:?}"))?; + + let ts_bytes = key_rest + .get(..8) + .ok_or_else(|| anyhow::anyhow!("invalid length: {key_rest:?}"))?; + let cid_suffix = key_rest + .get(9..) + .ok_or_else(|| anyhow::anyhow!("invalid length: {key_rest:?}"))?; + + let op_key = [ts_bytes, &[SEP][..], cid_suffix].concat(); + let ts = decode_timestamp(ts_bytes)?; + + let value = self + .inner + .ops + .get(&op_key)? + .ok_or_else(|| anyhow::anyhow!("op not found: {op_key:?}"))?; + + let op: DbOp = rmp_serde::from_slice(&value)?; + let mut full_cid = op.cid_prefix.clone(); + full_cid.extend_from_slice(cid_suffix); + + Ok((ts, PlcCid(full_cid), op)) + } + fn _ops_for_did( &self, did: &str, @@ -1015,34 +1060,7 @@ impl FjallDb { let (by_did_key, _) = guard .into_inner() .map_err(|e| anyhow::anyhow!("fjall read error: {e}"))?; - - let key_rest = by_did_key - .get(prefix.len()..) - .ok_or_else(|| anyhow::anyhow!("invalid by_did key {by_did_key:?}"))?; - - let ts_bytes = key_rest - .get(..8) - .ok_or_else(|| anyhow::anyhow!("invalid length: {key_rest:?}"))?; - let cid_suffix = key_rest - .get(9..) - .ok_or_else(|| anyhow::anyhow!("invalid length: {key_rest:?}"))?; - - let op_key = [ts_bytes, &[SEP][..], cid_suffix].concat(); - let ts = decode_timestamp(ts_bytes)?; - - let value = self - .inner - .ops - .get(&op_key)? - .ok_or_else(|| anyhow::anyhow!("op not found: {op_key:?}"))?; - - let op: DbOp = rmp_serde::from_slice(&value)?; - let mut full_cid_bytes = op.cid_prefix.clone(); - full_cid_bytes.extend_from_slice(cid_suffix); - - let cid = PlcCid(full_cid_bytes); - - Ok((ts, cid, op)) + self.decode_by_did_entry(&by_did_key, prefix.len()) })) } @@ -1111,14 +1129,168 @@ impl FjallDb { })) } - pub fn export_ops_week( + pub fn drop_op(&self, did_str: &str, created_at: &Dt, cid: &str) -> anyhow::Result<()> { + let cid = decode_cid_str(cid)?; + let cid_suffix = &cid[30..]; + + let op_key = op_key(created_at, cid_suffix); + let by_did_key = by_did_key(did_str, created_at, cid_suffix)?; + + let mut batch = self.inner.db.batch(); + batch.remove(&self.inner.ops, op_key); + batch.remove(&self.inner.by_did, by_did_key); + batch.commit()?; + + Ok(()) + } + + pub fn audit( &self, - week: Week, - ) -> anyhow::Result> + '_> { - let after: Dt = week.into(); - let before: Dt = week.next().into(); + invalid_ops_tx: mpsc::Sender<(String, Dt, String)>, + ) -> anyhow::Result<(usize, usize)> { + use std::sync::mpsc; + + let ops = self.inner.by_did.len()?; + + let workers = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(4); + + type Batch = (Vec, Vec<(Dt, PlcCid, DbOp)>); + let (result_tx, result_rx) = mpsc::sync_channel::>(workers); + + let channels: Vec<_> = (0..workers) + .map(|_| mpsc::sync_channel::(512)) + .collect(); + let senders: Vec<_> = channels.iter().map(|(tx, _)| tx.clone()).collect(); + + std::thread::scope(|s| { + for (_, rx) in channels { + let result_tx = result_tx.clone(); + let invalid_ops_tx = invalid_ops_tx.clone(); + s.spawn(move || { + let mut checked: usize = 0; + let mut failed: usize = 0; + while let Ok((did_prefix, ops)) = rx.recv() { + let did = decode_did(&did_prefix[..did_prefix.len() - 1]); + for (ts, cid, op) in &ops { + checked += 1; + let prev_op = op.operation.prev.as_ref().and_then(|expected| { + ops.iter().find(|(_, c, _)| c == expected) + }); + let prev_cid_ok = op.operation.prev.is_none() || prev_op.is_some(); + if !prev_cid_ok { + log::error!("audit: op {did} {cid} prev cid mismatch or missing predecessor, is db corrupted?"); + failed += 1; + let _ = invalid_ops_tx.blocking_send((did.clone(), ts.clone(), cid.to_string())); + continue; + } + let prev_stored = prev_op.map(|(_, _, p)| &p.operation); + match verify_op_sig(&op.operation, prev_stored) { + Ok(results) => { + if !results.valid { + let msg = results + .errors + .iter() + .map(|e| e.to_string()) + .collect::>() + .join("\n "); + log::warn!("audit: invalid op {} {}:\n {msg}", did, cid); + failed += 1; + let _ = invalid_ops_tx.blocking_send((did.clone(), ts.clone(), cid.to_string())); + } + } + Err(e) => { + log::warn!("audit: invalid op {} {}: {e}", did, cid); + failed += 1; + let _ = invalid_ops_tx.blocking_send((did.clone(), ts.clone(), cid.to_string())); + } + } + } + } + let _ = result_tx.send(Ok((checked, failed))); + }); + } + drop(result_tx); + + // todo: probably dont use a macro... + macro_rules! spawn_scan_thread { + ($iter_method:ident, $start_idx:expr, $reverse:expr, $limit:expr) => {{ + let senders = senders.clone(); + let mut iter = self.inner.by_did.iter(); + + s.spawn(move || -> anyhow::Result<()> { + let mut current_prefix: Option<[u8; 16]> = None; + let mut did_ops: Vec<(Dt, PlcCid, DbOp)> = Vec::new(); + let mut idx = $start_idx; + let mut processed_ops: usize = 0; + + while let Some(guard) = iter.$iter_method() { + let (by_did_key, _) = guard + .into_inner() + .map_err(|e| anyhow::anyhow!("fjall read error: {e}"))?; + + let mut prefix_array = [0u8; 16]; + prefix_array.copy_from_slice(by_did_key.get(..16).ok_or_else( + || anyhow::anyhow!("by_did key too short: {by_did_key:?}"), + )?); + + let op = self.decode_by_did_entry(&by_did_key, 16)?; + + if current_prefix.map_or(true, |cp| cp != prefix_array) { + // new did, push the ops + if let Some(prefix) = current_prefix.take() { + if $reverse { + did_ops.reverse(); + } + senders[idx % workers] + .send((prefix.to_vec(), std::mem::take(&mut did_ops))) + .ok(); + idx += 1; + + if processed_ops >= $limit { + break; + } + } + current_prefix = Some(prefix_array); + } + + did_ops.push(op); + processed_ops += 1; + } - self.export_ops(after..before) + if let Some(prefix) = current_prefix { + if $reverse { + did_ops.reverse(); + } + senders[idx % workers].send((prefix.to_vec(), did_ops)).ok(); + } + + Ok(()) + }) + }}; + } + + // we can start two threads, one for forward iteration and one for reverse iteration + // this way we have two scans in parallel which should be faster! + let f_handle = spawn_scan_thread!(next, 0, false, ops / 2); + let b_handle = spawn_scan_thread!(next_back, workers / 2, true, ops - (ops / 2)); + + f_handle.join().unwrap()?; + b_handle.join().unwrap()?; + + drop(senders); + + let mut total_checked: usize = 0; + let mut total_failed: usize = 0; + for res in result_rx { + let (c, f) = res?; + total_checked += c; + total_failed += f; + } + + Ok((total_checked, total_failed)) + }) } } @@ -1130,10 +1302,13 @@ impl BundleSource for FjallDb { let db = self.clone(); async move { - let (mut tx, rx) = tokio::io::duplex(1024 * 1024 * 64); + let (mut tx, rx) = tokio::io::duplex(1024 * 1024 * 16); tokio::task::spawn_blocking(move || -> anyhow::Result<()> { - let iter = db.export_ops_week(week)?; + let after: Dt = week.into(); + let before: Dt = week.next().into(); + + let iter = db.export_ops(after..before)?; let rt = tokio::runtime::Handle::current(); @@ -1184,6 +1359,7 @@ pub async fn backfill_to_fjall( loop { let pages_finished = pages.is_closed(); + // we can stop if we have no more pages and all the insert tasks are finished if pages_finished && insert_tasks.is_empty() { break; } @@ -1268,6 +1444,33 @@ pub async fn pages_to_fjall( Ok("pages_to_fjall") } +pub async fn audit( + db: FjallDb, + invalid_ops_tx: mpsc::Sender<(String, Dt, String)>, +) -> anyhow::Result<&'static str> { + log::info!("starting fjall audit..."); + let t0 = std::time::Instant::now(); + let (checked, failed) = tokio::task::spawn_blocking(move || db.audit(invalid_ops_tx)).await??; + log::info!( + "fjall audit complete in {:?}, {checked} ops checked", + t0.elapsed() + ); + if failed > 0 { + anyhow::bail!("audit found {failed} invalid operations"); + } + Ok("audit_fjall") +} + +pub async fn drop_invalid_ops( + db: FjallDb, + mut invalid_ops_rx: mpsc::Receiver<(String, Dt, String)>, +) -> anyhow::Result<&'static str> { + while let Some((did, at, cid)) = invalid_ops_rx.recv().await { + db.drop_op(&did, &at, &cid)?; + } + Ok("drop_invalid_ops") +} + #[cfg(test)] mod tests { use super::*; -- 2.51.2