use crate::error::{AppError, AppResult};
use crate::{ListItem, LogEvent, PromotionItem, Resolution, format_timestamp, normalized};
use serde::Serialize;
use serde_json::{Value, json};
use std::collections::{BTreeMap, HashMap};
use std::fs::{self, File, OpenOptions, Permissions};
use std::io::{ErrorKind, Read, Seek, SeekFrom, Write};
#[cfg(unix)]
use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
use std::path::{Component, Path, PathBuf};
use std::thread;
use std::time::Duration;
const LOCK_ATTEMPTS: usize = 50;
const LOCK_DELAY: Duration = Duration::from_millis(100);
#[derive(Debug, Clone)]
pub struct ResolvedFile {
pub path: PathBuf,
pub cwd: PathBuf,
pub explicit: bool,
pub repo: Option<PathBuf>,
pub warnings: Vec<String>,
}
impl ResolvedFile {
pub fn cwd_repo(&self) -> Option<&Path> {
self.repo
.as_deref()
.filter(|root| self.path.starts_with(root))
}
}
#[derive(Debug, Default)]
pub struct FoldResult {
pub items: Vec<ListItem>,
pub promotions: Vec<PromotionItem>,
pub warnings: Vec<String>,
records: BTreeMap<String, LogEvent>,
winning_amends: HashMap<String, LogEvent>,
lines: Vec<FoldedLine>,
}
#[derive(Debug, Clone)]
pub struct FoldedLine {
pub line: usize,
pub id: String,
pub ts: jiff::Timestamp,
}
pub struct LoadedFold {
pub items: Vec<ListItem>,
pub promotions: Vec<PromotionItem>,
pub warnings: Vec<String>,
}
impl FoldResult {
pub fn record(&self, id: &str) -> Option<&LogEvent> {
self.records.get(id)
}
pub fn lines(&self) -> &[FoldedLine] {
&self.lines
}
pub(crate) fn materialized_appended_resolution(&self, event: &LogEvent) -> Resolution {
let LogEvent::Resolve { id, amend, .. } = event else {
unreachable!("only resolve events materialize resolutions")
};
let effective = match self.winning_amends.get(id) {
Some(stored) if !*amend => stored,
Some(stored) if later_resolve(stored, event) => stored,
_ => event,
};
resolution_from_event(effective)
}
}
#[derive(Default)]
struct WarningCounts {
torn: usize,
malformed: usize,
unknown: usize,
duplicate_cuts: usize,
duplicate_dogears: usize,
duplicate_promotions: usize,
duplicate_resolves: usize,
orphans: usize,
invalid_resolutions: usize,
}
pub(crate) struct ScannedLine<'a> {
pub line: usize,
pub raw: &'a [u8],
pub event: Result<LogEvent, ScanIssue>,
}
pub(crate) enum ScanIssue {
Malformed(String),
Unknown(Option<String>),
Torn,
}
pub const RECORD_VERSION: u64 = 2;
const PROBE_KINDS: [&str; 4] = ["cut", "dogear", "resolve", "promotion"];
#[derive(Debug, Clone)]
pub struct VersionProbe {
pub line: usize,
pub found_version: Option<Value>,
}
pub fn probe_version(bytes: &[u8]) -> Option<VersionProbe> {
for (line, raw) in physical_lines(bytes).1 {
let Ok(value) = serde_json::from_slice::<Value>(raw) else {
continue;
};
let known = value
.get("kind")
.and_then(Value::as_str)
.is_some_and(|kind| PROBE_KINDS.contains(&kind));
if !known {
continue;
}
match value.get("v") {
Some(found) if found.as_u64() == Some(RECORD_VERSION) => {}
found => {
return Some(VersionProbe {
line,
found_version: found.cloned(),
});
}
}
}
None
}
pub fn check_version(bytes: &[u8], path: &Path) -> AppResult<()> {
match probe_version(bytes) {
None => Ok(()),
Some(probe) => Err(AppError::unsupported_log_version(
path,
probe.line,
probe.found_version.as_ref(),
)),
}
}
#[derive(Serialize)]
struct Stored<'a> {
v: u64,
#[serde(flatten)]
event: &'a LogEvent,
}
impl<'a> Stored<'a> {
fn new(event: &'a LogEvent) -> Self {
Self {
v: RECORD_VERSION,
event,
}
}
}
pub fn discover(flag: Option<PathBuf>) -> AppResult<ResolvedFile> {
let cwd = std::env::current_dir().map_err(|error| AppError::from_io(error, Path::new(".")))?;
discover_from(&cwd, flag)
}
pub fn discover_from(cwd: &Path, flag: Option<PathBuf>) -> AppResult<ResolvedFile> {
let repo = find_repo_root(cwd);
if let Some(path) = flag {
return Ok(resolved_file(cwd, absolute(cwd, path), true, repo));
}
if let Some(path) = std::env::var_os("BLOTTER_FILE")
&& !path.is_empty()
{
return Ok(resolved_file(
cwd,
absolute(cwd, PathBuf::from(path)),
true,
repo,
));
}
if let Some(root) = repo.clone() {
let path = default_log_path(&root);
return Ok(resolved_file(cwd, path, false, Some(root)));
}
let home = home_dir(cwd).ok_or_else(|| {
AppError::config(
"cannot resolve the home directory for the default blotter file",
"Set HOME or pass --file PATH.",
)
})?;
Ok(resolved_file(
cwd,
home.join(".blotter/log.jsonl"),
false,
None,
))
}
fn resolved_file(cwd: &Path, path: PathBuf, explicit: bool, repo: Option<PathBuf>) -> ResolvedFile {
ResolvedFile {
warnings: Vec::new(),
path,
cwd: cwd.to_path_buf(),
explicit,
repo,
}
}
pub fn default_log_path(root: &Path) -> PathBuf {
root.join(".blotter.jsonl")
}
pub fn find_repo_root(start: &Path) -> Option<PathBuf> {
start
.ancestors()
.find(|candidate| candidate.join(".git").exists())
.map(Path::to_path_buf)
}
pub fn home_dir(cwd: &Path) -> Option<PathBuf> {
std::env::var_os("HOME")
.filter(|value| !value.is_empty())
.map(PathBuf::from)
.map(|home| absolute(cwd, home))
}
pub fn record_cwd(cwd: &Path, repo: Option<&Path>, home: Option<&Path>) -> String {
if let Some(relative) = repo.and_then(|root| cwd.strip_prefix(root).ok()) {
return match relative.as_os_str().is_empty() {
true => ".".into(),
false => relative.to_string_lossy().into_owned(),
};
}
crate::redact::rewrite_home_paths(&cwd.to_string_lossy(), home)
}
fn absolute(cwd: &Path, path: PathBuf) -> PathBuf {
let joined = if path.is_absolute() {
path
} else {
cwd.join(path)
};
let components: Vec<Component> = joined.components().collect();
if !components
.iter()
.any(|component| matches!(component, Component::ParentDir))
{
return fold_lexically(PathBuf::new(), &components);
}
let trailing = match components.last() {
Some(Component::Normal(_)) => components.len() - 1,
_ => components.len(),
};
let mut resolved = resolve_existing_prefix(&components[..trailing]);
if let Some(Component::Normal(name)) = components.get(trailing) {
resolved.push(name);
}
resolved
}
fn resolve_existing_prefix(components: &[Component]) -> PathBuf {
for split in (1..=components.len()).rev() {
let mut candidate = PathBuf::new();
for component in &components[..split] {
candidate.push(component.as_os_str());
}
if let Ok(canonical) = fs::canonicalize(&candidate) {
return fold_lexically(canonical, &components[split..]);
}
}
fold_lexically(PathBuf::new(), components)
}
fn fold_lexically(mut base: PathBuf, components: &[Component]) -> PathBuf {
for component in components {
match component {
Component::CurDir => {}
Component::ParentDir => {
base.pop();
}
other => base.push(other.as_os_str()),
}
}
base
}
pub fn with_shared<T>(path: &Path, action: impl FnOnce(&mut File) -> AppResult<T>) -> AppResult<T> {
let mut file = open_locked(path, false, || {
#[cfg(unix)]
let opened = OpenOptions::new()
.read(true)
.custom_flags(libc::O_NONBLOCK)
.open(path);
#[cfg(not(unix))]
let opened = File::open(path);
opened.map_err(|error| AppError::from_log_open(error, path))
})?;
let result = action(&mut file);
let unlock = file
.unlock()
.map_err(|error| AppError::from_io(error, path));
match (result, unlock) {
(Err(error), _) | (Ok(_), Err(error)) => Err(error),
(Ok(value), Ok(())) => Ok(value),
}
}
pub fn read_or_empty<T>(
path: &Path,
explicit: bool,
warnings: &mut Vec<String>,
warning: &str,
suggested_fix: &str,
empty: impl FnOnce() -> T,
read: impl FnOnce(&mut File) -> AppResult<T>,
) -> AppResult<(T, bool)> {
match with_shared(path, read) {
Ok(value) => Ok((value, true)),
Err(error) if error.code == "not_found" && error.exit_code == 66 && !explicit => {
warnings.push(warning.into());
Ok((empty(), false))
}
Err(error) if error.code == "not_found" && error.exit_code == 66 => {
Err(AppError::not_found(
format!("blotter file not found: {}", path.display()),
suggested_fix,
))
}
Err(error) => Err(error),
}
}
pub fn load_folded(resolved: &ResolvedFile) -> AppResult<LoadedFold> {
let mut warnings = resolved.warnings.clone();
let (folded, _) = read_or_empty(
&resolved.path,
resolved.explicit,
&mut warnings,
"no blotter file yet; blotter add creates it",
"Pass an existing --file PATH or run `blotter add` to create a discovered default file.",
FoldResult::default,
|log| {
let bytes = read_bytes(log, &resolved.path)?;
check_version(&bytes, &resolved.path)?;
Ok(fold_bytes(&bytes))
},
)?;
warnings.extend(folded.warnings);
Ok(LoadedFold {
items: folded.items,
promotions: folded.promotions,
warnings,
})
}
pub fn with_exclusive<T>(
path: &Path,
create: bool,
action: impl FnOnce(&mut File) -> AppResult<T>,
) -> AppResult<T> {
if create && let Some(parent) = path.parent() {
std::fs::create_dir_all(parent).map_err(|error| AppError::from_io(error, parent))?;
}
let mut file = open_locked(path, true, || {
let mut options = OpenOptions::new();
options.read(true).append(true).create(create);
#[cfg(unix)]
options.custom_flags(libc::O_NONBLOCK);
options
.open(path)
.map_err(|error| AppError::from_log_open(error, path))
})?;
let result = action(&mut file);
let unlock = file
.unlock()
.map_err(|error| AppError::from_io(error, path));
match (result, unlock) {
(Err(error), _) | (Ok(_), Err(error)) => Err(error),
(Ok(value), Ok(())) => Ok(value),
}
}
fn open_locked(
path: &Path,
exclusive: bool,
mut open: impl FnMut() -> AppResult<File>,
) -> AppResult<File> {
let mut file = Some(regular_file(open()?, path)?);
let mut missing: Option<AppError> = None;
for attempt in 0..LOCK_ATTEMPTS {
if file.is_none() {
match open() {
Ok(opened) => file = Some(regular_file(opened, path)?),
Err(error) if error.code == "not_found" => {
missing = Some(error);
delay_before_retry(attempt);
continue;
}
Err(error) => return Err(error),
}
}
let result = if exclusive {
file.as_ref().expect("file is open").try_lock()
} else {
file.as_ref().expect("file is open").try_lock_shared()
};
match result {
Ok(()) => {
if path_identity_matches(file.as_ref().expect("file is open"), path)? {
return Ok(file.take().expect("file is open"));
}
let stale = file.take().expect("file is open");
let _ = stale.unlock();
missing = None;
delay_before_retry(attempt);
}
Err(error) => {
let error: std::io::Error = error.into();
if error.kind() != std::io::ErrorKind::WouldBlock {
return Err(AppError::from_io(error, path));
}
missing = None;
delay_before_retry(attempt);
}
}
}
Err(missing.unwrap_or_else(|| AppError::lock_timeout(path)))
}
fn delay_before_retry(attempt: usize) {
if attempt + 1 < LOCK_ATTEMPTS {
thread::sleep(LOCK_DELAY);
}
}
fn regular_file(file: File, path: &Path) -> AppResult<File> {
let metadata = file
.metadata()
.map_err(|error| AppError::from_io(error, path))?;
if !metadata.is_file() {
return Err(AppError::invalid_input(
format!("blotter file is not a regular file: {}", path.display()),
"Point --file PATH or BLOTTER_FILE at a regular JSONL file; FIFOs and devices are not accepted.",
));
}
Ok(file)
}
#[cfg(unix)]
fn path_identity_matches(file: &File, path: &Path) -> AppResult<bool> {
let locked = file
.metadata()
.map_err(|error| AppError::from_io(error, path))?;
match std::fs::metadata(path) {
Ok(current) => Ok(locked.dev() == current.dev() && locked.ino() == current.ino()),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
Err(error) => Err(AppError::from_io(error, path)),
}
}
#[cfg(not(unix))]
fn path_identity_matches(_file: &File, _path: &Path) -> AppResult<bool> {
Ok(true)
}
pub fn read_bytes(file: &mut File, path: &Path) -> AppResult<Vec<u8>> {
file.seek(SeekFrom::Start(0))
.and_then(|_| {
let mut bytes = Vec::new();
file.read_to_end(&mut bytes).map(|_| bytes)
})
.map_err(|error| AppError::from_io(error, path))
}
pub fn write_new_file(path: &Path, bytes: &[u8], permissions: &Permissions) -> AppResult<PathBuf> {
let mut file = create_new_file(path, permissions, false)
.map_err(|error| AppError::from_io(error, path))?;
if let Err(error) = file.write_all(bytes) {
discard_new_file(file, path);
return Err(AppError::from_io(error, path));
}
if let Err(error) = file.sync_all() {
discard_new_file(file, path);
return Err(AppError::from_io(error, path));
}
Ok(path.to_path_buf())
}
pub fn append_file(path: &Path, bytes: &[u8], permissions: &Permissions) -> AppResult<PathBuf> {
let (mut file, created) = match create_new_file(path, permissions, false) {
Ok(file) => (file, true),
Err(error) if error.kind() == ErrorKind::AlreadyExists => (
OpenOptions::new()
.append(true)
.open(path)
.map_err(|error| AppError::from_io(error, path))?,
false,
),
Err(error) => return Err(AppError::from_io(error, path)),
};
if let Err(error) = file.write_all(bytes) {
if created {
discard_new_file(file, path);
}
return Err(AppError::from_io(error, path));
}
if let Err(error) = file.sync_all() {
if created {
discard_new_file(file, path);
}
return Err(AppError::from_io(error, path));
}
Ok(path.to_path_buf())
}
pub fn replace_log(
path: &Path,
bytes: &[u8],
permissions: &Permissions,
temporary_suffix: &str,
) -> AppResult<()> {
let temporary = suffixed_path(path, temporary_suffix);
let mut file = create_new_file(&temporary, permissions, true)
.map_err(|error| AppError::from_io(error, &temporary))?;
if let Err(error) = file.write_all(bytes) {
discard_new_file(file, &temporary);
return Err(AppError::from_io(error, &temporary));
}
if let Err(error) = file.sync_all() {
discard_new_file(file, &temporary);
return Err(AppError::from_io(error, &temporary));
}
drop(file);
if let Err(error) = fs::rename(&temporary, path) {
let _ = fs::remove_file(&temporary);
return Err(AppError::from_io(error, path));
}
if let Some(parent) = path.parent()
&& let Ok(directory) = File::open(parent)
{
let _ = directory.sync_all();
}
Ok(())
}
pub fn resolve_symlinked_log(path: &Path) -> AppResult<PathBuf> {
let mut current = path.to_path_buf();
for _ in 0..40 {
let metadata =
fs::symlink_metadata(¤t).map_err(|error| AppError::from_io(error, ¤t))?;
if !metadata.file_type().is_symlink() {
return Ok(current);
}
let target = fs::read_link(¤t).map_err(|error| AppError::from_io(error, ¤t))?;
current = if target.is_absolute() {
target
} else {
match current.parent() {
Some(parent) => parent.join(&target),
None => target,
}
};
}
Err(AppError::from_io(
std::io::Error::other("too many levels of symbolic links"),
path,
))
}
pub fn suffixed_path(path: &Path, suffix: &str) -> PathBuf {
let mut value = path.as_os_str().to_os_string();
value.push(suffix);
PathBuf::from(value)
}
pub fn backup_timestamp(now: jiff::Timestamp) -> String {
format_timestamp(now)
.chars()
.filter(|character| !matches!(character, '-' | ':' | '.'))
.collect()
}
pub fn restore_hint(backup: &Path, path: &Path) -> String {
format!("cp {} {}", shell_quote(backup), shell_quote(path))
}
fn create_new_file(
path: &Path,
permissions: &Permissions,
set_permissions_on_non_unix: bool,
) -> std::io::Result<File> {
let mut options = OpenOptions::new();
options.write(true).create_new(true);
#[cfg(unix)]
options.mode(permissions.mode());
let file = options.open(path)?;
#[cfg(unix)]
let permissions_result = {
let _ = set_permissions_on_non_unix;
file.set_permissions(permissions.clone())
};
#[cfg(not(unix))]
let permissions_result = set_permissions_on_non_unix
.then(|| file.set_permissions(permissions.clone()))
.transpose()
.map(|_| ());
if let Err(error) = permissions_result {
drop(file);
let _ = fs::remove_file(path);
return Err(error);
}
Ok(file)
}
fn discard_new_file(file: File, path: &Path) {
drop(file);
let _ = fs::remove_file(path);
}
fn shell_quote(path: &Path) -> String {
format!("'{}'", path.to_string_lossy().replace('\'', "'\\''"))
}
pub fn append_json(file: &mut File, path: &Path, prior: &[u8], record: &LogEvent) -> AppResult<()> {
let mut record_bytes = Vec::new();
serde_json::to_writer(&mut record_bytes, &Stored::new(record))
.map_err(|error| AppError::internal(error.to_string()))?;
record_bytes.push(b'\n');
append_bytes(file, path, prior, &record_bytes)
}
pub fn append_unique(path: &Path, record: LogEvent, dry_run: bool) -> AppResult<(bool, LogEvent)> {
if dry_run {
return Ok((false, record));
}
let id = record.id().expect("new records have IDs").to_owned();
let kind = match &record {
LogEvent::Cut { .. } => "cut",
LogEvent::Dogear { .. } => "dogear",
_ => unreachable!("append_unique only receives cut or dogear records"),
};
with_exclusive(path, true, |log| {
let bytes = read_bytes(log, path)?;
check_version(&bytes, path)?;
let records = fold_records(&bytes);
if let Some(existing) = records.get(&id) {
return if std::mem::discriminant(&record) == std::mem::discriminant(existing) {
Ok((false, existing.clone()))
} else {
Err(AppError::internal(format!(
"{kind} ID collides with an existing non-{kind} record"
)))
};
}
append_json(log, path, &bytes, &record)?;
Ok((true, record))
})
}
pub fn append_json_batch(
file: &mut File,
path: &Path,
prior: &[u8],
records: &[LogEvent],
) -> AppResult<()> {
let mut record_bytes = Vec::new();
for record in records {
serde_json::to_writer(&mut record_bytes, &Stored::new(record))
.map_err(|error| AppError::internal(error.to_string()))?;
record_bytes.push(b'\n');
}
append_bytes(file, path, prior, &record_bytes)
}
fn append_bytes(file: &mut File, path: &Path, prior: &[u8], record_bytes: &[u8]) -> AppResult<()> {
append_bytes_with(file, path, prior, record_bytes, |file, bytes| {
file.write_all(bytes)
})
}
fn append_bytes_with(
file: &mut File,
path: &Path,
prior: &[u8],
record_bytes: &[u8],
write: impl FnOnce(&mut File, &[u8]) -> std::io::Result<()>,
) -> AppResult<()> {
let original_len = file
.metadata()
.map_err(|error| AppError::from_io(error, path))?
.len();
let mut bytes = Vec::new();
if !is_empty_log(prior) && !prior.ends_with(b"\n") {
bytes.push(b'\n');
}
bytes.extend_from_slice(record_bytes);
if let Err(error) = write(file, &bytes) {
if let Err(rollback) = file.set_len(original_len) {
return Err(AppError {
code: "io_error",
message: format!(
"append failed: {error}; rollback to original length {original_len} failed: {rollback}"
),
details: json!({}),
retryable: false,
suggested_fix: "Check the blotter file and filesystem, then retry.".into(),
exit_code: 74,
});
}
return Err(AppError::from_io(error, path));
}
Ok(())
}
pub(crate) fn is_empty_log(bytes: &[u8]) -> bool {
bytes.is_empty() || bytes == b"\n"
}
fn physical_lines(bytes: &[u8]) -> (bool, impl Iterator<Item = (usize, &[u8])> + '_) {
let terminated = bytes.ends_with(b"\n");
let body = if terminated {
&bytes[..bytes.len() - 1]
} else {
bytes
};
let lines = body
.split(|byte| *byte == b'\n')
.enumerate()
.filter(|(index, raw)| !(raw.is_empty() && *index == 0))
.map(|(index, raw)| (index + 1, raw));
(terminated, lines)
}
pub(crate) fn scan(bytes: &[u8]) -> impl Iterator<Item = ScannedLine<'_>> + '_ {
let (terminated, lines) = physical_lines(bytes);
let last_line = physical_lines(bytes).1.map(|(line, _)| line).last();
lines.map(move |(line, raw)| {
let final_line = Some(line) == last_line;
let decoded = serde_json::from_slice::<Value>(raw);
let known = decoded.as_ref().ok().and_then(known_kind);
let event = if final_line && !terminated && known.is_none() {
Err(ScanIssue::Torn)
} else {
match decoded {
Ok(value) => parse_event(value, known),
Err(_) => Err(ScanIssue::Malformed("line is not valid JSON".into())),
}
};
ScannedLine { line, raw, event }
})
}
fn known_kind(value: &Value) -> Option<&'static str> {
match value.get("kind").and_then(Value::as_str) {
Some("cut") => Some("cut"),
Some("dogear") => Some("dogear"),
Some("resolve") => Some("resolve"),
Some("promotion") => Some("promotion"),
_ => None,
}
}
fn parse_event(value: Value, known: Option<&'static str>) -> Result<LogEvent, ScanIssue> {
let unknown = value.get("kind").and_then(Value::as_str).map(str::to_owned);
match serde_json::from_value::<LogEvent>(value) {
Ok(LogEvent::Unknown) => Err(ScanIssue::Unknown(unknown)),
Ok(event) => {
let ts = match &event {
LogEvent::Cut { ts, .. }
| LogEvent::Dogear { ts, .. }
| LogEvent::Resolve { ts, .. }
| LogEvent::Promotion { ts, .. } => ts,
LogEvent::Unknown => unreachable!("unknown events are classified above"),
};
match ts.parse::<jiff::Timestamp>() {
Ok(_) => Ok(event),
Err(_) => Err(ScanIssue::Malformed(format!(
"{} ts is not a full RFC3339 timestamp",
known.expect("parsed events have a known kind")
))),
}
}
Err(error) => match known {
Some(kind) => Err(ScanIssue::Malformed(format!(
"invalid {kind} record: {error}"
))),
None => Err(ScanIssue::Unknown(unknown)),
},
}
}
fn later_resolve(stored: &LogEvent, candidate: &LogEvent) -> bool {
let timestamp = |event: &LogEvent| match event {
LogEvent::Resolve { ts, .. } => ts.parse::<jiff::Timestamp>().ok(),
_ => None,
};
match (timestamp(stored), timestamp(candidate)) {
(Some(stored), Some(candidate)) => stored > candidate,
_ => false,
}
}
fn resolution_from_event(event: &LogEvent) -> Resolution {
let LogEvent::Resolve {
ts,
agent,
note,
task,
pr,
commit,
url,
dropped,
amend,
disposition,
disposition_ts,
promotion,
..
} = event
else {
unreachable!("only resolve events materialize resolutions")
};
Resolution {
ts: ts.clone(),
agent: agent.clone(),
note: note.clone(),
task: task.clone(),
pr: pr.clone(),
commit: commit.clone(),
url: url.clone(),
dropped: *dropped,
amended: *amend,
disposition: *disposition,
disposition_ts: disposition_ts.clone(),
promotion: promotion.clone(),
}
}
pub type PromotionSources = HashMap<String, Vec<String>>;
pub(crate) fn broken_resolution_rules(
event: &LogEvent,
record_kind: &str,
promotions: &PromotionSources,
) -> Vec<&'static str> {
let LogEvent::Resolve {
id,
disposition,
disposition_ts,
promotion,
..
} = event
else {
unreachable!("only resolve events are validated")
};
let mut broken = Vec::new();
if record_kind == "cut" && disposition.is_none() {
broken.push("resolve targets a cut without a disposition");
}
if record_kind == "dogear" && disposition.is_some() {
broken.push("resolve targets a dogear with a disposition");
}
if disposition.is_some() != disposition_ts.is_some() {
broken.push("disposition and disposition_ts must be present together");
}
if let Some(promotion) = promotion {
if *disposition != Some(crate::Disposition::Promoted) {
broken.push("a promotion link requires disposition promoted");
}
match promotions.get(promotion) {
None => broken.push("promotion link names no promotion in this log"),
Some(sources) if !sources.contains(id) => {
broken.push("promotion does not name this record as a source");
}
Some(_) => {}
}
}
broken
}
fn fold_records(bytes: &[u8]) -> BTreeMap<String, LogEvent> {
let mut records = BTreeMap::<String, LogEvent>::new();
for scanned in scan(bytes) {
let Ok(mut event) = scanned.event else {
continue;
};
match &mut event {
LogEvent::Cut { tags, .. } | LogEvent::Dogear { tags, .. } => {
tags.sort();
tags.dedup();
}
LogEvent::Promotion { sources, .. } => *sources = normalized(sources),
LogEvent::Resolve { .. } | LogEvent::Unknown => continue,
}
let id = event.id().expect("parsed records have IDs").to_owned();
records.entry(id).or_insert(event);
}
records
}
pub fn fold_bytes(bytes: &[u8]) -> FoldResult {
fold_bytes_inner(bytes, false)
}
pub fn fold_bytes_with_lines(bytes: &[u8]) -> FoldResult {
fold_bytes_inner(bytes, true)
}
fn fold_bytes_inner(bytes: &[u8], collect_lines: bool) -> FoldResult {
let mut lines = Vec::new();
let mut records = BTreeMap::<String, LogEvent>::new();
let mut resolves = HashMap::<String, LogEvent>::new();
let mut amends = HashMap::<String, (jiff::Timestamp, LogEvent)>::new();
let mut resolve_events = Vec::<LogEvent>::new();
let mut counts = WarningCounts::default();
for scanned in scan(bytes) {
let line = scanned.line;
match scanned.event {
Err(ScanIssue::Malformed(_)) => counts.malformed += 1,
Err(ScanIssue::Unknown(_)) => counts.unknown += 1,
Err(ScanIssue::Torn) => counts.torn += 1,
Ok(mut event) => {
if collect_lines
&& let Some(id) = event.id()
&& let Some(ts) = event_timestamp(&event)
{
lines.push(FoldedLine {
line,
id: id.to_owned(),
ts,
});
}
match &mut event {
LogEvent::Cut { tags, .. } => {
tags.sort();
tags.dedup();
let id = event.id().expect("parsed cuts have IDs").to_owned();
if let std::collections::btree_map::Entry::Vacant(entry) = records.entry(id)
{
entry.insert(event);
} else {
counts.duplicate_cuts += 1;
}
}
LogEvent::Dogear { tags, .. } => {
tags.sort();
tags.dedup();
let id = event.id().expect("parsed dogears have IDs").to_owned();
if let std::collections::btree_map::Entry::Vacant(entry) = records.entry(id)
{
entry.insert(event);
} else {
counts.duplicate_dogears += 1;
}
}
LogEvent::Promotion { sources, .. } => {
*sources = normalized(sources);
let id = event.id().expect("parsed promotions have IDs").to_owned();
if let std::collections::btree_map::Entry::Vacant(entry) = records.entry(id)
{
entry.insert(event);
} else {
counts.duplicate_promotions += 1;
}
}
LogEvent::Resolve { .. } => resolve_events.push(event),
LogEvent::Unknown => counts.unknown += 1,
}
}
}
}
let promotion_sources = promotion_sources(&records);
for event in resolve_events {
let LogEvent::Resolve { id, ts, amend, .. } = &event else {
unreachable!("only resolve events are held back")
};
let id = id.clone();
let amend = *amend;
if let Some(kind) = records.get(&id).and_then(record_kind)
&& !broken_resolution_rules(&event, kind, &promotion_sources).is_empty()
{
counts.invalid_resolutions += 1;
continue;
}
if amend {
let timestamp = ts
.parse::<jiff::Timestamp>()
.expect("parsed resolves have valid RFC3339 timestamps");
match amends.entry(id) {
std::collections::hash_map::Entry::Occupied(mut entry) => {
if timestamp >= entry.get().0 {
entry.insert((timestamp, event));
}
}
std::collections::hash_map::Entry::Vacant(entry) => {
entry.insert((timestamp, event));
}
}
} else if let std::collections::hash_map::Entry::Vacant(entry) = resolves.entry(id) {
entry.insert(event);
} else {
counts.duplicate_resolves += 1;
}
}
let mut winning_amends = HashMap::new();
for (id, (_, amend)) in amends {
winning_amends.insert(id.clone(), amend.clone());
match resolves.entry(id) {
std::collections::hash_map::Entry::Occupied(mut entry) => {
entry.insert(amend);
}
std::collections::hash_map::Entry::Vacant(_) => counts.orphans += 1,
}
}
for id in resolves.keys() {
if !records.contains_key(id) {
counts.orphans += 1;
}
}
let mut items: Vec<_> = records
.values()
.filter(|record| !matches!(record, LogEvent::Promotion { .. }))
.cloned()
.map(|record| {
let resolution = record
.id()
.and_then(|id| resolves.get(id))
.map(resolution_from_event);
let item = ListItem::from_record(record, resolution);
let timestamp = item
.ts
.parse::<jiff::Timestamp>()
.expect("folded items have valid RFC3339 timestamps");
(item, timestamp)
})
.collect();
items.sort_by(|(left, left_timestamp), (right, right_timestamp)| {
match (left.kind.as_str(), right.kind.as_str()) {
("cut", "cut") => right
.impact
.expect("cut has impact")
.rank()
.cmp(&left.impact.expect("cut has impact").rank())
.then_with(|| right_timestamp.cmp(left_timestamp))
.then_with(|| left.id.cmp(&right.id)),
("dogear", "dogear") => right_timestamp
.cmp(left_timestamp)
.then_with(|| left.id.cmp(&right.id)),
("cut", "dogear") => std::cmp::Ordering::Less,
("dogear", "cut") => std::cmp::Ordering::Greater,
_ => left.kind.cmp(&right.kind),
}
});
let items = items.into_iter().map(|(item, _)| item).collect();
let mut promotions: Vec<_> = records
.values()
.filter(|record| matches!(record, LogEvent::Promotion { .. }))
.cloned()
.map(|record| {
let item = PromotionItem::from_record(record);
let timestamp = item
.ts
.parse::<jiff::Timestamp>()
.expect("folded promotions have valid RFC3339 timestamps");
(item, timestamp)
})
.collect();
promotions.sort_by(|(left, left_ts), (right, right_ts)| {
right_ts.cmp(left_ts).then_with(|| left.id.cmp(&right.id))
});
let promotions = promotions.into_iter().map(|(item, _)| item).collect();
let mut warnings = Vec::new();
warning(&mut warnings, counts.torn, "torn final line");
warning(&mut warnings, counts.malformed, "malformed line");
warning(&mut warnings, counts.unknown, "unknown event");
warning(&mut warnings, counts.duplicate_cuts, "duplicate cut");
warning(&mut warnings, counts.duplicate_dogears, "duplicate dogear");
warning(
&mut warnings,
counts.duplicate_promotions,
"duplicate promotion",
);
warning(
&mut warnings,
counts.duplicate_resolves,
"duplicate resolve",
);
warning(&mut warnings, counts.orphans, "orphan resolve");
warning(
&mut warnings,
counts.invalid_resolutions,
"invalid resolution",
);
FoldResult {
items,
promotions,
warnings,
records,
winning_amends,
lines,
}
}
fn record_kind(event: &LogEvent) -> Option<&'static str> {
match event {
LogEvent::Cut { .. } => Some("cut"),
LogEvent::Dogear { .. } => Some("dogear"),
LogEvent::Promotion { .. } => Some("promotion"),
LogEvent::Resolve { .. } | LogEvent::Unknown => None,
}
}
fn promotion_sources(records: &BTreeMap<String, LogEvent>) -> PromotionSources {
records
.iter()
.filter_map(|(id, event)| match event {
LogEvent::Promotion { sources, .. } => Some((id.clone(), sources.clone())),
_ => None,
})
.collect()
}
fn event_timestamp(event: &LogEvent) -> Option<jiff::Timestamp> {
match event {
LogEvent::Cut { ts, .. }
| LogEvent::Dogear { ts, .. }
| LogEvent::Resolve { ts, .. }
| LogEvent::Promotion { ts, .. } => ts.parse().ok(),
LogEvent::Unknown => None,
}
}
fn warning(warnings: &mut Vec<String>, count: usize, label: &str) {
if count > 0 {
warnings.push(format!(
"skipped {count} {label}{}",
if count == 1 { "" } else { "s" }
));
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{Impact, ItemStatus, compute_id};
use std::io::Write;
use tempfile::TempDir;
fn cut(id: &str) -> String {
cut_with_text(id, "x")
}
fn cut_with_text(id: &str, text: &str) -> String {
serde_json::json!({
"v":2, "kind":"cut", "id":id, "ts":"2026-07-09T00:00:00.000Z",
"agent":"a", "text":text, "tags":[], "impact":"low",
"cwd":"/tmp", "repo":null
})
.to_string()
}
fn resolve(id: &str) -> String {
serde_json::json!({
"v":2, "kind":"resolve", "id":id, "ts":"2026-07-10T00:00:00.000Z",
"agent":"a", "note":null,
"disposition":"fixed", "disposition_ts":"2026-07-10T00:00:00.000Z"
})
.to_string()
}
#[cfg(unix)]
#[test]
fn exclusive_lock_reopens_a_replaced_path_before_appending() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("cuts.jsonl");
std::fs::write(&path, b"old\n").unwrap();
let holder = OpenOptions::new()
.read(true)
.write(true)
.open(&path)
.unwrap();
holder.lock().unwrap();
let preopened = OpenOptions::new()
.read(true)
.append(true)
.open(&path)
.unwrap();
let (opened_tx, opened_rx) = std::sync::mpsc::channel();
let writer_path = path.clone();
let writer = std::thread::spawn(move || {
let mut first_open = Some(preopened);
let mut file = open_locked(&writer_path, true, || {
if let Some(file) = first_open.take() {
opened_tx.send(()).unwrap();
Ok(file)
} else {
OpenOptions::new()
.read(true)
.append(true)
.open(&writer_path)
.map_err(|error| AppError::from_log_open(error, &writer_path))
}
})
.unwrap();
file.write_all(b"writer\n").unwrap();
file.unlock().unwrap();
});
opened_rx
.recv_timeout(std::time::Duration::from_secs(2))
.unwrap();
let replacement = temp.path().join("replacement.jsonl");
std::fs::write(&replacement, b"replacement\n").unwrap();
std::fs::rename(&replacement, &path).unwrap();
holder.unlock().unwrap();
writer.join().unwrap();
assert_eq!(std::fs::read(&path).unwrap(), b"replacement\nwriter\n");
}
#[cfg(unix)]
#[test]
fn a_permanent_path_identity_mismatch_still_pays_the_retry_delay() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("cuts.jsonl");
let other = temp.path().join("other.jsonl");
std::fs::write(&path, b"").unwrap();
std::fs::write(&other, b"").unwrap();
let started = std::time::Instant::now();
let error = open_locked(&path, true, || {
OpenOptions::new()
.read(true)
.append(true)
.open(&other)
.map_err(|error| AppError::from_log_open(error, &other))
})
.expect_err("a permanent identity mismatch never locks the path");
let elapsed = started.elapsed();
assert_eq!(error.code, "lock_timeout");
assert_eq!(error.exit_code, 75);
assert!(
elapsed >= LOCK_DELAY * (LOCK_ATTEMPTS as u32 - 1),
"gave up after {elapsed:?}"
);
}
#[cfg(unix)]
#[test]
fn a_log_that_vanishes_during_the_retry_budget_reports_not_found() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("cuts.jsonl");
let other = temp.path().join("other.jsonl");
std::fs::write(&other, b"").unwrap();
let mut first = true;
let error = open_locked(&path, true, || {
let target = if std::mem::take(&mut first) {
other.as_path()
} else {
path.as_path()
};
OpenOptions::new()
.read(true)
.append(true)
.open(target)
.map_err(|error| AppError::from_log_open(error, target))
})
.expect_err("a log that never appears cannot be locked");
assert_eq!(error.code, "not_found");
assert_eq!(error.exit_code, 66);
}
#[test]
fn batch_append_rollback_restores_a_torn_tail_after_partial_write_failure() {
let temp = TempDir::new().unwrap();
let path = temp.path().join("cuts.jsonl");
let original = b"{\"kind\":\"cut\"}\n{\"kind\":";
std::fs::write(&path, original).unwrap();
let mut file = OpenOptions::new()
.read(true)
.append(true)
.open(&path)
.unwrap();
let error = append_bytes_with(
&mut file,
&path,
original,
b"{\"kind\":\"resolve\"}\n{\"kind\":\"resolve\"}\n",
|file, bytes| {
file.write_all(&bytes[..8])?;
Err(std::io::Error::other("injected partial write failure"))
},
)
.unwrap_err();
assert_eq!(error.code, "io_error");
assert_eq!(std::fs::read(&path).unwrap(), original);
}
#[test]
fn fold_matrix() {
let id = compute_id("2026-07-09T00:00:00.000Z", "a", "x", Impact::Low, &[]);
let cases = [
("cut", format!("{}\n", cut(&id)), 1, ItemStatus::Open, 0),
(
"resolve before cut",
format!("{}\n{}\n", resolve(&id), cut(&id)),
1,
ItemStatus::Resolved,
0,
),
(
"duplicates",
format!(
"{}\n{}\n{}\n{}\n",
cut(&id),
cut(&id),
resolve(&id),
resolve(&id)
),
1,
ItemStatus::Resolved,
2,
),
(
"unknown malformed orphan",
format!(
"{{\"v\":2,\"kind\":\"future\"}}\nnope\n{}\n{}\n",
resolve("bl_deadbeef000000000000"),
cut(&id)
),
1,
ItemStatus::Open,
3,
),
(
"torn tail",
format!("{}\n{{\"kind\":", cut(&id)),
1,
ItemStatus::Open,
1,
),
(
"all adversarial orderings interleaved",
format!(
"{}\n{{\"v\":2,\"kind\":\"future\"}}\n{}\n{}\n{}\n{}\n{}\nnope\n{{\"kind\":",
resolve(&id),
cut(&id),
cut(&id),
cut_with_text(&id, "conflicting payload"),
resolve(&id),
resolve("bl_deadbeef000000000000"),
),
1,
ItemStatus::Resolved,
6,
),
];
for (name, input, item_count, status, warning_count) in cases {
let folded = fold_bytes(input.as_bytes());
assert_eq!(folded.items.len(), item_count, "{name}");
if !folded.items.is_empty() {
assert_eq!(folded.items[0].status, status, "{name}");
assert_eq!(folded.items[0].text, "x", "{name}");
}
assert_eq!(folded.warnings.len(), warning_count, "{name}");
}
}
}