From bb4a7328dd6bb065ec9c6599a26edcf9e0ecdb0a Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Sat, 28 Feb 2026 07:27:10 +0300 Subject: [PATCH] allow audit to fix ops using upstream --- src/bin/allegedly.rs | 2 +- src/bin/audit.rs | 17 ++++--- src/lib.rs | 37 ++++++++------- src/plc_fjall.rs | 109 +++++++++++++++++++++++++++++++++++++------ 4 files changed, 126 insertions(+), 39 deletions(-) diff --git a/src/bin/allegedly.rs b/src/bin/allegedly.rs index bbac072..4a5050a 100644 --- a/src/bin/allegedly.rs +++ b/src/bin/allegedly.rs @@ -126,7 +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::Audit { args, .. } => audit::run(globals, 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 index 78d1fc7..3e71670 100644 --- a/src/bin/audit.rs +++ b/src/bin/audit.rs @@ -1,7 +1,7 @@ use allegedly::{ FjallDb, audit_fjall, - bin::{InstrumentationArgs, bin_init}, - drop_invalid_ops_fjall, file_to_invalid_ops, invalid_ops_to_stdout, logo, + bin::{GlobalArgs, InstrumentationArgs, bin_init}, + file_to_invalid_ops, fix_ops_fjall, invalid_ops_to_stdout, logo, }; use clap::Parser; use std::path::PathBuf; @@ -12,12 +12,15 @@ 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 + /// path to a file containing invalid ops to fix using upstream #[arg(long, env = "ALLEGEDLY_FIX")] fix: Option, + /// drop invalid ops instead of trying to fix them from upstream + #[arg(long, env = "ALLEGEDLY_DROP")] + drop: bool, } -pub async fn run(Args { fjall, fix }: Args) -> anyhow::Result<()> { +pub async fn run(globals: GlobalArgs, Args { fjall, fix, drop }: Args) -> anyhow::Result<()> { let mut tasks = JoinSet::new(); if let Some(fjall) = fjall { @@ -26,7 +29,7 @@ pub async fn run(Args { fjall, fix }: Args) -> anyhow::Result<()> { 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)); + tasks.spawn(fix_ops_fjall(db, globals.upstream, drop, invalid_ops_rx)); } else { tasks.spawn(audit_fjall(db, invalid_ops_tx)); tasks.spawn(invalid_ops_to_stdout(invalid_ops_rx)); @@ -60,6 +63,8 @@ pub async fn run(Args { fjall, fix }: Args) -> anyhow::Result<()> { #[derive(Debug, Parser)] struct CliArgs { + #[command(flatten)] + globals: GlobalArgs, #[command(flatten)] instrumentation: InstrumentationArgs, #[command(flatten)] @@ -72,6 +77,6 @@ async fn main() -> anyhow::Result<()> { let args = CliArgs::parse(); bin_init(args.instrumentation.enable_opentelemetry); log::info!("{}", logo("audit")); - run(args.args).await?; + run(args.globals, args.args).await?; Ok(()) } diff --git a/src/lib.rs b/src/lib.rs index 8a63cf6..cdb059b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -20,7 +20,9 @@ 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, audit as audit_fjall, backfill_to_fjall, pages_to_fjall, drop_invalid_ops as drop_invalid_ops_fjall}; +pub use plc_fjall::{ + FjallDb, audit as audit_fjall, backfill_to_fjall, fix_ops as fix_ops_fjall, pages_to_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,37 +140,36 @@ pub async fn pages_to_stdout( Ok("pages_to_stdout") } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct InvalidOp { + pub did: String, + pub at: Dt, + pub cid: String, +} + pub async fn invalid_ops_to_stdout( - mut rx: mpsc::Receiver<(String, Dt, String)>, + mut rx: mpsc::Receiver, ) -> 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}"); + while let Some(op) = rx.recv().await { + use std::io::{Write, stdout}; + let mut stdout = stdout().lock(); + serde_json::to_writer(&mut stdout, &op)?; + stdout.write_all(b"\n")?; } Ok("invalid_ops_to_stdout") } pub async fn file_to_invalid_ops( path: impl AsRef, - tx: mpsc::Sender<(String, Dt, String)>, + tx: mpsc::Sender, ) -> 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?; + let op: InvalidOp = serde_json::from_str(&line)?; + tx.send(op).await?; } Ok("invalid_ops_to_stdout") diff --git a/src/plc_fjall.rs b/src/plc_fjall.rs index 38eb718..2bafc0e 100644 --- a/src/plc_fjall.rs +++ b/src/plc_fjall.rs @@ -1,5 +1,5 @@ use crate::{ - BundleSource, Dt, ExportPage, Op as CommonOp, PageBoundaryState, Week, + BundleSource, Dt, ExportPage, InvalidOp, Op as CommonOp, PageBoundaryState, Week, crypto::{AssuranceResults, DidKey, Signature, assure_valid_sig}, }; use anyhow::Context; @@ -1148,10 +1148,7 @@ impl FjallDb { Ok(()) } - pub fn audit( - &self, - invalid_ops_tx: mpsc::Sender<(String, Dt, String)>, - ) -> anyhow::Result<(usize, usize)> { + pub fn audit(&self, invalid_ops_tx: mpsc::Sender) -> anyhow::Result<(usize, usize)> { use std::sync::mpsc; let ops = self.inner.by_did.len()?; @@ -1178,6 +1175,13 @@ impl FjallDb { while let Ok((did_prefix, ops)) = rx.recv() { let did = decode_did(&did_prefix[..did_prefix.len() - 1]); for (ts, cid, op) in &ops { + let send_invalid = || { + let _ = invalid_ops_tx.blocking_send(InvalidOp { + did: did.clone(), + at: ts.clone(), + cid: cid.to_string(), + }); + }; checked += 1; let prev_op = op.operation.prev.as_ref().and_then(|expected| { ops.iter().find(|(_, c, _)| c == expected) @@ -1186,7 +1190,7 @@ impl FjallDb { 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())); + send_invalid(); continue; } let prev_stored = prev_op.map(|(_, _, p)| &p.operation); @@ -1201,13 +1205,13 @@ impl FjallDb { .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())); + send_invalid(); } } 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())); + send_invalid(); } } } @@ -1450,7 +1454,7 @@ pub async fn pages_to_fjall( pub async fn audit( db: FjallDb, - invalid_ops_tx: mpsc::Sender<(String, Dt, String)>, + invalid_ops_tx: mpsc::Sender, ) -> anyhow::Result<&'static str> { log::info!("starting fjall audit..."); let t0 = std::time::Instant::now(); @@ -1465,14 +1469,91 @@ pub async fn audit( Ok("audit_fjall") } -pub async fn drop_invalid_ops( +pub async fn fix_ops( db: FjallDb, - mut invalid_ops_rx: mpsc::Receiver<(String, Dt, String)>, + upstream: reqwest::Url, + only_drop: bool, + mut invalid_ops_rx: mpsc::Receiver, ) -> anyhow::Result<&'static str> { - while let Some((did, at, cid)) = invalid_ops_rx.recv().await { - db.drop_op(&did, &at, &cid)?; + log::info!("starting fjall fix ops..."); + let mut fixed_dids = std::collections::HashSet::new(); + let mut count = 0; + + let latest_at = db + .get_latest()? + .ok_or_else(|| anyhow::anyhow!("db not backfilled? expected at least one op"))?; + + while let Some(op) = invalid_ops_rx.recv().await { + let InvalidOp { did, at, cid, .. } = op; + + if only_drop { + db.drop_op(&did, &at, &cid)?; + db.persist(PersistMode::Buffer)?; + count += 1; + continue; + } + + if fixed_dids.contains(&did) { + continue; + } + + log::trace!("fetching upstream ops to fix did: {did}"); + let mut url = upstream.clone(); + url.set_path(&format!("/{did}/log/audit")); + + let resp = crate::CLIENT.get(url).send().await?; + + use reqwest::StatusCode; + let ops: Vec = match resp.status() { + StatusCode::OK => match resp.json().await { + Ok(ops) => ops, + Err(e) => { + log::warn!("failed to parse upstream ops for {did}: {e}"); + continue; + } + }, + StatusCode::NOT_FOUND => { + log::trace!("did not found upstream: {did}"); + Vec::new() // this essentially means drop the whole did + } + s => { + log::warn!("failed to fetch upstream for {did}: {s}"); + continue; + } + }; + + log::trace!("fetched {} ops for {did}", ops.len()); + + // we drop all ops first just to be safe + let existing = db.ops_for_did(&did)?; + for op in existing { + let op = op?; + db.drop_op(&did, &op.created_at, &op.cid)?; + } + + // then insert the fresh ops + for op in ops { + // skip newer ops, since we will fill them in later anyway + // if we don't skip these we might miss some ops in between + // the latest_at we started with vs the one we ended up with + if op.created_at > latest_at { + log::trace!( + "skipping op {} for {did} because it is newer than latest_at {latest_at}", + op.cid + ); + continue; + } + + count += db.insert_op::(&op)?; + } + + db.persist(PersistMode::Buffer)?; + fixed_dids.insert(did); } - Ok("drop_invalid_ops") + + log::info!("fixed {count} ops"); + + Ok("fix_ops_fjall") } #[cfg(test)] -- 2.51.2