Something went wrong. Try again.
Server tools to backfill, tail, mirror, and verify PLC logs forked from microcosm.blue/Allegedly
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440use crate::{CLIENT, Dt, ExportPage, Op, OpKey};use reqwest::Url;use std::time::Duration;use thiserror::Error;use tokio::sync::mpsc;
// plc.directory ratelimit on /export is 500 per 5 minsconst UPSTREAM_REQUEST_INTERVAL: Duration = Duration::from_millis(600);
#[derive(Debug, Error)]pub enum GetPageError { #[error(transparent)] Reqwest(#[from] reqwest::Error), #[error(transparent)] ReqwestMiddleware(#[from] reqwest_middleware::Error), #[error(transparent)] Serde(#[from] serde_json::Error),}
/// ops are primary-keyed by (did, cid)/// plc orders by `created_at` but does not guarantee distinct times per op/// we assume that the order will at least be deterministic: this may be unsound#[derive(Debug, PartialEq)]pub struct LastOp { pub created_at: Dt, // any op greater is definitely not duplicated pk: (String, String), // did, cid}
impl From<Op> for LastOp { fn from(op: Op) -> Self { Self { created_at: op.created_at, pk: (op.did, op.cid), } }}
impl From<&Op> for LastOp { fn from(op: &Op) -> Self { Self { created_at: op.created_at, pk: (op.did.clone(), op.cid.clone()), } }}
// bit of a hackimpl From<Dt> for LastOp { fn from(dt: Dt) -> Self { Self { created_at: dt, pk: ("".to_string(), "".to_string()), } }}
/// PLC#[derive(Debug, PartialEq)]pub struct PageBoundaryState { pub last_at: Dt, keys_at: Vec<OpKey>, // expected to ~always be length one}
/// track keys at final createdAt to deduplicate the start of the next pageimpl PageBoundaryState { pub fn new(page: &ExportPage) -> Option<Self> { // grab the very last op let (last_at, last_key) = page.ops.last().map(|op| (op.created_at, op.into()))?;
// set initial state let mut me = Self { last_at, keys_at: vec![last_key], };
// and make sure all keys at this time are captured from the back me.capture_nth_last_at(page, last_at, 1);
Some(me) } fn apply_to_next(&mut self, page: &mut ExportPage) { // walk ops forward, kicking previously-seen ops until created_at advances let to_remove: Vec<usize> = page .ops .iter() .enumerate() .take_while(|(_, op)| op.created_at == self.last_at) .filter(|(_, op)| self.keys_at.contains(&(*op).into())) .map(|(i, _)| i) .collect();
// actually remove them. last to first so indices don't shift for dup_idx in to_remove.into_iter().rev() { page.ops.remove(dup_idx); }
// grab the very last op let Some((last_at, last_key)) = page.ops.last().map(|op| (op.created_at, op.into())) else { // there are no ops left? oop. bail. // last_at and existing keys remain in tact return; };
// reset state (as long as time actually moved forward on this page) if last_at > self.last_at { self.last_at = last_at; self.keys_at = vec![last_key]; } else { // weird cases: either time didn't move (fine...) or went backwards (not fine) assert_eq!(last_at, self.last_at, "time moved backwards on a page"); self.keys_at.push(last_key); } // and make sure all keys at this time are captured from the back self.capture_nth_last_at(page, last_at, 1); }
/// walk backwards from 2nd last and collect keys until created_at changes fn capture_nth_last_at(&mut self, page: &ExportPage, last_at: Dt, skips: usize) { page.ops .iter() .rev() .skip(skips) .take_while(|op| op.created_at == last_at) .for_each(|op| { self.keys_at.push(op.into()); }); }}
pub async fn get_page(url: Url) -> Result<(ExportPage, Option<LastOp>), GetPageError> { log::trace!("Getting page: {url}");
let ops: Vec<Op> = CLIENT .get(url) .send() .await? .error_for_status()? .text() .await? .trim() .split('\n') .filter_map(|s| { serde_json::from_str::<Op>(s) .inspect_err(|e| log::warn!("failed to parse op: {e} ({s})")) .ok() }) .collect();
let last_op = ops.last().map(Into::into);
Ok((ExportPage { ops }, last_op))}
pub async fn poll_upstream( after: Option<Dt>, base: Url, dest: mpsc::Sender<ExportPage>,) -> anyhow::Result<&'static str> { log::info!("starting upstream poller after {after:?}"); let mut tick = tokio::time::interval(UPSTREAM_REQUEST_INTERVAL); let mut prev_last: Option<LastOp> = after.map(Into::into); let mut boundary_state: Option<PageBoundaryState> = None; loop { tick.tick().await;
let mut url = base.clone(); if let Some(ref pl) = prev_last { url.query_pairs_mut() .append_pair("after", &pl.created_at.to_rfc3339()); };
let (mut page, next_last) = get_page(url).await?; if let Some(ref mut state) = boundary_state { state.apply_to_next(&mut page); } else { boundary_state = PageBoundaryState::new(&page); } if !page.is_empty() { match dest.try_send(page) { Ok(()) => {} Err(mpsc::error::TrySendError::Full(page)) => { log::warn!("export: destination channel full, awaiting..."); dest.send(page).await?; } e => e?, }; }
prev_last = next_last.or(prev_last); }}
#[cfg(test)]mod test { use super::*;
const FIVES_TS: i64 = 1431648000; const NEXT_TS: i64 = 1431648001;
fn valid_op() -> Op { serde_json::from_value(serde_json::json!({ "did": "did", "cid": "cid", "createdAt": "2015-05-15T00:00:00Z", "nullified": false, "operation": {}, })) .unwrap() }
fn next_op() -> Op { serde_json::from_value(serde_json::json!({ "did": "didnext", "cid": "cidnext", "createdAt": "2015-05-15T00:00:01Z", "nullified": false, "operation": {}, })) .unwrap() }
fn base_state() -> PageBoundaryState { let page = ExportPage { ops: vec![valid_op()], }; PageBoundaryState::new(&page).expect("to have a base page boundary state") }
#[test] fn test_boundary_new_empty() { let page = ExportPage { ops: vec![] }; let state = PageBoundaryState::new(&page); assert!(state.is_none()); }
#[test] fn test_boundary_new_one_op() { let page = ExportPage { ops: vec![valid_op()], }; let state = PageBoundaryState::new(&page).unwrap(); assert_eq!(state.last_at, Dt::from_timestamp(FIVES_TS, 0).unwrap()); assert_eq!( state.keys_at, vec![OpKey { cid: "cid".to_string(), did: "did".to_string(), }] ); }
#[test] fn test_add_new_empty() { let mut state = base_state(); state.apply_to_next(&mut ExportPage { ops: vec![] }); assert_eq!(state, base_state()); }
#[test] fn test_add_new_same_op() { let mut page = ExportPage { ops: vec![valid_op()], }; let mut state = base_state(); state.apply_to_next(&mut page); assert_eq!(state, base_state()); }
#[test] fn test_add_new_same_time() { // make an op with a different OpKey let mut op = valid_op(); op.cid = "cid2".to_string(); let mut page = ExportPage { ops: vec![op] };
let mut state = base_state(); state.apply_to_next(&mut page); assert_eq!(state.last_at, Dt::from_timestamp(FIVES_TS, 0).unwrap()); assert_eq!( state.keys_at, vec![ OpKey { cid: "cid".to_string(), did: "did".to_string(), }, OpKey { cid: "cid2".to_string(), did: "did".to_string(), }, ] ); }
#[test] fn test_add_new_same_time_dup_before() { // make an op with a different OpKey let mut op = valid_op(); op.cid = "cid2".to_string(); let mut page = ExportPage { ops: vec![valid_op(), op], };
let mut state = base_state(); state.apply_to_next(&mut page); assert_eq!(state.last_at, Dt::from_timestamp(FIVES_TS, 0).unwrap()); assert_eq!( state.keys_at, vec![ OpKey { cid: "cid".to_string(), did: "did".to_string(), }, OpKey { cid: "cid2".to_string(), did: "did".to_string(), }, ] ); }
#[test] fn test_add_new_same_time_dup_after() { // make an op with a different OpKey let mut op = valid_op(); op.cid = "cid2".to_string(); let mut page = ExportPage { ops: vec![op, valid_op()], };
let mut state = base_state(); state.apply_to_next(&mut page); assert_eq!(state.last_at, Dt::from_timestamp(FIVES_TS, 0).unwrap()); assert_eq!( state.keys_at, vec![ OpKey { cid: "cid".to_string(), did: "did".to_string(), }, OpKey { cid: "cid2".to_string(), did: "did".to_string(), }, ] ); }
#[test] fn test_add_new_next_time() { let mut page = ExportPage { ops: vec![next_op()], }; let mut state = base_state(); state.apply_to_next(&mut page); assert_eq!(state.last_at, Dt::from_timestamp(NEXT_TS, 0).unwrap()); assert_eq!( state.keys_at, vec![OpKey { cid: "cidnext".to_string(), did: "didnext".to_string(), },] ); }
#[test] fn test_add_new_next_time_with_dup() { let mut page = ExportPage { ops: vec![valid_op(), next_op()], }; let mut state = base_state(); state.apply_to_next(&mut page); assert_eq!(state.last_at, Dt::from_timestamp(NEXT_TS, 0).unwrap()); assert_eq!( state.keys_at, vec![OpKey { cid: "cidnext".to_string(), did: "didnext".to_string(), },] ); assert_eq!(page.ops.len(), 1); assert_eq!(page.ops[0], next_op()); }
#[test] fn test_add_new_next_time_with_dup_and_new_prev_same_time() { // make an op with a different OpKey let mut op = valid_op(); op.cid = "cid2".to_string();
let mut page = ExportPage { ops: vec![ valid_op(), // should get dropped op.clone(), // should be kept next_op(), ], }; let mut state = base_state(); state.apply_to_next(&mut page); assert_eq!(state.last_at, Dt::from_timestamp(NEXT_TS, 0).unwrap()); assert_eq!( state.keys_at, vec![OpKey { cid: "cidnext".to_string(), did: "didnext".to_string(), },] ); assert_eq!(page.ops.len(), 2); assert_eq!(page.ops[0], op); assert_eq!(page.ops[1], next_op()); }
#[test] fn test_add_new_next_time_with_dup_later_and_new_prev_same_time() { // make an op with a different OpKey let mut op = valid_op(); op.cid = "cid2".to_string();
let mut page = ExportPage { ops: vec![ op.clone(), // should be kept valid_op(), // should get dropped next_op(), ], }; let mut state = base_state(); state.apply_to_next(&mut page); assert_eq!(state.last_at, Dt::from_timestamp(NEXT_TS, 0).unwrap()); assert_eq!( state.keys_at, vec![OpKey { cid: "cidnext".to_string(), did: "didnext".to_string(), },] ); assert_eq!(page.ops.len(), 2); assert_eq!(page.ops[0], op); assert_eq!(page.ops[1], next_op()); }}