the firehose counts the pushes past an offset by the day they landed
Signed-off-by: dawn <90008@klbr.net>
6 unmodified lines789101112131415266 unmodified linesExpand all282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345229 unmodified linesExpand all5755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056 unmodified linesuse cid::Cid;use ipld_core::ipld::Ipld;use jacquard_common::types::tid::{Ticker, Tid};use knot_record::chain::{self, CommitChunk, Emit};use knot_record::git_refs::GIT_REFS_COLLECTION;use knot_types::{RepoDid, UnixMicros};use serde::{Deserialize, Serialize};
use crate::cas::atomic_write;266 unmodified linesExpand all out.extend(serde_ipld_dagcbor::to_vec(&Ipld::Map(body))?); Ok(out)}
/// the frames a reader counts: whole ones from an offset up to the recorded length#[derive(Debug, Default, PartialEq)]pub(crate) struct Pushes { /// the log is shorter than the offset asked for, so these start at its beginning pub(crate) restarted: bool, pub(crate) end: u64, /// the utc day of each frame that moves a repo's refs, `yyyy-mm-dd` pub(crate) days: Vec<String>,}
pub(crate) fn pushes_after(output_root: &Path, offset: u64) -> anyhow::Result<Pushes> { let dir = output_root.join(DIR); let Some(state) = std::fs::read(dir.join(STATE)) .ok() .map(|bytes| serde_json::from_slice::<State>(&bytes)) .transpose() .context("read the firehose state")? else { return Ok(Pushes::default()); }; let restarted = offset > state.len; let start = if restarted { 0 } else { offset }; let mut bytes = Vec::new(); if state.len > start { let mut frames = std::fs::File::open(dir.join(FRAMES)).context("open the firehose frames")?; frames.seek(SeekFrom::Start(start))?; frames .take(state.len - start) .read_to_end(&mut bytes) .context("read the firehose frames")?; } let mut days = Vec::new(); let mut at = 0; while let Some((header, payload)) = ws_header(&bytes[at..]) { let end = at + header + payload as usize; if end > bytes.len() { break; } let (_, body) = split(&bytes[at + header..end])?; let moves_refs = matches!(body.get("ops"), Some(Ipld::List(ops)) if ops.iter().any(|op| { matches!(op, Ipld::Map(op) if matches!(op.get("path"), Some(Ipld::String(path)) if path.split('/').next() == Some(GIT_REFS_COLLECTION))) })); if let (true, Some(Ipld::String(time))) = (moves_refs, body.get("time")) { days.extend(time.get(..10).map(str::to_string)); } at = end; } Ok(Pushes { restarted, end: start + at as u64, days, })}
fn ws_frame(payload: &[u8]) -> Vec<u8> { let mut out = Vec::with_capacity(payload.len() + 10); out.push(0x82); match payload.len() {229 unmodified linesExpand all let frames = read_frames(root); assert_eq!(frames[0].1["commit"], Ipld::Link(second.commit)); assert_eq!(frames[0].1["since"], Ipld::String(first.rev.clone())); }
#[test] fn pushes_are_counted_from_an_offset_by_the_day_their_refs_moved() { let (fixture, _, second, _) = two_commits(); let root = fixture.chain_root.parent().unwrap(); assert_eq!(pushes_after(root, 0).unwrap(), Pushes::default()); assert_eq!(append(root, served(&fixture, &second)).unwrap(), 1); let Ipld::String(time) = &read_frames(root)[0].1["time"] else { panic!("time") }; let len = std::fs::metadata(frames_path(root)).unwrap().len();
let all = pushes_after(root, 0).unwrap(); assert_eq!( all, Pushes { restarted: false, end: len, days: vec![time[..10].to_string()], } ); assert_eq!(pushes_after(root, len).unwrap().days, Vec::<String>::new()); let past = pushes_after(root, len + 1).unwrap(); assert!(past.restarted); assert_eq!(past.days.len(), 1); }}