Miguel PalhasandClaude Sonnet 5
0b34d1608f
write() reported Ok(partial) and Err from the same call, which Write forbids. Defer the error to the following call so write_all observes a genuine short write, then a failure — matching what a real fd does. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
643 lines
23 KiB
Rust
643 lines
23 KiB
Rust
//! Per-session JSON metadata store.
|
|
//!
|
|
//! On `CreateSession`, the daemon writes a `<id>.json` sidecar so that on
|
|
//! restart it can pair each surviving `maestro-<id>` tmux session with its
|
|
//! original `cwd`, command, env, metadata, etc. Files are deleted on
|
|
//! `KillSession`. Sessions whose sidecar is missing at recovery time are still
|
|
//! advertised but with empty metadata (handled by the caller, not here).
|
|
//!
|
|
//! A structured session (#135) also owns `<id>.events.jsonl` next to its
|
|
//! sidecar: every `session/update` notification and daemon approval row is one
|
|
//! JSON object, appended before it is broadcast. The log is the source of
|
|
//! truth for the transcript; `seq` is the line's 1-based position, so a
|
|
//! client that holds seq `n` asks for [`EventCursor::Since`]`(n)`. A line
|
|
//! that does not parse — including the partial line a hard kill leaves at
|
|
//! the end — keeps its seq and is skipped on read. `delete` removes the log
|
|
//! with the sidecar; `load_all` never reads one as a sidecar.
|
|
|
|
use std::collections::VecDeque;
|
|
use std::fs::{File, OpenOptions};
|
|
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
|
|
use std::path::{Path, PathBuf};
|
|
|
|
use maestro_acp::schema::SessionNotification;
|
|
use maestro_protocol::{Session, TranscriptRow};
|
|
use uuid::Uuid;
|
|
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum SidecarError {
|
|
#[error("io: {0}")]
|
|
Io(#[from] std::io::Error),
|
|
#[error("json: {0}")]
|
|
Json(#[from] serde_json::Error),
|
|
}
|
|
|
|
/// Filesystem-backed session metadata store, rooted at a directory.
|
|
#[derive(Debug, Clone)]
|
|
pub struct SidecarStore {
|
|
dir: PathBuf,
|
|
}
|
|
|
|
impl SidecarStore {
|
|
pub fn new(dir: impl Into<PathBuf>) -> Self {
|
|
Self { dir: dir.into() }
|
|
}
|
|
|
|
pub fn dir(&self) -> &Path {
|
|
&self.dir
|
|
}
|
|
|
|
pub fn path_for(&self, id: Uuid) -> PathBuf {
|
|
self.dir.join(format!("{id}.json"))
|
|
}
|
|
|
|
/// Where the per-session event log lives (#135).
|
|
pub fn events_path_for(&self, id: Uuid) -> PathBuf {
|
|
self.dir.join(format!("{id}.events.jsonl"))
|
|
}
|
|
|
|
/// Durable evidence that graceful daemon shutdown ended this structured
|
|
/// session and the next daemon may resume it.
|
|
pub fn recovery_path_for(&self, id: Uuid) -> PathBuf {
|
|
self.dir.join(format!("{id}.recover"))
|
|
}
|
|
|
|
/// Mark `id` as eligible for one structured-session recovery attempt.
|
|
pub fn mark_for_recovery(&self, id: Uuid) -> Result<(), SidecarError> {
|
|
std::fs::create_dir_all(&self.dir)?;
|
|
std::fs::write(self.recovery_path_for(id), [])?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Consume the graceful-shutdown recovery marker for `id`.
|
|
pub fn take_recovery_marker(&self, id: Uuid) -> Result<bool, SidecarError> {
|
|
match std::fs::remove_file(self.recovery_path_for(id)) {
|
|
Ok(()) => Ok(true),
|
|
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
|
|
Err(error) => Err(error.into()),
|
|
}
|
|
}
|
|
|
|
/// Open the event log for `id` for appending, creating it (and the
|
|
/// directory) when missing. One handle per live session: the caller
|
|
/// keeps it for the session's lifetime rather than reopening per
|
|
/// notification.
|
|
///
|
|
/// A log whose last line has no newline — what a hard kill mid-write
|
|
/// leaves behind — is terminated first, so the next append starts a
|
|
/// fresh line instead of gluing onto the fragment. The fragment keeps
|
|
/// the seq of its line and reads as malformed.
|
|
pub fn open_events(&self, id: Uuid) -> Result<EventLog, SidecarError> {
|
|
std::fs::create_dir_all(&self.dir)?;
|
|
let mut file = OpenOptions::new()
|
|
.create(true)
|
|
.read(true)
|
|
.append(true)
|
|
.open(self.events_path_for(id))?;
|
|
let mut lines = 0u64;
|
|
let mut last = None;
|
|
let mut reader = BufReader::new(&mut file);
|
|
let mut buf = [0u8; 64 * 1024];
|
|
loop {
|
|
let n = reader.read(&mut buf)?;
|
|
if n == 0 {
|
|
break;
|
|
}
|
|
lines += buf[..n].iter().filter(|&&b| b == b'\n').count() as u64;
|
|
last = Some(buf[n - 1]);
|
|
}
|
|
if last.is_some_and(|b| b != b'\n') {
|
|
file.write_all(b"\n")?;
|
|
lines += 1;
|
|
}
|
|
file.seek(SeekFrom::End(0))?;
|
|
Ok(EventLog {
|
|
file: Box::new(file),
|
|
last_seq: lines,
|
|
})
|
|
}
|
|
|
|
/// Read events from the log for `id`, at most `limit`, in ascending seq
|
|
/// order. A missing log is an empty log. Lines that fail to parse are
|
|
/// skipped but still consume a seq, so cursors stay stable.
|
|
pub fn read_events(
|
|
&self,
|
|
id: Uuid,
|
|
cursor: EventCursor,
|
|
limit: usize,
|
|
) -> Result<Vec<LoggedEvent>, SidecarError> {
|
|
let file = match File::open(self.events_path_for(id)) {
|
|
Ok(f) => f,
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
|
|
Err(e) => return Err(e.into()),
|
|
};
|
|
if limit == 0 {
|
|
return Ok(Vec::new());
|
|
}
|
|
let mut out: VecDeque<LoggedEvent> = VecDeque::new();
|
|
let mut seq = 0u64;
|
|
for line in BufReader::new(file).split(b'\n') {
|
|
let line = line?;
|
|
seq += 1;
|
|
let keep = match cursor {
|
|
EventCursor::Since(since) => seq > since,
|
|
EventCursor::Before(Some(before)) => seq < before,
|
|
EventCursor::Before(None) => true,
|
|
};
|
|
if !keep {
|
|
if matches!(cursor, EventCursor::Before(Some(_))) {
|
|
break;
|
|
}
|
|
continue;
|
|
}
|
|
let Ok(event) = PersistedEvent::decode(&line) else {
|
|
continue;
|
|
};
|
|
out.push_back(LoggedEvent { seq, event });
|
|
match cursor {
|
|
EventCursor::Since(_) if out.len() >= limit => break,
|
|
EventCursor::Before(_) if out.len() > limit => {
|
|
out.pop_front();
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
Ok(out.into())
|
|
}
|
|
|
|
/// Atomically write `session` to `<dir>/<id>.json`. Creates the directory
|
|
/// if it does not exist.
|
|
pub fn write(&self, session: &Session) -> Result<(), SidecarError> {
|
|
std::fs::create_dir_all(&self.dir)?;
|
|
let final_path = self.path_for(session.id);
|
|
let tmp = self.dir.join(format!("{}.json.tmp", session.id));
|
|
let json = serde_json::to_vec_pretty(session)?;
|
|
std::fs::write(&tmp, json)?;
|
|
std::fs::rename(&tmp, &final_path)?;
|
|
Ok(())
|
|
}
|
|
|
|
/// Read the sidecar for `id`. `Ok(None)` if the file is missing.
|
|
pub fn read(&self, id: Uuid) -> Result<Option<Session>, SidecarError> {
|
|
match std::fs::read(self.path_for(id)) {
|
|
Ok(bytes) => Ok(Some(serde_json::from_slice(&bytes)?)),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
|
|
Err(e) => Err(e.into()),
|
|
}
|
|
}
|
|
|
|
/// Delete the sidecar, event log, and recovery marker for `id`. No-op for
|
|
/// whichever does not exist.
|
|
pub fn delete(&self, id: Uuid) -> Result<(), SidecarError> {
|
|
remove_if_present(&self.events_path_for(id))?;
|
|
remove_if_present(&self.dir.join(format!("{id}.approvals.jsonl")))?;
|
|
remove_if_present(&self.recovery_path_for(id))?;
|
|
remove_if_present(&self.path_for(id))
|
|
}
|
|
|
|
/// Read every sidecar JSON in the directory. Files that fail to parse are
|
|
/// skipped silently — the daemon will still surface the live tmux session
|
|
/// with empty metadata, which matches the spec's recovery behavior.
|
|
pub fn load_all(&self) -> Result<Vec<Session>, SidecarError> {
|
|
let mut out = Vec::new();
|
|
let entries = match std::fs::read_dir(&self.dir) {
|
|
Ok(e) => e,
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(out),
|
|
Err(e) => return Err(e.into()),
|
|
};
|
|
for entry in entries {
|
|
let entry = entry?;
|
|
let path = entry.path();
|
|
// `.events.jsonl` logs (#135) sit in the same directory and are
|
|
// never sidecars.
|
|
if path.extension().and_then(|s| s.to_str()) != Some("json") {
|
|
continue;
|
|
}
|
|
let Ok(bytes) = std::fs::read(&path) else {
|
|
continue;
|
|
};
|
|
if let Ok(session) = serde_json::from_slice::<Session>(&bytes) {
|
|
out.push(session);
|
|
}
|
|
}
|
|
out.sort_by_key(|s| s.id);
|
|
Ok(out)
|
|
}
|
|
}
|
|
|
|
fn remove_if_present(path: &Path) -> Result<(), SidecarError> {
|
|
match std::fs::remove_file(path) {
|
|
Ok(()) => Ok(()),
|
|
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
|
|
Err(e) => Err(e.into()),
|
|
}
|
|
}
|
|
|
|
/// Which slice of a session's event log to read.
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum EventCursor {
|
|
/// Forward: events with `seq` strictly greater than the cursor, the
|
|
/// first `limit` of them. `Since(0)` is the whole log from the start.
|
|
/// What a client catching up after a reconnect asks for.
|
|
Since(u64),
|
|
/// Backward: events with `seq` strictly less than the cursor, the last
|
|
/// `limit` of them, still returned in ascending order. `None` means the
|
|
/// tail of the log, for a first paint. The next page passes the smallest
|
|
/// seq it received.
|
|
Before(Option<u64>),
|
|
}
|
|
|
|
/// One line of a session's event log.
|
|
#[derive(Debug, Clone, PartialEq)]
|
|
pub struct LoggedEvent {
|
|
/// 1-based line position in the log; the cursor for [`EventCursor`].
|
|
pub seq: u64,
|
|
pub event: PersistedEvent,
|
|
}
|
|
|
|
/// A single entry in the structured event history.
|
|
///
|
|
/// ACP notifications retain their original JSON representation, so existing
|
|
/// event logs remain readable. Daemon-generated rows use an envelope so they
|
|
/// can share the same cursor sequence without pretending to be ACP traffic.
|
|
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
|
|
#[serde(tag = "maestroEvent", rename_all = "snake_case")]
|
|
pub enum PersistedEvent {
|
|
Approval {
|
|
row: TranscriptRow,
|
|
},
|
|
#[serde(skip)]
|
|
Notification(Box<SessionNotification>),
|
|
}
|
|
|
|
impl PersistedEvent {
|
|
fn decode(line: &[u8]) -> Result<Self, serde_json::Error> {
|
|
serde_json::from_slice::<SessionNotification>(line)
|
|
.map(|notification| Self::Notification(Box::new(notification)))
|
|
.or_else(|_| serde_json::from_slice(line))
|
|
}
|
|
}
|
|
|
|
impl From<SessionNotification> for PersistedEvent {
|
|
fn from(notification: SessionNotification) -> Self {
|
|
Self::Notification(Box::new(notification))
|
|
}
|
|
}
|
|
|
|
impl From<&SessionNotification> for PersistedEvent {
|
|
fn from(notification: &SessionNotification) -> Self {
|
|
Self::Notification(Box::new(notification.clone()))
|
|
}
|
|
}
|
|
|
|
impl From<&PersistedEvent> for PersistedEvent {
|
|
fn from(event: &PersistedEvent) -> Self {
|
|
event.clone()
|
|
}
|
|
}
|
|
|
|
/// What an [`EventLog`] appends onto: something seekable by length and
|
|
/// truncatable back to one, so a failed append can be rolled off the end
|
|
/// again. A real [`File`] in production; an in-memory fault double in tests
|
|
/// (#184) — the same [`EventLog::append`] rollback logic runs over both, so
|
|
/// the behavior under test is the production behavior, not a stand-in for
|
|
/// it.
|
|
pub(crate) trait LogSink: Write + Send {
|
|
fn len(&mut self) -> std::io::Result<u64>;
|
|
fn truncate(&mut self, len: u64) -> std::io::Result<()>;
|
|
}
|
|
|
|
impl LogSink for File {
|
|
fn len(&mut self) -> std::io::Result<u64> {
|
|
Ok(self.metadata()?.len())
|
|
}
|
|
|
|
fn truncate(&mut self, len: u64) -> std::io::Result<()> {
|
|
self.set_len(len)
|
|
}
|
|
}
|
|
|
|
/// Outcome of a failed [`EventLog::append`].
|
|
#[derive(Debug, thiserror::Error)]
|
|
pub enum AppendError {
|
|
/// The write or flush failed, but the log was rolled back to exactly the
|
|
/// byte it was at before the call: `last_seq` and the file agree, and
|
|
/// this same [`EventLog`] is safe to keep appending to. The event itself
|
|
/// was never persisted or broadcast.
|
|
#[error("event append failed, rolled back cleanly: {0}")]
|
|
Failed(#[source] SidecarError),
|
|
/// The append failed *and* the rollback did not restore the log to its
|
|
/// prior length, so the file's line count and `last_seq` may now
|
|
/// disagree. The caller must not append to this [`EventLog`] again —
|
|
/// treat the session as dead rather than promise consistency it cannot
|
|
/// keep.
|
|
#[error("event append failed ({cause}) and rollback did not restore consistency ({rollback})")]
|
|
Fatal {
|
|
cause: SidecarError,
|
|
rollback: SidecarError,
|
|
},
|
|
}
|
|
|
|
impl From<SidecarError> for AppendError {
|
|
fn from(error: SidecarError) -> Self {
|
|
AppendError::Failed(error)
|
|
}
|
|
}
|
|
|
|
/// An open, append-only handle on one session's event log. Obtained from
|
|
/// [`SidecarStore::open_events`]; held for the session's lifetime.
|
|
pub struct EventLog {
|
|
file: Box<dyn LogSink>,
|
|
last_seq: u64,
|
|
}
|
|
|
|
impl std::fmt::Debug for EventLog {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("EventLog")
|
|
.field("last_seq", &self.last_seq)
|
|
.finish_non_exhaustive()
|
|
}
|
|
}
|
|
|
|
impl EventLog {
|
|
/// Wrap an arbitrary sink as an [`EventLog`], for tests that need to
|
|
/// inject a deterministic append failure (#184) — a partial write, a
|
|
/// flush that fails after the bytes are already down, or a rollback that
|
|
/// cannot itself complete.
|
|
#[cfg(test)]
|
|
pub(crate) fn for_test(file: impl LogSink + 'static, last_seq: u64) -> Self {
|
|
Self {
|
|
file: Box::new(file),
|
|
last_seq,
|
|
}
|
|
}
|
|
|
|
/// The seq of the last line, `0` for an empty log.
|
|
pub fn last_seq(&self) -> u64 {
|
|
self.last_seq
|
|
}
|
|
|
|
/// Append one notification as a line and return its seq. The line is
|
|
/// handed to the kernel in a single write before this returns, so a
|
|
/// caller that broadcasts afterwards never announces an event the log
|
|
/// would not replay — short of the machine losing power.
|
|
///
|
|
/// A `write_all` that errors partway can leave an unterminated prefix on
|
|
/// disk; a `flush` that errors after a *complete* `write_all` leaves
|
|
/// ambiguity about whether that full, well-formed line should count —
|
|
/// counting it would silently shift every later seq by one, since
|
|
/// `last_seq` was never bumped for it. Both cases are rolled back to the
|
|
/// exact pre-append length before returning, so `last_seq` and the file
|
|
/// never disagree and the same [`EventLog`] can be appended to again
|
|
/// immediately. If the rollback itself fails, the log's consistency can
|
|
/// no longer be promised and [`AppendError::Fatal`] is returned instead.
|
|
pub fn append(&mut self, event: impl Into<PersistedEvent>) -> Result<u64, AppendError> {
|
|
let event = event.into();
|
|
let mut line = match &event {
|
|
PersistedEvent::Notification(notification) => serde_json::to_vec(notification)
|
|
.map_err(|e| AppendError::from(SidecarError::from(e)))?,
|
|
_ => {
|
|
serde_json::to_vec(&event).map_err(|e| AppendError::from(SidecarError::from(e)))?
|
|
}
|
|
};
|
|
line.push(b'\n');
|
|
|
|
let pre_len = self
|
|
.file
|
|
.len()
|
|
.map_err(|e| AppendError::from(SidecarError::from(e)))?;
|
|
|
|
if let Err(cause) = self.file.write_all(&line) {
|
|
return Err(self.rollback_after(pre_len, cause));
|
|
}
|
|
if let Err(cause) = self.file.flush() {
|
|
return Err(self.rollback_after(pre_len, cause));
|
|
}
|
|
|
|
self.last_seq += 1;
|
|
Ok(self.last_seq)
|
|
}
|
|
|
|
fn rollback_after(&mut self, pre_len: u64, cause: std::io::Error) -> AppendError {
|
|
match self.file.truncate(pre_len) {
|
|
Ok(()) => AppendError::Failed(SidecarError::Io(cause)),
|
|
Err(rollback) => AppendError::Fatal {
|
|
cause: SidecarError::Io(cause),
|
|
rollback: SidecarError::Io(rollback),
|
|
},
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Bytes shared between a [`FaultyLogSink`] moved into an [`EventLog`] and
|
|
/// the test that planted it, so the test can inspect what actually ended up
|
|
/// on the "disk" after a fault without reaching into `EventLog`'s private
|
|
/// `file` field.
|
|
#[cfg(test)]
|
|
#[derive(Clone, Default)]
|
|
pub(crate) struct SharedBuf(std::sync::Arc<std::sync::Mutex<Vec<u8>>>);
|
|
|
|
#[cfg(test)]
|
|
impl SharedBuf {
|
|
pub(crate) fn snapshot(&self) -> Vec<u8> {
|
|
self.0.lock().unwrap().clone()
|
|
}
|
|
}
|
|
|
|
/// An in-memory [`LogSink`] double that injects one deterministic fault — a
|
|
/// short write, a flush that fails after a full write, or a rollback that
|
|
/// cannot itself complete — then behaves like a healthy sink from then on.
|
|
/// Lets a test prove the *same* [`EventLog`] recovers, rather than one with a
|
|
/// handle swapped in from a healthy disk, which production never does
|
|
/// (#184).
|
|
#[cfg(test)]
|
|
#[derive(Default)]
|
|
pub(crate) struct FaultyLogSink {
|
|
pub(crate) buf: SharedBuf,
|
|
/// Accept this many bytes on the next `write` call, then fail the
|
|
/// *following* call — a short write followed by an error, exactly what
|
|
/// `write_all` sees from a real fd that dies mid-line. `Write::write`
|
|
/// must not report an error on a call that made partial progress, so
|
|
/// the fault is delivered one call later, via `pending_write_error`.
|
|
pub(crate) fail_write_after: Option<usize>,
|
|
pub(crate) pending_write_error: bool,
|
|
pub(crate) fail_flush_once: bool,
|
|
pub(crate) fail_truncate: bool,
|
|
}
|
|
|
|
#[cfg(test)]
|
|
impl Write for FaultyLogSink {
|
|
fn write(&mut self, data: &[u8]) -> std::io::Result<usize> {
|
|
if self.pending_write_error {
|
|
self.pending_write_error = false;
|
|
return Err(std::io::Error::other("fault: write failed mid-line"));
|
|
}
|
|
if let Some(allowed) = self.fail_write_after.take() {
|
|
let allowed = allowed.min(data.len());
|
|
if allowed == 0 {
|
|
// No progress was made, so it's fine to report the error
|
|
// from this same call.
|
|
return Err(std::io::Error::other("fault: write failed mid-line"));
|
|
}
|
|
self.buf
|
|
.0
|
|
.lock()
|
|
.unwrap()
|
|
.extend_from_slice(&data[..allowed]);
|
|
self.pending_write_error = true;
|
|
return Ok(allowed);
|
|
}
|
|
self.buf.0.lock().unwrap().extend_from_slice(data);
|
|
Ok(data.len())
|
|
}
|
|
|
|
fn flush(&mut self) -> std::io::Result<()> {
|
|
if self.fail_flush_once {
|
|
self.fail_flush_once = false;
|
|
return Err(std::io::Error::other(
|
|
"fault: flush failed after a full write",
|
|
));
|
|
}
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
impl LogSink for FaultyLogSink {
|
|
fn len(&mut self) -> std::io::Result<u64> {
|
|
Ok(self.buf.0.lock().unwrap().len() as u64)
|
|
}
|
|
|
|
fn truncate(&mut self, len: u64) -> std::io::Result<()> {
|
|
if self.fail_truncate {
|
|
return Err(std::io::Error::other("fault: rollback truncate failed"));
|
|
}
|
|
self.buf.0.lock().unwrap().truncate(len as usize);
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn sample_event(id: &str) -> PersistedEvent {
|
|
PersistedEvent::Approval {
|
|
row: TranscriptRow {
|
|
id: id.to_owned(),
|
|
kind: maestro_protocol::TranscriptRowKind::Approval,
|
|
data: serde_json::json!({"tool": "bash"}),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Replays `bytes` as a real on-disk log via the production read path,
|
|
/// so "live and durable agree" is checked with the same code a
|
|
/// reconnecting client would hit, not a hand-rolled line splitter.
|
|
fn replay(bytes: &[u8]) -> Vec<LoggedEvent> {
|
|
let dir = tempfile::tempdir().unwrap();
|
|
let store = SidecarStore::new(dir.path());
|
|
let id = Uuid::new_v4();
|
|
std::fs::create_dir_all(store.dir()).unwrap();
|
|
std::fs::write(store.events_path_for(id), bytes).unwrap();
|
|
store.read_events(id, EventCursor::Since(0), 100).unwrap()
|
|
}
|
|
|
|
#[test]
|
|
fn append_rolls_back_a_partial_write_and_the_same_instance_recovers() {
|
|
let buf = SharedBuf::default();
|
|
let writer = FaultyLogSink {
|
|
buf: buf.clone(),
|
|
fail_write_after: Some(5),
|
|
..Default::default()
|
|
};
|
|
let mut log = EventLog::for_test(writer, 0);
|
|
|
|
let err = log.append(sample_event("r1")).unwrap_err();
|
|
assert!(
|
|
matches!(err, AppendError::Failed(_)),
|
|
"a repairable rollback must not be reported fatal"
|
|
);
|
|
assert_eq!(log.last_seq(), 0, "cursor must stay put on a failed append");
|
|
assert!(
|
|
buf.snapshot().is_empty(),
|
|
"the partial prefix a failed write_all leaves behind must be truncated away"
|
|
);
|
|
|
|
let seq = log
|
|
.append(sample_event("r1"))
|
|
.expect("the same instance must be able to append again after a clean rollback");
|
|
assert_eq!(seq, 1);
|
|
assert_eq!(log.last_seq(), 1);
|
|
|
|
let bytes = buf.snapshot();
|
|
assert_eq!(
|
|
bytes.iter().filter(|&&b| b == b'\n').count(),
|
|
1,
|
|
"exactly one well-formed line, not a partial prefix glued to the recovered append"
|
|
);
|
|
|
|
let replayed = replay(&bytes);
|
|
assert_eq!(replayed.len(), 1);
|
|
assert_eq!(replayed[0].seq, log.last_seq());
|
|
let PersistedEvent::Approval { row } = &replayed[0].event else {
|
|
panic!("approval event");
|
|
};
|
|
assert_eq!(row.id, "r1");
|
|
}
|
|
|
|
#[test]
|
|
fn append_rolls_back_when_flush_fails_after_a_full_write() {
|
|
let buf = SharedBuf::default();
|
|
let writer = FaultyLogSink {
|
|
buf: buf.clone(),
|
|
fail_flush_once: true,
|
|
..Default::default()
|
|
};
|
|
let mut log = EventLog::for_test(writer, 0);
|
|
|
|
let err = log.append(sample_event("r1")).unwrap_err();
|
|
assert!(matches!(err, AppendError::Failed(_)));
|
|
assert_eq!(log.last_seq(), 0);
|
|
assert!(
|
|
buf.snapshot().is_empty(),
|
|
"a full-but-unflushed line must not survive uncounted by last_seq — it would \
|
|
silently shift every later seq by one"
|
|
);
|
|
|
|
let seq = log
|
|
.append(sample_event("r1"))
|
|
.expect("the same instance must recover once flush stops failing");
|
|
assert_eq!(seq, 1);
|
|
|
|
let bytes = buf.snapshot();
|
|
let replayed = replay(&bytes);
|
|
assert_eq!(replayed.len(), 1);
|
|
assert_eq!(replayed[0].seq, log.last_seq());
|
|
}
|
|
|
|
#[test]
|
|
fn append_reports_fatal_when_rollback_cannot_restore_consistency() {
|
|
let buf = SharedBuf::default();
|
|
let writer = FaultyLogSink {
|
|
buf,
|
|
fail_write_after: Some(3),
|
|
fail_truncate: true,
|
|
..Default::default()
|
|
};
|
|
let mut log = EventLog::for_test(writer, 0);
|
|
|
|
let err = log.append(sample_event("r1")).unwrap_err();
|
|
assert!(
|
|
matches!(err, AppendError::Fatal { .. }),
|
|
"an unrepairable rollback must be surfaced as fatal, never silently treated as \
|
|
a clean recoverable failure"
|
|
);
|
|
assert_eq!(
|
|
log.last_seq(),
|
|
0,
|
|
"a fatal append must not advance the cursor either"
|
|
);
|
|
}
|
|
}
|