diff --git a/SPEC.md b/SPEC.md index ca9018857e..673c4e1bbb 100644 --- a/SPEC.md +++ b/SPEC.md @@ -8736,7 +8736,9 @@ field unchanged; the bot pins that list against the rules. A kind the extractor does not know yet pages as unclassified until the extractor and the rules learn it. A wrapper alert that embeds a rendered error (a dead-lettered job, a supervised worker's terminal failure) takes the most specific known kind named -in its text, as the extractor does, and falls back to its own kind. The message +in its text, as the extractor does. Each job's terminal failure message carries +its own kind's phrase, so that kind applies when the error names nothing more +specific; the bot pins that every such message holds a known phrase. The message text stays the same. A text log line also renders `kind="..."` after the message, and the extractor takes the rightmost phrase on the line, so a text line now classifies by the field unless a later field names another phrase. This diff --git a/docs/observability.md b/docs/observability.md index b70d8b5cdd..d0cfbb642c 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -231,7 +231,10 @@ stays lost, and so does the `liq_trade` line of a hedge release. Every `operational_alert` line carries a `kind` field, an `AlertKind` from `src/alerts/mod.rs`. Send alerts through `Notifier::notify(kind, message)`. A direct `error!(target: "operational_alert", ...)` line must also log -`kind = AlertKind::...as_str()`; a test fails on one that does not. +`kind = AlertKind::Variant.as_str()` over a plain string literal message. A test +scans every direct line: it fails on a line without a kind, and it checks the +kind against the message literal with the same rules `CapturingNotifier` uses +(below). - An existing kind's string is exactly the label the `operational_alerts` extractor (t0.devops `observability-consumer`) gives the line once that @@ -242,9 +245,21 @@ direct `error!(target: "operational_alert", ...)` line must also log - A new alert gets a new kind in the `not_extracted` group, whose string is a fixed phrase of its message. Teach the extractor that phrase and add a rule; then move the kind into the `extracted` group in the extractor's order. -- `CapturingNotifier` checks every alert a test drives: an extracted kind must - be what the extractor reads from the message, and a not-extracted kind must - match no extracted phrase. +- Every kind's string must appear in its message. `CapturingNotifier` checks + every alert a test drives: an extracted kind must be what the extractor reads + from the message, and the message of a not-extracted kind must hold no + extracted phrase. +- A job's terminal failure page (a dead letter or a supervised worker's terminal + failure) takes its kind from its text: the most specific extracted phrase in + the rendered alert, else `JobFailedAfterRetries`. The job's + `TERMINAL_FAILURE_MSG` decides the kind when the embedded error names nothing + more specific, so an override must carry the phrase of its own extracted kind. + `build_supervised_worker!` and `build_best_effort_worker!` fail the build on a + message with no extracted phrase, and a test checks every + `TERMINAL_FAILURE_MSG` in the source. So a job's terminal page cannot use a + `not_extracted` kind: the rule above for a new alert does not apply to it + until the extractor knows its phrase and the kind moves to the `extracted` + group. The phrase extractor is transitional. Once the `operational_alerts` pager reads `jsonPayload.kind`, a new alert needs only a new `AlertKind` and a rule in diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs index 87c575821f..640755ac69 100644 --- a/src/alerts/mod.rs +++ b/src/alerts/mod.rs @@ -189,6 +189,34 @@ impl AlertKind { } } +/// Whether `text` holds any extracted phrase, that is whether +/// [`AlertKind::most_specific_in`] gives it a kind. +/// +/// A `const fn` so the job worker macros (`build_supervised_worker!`, +/// `build_best_effort_worker!`) can fail the build when a job's +/// `TERMINAL_FAILURE_MSG` names no extracted kind, however the const is +/// written. +pub(crate) const fn names_extracted_kind(text: &str) -> bool { + let text = text.as_bytes(); + let mut kind = 0; + while kind < EXTRACTED.len() { + let phrase = EXTRACTED[kind].as_str().as_bytes(); + let mut start = 0; + while start + phrase.len() <= text.len() { + let mut matched = 0; + while matched < phrase.len() && text[start + matched] == phrase[matched] { + matched += 1; + } + if matched == phrase.len() { + return true; + } + start += 1; + } + kind += 1; + } + false +} + /// Sends an operational alert over some channel. /// /// Kept as a trait so monitors depend on the capability, not the concrete @@ -231,7 +259,7 @@ impl Notifier for LogNotifier { } #[cfg(test)] -pub(crate) use test_support::CapturingNotifier; +pub(crate) use test_support::{CapturingNotifier, assert_kind_matches_extractor}; /// Test-only notifier helpers. Lives in a `#[cfg(test)]` module (rather than /// bare `#[cfg(test)]` items) so clippy's `allow-unwrap-in-tests` applies to the @@ -248,8 +276,9 @@ mod test_support { /// /// Every capture also checks the kind against the extractor: an extracted /// kind must be what the extractor reads from the message, and a kind the - /// extractor does not know must not collide with one it does. So every - /// test that drives an alert path pins its call site's kind. + /// extractor does not know must not collide with one it does and must + /// appear in the message. So every test that drives an alert path pins + /// its call site's kind. #[derive(Default)] pub(crate) struct CapturingNotifier { captured: std::sync::Mutex>, @@ -289,7 +318,12 @@ mod test_support { /// Fails when `kind` is not the label the extractor gives `message`: the /// extracted phrase itself, or, for a kind the extractor does not know, - /// no phrase at all. + /// no phrase at all. A kind the extractor does not know must still carry + /// its own phrase, so the message names the kind it is logged under. + /// + /// Test notifiers that do not capture through [`CapturingNotifier`] call + /// this from their own `notify`, so every alert path a test drives pins + /// its call site's kind. pub(crate) fn assert_kind_matches_extractor(kind: AlertKind, message: &str) { let expected = EXTRACTED.contains(&kind).then_some(kind); assert_eq!( @@ -297,6 +331,10 @@ mod test_support { expected, "alert kind {kind:?} disagrees with the extractor on {message:?}" ); + assert!( + message.contains(kind.as_str()), + "alert kind {kind:?} names a phrase its message lacks: {message:?}" + ); } } @@ -451,6 +489,20 @@ mod tests { regex_kind ); } + + /// The compile-time check the worker macros run agrees with the + /// classifier on whether a text names a kind. + #[test] + fn names_extracted_kind_agrees_with_the_classifier( + parts in prop::collection::vec(phrase_or_noise(), 0..8) + ) { + let text = parts.concat(); + + prop_assert_eq!( + names_extracted_kind(&text), + AlertKind::most_specific_in(&text).is_some() + ); + } } /// The target string is the delivery contract (the downstream metric @@ -483,15 +535,10 @@ mod tests { ); } - /// Every direct `operational_alert` line in the crate carries a `kind` - /// field, as `Notifier::notify` does. - #[test] - fn every_direct_operational_alert_line_carries_a_kind() { - // Built with concat! so this test's own source does not match it. - const DIRECT_ALERT_TARGET: &str = concat!("target: \"", "operational_alert", "\""); - let src = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src"); - let mut pending = vec![src]; - let mut sites = 0; + /// Every `.rs` file under `src`, with its path and contents. + fn crate_sources() -> Vec<(std::path::PathBuf, String)> { + let mut pending = vec![std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("src")]; + let mut sources = Vec::new(); while let Some(path) = pending.pop() { if path.is_dir() { @@ -502,23 +549,254 @@ mod tests { ); continue; } - if path.extension().is_none_or(|extension| extension != "rs") { - continue; + if path.extension().is_some_and(|extension| extension == "rs") { + let source = std::fs::read_to_string(&path).unwrap(); + sources.push((path, source)); + } + } + + sources + } + + /// The value of the Rust string literal whose opening quote is at byte + /// `start` of `source`, and the byte offset just past its closing quote. + /// Folds `\` line continuations and the `\n`, `\t`, `\"` and `\\` + /// escapes; other escapes keep only their letter, and raw strings are not + /// read. + fn string_literal(source: &str, start: usize) -> (String, usize) { + let body_start = start + 1; + let mut value = String::new(); + let mut chars = source[body_start..].char_indices().peekable(); + + while let Some((offset, character)) = chars.next() { + match character { + '"' => return (value, body_start + offset + 1), + '\\' => match chars.next().map(|(_, escaped)| escaped) { + Some('\n') => { + while chars.next_if(|(_, next)| next.is_whitespace()).is_some() {} + } + Some('n') => value.push('\n'), + Some('t') => value.push('\t'), + Some(escaped) => value.push(escaped), + None => break, + }, + other => value.push(other), + } + } + + panic!("unterminated string literal at byte {start}"); + } + + /// The direct `operational_alert` macro invocation whose target starts at + /// byte `target_offset` of `source`: its text from the target to the + /// macro's closing bracket, and its message, the first positional string + /// literal at the macro's top level after the target. A literal after + /// `=`, `= %` or `= ?` is a field value, not the message. Brackets inside + /// string literals are skipped; brackets inside char literals are not. + fn direct_alert_invocation(source: &str, target_offset: usize) -> (&str, String) { + let mut depth = 1_usize; + let mut positional_literals = Vec::new(); + let mut cursor = target_offset; + + while depth > 0 { + let character = source[cursor..].chars().next().unwrap(); + match character { + '"' => { + let (value, end) = string_literal(source, cursor); + let is_field_value = source[..cursor] + .trim_end() + .trim_end_matches(['%', '?']) + .trim_end() + .ends_with('='); + if depth == 1 && !is_field_value { + positional_literals.push(value); + } + cursor = end; + continue; + } + '(' | '[' | '{' => depth += 1, + ')' | ']' | '}' => depth -= 1, + _ => {} } - let source = std::fs::read_to_string(&path).unwrap(); + cursor += character.len_utf8(); + } + + let invocation = &source[target_offset..cursor]; + // The first positional literal is the target string itself. + let message = positional_literals + .into_iter() + .nth(1) + .unwrap_or_else(|| panic!("no message literal in {invocation}")); + (invocation, message) + } + + #[test] + fn direct_alert_invocation_skips_field_values_with_and_without_sigils() { + let target = concat!("target: \"", "operational_alert", "\""); + for field in [ + r#"detail = "Low gas""#, + r#"detail = %"Low gas""#, + r#"detail = ?"Low gas""#, + r#"detail=%"Low gas""#, + ] { + let source = format!( + "error!({target}, alert = true, {field}, \"Portfolio snapshot mark stale\");" + ); + let target_offset = source.find(target).unwrap(); + + let (invocation, message) = direct_alert_invocation(&source, target_offset); + + assert_eq!(message, "Portfolio snapshot mark stale", "for {field}"); + assert!(invocation.ends_with(')'), "for {field}: {invocation}"); + } + } + + /// Every direct `operational_alert` line in the crate carries a `kind` + /// field naming an [`AlertKind`] variant, and that kind agrees with the + /// line's message literal as the extractor reads it, the same check + /// [`CapturingNotifier`] makes on `Notifier::notify`. + /// + /// The scan reads source text, so it has limits: the kind must be written + /// `kind = AlertKind::Variant`, the message must be a plain string literal + /// (not a raw string or a `concat!`), and a char literal holding a bracket + /// would end the invocation early. Runtime arguments interpolated into the + /// message are not seen. `LogNotifier`'s forwarding line (`kind = + /// kind.as_str()` over `"{message}"`) is the one site naming no variant. + #[test] + fn every_direct_operational_alert_line_carries_its_messages_kind() { + // Built with concat! so this test's own source does not match them. + const DIRECT_ALERT_TARGET: &str = concat!("target: \"", "operational_alert", "\""); + const KIND_FIELD: &str = concat!("kind = ", "AlertKind::"); + let mut sites = 0; + + for (path, source) in crate_sources() { for (offset, _) in source.match_indices(DIRECT_ALERT_TARGET) { - let invocation = &source[offset..]; - let invocation = &invocation[..invocation.find(");").unwrap()]; + let (invocation, message) = direct_alert_invocation(&source, offset); sites += 1; + // `LogNotifier` forwards a caller's kind and message. + let forwarded_message = concat!("{", "message", "}"); + if invocation.contains("kind = kind.as_str()") && message == forwarded_message { + continue; + } + let Some(field) = invocation.find(KIND_FIELD) else { + panic!( + "{} has an operational_alert line without a kind: {invocation}", + path.display() + ); + }; + let variant: String = invocation[field + KIND_FIELD.len()..] + .chars() + .take_while(|character| character.is_alphanumeric() || *character == '_') + .collect(); + let kind = EXTRACTED + .iter() + .chain(NOT_EXTRACTED) + .find(|candidate| format!("{candidate:?}") == variant) + .unwrap_or_else(|| { + panic!("{} names unknown AlertKind::{variant}", path.display()) + }); + assert_kind_matches_extractor(*kind, &message); + } + } + + assert!(sites > 20, "found only {sites} direct sites"); + } + + /// The value of each `const TERMINAL_FAILURE_MSG` declaration in + /// `source`, or `None` for one whose value is not a plain string literal. + /// Any type text between the name and the `=` is skipped, so `&str` and + /// `&'static str` both count. + fn terminal_failure_messages(source: &str) -> Vec> { + // Built with concat! so this test's own source does not match it. + const MESSAGE_CONST: &str = concat!("const TERMINAL_FAILURE", "_MSG"); + + source + .match_indices(MESSAGE_CONST) + .filter(|(offset, _)| { + // A declaration, not a mention in a comment or a string. + let line_start = source[..*offset] + .rfind('\n') + .map_or(0, |newline| newline + 1); + let declaration = source[line_start..*offset] + .trim() + .trim_start_matches("pub(crate)") + .trim_start_matches("pub") + .trim() + .is_empty(); + declaration + && source[offset + MESSAGE_CONST.len()..] + .chars() + .next() + .is_some_and(|next| next != '_' && !next.is_alphanumeric()) + }) + .map(|(offset, _)| { + let name_end = offset + MESSAGE_CONST.len(); + let value_start = name_end + source[name_end..].find('=').unwrap() + 1; + source[value_start..] + .find(|character: char| !character.is_whitespace()) + .map(|skipped| value_start + skipped) + .filter(|literal_start| source[*literal_start..].starts_with('"')) + .map(|literal_start| string_literal(source, literal_start).0) + }) + .collect() + } + + #[test] + fn terminal_failure_messages_reads_any_type_spelling() { + let name = concat!("const TERMINAL_FAILURE", "_MSG"); + for declaration in [ + format!("{name}: &'static str = \"Low gas\";"), + format!("{name}: &str = \"Low gas\";"), + format!("{name}:&str=\"Low gas\";"), + format!("{name}: &'static str =\n \"Low gas\";"), + ] { + let messages = terminal_failure_messages(&declaration); + + assert_eq!( + messages, + vec![Some("Low gas".to_string())], + "for {declaration}" + ); + } + + let not_a_literal = format!("{name}: &str = OTHER_MESSAGE;"); + assert_eq!(terminal_failure_messages(¬_a_literal), vec![None]); + + let longer_name = format!("{name}_PREFIX: &str = \"Low gas\";"); + assert!(terminal_failure_messages(&longer_name).is_empty()); + } + + /// Every job's terminal failure message, the trait default and each + /// override, carries an extracted phrase: a terminal job failure page + /// takes its kind from that text (`conductor::job::terminal_failure_kind`), + /// so a message without one would page as the generic job failure while + /// the extractor reads no kind at all. The worker macros check the same + /// at compile time with [`names_extracted_kind`]; this scan also covers + /// jobs no worker macro builds. + #[test] + fn every_terminal_failure_message_names_an_extracted_kind() { + let mut messages = 0; + + for (path, source) in crate_sources() { + for message in terminal_failure_messages(&source) { + let Some(message) = message else { + panic!( + "{} sets TERMINAL_FAILURE_MSG to something other than a string literal", + path.display() + ); + }; + messages += 1; assert!( - invocation.contains("kind = AlertKind::") - || invocation.contains("kind = kind.as_str()"), - "{} has an operational_alert line without a kind: {invocation}", + AlertKind::most_specific_in(&message).is_some(), + "{} has a terminal failure message with no extracted phrase: {message:?}", path.display() ); } } - assert!(sites > 20, "found only {sites} direct sites"); + assert!( + messages >= 4, + "found only {messages} terminal failure messages" + ); } } diff --git a/src/conductor.rs b/src/conductor.rs index 00eef47314..cc49ca1240 100644 --- a/src/conductor.rs +++ b/src/conductor.rs @@ -80,7 +80,9 @@ use crate::cctp_burn::{ CctpBurnOperation, RestoredCctpBurns, bot_cctp_bridge, restore_pending_cctp_burns, }; use crate::conductor::exit::{ConductorExit, ConductorExitError, MonitorTaskError}; -use crate::conductor::job::{BACKPRESSURE_RESCHEDULE_LIMIT, BackpressureStreak}; +use crate::conductor::job::{ + BACKPRESSURE_RESCHEDULE_LIMIT, BackpressureStreak, terminal_failure_kind, +}; use crate::conductor::monitor::order_fills::{CutoffProbe, probe_cutoff_block_support}; use crate::dashboard::equity_price::{EquityPriceStore, MarkListener}; use crate::dashboard::pnl::{LedgerHead, PnlLedger, PnlLedgerReactor}; @@ -1478,10 +1480,9 @@ impl Conductor { MonitorTaskError::TerminalJobFailure { worker, context, - kind, source, } => { - self.alert_terminal_job_failure(worker, context, *kind, source) + self.alert_terminal_job_failure(worker, context, source) .await; } // Not a per-worker failure: the propagated error is the operator @@ -1501,7 +1502,6 @@ impl Conductor { &self, worker: &str, context: &'static str, - kind: AlertKind, source: &BoxDynError, ) { // Deliberately states only what this process controls. The restart @@ -1510,9 +1510,7 @@ impl Conductor { // page an operator with a stale SLA. let alert = format!("st0x-hedge: {worker}: {context}: {source}; process will exit for restart"); - // The most specific known cause in the rendered error names the page, - // as the log extractor reads it; the job's own kind otherwise. - let kind = AlertKind::most_specific_in(&alert).unwrap_or(kind); + let kind = terminal_failure_kind(&alert); match tokio::time::timeout( TERMINAL_FAILURE_ALERT_TIMEOUT, @@ -20752,7 +20750,6 @@ mod tests { Err(MonitorTaskError::TerminalJobFailure { worker: "test-worker-0".to_string(), context: "Job failed after retries", - kind: AlertKind::JobFailedAfterRetries, source, }) }); @@ -20804,7 +20801,6 @@ mod tests { Err(MonitorTaskError::TerminalJobFailure { worker: "test-worker-0".to_string(), context: "Job failed after retries", - kind: AlertKind::JobFailedAfterRetries, source, }) }); @@ -20908,7 +20904,6 @@ mod tests { Err(MonitorTaskError::TerminalJobFailure { worker: "test-worker-0".to_string(), context: "Job failed after retries", - kind: AlertKind::JobFailedAfterRetries, source, }) }); @@ -20958,9 +20953,10 @@ mod tests { impl Notifier for HangingNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, - _message: &str, + kind: crate::alerts::AlertKind, + message: &str, ) -> Result<(), NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); std::future::pending().await } } @@ -20974,9 +20970,10 @@ mod tests { impl Notifier for ErroringNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, - _message: &str, + kind: crate::alerts::AlertKind, + message: &str, ) -> Result<(), NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); Err(NotifierError::Simulated) } } @@ -20996,7 +20993,6 @@ mod tests { Err(MonitorTaskError::TerminalJobFailure { worker: "test-worker-0".to_string(), context: "Job failed after retries", - kind: AlertKind::JobFailedAfterRetries, source, }) }); @@ -21041,7 +21037,6 @@ mod tests { Err(MonitorTaskError::TerminalJobFailure { worker: "test-worker-0".to_string(), context: "Job failed after retries", - kind: AlertKind::JobFailedAfterRetries, source, }) }); @@ -21078,7 +21073,6 @@ mod tests { let error = check_monitor_drain_result(Ok(Err(MonitorTaskError::TerminalJobFailure { worker: "test-worker-0".to_string(), context: "Job failed after retries", - kind: AlertKind::JobFailedAfterRetries, source, }))) .unwrap_err(); diff --git a/src/conductor/builder.rs b/src/conductor/builder.rs index 043bd6c2a9..983a2d96b1 100644 --- a/src/conductor/builder.rs +++ b/src/conductor/builder.rs @@ -1329,7 +1329,6 @@ where TerminalFailureInfo { worker, context, - kind, source, } = failure_notify_for_select.notified(), if !is_draining.is_cancelled() => @@ -1337,7 +1336,6 @@ where Err(MonitorTaskError::TerminalJobFailure { worker, context, - kind, source, }) } diff --git a/src/conductor/exit.rs b/src/conductor/exit.rs index c913675ab1..e3fb15eec1 100644 --- a/src/conductor/exit.rs +++ b/src/conductor/exit.rs @@ -10,15 +10,12 @@ use task_supervisor::SupervisorError; use tokio::task::JoinError; use tracing::{error, info}; -use crate::alerts::AlertKind; - #[derive(Debug, thiserror::Error)] pub(crate) enum MonitorTaskError { #[error("Apalis worker failed after retries: {worker}: {context}")] TerminalJobFailure { worker: String, context: &'static str, - kind: AlertKind, #[source] source: Arc, }, @@ -146,7 +143,6 @@ mod tests { ConductorExit::Monitor(Ok(Err(MonitorTaskError::TerminalJobFailure { worker: "test-worker-0".to_string(), context: "terminal failure", - kind: AlertKind::JobFailedAfterRetries, source: Arc::new(source), }))) .handle() diff --git a/src/conductor/job.rs b/src/conductor/job.rs index 667eef498a..caaad4fa5c 100644 --- a/src/conductor/job.rs +++ b/src/conductor/job.rs @@ -656,13 +656,14 @@ where /// Logged when retries are exhausted and the supervisor receives /// a terminal failure for this job. + /// + /// It is also the alert's sole kind source: the page takes the kind of + /// the most specific extracted phrase in the rendered alert (see + /// [`terminal_failure_kind`]), so an override must carry the phrase of + /// its own extracted [`AlertKind`]. The worker macros fail the build on + /// a message with no extracted phrase. const TERMINAL_FAILURE_MSG: &'static str = "Job failed after retries"; - /// The alert kind of [`TERMINAL_FAILURE_MSG`](Job::TERMINAL_FAILURE_MSG): - /// the kind of its page unless the failure's error names a more specific - /// one. - const TERMINAL_FAILURE_KIND: AlertKind = AlertKind::JobFailedAfterRetries; - /// Upper bound on a single [`perform`](Job::perform) invocation, past /// which the attempt fails instead of parking the worker (RAI-2218). No /// default: every impl states its own bound so the choice is conscious, @@ -750,6 +751,28 @@ macro_rules! build_worker_inner { pub(crate) use build_worker_inner; +/// Fails the build when a job's +/// [`TERMINAL_FAILURE_MSG`](Job::TERMINAL_FAILURE_MSG) carries no extracted +/// [`AlertKind`] phrase. The terminal failure page takes its kind from that +/// text (see [`terminal_failure_kind`]), so a message without one would page +/// as the generic job failure while the `operational_alerts` extractor reads +/// no kind at all. Both worker macros run it, so every production worker is +/// checked however the const is written. +macro_rules! assert_terminal_failure_msg_names_a_kind { + (::<$ctx_type:ty, $job:ty>) => { + const { + assert!( + $crate::alerts::names_extracted_kind( + <$job as $crate::conductor::job::Job<$ctx_type>>::TERMINAL_FAILURE_MSG + ), + "TERMINAL_FAILURE_MSG must contain the phrase of an extracted AlertKind" + ); + } + }; +} + +pub(crate) use assert_terminal_failure_msg_names_a_kind; + /// Builds a `Worker` for a `Job` impl. /// /// Mirrors the `work::` turbofish style: pass the same two @@ -777,6 +800,8 @@ macro_rules! build_supervised_worker { $failure_notify:expr $(, $failure_injector:expr)? $(,)? ) => {{ + $crate::conductor::job::assert_terminal_failure_msg_names_a_kind!(::<$ctx_type, $job>); + build_worker_inner!( ::<$ctx_type, $job>, $index, @@ -785,7 +810,6 @@ macro_rules! build_supervised_worker { $crate::conductor::job::on_terminal_failure( $failure_notify, <$job as $crate::conductor::job::Job<$ctx_type>>::TERMINAL_FAILURE_MSG, - <$job as $crate::conductor::job::Job<$ctx_type>>::TERMINAL_FAILURE_KIND, ) $(, $failure_injector)? ) @@ -821,6 +845,8 @@ macro_rules! build_best_effort_worker { use ::apalis::prelude::WorkerBuilder; use ::apalis_core::worker::ext::event_listener::EventListenerExt; + $crate::conductor::job::assert_terminal_failure_msg_names_a_kind!(::<$ctx_type, $job>); + let builder = WorkerBuilder::new(format!( "{}-{}", <$job as $crate::conductor::job::Job<$ctx_type>>::WORKER_NAME, @@ -845,7 +871,6 @@ macro_rules! build_best_effort_worker { .on_event($crate::conductor::job::on_best_effort_terminal_failure( $notifier, <$job as $crate::conductor::job::Job<$ctx_type>>::TERMINAL_FAILURE_MSG, - <$job as $crate::conductor::job::Job<$ctx_type>>::TERMINAL_FAILURE_KIND, )) .build($crate::conductor::job::work::<$ctx_type, $job>) }}; @@ -1371,7 +1396,6 @@ fn attempt_is_terminal(current_attempt: usize, max_attempts: usize) -> bool { pub(crate) struct TerminalFailureInfo { pub(crate) worker: String, pub(crate) context: &'static str, - pub(crate) kind: AlertKind, pub(crate) source: Arc, } @@ -1447,7 +1471,6 @@ impl TerminalFailureSignal { pub(crate) fn on_terminal_failure( failure_signal: Arc, error_msg: &'static str, - error_kind: AlertKind, ) -> impl Fn(&WorkerContext, &Event) + Send + Sync + 'static { move |ctx, event| { if let Event::Error(err) = event { @@ -1456,7 +1479,6 @@ pub(crate) fn on_terminal_failure( failure_signal.record_and_notify(TerminalFailureInfo { worker, context: error_msg, - kind: error_kind, source: Arc::clone(err), }); let _ = ctx.stop(); @@ -1479,7 +1501,6 @@ pub(crate) fn on_terminal_failure( pub(crate) fn on_best_effort_terminal_failure( notifier: Arc, error_msg: &'static str, - error_kind: AlertKind, ) -> impl Fn(&WorkerContext, &Event) + Send + Sync + 'static { let last_alerted_task = Arc::new(std::sync::Mutex::new(None::)); @@ -1511,9 +1532,7 @@ pub(crate) fn on_best_effort_terminal_failure( let message = format!("st0x-hedge: {worker}: {error_msg}: {err}; worker and conductor will continue"); - // The most specific known cause in the rendered error names the page, - // as the log extractor reads it; the job's own kind otherwise. - let kind = AlertKind::most_specific_in(&message).unwrap_or(error_kind); + let kind = terminal_failure_kind(&message); let notifier = Arc::clone(¬ifier); tokio::spawn(async move { match tokio::time::timeout( @@ -1534,6 +1553,16 @@ pub(crate) fn on_best_effort_terminal_failure( } } +/// The kind of a terminal job failure alert: the most specific extracted +/// phrase in the rendered `alert`, as the log extractor reads it. The alert +/// embeds the job's [`TERMINAL_FAILURE_MSG`](Job::TERMINAL_FAILURE_MSG), which +/// carries its own kind's phrase, so a deeper cause in the rendered error +/// wins and the job's message decides otherwise. The generic job failure kind +/// covers an alert that holds no phrase at all. +pub(crate) fn terminal_failure_kind(alert: &str) -> AlertKind { + AlertKind::most_specific_in(alert).unwrap_or(AlertKind::JobFailedAfterRetries) +} + fn find_job_error<'a>(mut error: &'a (dyn std::error::Error + 'static)) -> Option<&'a JobError> { loop { if let Some(job_error) = error.downcast_ref::() { @@ -1611,7 +1640,6 @@ mod tests { .on_event(on_best_effort_terminal_failure( Arc::new(CapturingNotifier::default()), "Job failed after retries", - AlertKind::JobFailedAfterRetries, )) .build(work::) }); @@ -1632,11 +1660,7 @@ mod tests { #[tokio::test] async fn best_effort_alerts_once_per_task_when_rendered_errors_match() { let notifier = Arc::new(CapturingNotifier::default()); - let handler = on_best_effort_terminal_failure( - notifier.clone(), - "Job failed after retries", - AlertKind::JobFailedAfterRetries, - ); + let handler = on_best_effort_terminal_failure(notifier.clone(), "Job failed after retries"); let ctx = WorkerContext::new::("test-best-effort-worker"); let first = terminal_job_event("first-task"); let second = terminal_job_event("second-task"); @@ -1678,11 +1702,7 @@ mod tests { #[tokio::test] async fn best_effort_alert_takes_the_kind_of_the_most_specific_cause() { let notifier = Arc::new(CapturingNotifier::default()); - let handler = on_best_effort_terminal_failure( - notifier.clone(), - "Job failed after retries", - AlertKind::JobFailedAfterRetries, - ); + let handler = on_best_effort_terminal_failure(notifier.clone(), "Job failed after retries"); let ctx = WorkerContext::new::("test-best-effort-worker"); let error: BoxDynError = Box::new(UnderFundedJobError(JobError::Injected { task_identity: TaskIdentity("under-funded-task".to_owned()), @@ -1701,14 +1721,38 @@ mod tests { assert_eq!(notifier.kinds(), [AlertKind::InventoryVaultUnderFunded]); } - /// The default terminal failure text carries its kind's phrase, so a - /// dead letter pages under that kind when its error names nothing more - /// specific. + /// The default terminal failure text carries the generic job failure + /// phrase, so a dead letter pages under that kind when its error names + /// nothing more specific. #[test] fn default_terminal_failure_message_carries_its_kind() { assert_eq!( AlertKind::most_specific_in(>::TERMINAL_FAILURE_MSG), - Some(>::TERMINAL_FAILURE_KIND) + Some(AlertKind::JobFailedAfterRetries) + ); + } + + /// The rendered error's deeper cause names the page; the job's own + /// message names it otherwise; an alert with no phrase at all pages as + /// the generic job failure. + #[test] + fn terminal_failure_kind_prefers_the_most_specific_phrase() { + assert_eq!( + terminal_failure_kind( + "st0x-hedge: worker-0: Job failed after retries: Raindex vault withdraw \ + failed: inventory vault under-funded" + ), + AlertKind::InventoryVaultUnderFunded + ); + assert_eq!( + terminal_failure_kind( + "st0x-hedge: worker-0: Dashboard trade delivery failed after retries: boom" + ), + AlertKind::DashboardTradeDeliveryFailed + ); + assert_eq!( + terminal_failure_kind("st0x-hedge: worker-0: stress test terminal failure: boom"), + AlertKind::JobFailedAfterRetries ); } @@ -2935,7 +2979,6 @@ mod tests { .on_event(on_terminal_failure( failure_notify.clone(), "stress test terminal failure", - AlertKind::JobFailedAfterRetries, )) .build(work::) }); @@ -3366,7 +3409,6 @@ mod tests { signal.record_and_notify(TerminalFailureInfo { worker: "worker-a".to_string(), context: "context-a", - kind: AlertKind::JobFailedAfterRetries, source: source_a, }); } @@ -3379,7 +3421,6 @@ mod tests { signal.record_and_notify(TerminalFailureInfo { worker: "worker-b".to_string(), context: "context-b", - kind: AlertKind::JobFailedAfterRetries, source: source_b, }); } diff --git a/src/conductor/monitor/gas.rs b/src/conductor/monitor/gas.rs index 4c8baf5976..dee1bb6c9d 100644 --- a/src/conductor/monitor/gas.rs +++ b/src/conductor/monitor/gas.rs @@ -261,7 +261,8 @@ mod tests { use crate::startup::{StartupBarrier, StartupTask}; /// Notifier that records every message it is asked to deliver, so tests can - /// assert exactly which alerts fired. + /// assert exactly which alerts fired, and checks each alert's kind + /// against its message. struct CapturingNotifier { messages: Mutex>, } @@ -282,9 +283,10 @@ mod tests { impl Notifier for CapturingNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, + kind: crate::alerts::AlertKind, message: &str, ) -> Result<(), crate::alerts::NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); self.messages.lock().unwrap().push(message.to_owned()); Ok(()) } diff --git a/src/conductor/monitor/hedge_stall.rs b/src/conductor/monitor/hedge_stall.rs index 298de5351c..92ff9af30d 100644 --- a/src/conductor/monitor/hedge_stall.rs +++ b/src/conductor/monitor/hedge_stall.rs @@ -1586,9 +1586,10 @@ mod tests { impl Notifier for HangingNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, - _message: &str, + kind: crate::alerts::AlertKind, + message: &str, ) -> Result<(), NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); std::future::pending().await } } diff --git a/src/dashboard/event.rs b/src/dashboard/event.rs index fd49101431..0cdd31436f 100644 --- a/src/dashboard/event.rs +++ b/src/dashboard/event.rs @@ -30,7 +30,6 @@ use super::equity_price::EquityPriceStore; use super::event_lines::{ EventId, EventParent, TransferLineSweep, TransferLines, log_event, log_trade, transfer_kind, }; -use crate::alerts::AlertKind; use crate::conductor::job::{Job, JobQueue, Label, QueuePushError}; use crate::equity_redemption::EquityRedemption; use crate::offchain::order::{ @@ -950,7 +949,6 @@ impl Job for DeliverDashboardTrade { Some(crate::conductor::job::DEFAULT_PERFORM_TIMEOUT); const TERMINAL_FAILURE_MSG: &'static str = "Dashboard trade delivery failed after retries; terminal update remains undelivered"; - const TERMINAL_FAILURE_KIND: AlertKind = AlertKind::DashboardTradeDeliveryFailed; #[cfg(any(test, feature = "test-support"))] const JOB_KIND: crate::conductor::job::JobKind = @@ -1489,6 +1487,7 @@ mod tests { use st0x_finance::Usdc; use super::*; + use crate::alerts::AlertKind; use crate::conductor::job::{ FailureInjector, TerminalFailureSignal, build_supervised_worker, build_worker_inner, }; @@ -1511,7 +1510,7 @@ mod tests { AlertKind::most_specific_in( >::TERMINAL_FAILURE_MSG ), - Some(>::TERMINAL_FAILURE_KIND) + Some(AlertKind::DashboardTradeDeliveryFailed) ); } diff --git a/src/portfolio_snapshot/write.rs b/src/portfolio_snapshot/write.rs index 469ebf7627..5e3c62c44f 100644 --- a/src/portfolio_snapshot/write.rs +++ b/src/portfolio_snapshot/write.rs @@ -1440,9 +1440,10 @@ mod tests { impl Notifier for FailOnceNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, - _message: &str, + kind: crate::alerts::AlertKind, + message: &str, ) -> Result<(), NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); if self.attempts.fetch_add(1, Ordering::SeqCst) == 0 { return Err(NotifierError::Simulated); } diff --git a/src/rebalancing/equity/authorization_job.rs b/src/rebalancing/equity/authorization_job.rs index 365e5aabdf..9552bfaa89 100644 --- a/src/rebalancing/equity/authorization_job.rs +++ b/src/rebalancing/equity/authorization_job.rs @@ -115,7 +115,6 @@ impl Job for DeliverMintAuthorization { const TERMINAL_FAILURE_MSG: &'static str = "Mint authorization delivery failed all retries; the orchestrator-mode \ mint cannot proceed until the authorization reaches issuance. \ Operator action required."; - const TERMINAL_FAILURE_KIND: AlertKind = AlertKind::MintAuthorizationDeliveryFailed; #[cfg(any(test, feature = "test-support"))] const JOB_KIND: crate::conductor::job::JobKind = @@ -402,7 +401,7 @@ mod tests { fn terminal_failure_message_carries_its_kind() { assert_eq!( AlertKind::most_specific_in(>::TERMINAL_FAILURE_MSG), - Some(>::TERMINAL_FAILURE_KIND) + Some(AlertKind::MintAuthorizationDeliveryFailed) ); } diff --git a/src/rebalancing/equity/resume_job.rs b/src/rebalancing/equity/resume_job.rs index 8a499c1183..e6479aa120 100644 --- a/src/rebalancing/equity/resume_job.rs +++ b/src/rebalancing/equity/resume_job.rs @@ -33,7 +33,7 @@ use super::{ WITHDRAWAL_RECONCILIATION_REDRIVE_DELAY, issuer_send_redrive, page_underlying_mismatch, withdrawal_reconciliation_redrive_delay, }; -use crate::alerts::{AlertKind, Notifier}; +use crate::alerts::Notifier; #[cfg(test)] use crate::bot_gas::BotGasReceiptCostEnqueuer; use crate::bot_gas::redrive::{BotGasFailureClassifier, redrive_on_bot_gas_failure}; @@ -164,7 +164,6 @@ impl Job for ResumeTokenizationAggregate { const PERFORM_TIMEOUT: Option = None; const TERMINAL_FAILURE_MSG: &'static str = "Interrupted tokenization aggregate failed all resume retries; \ the aggregate remains stuck. Operator action required."; - const TERMINAL_FAILURE_KIND: AlertKind = AlertKind::TokenizationResumeFailed; #[cfg(any(test, feature = "test-support"))] const JOB_KIND: crate::conductor::job::JobKind = @@ -432,6 +431,7 @@ mod tests { use std::collections::{BTreeMap, HashMap}; use super::*; + use crate::alerts::AlertKind; use crate::equity_redemption::{ EquityRedemption, EquityRedemptionCommand, redemption_aggregate_id, }; @@ -451,9 +451,7 @@ mod tests { AlertKind::most_specific_in( >::TERMINAL_FAILURE_MSG ), - Some( - >::TERMINAL_FAILURE_KIND - ) + Some(AlertKind::TokenizationResumeFailed) ); } diff --git a/src/rebalancing/trigger/mod.rs b/src/rebalancing/trigger/mod.rs index 57129a8dfd..b327ae9742 100644 --- a/src/rebalancing/trigger/mod.rs +++ b/src/rebalancing/trigger/mod.rs @@ -31736,9 +31736,10 @@ mod tests { impl crate::alerts::Notifier for FlakyNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, + kind: crate::alerts::AlertKind, message: &str, ) -> Result<(), crate::alerts::NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); let should_fail = self .remaining_failures .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |remaining| { diff --git a/src/rebalancing/usdc/job.rs b/src/rebalancing/usdc/job.rs index 8918e31581..8c78f9eaa9 100644 --- a/src/rebalancing/usdc/job.rs +++ b/src/rebalancing/usdc/job.rs @@ -6095,9 +6095,10 @@ mod tests { impl crate::alerts::Notifier for FailingNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, - _message: &str, + kind: crate::alerts::AlertKind, + message: &str, ) -> Result<(), crate::alerts::NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); Err(crate::alerts::NotifierError::Simulated) } } diff --git a/src/trading/offchain/hedge.rs b/src/trading/offchain/hedge.rs index b5376670a7..a21ca3626b 100644 --- a/src/trading/offchain/hedge.rs +++ b/src/trading/offchain/hedge.rs @@ -2385,9 +2385,10 @@ mod tests { impl crate::alerts::Notifier for FlakyNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, + kind: crate::alerts::AlertKind, message: &str, ) -> Result<(), crate::alerts::NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); if self.failing.load(Ordering::SeqCst) { return Err(crate::alerts::NotifierError::Simulated); } @@ -2408,9 +2409,10 @@ mod tests { impl crate::alerts::Notifier for PausingNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, - _message: &str, + kind: crate::alerts::AlertKind, + message: &str, ) -> Result<(), crate::alerts::NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); self.calls.fetch_add(1, Ordering::SeqCst); self.started.notify_one(); self.release.notified().await; @@ -2427,9 +2429,10 @@ mod tests { impl crate::alerts::Notifier for HangingNotifier { async fn notify( &self, - _kind: crate::alerts::AlertKind, - _message: &str, + kind: crate::alerts::AlertKind, + message: &str, ) -> Result<(), crate::alerts::NotifierError> { + crate::alerts::assert_kind_matches_extractor(kind, message); self.calls.fetch_add(1, Ordering::SeqCst); std::future::pending().await } @@ -3761,18 +3764,26 @@ mod tests { ); } + /// A synthetic dead-letter page carrying the standing-delta phrase its + /// symbol-scoped reason pages under, told apart by `label`. + fn standing_delta_page(label: &str) -> String { + format!("{label}: the symbol carries a standing delta") + } + #[tokio::test] async fn concurrent_dead_letter_pages_reserve_one_delivery_slot() { let notifier = PausingNotifier::default(); let alerted = Mutex::new(HashSet::new()); let symbol = Symbol::new("AAPL").unwrap(); let reason = DeadLetterReason::SymbolScoped(SymbolScopedReason::LimitQuoteFetch); + let first = standing_delta_page("first"); + let second = standing_delta_page("second"); tokio::join!( - alert_dead_letter(¬ifier, &alerted, &symbol, reason, "first"), + alert_dead_letter(¬ifier, &alerted, &symbol, reason, &first), async { notifier.started.notified().await; - alert_dead_letter(¬ifier, &alerted, &symbol, reason, "second").await; + alert_dead_letter(¬ifier, &alerted, &symbol, reason, &second).await; notifier.release.notify_one(); } ); @@ -3795,9 +3806,10 @@ mod tests { let symbol = Symbol::new("AAPL").unwrap(); let reason = DeadLetterReason::SymbolScoped(SymbolScopedReason::LimitQuoteFetch); let key = (symbol.clone(), reason); + let page = standing_delta_page("page"); tokio::join!( - alert_dead_letter(¬ifier, &alerted, &symbol, reason, "page"), + alert_dead_letter(¬ifier, &alerted, &symbol, reason, &page), async { notifier.started.notified().await; assert!( @@ -3820,9 +3832,10 @@ mod tests { let alerted = Mutex::new(HashSet::new()); let symbol = Symbol::new("AAPL").unwrap(); let reason = DeadLetterReason::SymbolScoped(SymbolScopedReason::LimitQuoteFetch); + let page = standing_delta_page("page"); for _ in 0..2 { - alert_dead_letter(¬ifier, &alerted, &symbol, reason, "page").await; + alert_dead_letter(¬ifier, &alerted, &symbol, reason, &page).await; } assert_eq!(