diff --git a/crates/services/src/inbuxa_relay_link.rs b/crates/services/src/inbuxa_relay_link.rs index 5f78fb4c..a5e824be 100644 --- a/crates/services/src/inbuxa_relay_link.rs +++ b/crates/services/src/inbuxa_relay_link.rs @@ -21,6 +21,11 @@ //! answered, a held one is delivered, or reported to a sender who passed //! SPF or DMARC (RL-17, RL-18). A message taken is recorded before the //! relay is told, so one offered again isn't delivered twice (RL-16). +//! +//! The relay's own events become this server's `relay.*` trace events, so +//! they reach the log, webhooks and OpenTelemetry. On a cluster only the +//! linked node with the lowest id takes them. A node with no link to a +//! joined relay for an hour says so with `relay.link-down` (RL-T23). use common::{ BuildServer, Inner, Server, @@ -49,8 +54,8 @@ use inbuxa_relay_proto::{ recipient::{RecipientDomain, RecipientHash}, seal, token, types::{ - Address, Capability, Inbound, InboundState, RELAY_SINGLETON, RelaySettings, RelayStatus, - Reply, Verdict, + Address, Capability, Inbound, InboundState, RELAY_SINGLETON, RelayEvent as ProtoRelayEvent, + RelaySettings, RelayStatus, Reply, Verdict, }, }; use registry::schema::{ @@ -107,6 +112,8 @@ const TAKE_MAX: usize = 8; const LEASE_SECONDS: u64 = 300; /// How often messages taken long ago are forgotten (RL-16). const FORGET_EVERY: u64 = 3_600; +/// No link this long to a joined relay raises `relay.link-down`. +const LINK_DOWN_ALERT: u64 = 3_600; pub fn spawn_relay_links(inner: Arc) { tokio::spawn(async move { @@ -343,6 +350,9 @@ async fn run_link( wait } }; + if relay.join.state(now()) == JoinState::Joined { + status.alert_if_down(relay.relay_id(), &relay.settings.hostname); + } if !wait.is_zero() { tokio::select! { _ = tokio::time::sleep(wait) => {} @@ -368,6 +378,10 @@ async fn link_keys( struct StatusWriter { current: NodeLink, written_at: u64, + /// Since when this node has had no link, if it hasn't. + unlinked_since: Option, + /// `relay.link-down` was raised for this outage, for this relay. + alerted: Option<(String, String)>, } impl StatusWriter { @@ -378,6 +392,26 @@ impl StatusWriter { ..Default::default() }, written_at: 0, + unlinked_since: None, + alerted: None, + } + } + + /// RL-T23: an hour with no link to a joined relay is said once, and + /// so is its end. + fn alert_if_down(&mut self, relay_id: String, hostname: &str) { + let Some(since) = self.unlinked_since else { + return; + }; + if self.alerted.is_none() && now() >= since + LINK_DOWN_ALERT { + trc::event!( + Relay(trc::RelayEvent::LinkDown), + Id = relay_id.clone(), + Hostname = hostname.to_string(), + Reason = self.current.last_error.clone().unwrap_or_default(), + Elapsed = trc::Value::Duration(now().saturating_sub(since) * 1000), + ); + self.alerted = Some((relay_id, hostname.to_string())); } } @@ -392,6 +426,14 @@ impl StatusWriter { if state != self.current.state { self.current.since = now; } + if state == LinkState::Linked { + self.unlinked_since = None; + if let Some((id, hostname)) = self.alerted.take() { + trc::event!(Relay(trc::RelayEvent::LinkUp), Id = id, Hostname = hostname); + } + } else if self.unlinked_since.is_none() { + self.unlinked_since = Some(now); + } let changed = state != self.current.state || error != self.current.last_error; self.current.state = state; if error.is_some() || state == LinkState::Linked { @@ -730,6 +772,7 @@ struct Ws { next_id: u64, relay_changed: bool, inbound_changed: bool, + events_changed: bool, } impl Ws { @@ -760,6 +803,7 @@ impl Ws { next_id: 0, relay_changed: false, inbound_changed: false, + events_changed: false, }) } @@ -795,6 +839,7 @@ impl Ws { for types in changed.values() { self.relay_changed |= types.contains_key("Relay"); self.inbound_changed |= types.contains_key("Inbound"); + self.events_changed |= types.contains_key("RelayEvent"); } } } @@ -891,7 +936,7 @@ async fn session( let mut ws = Ws::open(&dialer).await?; // Inbound too: live mode (RL-12) ws.send(&ClientFrame::WebSocketPushEnable { - data_types: Some(vec!["Relay".into(), "Inbound".into()]), + data_types: Some(vec!["Relay".into(), "Inbound".into(), "RelayEvent".into()]), push_state: None, }) .await?; @@ -913,6 +958,7 @@ async fn session( answers: answers_tx, }; taker.take(&mut ws).await?; + take_events(&mut ws, data, &relay, status.current.node_id).await?; let mut tick = tokio::time::interval(SYNC); tick.reset(); @@ -927,6 +973,9 @@ async fn session( if ws.inbound_changed { taker.take(&mut ws).await?; } + if ws.events_changed { + take_events(&mut ws, data, &relay, status.current.node_id).await?; + } continue; } Some(answer) = answers.recv() => { @@ -957,6 +1006,108 @@ async fn session( refresh_status(&mut ws, data, &relay, &sent, status).await?; taker.relay_name = relay.name.clone(); taker.take(&mut ws).await?; + take_events(&mut ws, data, &relay, status.current.node_id).await?; + } +} + +// ---- Relay events ----------------------------------------------------------- + +/// The relay's events, as this server's trace events, then acknowledged so +/// the relay can forget them. One node takes them: the lowest linked one. +async fn take_events( + ws: &mut Ws, + data: &store::Store, + relay: &Relay, + node_id: u64, +) -> Result<(), LinkError> { + ws.events_changed = false; + let first = relays::link::statuses(data, relay.id) + .await? + .iter() + .filter(|link| { + link.state == LinkState::Linked && link.updated_at + 3 * STATUS_EVERY >= now() + }) + .map(|link| link.node_id) + .min() + .is_none_or(|first| first >= node_id); + if !first { + return Ok(()); + } + let answer = ws.ok(method::RELAY_EVENT_GET, json!({})).await?; + let mut ids = Vec::new(); + for event in answer["list"].as_array().into_iter().flatten() { + if let Ok(event) = serde_json::from_value::(event.clone()) { + trace_event(relay, &event); + ids.push(event.id); + } + } + if !ids.is_empty() { + ws.ok(method::RELAY_EVENT_SET, json!({"destroy": ids})) + .await?; + } + Ok(()) +} + +/// One relay event, as a `relay.*` trace event naming the relay. +fn trace_event(relay: &Relay, event: &ProtoRelayEvent) { + let id = relay.relay_id(); + let hostname = relay.settings.hostname.clone(); + let data = &event.data; + let text = |key: &str| { + data.get(key) + .and_then(Value::as_str) + .unwrap_or_default() + .to_string() + }; + let number = |key: &str| data.get(key).and_then(Value::as_u64).unwrap_or_default(); + match event.kind.as_str() { + "smtp.session" => trc::event!( + Relay(trc::RelayEvent::Session), + Id = id, + Hostname = hostname, + RemoteIp = text("remoteIp"), + Tls = data.get("tls").and_then(Value::as_bool).unwrap_or_default(), + Total = number("accepted"), + TotalFailures = number("refusedRecipients"), + ), + "inbound.accepted" => trc::event!( + Relay(trc::RelayEvent::MessageAccepted), + Id = id, + Hostname = hostname, + Details = text("id"), + Type = text("mode"), + Size = number("size"), + ), + "inbound.refused" => trc::event!( + Relay(trc::RelayEvent::MessageRefused), + Id = id, + Hostname = hostname, + Details = text("id"), + Code = number("code"), + ), + "inbound.expired" => trc::event!( + Relay(trc::RelayEvent::MessageExpired), + Id = id, + Hostname = hostname, + Details = text("id"), + Value = text("receivedAt"), + ), + "relay.joined" => { + trc::event!(Relay(trc::RelayEvent::Joined), Id = id, Hostname = hostname,) + } + "recipients.replaced" => trc::event!( + Relay(trc::RelayEvent::RecipientsReplaced), + Id = id, + Hostname = hostname, + Total = number("count"), + ), + kind => trc::event!( + Relay(trc::RelayEvent::Event), + Id = id, + Hostname = hostname, + Type = kind.to_string(), + Details = Value::Object(data.clone()).to_string(), + ), } } @@ -1037,7 +1188,7 @@ impl Taker { let answers = self.answers.clone(); tokio::spawn(async move { let id = inbound.id.clone(); - let action = job.judge(inbound).await; + let action = Box::pin(job.judge(inbound)).await; let _ = answers.send(Answer { id, action }); }); } @@ -1106,7 +1257,12 @@ impl Job { }; let input = relayed_message(&self.relay_name, &inbound, !live, message); - let outcome = ingest_relayed(self.server.clone(), self.instance.clone(), input).await; + let outcome = Box::pin(ingest_relayed( + self.server.clone(), + self.instance.clone(), + input, + )) + .await; let action = action_for(&outcome, live); if action == Action::Destroy && let Err(err) = record_ingested(&data, self.relay_id, &inbound.id, now()).await @@ -1497,6 +1653,25 @@ mod tests { assert!(verdict_for("550 5.1.1").is_valid()); } + #[test] + fn link_down_alert() { + let alert = |status: &mut StatusWriter| status.alert_if_down("a".into(), "mx1.example.com"); + let mut status = StatusWriter::new(7); + // Never unlinked, or not for an hour yet: nothing + alert(&mut status); + assert!(status.alerted.is_none()); + status.unlinked_since = Some(now() - LINK_DOWN_ALERT + 60); + alert(&mut status); + assert!(status.alerted.is_none()); + // An hour: once + status.unlinked_since = Some(now() - LINK_DOWN_ALERT); + alert(&mut status); + let first = status.alerted.clone(); + assert!(first.is_some()); + alert(&mut status); + assert_eq!(status.alerted, first); + } + #[test] fn actions() { let accepted = RelayedOutcome::Accepted { diff --git a/crates/trc/src/event/enums.rs b/crates/trc/src/event/enums.rs index fb1e3ed4..3c48b0b5 100644 --- a/crates/trc/src/event/enums.rs +++ b/crates/trc/src/event/enums.rs @@ -12,10 +12,11 @@ // auth.legacy-protocol-refused (legacy-protocols LP-6); 643 is // security.legacy-protocols-changed (LP-8); 644 to 646 are the cluster // coordinator's connection events; 647 and 648 are the audit log's -// (audit-hold-lock spec, AU-3, AU-8); 649 is cluster.node-id-reassigned. +// (audit-hold-lock spec, AU-3, AU-8); 649 is cluster.node-id-reassigned; +// 650 to 658 are inbuxa Relay's (relay spec, RL-23, RL-T23). // TOTAL_EVENT_COUNT sizes the collector's per-event tables, so it is one // more than the highest id; the test below holds it to the variant list -pub const TOTAL_EVENT_COUNT: usize = 650; +pub const TOTAL_EVENT_COUNT: usize = 659; pub const TOTAL_METRIC_COUNT: usize = 369; #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] @@ -50,6 +51,8 @@ pub enum EventType { PushSubscription(PushSubscriptionEvent), Queue(QueueEvent), Registry(RegistryEvent), + // inbuxa: relay spec + Relay(RelayEvent), Resource(ResourceEvent), // inbuxa: SCIM-54 Scim(ScimEvent), @@ -642,6 +645,22 @@ pub enum ResourceEvent { ApplicationUnpacked = 602, } +// inbuxa: relay spec: what an inbuxa Relay reports, mapped by the link +// service, and the link-down alert (RL-T23) +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +#[repr(u16)] +pub enum RelayEvent { + Session = 650, + MessageAccepted = 651, + MessageRefused = 652, + MessageExpired = 653, + Joined = 654, + RecipientsReplaced = 655, + Event = 656, + LinkDown = 657, + LinkUp = 658, +} + // inbuxa: SCIM-54: every write an identity provider makes #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] #[repr(u16)] diff --git a/crates/trc/src/event/enums_impl.rs b/crates/trc/src/event/enums_impl.rs index d198d179..18cfa2e1 100644 --- a/crates/trc/src/event/enums_impl.rs +++ b/crates/trc/src/event/enums_impl.rs @@ -38,6 +38,16 @@ impl EventType { b"acme.error" => EventType::Acme(AcmeEvent::Error), b"ai.llm-response" => EventType::Ai(AiEvent::LlmResponse), b"ai.api-error" => EventType::Ai(AiEvent::ApiError), + // inbuxa: relay spec + b"relay.session" => EventType::Relay(RelayEvent::Session), + b"relay.message-accepted" => EventType::Relay(RelayEvent::MessageAccepted), + b"relay.message-refused" => EventType::Relay(RelayEvent::MessageRefused), + b"relay.message-expired" => EventType::Relay(RelayEvent::MessageExpired), + b"relay.joined" => EventType::Relay(RelayEvent::Joined), + b"relay.recipients-replaced" => EventType::Relay(RelayEvent::RecipientsReplaced), + b"relay.event" => EventType::Relay(RelayEvent::Event), + b"relay.link-down" => EventType::Relay(RelayEvent::LinkDown), + b"relay.link-up" => EventType::Relay(RelayEvent::LinkUp), // inbuxa: SCIM-54 b"scim.resource-created" => EventType::Scim(ScimEvent::ResourceCreated), b"scim.resource-updated" => EventType::Scim(ScimEvent::ResourceUpdated), @@ -699,6 +709,16 @@ impl EventType { EventType::Acme(AcmeEvent::Error) => "acme.error", EventType::Ai(AiEvent::LlmResponse) => "ai.llm-response", EventType::Ai(AiEvent::ApiError) => "ai.api-error", + // inbuxa: relay spec + EventType::Relay(RelayEvent::Session) => "relay.session", + EventType::Relay(RelayEvent::MessageAccepted) => "relay.message-accepted", + EventType::Relay(RelayEvent::MessageRefused) => "relay.message-refused", + EventType::Relay(RelayEvent::MessageExpired) => "relay.message-expired", + EventType::Relay(RelayEvent::Joined) => "relay.joined", + EventType::Relay(RelayEvent::RecipientsReplaced) => "relay.recipients-replaced", + EventType::Relay(RelayEvent::Event) => "relay.event", + EventType::Relay(RelayEvent::LinkDown) => "relay.link-down", + EventType::Relay(RelayEvent::LinkUp) => "relay.link-up", // inbuxa: SCIM-54 EventType::Scim(ScimEvent::ResourceCreated) => "scim.resource-created", EventType::Scim(ScimEvent::ResourceUpdated) => "scim.resource-updated", @@ -1501,6 +1521,16 @@ impl EventType { EventType::Acme(AcmeEvent::Error) => 15, EventType::Ai(AiEvent::LlmResponse) => 556, EventType::Ai(AiEvent::ApiError) => 557, + // inbuxa: relay spec + EventType::Relay(RelayEvent::Session) => 650, + EventType::Relay(RelayEvent::MessageAccepted) => 651, + EventType::Relay(RelayEvent::MessageRefused) => 652, + EventType::Relay(RelayEvent::MessageExpired) => 653, + EventType::Relay(RelayEvent::Joined) => 654, + EventType::Relay(RelayEvent::RecipientsReplaced) => 655, + EventType::Relay(RelayEvent::Event) => 656, + EventType::Relay(RelayEvent::LinkDown) => 657, + EventType::Relay(RelayEvent::LinkUp) => 658, // inbuxa: SCIM-54 EventType::Scim(ScimEvent::ResourceCreated) => 637, EventType::Scim(ScimEvent::ResourceUpdated) => 638, @@ -2161,6 +2191,16 @@ impl EventType { 15 => Some(EventType::Acme(AcmeEvent::Error)), 556 => Some(EventType::Ai(AiEvent::LlmResponse)), 557 => Some(EventType::Ai(AiEvent::ApiError)), + // inbuxa: relay spec + 650 => Some(EventType::Relay(RelayEvent::Session)), + 651 => Some(EventType::Relay(RelayEvent::MessageAccepted)), + 652 => Some(EventType::Relay(RelayEvent::MessageRefused)), + 653 => Some(EventType::Relay(RelayEvent::MessageExpired)), + 654 => Some(EventType::Relay(RelayEvent::Joined)), + 655 => Some(EventType::Relay(RelayEvent::RecipientsReplaced)), + 656 => Some(EventType::Relay(RelayEvent::Event)), + 657 => Some(EventType::Relay(RelayEvent::LinkDown)), + 658 => Some(EventType::Relay(RelayEvent::LinkUp)), // inbuxa: SCIM-54 637 => Some(EventType::Scim(ScimEvent::ResourceCreated)), 638 => Some(EventType::Scim(ScimEvent::ResourceUpdated)), @@ -3143,6 +3183,16 @@ impl EventType { EventType::Acme(AcmeEvent::TlsAlpnError) => Level::Warn, EventType::Acme(AcmeEvent::TokenNotFound) => Level::Warn, EventType::Ai(AiEvent::ApiError) => Level::Warn, + // inbuxa: relay spec + EventType::Relay(RelayEvent::Session) => Level::Info, + EventType::Relay(RelayEvent::MessageAccepted) => Level::Info, + EventType::Relay(RelayEvent::MessageRefused) => Level::Info, + EventType::Relay(RelayEvent::MessageExpired) => Level::Warn, + EventType::Relay(RelayEvent::Joined) => Level::Info, + EventType::Relay(RelayEvent::RecipientsReplaced) => Level::Debug, + EventType::Relay(RelayEvent::Event) => Level::Info, + EventType::Relay(RelayEvent::LinkDown) => Level::Warn, + EventType::Relay(RelayEvent::LinkUp) => Level::Info, // inbuxa: SCIM-54 EventType::Scim(ScimEvent::ResourceCreated) => Level::Info, EventType::Scim(ScimEvent::ResourceUpdated) => Level::Info, @@ -3234,6 +3284,16 @@ impl EventType { EventType::Acme(AcmeEvent::Error) => "ACME error", EventType::Ai(AiEvent::LlmResponse) => "LLM response", EventType::Ai(AiEvent::ApiError) => "AI API error", + // inbuxa: relay spec + EventType::Relay(RelayEvent::Session) => "Relay SMTP session ended", + EventType::Relay(RelayEvent::MessageAccepted) => "Relay accepted a message", + EventType::Relay(RelayEvent::MessageRefused) => "Relay refused a message", + EventType::Relay(RelayEvent::MessageExpired) => "Relay held mail expired", + EventType::Relay(RelayEvent::Joined) => "Relay joined", + EventType::Relay(RelayEvent::RecipientsReplaced) => "Relay recipient list replaced", + EventType::Relay(RelayEvent::Event) => "Relay event", + EventType::Relay(RelayEvent::LinkDown) => "Relay link down for an hour", + EventType::Relay(RelayEvent::LinkUp) => "Relay link back", // inbuxa: SCIM-54 EventType::Scim(ScimEvent::ResourceCreated) => "SCIM resource created", EventType::Scim(ScimEvent::ResourceUpdated) => "SCIM resource updated", @@ -4339,6 +4399,16 @@ impl EventType { EventType::Acme(AcmeEvent::Error), EventType::Ai(AiEvent::LlmResponse), EventType::Ai(AiEvent::ApiError), + // inbuxa: relay spec + EventType::Relay(RelayEvent::Session), + EventType::Relay(RelayEvent::MessageAccepted), + EventType::Relay(RelayEvent::MessageRefused), + EventType::Relay(RelayEvent::MessageExpired), + EventType::Relay(RelayEvent::Joined), + EventType::Relay(RelayEvent::RecipientsReplaced), + EventType::Relay(RelayEvent::Event), + EventType::Relay(RelayEvent::LinkDown), + EventType::Relay(RelayEvent::LinkUp), // inbuxa: SCIM-54 EventType::Scim(ScimEvent::ResourceCreated), EventType::Scim(ScimEvent::ResourceUpdated), diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index a522edf8..4f03c258 100644 --- a/docs/spec/features/relay.md +++ b/docs/spec/features/relay.md @@ -690,8 +690,16 @@ In progress. Done so far: is retried after the lease, and spam over the reject threshold goes to Junk (RL-17). Every message taken is recorded until `heldMaxAge` and a day have passed (RL-16). A pending message stays `pending` when claimed. -- Not done yet: `inbuxa:RelayMessage`, relay events in the trace, and the - link-down alert. +- Relay events: the link service asks for `RelayEvent` push too, and the + linked node with the lowest id turns each event into a `relay.*` trace + event (`relay.session`, `relay.message-accepted`, `-refused`, + `-expired`, `relay.joined`, `relay.recipients-replaced`, and + `relay.event` for any other type), then acknowledges it. A node with no + link to a joined relay for an hour raises `relay.link-down` (a warning), + once, and `relay.link-up` when it is back (RL-T23). Both reach the log, + webhooks and OpenTelemetry. +- Not done yet: `inbuxa:RelayMessage`, which comes with inbuxa-admin's + Relays page (step 5), once `Outbound` exists to show beside `Inbound`. Order: diff --git a/resources/schema/schema.json.gz b/resources/schema/schema.json.gz index d1e2a536..4fd5e892 100644 Binary files a/resources/schema/schema.json.gz and b/resources/schema/schema.json.gz differ diff --git a/resources/schema/schema.json.sha256 b/resources/schema/schema.json.sha256 index 1c23e561..ab036e1b 100644 --- a/resources/schema/schema.json.sha256 +++ b/resources/schema/schema.json.sha256 @@ -1 +1 @@ -B5ARrqKFTX10Ib2fIAsGacr8yxF4NyI3IwmJEPFiTP8 \ No newline at end of file +D6Ad4OELCzUv1LCDwL_v0XRNhdN-2jaO4HuYVRXzxu4 \ No newline at end of file diff --git a/tests/src/system/relay.rs b/tests/src/system/relay.rs index 2330e114..49e8c064 100644 --- a/tests/src/system/relay.rs +++ b/tests/src/system/relay.rs @@ -16,7 +16,8 @@ //! port 25 takes mail only for this server's real recipients. Live, a //! message is delivered before the sender hears 250, and this server's //! refusal is the sender's reply; held while the link is off, it is -//! delivered once the link is back, exactly once. +//! delivered once the link is back, exactly once. The relay's events are +//! taken and acknowledged. use crate::utils::{ account::Account, @@ -730,6 +731,11 @@ pub async fn link(test: &mut TestServer) { // message is delivered once the link is back, once let reply = send_mail(smtp_port, "jane@example.com", &message("Held one", "")).await; assert!(reply.starts_with("250"), "{reply}"); + let pending_events = std::fs::read_to_string(state.join("events.jsonl")).unwrap_or_default(); + assert!( + pending_events.contains("inbound.accepted"), + "the relay recorded the held message: {pending_events}" + ); call( admin, "inbuxa:Relay/set", @@ -751,6 +757,26 @@ pub async fn link(test: &mut TestServer) { 2 ); + // The relay's events were taken as this server's relay.* trace events, + // and acknowledged, so the relay forgot them + let events = state.join("events.jsonl"); + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); + loop { + let left = std::fs::read_to_string(&events) + .unwrap_or_default() + .lines() + .filter(|line| !line.trim().is_empty()) + .count(); + if left == 0 { + break; + } + assert!( + std::time::Instant::now() < deadline, + "{left} relay events left unacknowledged" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + call(admin, "inbuxa:Relay/set", json!({"destroy": [&id]})).await; }