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.
This commit is contained in:
Miguel Palhas
2026-09-05 04:55:26 +01:00
parent c067f48a24
commit 075dea509a
4 changed files with 443 additions and 68 deletions
+97 -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,11 +2060,23 @@ fn start_acp_permission_pump(
let _ = response.send(maestro_acp::schema::RequestPermissionOutcome::Cancelled);
continue;
}
let update = record_permission_request(&mut events, request, response);
if let Some(update) = update {
let _ = events.projected_live.send(update);
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);
}
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;
}
@@ -2071,25 +2084,36 @@ 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.
/// Returns `None` when nothing should be broadcast live: an unserializable
/// request, or a persistence failure.
fn record_permission_request(
events: &mut AcpEventStream,
request: maestro_acp::schema::RequestPermissionRequest,
response: oneshot::Sender<maestro_acp::schema::RequestPermissionOutcome>,
) -> Option<ProjectionUpdate> {
) -> 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 None;
return RecordOutcome::Dropped;
}
};
cancel_pending_approvals(events);
@@ -2102,18 +2126,27 @@ fn record_permission_request(
);
let update = events.projection.request_approval(nonce, data);
let Some(TranscriptDelta::Append { row }) = &update.transcript else {
return Some(update);
return RecordOutcome::Live(Box::new(update));
};
let event = PersistedEvent::Approval { row: row.clone() };
if let Err(error) = events.log.append(&event) {
tracing::error!(%error, "approval history append failed");
events.projection.rollback_row(&row.id);
cancel_pending_approvals(events);
return None;
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
}
}
let seq = events.log.last_seq();
let _ = events.live.send(LoggedEvent { seq, event });
Some(update)
}
fn cancel_pending_approvals(events: &mut AcpEventStream) {
@@ -2677,10 +2710,12 @@ mod tests {
use super::{
cancel_pending_approvals, permission_outcome, record_permission_request, AcpDispatchState,
AcpEventStream, ApprovalDecision,
AcpEventStream, ApprovalDecision, RecordOutcome,
};
use crate::projection::Projection;
use crate::sidecar::{EventCursor, EventLog, PersistedEvent, SidecarStore};
use crate::sidecar::{
EventCursor, EventLog, FaultyLogSink, PersistedEvent, SharedBuf, SidecarStore,
};
fn option(id: &str, kind: PermissionOptionKind) -> PermissionOption {
PermissionOption::new(id.to_owned(), id, kind)
@@ -2708,15 +2743,17 @@ mod tests {
}
}
/// A handle that can never append: opened read-only, so `write_all`
/// fails with the OS's own error instead of a hand-rolled one — the
/// deterministic fault injection #184 asks for.
/// 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_leaves_no_phantom_row_and_replay_agrees() {
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();
@@ -2727,11 +2764,11 @@ mod tests {
let mut events = stream(unwritable_log(&store.events_path_for(id), 0));
let (response_tx, response_rx) = tokio::sync::oneshot::channel();
let update = record_permission_request(&mut events, permission_request(), response_tx);
let outcome = record_permission_request(&mut events, permission_request(), response_tx);
assert!(
update.is_none(),
"a failed persist must not be broadcast live"
matches!(outcome, RecordOutcome::Fatal),
"a write failure whose rollback also fails must be reported fatal, not dropped"
);
assert!(
events.pending_approvals.is_empty(),
@@ -2774,29 +2811,53 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let store = SidecarStore::new(dir.path());
let id = uuid::Uuid::new_v4();
drop(store.open_events(id).unwrap());
let mut events = stream(unwritable_log(&store.events_path_for(id), 0));
// 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!(record_permission_request(&mut events, permission_request(), first_tx).is_none());
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"
);
// A real disk recovers; the next request over the same connection
// must not be poisoned by the earlier failure (concurrent/ordering
// audit for #184).
events.log = store.open_events(id).unwrap();
// 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 update = record_permission_request(&mut events, permission_request(), second_tx);
let outcome = record_permission_request(&mut events, permission_request(), second_tx);
assert!(update.is_some(), "a healthy append must be exposed live");
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);
assert_eq!(replayed.len(), 1, "the failed attempt left no line behind");
let PersistedEvent::Approval { row } = &replayed[0].event else {
panic!("approval event")
};