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<Inner>) {
     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<u64>,
+    /// `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::<ProtoRelayEvent>(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 {

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)]

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),

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:
 

resources/schema/schema.json.sha256

@@ -1 +1 @@
-B5ARrqKFTX10Ib2fIAsGacr8yxF4NyI3IwmJEPFiTP8
+D6Ad4OELCzUv1LCDwL_v0XRNhdN-2jaO4HuYVRXzxu4

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, "[email protected]", &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;
 }