gate stale todo wakes on an emptiness check (#2678)
This commit is contained in:
parent
dfadacd45f
commit
001ea38ea4
2 changed files with 65 additions and 3 deletions
|
|
@ -500,22 +500,32 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
|
||||||
// acceptable since a from-scratch harness boot either has a writable
|
// acceptable since a from-scratch harness boot either has a writable
|
||||||
// harness dir or doesn't).
|
// harness dir or doesn't).
|
||||||
let todo_wake = Arc::new(tokio::sync::Notify::new());
|
let todo_wake = Arc::new(tokio::sync::Notify::new());
|
||||||
match todos::Todos::open(&paths::state_db()) {
|
// Kept alongside `todo_wake` so the serve loop's `LocalTodo` arm can
|
||||||
|
// gate a wake on `has_any()` before spawning a turn — see its doc
|
||||||
|
// comment (the phantom-todo-wake issue: a burst of same-turn upserts
|
||||||
|
// can arm a second `Notify` permit that outlives the turn that
|
||||||
|
// already drained its payload).
|
||||||
|
let todos_store: Option<Arc<todos::Todos>> = match todos::Todos::open(&paths::state_db()) {
|
||||||
Ok(store) => {
|
Ok(store) => {
|
||||||
let store = Arc::new(store);
|
let store = Arc::new(store);
|
||||||
let wake = todo_wake.clone();
|
let wake = todo_wake.clone();
|
||||||
let reminders = reminder_store.clone();
|
let reminders = reminder_store.clone();
|
||||||
let bus_for_socket = bus.clone();
|
let bus_for_socket = bus.clone();
|
||||||
|
let store_for_socket = store.clone();
|
||||||
tokio::spawn(async move {
|
tokio::spawn(async move {
|
||||||
if let Err(e) = todo_server::run(store, wake, reminders, bus_for_socket).await {
|
if let Err(e) =
|
||||||
|
todo_server::run(store_for_socket, wake, reminders, bus_for_socket).await
|
||||||
|
{
|
||||||
tracing::error!(error = %e, "in-agent todo socket exited with error");
|
tracing::error!(error = %e, "in-agent todo socket exited with error");
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
Some(store)
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled");
|
tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled");
|
||||||
|
None
|
||||||
}
|
}
|
||||||
}
|
};
|
||||||
if matches!(initial, LoginState::NeedsLogin) {
|
if matches!(initial, LoginState::NeedsLogin) {
|
||||||
login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
|
login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
|
||||||
} else {
|
} else {
|
||||||
|
|
@ -533,6 +543,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
|
||||||
stats,
|
stats,
|
||||||
&files,
|
&files,
|
||||||
todo_wake,
|
todo_wake,
|
||||||
|
todos_store,
|
||||||
reminder_rx,
|
reminder_rx,
|
||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
|
|
@ -556,6 +567,7 @@ async fn serve_loop<S: Surface>(
|
||||||
stats: Option<TurnStats>,
|
stats: Option<TurnStats>,
|
||||||
files: &turn::TurnFiles,
|
files: &turn::TurnFiles,
|
||||||
todo_wake: Arc<tokio::sync::Notify>,
|
todo_wake: Arc<tokio::sync::Notify>,
|
||||||
|
todos_store: Option<Arc<todos::Todos>>,
|
||||||
mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::DeliveredMessage>,
|
mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::DeliveredMessage>,
|
||||||
) -> Result<()> {
|
) -> Result<()> {
|
||||||
tracing::info!(socket = %socket.display(), "harness serve");
|
tracing::info!(socket = %socket.display(), "harness serve");
|
||||||
|
|
@ -586,6 +598,22 @@ async fn serve_loop<S: Surface>(
|
||||||
} {
|
} {
|
||||||
RecvOutcome::Message(first) => first,
|
RecvOutcome::Message(first) => first,
|
||||||
RecvOutcome::LocalTodo => {
|
RecvOutcome::LocalTodo => {
|
||||||
|
// Gate on `has_any()` before spawning a turn: a burst of
|
||||||
|
// same-turn upserts can arm a second `Notify` permit that
|
||||||
|
// outlives the turn which already drained its payload
|
||||||
|
// (the phantom-todo-wake issue) — `notify_one` doesn't
|
||||||
|
// coalesce once the first permit's been consumed, so the surplus wake
|
||||||
|
// fires the instant the loop is back here even though
|
||||||
|
// there's nothing left to show. Fail open (drive a turn
|
||||||
|
// anyway) on a `has_any` error so a flaky sqlite read
|
||||||
|
// never silently swallows a real wake.
|
||||||
|
let has_any = todos_store
|
||||||
|
.as_ref()
|
||||||
|
.is_none_or(|store| store.has_any().unwrap_or(true));
|
||||||
|
if !has_any {
|
||||||
|
tracing::debug!("todo wake fired against an empty store — stale, skipping");
|
||||||
|
continue;
|
||||||
|
}
|
||||||
tracing::debug!("todo wake consumed, sending synthetic todo message");
|
tracing::debug!("todo wake consumed, sending synthetic todo message");
|
||||||
synthetic_todo_message()
|
synthetic_todo_message()
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -223,6 +223,27 @@ impl Todos {
|
||||||
.collect::<rusqlite::Result<Vec<_>>>()?;
|
.collect::<rusqlite::Result<Vec<_>>>()?;
|
||||||
Ok(rows)
|
Ok(rows)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Cheap existence check — `true` when at least one todo row exists
|
||||||
|
/// (across every subsystem). Used to gate a `todo_wake` notification
|
||||||
|
/// against being turned into a turn when its payload has already been
|
||||||
|
/// drained by an earlier turn (see the serve loop's `LocalTodo` arm):
|
||||||
|
/// an `EXISTS` probe, not a full `list` + row materialization, since
|
||||||
|
/// this runs on every wake.
|
||||||
|
///
|
||||||
|
/// # Errors
|
||||||
|
///
|
||||||
|
/// Propagates the sqlite query failure.
|
||||||
|
///
|
||||||
|
/// # Panics
|
||||||
|
///
|
||||||
|
/// Panics if the connection mutex is poisoned.
|
||||||
|
pub fn has_any(&self) -> Result<bool> {
|
||||||
|
let conn = self.conn.lock().unwrap();
|
||||||
|
let any: bool =
|
||||||
|
conn.query_row("SELECT EXISTS(SELECT 1 FROM todos)", [], |row| row.get(0))?;
|
||||||
|
Ok(any)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
|
|
@ -279,6 +300,19 @@ mod tests {
|
||||||
assert!(s.list(None).unwrap().is_empty());
|
assert!(s.list(None).unwrap().is_empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn has_any_reflects_emptiness() {
|
||||||
|
let (_dir, s) = store();
|
||||||
|
assert!(!s.has_any().unwrap(), "fresh store has no todos");
|
||||||
|
let (id, _) = s.upsert("bash", None, "task done", None).unwrap();
|
||||||
|
assert!(s.has_any().unwrap());
|
||||||
|
s.mark_done(id).unwrap();
|
||||||
|
assert!(
|
||||||
|
!s.has_any().unwrap(),
|
||||||
|
"empty again after draining the only row"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn clear_subsystem_wipes_only_its_own() {
|
fn clear_subsystem_wipes_only_its_own() {
|
||||||
let (_dir, s) = store();
|
let (_dir, s) = store();
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue