crates/features/src/relay/messages.rs

@@ -0,0 +1,53 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! `inbuxa:RelayMessage` ids (relay spec, "Registry objects"): a relay's
+//! `Inbound` and `Outbound` messages, as this server names them. Nothing
+//! is stored: the relay keeps its queues, and each id is made from the
+//! relay's id and the relay's own message id, so every node names a
+//! message the same way. The top 16 bits are the relay, the rest a hash
+//! of the message's id at the relay, which a lookup checks against the
+//! relay's queues.
+
+use xxhash_rust::xxh3::xxh3_64;
+
+const HASH_BITS: u32 = 48;
+const HASH_MASK: u64 = (1 << HASH_BITS) - 1;
+
+/// Relays numbered above this have no message ids. Relays are numbered
+/// from 1, one more for each made, so there are never that many.
+pub const MAX_RELAY_ID: u32 = u16::MAX as u32;
+
+/// The id of a relay's message, from its id at the relay.
+pub fn message_id(relay_id: u32, relay_message_id: &str) -> Option<u64> {
+    (relay_id <= MAX_RELAY_ID).then(|| {
+        (u64::from(relay_id) << HASH_BITS) | (xxh3_64(relay_message_id.as_bytes()) & HASH_MASK)
+    })
+}
+
+/// The relay a message id is at.
+pub fn relay_of(id: u64) -> u32 {
+    (id >> HASH_BITS) as u32
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn ids_name_their_relay() {
+        let id = message_id(7, "o0123456789abcdef01234567").unwrap();
+        assert_eq!(relay_of(id), 7);
+        assert_eq!(message_id(7, "o0123456789abcdef01234567"), Some(id));
+        assert_ne!(message_id(8, "o0123456789abcdef01234567"), Some(id));
+        assert_ne!(message_id(7, "i0123456789abcdef01234567"), Some(id));
+        assert_eq!(
+            relay_of(message_id(MAX_RELAY_ID, "x").unwrap()),
+            MAX_RELAY_ID
+        );
+        assert_eq!(message_id(MAX_RELAY_ID + 1, "x"), None);
+    }
+}

crates/features/src/relay/mod.rs

@@ -24,6 +24,7 @@
 //! Numbers are big-endian. There are few relays, so they're read whole.
 
 pub mod link;
+pub mod messages;
 pub mod outbound;
 
 use inbuxa_relay_proto::{

crates/features/src/relay/outbound.rs

@@ -47,10 +47,16 @@ pub enum Refused {
 
 pub type HandOver<'a> = Pin<Box<dyn Future<Output = Result<String, Refused>> + Send + 'a>>;
 
+pub type Call<'a> = Pin<Box<dyn Future<Output = Result<serde_json::Value, String>> + Send + 'a>>;
+
 /// One relay link on this node, able to take a message: uploads it and
 /// creates an `Outbound`, whose id it returns.
 pub trait Sender: Send + Sync {
     fn hand_over(&self, message: Vec<u8>, envelope: Envelope) -> HandOver<'_>;
+
+    /// One method on the relay's API, on its account: the answer's
+    /// arguments, or why there are none (the relay's error included).
+    fn call(&self, method: &'static str, args: serde_json::Value) -> Call<'_>;
 }
 
 /// A relay this node is linked to.
@@ -121,6 +127,17 @@ impl Senders {
             .insert(linked.relay_id, linked);
     }
 
+    /// This node's link to a relay, while it is up, whether the relay
+    /// sends mail or not.
+    pub fn linked(&self, relay_id: u32) -> Option<Arc<Linked>> {
+        self.state
+            .lock()
+            .expect("senders")
+            .linked
+            .get(&relay_id)
+            .cloned()
+    }
+
     /// Forgets a link, unless a newer one to the same relay replaced it.
     pub fn link_down(&self, linked: &Arc<Linked>) {
         let mut state = self.state.lock().expect("senders");
@@ -245,6 +262,10 @@ mod tests {
         fn hand_over(&self, _: Vec<u8>, _: Envelope) -> HandOver<'_> {
             Box::pin(async { Err(Refused::Temporary("no".into())) })
         }
+
+        fn call(&self, _: &'static str, _: serde_json::Value) -> Call<'_> {
+            Box::pin(async { Err("no".into()) })
+        }
     }
 
     fn relay(id: u32, send_through: bool, inbound_only: bool) -> Relay {

crates/jmap-proto/src/object/inbuxa_relay_message.rs

@@ -0,0 +1,268 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! `inbuxa:RelayMessage/get`, `/query` and `/set` under `urn:inbuxa:jmap`:
+//! what waits in a relay's `Inbound` and `Outbound` queues, fetched over
+//! this node's link when asked (docs/spec/features/relay.md, "Registry
+//! objects"). Never message content. Set only destroys: discards held
+//! mail, or cancels an outbound message. The set call's `reason`, if
+//! given, goes into the audit log with the change.
+
+use crate::{
+    object::{AnyId, JmapObject, JmapObjectId},
+    request::deserialize::DeserializeArguments,
+};
+use jmap_tools::{Element, Key, Property};
+use std::{borrow::Cow, str::FromStr};
+use types::id::Id;
+
+#[derive(Debug, Clone, Default)]
+pub struct RelayMessage;
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum RelayMessageProperty {
+    Id,
+    /// The `inbuxa:Relay` it is at.
+    RelayId,
+    /// `inbound` or `outbound`.
+    Direction,
+    /// Its id at the relay.
+    RelayMessageId,
+    CreatedAt,
+    Size,
+    /// RFC 8621 `Envelope`.
+    Envelope,
+    /// Inbound: `held`, `pending` and so on, as the relay says.
+    State,
+    ClaimedUntil,
+    /// Inbound: the SMTP session it came in on.
+    Session,
+    DoorChecks,
+    /// Outbound: RFC 8621 `DeliveryStatus` per recipient.
+    DeliveryStatus,
+    NextAttemptAt,
+    Attempts,
+    ExpiresAt,
+    /// Outbound: the message in this server's queue it was handed from.
+    QueueId,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum RelayMessageValue {
+    Id(Id),
+}
+
+impl Property for RelayMessageProperty {
+    fn try_parse(parent: Option<&Key<'_, Self>>, value: &str) -> Option<Self> {
+        // Keys inside the envelope, session and statuses stay plain keys
+        match parent {
+            None => RelayMessageProperty::parse(value),
+            Some(_) => None,
+        }
+    }
+
+    fn to_cow(&self) -> Cow<'static, str> {
+        match self {
+            RelayMessageProperty::Id => "id",
+            RelayMessageProperty::RelayId => "relayId",
+            RelayMessageProperty::Direction => "direction",
+            RelayMessageProperty::RelayMessageId => "relayMessageId",
+            RelayMessageProperty::CreatedAt => "createdAt",
+            RelayMessageProperty::Size => "size",
+            RelayMessageProperty::Envelope => "envelope",
+            RelayMessageProperty::State => "state",
+            RelayMessageProperty::ClaimedUntil => "claimedUntil",
+            RelayMessageProperty::Session => "session",
+            RelayMessageProperty::DoorChecks => "doorChecks",
+            RelayMessageProperty::DeliveryStatus => "deliveryStatus",
+            RelayMessageProperty::NextAttemptAt => "nextAttemptAt",
+            RelayMessageProperty::Attempts => "attempts",
+            RelayMessageProperty::ExpiresAt => "expiresAt",
+            RelayMessageProperty::QueueId => "queueId",
+        }
+        .into()
+    }
+}
+
+impl RelayMessageProperty {
+    fn parse(value: &str) -> Option<Self> {
+        hashify::tiny_map!(value.as_bytes(),
+            b"id" => RelayMessageProperty::Id,
+            b"relayId" => RelayMessageProperty::RelayId,
+            b"direction" => RelayMessageProperty::Direction,
+            b"relayMessageId" => RelayMessageProperty::RelayMessageId,
+            b"createdAt" => RelayMessageProperty::CreatedAt,
+            b"size" => RelayMessageProperty::Size,
+            b"envelope" => RelayMessageProperty::Envelope,
+            b"state" => RelayMessageProperty::State,
+            b"claimedUntil" => RelayMessageProperty::ClaimedUntil,
+            b"session" => RelayMessageProperty::Session,
+            b"doorChecks" => RelayMessageProperty::DoorChecks,
+            b"deliveryStatus" => RelayMessageProperty::DeliveryStatus,
+            b"nextAttemptAt" => RelayMessageProperty::NextAttemptAt,
+            b"attempts" => RelayMessageProperty::Attempts,
+            b"expiresAt" => RelayMessageProperty::ExpiresAt,
+            b"queueId" => RelayMessageProperty::QueueId,
+        )
+    }
+}
+
+impl FromStr for RelayMessageProperty {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        RelayMessageProperty::parse(s).ok_or(())
+    }
+}
+
+impl Element for RelayMessageValue {
+    type Property = RelayMessageProperty;
+
+    fn try_parse<P>(key: &Key<'_, Self::Property>, value: &str) -> Option<Self> {
+        match key {
+            Key::Property(
+                RelayMessageProperty::Id
+                | RelayMessageProperty::RelayId
+                | RelayMessageProperty::QueueId,
+            ) => Id::from_str(value).ok().map(RelayMessageValue::Id),
+            _ => None,
+        }
+    }
+
+    fn to_cow(&self) -> Cow<'static, str> {
+        match self {
+            RelayMessageValue::Id(id) => id.to_string().into(),
+        }
+    }
+}
+
+/// One condition of an `inbuxa:RelayMessage/query` filter. Several in one
+/// filter object must all hold.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RelayMessageFilter {
+    RelayId(Id),
+    /// `inbound` or `outbound`.
+    Direction(String),
+    _T(String),
+}
+
+impl Default for RelayMessageFilter {
+    fn default() -> Self {
+        RelayMessageFilter::_T(String::new())
+    }
+}
+
+impl<'de> DeserializeArguments<'de> for RelayMessageFilter {
+    fn deserialize_argument<A>(&mut self, key: &str, map: &mut A) -> Result<(), A::Error>
+    where
+        A: serde::de::MapAccess<'de>,
+    {
+        match key {
+            "relayId" => *self = RelayMessageFilter::RelayId(map.next_value()?),
+            "direction" => *self = RelayMessageFilter::Direction(map.next_value()?),
+            _ => {
+                *self = RelayMessageFilter::_T(key.to_string());
+                let _ = map.next_value::<serde::de::IgnoredAny>()?;
+            }
+        }
+        Ok(())
+    }
+}
+
+/// The set call's own argument: why, for the audit log.
+#[derive(Debug, Clone, Default)]
+pub struct RelayMessageSetArguments {
+    pub reason: Option<String>,
+}
+
+impl<'de> DeserializeArguments<'de> for RelayMessageSetArguments {
+    fn deserialize_argument<A>(&mut self, key: &str, map: &mut A) -> Result<(), A::Error>
+    where
+        A: serde::de::MapAccess<'de>,
+    {
+        if key == "reason" {
+            self.reason = map.next_value()?;
+        } else {
+            let _ = map.next_value::<serde::de::IgnoredAny>()?;
+        }
+        Ok(())
+    }
+}
+
+impl JmapObject for RelayMessage {
+    type Property = RelayMessageProperty;
+
+    type Element = RelayMessageValue;
+
+    type Id = Id;
+
+    type Filter = RelayMessageFilter;
+
+    type Comparator = ();
+
+    type GetArguments = ();
+
+    type SetArguments<'de> = RelayMessageSetArguments;
+
+    type QueryArguments = ();
+
+    type CopyArguments = ();
+
+    type ParseArguments = ();
+
+    const ID_PROPERTY: Self::Property = RelayMessageProperty::Id;
+}
+
+impl From<Id> for RelayMessageValue {
+    fn from(id: Id) -> Self {
+        RelayMessageValue::Id(id)
+    }
+}
+
+impl JmapObjectId for RelayMessageValue {
+    fn as_id(&self) -> Option<Id> {
+        match self {
+            RelayMessageValue::Id(id) => Some(*id),
+        }
+    }
+
+    fn as_any_id(&self) -> Option<AnyId> {
+        match self {
+            RelayMessageValue::Id(id) => Some(AnyId::Id(*id)),
+        }
+    }
+
+    fn as_id_ref(&self) -> Option<&str> {
+        None
+    }
+
+    fn try_set_id(&mut self, new_id: AnyId) -> bool {
+        if let AnyId::Id(id) = new_id {
+            *self = RelayMessageValue::Id(id);
+            true
+        } else {
+            false
+        }
+    }
+}
+
+impl JmapObjectId for RelayMessageProperty {
+    fn as_id(&self) -> Option<Id> {
+        None
+    }
+
+    fn as_any_id(&self) -> Option<AnyId> {
+        None
+    }
+
+    fn as_id_ref(&self) -> Option<&str> {
+        None
+    }
+
+    fn try_set_id(&mut self, _: AnyId) -> bool {
+        false
+    }
+}

crates/jmap-proto/src/object/mod.rs

@@ -43,6 +43,7 @@ pub mod inbuxa_log_settings; // inbuxa: personal-data catalog, D1
 pub mod inbuxa_mail_rule; // inbuxa: DLP and mail flow rules
 pub mod inbuxa_protocol_policy; // inbuxa: legacy protocols off
 pub mod inbuxa_relay; // inbuxa: inbuxa Relay
+pub mod inbuxa_relay_message; // inbuxa: a relay's queues
 pub mod inbuxa_report_export; // inbuxa: scheduled reports, RP-19
 pub mod inbuxa_scheduled_report; // inbuxa: scheduled reports
 pub mod inbuxa_scheduled_report_settings; // inbuxa: scheduled reports

crates/jmap-proto/src/references/eval.rs

@@ -91,6 +91,9 @@ impl Response<'_> {
                         GetResponseMethod::Relay(response) => {
                             response.eval_jptr(path, &mut results)
                         }
+                        GetResponseMethod::RelayMessage(response) => {
+                            response.eval_jptr(path, &mut results)
+                        }
                         GetResponseMethod::SecurityAcceptance(response) => {
                             response.eval_jptr(path, &mut results)
                         }

crates/jmap-proto/src/references/resolve.rs

@@ -56,6 +56,7 @@ impl Response<'_> {
                 GetRequestMethod::LegalHold(request) => request.resolve_references(self)?,
                 GetRequestMethod::MailRule(request) => request.resolve_references(self)?,
                 GetRequestMethod::Relay(request) => request.resolve_references(self)?,
+                GetRequestMethod::RelayMessage(request) => request.resolve_references(self)?,
                 GetRequestMethod::SecurityAcceptance(request) => {
                     request.resolve_references(self)?
                 }
@@ -151,6 +152,9 @@ impl Response<'_> {
                     request.resolve_references(self, 1, false)?
                 }
                 SetRequestMethod::Relay(request) => request.resolve_references(self, 1, false)?,
+                SetRequestMethod::RelayMessage(request) => {
+                    request.resolve_references(self, 1, false)?
+                }
                 SetRequestMethod::SecurityAcceptance(request) => {
                     request.resolve_references(self, 1, false)?
                 }

crates/jmap-proto/src/request/method.rs

@@ -69,6 +69,7 @@ pub enum MethodObject {
     // inbuxa: DLP and mail flow rules
     MailRule,
     Relay,
+    RelayMessage,
     // inbuxa: accepted security to-do items
     SecurityAcceptance,
     // inbuxa: the deliverability check
@@ -126,6 +127,7 @@ impl MethodObject {
             | MethodObject::HoldExport
             | MethodObject::MailRule
             | MethodObject::Relay
+            | MethodObject::RelayMessage
             | MethodObject::SecurityAcceptance
             | MethodObject::HeldMessage
             | MethodObject::DeliverabilityReport
@@ -340,6 +342,9 @@ impl MethodName {
             (MethodFunction::Set, MethodObject::MailRule) => "inbuxa:MailRule/set",
             (MethodFunction::Get, MethodObject::Relay) => "inbuxa:Relay/get",
             (MethodFunction::Set, MethodObject::Relay) => "inbuxa:Relay/set",
+            (MethodFunction::Get, MethodObject::RelayMessage) => "inbuxa:RelayMessage/get",
+            (MethodFunction::Query, MethodObject::RelayMessage) => "inbuxa:RelayMessage/query",
+            (MethodFunction::Set, MethodObject::RelayMessage) => "inbuxa:RelayMessage/set",
             (MethodFunction::Get, MethodObject::SecurityAcceptance) => {
                 "inbuxa:SecurityAcceptance/get"
             }
@@ -540,6 +545,9 @@ impl MethodName {
             "inbuxa:MailRule/set" => (MethodObject::MailRule, MethodFunction::Set),
             "inbuxa:Relay/get" => (MethodObject::Relay, MethodFunction::Get),
             "inbuxa:Relay/set" => (MethodObject::Relay, MethodFunction::Set),
+            "inbuxa:RelayMessage/get" => (MethodObject::RelayMessage, MethodFunction::Get),
+            "inbuxa:RelayMessage/query" => (MethodObject::RelayMessage, MethodFunction::Query),
+            "inbuxa:RelayMessage/set" => (MethodObject::RelayMessage, MethodFunction::Set),
             "inbuxa:SecurityAcceptance/get" => (MethodObject::SecurityAcceptance, MethodFunction::Get),
             "inbuxa:SecurityAcceptance/set" => (MethodObject::SecurityAcceptance, MethodFunction::Set),
             "inbuxa:DeliverabilityReport/get" => (MethodObject::DeliverabilityReport, MethodFunction::Get),
@@ -637,6 +645,7 @@ impl Display for MethodObject {
             MethodObject::LegalHold => "inbuxa:LegalHold",
             MethodObject::MailRule => "inbuxa:MailRule",
             MethodObject::Relay => "inbuxa:Relay",
+            MethodObject::RelayMessage => "inbuxa:RelayMessage",
             MethodObject::SecurityAcceptance => "inbuxa:SecurityAcceptance",
             MethodObject::DeliverabilityReport => "inbuxa:DeliverabilityReport",
             MethodObject::DeliverabilitySettings => "inbuxa:DeliverabilitySettings",

crates/jmap-proto/src/request/mod.rs

@@ -128,6 +128,7 @@ pub enum GetRequestMethod {
     LegalHold(Box<GetRequest<crate::object::inbuxa_legal_hold::LegalHold>>),
     MailRule(Box<GetRequest<crate::object::inbuxa_mail_rule::MailRule>>),
     Relay(Box<GetRequest<crate::object::inbuxa_relay::Relay>>),
+    RelayMessage(Box<GetRequest<crate::object::inbuxa_relay_message::RelayMessage>>),
     SecurityAcceptance(
         Box<GetRequest<crate::object::inbuxa_security_acceptance::SecurityAcceptance>>,
     ),
@@ -185,6 +186,7 @@ pub enum SetRequestMethod<'x> {
     LegalHold(Box<SetRequest<'x, crate::object::inbuxa_legal_hold::LegalHold>>),
     MailRule(Box<SetRequest<'x, crate::object::inbuxa_mail_rule::MailRule>>),
     Relay(Box<SetRequest<'x, crate::object::inbuxa_relay::Relay>>),
+    RelayMessage(Box<SetRequest<'x, crate::object::inbuxa_relay_message::RelayMessage>>),
     SecurityAcceptance(
         Box<SetRequest<'x, crate::object::inbuxa_security_acceptance::SecurityAcceptance>>,
     ),
@@ -246,6 +248,7 @@ pub enum QueryRequestMethod {
     Registry(Box<QueryRequest<Registry>>),
     AuditEvent(Box<QueryRequest<crate::object::inbuxa_audit::AuditEvent>>),
     JournalEntry(Box<QueryRequest<crate::object::inbuxa_journal_entry::JournalEntry>>),
+    RelayMessage(Box<QueryRequest<crate::object::inbuxa_relay_message::RelayMessage>>),
 }
 
 #[derive(Debug)]

crates/jmap-proto/src/request/parser.rs

@@ -704,6 +704,27 @@ impl<'de> Visitor<'de> for CallVisitor {
                     return Err(de::Error::invalid_length(1, &self));
                 }
             },
+            (MethodFunction::Get, MethodObject::RelayMessage) => match seq.next_element() {
+                Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::RelayMessage(value)),
+                Err(err) => RequestMethod::invalid(err),
+                Ok(None) => {
+                    return Err(de::Error::invalid_length(1, &self));
+                }
+            },
+            (MethodFunction::Query, MethodObject::RelayMessage) => match seq.next_element() {
+                Ok(Some(value)) => RequestMethod::Query(QueryRequestMethod::RelayMessage(value)),
+                Err(err) => RequestMethod::invalid(err),
+                Ok(None) => {
+                    return Err(de::Error::invalid_length(1, &self));
+                }
+            },
+            (MethodFunction::Set, MethodObject::RelayMessage) => match seq.next_element() {
+                Ok(Some(value)) => RequestMethod::Set(SetRequestMethod::RelayMessage(value)),
+                Err(err) => RequestMethod::invalid(err),
+                Ok(None) => {
+                    return Err(de::Error::invalid_length(1, &self));
+                }
+            },
             // inbuxa: accepted security to-do items
             (MethodFunction::Get, MethodObject::SecurityAcceptance) => match seq.next_element() {
                 Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::SecurityAcceptance(value)),

crates/jmap-proto/src/response/mod.rs

@@ -113,6 +113,7 @@ pub enum GetResponseMethod {
     LegalHold(GetResponse<crate::object::inbuxa_legal_hold::LegalHold>),
     MailRule(GetResponse<crate::object::inbuxa_mail_rule::MailRule>),
     Relay(GetResponse<crate::object::inbuxa_relay::Relay>),
+    RelayMessage(GetResponse<crate::object::inbuxa_relay_message::RelayMessage>),
     SecurityAcceptance(GetResponse<crate::object::inbuxa_security_acceptance::SecurityAcceptance>),
     DeliverabilityReport(
         GetResponse<crate::object::inbuxa_deliverability_report::DeliverabilityReport>,
@@ -168,6 +169,7 @@ pub enum SetResponseMethod {
     LegalHold(Box<SetResponse<crate::object::inbuxa_legal_hold::LegalHold>>),
     MailRule(Box<SetResponse<crate::object::inbuxa_mail_rule::MailRule>>),
     Relay(Box<SetResponse<crate::object::inbuxa_relay::Relay>>),
+    RelayMessage(Box<SetResponse<crate::object::inbuxa_relay_message::RelayMessage>>),
     SecurityAcceptance(
         Box<SetResponse<crate::object::inbuxa_security_acceptance::SecurityAcceptance>>,
     ),
@@ -1047,6 +1049,22 @@ impl<'x> From<SetResponse<crate::object::inbuxa_relay::Relay>> for ResponseMetho
     }
 }
 
+impl<'x> From<GetResponse<crate::object::inbuxa_relay_message::RelayMessage>>
+    for ResponseMethod<'x>
+{
+    fn from(value: GetResponse<crate::object::inbuxa_relay_message::RelayMessage>) -> Self {
+        ResponseMethod::Get(GetResponseMethod::RelayMessage(value))
+    }
+}
+
+impl<'x> From<SetResponse<crate::object::inbuxa_relay_message::RelayMessage>>
+    for ResponseMethod<'x>
+{
+    fn from(value: SetResponse<crate::object::inbuxa_relay_message::RelayMessage>) -> Self {
+        ResponseMethod::Set(SetResponseMethod::RelayMessage(Box::new(value)))
+    }
+}
+
 // inbuxa: journaling
 impl<'x> From<GetResponse<crate::object::inbuxa_journal_entry::JournalEntry>>
     for ResponseMethod<'x>

crates/jmap/src/api/auth.rs

@@ -118,6 +118,7 @@ impl JmapAuthorization for AccessToken {
                 }
                 // inbuxa: inbuxa Relay
                 GetRequestMethod::Relay(_) => Permission::SysRelayGet,
+                GetRequestMethod::RelayMessage(_) => Permission::SysRelayQueueGet,
                 // inbuxa: journaling (JR-18)
                 GetRequestMethod::Journal(_) => Permission::SysJournalGet,
                 GetRequestMethod::JournalEntry(_) => Permission::SysJournalSearch,
@@ -320,6 +321,14 @@ impl JmapAuthorization for AccessToken {
                         Permission::SysRelayUpdate,
                         Permission::SysRelayUpdate,
                     ),
+                    // inbuxa: a relay's queues; set only destroys
+                    SetRequestMethod::RelayMessage(s) => validate_set(
+                        s,
+                        self,
+                        Permission::SysRelayQueueUpdate,
+                        Permission::SysRelayQueueUpdate,
+                        Permission::SysRelayQueueUpdate,
+                    ),
                     // inbuxa: journaling (JR-18)
                     SetRequestMethod::Journal(s) => validate_set(
                         s,
@@ -577,6 +586,7 @@ impl JmapAuthorization for AccessToken {
                 | MethodObject::HoldExport
                 | MethodObject::MailRule
                 | MethodObject::Relay
+                | MethodObject::RelayMessage
                 | MethodObject::SecurityAcceptance
                 | MethodObject::DeliverabilityReport
                 | MethodObject::DeliverabilitySettings
@@ -650,6 +660,8 @@ impl JmapAuthorization for AccessToken {
                 QueryRequestMethod::AuditEvent(_) => Permission::SysAuditGet,
                 // inbuxa: journaling (JR-15)
                 QueryRequestMethod::JournalEntry(_) => Permission::SysJournalSearch,
+                // inbuxa: a relay's queues
+                QueryRequestMethod::RelayMessage(_) => Permission::SysRelayQueueGet,
                 QueryRequestMethod::Registry(_) => {
                     let MethodObject::Registry(object_type) = object else {
                         unreachable!()

crates/jmap/src/api/request.rs

@@ -293,6 +293,9 @@ impl RequestHandler for Server {
                                     SetResponseMethod::Relay(set_response) => {
                                         set_response.update_created_ids(&mut response);
                                     }
+                                    SetResponseMethod::RelayMessage(set_response) => {
+                                        set_response.update_created_ids(&mut response);
+                                    }
                                     SetResponseMethod::SecurityAcceptance(set_response) => {
                                         set_response.update_created_ids(&mut response);
                                     }
@@ -566,6 +569,13 @@ impl RequestHandler for Server {
                         .await?
                         .into()
                 }
+                // inbuxa: a relay's queues, fetched over the link
+                GetRequestMethod::RelayMessage(mut req) => {
+                    resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
+                    crate::inbuxa::relay_message::get(self, access_token, *req)
+                        .await?
+                        .into()
+                }
                 // inbuxa: accepted security to-do items
                 GetRequestMethod::SecurityAcceptance(mut req) => {
                     resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
@@ -835,6 +845,13 @@ impl RequestHandler for Server {
                         .await?
                         .into()
                 }
+                // inbuxa: a relay's queues
+                QueryRequestMethod::RelayMessage(mut req) => {
+                    resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
+                    crate::inbuxa::relay_message::query(self, access_token, *req)
+                        .await?
+                        .into()
+                }
                 QueryRequestMethod::Registry(mut req) => {
                     resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
                     assert_registry_account(self, method_name.obj, access_token, req.account_id)
@@ -1138,6 +1155,22 @@ impl RequestHandler for Server {
                     .await?
                     .into()
                 }
+                SetRequestMethod::RelayMessage(mut req) => {
+                    resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
+                    let reason = req.arguments.reason.clone();
+                    crate::inbuxa::audit::recorded(
+                        self,
+                        access_token,
+                        session,
+                        &method_name.obj.to_string(),
+                        None,
+                        reason,
+                        *req,
+                        |req| Box::pin(crate::inbuxa::relay_message::set(self, access_token, req)),
+                    )
+                    .await?
+                    .into()
+                }
                 // inbuxa: SS-26, every acceptance made or removed is in the
                 // audit log
                 SetRequestMethod::SecurityAcceptance(mut req) => {

crates/jmap/src/changes/get.rs

@@ -432,6 +432,7 @@ impl IntermediateChangesResponse {
             | MethodObject::HoldExport
             | MethodObject::MailRule
             | MethodObject::Relay
+            | MethodObject::RelayMessage
             | MethodObject::SecurityAcceptance
             | MethodObject::DeliverabilityReport
             | MethodObject::DeliverabilitySettings

crates/jmap/src/inbuxa/mod.rs

@@ -33,6 +33,7 @@ pub mod masked_secrets; // inbuxa: "****" is never saved as a secret
 pub mod oidc_directory; // inbuxa: http-directory-services-13
 pub mod protocol_policy;
 pub mod relay; // inbuxa: inbuxa Relay
+pub mod relay_message; // inbuxa: a relay's queues
 pub mod scheduled_reports; // inbuxa: scheduled reports and the weekly digest
 pub mod security_acceptance;
 pub mod sharing_policy;

crates/jmap/src/inbuxa/relay.rs

@@ -72,7 +72,7 @@ const SERVER_SET: &[P] = &[
     P::UpdatedAt,
 ];
 
-fn server_level(access_token: &AccessToken) -> trc::Result<()> {
+pub(super) fn server_level(access_token: &AccessToken) -> trc::Result<()> {
     if access_token.tenant_id().is_some() {
         Err(trc::JmapEvent::Forbidden
             .into_err()
@@ -82,7 +82,12 @@ fn server_level(access_token: &AccessToken) -> trc::Result<()> {
     }
 }
 
-fn json_to_value(json: serde_json::Value) -> RValue {
+/// A JSON value as a JMAP one, its object keys kept as plain keys.
+pub(super) fn json_to_value<P, E>(json: serde_json::Value) -> Value<'static, P, E>
+where
+    P: Property,
+    E: jmap_tools::Element<Property = P>,
+{
     match json {
         serde_json::Value::Null => Value::Null,
         serde_json::Value::Bool(b) => Value::Bool(b),

crates/jmap/src/inbuxa/relay_message.rs

@@ -0,0 +1,476 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! `inbuxa:RelayMessage` (docs/spec/features/relay.md, "Registry
+//! objects"): what waits in each relay's `Inbound` and `Outbound` queues,
+//! fetched over this node's link when asked, and never the message itself.
+//! Destroying one discards held mail at the relay, or cancels an outbound
+//! message: its recipients fail here, with no DSN, as a delivery cancelled
+//! in this server's queue does, and the relay drops its copy (at its next
+//! status, if it is sending it just then). The request layer records every
+//! destroy in the audit log.
+
+use super::relay::{json_to_value, server_level};
+use common::{Server, auth::AccessToken};
+use inbuxa_features::relay::{
+    self as relays, Relay as StoredRelay,
+    messages::{message_id, relay_of},
+    outbound::{self, Handed, Linked},
+};
+use inbuxa_relay_proto::jmap::method;
+use jmap_proto::{
+    error::set::SetError,
+    method::{
+        get::{GetRequest, GetResponse},
+        query::{Filter as QueryFilter, QueryRequest, QueryResponse},
+        set::{SetRequest, SetResponse},
+    },
+    object::inbuxa_relay_message::{
+        RelayMessage, RelayMessageFilter, RelayMessageProperty as P, RelayMessageValue,
+    },
+    request::IntoValid,
+    types::state::State,
+};
+use jmap_tools::{Key, Map, Value};
+use serde_json::json;
+use smtp::outbound::relay::{Applied, RelayedStatus, apply_relayed};
+use std::{collections::HashMap, sync::Arc};
+use types::id::Id;
+
+type RValue = Value<'static, P, RelayMessageValue>;
+
+const ALL: &[P] = &[
+    P::Id,
+    P::RelayId,
+    P::Direction,
+    P::RelayMessageId,
+    P::CreatedAt,
+    P::Size,
+    P::Envelope,
+    P::State,
+    P::ClaimedUntil,
+    P::Session,
+    P::DoorChecks,
+    P::DeliveryStatus,
+    P::NextAttemptAt,
+    P::Attempts,
+    P::ExpiresAt,
+    P::QueueId,
+];
+
+/// The most ids one query returns.
+const MAX_QUERY_LIMIT: usize = 1_000;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Direction {
+    Inbound,
+    Outbound,
+}
+
+impl Direction {
+    fn as_str(self) -> &'static str {
+        match self {
+            Direction::Inbound => "inbound",
+            Direction::Outbound => "outbound",
+        }
+    }
+}
+
+/// One message at a relay, as the relay describes it.
+struct Found {
+    id: u64,
+    relay_id: u32,
+    direction: Direction,
+    relay_message_id: String,
+    created_at: String,
+    object: serde_json::Value,
+}
+
+/// One relay's queues, read over this node's link.
+struct Queues {
+    linked: Arc<Linked>,
+    messages: Vec<Found>,
+    /// Its outbound messages' places in this server's queue.
+    handed: HashMap<String, Handed>,
+}
+
+fn unavailable(relay: &StoredRelay, why: impl std::fmt::Display) -> trc::Error {
+    trc::EventType::Relay(trc::RelayEvent::LinkDown)
+        .into_err()
+        .details(format!("Relay {}: {why}", relay.name))
+        .ctx(trc::Key::Id, Id::from(relay.id).to_string())
+}
+
+async fn queues(server: &Server, relay: &StoredRelay) -> trc::Result<Queues> {
+    let linked = server
+        .inner
+        .data
+        .relay_senders
+        .linked(relay.id)
+        .ok_or_else(|| unavailable(relay, "this node has no link to it now"))?;
+    let mut messages = Vec::new();
+    for (direction, get) in [
+        (Direction::Inbound, method::INBOUND_GET),
+        (Direction::Outbound, method::OUTBOUND_GET),
+    ] {
+        let answer = linked
+            .sender
+            .call(get, json!({"ids": null}))
+            .await
+            .map_err(|why| unavailable(relay, why))?;
+        for object in answer["list"].as_array().into_iter().flatten() {
+            let Some(relay_message_id) = object["id"].as_str() else {
+                continue;
+            };
+            let Some(id) = message_id(relay.id, relay_message_id) else {
+                continue;
+            };
+            let created_at = match direction {
+                Direction::Inbound => &object["receivedAt"],
+                Direction::Outbound => &object["createdAt"],
+            };
+            messages.push(Found {
+                id,
+                relay_id: relay.id,
+                direction,
+                relay_message_id: relay_message_id.to_string(),
+                created_at: created_at.as_str().unwrap_or_default().to_string(),
+                object: object.clone(),
+            });
+        }
+    }
+    let handed = outbound::handed(server.store(), relay.id)
+        .await?
+        .into_iter()
+        .map(|h| (h.id.clone(), h))
+        .collect();
+    Ok(Queues {
+        linked,
+        messages,
+        handed,
+    })
+}
+
+fn to_value(found: &Found, handed: &HashMap<String, Handed>, properties: &[P]) -> RValue {
+    let field = |name: &str| -> RValue {
+        match found.object.get(name) {
+            Some(value) if !value.is_null() => json_to_value(value.clone()),
+            _ => Value::Null,
+        }
+    };
+    let inbound = found.direction == Direction::Inbound;
+    let mut out = Map::with_capacity(properties.len());
+    for property in properties {
+        let value = match property {
+            P::Id => Value::Element(RelayMessageValue::Id(Id::new(found.id))),
+            P::RelayId => Value::Element(RelayMessageValue::Id(Id::from(found.relay_id))),
+            P::Direction => Value::Str(found.direction.as_str().into()),
+            P::RelayMessageId => Value::Str(found.relay_message_id.clone().into()),
+            P::CreatedAt => Value::Str(found.created_at.clone().into()),
+            P::Size => field("size"),
+            P::Envelope => field("envelope"),
+            P::State if inbound => field("state"),
+            P::ClaimedUntil if inbound => field("claimedUntil"),
+            P::Session if inbound => field("session"),
+            P::DoorChecks if inbound => field("doorChecks"),
+            P::DeliveryStatus if !inbound => field("deliveryStatus"),
+            P::NextAttemptAt if !inbound => field("nextAttemptAt"),
+            P::Attempts if !inbound => field("attempts"),
+            P::ExpiresAt if !inbound => field("expiresAt"),
+            P::QueueId if !inbound => handed
+                .get(&found.relay_message_id)
+                .map_or(Value::Null, |h| {
+                    Value::Element(RelayMessageValue::Id(Id::new(h.queue_id)))
+                }),
+            _ => Value::Null,
+        };
+        out.insert_unchecked(Key::Property(property.clone()), value);
+    }
+    Value::Object(out)
+}
+
+/// The relays with these messages, or every enabled relay.
+fn wanted_relays(all: &[StoredRelay], ids: Option<&[Id]>) -> Vec<StoredRelay> {
+    all.iter()
+        .filter(|relay| match ids {
+            Some(ids) => ids.iter().any(|id| relay_of(id.id()) == relay.id),
+            None => relay.enabled,
+        })
+        .cloned()
+        .collect()
+}
+
+pub async fn get(
+    server: &Server,
+    access_token: &AccessToken,
+    mut request: GetRequest<RelayMessage>,
+) -> trc::Result<GetResponse<RelayMessage>> {
+    server_level(access_token)?;
+    let properties = request.unwrap_properties(ALL);
+    let (ids, not_found) = request.unwrap_ids(server.core.jmap.get_max_objects)?;
+    let mut response = GetResponse {
+        account_id: request.account_id.into(),
+        state: None,
+        list: Vec::new(),
+        not_found,
+    };
+    let all = relays::all(server.store()).await?;
+    let mut fetched = Vec::new();
+    for relay in wanted_relays(&all, ids.as_deref()) {
+        match queues(server, &relay).await {
+            Ok(queues) => fetched.push(queues),
+            // Asked for all: a relay this node can't reach now is left out
+            Err(_) if ids.is_none() => {}
+            Err(err) => return Err(err),
+        }
+    }
+    match ids {
+        None => {
+            for queues in &fetched {
+                for found in &queues.messages {
+                    response
+                        .list
+                        .push(to_value(found, &queues.handed, &properties));
+                }
+            }
+        }
+        Some(ids) => {
+            for id in ids {
+                match fetched.iter().find_map(|queues| {
+                    queues
+                        .messages
+                        .iter()
+                        .find(|found| found.id == id.id())
+                        .map(|found| (found, &queues.handed))
+                }) {
+                    Some((found, handed)) => {
+                        response.list.push(to_value(found, handed, &properties))
+                    }
+                    None => response.push_not_found(id),
+                }
+            }
+        }
+    }
+    Ok(response)
+}
+
+/// `inbuxa:RelayMessage/query`: newest first, of every enabled relay this
+/// node is linked to, or of the one asked for, which must be.
+pub async fn query(
+    server: &Server,
+    access_token: &AccessToken,
+    request: QueryRequest<RelayMessage>,
+) -> trc::Result<QueryResponse> {
+    server_level(access_token)?;
+    let unsupported = |why: String| trc::JmapEvent::UnsupportedFilter.into_err().details(why);
+    let mut relay_id = None;
+    let mut direction = None;
+    for condition in request.filter {
+        match condition {
+            QueryFilter::Property(RelayMessageFilter::RelayId(id)) => relay_id = Some(id),
+            QueryFilter::Property(RelayMessageFilter::Direction(d)) => {
+                direction = Some(match d.as_str() {
+                    "inbound" => Direction::Inbound,
+                    "outbound" => Direction::Outbound,
+                    _ => {
+                        return Err(unsupported("direction is inbound or outbound.".to_string()));
+                    }
+                })
+            }
+            QueryFilter::Property(RelayMessageFilter::_T(other)) => {
+                return Err(unsupported(format!("Unknown filter property {other}.")));
+            }
+            QueryFilter::And | QueryFilter::Close => {}
+            QueryFilter::Or | QueryFilter::Not => {
+                return Err(unsupported(
+                    "Conditions must all hold; OR and NOT aren't supported.".to_string(),
+                ));
+            }
+        }
+    }
+    let position = request.position.unwrap_or(0);
+    if position < 0 || request.anchor.is_some() {
+        return Err(unsupported(
+            "Relay messages page by a position from the start.".to_string(),
+        ));
+    }
+    let limit = request
+        .limit
+        .unwrap_or(MAX_QUERY_LIMIT)
+        .min(MAX_QUERY_LIMIT);
+
+    let all = relays::all(server.store()).await?;
+    let mut found = Vec::new();
+    match relay_id {
+        Some(id) => {
+            if let Some(relay) = all.iter().find(|r| u64::from(r.id) == id.id()) {
+                found.extend(queues(server, relay).await?.messages);
+            }
+        }
+        None => {
+            for relay in all.iter().filter(|r| r.enabled) {
+                if let Ok(queues) = queues(server, relay).await {
+                    found.extend(queues.messages);
+                }
+            }
+        }
+    }
+    found.retain(|f| direction.is_none_or(|d| f.direction == d));
+    found.sort_by(|a, b| b.created_at.cmp(&a.created_at).then(b.id.cmp(&a.id)));
+    let total = found.len();
+    Ok(QueryResponse {
+        account_id: request.account_id,
+        query_state: State::Initial,
+        can_calculate_changes: false,
+        position,
+        ids: found
+            .into_iter()
+            .skip(position as usize)
+            .take(limit)
+            .map(|f| Id::new(f.id))
+            .collect(),
+        total: request.calculate_total.unwrap_or(false).then_some(total),
+        limit: Some(limit),
+    })
+}
+
+/// What a relay's destroy did with one id.
+enum Destroyed {
+    Yes,
+    /// It was gone already.
+    NotFound,
+    Refused(SetError<P>),
+}
+
+fn destroyed(answer: &serde_json::Value, id: &str) -> Destroyed {
+    let Some(error) = answer["notDestroyed"].get(id) else {
+        return Destroyed::Yes;
+    };
+    match error["type"].as_str() {
+        Some("notFound") => Destroyed::NotFound,
+        _ => Destroyed::Refused(
+            SetError::forbidden().with_description(
+                error["description"]
+                    .as_str()
+                    .unwrap_or("The relay refused.")
+                    .to_string(),
+            ),
+        ),
+    }
+}
+
+/// `inbuxa:RelayMessage/set`: destroy only.
+pub async fn set(
+    server: &Server,
+    access_token: &AccessToken,
+    mut request: SetRequest<'_, RelayMessage>,
+) -> trc::Result<SetResponse<RelayMessage>> {
+    server_level(access_token)?;
+    let mut response = SetResponse::from_request(&request, server.core.jmap.set_max_objects)?;
+    for (client_id, _) in request.unwrap_create() {
+        response.not_created.append(
+            client_id,
+            SetError::forbidden().with_description("Only a relay makes its messages."),
+        );
+    }
+    for (id, _) in request.unwrap_update().into_valid() {
+        response.not_updated.append(
+            id,
+            SetError::forbidden()
+                .with_description("A relay's message can't be changed, only destroyed."),
+        );
+    }
+
+    let data = server.store();
+    let all = relays::all(data).await?;
+    let mut fetched: HashMap<u32, Queues> = HashMap::new();
+    for id in request.unwrap_destroy().into_valid() {
+        let Some(relay) = all.iter().find(|r| r.id == relay_of(id.id())) else {
+            response.not_destroyed.append(id, SetError::not_found());
+            continue;
+        };
+        if let std::collections::hash_map::Entry::Vacant(entry) = fetched.entry(relay.id) {
+            entry.insert(queues(server, relay).await?);
+        }
+        let queues = &fetched[&relay.id];
+        let Some(found) = queues.messages.iter().find(|f| f.id == id.id()) else {
+            response.not_destroyed.append(id, SetError::not_found());
+            continue;
+        };
+        let rid = found.relay_message_id.as_str();
+        match found.direction {
+            Direction::Inbound => {
+                // Being taken just now: it is this server's mail already
+                if found.object["state"].as_str() == Some("claimed") {
+                    response.not_destroyed.append(
+                        id,
+                        SetError::forbidden()
+                            .with_description("This server is taking it now; try again."),
+                    );
+                    continue;
+                }
+                let answer = queues
+                    .linked
+                    .sender
+                    .call(method::INBOUND_SET, json!({"destroy": [rid]}))
+                    .await
+                    .map_err(|why| unavailable(relay, why))?;
+                match destroyed(&answer, rid) {
+                    Destroyed::Yes => response.destroyed.push(id),
+                    Destroyed::NotFound => response.not_destroyed.append(id, SetError::not_found()),
+                    Destroyed::Refused(error) => response.not_destroyed.append(id, error),
+                }
+            }
+            Direction::Outbound => {
+                // Cancelled here first, so the queue never hands it over
+                // again once the relay drops it
+                let handed = queues.handed.get(rid);
+                if let Some(handed) = handed {
+                    let cancelled: Vec<(String, RelayedStatus)> = handed
+                        .recipients
+                        .iter()
+                        .map(|r| (r.clone(), RelayedStatus::Cancelled))
+                        .collect();
+                    if apply_relayed(
+                        server,
+                        handed.queue_id,
+                        &handed.queue_name,
+                        &relay.name,
+                        &cancelled,
+                    )
+                    .await
+                        == Applied::Locked
+                    {
+                        response.not_destroyed.append(
+                            id,
+                            SetError::forbidden().with_description(
+                                "This server's queue is working on the message; try again.",
+                            ),
+                        );
+                        continue;
+                    }
+                }
+                let answer = queues
+                    .linked
+                    .sender
+                    .call(method::OUTBOUND_SET, json!({"destroy": [rid]}))
+                    .await
+                    .map_err(|why| unavailable(relay, why))?;
+                match destroyed(&answer, rid) {
+                    // Cancelled here; the relay drops it after this attempt
+                    Destroyed::Refused(error) => response.not_destroyed.append(id, error),
+                    Destroyed::Yes | Destroyed::NotFound => {
+                        if handed.is_some() {
+                            outbound::forget_handed(data, relay.id, &[rid.to_string()]).await?;
+                        }
+                        response.destroyed.push(id);
+                    }
+                }
+            }
+        }
+    }
+    Ok(response)
+}

crates/services/src/inbuxa_relay_link.rs

@@ -46,7 +46,7 @@ use inbuxa_features::{
             RecipientPlan, forget_ingested, is_inbound_id, issue_node_certificate, pushed_settings,
             record_ingested, record_join, set_status, was_ingested,
         },
-        outbound::{self, HandOver, Handed, Linked, Refused, Sender},
+        outbound::{self, Call, HandOver, Handed, Linked, Refused, Sender},
     },
 };
 use inbuxa_relay_proto::{
@@ -1214,6 +1214,32 @@ impl Sender for LinkSender {
             Err(refusal(&args["notCreated"]["m"]))
         })
     }
+
+    fn call(&self, method: &'static str, mut args: Value) -> Call<'_> {
+        Box::pin(async move {
+            if args.get("accountId").is_none() {
+                args["accountId"] = ACCOUNT_ID.into();
+            }
+            let answer = self
+                .dialer
+                .api(vec![Invocation::new(method, args, "c")])
+                .await
+                .map_err(|err| err.message)?
+                .into_iter()
+                .next()
+                .ok_or_else(|| format!("The relay didn't answer {method}."))?;
+            if answer.name() == "error" {
+                return Err(format!(
+                    "The relay refused {method}: {}",
+                    answer.args()["description"]
+                        .as_str()
+                        .or(answer.args()["type"].as_str())
+                        .unwrap_or("no reason")
+                ));
+            }
+            Ok(Value::Object(answer.1))
+        })
+    }
 }
 
 /// A relay's `notCreated` for a message, as this server's refusal.

crates/smtp/src/outbound/relay.rs

@@ -17,8 +17,8 @@
 
 use super::DeliveryResult;
 use crate::queue::{
-    Error, ErrorDetails, HostResponse, MessageWrapper, RCPT_RELAYED, Status, UnexpectedResponse,
-    spool::SmtpSpool,
+    Error, ErrorDetails, HostResponse, MessageWrapper, RCPT_DSN_SENT, RCPT_RELAYED, Status,
+    UnexpectedResponse, spool::SmtpSpool,
 };
 use crate::{queue::RecipientDomain, reporting::send::MtaReportSend};
 use common::{Server, config::smtp::queue::QueueName, ipc::QueueEvent};
@@ -274,6 +274,10 @@ pub enum RelayedStatus {
     },
     /// The relay no longer has the message: hand it over again.
     Lost,
+    /// An administrator cancelled it (`inbuxa:RelayMessage/set` destroy):
+    /// failed, like a delivery cancelled in this server's queue, and with
+    /// no DSN.
+    Cancelled,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -281,7 +285,9 @@ pub enum Applied {
     Done,
     /// The queue is working on the message: try again later.
     Locked,
-    /// The message is no longer queued here.
+    /// The message is no longer queued here, or none of these recipients
+    /// waits on the relay any more (cancelled, or expired here): the relay
+    /// can stop.
     Gone,
 }
 
@@ -308,6 +314,7 @@ pub async fn apply_relayed(
     let now = now();
     let mut changed = false;
     let mut wake = false;
+    let mut waiting = false;
     for rcpt in message.message.recipients.iter_mut() {
         // Expired here meanwhile: this server's word stands
         if rcpt.flags & RCPT_RELAYED == 0
@@ -324,6 +331,7 @@ pub async fn apply_relayed(
         else {
             continue;
         };
+        waiting = true;
         let entity = |mx: &Option<String>| -> Box<str> { mx.as_deref().unwrap_or(relay).into() };
         let new = match status {
             RelayedStatus::Delivered { reply, mx } => Status::Completed(HostResponse {
@@ -351,10 +359,20 @@ pub async fn apply_relayed(
                 entity: relay.into(),
                 details: Error::Io("The relay no longer has the message.".into()),
             }),
+            RelayedStatus::Cancelled => {
+                rcpt.flags |= RCPT_DSN_SENT;
+                Status::PermanentFailure(ErrorDetails {
+                    entity: relay.into(),
+                    details: Error::Io("Delivery canceled.".into()),
+                })
+            }
         };
         if matches!(
             status,
-            RelayedStatus::Delivered { .. } | RelayedStatus::Refused { .. } | RelayedStatus::Lost
+            RelayedStatus::Delivered { .. }
+                | RelayedStatus::Refused { .. }
+                | RelayedStatus::Lost
+                | RelayedStatus::Cancelled
         ) {
             rcpt.flags &= !RCPT_RELAYED;
             rcpt.retry.due = now;
@@ -365,6 +383,10 @@ pub async fn apply_relayed(
             changed = true;
         }
     }
+    if !waiting {
+        server.unlock_event(queue_id, queue_name).await;
+        return Applied::Gone;
+    }
     if changed || wake {
         message
             .save_changes(server, None, wake.then_some(now))

docs/spec/features/relay.md

@@ -508,7 +508,7 @@ When no link is up, or after RL-13:
 | Object | What it holds |
 |---|---|
 | `inbuxa:Relay` | One per relay: `name`, `address`, `linkPort`, `enabled`, the settings pushed to the relay, `sendThrough` (route outbound mail through relays), `inboundOnly` (takes mail in but sends none, e.g. while its IP isn't warmed up or isn't in SPF yet), and read-only status mirrored from the relay: link state per node, version, queue counts, `lastSeen`. Create, or an update with `rejoin: true`, returns `joinToken` once |
-| `inbuxa:RelayMessage` | Read-only view of a relay's `Inbound` and `Outbound` queues, fetched over the link when asked. Never message content |
+| `inbuxa:RelayMessage` | Read-only view of a relay's `Inbound` and `Outbound` queues, fetched over the link when asked. Never message content. `/get`, `/query` (filter `relayId`, `direction`; newest first) and `/set` destroy only: discards held mail, or cancels an outbound message, whose recipients fail here with no DSN as a delivery cancelled in the queue does. Each id is the relay's id and a hash of the relay's own message id, so nothing is stored |
 
 Permissions: `sysRelayGet`, `sysRelayUpdate`, `sysRelayQueueGet`,
 `sysRelayQueueUpdate` (cancel an outbound message, discard held mail). The
@@ -858,8 +858,17 @@ In progress. Done so far:
   `crates/common`. The server keeps its caches, trace events, policy fetch
   and test mocks around them, so its behavior is unchanged. The relay will
   use the same code to deliver.
-- Not done yet: `inbuxa:RelayMessage`, which comes with inbuxa-admin's
-  Relays page (step 5), once `Outbound` exists to show beside `Inbound`.
+- `inbuxa:RelayMessage` (step 5, for inbuxa-admin's Relays page): get,
+  query and destroy, over the asking node's link, with `sysRelayQueueGet`
+  and `sysRelayQueueUpdate`, and every destroy audited. A relay this node
+  can't reach is left out of a query of all relays; asked for by name, it
+  is an error (`serverUnavailable`). An outbound message cancelled while
+  the relay is sending it is dropped at the relay's next status. Tested
+  end to end: the deferred message read back with the remote server's
+  reply and its queue id, then cancelled, gone from both queues, with no
+  DSN.
+- Still to come in step 5: inbuxa-admin's Relays page, certificates
+  (RL-27), DNS (RL-25) and upgrades (RL-29).
 
 Order:
 

resources/privacy/catalog.toml

@@ -131,6 +131,20 @@ joinToken = ["credential"]
 links = ["network"]
 createdBy = ["identifier"]
 
+# Kept at the relay, read over the link when asked; nothing stored here
+[object."inbuxa:RelayMessage"]
+file = "inbuxa_relay_message.rs"
+default = "none"
+whose = ["holder", "correspondent"]
+where = ["external"]
+scope = "server"
+retention = "object-life"
+[object."inbuxa:RelayMessage".properties]
+envelope = ["identifier", "contact"]
+session = ["network", "identifier"]
+deliveryStatus = ["identifier", "network"]
+doorChecks = ["network"]
+
 [object."inbuxa:Journal"]
 file = "inbuxa_journal.rs"
 default = "none"

tests/src/system/relay.rs

@@ -1026,6 +1026,92 @@ pub async fn link(test: &mut TestServer) {
         tokio::time::sleep(std::time::Duration::from_millis(500)).await;
     }
 
+    // The relay's queues, read over the link (inbuxa:RelayMessage): the
+    // deferred message, with the remote server's reply and its place in
+    // this server's queue, and never the message itself
+    let (name, response) = call(
+        admin,
+        "inbuxa:RelayMessage/query",
+        json!({"filter": {"relayId": &id, "direction": "outbound"}, "calculateTotal": true}),
+    )
+    .await;
+    assert_eq!(name, "inbuxa:RelayMessage/query", "{response}");
+    let ids = response["ids"].as_array().cloned().unwrap_or_default();
+    assert!(ids.len() >= 2, "{response}");
+    let (_, response) = call(admin, "inbuxa:RelayMessage/get", json!({"ids": ids})).await;
+    assert!(!response.to_string().contains("blobId"), "{response}");
+    let later = response["list"]
+        .as_array()
+        .into_iter()
+        .flatten()
+        .find(|m| m["envelope"]["rcptTo"][0]["email"] == "[email protected]")
+        .cloned()
+        .unwrap_or_else(|| panic!("the deferred message: {response}"));
+    assert_eq!(later["relayId"], json!(&id), "{later}");
+    assert_eq!(later["direction"], "outbound", "{later}");
+    assert!(
+        later["deliveryStatus"]["[email protected]"]["smtpReply"]
+            .as_str()
+            .is_some_and(|r| r.contains("Try again later")),
+        "{later}"
+    );
+    assert!(later["queueId"].is_string(), "{later}");
+    assert!(
+        later["state"].is_null() && later["session"].is_null(),
+        "{later}"
+    );
+    // Cancelled: gone from the relay and from this server's queue, with no
+    // DSN (the count below)
+    let (_, response) = call(
+        admin,
+        "inbuxa:RelayMessage/set",
+        json!({"destroy": [&later["id"]], "reason": "Sent by mistake"}),
+    )
+    .await;
+    assert_eq!(response["destroyed"], json!([&later["id"]]), "{response}");
+    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
+    loop {
+        let (_, queued) = call(admin, "x:QueuedMessage/get", json!({"ids": null})).await;
+        let (_, at_relay) = call(
+            admin,
+            "inbuxa:RelayMessage/get",
+            json!({"ids": [&later["id"]]}),
+        )
+        .await;
+        let here = queued.to_string().contains("[email protected]");
+        let there = at_relay["notFound"] != json!([&later["id"]]);
+        if !here && !there {
+            break;
+        }
+        assert!(
+            std::time::Instant::now() < deadline,
+            "queued here: {here}, at the relay: {there}"
+        );
+        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
+    }
+    // A relay message that isn't there, and one that can't be changed
+    let other = ids
+        .iter()
+        .filter_map(|i| i.as_str())
+        .find(|i| *i != later["id"])
+        .unwrap()
+        .to_string();
+    let (_, response) = call(
+        admin,
+        "inbuxa:RelayMessage/set",
+        json!({"destroy": [&later["id"]], "update": {&other: {"size": 1}}}),
+    )
+    .await;
+    assert_eq!(
+        response["notDestroyed"][later["id"].as_str().unwrap()]["type"],
+        "notFound",
+        "{response}"
+    );
+    assert_eq!(
+        response["notUpdated"][&other]["type"], "forbidden",
+        "{response}"
+    );
+
     // A route naming relays (RL-24): with sendThrough off, only the
     // domains routed to it go through a relay, and only the relays named.
     // One named relay that isn't there: the mail waits, saying so