diff --git a/crates/daemon/src/service/ingress.rs b/crates/daemon/src/service/ingress.rs index 7699f6df..dd482690 100644 --- a/crates/daemon/src/service/ingress.rs +++ b/crates/daemon/src/service/ingress.rs @@ -20,7 +20,8 @@ use tokio_util::sync::CancellationToken; use uuid::Uuid; const REQUEST_DEDUP_CAP: usize = 4096; -const PENDING_DELIVERY_TTL: std::time::Duration = std::time::Duration::from_secs(30 * 60); +pub(super) const PENDING_DELIVERY_TTL: std::time::Duration = + std::time::Duration::from_secs(30 * 60); /// How long a session must sit idle and quiet, with no approval pending, /// before a turn that produced no answer is called finished-and-failed. @@ -41,6 +42,32 @@ struct PendingDelivery { created_at: tokio::time::Instant, } +/// A delivery this service accepted and has not answered yet. +/// +/// The task waiting on a turn lives exactly as long as the daemon process. A +/// restart mid-turn therefore takes the waiter with it, and everything the +/// channel put in front of the person waiting — a placeholder, a reaction — +/// is left standing with nothing left alive to replace it. Recording the +/// delivery is what lets the next daemon pick the wait back up. +#[derive(Clone, Serialize, Deserialize)] +pub(super) struct OutstandingDelivery { + /// Which channel of this service accepted it. A service may run several. + pub(super) channel_id: String, + pub(super) session: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub(super) delivery_id: Option, + /// Transcript position the turn began at, so a resumed wait does not + /// mistake the previous answer in a long-lived session for this one. + pub(super) event_cursor: usize, + pub(super) submitted_at: chrono::DateTime, + /// Whatever the channel needs to find what it already put on screen. + /// Opaque here on purpose: the ingress does not render anything, and a + /// second kind of channel must not require a schema change in state + /// shared by all of them. + #[serde(default)] + pub(super) context: serde_json::Value, +} + #[derive(Default, Serialize, Deserialize)] pub(super) struct PersistedState { #[serde(default)] @@ -50,6 +77,11 @@ pub(super) struct PersistedState { /// but still need to remain queryable by the channel that created them. #[serde(default)] pub(super) owned_sessions: HashSet, + /// Accepted-but-unanswered deliveries, keyed by whatever the channel uses + /// to identify a request. Empty in the steady state — a record lives only + /// between accepting a delivery and resolving it. + #[serde(default)] + pub(super) outstanding: HashMap, } impl PersistedState { @@ -209,6 +241,111 @@ impl ServiceIngress { .any(|key| key.starts_with(&lookup_prefix)) } + /// Record a delivery as accepted-but-unanswered, so a daemon that restarts + /// mid-turn can pick the wait back up instead of abandoning it. + pub(super) async fn record_outstanding( + &self, + key: &str, + receipt: &IngressReceipt, + context: serde_json::Value, + ) { + let record = OutstandingDelivery { + channel_id: self.channel_id.clone(), + session: receipt.session.clone(), + delivery_id: receipt.delivery_id.clone(), + event_cursor: receipt.event_cursor, + submitted_at: chrono::Utc::now(), + context, + }; + let mut state = self.shared.state.lock().await; + state.outstanding.insert(self.outstanding_key(key), record); + if let Err(error) = self.persist_state(&state).await { + tracing::warn!( + service = %self.shared.name, + %error, + "could not record an outstanding delivery; a restart would abandon it" + ); + } + } + + /// Amend what the channel has left on screen for an outstanding delivery. + /// + /// A channel does not know everything it will need to clean up at the + /// moment it accepts a delivery — a progress placeholder, for one, only + /// exists once a turn has run long enough to deserve it. + pub(super) async fn amend_outstanding(&self, key: &str, context: serde_json::Value) { + let mut state = self.shared.state.lock().await; + let Some(record) = state.outstanding.get_mut(&self.outstanding_key(key)) else { + return; + }; + record.context = context; + if let Err(error) = self.persist_state(&state).await { + tracing::warn!(service = %self.shared.name, %error, "could not amend an outstanding delivery"); + } + } + + /// Forget a delivery that has been resolved one way or the other. + pub(super) async fn clear_outstanding(&self, key: &str) { + let mut state = self.shared.state.lock().await; + if state.outstanding.remove(&self.outstanding_key(key)).is_none() { + return; + } + if let Err(error) = self.persist_state(&state).await { + tracing::warn!(service = %self.shared.name, %error, "could not clear a resolved delivery"); + } + } + + /// Every delivery this channel accepted and never answered. + /// + /// Records are left in place rather than consumed: a channel interrupted + /// again while reconciling must find them a second time, and resolving one + /// twice edits the same message rather than losing it. + pub(super) async fn outstanding(&self) -> Vec<(String, OutstandingDelivery)> { + let prefix = format!("{}:", self.channel_id); + self.shared + .state + .lock() + .await + .outstanding + .iter() + .filter(|(_, record)| record.channel_id == self.channel_id) + .map(|(key, record)| { + ( + key.strip_prefix(&prefix).unwrap_or(key).to_string(), + record.clone(), + ) + }) + .collect() + } + + fn outstanding_key(&self, key: &str) -> String { + format!("{}:{key}", self.channel_id) + } + + /// Rebuild a receipt for a delivery recorded before a restart. + /// + /// `None` when the session it was routed to no longer exists, which is the + /// one case a resumed wait cannot recover from. + pub(super) async fn resume_outstanding( + &self, + record: &OutstandingDelivery, + ) -> Option { + self.shared.manager.get_entry(&record.session).await?; + // The reply tool authorizes against this map, which did not survive + // the restart either. Without it a turn still running would be told + // its delivery is unknown at the moment it tries to answer. + if let Some(delivery_id) = record.delivery_id.as_ref() { + self.shared + .restore_delivery(delivery_id.clone(), record.session.clone()) + .await; + } + Some(IngressReceipt { + session: record.session.clone(), + event_cursor: record.event_cursor, + delivery_id: record.delivery_id.clone(), + }) + } + /// Submit a native channel delivery and retain the transcript position at /// which its turn began. Long-lived adapters use this cursor to avoid /// mistaking the previous turn's final answer for the new one. @@ -366,18 +503,22 @@ impl ServiceIngress { /// can tell its user rather than going silent for the whole turn. It is /// advisory: nothing here waits on a reader, and a channel that does not /// render progress simply ignores it. + /// `budget` is how much longer this turn may run. A delivery picked back up + /// after a restart passes what is left of its original allowance rather + /// than a fresh one, so surviving a restart cannot extend a turn's life. pub(super) async fn wait_for_final( &self, receipt: &IngressReceipt, cancel: &CancellationToken, progress: &watch::Sender, + budget: std::time::Duration, ) -> Result { if let Some(delivery_id) = receipt.delivery_id.as_deref() { return self - .wait_for_explicit_reply(receipt, delivery_id, cancel, progress) + .wait_for_explicit_reply(receipt, delivery_id, cancel, progress, budget) .await; } - let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(30 * 60); + let deadline = tokio::time::Instant::now() + budget; let mut watch = TurnWatch::default(); loop { tokio::select! { @@ -430,8 +571,9 @@ impl ServiceIngress { delivery_id: &str, cancel: &CancellationToken, progress: &watch::Sender, + budget: std::time::Duration, ) -> Result { - let deadline = tokio::time::Instant::now() + PENDING_DELIVERY_TTL; + let deadline = tokio::time::Instant::now() + budget; let mut watch = TurnWatch::default(); loop { tokio::select! { @@ -935,10 +1077,10 @@ pub(super) fn pending_approval( } #[cfg(test)] -mod tests { +pub(super) mod tests { use super::*; - async fn shared_for_delivery_tests() -> Arc { + pub(in crate::service) async fn shared_for_delivery_tests() -> Arc { let tmp = Box::leak(Box::new(tempfile::tempdir().expect("tempdir"))); let storage = Arc::new(crate::storage::Storage::new(tmp.path().join("data")).expect("storage")); @@ -1108,6 +1250,132 @@ mod tests { assert!(!other.has_session_under("T1:C1:").await); } + #[tokio::test] + async fn an_accepted_delivery_stays_recorded_until_it_is_answered() { + // The record is what a restarted daemon reads. It has to exist for the + // whole window in which a restart could happen — from the moment the + // delivery is accepted to the moment something is finally said about + // it — because that window is exactly when a waiter can be lost. + let shared = shared_for_delivery_tests().await; + let ingress = ServiceIngress::new("C1".to_string(), shared); + let receipt = IngressReceipt { + session: "s1".to_string(), + event_cursor: 7, + delivery_id: Some("d1".to_string()), + }; + let context = serde_json::json!({"channel": "C1", "thread_ts": "1.1"}); + + ingress + .record_outstanding("C1:1.1", &receipt, context.clone()) + .await; + let outstanding = ingress.outstanding().await; + assert_eq!(outstanding.len(), 1); + let (key, record) = &outstanding[0]; + assert_eq!(key, "C1:1.1", "the channel's own key comes back"); + assert_eq!(record.session, "s1"); + assert_eq!(record.event_cursor, 7, "a resumed wait starts where it left"); + assert_eq!(record.delivery_id.as_deref(), Some("d1")); + assert_eq!(record.context, context); + + // The affordance appears later than acceptance, so what the channel + // has on screen has to be amendable after the fact. + let amended = serde_json::json!({"channel": "C1", "thread_ts": "1.1", "placeholder_ts": "P1"}); + ingress.amend_outstanding("C1:1.1", amended.clone()).await; + assert_eq!(ingress.outstanding().await[0].1.context, amended); + + ingress.clear_outstanding("C1:1.1").await; + assert!( + ingress.outstanding().await.is_empty(), + "a resolved delivery stops being outstanding" + ); + } + + #[tokio::test] + async fn an_outstanding_delivery_survives_being_written_and_read_back() { + // The record is only worth keeping if it reaches disk: an in-memory + // one is lost by exactly the event it exists to survive. + let shared = shared_for_delivery_tests().await; + let ingress = ServiceIngress::new("C1".to_string(), shared.clone()); + ingress + .record_outstanding( + "C1:1.1", + &IngressReceipt { + session: "s1".to_string(), + event_cursor: 3, + delivery_id: None, + }, + serde_json::json!({"channel": "C1"}), + ) + .await; + + let raw = tokio::fs::read(&shared.state_path) + .await + .expect("state was persisted"); + let reloaded: PersistedState = serde_json::from_slice(&raw).expect("state parses"); + let record = reloaded + .outstanding + .get("C1:C1:1.1") + .expect("the delivery is on disk"); + assert_eq!(record.session, "s1"); + assert_eq!(record.channel_id, "C1"); + } + + #[tokio::test] + async fn a_delivery_whose_session_is_gone_cannot_be_resumed() { + // The one thing a resumed wait cannot recover from: there is nothing + // left to wait on, so the channel has to report it instead. + let shared = shared_for_delivery_tests().await; + let ingress = ServiceIngress::new("C1".to_string(), shared); + let record = OutstandingDelivery { + channel_id: "C1".into(), + session: "s-does-not-exist".into(), + delivery_id: None, + event_cursor: 0, + submitted_at: chrono::Utc::now(), + context: serde_json::Value::Null, + }; + assert!(ingress.resume_outstanding(&record).await.is_none()); + } + + #[tokio::test] + async fn one_channels_outstanding_deliveries_are_not_anothers() { + // A service runs several channels against one state file. A channel + // that reconciled another's deliveries would answer into a + // conversation it does not own. + let shared = shared_for_delivery_tests().await; + let first = ServiceIngress::new("C1".to_string(), shared.clone()); + let second = ServiceIngress::new("C2".to_string(), shared); + first + .record_outstanding( + "1.1", + &IngressReceipt { + session: "s1".to_string(), + event_cursor: 0, + delivery_id: None, + }, + serde_json::Value::Null, + ) + .await; + + assert_eq!(first.outstanding().await.len(), 1); + assert!(second.outstanding().await.is_empty()); + + // Same key, different channels: the records must not collide. + second + .record_outstanding( + "1.1", + &IngressReceipt { + session: "s2".to_string(), + event_cursor: 0, + delivery_id: None, + }, + serde_json::Value::Null, + ) + .await; + assert_eq!(first.outstanding().await[0].1.session, "s1"); + assert_eq!(second.outstanding().await[0].1.session, "s2"); + } + #[test] fn progress_reports_the_approval_a_turn_is_stopped_at() { let at = chrono::Utc::now(); diff --git a/crates/daemon/src/service/slack.rs b/crates/daemon/src/service/slack.rs index 24667c13..a219766a 100644 --- a/crates/daemon/src/service/slack.rs +++ b/crates/daemon/src/service/slack.rs @@ -4,11 +4,13 @@ //! token for a short-lived WebSocket URL, acknowledges each envelope before //! doing work, then posts the completed answer with the bot token. -use super::ingress::{IngressProgress, IngressRequest, ServiceIngress}; +use super::ingress::{ + IngressProgress, IngressReceipt, IngressRequest, ServiceIngress, PENDING_DELIVERY_TTL, +}; use super::{SlackFollowUp, SlackProgress}; use anyhow::{anyhow, Context, Result}; use futures::{SinkExt, StreamExt}; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use std::sync::Arc; use tokio::sync::watch; use tokio_tungstenite::tungstenite::Message; @@ -216,6 +218,10 @@ async fn serve_with_api( cancel: CancellationToken, api: SlackApi, ) -> Result<()> { + // Before taking anything new: finish whatever a previous daemon accepted + // and never answered. Doing this ahead of the socket keeps a reconnect + // from racing a resumed turn for the same thread. + reconcile_outstanding(&ingress, &config, &cancel, &api).await; let mut backoff = std::time::Duration::from_secs(1); loop { if cancel.is_cancelled() { @@ -378,10 +384,21 @@ fn progress_text(progress: &IngressProgress, elapsed: std::time::Duration) -> St } } -/// What the affordance left behind in Slack, so the answer can replace it. -#[derive(Default)] -struct Affordance { +/// Everything needed to finish one delivery in Slack: where the conversation +/// is, and what has already been placed in it on this delivery's behalf. +/// +/// This is what the ingress persists as the delivery's channel context, so a +/// daemon that restarts mid-turn can find the placeholder it left behind +/// instead of adding a second one below it. +#[derive(Clone, Default, Serialize, Deserialize)] +struct DeliveryTrace { + channel: String, + thread_ts: String, + /// The message being answered — where reactions belong. + message_ts: String, + #[serde(default, skip_serializing_if = "Option::is_none")] placeholder_ts: Option, + #[serde(default)] reacted: bool, } @@ -392,37 +409,49 @@ struct Affordance { async fn run_affordance( api: SlackApi, config: SlackConfig, - channel: String, - thread_ts: String, - message_ts: String, + ingress: Arc, + key: String, + mut state: DeliveryTrace, after: std::time::Duration, + already_waited: std::time::Duration, mut progress: watch::Receiver, cancel: CancellationToken, -) -> Affordance { - let mut state = Affordance::default(); +) -> DeliveryTrace { if config.progress == SlackProgress::Off { return state; } // Measured from when the turn was submitted, not from when the placeholder // appeared: the elapsed time this reports is the wait the person in Slack - // has actually had, which started when they hit send. - let started = tokio::time::Instant::now(); - tokio::select! { - _ = cancel.cancelled() => return state, - _ = tokio::time::sleep(after) => {} + // has actually had, which started when they hit send — including, for a + // delivery picked back up after a restart, the part served by the daemon + // that took it. + let now = tokio::time::Instant::now(); + let started = now.checked_sub(already_waited).unwrap_or(now); + // That same earlier daemon may already have put the affordance in the + // channel, so only the unserved remainder of the delay is waited out. + let remaining = after.saturating_sub(already_waited); + if !remaining.is_zero() { + tokio::select! { + _ = cancel.cancelled() => return state, + _ = tokio::time::sleep(remaining) => {} + } } - if config.progress.reacts() { + let mut placed = false; + if config.progress.reacts() && !state.reacted { match api .set_reaction( &config.bot_token, "reactions.add", - &channel, - &message_ts, + &state.channel, + &state.message_ts, WORKING_EMOJI, ) .await { - Ok(_) => state.reacted = true, + Ok(_) => { + state.reacted = true; + placed = true; + } Err(error) => tracing::warn!( %error, "Slack progress reaction failed; the answer is unaffected \ @@ -430,16 +459,26 @@ async fn run_affordance( ), } } - if config.progress.posts_placeholder() { + if config.progress.posts_placeholder() && state.placeholder_ts.is_none() { let text = progress_text(&progress.borrow_and_update().clone(), started.elapsed()); match api - .post_message(&config.bot_token, &channel, &thread_ts, &text) + .post_message(&config.bot_token, &state.channel, &state.thread_ts, &text) .await { - Ok(ts) => state.placeholder_ts = ts, + Ok(ts) => { + state.placeholder_ts = ts; + placed = true; + } Err(error) => tracing::warn!(%error, "Slack progress placeholder failed"), } } + // Record what is now standing in the channel *before* waiting on the turn. + // Anything placed and not recorded is exactly what a restart would strand. + if placed { + ingress + .amend_outstanding(&key, serde_json::to_value(&state).unwrap_or_default()) + .await; + } // Keep the placeholder honest on both counts: a turn that stops at an // approval must stop claiming it is working, and a placeholder that has // said the same words for twenty minutes is indistinguishable from one @@ -462,7 +501,7 @@ async fn run_affordance( }; let text = progress_text(&phase, started.elapsed()); if let Err(error) = api - .update_message(&config.bot_token, &channel, ts, &text) + .update_message(&config.bot_token, &state.channel, ts, &text) .await { tracing::warn!(%error, "Slack progress update failed"); @@ -470,6 +509,74 @@ async fn run_affordance( } } +/// Finish deliveries this channel accepted but never answered. +/// +/// Called once per connection attempt, before any new delivery is taken. A +/// turn interrupted by a restart is usually still running — the harness +/// outlives the daemon and is reattached — so the wait is picked back up +/// rather than declared lost, on what remains of the delivery's original +/// allowance. Only a delivery that cannot be resumed is reported as over. +async fn reconcile_outstanding( + ingress: &Arc, + config: &SlackConfig, + cancel: &CancellationToken, + api: &SlackApi, +) { + for (key, record) in ingress.outstanding().await { + let Ok(trace) = serde_json::from_value::(record.context.clone()) else { + // Written by a version that framed its context differently; there + // is nothing here to find in Slack, so stop tracking it. + ingress.clear_outstanding(&key).await; + continue; + }; + if trace.channel.is_empty() { + ingress.clear_outstanding(&key).await; + continue; + } + let waited = (chrono::Utc::now() - record.submitted_at) + .to_std() + .unwrap_or_default(); + let receipt = ingress.resume_outstanding(&record).await; + let unfinished = match receipt { + None => Some("_The session answering this is gone; the turn never finished._"), + Some(_) if waited >= PENDING_DELIVERY_TTL => { + Some("_The turn was interrupted and did not finish._") + } + Some(_) => None, + }; + if let Some(text) = unfinished { + tracing::info!( + service = %ingress.service_name(), + channel = %ingress.channel_id(), + session = %record.session, + waited_seconds = waited.as_secs(), + "outstanding service delivery cannot be resumed; reporting it" + ); + settle(api, config, &trace, text, false).await; + ingress.clear_outstanding(&key).await; + continue; + } + let Some(receipt) = receipt else { continue }; + let (ingress, config, cancel, api) = + (ingress.clone(), config.clone(), cancel.clone(), api.clone()); + tracing::info!( + service = %ingress.service_name(), + channel = %ingress.channel_id(), + session = %record.session, + waited_seconds = waited.as_secs(), + "resuming a service delivery left outstanding by a restart" + ); + tokio::spawn(async move { + if let Err(error) = + resolve_delivery(&ingress, &config, &cancel, &api, &key, trace, receipt, waited) + .await + { + tracing::warn!(%error, "resumed Slack delivery failed"); + } + }); + } +} + /// The thread the bot was just pulled into, if it is being pulled into one. /// /// Returns `None` — leaving the delivery exactly as it is today — when the @@ -513,7 +620,7 @@ async fn first_engagement_context( } async fn process_delivery( - ingress: &ServiceIngress, + ingress: &Arc, config: &SlackConfig, cancel: &CancellationToken, api: &SlackApi, @@ -528,29 +635,82 @@ async fn process_delivery( Some(context) => format!("{context}\n\n{}", delivery.text), None => delivery.text.clone(), }; + let key = delivery.request_id(); let receipt = ingress .submit_tracked(IngressRequest { message, session_key: Some(session_key), - request_id: Some(delivery.request_id()), + request_id: Some(key.clone()), }) .await?; + let trace = DeliveryTrace { + channel: delivery.channel.clone(), + thread_ts: delivery.thread_ts.clone(), + message_ts: delivery.message_ts.clone(), + placeholder_ts: None, + reacted: false, + }; + // Recorded before the wait begins: from here until it resolves, this + // delivery is recoverable by whichever daemon is running when it does. + ingress + .record_outstanding( + &key, + &receipt, + serde_json::to_value(&trace).unwrap_or_default(), + ) + .await; + resolve_delivery( + ingress, + config, + cancel, + api, + &key, + trace, + receipt, + std::time::Duration::ZERO, + ) + .await +} + +/// Wait out one delivery's turn and put the outcome in the thread. +/// +/// Shared by a delivery that just arrived and one picked back up after a +/// restart; `already_waited` is what the second kind has already spent, and is +/// zero for the first. The delivery stops being outstanding only once +/// something has been said about it. +#[allow(clippy::too_many_arguments)] +async fn resolve_delivery( + ingress: &Arc, + config: &SlackConfig, + cancel: &CancellationToken, + api: &SlackApi, + key: &str, + trace: DeliveryTrace, + receipt: IngressReceipt, + already_waited: std::time::Duration, +) -> Result<()> { let (progress_tx, progress_rx) = watch::channel(IngressProgress::default()); let affordance_cancel = CancellationToken::new(); let affordance = tokio::spawn(run_affordance( api.clone(), config.clone(), - delivery.channel.clone(), - delivery.thread_ts.clone(), - delivery.message_ts.clone(), + ingress.clone(), + key.to_string(), + trace.clone(), PROGRESS_AFTER, + already_waited, progress_rx, affordance_cancel.clone(), )); - let reply = ingress.wait_for_final(&receipt, cancel, &progress_tx).await; + // Surviving a restart must not extend a turn's life, so what is left of + // the original allowance is what it gets. + let budget = PENDING_DELIVERY_TTL.saturating_sub(already_waited); + let reply = ingress + .wait_for_final(&receipt, cancel, &progress_tx, budget) + .await; affordance_cancel.cancel(); - let affordance = affordance.await.unwrap_or_default(); + let trace = affordance.await.unwrap_or(trace); // A turn that failed used to leave the thread silent forever. Now that // something in the channel says "working", it must not keep saying that. @@ -559,27 +719,39 @@ async fn process_delivery( Err(error) => format!("_The turn ended without an answer: {error}_"), }; if cancel.is_cancelled() { + // The channel is stopping, not the turn. Leave the delivery recorded: + // this is precisely the case the record exists for. return Err(anyhow!("channel stopped")); } - match affordance.placeholder_ts.as_deref() { - Some(ts) => { - api.update_message(&config.bot_token, &delivery.channel, ts, &text) - .await?; - } - None => { - api.post_message( - &config.bot_token, - &delivery.channel, - &delivery.thread_ts, - &text, - ) - .await?; - } + settle(api, config, &trace, &text, reply.is_ok()).await; + ingress.clear_outstanding(key).await; + reply.map(|_| ()) +} + +/// Put the outcome where the affordance was, and mark the message with it. +async fn settle( + api: &SlackApi, + config: &SlackConfig, + trace: &DeliveryTrace, + text: &str, + answered: bool, +) { + let posted = match trace.placeholder_ts.as_deref() { + Some(ts) => api + .update_message(&config.bot_token, &trace.channel, ts, text) + .await + .map(|_| ()), + None => api + .post_message(&config.bot_token, &trace.channel, &trace.thread_ts, text) + .await + .map(|_| ()), + }; + if let Err(error) = posted { + tracing::warn!(%error, "Slack outcome delivery failed"); } - if affordance.reacted { - settle_reaction(api, config, &delivery, reply.is_ok()).await; + if trace.reacted { + settle_reaction(api, config, trace, answered).await; } - reply.map(|_| ()) } /// Swap the working reaction for the outcome. Cosmetic: a workspace that @@ -587,15 +759,15 @@ async fn process_delivery( async fn settle_reaction( api: &SlackApi, config: &SlackConfig, - delivery: &SlackDelivery, + trace: &DeliveryTrace, answered: bool, ) { let _ = api .set_reaction( &config.bot_token, "reactions.remove", - &delivery.channel, - &delivery.message_ts, + &trace.channel, + &trace.message_ts, WORKING_EMOJI, ) .await; @@ -608,8 +780,8 @@ async fn settle_reaction( .set_reaction( &config.bot_token, "reactions.add", - &delivery.channel, - &delivery.message_ts, + &trace.channel, + &trace.message_ts, settled, ) .await @@ -1156,10 +1328,11 @@ mod tests { let state = run_affordance( api, config(), - "C1".into(), - "1.1".into(), - "1.1".into(), + test_ingress().await, + "C1:1.1".to_string(), + trace(), PROGRESS_AFTER, + std::time::Duration::ZERO, rx, cancel, ) @@ -1184,10 +1357,11 @@ mod tests { let state = run_affordance( api, config, - "C1".into(), - "1.1".into(), - "1.1".into(), + test_ingress().await, + "C1:1.1".to_string(), + trace(), PROGRESS_AFTER, + std::time::Duration::ZERO, rx, CancellationToken::new(), ) @@ -1197,6 +1371,51 @@ mod tests { assert!(!state.reacted); } + async fn test_ingress() -> Arc { + let shared = super::super::ingress::tests::shared_for_delivery_tests().await; + Arc::new(ServiceIngress::new("C1".to_string(), shared)) + } + + fn trace() -> DeliveryTrace { + DeliveryTrace { + channel: "C1".into(), + thread_ts: "1.1".into(), + message_ts: "1.1".into(), + placeholder_ts: None, + reacted: false, + } + } + + #[test] + fn what_the_channel_left_on_screen_survives_a_round_trip() { + // This is the payload a restart reads back to find the placeholder it + // has to replace. If it does not round-trip, a resumed delivery posts + // a second message below the stale one instead of editing it. + let trace = DeliveryTrace { + channel: "C1".into(), + thread_ts: "111.11".into(), + message_ts: "222.22".into(), + placeholder_ts: Some("P1".into()), + reacted: true, + }; + let encoded = serde_json::to_value(&trace).expect("encode"); + let decoded: DeliveryTrace = serde_json::from_value(encoded).expect("decode"); + assert_eq!(decoded.channel, "C1"); + assert_eq!(decoded.thread_ts, "111.11"); + assert_eq!(decoded.message_ts, "222.22"); + assert_eq!(decoded.placeholder_ts.as_deref(), Some("P1")); + assert!(decoded.reacted); + + // A delivery recorded before the affordance appeared carries neither, + // and must still decode — that is the common case at acceptance. + let bare: DeliveryTrace = serde_json::from_value( + serde_json::json!({"channel": "C1", "thread_ts": "1.1", "message_ts": "1.1"}), + ) + .expect("decode without an affordance"); + assert!(bare.placeholder_ts.is_none()); + assert!(!bare.reacted); + } + /// Stub Slack Web API: answers every method `ok` and records the calls. async fn stub_slack() -> (SlackApi, Arc>>) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -1288,10 +1507,17 @@ mod tests { let affordance = tokio::spawn(run_affordance( api, config, - "C1".into(), - "111.11".into(), - "222.22".into(), + test_ingress().await, + "C1:222.22".to_string(), + DeliveryTrace { + channel: "C1".into(), + thread_ts: "111.11".into(), + message_ts: "222.22".into(), + placeholder_ts: None, + reacted: false, + }, std::time::Duration::from_millis(10), + std::time::Duration::ZERO, rx, cancel.clone(), )); diff --git a/specs/0182-an-accepted-delivery-outlives-the-daemon.md b/specs/0182-an-accepted-delivery-outlives-the-daemon.md new file mode 100644 index 00000000..6c3035a7 --- /dev/null +++ b/specs/0182-an-accepted-delivery-outlives-the-daemon.md @@ -0,0 +1,97 @@ +# 0182-an-accepted-delivery-outlives-the-daemon + +Status: accepted +Date: 2026-08-02 +Area: persistence +Scope: What happens to a service delivery whose turn is still running when the daemon restarts. + +## Decision + +Accepting a delivery is a promise to say something about it. That promise +outlives the daemon process that made it. + +A delivery is recorded the moment it is accepted and stays recorded until +something has been said in the channel about how it ended. The record carries +what a later daemon needs to pick the wait back up — which session, where in +its transcript the turn began, when it was accepted — plus a channel-owned +blob describing whatever the channel already put in front of the person +waiting. + +On starting a channel, every delivery it left outstanding is finished before +any new one is taken: + +**Resume by default.** A harness outlives the daemon and is reattached, so a +turn interrupted by a restart is usually still running. The wait is picked back +up rather than declared lost. + +**On what is left of the original allowance.** Surviving a restart must not +extend a turn's life. A delivery whose allowance is already spent is reported, +not resumed. + +**Report only what cannot be resumed.** A delivery whose session no longer +exists has nothing left to wait on, and says so. + +The channel context is opaque to the ingress. The ingress routes and waits; it +does not render, and a second kind of channel must not require a schema change +in state shared by all of them. + +## Reason + +The task waiting on a turn lives exactly as long as the daemon process. A +restart takes it with it — silently, because nothing is watching the waiter. +Whatever the channel had already placed on behalf of that delivery is then +permanent: a progress placeholder that will never be replaced, a reaction that +will never be settled. No timeout fires, because the thing that would have +timed out is gone. + +This is not a rare edge. Restarts happen on upgrade, on configuration change, +and on operator command, and a service channel accepts deliveries the whole +time. Every restart that lands mid-turn strands one. + +Resuming rather than reporting is what makes the recovery worth having. The +harness did not restart, its work was not lost, and in most cases the answer is +still coming. Declaring the turn dead would throw away a completed turn's +output to report an interruption the person waiting never needed to know about. + +Records are left in place while a delivery is being resolved rather than +consumed on read, so a daemon interrupted *again* mid-recovery finds them a +second time. Resolving one twice edits the message already there; losing one +strands it forever. The asymmetry decides it. + +## Consequences + +- The window between accepting a delivery and resolving it must be fully + covered by the record. Anything a channel places during that window — a + placeholder, a reaction — has to be recorded as it is placed, or a restart + strands exactly that. +- Reconciliation runs before the channel takes new work, so a reconnect cannot + race a resumed turn for the same conversation. +- A resumed delivery inherits its original deadline. Restart loops cannot + extend a turn indefinitely. +- Shared service state is now written on delivery acceptance, not only on + routing changes. +- Records are keyed per channel. Two channels of one service using the same + request id must not collide, and neither may reconcile the other's. +- Recovery is best-effort at the edges: a record whose channel context cannot + be understood is dropped rather than acted on blindly. + +## Non-Goals + +- Surviving anything other than the daemon going away and coming back. A + harness that dies with its session is a different failure. +- Replaying or re-submitting a turn. The delivery is waited on again, never + sent again. +- Guaranteeing exactly-once channel output across repeated restarts mid- + recovery. Editing the same message twice is accepted; stranding it is not. + +## Examples + +A question arrives in a channel and its turn is a long one. Four minutes in, +the daemon is restarted to pick up a new build. The channel comes back, finds +the delivery outstanding, reattaches to the still-running session, and — when +the turn finishes two minutes later — replaces the same placeholder with the +answer. Nobody in the channel learns a restart happened. + +The same, except the session was deleted while the daemon was down. The +placeholder is replaced with a statement that the turn never finished, and the +delivery stops being tracked.