Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0b34d1608f | ||
|
|
075dea509a | ||
|
|
c067f48a24 | ||
|
|
51f96cfb98 |
@@ -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();
|
||||
|
||||
@@ -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()),
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user