use std::fs::{File, Metadata};
use std::io::{self, BufRead, BufReader, Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};
use chrono::{DateTime, Utc};
use super::{Filter, GhCounts};
use crate::request_log::LogRecord;
const TAIL_LEN: usize = 64;
const BACKSCAN_BLOCK: usize = 8 * 1024;
const BASELINE_SCAN_LIMIT: u64 = 4 * 1024 * 1024;
const READ_BUFFER: usize = 64 * 1024;
type FileId = Option<(u64, u64)>;
#[cfg(unix)]
fn file_id(meta: &Metadata) -> FileId {
use std::os::unix::fs::MetadataExt;
Some((meta.dev(), meta.ino()))
}
#[cfg(not(unix))]
fn file_id(_meta: &Metadata) -> FileId {
None
}
#[derive(Debug)]
struct Cursor {
id: FileId,
offset: u64,
tail: Vec<u8>,
}
impl Cursor {
fn start_of(id: FileId) -> Self {
Self {
id,
offset: 0,
tail: Vec::new(),
}
}
fn is_intact(&self, file: &mut File, id: FileId, len: u64) -> io::Result<bool> {
if self.id != id || len < self.offset {
return Ok(false);
}
let mut seen = vec![0; self.tail.len()];
file.seek(SeekFrom::Start(self.offset - self.tail.len() as u64))?;
match file.read_exact(&mut seen) {
Ok(()) => Ok(seen == self.tail),
Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => Ok(false),
Err(e) => Err(e),
}
}
fn advance(&mut self, line: &[u8]) {
self.offset += line.len() as u64;
if line.len() >= TAIL_LEN {
self.tail.clear();
self.tail.extend_from_slice(&line[line.len() - TAIL_LEN..]);
} else {
self.tail.extend_from_slice(line);
let excess = self.tail.len().saturating_sub(TAIL_LEN);
self.tail.drain(..excess);
}
}
fn baseline(path: &Path) -> Option<Self> {
match Self::try_baseline(path) {
Ok(cursor) => cursor,
Err(e) => {
tracing::warn!(
"github counters: cannot take a baseline of {}: {e}; the first summary \
will scan the whole log",
path.display()
);
None
}
}
}
fn try_baseline(path: &Path) -> io::Result<Option<Self>> {
let mut file = match File::open(path) {
Ok(file) => file,
Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),
Err(e) => return Err(e),
};
let meta = file.metadata()?;
let offset = end_of_last_line(&mut file, meta.len(), BASELINE_SCAN_LIMIT)?;
let tail = read_tail(&mut file, offset)?;
Ok(Some(Self {
id: file_id(&meta),
offset,
tail,
}))
}
}
fn end_of_last_line(file: &mut File, len: u64, limit: u64) -> io::Result<u64> {
let floor = len.saturating_sub(limit);
let mut block = [0u8; BACKSCAN_BLOCK];
let mut end = len;
while end > floor {
let start = end.saturating_sub(BACKSCAN_BLOCK as u64).max(floor);
let want = usize::try_from(end - start).unwrap_or(BACKSCAN_BLOCK);
let chunk = &mut block[..want];
file.seek(SeekFrom::Start(start))?;
file.read_exact(chunk)?;
if let Some(i) = chunk.iter().rposition(|&b| b == b'\n') {
return Ok(start + i as u64 + 1);
}
end = start;
}
Ok(if floor == 0 { 0 } else { len })
}
fn read_tail(file: &mut File, offset: u64) -> io::Result<Vec<u8>> {
let len = usize::try_from(offset).map_or(TAIL_LEN, |o| o.min(TAIL_LEN));
let mut tail = vec![0; len];
file.seek(SeekFrom::Start(offset - len as u64))?;
file.read_exact(&mut tail)?;
Ok(tail)
}
#[derive(Debug)]
pub struct IncrementalCounts {
path: PathBuf,
since: DateTime<Utc>,
filter: Filter,
counts: GhCounts,
cursor: Option<Cursor>,
}
impl IncrementalCounts {
pub fn start(path: PathBuf) -> Self {
let cursor = Cursor::baseline(&path);
Self::new(path, Utc::now(), cursor)
}
#[cfg(test)]
fn start_since(path: PathBuf, since: DateTime<Utc>) -> Self {
let cursor = Cursor::baseline(&path);
Self::new(path, since, cursor)
}
fn new(path: PathBuf, since: DateTime<Utc>, cursor: Option<Cursor>) -> Self {
Self {
path,
since,
filter: Filter {
since: Some(since),
until: None,
source: None,
},
counts: GhCounts::default(),
cursor,
}
}
pub fn since(&self) -> DateTime<Utc> {
self.since
}
pub fn counts(&self) -> &GhCounts {
&self.counts
}
pub fn refresh(&mut self, mut stop: impl FnMut() -> bool) -> bool {
match self.try_refresh(&mut stop) {
Ok(complete) => complete,
Err(e) => {
tracing::debug!(
"github counters: reading {} failed: {e}",
self.path.display()
);
false
}
}
}
fn try_refresh(&mut self, stop: &mut impl FnMut() -> bool) -> io::Result<bool> {
let mut file = match File::open(&self.path) {
Ok(file) => file,
Err(e) if e.kind() == io::ErrorKind::NotFound => {
self.counts = GhCounts::default();
self.cursor = None;
return Ok(true);
}
Err(e) => return Err(e),
};
let meta = file.metadata()?;
let mut cursor = self.resume(&mut file, &meta);
let result = self.read_new_lines(&mut file, &mut cursor, stop);
self.cursor = Some(cursor);
result
}
fn resume(&mut self, file: &mut File, meta: &Metadata) -> Cursor {
let id = file_id(meta);
if let Some(cursor) = self.cursor.take() {
match cursor.is_intact(file, id, meta.len()) {
Ok(true) => return cursor,
Ok(false) => {}
Err(e) => tracing::debug!(
"github counters: cannot verify the position in {}, rescanning: {e}",
self.path.display()
),
}
}
self.counts = GhCounts::default();
Cursor::start_of(id)
}
fn read_new_lines(
&mut self,
file: &mut File,
cursor: &mut Cursor,
stop: &mut impl FnMut() -> bool,
) -> io::Result<bool> {
file.seek(SeekFrom::Start(cursor.offset))?;
let mut reader = BufReader::with_capacity(READ_BUFFER, file);
let mut line = Vec::new();
loop {
if stop() {
return Ok(false);
}
line.clear();
reader.read_until(b'\n', &mut line)?;
if line.last() != Some(&b'\n') {
return Ok(true);
}
if let Ok(rec) = serde_json::from_slice::<LogRecord>(&line) {
self.filter.tally(&mut self.counts, &rec);
}
cursor.advance(&line);
}
}
#[cfg(test)]
fn offset(&self) -> Option<u64> {
self.cursor.as_ref().map(|c| c.offset)
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
use crate::github_metrics::aggregate;
use chrono::{Duration, SecondsFormat};
use std::io::Write;
fn since() -> DateTime<Utc> {
DateTime::parse_from_rfc3339("2026-07-21T10:00:00.000Z")
.unwrap()
.with_timezone(&Utc)
}
fn stamp(secs: i64) -> String {
(since() + Duration::seconds(secs)).to_rfc3339_opts(SecondsFormat::Millis, true)
}
fn gh(secs: i64, command: &[&str], source: &str) -> String {
format!(
"{{\"kind\":\"gh\",\"timestamp\":\"{}\",\"command\":{},\"source\":\"{source}\"}}\n",
stamp(secs),
serde_json::to_string(command).unwrap(),
)
}
struct Log {
_dir: tempfile::TempDir,
path: PathBuf,
}
impl Log {
fn new() -> Self {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("log.jsonl");
Self { _dir: dir, path }
}
fn write(&self, text: &str) {
std::fs::write(&self.path, text).unwrap();
}
fn append(&self, text: &str) {
let mut file = std::fs::OpenOptions::new()
.append(true)
.create(true)
.open(&self.path)
.unwrap();
file.write_all(text.as_bytes()).unwrap();
}
fn replace(&self, text: &str) {
let tmp = self.path.with_extension("tmp");
std::fs::write(&tmp, text).unwrap();
std::fs::rename(&tmp, &self.path).unwrap();
}
fn len(&self) -> u64 {
std::fs::metadata(&self.path).unwrap().len()
}
fn full_scan(&self) -> GhCounts {
aggregate(&self.path, Some(since()), None, None)
}
fn start(&self) -> IncrementalCounts {
IncrementalCounts::start_since(self.path.clone(), since())
}
}
fn never() -> bool {
false
}
#[test]
fn records_before_the_baseline_are_never_read() {
let log = Log::new();
let pre: String = (0..5).map(|n| gh(10 + n, &["pr", "list"], "cli")).collect();
log.write(&pre);
assert_eq!(log.full_scan().total(), 5);
let mut tally = log.start();
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 0);
assert_eq!(tally.offset(), Some(log.len()));
}
#[test]
fn appended_records_match_a_full_scan() {
let log = Log::new();
log.write(&gh(-100, &["pr", "list"], "cli")); let mut tally = log.start();
log.append(&gh(1, &["api", "graphql"], "daemon"));
log.append(&gh(2, &["pr", "list"], "cli"));
log.append(&gh(3, &["--version"], "cli"));
log.append("{\"kind\":\"http\",\"timestamp\":\"2026-07-21T10:00:04.000Z\"}\n");
log.append("not json\n\n");
log.append(&gh(-5, &["pr", "list"], "cli")); log.append("{\"kind\":\"gh\",\"timestamp\":\"bad\",\"command\":[\"pr\"]}\n");
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 3);
assert_eq!(*tally.counts(), log.full_scan());
log.append(&gh(6, &["api", "graphql"], "mcp"));
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 4);
assert_eq!(*tally.counts(), log.full_scan());
assert!(tally.refresh(never));
assert_eq!(*tally.counts(), log.full_scan());
assert_eq!(tally.offset(), Some(log.len()));
}
#[test]
fn a_partial_trailing_line_waits_for_its_newline() {
let log = Log::new();
let mut tally = log.start();
let line = gh(1, &["pr", "list"], "cli");
let (head, rest) = line.split_at(line.len() / 2);
log.append(head);
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 0);
assert_eq!(tally.offset(), Some(0));
log.append(rest);
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 1);
assert_eq!(*tally.counts(), log.full_scan());
let line = gh(2, &["pr", "view"], "cli");
log.append(line.trim_end());
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 1);
log.append("\n");
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 2);
assert_eq!(*tally.counts(), log.full_scan());
}
#[test]
fn a_baseline_inside_a_line_starts_at_that_line() {
let log = Log::new();
let first = gh(-1, &["pr", "list"], "cli");
let second = gh(2, &["pr", "view"], "cli");
let (head, rest) = second.split_at(second.len() / 2);
log.write(&format!("{first}{head}"));
let mut tally = log.start();
assert_eq!(tally.offset(), Some(first.len() as u64));
log.append(rest);
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 1);
assert_eq!(tally.counts().by_subcommand["pr view"], 1);
}
#[test]
fn the_last_line_is_found_across_search_blocks() {
let log = Log::new();
let first = gh(1, &["pr", "list"], "cli");
let long = "x".repeat(3 * BACKSCAN_BLOCK + 17);
log.write(&format!("{first}{long}"));
let mut file = File::open(&log.path).unwrap();
assert_eq!(
end_of_last_line(&mut file, log.len(), u64::MAX).unwrap(),
first.len() as u64
);
assert_eq!(end_of_last_line(&mut file, 0, u64::MAX).unwrap(), 0);
log.write(&long); let mut file = File::open(&log.path).unwrap();
assert_eq!(end_of_last_line(&mut file, log.len(), u64::MAX).unwrap(), 0);
}
#[test]
fn the_search_for_the_last_line_gives_up_at_its_limit() {
let log = Log::new();
let first = gh(1, &["pr", "list"], "cli");
let long = "x".repeat(3 * BACKSCAN_BLOCK + 17);
log.write(&format!("{first}{long}"));
let mut file = File::open(&log.path).unwrap();
let limit = 2 * BACKSCAN_BLOCK as u64;
assert_eq!(
end_of_last_line(&mut file, log.len(), limit).unwrap(),
log.len()
);
let reach = long.len() as u64 + 1;
assert_eq!(
end_of_last_line(&mut file, log.len(), reach).unwrap(),
first.len() as u64
);
}
#[test]
fn a_baseline_inside_a_huge_line_skips_that_fragment_and_keeps_counting() {
let log = Log::new();
let huge = "x".repeat(usize::try_from(BASELINE_SCAN_LIMIT).unwrap() + 1024);
log.write(&format!("{}{huge}", gh(1, &["pr", "list"], "cli")));
let mut tally = log.start();
assert_eq!(tally.offset(), Some(log.len()));
log.append("\n");
log.append(&gh(2, &["pr", "view"], "cli"));
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 1);
assert_eq!(tally.counts().by_subcommand["pr view"], 1);
}
#[test]
fn a_replaced_log_is_rescanned_from_its_first_byte() {
let log = Log::new();
let mut tally = log.start();
log.append(&gh(1, &["pr", "list"], "cli"));
log.append(&gh(2, &["pr", "list"], "cli"));
log.append(&gh(3, &["pr", "list"], "cli"));
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 3);
log.replace(&format!(
"{}{}{}",
gh(4, &["api", "graphql"], "daemon"),
gh(5, &["api", "graphql"], "daemon"),
gh(-9, &["pr", "list"], "cli"),
));
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 2);
assert_eq!(*tally.counts(), log.full_scan());
}
#[test]
fn a_rotated_log_is_followed_to_its_replacement() {
let log = Log::new();
let mut tally = log.start();
log.append(&gh(1, &["pr", "list"], "cli"));
assert!(tally.refresh(never));
std::fs::rename(&log.path, log.path.with_extension("jsonl.1")).unwrap();
log.append(&gh(2, &["api", "graphql"], "daemon"));
assert!(tally.refresh(never));
assert_eq!(*tally.counts(), log.full_scan());
assert_eq!(tally.counts().total(), 1);
std::fs::remove_file(&log.path).unwrap();
assert!(tally.refresh(never));
assert_eq!(*tally.counts(), GhCounts::default());
assert_eq!(tally.offset(), None);
log.append(&gh(3, &["pr", "view"], "cli"));
log.append(&gh(4, &["pr", "view"], "cli"));
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 2);
assert_eq!(*tally.counts(), log.full_scan());
}
#[test]
fn a_log_truncated_below_the_offset_is_rescanned() {
let log = Log::new();
let mut tally = log.start();
for n in 1..=4 {
log.append(&gh(n, &["pr", "list"], "cli"));
}
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 4);
log.write(&gh(9, &["api", "graphql"], "daemon")); assert!(log.len() < tally.offset().unwrap());
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 1);
assert_eq!(*tally.counts(), log.full_scan());
}
#[test]
fn a_log_rewritten_in_place_and_regrown_past_the_offset_is_rescanned() {
let log = Log::new();
let mut tally = log.start();
log.append(&gh(1, &["pr", "list"], "cli"));
log.append(&gh(2, &["pr", "list"], "cli"));
assert!(tally.refresh(never));
let stored = tally.offset().unwrap();
let rewritten: String = (10..16)
.map(|n| gh(n, &["api", "graphql"], "daemon"))
.collect();
log.write(&rewritten);
assert!(log.len() > stored);
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 6);
assert_eq!(*tally.counts(), log.full_scan());
}
#[test]
fn a_log_created_after_the_baseline_is_scanned_from_its_first_byte() {
let log = Log::new();
let mut tally = log.start();
assert_eq!(tally.offset(), None);
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 0);
log.append(&gh(-10, &["pr", "list"], "cli"));
log.append(&gh(1, &["pr", "list"], "cli"));
log.append(&gh(2, &["pr", "view"], "cli"));
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 2);
assert_eq!(*tally.counts(), log.full_scan());
}
#[test]
fn a_stopped_refresh_resumes_where_it_stopped() {
let log = Log::new();
let mut tally = log.start();
for n in 1..=5 {
log.append(&gh(n, &["pr", "list"], "cli"));
}
let mut calls = 0;
let complete = tally.refresh(|| {
calls += 1;
calls > 2
});
assert!(!complete);
assert_eq!(tally.counts().total(), 2);
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 5);
assert_eq!(*tally.counts(), log.full_scan());
}
#[test]
fn a_stopped_rebuild_keeps_what_it_read_and_continues() {
let log = Log::new();
let mut tally = log.start();
log.append(&gh(1, &["pr", "list"], "cli"));
assert!(tally.refresh(never));
let replacement: String = (10..16).map(|n| gh(n, &["pr", "view"], "cli")).collect();
log.replace(&replacement);
let mut calls = 0;
assert!(!tally.refresh(|| {
calls += 1;
calls > 3
}));
assert_eq!(tally.counts().total(), 3);
assert!(tally.refresh(never));
assert_eq!(tally.counts().total(), 6);
assert_eq!(*tally.counts(), log.full_scan());
}
#[test]
fn the_tail_keeps_the_last_bytes_across_short_and_long_lines() {
let mut cursor = Cursor::start_of(None);
cursor.advance(b"ab\n");
cursor.advance(b"cd\n");
assert_eq!(cursor.tail, b"ab\ncd\n");
assert_eq!(cursor.offset, 6);
let long: Vec<u8> = (0..200u8).collect();
cursor.advance(&long);
assert_eq!(cursor.tail, &long[long.len() - TAIL_LEN..]);
cursor.advance(b"z\n");
assert_eq!(cursor.tail.len(), TAIL_LEN);
assert!(cursor.tail.ends_with(b"z\n"));
assert_eq!(cursor.offset, 6 + 200 + 2);
}
}