From 00db2d8a7a56e49fc6652456b6beb5039f2366a5 Mon Sep 17 00:00:00 2001 From: John Coffey Date: Fri, 30 Oct 2026 13:25:12 -0700 Subject: [PATCH] Relay: inbuxa:RelayMessage, a relay's queues read over the link The Relays page needs to show what waits at each relay, and to let an administrator discard held mail or cancel an outbound message, without ever talking to a relay itself (relay spec, "Registry objects"). inbuxa:RelayMessage has get, query and set (destroy only) under urn:inbuxa:jmap, with sysRelayQueueGet and sysRelayQueueUpdate. The asking node reads the relay's Inbound and Outbound over its own link, through a new call on the link's sender, and shows each message's envelope, size, times and state: the session and door checks for inbound mail, the per-recipient status and retry state for outbound, and the message in this server's queue it was handed from. Never the blob. Query filters by relayId and direction, newest first. Nothing is stored: an id is the relay's id (top 16 bits) and a hash of the relay's own message id, so every node names a message the same way, and a lookup checks it against the relay's queues. A relay this node can't reach is left out of a query of all relays and is an error when asked for by itself. Destroying held mail destroys it at the relay, unless a node is taking it just then. Cancelling an outbound message fails its recipients here first, with no DSN, the way a delivery cancelled in the queue does, so the queue never hands it over again; then the relay drops it. If the relay is sending it at that moment, it drops it at its next status: the follower now also destroys the relay's copy when none of its recipients waits on the relay any more. Every destroy is audited, with its reason. Test: unit test for the ids. The relay system test reads the relay's outbound queue, finds the deferred message with the remote server's reply and its queue id and no blob, cancels it, and sees it gone from both queues with no DSN; a second destroy is notFound and an update is forbidden. --- crates/features/src/relay/messages.rs | 53 ++ crates/features/src/relay/mod.rs | 1 + crates/features/src/relay/outbound.rs | 21 + .../src/object/inbuxa_relay_message.rs | 268 ++++++++++ crates/jmap-proto/src/object/mod.rs | 1 + crates/jmap-proto/src/references/eval.rs | 3 + crates/jmap-proto/src/references/resolve.rs | 4 + crates/jmap-proto/src/request/method.rs | 9 + crates/jmap-proto/src/request/mod.rs | 3 + crates/jmap-proto/src/request/parser.rs | 21 + crates/jmap-proto/src/response/mod.rs | 18 + crates/jmap/src/api/auth.rs | 12 + crates/jmap/src/api/request.rs | 33 ++ crates/jmap/src/changes/get.rs | 1 + crates/jmap/src/inbuxa/mod.rs | 1 + crates/jmap/src/inbuxa/relay.rs | 9 +- crates/jmap/src/inbuxa/relay_message.rs | 476 ++++++++++++++++++ crates/services/src/inbuxa_relay_link.rs | 28 +- crates/smtp/src/outbound/relay.rs | 30 +- docs/spec/features/relay.md | 15 +- resources/privacy/catalog.toml | 14 + tests/src/system/relay.rs | 86 ++++ 22 files changed, 1097 insertions(+), 10 deletions(-) create mode 100644 crates/features/src/relay/messages.rs create mode 100644 crates/jmap-proto/src/object/inbuxa_relay_message.rs create mode 100644 crates/jmap/src/inbuxa/relay_message.rs diff --git a/crates/features/src/relay/messages.rs b/crates/features/src/relay/messages.rs new file mode 100644 index 00000000..dc379c53 --- /dev/null +++ b/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 { + (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); + } +} diff --git a/crates/features/src/relay/mod.rs b/crates/features/src/relay/mod.rs index 6c17f230..f28138e3 100644 --- a/crates/features/src/relay/mod.rs +++ b/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::{ diff --git a/crates/features/src/relay/outbound.rs b/crates/features/src/relay/outbound.rs index a44eca2f..736e2e95 100644 --- a/crates/features/src/relay/outbound.rs +++ b/crates/features/src/relay/outbound.rs @@ -47,10 +47,16 @@ pub enum Refused { pub type HandOver<'a> = Pin> + Send + 'a>>; +pub type Call<'a> = Pin> + 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, 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> { + 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) { let mut state = self.state.lock().expect("senders"); @@ -245,6 +262,10 @@ mod tests { fn hand_over(&self, _: Vec, _: 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 { diff --git a/crates/jmap-proto/src/object/inbuxa_relay_message.rs b/crates/jmap-proto/src/object/inbuxa_relay_message.rs new file mode 100644 index 00000000..c2fd9837 --- /dev/null +++ b/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 { + // 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 { + 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 { + RelayMessageProperty::parse(s).ok_or(()) + } +} + +impl Element for RelayMessageValue { + type Property = RelayMessageProperty; + + fn try_parse

(key: &Key<'_, Self::Property>, value: &str) -> Option { + 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(&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::()?; + } + } + Ok(()) + } +} + +/// The set call's own argument: why, for the audit log. +#[derive(Debug, Clone, Default)] +pub struct RelayMessageSetArguments { + pub reason: Option, +} + +impl<'de> DeserializeArguments<'de> for RelayMessageSetArguments { + fn deserialize_argument(&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::()?; + } + 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 for RelayMessageValue { + fn from(id: Id) -> Self { + RelayMessageValue::Id(id) + } +} + +impl JmapObjectId for RelayMessageValue { + fn as_id(&self) -> Option { + match self { + RelayMessageValue::Id(id) => Some(*id), + } + } + + fn as_any_id(&self) -> Option { + 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 { + None + } + + fn as_any_id(&self) -> Option { + None + } + + fn as_id_ref(&self) -> Option<&str> { + None + } + + fn try_set_id(&mut self, _: AnyId) -> bool { + false + } +} diff --git a/crates/jmap-proto/src/object/mod.rs b/crates/jmap-proto/src/object/mod.rs index ad828cd8..b158c737 100644 --- a/crates/jmap-proto/src/object/mod.rs +++ b/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 diff --git a/crates/jmap-proto/src/references/eval.rs b/crates/jmap-proto/src/references/eval.rs index 193e234b..e1af0805 100644 --- a/crates/jmap-proto/src/references/eval.rs +++ b/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) } diff --git a/crates/jmap-proto/src/references/resolve.rs b/crates/jmap-proto/src/references/resolve.rs index 40200f36..5c7bdafb 100644 --- a/crates/jmap-proto/src/references/resolve.rs +++ b/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)? } diff --git a/crates/jmap-proto/src/request/method.rs b/crates/jmap-proto/src/request/method.rs index ddc2dcc5..adca09cc 100644 --- a/crates/jmap-proto/src/request/method.rs +++ b/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", diff --git a/crates/jmap-proto/src/request/mod.rs b/crates/jmap-proto/src/request/mod.rs index 807b13ec..01bde4cf 100644 --- a/crates/jmap-proto/src/request/mod.rs +++ b/crates/jmap-proto/src/request/mod.rs @@ -128,6 +128,7 @@ pub enum GetRequestMethod { LegalHold(Box>), MailRule(Box>), Relay(Box>), + RelayMessage(Box>), SecurityAcceptance( Box>, ), @@ -185,6 +186,7 @@ pub enum SetRequestMethod<'x> { LegalHold(Box>), MailRule(Box>), Relay(Box>), + RelayMessage(Box>), SecurityAcceptance( Box>, ), @@ -246,6 +248,7 @@ pub enum QueryRequestMethod { Registry(Box>), AuditEvent(Box>), JournalEntry(Box>), + RelayMessage(Box>), } #[derive(Debug)] diff --git a/crates/jmap-proto/src/request/parser.rs b/crates/jmap-proto/src/request/parser.rs index a598c0bf..3d4144fb 100644 --- a/crates/jmap-proto/src/request/parser.rs +++ b/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)), diff --git a/crates/jmap-proto/src/response/mod.rs b/crates/jmap-proto/src/response/mod.rs index ccbab85e..68f3d69f 100644 --- a/crates/jmap-proto/src/response/mod.rs +++ b/crates/jmap-proto/src/response/mod.rs @@ -113,6 +113,7 @@ pub enum GetResponseMethod { LegalHold(GetResponse), MailRule(GetResponse), Relay(GetResponse), + RelayMessage(GetResponse), SecurityAcceptance(GetResponse), DeliverabilityReport( GetResponse, @@ -168,6 +169,7 @@ pub enum SetResponseMethod { LegalHold(Box>), MailRule(Box>), Relay(Box>), + RelayMessage(Box>), SecurityAcceptance( Box>, ), @@ -1047,6 +1049,22 @@ impl<'x> From> for ResponseMetho } } +impl<'x> From> + for ResponseMethod<'x> +{ + fn from(value: GetResponse) -> Self { + ResponseMethod::Get(GetResponseMethod::RelayMessage(value)) + } +} + +impl<'x> From> + for ResponseMethod<'x> +{ + fn from(value: SetResponse) -> Self { + ResponseMethod::Set(SetResponseMethod::RelayMessage(Box::new(value))) + } +} + // inbuxa: journaling impl<'x> From> for ResponseMethod<'x> diff --git a/crates/jmap/src/api/auth.rs b/crates/jmap/src/api/auth.rs index b1040249..985ace27 100644 --- a/crates/jmap/src/api/auth.rs +++ b/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!() diff --git a/crates/jmap/src/api/request.rs b/crates/jmap/src/api/request.rs index dd528df2..e770c653 100644 --- a/crates/jmap/src/api/request.rs +++ b/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) => { diff --git a/crates/jmap/src/changes/get.rs b/crates/jmap/src/changes/get.rs index bf105c9c..263f9ddd 100644 --- a/crates/jmap/src/changes/get.rs +++ b/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 diff --git a/crates/jmap/src/inbuxa/mod.rs b/crates/jmap/src/inbuxa/mod.rs index cfb20291..2fb0e7eb 100644 --- a/crates/jmap/src/inbuxa/mod.rs +++ b/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; diff --git a/crates/jmap/src/inbuxa/relay.rs b/crates/jmap/src/inbuxa/relay.rs index 64f303b0..eede336a 100644 --- a/crates/jmap/src/inbuxa/relay.rs +++ b/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(json: serde_json::Value) -> Value<'static, P, E> +where + P: Property, + E: jmap_tools::Element, +{ match json { serde_json::Value::Null => Value::Null, serde_json::Value::Bool(b) => Value::Bool(b), diff --git a/crates/jmap/src/inbuxa/relay_message.rs b/crates/jmap/src/inbuxa/relay_message.rs new file mode 100644 index 00000000..07c2d36d --- /dev/null +++ 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, + messages: Vec, + /// Its outbound messages' places in this server's queue. + handed: HashMap, +} + +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 { + 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, 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 { + 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, +) -> trc::Result> { + 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, +) -> trc::Result { + 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

), +} + +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> { + 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 = 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) +} diff --git a/crates/services/src/inbuxa_relay_link.rs b/crates/services/src/inbuxa_relay_link.rs index c62f7f5a..a67154a4 100644 --- a/crates/services/src/inbuxa_relay_link.rs +++ b/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. diff --git a/crates/smtp/src/outbound/relay.rs b/crates/smtp/src/outbound/relay.rs index 054d1e6c..8ba26f18 100644 --- a/crates/smtp/src/outbound/relay.rs +++ b/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| -> Box { 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)) diff --git a/docs/spec/features/relay.md b/docs/spec/features/relay.md index 1f208771..f1e2cfbc 100644 --- a/docs/spec/features/relay.md +++ b/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: diff --git a/resources/privacy/catalog.toml b/resources/privacy/catalog.toml index d48a3035..8d8a1720 100644 --- a/resources/privacy/catalog.toml +++ b/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" diff --git a/tests/src/system/relay.rs b/tests/src/system/relay.rs index 6cf5809b..2a4e61dd 100644 --- a/tests/src/system/relay.rs +++ b/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"] == "later@remote.test") + .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"]["later@remote.test"]["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("later@remote.test"); + 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 -- 2.54.0