Compare commits

..
Author SHA1 Message Date
Miguel PalhasandClaude Sonnet 5 0b34d1608f fix(maestro): make FaultyLogSink honor the Write contract on partial writes
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>
2026-09-05 05:02:01 +01:00
Miguel Palhas 075dea509a fix(maestro): make EventLog::append transactional under partial write/flush failure
A write_all that errors mid-line, or a flush that errors after a full
write, could leave last_seq and the log's on-disk line count out of
sync: a well-formed but uncounted line, or a partial one, would corrupt
every later cursor once the next append landed beside it. append now
captures the pre-append length and truncates back to it on either
failure, so the same EventLog can be appended to again immediately. If
the rollback itself can't restore that length, append returns a Fatal
variant so callers kill the session instead of promising consistency
they can no longer keep — record_permission_request and its pump now
route this through the session-death path used elsewhere.

Adds deterministic fault injection (a seekable/truncatable LogSink
double) to prove the same EventLog recovers with valid JSONL and
live/replay agreement, rather than swapping in a healthy handle. Fixes
the row-index rollback test to actually shift a later row, and
tightens the persistence-failure wording in STRUCTURED-VIEW.md.
2026-09-05 04:55:26 +01:00
Miguel Palhas c067f48a24 fix(maestro): roll back phantom approval row on log append failure
request_approval() appended the Approval row to the live projection
before the daemon tried to persist it. On append failure the pending
nonce was cancelled but the row stayed in the in-memory transcript, so
a WebSocket snapshot taken afterward showed a row durable replay could
never reproduce.

Add Projection::rollback_row to undo the append and call it alongside
the existing cancel_pending_approvals on failure. Extract the pump's
per-request handling into record_permission_request so the failure
path is directly testable without a live ACP connection.

Fault-injection regression tests use a read-only file handle
(EventLog::for_test, cfg(test)) to force append() to fail
deterministically, then assert the live snapshot and durable replay
agree (both empty) and a subsequent request still persists and
replays correctly.

Fixes #184.
2026-09-05 04:34:47 +01:00
Miguel Palhas 51f96cfb98 Merge structured-view/142-structured-activity into blitz/structured-view 2026-09-05 04:15:46 +01:00
4 changed files with 668 additions and 48 deletions
+83
View File
@@ -223,6 +223,21 @@ impl Projection {
})
}
/// Undo an `append`ed row that a caller decides never happened — used
/// when persisting the row's own event fails (#184), so a snapshot taken
/// after the failure cannot show a row durable replay would never
/// reproduce. Only ever called for a row that was just appended and
/// never patched, so removing it by id fully reverses `append`.
pub fn rollback_row(&mut self, id: &str) {
let Some(index) = self.row_index.remove(id) else {
return;
};
self.rows.remove(index);
for (shifted_index, row) in self.rows.iter().enumerate().skip(index) {
self.row_index.insert(row.id.clone(), shifted_index);
}
}
fn content(
&mut self,
seq: u64,
@@ -725,6 +740,74 @@ mod tests {
assert_eq!(projection.snapshot(0).rows, vec![row]);
}
#[test]
fn rollback_row_removes_only_that_row_and_keeps_others_addressable() {
let mut projection = Projection::default();
projection.apply(
1,
&event(serde_json::json!({
"sessionUpdate": "user_message_chunk",
"messageId": "m1",
"content": {"type": "text", "text": "hi"},
})),
);
let TranscriptDelta::Append { row } = projection
.request_approval("nonce".into(), serde_json::json!({"tool": "bash"}))
.transcript
.unwrap()
else {
panic!("approval row")
};
// A row after the one being rolled back, so removing the approval
// (index 1) shifts this one from index 2 down to index 1 — proving
// `row_index` gets reindexed, not just `rows`.
projection.apply(
2,
&event(serde_json::json!({
"sessionUpdate": "user_message_chunk",
"messageId": "m2",
"content": {"type": "text", "text": "hey"},
})),
);
assert_eq!(projection.snapshot(0).rows.len(), 3);
projection.rollback_row(&row.id);
assert_eq!(projection.snapshot(0).rows.len(), 2);
assert_eq!(projection.snapshot(0).rows[0].kind, TranscriptRowKind::User);
assert_eq!(projection.snapshot(0).rows[1].kind, TranscriptRowKind::User);
// The shifted row must still be independently patchable at its new
// index.
let update = projection.apply(
3,
&event(serde_json::json!({
"sessionUpdate": "user_message_chunk",
"messageId": "m2",
"content": {"type": "text", "text": " there"},
})),
);
let TranscriptDelta::Patch { row } = update.transcript.unwrap() else {
panic!("patched row")
};
assert_eq!(row.data.pointer("/content/text").unwrap(), "hey there");
assert_eq!(projection.snapshot(0).rows.len(), 2);
}
#[test]
fn rollback_row_is_a_no_op_for_an_unknown_id() {
let mut projection = Projection::default();
projection.apply(
1,
&event(serde_json::json!({
"sessionUpdate": "user_message_chunk",
"content": {"type": "text", "text": "hi"},
})),
);
projection.rollback_row("approval:does-not-exist");
assert_eq!(projection.snapshot(0).rows.len(), 1);
}
#[test]
fn recovered_approval_is_historical_not_pending() {
let mut live = Projection::default();
+247 -36
View File
@@ -25,7 +25,7 @@ use crate::panes::Panes;
use crate::projection::{Projection, ProjectionUpdate, TurnLifecycle};
use crate::pty::{pipe_pane_command, PtyBridge};
use crate::sidecar::{
EventCursor, EventLog, LoggedEvent, PersistedEvent, SidecarError, SidecarStore,
AppendError, EventCursor, EventLog, LoggedEvent, PersistedEvent, SidecarError, SidecarStore,
};
use crate::spawn_history::{RecentDir, SpawnHistory, SpawnHistoryRecord};
use crate::tmux::TmuxError;
@@ -2047,6 +2047,7 @@ fn start_acp_permission_pump(
return;
};
let weak_inner = Arc::downgrade(inner);
let weak_acp = Arc::downgrade(acp);
tokio::spawn(async move {
while let Some(PermissionRequest {
turn,
@@ -2059,41 +2060,23 @@ fn start_acp_permission_pump(
let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
continue;
}
let nonce = Uuid::new_v4().to_string();
let data = match serde_json::to_value(&request) {
Ok(value) => value,
Err(_) => {
let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
continue;
}
};
// ACP requests serialize through its connection. Keep the newest
// one authoritative if an adapter nonetheless overlaps them.
cancel_pending_approvals(&mut events);
events.pending_approvals.insert(
nonce.clone(),
PendingApproval {
response,
options: request.options,
},
);
let update = events.projection.request_approval(nonce, data);
if let Some(TranscriptDelta::Append { row }) = &update.transcript {
let event = PersistedEvent::Approval { row: row.clone() };
if let Err(error) = events.log.append(&event) {
tracing::error!(%error, "approval history append failed");
cancel_pending_approvals(&mut events);
drop(events);
if let Some(inner) = weak_inner.upgrade() {
sync_structured_activity(&inner, id).await;
}
continue;
}
let seq = events.log.last_seq();
let _ = events.live.send(LoggedEvent { seq, event });
let outcome = record_permission_request(&mut events, request, response);
let fatal = matches!(outcome, RecordOutcome::Fatal);
if let RecordOutcome::Live(update) = outcome {
let _ = events.projected_live.send(*update);
}
let _ = events.projected_live.send(update);
drop(events);
if fatal {
tracing::error!(%id, "approval history append rollback did not restore consistency");
if let Some(acp) = weak_acp.upgrade() {
acp.terminate(maestro_acp::CANCEL_GRACE, maestro_acp::TERM_GRACE)
.await;
}
if let Some(inner) = weak_inner.upgrade() {
mark_session_dead(&inner, id).await;
}
break;
}
if let Some(inner) = weak_inner.upgrade() {
sync_structured_activity(&inner, id).await;
}
@@ -2101,6 +2084,71 @@ fn start_acp_permission_pump(
});
}
/// What became of one [`record_permission_request`] call.
enum RecordOutcome {
/// Persisted; broadcast this live.
Live(Box<ProjectionUpdate>),
/// Not persisted, but the log rolled back cleanly: the approval was
/// cancelled and the session lives on.
Dropped,
/// Not persisted, and the rollback did not restore a consistent log.
/// The caller must treat the session as dead rather than keep appending
/// to an `EventLog` whose `last_seq` may no longer match its file.
Fatal,
}
/// Install `request` as the pending approval — cancelling whatever was
/// already pending, so the newest request wins (#139) — and persist its
/// transcript row before it is exposed live. On append failure the row is
/// rolled back from the projection so a later `TranscriptSnapshot` cannot
/// show a row durable replay could never reproduce (#184); the ACP callback
/// still resolves `Cancelled`, exactly as any other cancelled approval does.
fn record_permission_request(
events: &mut AcpEventStream,
request: maestro_acp::schema::RequestPermissionRequest,
response: oneshot::Sender<maestro_acp::schema::RequestPermissionOutcome>,
) -> RecordOutcome {
let nonce = Uuid::new_v4().to_string();
let data = match serde_json::to_value(&request) {
Ok(value) => value,
Err(_) => {
let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
return RecordOutcome::Dropped;
}
};
cancel_pending_approvals(events);
events.pending_approvals.insert(
nonce.clone(),
PendingApproval {
response,
options: request.options,
},
);
let update = events.projection.request_approval(nonce, data);
let Some(TranscriptDelta::Append { row }) = &update.transcript else {
return RecordOutcome::Live(Box::new(update));
};
let event = PersistedEvent::Approval { row: row.clone() };
match events.log.append(&event) {
Ok(seq) => {
let _ = events.live.send(LoggedEvent { seq, event });
RecordOutcome::Live(Box::new(update))
}
Err(AppendError::Failed(error)) => {
tracing::error!(%error, "approval history append failed");
events.projection.rollback_row(&row.id);
cancel_pending_approvals(events);
RecordOutcome::Dropped
}
Err(fatal @ AppendError::Fatal { .. }) => {
tracing::error!(error = %fatal, "approval history append failed unrecoverably");
events.projection.rollback_row(&row.id);
cancel_pending_approvals(events);
RecordOutcome::Fatal
}
}
}
fn cancel_pending_approvals(events: &mut AcpEventStream) {
for (_, pending) in events.pending_approvals.drain() {
let _ = pending
@@ -2653,13 +2701,176 @@ async fn cleanup_record(
#[cfg(test)]
mod tests {
use super::{permission_outcome, ApprovalDecision};
use maestro_acp::schema::{PermissionOption, PermissionOptionKind, RequestPermissionOutcome};
use std::fs::File;
use maestro_acp::schema::{
PermissionOption, PermissionOptionKind, RequestPermissionOutcome, RequestPermissionRequest,
ToolCallUpdate, ToolCallUpdateFields,
};
use super::{
cancel_pending_approvals, permission_outcome, record_permission_request, AcpDispatchState,
AcpEventStream, ApprovalDecision, RecordOutcome,
};
use crate::projection::Projection;
use crate::sidecar::{
EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore,
};
fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
PermissionOption::new(id.to_owned(), id, kind)
}
fn permission_request() -> RequestPermissionRequest {
RequestPermissionRequest::new(
"acp-session",
ToolCallUpdate::new("tool-1", ToolCallUpdateFields::new()),
vec![option("allow-once", PermissionOptionKind::AllowOnce)],
)
}
fn stream(log: EventLog) -> AcpEventStream {
let (live, _live_rx) = tokio::sync::broadcast::channel(16);
let (projected_live, _projected_rx) = tokio::sync::broadcast::channel(16);
AcpEventStream {
log,
live,
projection: Projection::default(),
projected_live,
pending_approvals: std::collections::HashMap::new(),
operation: std::sync::Arc::new(tokio::sync::Mutex::new(())),
dispatch: AcpDispatchState::default(),
}
}
/// A handle that can never append *or* roll back: opened read-only, so
/// both `write_all` and the rollback `truncate` fail with the OS's own
/// error (`ftruncate` requires a writable fd too) instead of a
/// hand-rolled one. A genuinely dead fd like this is exactly the case
/// where consistency cannot be promised, so it must come back fatal.
fn unwritable_log(path: &std::path::Path, last_seq: u64) -> EventLog {
EventLog::for_test(File::open(path).expect("open read-only handle"), last_seq)
}
#[tokio::test]
async fn failed_approval_append_with_no_usable_rollback_is_fatal_and_leaves_no_phantom_row() {
let dir = tempfile::tempdir().unwrap();
let store = SidecarStore::new(dir.path());
let id = uuid::Uuid::new_v4();
// Create the log file for real so a read-only handle onto it exists,
// then swap in the broken handle: the write is what must fail, not
// the open.
drop(store.open_events(id).unwrap());
let mut events = stream(unwritable_log(&store.events_path_for(id), 0));
let (response_tx, response_rx) = tokio::sync::oneshot::channel();
let outcome = record_permission_request(&mut events, permission_request(), response_tx);
assert!(
matches!(outcome, RecordOutcome::Fatal),
"a write failure whose rollback also fails must be reported fatal, not dropped"
);
assert!(
events.pending_approvals.is_empty(),
"the failed nonce must not linger as pending"
);
assert_eq!(
response_rx.await.unwrap(),
RequestPermissionOutcome::Cancelled,
"ACP must see the withdrawn-request outcome, not a hang"
);
let state = events.projection.state();
assert!(
state.pending_approval.is_none(),
"reduced state must not advertise a pending approval that never persisted"
);
assert!(
events.projection.snapshot(0).rows.is_empty(),
"the live projection must not carry a row durable replay cannot reproduce"
);
// Durable replay of the untouched real log must agree with the live
// projection: nothing happened.
let replayed = store
.read_events(id, EventCursor::Since(0), 100)
.expect("replay");
assert!(
replayed.is_empty(),
"durable replay must not diverge from the rolled-back live projection"
);
assert_eq!(
events.log.last_seq(),
0,
"cursor must stay put on a failed append"
);
}
#[tokio::test]
async fn approval_after_a_failed_one_persists_and_replays_cleanly() {
let dir = tempfile::tempdir().unwrap();
let store = SidecarStore::new(dir.path());
let id = uuid::Uuid::new_v4();
// A fault that fails the write once and then behaves like a healthy
// sink — the same `EventLog` instance must recover, unlike a handle
// swapped in from a healthy disk, which production never does
// (#184).
let buf = SharedBuf::default();
let sink = FaultyLogSink {
buf: buf.clone(),
fail_write_after: Some(0),
..Default::default()
};
let mut events = stream(EventLog::for_test(sink, 0));
let (first_tx, first_rx) = tokio::sync::oneshot::channel();
assert!(matches!(
record_permission_request(&mut events, permission_request(), first_tx),
RecordOutcome::Dropped
));
assert_eq!(first_rx.await.unwrap(), RequestPermissionOutcome::Cancelled);
assert_eq!(
events.log.last_seq(),
0,
"the failed attempt must not have advanced the cursor"
);
// Same instance, no swap: the sink is healthy again, so the next
// request over the same connection must not be poisoned by the
// earlier failure (concurrent/ordering audit for #184).
let (second_tx, _second_rx) = tokio::sync::oneshot::channel();
let outcome = record_permission_request(&mut events, permission_request(), second_tx);
assert!(
matches!(outcome, RecordOutcome::Live(_)),
"a healthy append must be exposed live"
);
assert_eq!(events.pending_approvals.len(), 1);
let rows = events.projection.snapshot(0).rows;
assert_eq!(rows.len(), 1, "no phantom row survives from the failure");
// Durable replay via the real read path, over the bytes the same
// instance actually wrote.
std::fs::create_dir_all(store.dir()).unwrap();
std::fs::write(store.events_path_for(id), buf.snapshot()).unwrap();
let replayed = store
.read_events(id, EventCursor::Since(0), 100)
.expect("replay");
assert_eq!(replayed.len(), 1, "the failed attempt left no line behind");
let PersistedEvent::Approval { row } = &replayed[0].event else {
panic!("approval event")
};
assert_eq!(
row.id, rows[0].id,
"live and durable must agree on identity"
);
assert_eq!(replayed[0].seq, events.log.last_seq());
cancel_pending_approvals(&mut events);
assert!(events.pending_approvals.is_empty());
}
fn selected_id(outcome: RequestPermissionOutcome) -> Option<String> {
match outcome {
RequestPermissionOutcome::Selected(selected) => Some(selected.option_id.to_string()),
+321 -8
View File
@@ -112,7 +112,7 @@ impl SidecarStore {
}
file.seek(SeekFrom::End(0))?;
Ok(EventLog {
file,
file: Box::new(file),
last_seq: lines,
})
}
@@ -296,15 +296,82 @@ impl From<&PersistedEvent> for PersistedEvent {
}
}
/// 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.
#[derive(Debug)]
pub struct EventLog {
file: File,
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
@@ -314,16 +381,262 @@ impl EventLog {
/// 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.
pub fn append(&mut self, event: impl Into<PersistedEvent>) -> Result<u64, SidecarError> {
///
/// 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)?,
_ => serde_json::to_vec(&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');
self.file.write_all(&line)?;
self.file.flush()?;
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"
);
}
}
+17 -4
View File
@@ -302,7 +302,14 @@ directory: the sidecar itself (`<id>.json`), plus two next to it:
daemon-generated row. Both share the same 1-based line-position cursor. A
line a hard kill left half-written is terminated before the next append, so
it keeps its position and is skipped as malformed rather than corrupting
what follows.
what follows. An append that fails *while the daemon is running* — a
`write_all` that errors after only a prefix reached disk, or a `flush` that
errors after a complete write — is rolled back to the exact byte length the
file was at before the call, so `last_seq` and the file's line count never
disagree and the same log handle can be appended to again immediately (#184).
If that rollback itself cannot restore the prior length, the append is
fatal: the log's consistency can no longer be promised, so the caller kills
the session instead of appending to it again.
- **`<id>.recover`** — an empty marker file. Its only meaning is "the daemon
that wrote this shut down gracefully while this session was running, and
the next daemon may attempt to resume it." It is consumed (deleted) the
@@ -314,9 +321,15 @@ persistence consequences:
1. **The adapter exits, or a notification append to the event log fails,
while the daemon keeps running.** (An approval-row append failure is
narrower: it cancels the pending approval and lets the session continue,
without marking it dead.) `mark_session_dead` applies the same
natural-exit worktree rule
narrower: as long as the log rolled back cleanly, it only cancels the
pending approval and rolls the unpersisted `Approval` row back out of the
live transcript, without marking the session dead — a snapshot taken after
the failure must never show a row durable replay could not reproduce
(#184). If the rollback itself did not restore the log's prior length,
this narrower path does not apply either: the session is marked dead the
same as any other append failure, since the log can no longer promise
`last_seq` matches its file.)
`mark_session_dead` applies the same natural-exit worktree rule
as a terminal session's own pane exiting (`docs/WORKTREES.md`): a pristine
worktree is removed, one holding unsaved work is left on disk. No recovery
marker is written. This death is permanent — a later daemon restart's