relay: hand outbound mail to relays and follow its status #261
No files matched your search
Generated
+1
@@ -8162,6 +8162,7 @@ dependencies = [
|
||||
"hashify",
|
||||
"inbuxa-features",
|
||||
"inbuxa-outbound-tls",
|
||||
"inbuxa-relay-proto",
|
||||
"mail-auth",
|
||||
"mail-builder 1.0.0",
|
||||
"mail-parser",
|
||||
|
||||
@@ -100,6 +100,7 @@ impl Data {
|
||||
smtp_connectors: TlsConnectors::try_new().failed("Failed to build TLS connectors"),
|
||||
build_errors: Default::default(),
|
||||
audit: Default::default(),
|
||||
relay_senders: Default::default(),
|
||||
asn_geo_data: Default::default(),
|
||||
}
|
||||
}
|
||||
@@ -246,6 +247,7 @@ impl Default for Data {
|
||||
smtp_connectors: TlsConnectors::try_new().unwrap(),
|
||||
build_errors: Default::default(),
|
||||
audit: Default::default(),
|
||||
relay_senders: Default::default(),
|
||||
asn_geo_data: Default::default(),
|
||||
lookup_stores: Default::default(),
|
||||
}
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use super::*;
|
||||
@@ -78,6 +80,8 @@ pub enum RoutingStrategy {
|
||||
Local,
|
||||
Mx(MxConfig),
|
||||
Relay(RelayConfig),
|
||||
// inbuxa: handed to an inbuxa Relay over its link (relay spec, RL-24)
|
||||
InbuxaRelay,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
|
||||
@@ -194,6 +194,10 @@ pub struct Data {
|
||||
|
||||
// inbuxa: the audit log's chain heads and recent-access marks (AU)
|
||||
pub audit: inbuxa_features::audit::AuditLog,
|
||||
|
||||
// inbuxa: the relays outbound mail goes through, and this node's links
|
||||
// to them (relay spec, RL-20, RL-24)
|
||||
pub relay_senders: inbuxa_features::relay::outbound::Senders,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
|
||||
@@ -17,10 +17,14 @@
|
||||
//! - `r` + relay id (u32): a relay ([`Relay`]), as JSON.
|
||||
//! - `s` + relay id (u32) + node id (u64): that node's link to the relay
|
||||
//! ([`link::NodeLink`]), as JSON.
|
||||
//! - `i` + relay id (u32) + `Inbound` id: a message taken from the relay.
|
||||
//! - `o` + relay id (u32) + `Outbound` id: a message handed to the relay
|
||||
//! ([`outbound::Handed`]), as JSON.
|
||||
//!
|
||||
//! Numbers are big-endian. There are few relays, so they're read whole.
|
||||
|
||||
pub mod link;
|
||||
pub mod outbound;
|
||||
|
||||
use inbuxa_relay_proto::{
|
||||
b64,
|
||||
@@ -52,6 +56,7 @@ const KIND_KEYS: u8 = b'k';
|
||||
const KIND_RELAY: u8 = b'r';
|
||||
const KIND_STATUS: u8 = b's';
|
||||
const KIND_INGESTED: u8 = b'i';
|
||||
const KIND_HANDED: u8 = b'o';
|
||||
const CREATE_ATTEMPTS: usize = 5;
|
||||
|
||||
const MAX_NAME: usize = 100;
|
||||
|
||||
@@ -0,0 +1,305 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs LLC
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! Outbound mail through relays, the downstream's side (relay spec,
|
||||
//! "Outbound routing", RL-20, RL-24): whether mail goes through relays,
|
||||
//! which of them this node can hand a message to, and the messages handed
|
||||
//! to each until it reports a final status.
|
||||
//!
|
||||
//! A message handed over is kept under `o` + relay id (u32) + the relay's
|
||||
//! `Outbound` id ([`Handed`]), as JSON, in the shared store: on a cluster
|
||||
//! the node that follows a relay's statuses needn't be the one that handed
|
||||
//! the message over.
|
||||
|
||||
use super::{
|
||||
FEATURE, Json, KIND_HANDED, Relay,
|
||||
link::{INBOUND_ID_MAX, is_inbound_id},
|
||||
};
|
||||
use inbuxa_relay_proto::types::Envelope;
|
||||
use serde::{Deserialize as SerdeDeserialize, Serialize as SerdeSerialize};
|
||||
use std::{
|
||||
collections::{BTreeMap, HashSet},
|
||||
future::Future,
|
||||
pin::Pin,
|
||||
sync::{
|
||||
Arc, Mutex,
|
||||
atomic::{AtomicUsize, Ordering},
|
||||
},
|
||||
};
|
||||
use store::{
|
||||
Deserialize, IterateParams, SUBSPACE_INBUXA, Serialize, Store, ValueKey,
|
||||
write::{AnyClass, BatchBuilder, ValueClass},
|
||||
};
|
||||
use trc::AddContext;
|
||||
|
||||
/// Why a relay didn't take a message.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Refused {
|
||||
/// No relay will: the message is too large, or its sender isn't at
|
||||
/// one of the relays' domains. An SMTP reply, e.g. `552 5.3.4 ...`.
|
||||
Permanent(String),
|
||||
/// Not this relay, or not now: the link failed, or its spool is full.
|
||||
Temporary(String),
|
||||
}
|
||||
|
||||
pub type HandOver<'a> = Pin<Box<dyn Future<Output = Result<String, Refused>> + 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<'_>;
|
||||
}
|
||||
|
||||
/// A relay this node is linked to.
|
||||
pub struct Linked {
|
||||
pub relay_id: u32,
|
||||
pub name: String,
|
||||
pub sender: Box<dyn Sender>,
|
||||
}
|
||||
|
||||
/// The relays mail goes through, as this node sees them. Read from the
|
||||
/// registry by the link service, and on first use.
|
||||
#[derive(Default)]
|
||||
pub struct Senders {
|
||||
state: Mutex<State>,
|
||||
next: AtomicUsize,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct State {
|
||||
/// The relays were read at least once.
|
||||
known: bool,
|
||||
/// An enabled relay has `sendThrough` on (RL-24).
|
||||
send_through: bool,
|
||||
/// Enabled relays not marked `inboundOnly` (RL-20).
|
||||
may_send: HashSet<u32>,
|
||||
linked: BTreeMap<u32, Arc<Linked>>,
|
||||
}
|
||||
|
||||
impl Senders {
|
||||
/// Takes the relays as they are in the registry now.
|
||||
pub fn set_relays(&self, relays: &[Relay]) {
|
||||
let mut state = self.state.lock().expect("senders");
|
||||
state.known = true;
|
||||
state.send_through = relays.iter().any(|r| r.enabled && r.send_through);
|
||||
state.may_send = relays
|
||||
.iter()
|
||||
.filter(|r| r.enabled && !r.inbound_only)
|
||||
.map(|r| r.id)
|
||||
.collect();
|
||||
}
|
||||
|
||||
/// Whether the relays were read yet.
|
||||
pub fn known(&self) -> bool {
|
||||
self.state.lock().expect("senders").known
|
||||
}
|
||||
|
||||
/// Reads the relays, if nothing has yet.
|
||||
pub async fn load(&self, data: &Store) -> trc::Result<()> {
|
||||
if !self.known() {
|
||||
let relays = super::all(data).await?;
|
||||
self.set_relays(&relays);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Whether every non-local recipient goes through a relay (RL-24). With
|
||||
/// none of them linked, mail waits in this server's queue (RL-20).
|
||||
pub fn sending(&self) -> bool {
|
||||
self.state.lock().expect("senders").send_through
|
||||
}
|
||||
|
||||
pub fn link_up(&self, linked: Arc<Linked>) {
|
||||
self.state
|
||||
.lock()
|
||||
.expect("senders")
|
||||
.linked
|
||||
.insert(linked.relay_id, linked);
|
||||
}
|
||||
|
||||
/// 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");
|
||||
if state
|
||||
.linked
|
||||
.get(&linked.relay_id)
|
||||
.is_some_and(|current| Arc::ptr_eq(current, linked))
|
||||
{
|
||||
state.linked.remove(&linked.relay_id);
|
||||
}
|
||||
}
|
||||
|
||||
/// The relays to try for one message, in order: those this node is
|
||||
/// linked to and that may send, each message starting with the next
|
||||
/// one in turn (RL-20).
|
||||
pub fn candidates(&self) -> Vec<Arc<Linked>> {
|
||||
let state = self.state.lock().expect("senders");
|
||||
let mut linked: Vec<Arc<Linked>> = state
|
||||
.linked
|
||||
.values()
|
||||
.filter(|l| state.may_send.contains(&l.relay_id))
|
||||
.cloned()
|
||||
.collect();
|
||||
if !linked.is_empty() {
|
||||
let first = self.next.fetch_add(1, Ordering::Relaxed) % linked.len();
|
||||
linked.rotate_left(first);
|
||||
}
|
||||
linked
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Messages handed over ---------------------------------------------------
|
||||
|
||||
/// A message handed to a relay: which queued message, and which of its
|
||||
/// recipients, as the envelope named them.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct Handed {
|
||||
/// The relay's `Outbound` id.
|
||||
pub id: String,
|
||||
pub queue_id: u64,
|
||||
pub queue_name: String,
|
||||
pub recipients: Vec<String>,
|
||||
pub at: u64,
|
||||
}
|
||||
|
||||
fn handed_class(relay_id: u32, outbound_id: &[u8]) -> ValueClass {
|
||||
let mut key = Vec::with_capacity(6 + outbound_id.len());
|
||||
key.push(FEATURE);
|
||||
key.push(KIND_HANDED);
|
||||
key.extend_from_slice(&relay_id.to_be_bytes());
|
||||
key.extend_from_slice(outbound_id);
|
||||
ValueClass::Any(AnyClass {
|
||||
subspace: SUBSPACE_INBUXA,
|
||||
key,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn record_handed(data: &Store, relay_id: u32, handed: &Handed) -> trc::Result<()> {
|
||||
if !is_inbound_id(&handed.id) {
|
||||
return Err(trc::StoreEvent::UnexpectedError
|
||||
.into_err()
|
||||
.details("A relay's outbound id isn't one")
|
||||
.ctx(trc::Key::Id, handed.id.clone()));
|
||||
}
|
||||
let mut batch = BatchBuilder::new();
|
||||
batch.set(
|
||||
handed_class(relay_id, handed.id.as_bytes()),
|
||||
Json(handed).serialize()?,
|
||||
);
|
||||
data.write(batch.build_all())
|
||||
.await
|
||||
.caused_by(trc::location!())?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// The messages handed to a relay with no final status yet.
|
||||
pub async fn handed(data: &Store, relay_id: u32) -> trc::Result<Vec<Handed>> {
|
||||
let mut handed = Vec::new();
|
||||
data.iterate(
|
||||
IterateParams::new(
|
||||
ValueKey::from(handed_class(relay_id, &[])),
|
||||
ValueKey::from(handed_class(relay_id, &[u8::MAX; INBOUND_ID_MAX + 1])),
|
||||
),
|
||||
|_, value| {
|
||||
if let Ok(Json(one)) = Json::<Handed>::deserialize(value) {
|
||||
handed.push(one);
|
||||
}
|
||||
Ok(true)
|
||||
},
|
||||
)
|
||||
.await
|
||||
.caused_by(trc::location!())?;
|
||||
Ok(handed)
|
||||
}
|
||||
|
||||
pub async fn forget_handed(data: &Store, relay_id: u32, ids: &[String]) -> trc::Result<()> {
|
||||
for chunk in ids.chunks(500) {
|
||||
let mut batch = BatchBuilder::new();
|
||||
for id in chunk {
|
||||
batch.clear(handed_class(relay_id, id.as_bytes()));
|
||||
}
|
||||
data.write(batch.build_all())
|
||||
.await
|
||||
.caused_by(trc::location!())?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
struct Nothing;
|
||||
|
||||
impl Sender for Nothing {
|
||||
fn hand_over(&self, _: Vec<u8>, _: Envelope) -> HandOver<'_> {
|
||||
Box::pin(async { Err(Refused::Temporary("no".into())) })
|
||||
}
|
||||
}
|
||||
|
||||
fn relay(id: u32, send_through: bool, inbound_only: bool) -> Relay {
|
||||
let mut relay: Relay = serde_json::from_value(serde_json::json!({
|
||||
"name": format!("r{id}"),
|
||||
"address": "relay.example.com",
|
||||
}))
|
||||
.unwrap();
|
||||
relay.id = id;
|
||||
relay.send_through = send_through;
|
||||
relay.inbound_only = inbound_only;
|
||||
relay
|
||||
}
|
||||
|
||||
fn linked(relay_id: u32) -> Arc<Linked> {
|
||||
Arc::new(Linked {
|
||||
relay_id,
|
||||
name: format!("r{relay_id}"),
|
||||
sender: Box::new(Nothing),
|
||||
})
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rotation_skips_inbound_only() {
|
||||
let senders = Senders::default();
|
||||
assert!(!senders.known() && !senders.sending());
|
||||
// Off until a relay has sendThrough on
|
||||
senders.set_relays(&[relay(1, false, false), relay(2, false, true)]);
|
||||
assert!(senders.known() && !senders.sending());
|
||||
senders.set_relays(&[
|
||||
relay(1, true, false),
|
||||
relay(2, false, true),
|
||||
relay(3, false, false),
|
||||
]);
|
||||
assert!(senders.sending());
|
||||
// Nothing linked: nothing to try, and the mail waits
|
||||
assert!(senders.candidates().is_empty());
|
||||
for id in [1, 2, 3] {
|
||||
senders.link_up(linked(id));
|
||||
}
|
||||
let mut firsts = Vec::new();
|
||||
for _ in 0..4 {
|
||||
let ids: Vec<u32> = senders.candidates().iter().map(|l| l.relay_id).collect();
|
||||
assert_eq!(ids.len(), 2, "{ids:?}");
|
||||
assert!(!ids.contains(&2), "inboundOnly: {ids:?}");
|
||||
firsts.push(ids[0]);
|
||||
}
|
||||
assert_eq!(firsts, [1, 3, 1, 3]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_old_link_leaves_a_new_one() {
|
||||
let senders = Senders::default();
|
||||
senders.set_relays(&[relay(1, true, false)]);
|
||||
let old = linked(1);
|
||||
let new = linked(1);
|
||||
senders.link_up(old.clone());
|
||||
senders.link_up(new.clone());
|
||||
senders.link_down(&old);
|
||||
assert_eq!(senders.candidates().len(), 1);
|
||||
senders.link_down(&new);
|
||||
assert!(senders.candidates().is_empty());
|
||||
}
|
||||
}
|
||||
@@ -401,6 +401,9 @@ pub struct DeliveryStatus {
|
||||
pub delivered: Delivered,
|
||||
/// The relay never learns this.
|
||||
pub displayed: Displayed,
|
||||
/// The MX host the reply came from, when one answered. Not in RFC 8621.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub mx: Option<String>,
|
||||
}
|
||||
|
||||
impl DeliveryStatus {
|
||||
@@ -409,6 +412,7 @@ impl DeliveryStatus {
|
||||
smtp_reply: String::new(),
|
||||
delivered: Delivered::Queued,
|
||||
displayed: Displayed::Unknown,
|
||||
mx: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -32,6 +32,11 @@ futures-util = "0.3.34"
|
||||
serde = { version = "1.0", features = ["derive"] }
|
||||
serde_json = "1.0"
|
||||
|
||||
[features]
|
||||
# For the server's system tests only: `INBUXA_RELAY_TEST_MX=<port>` sends
|
||||
# every domain's mail to 127.0.0.1 on that port. Never in a release.
|
||||
test-mx = []
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
|
||||
@@ -146,11 +146,12 @@ struct Target<'a, R> {
|
||||
rcpts: &'a [&'a Address],
|
||||
}
|
||||
|
||||
fn status(delivered: Delivered, smtp_reply: String) -> DeliveryStatus {
|
||||
fn status(delivered: Delivered, smtp_reply: String, mx: Option<&str>) -> DeliveryStatus {
|
||||
DeliveryStatus {
|
||||
smtp_reply,
|
||||
delivered,
|
||||
displayed: Displayed::Unknown,
|
||||
mx: mx.map(str::to_string),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -158,7 +159,7 @@ impl<R: Resolve> Target<'_, R> {
|
||||
fn every(&self, delivered: Delivered, reply: String) -> Vec<(String, DeliveryStatus)> {
|
||||
self.rcpts
|
||||
.iter()
|
||||
.map(|r| (r.email.clone(), status(delivered, reply.clone())))
|
||||
.map(|r| (r.email.clone(), status(delivered, reply.clone(), None)))
|
||||
.collect()
|
||||
}
|
||||
|
||||
@@ -288,7 +289,7 @@ impl<R: Resolve> Target<'_, R> {
|
||||
RcptOutcome::Deferred(r) => (Delivered::Queued, r, 2),
|
||||
};
|
||||
counts[n] += 1;
|
||||
(rcpt.email.clone(), status(delivered, reply))
|
||||
(rcpt.email.clone(), status(delivered, reply, Some(host)))
|
||||
})
|
||||
.collect();
|
||||
self.app.record(
|
||||
|
||||
@@ -301,3 +301,36 @@ impl Resolve for Dns {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The server's system tests: every domain is its own MX, at 127.0.0.1,
|
||||
/// unsigned and without MTA-STS.
|
||||
#[cfg(feature = "test-mx")]
|
||||
pub struct TestMx {
|
||||
pub port: u16,
|
||||
}
|
||||
|
||||
#[cfg(feature = "test-mx")]
|
||||
impl Resolve for TestMx {
|
||||
async fn mx(&self, domain: &str) -> Result<Mx, LookupError> {
|
||||
Ok(Mx::Hosts {
|
||||
hosts: vec![domain.to_string()],
|
||||
secure: false,
|
||||
})
|
||||
}
|
||||
|
||||
async fn addresses(&self, _: &str, _: bool) -> Result<(Vec<IpAddr>, bool), LookupError> {
|
||||
Ok((vec![IpAddr::V4(std::net::Ipv4Addr::LOCALHOST)], false))
|
||||
}
|
||||
|
||||
async fn tlsa(&self, _: &str) -> Result<TlsaResult, LookupError> {
|
||||
Ok(TlsaResult::Missing)
|
||||
}
|
||||
|
||||
async fn mta_sts(&self, _: &str) -> Option<Arc<Policy>> {
|
||||
None
|
||||
}
|
||||
|
||||
fn port(&self) -> u16 {
|
||||
self.port
|
||||
}
|
||||
}
|
||||
@@ -191,6 +191,15 @@ fn run(paths: &Paths, link_listen: SocketAddr) -> Result<(), String> {
|
||||
tokio::spawn(sweeper(app.clone()));
|
||||
let outbound = app.clone();
|
||||
tokio::spawn(async move {
|
||||
#[cfg(feature = "test-mx")]
|
||||
if let Some(port) = std::env::var("INBUXA_RELAY_TEST_MX")
|
||||
.ok()
|
||||
.and_then(|port| port.parse().ok())
|
||||
{
|
||||
eprintln!("inbuxa-relay: a test build: all mail goes to 127.0.0.1:{port}");
|
||||
let test = Arc::new(deliver::route::TestMx { port });
|
||||
return deliver::run(outbound, test).await;
|
||||
}
|
||||
let dns = Arc::new(deliver::route::Dns::new().await);
|
||||
deliver::run(outbound, dns).await
|
||||
});
|
||||
|
||||
@@ -45,6 +45,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},
|
||||
},
|
||||
};
|
||||
use inbuxa_relay_proto::{
|
||||
@@ -54,8 +55,9 @@ use inbuxa_relay_proto::{
|
||||
recipient::{RecipientDomain, RecipientHash},
|
||||
seal, token,
|
||||
types::{
|
||||
Address, Capability, Inbound, InboundState, RELAY_SINGLETON, RelayEvent as ProtoRelayEvent,
|
||||
RelaySettings, RelayStatus, Reply, Verdict,
|
||||
Address, Capability, Delivered, DeliveryStatus, Envelope, Inbound, InboundState,
|
||||
Outbound as ProtoOutbound, RELAY_SINGLETON, RelayEvent as ProtoRelayEvent, RelaySettings,
|
||||
RelayStatus, Reply, Verdict,
|
||||
},
|
||||
};
|
||||
use registry::schema::{
|
||||
@@ -71,6 +73,7 @@ use serde_json::{Value, json};
|
||||
use smtp::{
|
||||
core::Relayed,
|
||||
inbound::relayed::{RelayedMessage, RelayedOutcome, ingest_relayed},
|
||||
outbound::relay::{Applied, RelayedStatus, apply_relayed},
|
||||
};
|
||||
use std::{
|
||||
collections::{BTreeMap, BTreeSet, HashMap, HashSet},
|
||||
@@ -177,15 +180,17 @@ async fn reconcile(
|
||||
links: &mut HashMap<u32, Running>,
|
||||
) -> trc::Result<()> {
|
||||
let data = server.store();
|
||||
let wanted: HashMap<u32, (String, u16)> = match relays::keys(data).await? {
|
||||
Some(_) => relays::all(data)
|
||||
.await?
|
||||
.into_iter()
|
||||
.filter(|relay| relay.enabled)
|
||||
.map(|relay| (relay.id, (relay.address.clone(), relay.link_port)))
|
||||
.collect(),
|
||||
None => HashMap::new(),
|
||||
let all = match relays::keys(data).await? {
|
||||
Some(_) => relays::all(data).await?,
|
||||
None => Vec::new(),
|
||||
};
|
||||
// Which relays outbound mail goes through (RL-20, RL-24)
|
||||
server.inner.data.relay_senders.set_relays(&all);
|
||||
let wanted: HashMap<u32, (String, u16)> = all
|
||||
.into_iter()
|
||||
.filter(|relay| relay.enabled)
|
||||
.map(|relay| (relay.id, (relay.address.clone(), relay.link_port)))
|
||||
.collect();
|
||||
links.retain(|id, link| match wanted.get(id) {
|
||||
Some(target) if *target == link.target && !link.handle.is_finished() => true,
|
||||
// Moved: start over at the new address
|
||||
@@ -780,6 +785,7 @@ struct Ws {
|
||||
relay_changed: bool,
|
||||
inbound_changed: bool,
|
||||
events_changed: bool,
|
||||
outbound_changed: bool,
|
||||
}
|
||||
|
||||
impl Ws {
|
||||
@@ -812,6 +818,7 @@ impl Ws {
|
||||
relay_changed: false,
|
||||
inbound_changed: false,
|
||||
events_changed: false,
|
||||
outbound_changed: false,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -861,6 +868,7 @@ impl Ws {
|
||||
self.relay_changed |= types.contains_key("Relay");
|
||||
self.inbound_changed |= types.contains_key("Inbound");
|
||||
self.events_changed |= types.contains_key("RelayEvent");
|
||||
self.outbound_changed |= types.contains_key("Outbound");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -957,7 +965,12 @@ async fn session(
|
||||
let mut ws = Ws::open(&dialer).await?;
|
||||
// Inbound too: live mode (RL-12)
|
||||
ws.send(&ClientFrame::WebSocketPushEnable {
|
||||
data_types: Some(vec!["Relay".into(), "Inbound".into(), "RelayEvent".into()]),
|
||||
data_types: Some(vec![
|
||||
"Relay".into(),
|
||||
"Inbound".into(),
|
||||
"RelayEvent".into(),
|
||||
"Outbound".into(),
|
||||
]),
|
||||
push_state: None,
|
||||
})
|
||||
.await?;
|
||||
@@ -981,6 +994,13 @@ async fn session(
|
||||
taker.take(&mut ws).await?;
|
||||
take_events(&mut ws, data, &relay, status.current.node_id).await?;
|
||||
|
||||
// Outbound mail can go through this link now (RL-24), until it drops
|
||||
let _sending = Sending::up(server, &relay, dialer.clone());
|
||||
let mut follower = Follower::default();
|
||||
follower
|
||||
.follow(&mut ws, server, &relay, status.current.node_id)
|
||||
.await?;
|
||||
|
||||
let mut tick = tokio::time::interval(SYNC);
|
||||
tick.reset();
|
||||
let mut ping = tokio::time::interval(PING_EVERY);
|
||||
@@ -1001,6 +1021,9 @@ async fn session(
|
||||
if ws.events_changed {
|
||||
take_events(&mut ws, data, &relay, status.current.node_id).await?;
|
||||
}
|
||||
if ws.outbound_changed {
|
||||
follower.follow(&mut ws, server, &relay, status.current.node_id).await?;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
Some(answer) = answers.recv() => {
|
||||
@@ -1036,6 +1059,9 @@ async fn session(
|
||||
taker.relay_name = relay.name.clone();
|
||||
taker.take(&mut ws).await?;
|
||||
take_events(&mut ws, data, &relay, status.current.node_id).await?;
|
||||
follower
|
||||
.follow(&mut ws, server, &relay, status.current.node_id)
|
||||
.await?;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1050,16 +1076,7 @@ async fn take_events(
|
||||
node_id: u64,
|
||||
) -> Result<(), LinkError> {
|
||||
ws.events_changed = false;
|
||||
let first = relays::link::statuses(data, relay.id)
|
||||
.await?
|
||||
.iter()
|
||||
.filter(|link| {
|
||||
link.state == LinkState::Linked && link.updated_at + 3 * STATUS_EVERY >= now()
|
||||
})
|
||||
.map(|link| link.node_id)
|
||||
.min()
|
||||
.is_none_or(|first| first >= node_id);
|
||||
if !first {
|
||||
if !first_linked(data, relay.id, node_id).await? {
|
||||
return Ok(());
|
||||
}
|
||||
let answer = ws.ok(method::RELAY_EVENT_GET, json!({})).await?;
|
||||
@@ -1077,6 +1094,20 @@ async fn take_events(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Whether this node is the lowest linked to the relay, which takes what
|
||||
/// only one node should: events, and outbound statuses.
|
||||
async fn first_linked(data: &store::Store, relay_id: u32, node_id: u64) -> trc::Result<bool> {
|
||||
Ok(relays::link::statuses(data, relay_id)
|
||||
.await?
|
||||
.iter()
|
||||
.filter(|link| {
|
||||
link.state == LinkState::Linked && link.updated_at + 3 * STATUS_EVERY >= now()
|
||||
})
|
||||
.map(|link| link.node_id)
|
||||
.min()
|
||||
.is_none_or(|first| first >= node_id))
|
||||
}
|
||||
|
||||
/// One relay event, as a `relay.*` trace event naming the relay.
|
||||
fn trace_event(relay: &Relay, event: &ProtoRelayEvent) {
|
||||
let id = relay.relay_id();
|
||||
@@ -1140,6 +1171,230 @@ fn trace_event(relay: &Relay, event: &ProtoRelayEvent) {
|
||||
}
|
||||
}
|
||||
|
||||
// ---- Outbound mail ----------------------------------------------------------
|
||||
|
||||
/// Hands messages to the relay over this node's link (RL-24): the message
|
||||
/// as an upload, then `Outbound/set` create.
|
||||
struct LinkSender {
|
||||
dialer: Arc<Dialer>,
|
||||
}
|
||||
|
||||
impl Sender for LinkSender {
|
||||
fn hand_over(&self, message: Vec<u8>, envelope: Envelope) -> HandOver<'_> {
|
||||
Box::pin(async move {
|
||||
let blob_id = self
|
||||
.dialer
|
||||
.upload(message)
|
||||
.await
|
||||
.map_err(|err| Refused::Temporary(err.message))?;
|
||||
let answer = self
|
||||
.dialer
|
||||
.api(vec![Invocation::new(
|
||||
method::OUTBOUND_SET,
|
||||
json!({
|
||||
"accountId": ACCOUNT_ID,
|
||||
"create": {"m": {"blobId": blob_id, "envelope": envelope}},
|
||||
}),
|
||||
"c",
|
||||
)])
|
||||
.await
|
||||
.map_err(|err| Refused::Temporary(err.message))?
|
||||
.into_iter()
|
||||
.next()
|
||||
.ok_or_else(|| Refused::Temporary("The relay didn't answer.".into()))?;
|
||||
let args = answer.args();
|
||||
if let Some(id) = args["created"]["m"]["id"].as_str() {
|
||||
return Ok(id.to_string());
|
||||
}
|
||||
Err(refusal(&args["notCreated"]["m"]))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// A relay's `notCreated` for a message, as this server's refusal.
|
||||
fn refusal(error: &Value) -> Refused {
|
||||
let description = error["description"].as_str().unwrap_or("no reason given");
|
||||
match error["type"].as_str().unwrap_or_default() {
|
||||
"tooLarge" => Refused::Permanent(format!("552 5.3.4 {description}")),
|
||||
"forbiddenFrom" => Refused::Permanent(format!("550 5.7.1 {description}")),
|
||||
"invalidProperties" => Refused::Permanent(format!("550 5.5.2 {description}")),
|
||||
kind => Refused::Temporary(format!("{kind}: {description}")),
|
||||
}
|
||||
}
|
||||
|
||||
/// This node's link, offered for outbound mail while it is up.
|
||||
struct Sending {
|
||||
inner: Arc<Inner>,
|
||||
linked: Arc<Linked>,
|
||||
}
|
||||
|
||||
impl Sending {
|
||||
fn up(server: &Server, relay: &Relay, dialer: Arc<Dialer>) -> Self {
|
||||
let linked = Arc::new(Linked {
|
||||
relay_id: relay.id,
|
||||
name: relay.name.clone(),
|
||||
sender: Box::new(LinkSender { dialer }),
|
||||
});
|
||||
server.inner.data.relay_senders.link_up(linked.clone());
|
||||
Sending {
|
||||
inner: server.inner.clone(),
|
||||
linked,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Sending {
|
||||
fn drop(&mut self) {
|
||||
self.inner.data.relay_senders.link_down(&self.linked);
|
||||
}
|
||||
}
|
||||
|
||||
/// How many messages one `Outbound/get` asks for.
|
||||
const FOLLOW_BATCH: usize = 200;
|
||||
|
||||
/// Follows the messages handed to the relay, from any node, until each
|
||||
/// has a final status (RL-24): the lowest linked node does, and records
|
||||
/// the relay's replies in this server's queue, which sends the DSNs.
|
||||
#[derive(Default)]
|
||||
struct Follower {
|
||||
/// Each message's statuses as last recorded, to skip ones unchanged.
|
||||
recorded: HashMap<String, BTreeMap<String, DeliveryStatus>>,
|
||||
}
|
||||
|
||||
impl Follower {
|
||||
async fn follow(
|
||||
&mut self,
|
||||
ws: &mut Ws,
|
||||
server: &Server,
|
||||
relay: &Relay,
|
||||
node_id: u64,
|
||||
) -> Result<(), LinkError> {
|
||||
ws.outbound_changed = false;
|
||||
let data = server.store();
|
||||
if !first_linked(data, relay.id, node_id).await? {
|
||||
self.recorded.clear();
|
||||
return Ok(());
|
||||
}
|
||||
let handed: HashMap<String, Handed> = outbound::handed(data, relay.id)
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|h| (h.id.clone(), h))
|
||||
.collect();
|
||||
self.recorded.retain(|id, _| handed.contains_key(id));
|
||||
let ids: Vec<&String> = handed.keys().collect();
|
||||
for batch in ids.chunks(FOLLOW_BATCH) {
|
||||
let answer = ws.ok(method::OUTBOUND_GET, json!({"ids": batch})).await?;
|
||||
let mut done = Vec::new();
|
||||
for found in answer["list"].as_array().into_iter().flatten() {
|
||||
let Ok(found) = serde_json::from_value::<ProtoOutbound>(found.clone()) else {
|
||||
continue;
|
||||
};
|
||||
let Some(handed) = handed.get(&found.id) else {
|
||||
continue;
|
||||
};
|
||||
if self.recorded.get(&found.id) == Some(&found.delivery_status) {
|
||||
continue;
|
||||
}
|
||||
let statuses = relayed_statuses(handed, &found);
|
||||
let finished = statuses.iter().all(|(_, s)| {
|
||||
matches!(
|
||||
s,
|
||||
RelayedStatus::Delivered { .. } | RelayedStatus::Refused { .. }
|
||||
)
|
||||
});
|
||||
match apply_relayed(
|
||||
server,
|
||||
handed.queue_id,
|
||||
&handed.queue_name,
|
||||
&relay.name,
|
||||
&statuses,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Applied::Done if finished => done.push(found.id.clone()),
|
||||
Applied::Done => {
|
||||
self.recorded
|
||||
.insert(found.id.clone(), found.delivery_status.clone());
|
||||
}
|
||||
// Cancelled here: the relay stops too
|
||||
Applied::Gone => done.push(found.id.clone()),
|
||||
Applied::Locked => {}
|
||||
}
|
||||
}
|
||||
// Gone from the relay (a reset, or kept past seven days): the
|
||||
// queue hands the recipients over again
|
||||
for id in answer["notFound"].as_array().into_iter().flatten() {
|
||||
let Some(handed) = id.as_str().and_then(|id| handed.get(id)) else {
|
||||
continue;
|
||||
};
|
||||
let lost: Vec<(String, RelayedStatus)> = handed
|
||||
.recipients
|
||||
.iter()
|
||||
.map(|r| (r.clone(), RelayedStatus::Lost))
|
||||
.collect();
|
||||
match apply_relayed(
|
||||
server,
|
||||
handed.queue_id,
|
||||
&handed.queue_name,
|
||||
&relay.name,
|
||||
&lost,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Applied::Done | Applied::Gone => {
|
||||
outbound::forget_handed(data, relay.id, std::slice::from_ref(&handed.id))
|
||||
.await?;
|
||||
}
|
||||
Applied::Locked => {}
|
||||
}
|
||||
}
|
||||
if !done.is_empty() {
|
||||
let answer = ws
|
||||
.ok(method::OUTBOUND_SET, json!({"destroy": done}))
|
||||
.await?;
|
||||
// One being tried right now is cleared next time
|
||||
let busy: HashSet<&str> = answer["notDestroyed"]
|
||||
.as_object()
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.filter(|(_, e)| e["type"] == "forbidden")
|
||||
.map(|(id, _)| id.as_str())
|
||||
.collect();
|
||||
let forget: Vec<String> = done
|
||||
.into_iter()
|
||||
.filter(|id| !busy.contains(id.as_str()))
|
||||
.collect();
|
||||
for id in &forget {
|
||||
self.recorded.remove(id);
|
||||
}
|
||||
outbound::forget_handed(data, relay.id, &forget).await?;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// What the relay says of each recipient a message was handed over for.
|
||||
fn relayed_statuses(handed: &Handed, found: &ProtoOutbound) -> Vec<(String, RelayedStatus)> {
|
||||
handed
|
||||
.recipients
|
||||
.iter()
|
||||
.filter_map(|address| {
|
||||
let status = found.delivery_status.get(address)?;
|
||||
let reply = status.smtp_reply.clone();
|
||||
let mx = status.mx.clone();
|
||||
Some((
|
||||
address.clone(),
|
||||
match status.delivered {
|
||||
Delivered::Yes | Delivered::Unknown => RelayedStatus::Delivered { reply, mx },
|
||||
Delivered::No => RelayedStatus::Refused { reply, mx },
|
||||
Delivered::Queued => RelayedStatus::Queued { reply, mx },
|
||||
},
|
||||
))
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
// ---- Inbound mail -----------------------------------------------------------
|
||||
|
||||
/// Takes the relay's messages for this node: a few at a time, each judged in
|
||||
|
||||
@@ -19,6 +19,8 @@ inbuxa-outbound-tls = { path = "../outbound-tls" }
|
||||
email = { path = "../email" }
|
||||
registry = { path = "../registry" }
|
||||
inbuxa-features = { path = "../features" }
|
||||
# inbuxa: outbound through inbuxa Relay (relay spec, RL-24)
|
||||
inbuxa-relay-proto = { path = "../relay-proto" }
|
||||
spam-filter = { path = "../spam-filter" }
|
||||
trc = { path = "../trc" }
|
||||
mail-auth = { version = "0.13", features = ["rkyv"] }
|
||||
|
||||
@@ -15,13 +15,14 @@ use crate::outbound::dane::dnssec::{DnssecStatus, TlsaLookup, TlsaResult};
|
||||
use crate::outbound::error::ClientError;
|
||||
use crate::outbound::lookup::{DnsLookup, SourceIp};
|
||||
use crate::outbound::mta_sts::lookup::MtaStsLookup;
|
||||
use crate::outbound::relay::RELAYED_RECHECK;
|
||||
use crate::outbound::{client::StartTlsResult, dane::verify::TlsaVerify};
|
||||
use crate::queue::dsn::{DsnStatus, SendDsn};
|
||||
use crate::queue::spool::{DSN_RETRY, SmtpSpool};
|
||||
use crate::queue::throttle::IsAllowed;
|
||||
use crate::queue::{
|
||||
Error, FROM_REPORT, HostResponse, MessageWrapper, Metadata, QueueEnvelope, QueuedMessage,
|
||||
Status,
|
||||
RCPT_RELAYED, Status,
|
||||
};
|
||||
use crate::reporting::send::MtaReportSend;
|
||||
use crate::{queue::ErrorDetails, reporting::tls::TlsRptOptions};
|
||||
@@ -237,6 +238,16 @@ impl QueuedMessage {
|
||||
}
|
||||
}
|
||||
}
|
||||
// inbuxa: with a relay's sendThrough on, every non-local recipient
|
||||
// goes through a relay (relay spec, RL-24). REQUIRETLS mail doesn't
|
||||
// yet: the relay can't keep that promise for it.
|
||||
static THROUGH_RELAY: RoutingStrategy = RoutingStrategy::InbuxaRelay;
|
||||
let senders = &server.inner.data.relay_senders;
|
||||
if let Err(err) = senders.load(server.store()).await {
|
||||
trc::error!(err.details("Failed to read the relays").span_id(span_id));
|
||||
}
|
||||
let through_relay = senders.sending() && message.message.flags & MAIL_REQUIRETLS == 0;
|
||||
let mut relayed_waiting = Vec::new();
|
||||
for (rcpt_idx, rcpt) in message.message.recipients.iter().enumerate() {
|
||||
if matches!(
|
||||
&rcpt.status,
|
||||
@@ -244,6 +255,11 @@ impl QueuedMessage {
|
||||
) && rcpt.retry.due <= now_
|
||||
&& rcpt.queue == message.queue_name
|
||||
{
|
||||
// inbuxa: waiting on a relay's status, not handed over again
|
||||
if rcpt.flags & RCPT_RELAYED != 0 {
|
||||
relayed_waiting.push(rcpt_idx);
|
||||
continue;
|
||||
}
|
||||
let envelope = QueueEnvelope::new(&message.message, rcpt);
|
||||
let route = server.get_route_or_default(
|
||||
&server
|
||||
@@ -252,6 +268,11 @@ impl QueuedMessage {
|
||||
.unwrap_or_else(|| "default".to_string()),
|
||||
message.span_id,
|
||||
);
|
||||
let route = if through_relay && !matches!(route, RoutingStrategy::Local) {
|
||||
&THROUGH_RELAY
|
||||
} else {
|
||||
route
|
||||
};
|
||||
|
||||
// Map RCPT headers
|
||||
let mut rcpt_headers = default_rcpt_header;
|
||||
@@ -313,6 +334,13 @@ impl QueuedMessage {
|
||||
.await;
|
||||
continue 'next_route;
|
||||
}
|
||||
// inbuxa: handed to an inbuxa Relay (relay spec, RL-24)
|
||||
RoutingStrategy::InbuxaRelay => {
|
||||
message
|
||||
.hand_to_relay(&rcpt_idxs, rcpt_headers, &mut delivery_results, &server)
|
||||
.await;
|
||||
continue 'next_route;
|
||||
}
|
||||
RoutingStrategy::Mx(mx_config) => (Vec::with_capacity(0), Some(mx_config), true),
|
||||
RoutingStrategy::Relay(relay_config) => (
|
||||
vec![NextHop::Relay(relay_config)],
|
||||
@@ -1520,8 +1548,22 @@ impl QueuedMessage {
|
||||
message.set_rcpt_rate_limit(rcpt_idx, retry_at);
|
||||
}
|
||||
}
|
||||
// inbuxa: kept here until the relay reports a final status
|
||||
DeliveryResult::Relayed { relay, rcpt_idx } => {
|
||||
let rcpt = &mut message.message.recipients[rcpt_idx];
|
||||
rcpt.flags |= RCPT_RELAYED;
|
||||
rcpt.retry.due = now() + RELAYED_RECHECK;
|
||||
rcpt.status = Status::TemporaryFailure(ErrorDetails {
|
||||
entity: relay.clone(),
|
||||
details: Error::Io(format!("Handed to relay {relay}").into()),
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
// inbuxa: still waiting on a relay: looked at again later
|
||||
for rcpt_idx in relayed_waiting {
|
||||
message.message.recipients[rcpt_idx].retry.due = now() + RELAYED_RECHECK;
|
||||
}
|
||||
|
||||
// Send Delivery Status Notifications
|
||||
let dsn_status = server.send_dsn(&mut message).await;
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
* SPDX-FileCopyrightText: 2020 Stalwart Labs LLC <[email protected]>
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
|
||||
*
|
||||
* Modified by Coffey Labs in 2026 for INBUXA.
|
||||
*/
|
||||
|
||||
use crate::{
|
||||
@@ -24,6 +26,8 @@ pub mod error;
|
||||
pub mod local;
|
||||
pub mod lookup;
|
||||
pub mod mta_sts;
|
||||
// inbuxa: outbound through inbuxa Relay (relay spec, RL-24)
|
||||
pub mod relay;
|
||||
pub mod session;
|
||||
|
||||
pub(super) enum DeliveryResult {
|
||||
@@ -39,6 +43,11 @@ pub(super) enum DeliveryResult {
|
||||
rcpt_idxs: Vec<usize>,
|
||||
retry_at: u64,
|
||||
},
|
||||
// inbuxa: handed to the relay named, which reports what became of it
|
||||
Relayed {
|
||||
relay: Box<str>,
|
||||
rcpt_idx: usize,
|
||||
},
|
||||
}
|
||||
|
||||
impl Status<HostResponse<Box<str>>, ErrorDetails> {
|
||||
|
||||
@@ -0,0 +1,378 @@
|
||||
/*
|
||||
* SPDX-FileCopyrightText: 2026 Coffey Labs LLC
|
||||
*
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
//! Outbound mail through an inbuxa Relay (relay spec, "Outbound routing",
|
||||
//! RL-20, RL-24).
|
||||
//!
|
||||
//! With a relay's `sendThrough` on, every non-local recipient is handed to
|
||||
//! one of the relays this node is linked to, in turn. The recipients stay
|
||||
//! in this server's queue, marked [`RCPT_RELAYED`] and shown as handed to
|
||||
//! the relay, until the link service reports the relay's final status for
|
||||
//! them through [`apply_relayed`]. The usual DSNs follow from there. With
|
||||
//! no relay linked, they wait in the queue and are retried as usual.
|
||||
|
||||
use super::DeliveryResult;
|
||||
use crate::queue::{
|
||||
Error, ErrorDetails, HostResponse, MessageWrapper, RCPT_RELAYED, Status, UnexpectedResponse,
|
||||
spool::SmtpSpool,
|
||||
};
|
||||
use common::{Server, config::smtp::queue::QueueName, ipc::QueueEvent};
|
||||
use inbuxa_features::relay::outbound::{Handed, Refused, record_handed};
|
||||
use inbuxa_relay_proto::types::{Address, Envelope};
|
||||
use smtp_proto::{
|
||||
MAIL_BODY_8BITMIME, MAIL_RET_FULL, MAIL_RET_HDRS, RCPT_NOTIFY_DELAY, RCPT_NOTIFY_FAILURE,
|
||||
RCPT_NOTIFY_NEVER, RCPT_NOTIFY_SUCCESS, Response,
|
||||
};
|
||||
use std::collections::BTreeMap;
|
||||
use store::write::now;
|
||||
use trc::RelayEvent;
|
||||
use types::id::Id;
|
||||
|
||||
/// How long a recipient handed to a relay waits before the queue looks at
|
||||
/// it again. The relay's status normally comes first.
|
||||
pub const RELAYED_RECHECK: u64 = 3_600;
|
||||
|
||||
impl MessageWrapper {
|
||||
/// Hands one domain's recipients to the first relay that takes them.
|
||||
pub(super) async fn hand_to_relay(
|
||||
&self,
|
||||
rcpt_idxs: &[usize],
|
||||
rcpt_headers: Option<&[u8]>,
|
||||
results: &mut Vec<DeliveryResult>,
|
||||
server: &Server,
|
||||
) {
|
||||
let candidates = server.inner.data.relay_senders.candidates();
|
||||
if candidates.is_empty() {
|
||||
let status = Status::TemporaryFailure(ErrorDetails {
|
||||
entity: "relay".into(),
|
||||
details: Error::ConnectionError("No relay is linked from this node.".into()),
|
||||
});
|
||||
results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec()));
|
||||
return;
|
||||
}
|
||||
let blob = match server
|
||||
.get_message_blob(self.message.blob_hash.as_slice(), 0..usize::MAX)
|
||||
.await
|
||||
{
|
||||
Ok(Some(blob)) => blob,
|
||||
Ok(None) => {
|
||||
let status = Status::PermanentFailure(ErrorDetails {
|
||||
entity: "localhost".into(),
|
||||
details: Error::Io("The message is gone from the blob store.".into()),
|
||||
});
|
||||
results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec()));
|
||||
return;
|
||||
}
|
||||
Err(err) => {
|
||||
trc::error!(
|
||||
err.details("Failed to read a message to hand to a relay")
|
||||
.span_id(self.span_id)
|
||||
);
|
||||
let status = Status::TemporaryFailure(ErrorDetails {
|
||||
entity: "localhost".into(),
|
||||
details: Error::Io("The message couldn't be read.".into()),
|
||||
});
|
||||
results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec()));
|
||||
return;
|
||||
}
|
||||
};
|
||||
let mut message = Vec::with_capacity(blob.len() + rcpt_headers.map_or(0, |h| h.len()));
|
||||
if let Some(headers) = rcpt_headers {
|
||||
message.extend_from_slice(headers);
|
||||
}
|
||||
message.extend_from_slice(&blob);
|
||||
drop(blob);
|
||||
let envelope = self.relay_envelope(rcpt_idxs);
|
||||
let recipients: Vec<String> = envelope.rcpt_to.iter().map(|r| r.email.clone()).collect();
|
||||
|
||||
let mut last = String::new();
|
||||
for relay in candidates {
|
||||
match relay
|
||||
.sender
|
||||
.hand_over(message.clone(), envelope.clone())
|
||||
.await
|
||||
{
|
||||
Ok(outbound_id) => {
|
||||
let handed = Handed {
|
||||
id: outbound_id.clone(),
|
||||
queue_id: self.queue_id,
|
||||
queue_name: self.queue_name.as_str().to_string(),
|
||||
recipients: recipients.clone(),
|
||||
at: now(),
|
||||
};
|
||||
if let Err(err) = record_handed(server.store(), relay.relay_id, &handed).await {
|
||||
// The relay sends it anyway; only its status is lost
|
||||
trc::error!(
|
||||
err.details("Failed to record a message handed to a relay")
|
||||
.span_id(self.span_id)
|
||||
);
|
||||
}
|
||||
trc::event!(
|
||||
Relay(RelayEvent::Event),
|
||||
SpanId = self.span_id,
|
||||
Id = Id::from(relay.relay_id).to_string(),
|
||||
Type = "outbound.handed-over",
|
||||
Details = outbound_id,
|
||||
To = recipients
|
||||
.iter()
|
||||
.map(|r| trc::Value::String(r.as_str().into()))
|
||||
.collect::<Vec<_>>(),
|
||||
);
|
||||
for &rcpt_idx in rcpt_idxs {
|
||||
results.push(DeliveryResult::Relayed {
|
||||
relay: relay.name.as_str().into(),
|
||||
rcpt_idx,
|
||||
});
|
||||
}
|
||||
return;
|
||||
}
|
||||
Err(Refused::Permanent(reply)) => {
|
||||
let status = Status::PermanentFailure(ErrorDetails {
|
||||
entity: relay.name.as_str().into(),
|
||||
details: Error::UnexpectedResponse(UnexpectedResponse {
|
||||
command: "".into(),
|
||||
response: parse_reply(&reply),
|
||||
}),
|
||||
});
|
||||
results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec()));
|
||||
return;
|
||||
}
|
||||
Err(Refused::Temporary(reason)) => {
|
||||
last = format!("Relay {}: {reason}", relay.name);
|
||||
}
|
||||
}
|
||||
}
|
||||
let status = Status::TemporaryFailure(ErrorDetails {
|
||||
entity: "relay".into(),
|
||||
details: Error::ConnectionError(last.into()),
|
||||
});
|
||||
results.push(DeliveryResult::domain(status, rcpt_idxs.to_vec()));
|
||||
}
|
||||
|
||||
/// The envelope as the relay takes it: the DSN parameters this server
|
||||
/// would send itself, and 8BITMIME (RFC 8621 `Address` parameters).
|
||||
fn relay_envelope(&self, rcpt_idxs: &[usize]) -> Envelope {
|
||||
let flags = self.message.flags;
|
||||
let mut from = BTreeMap::new();
|
||||
if flags & MAIL_BODY_8BITMIME != 0 {
|
||||
from.insert("BODY".to_string(), Some("8BITMIME".to_string()));
|
||||
}
|
||||
if flags & MAIL_RET_FULL != 0 {
|
||||
from.insert("RET".to_string(), Some("FULL".to_string()));
|
||||
} else if flags & MAIL_RET_HDRS != 0 {
|
||||
from.insert("RET".to_string(), Some("HDRS".to_string()));
|
||||
}
|
||||
if let Some(env_id) = &self.message.env_id {
|
||||
from.insert("ENVID".to_string(), Some(env_id.to_string()));
|
||||
}
|
||||
let rcpt_to = rcpt_idxs
|
||||
.iter()
|
||||
.map(|&idx| {
|
||||
let rcpt = &self.message.recipients[idx];
|
||||
let mut notify = Vec::new();
|
||||
for (flag, name) in [
|
||||
(RCPT_NOTIFY_SUCCESS, "SUCCESS"),
|
||||
(RCPT_NOTIFY_DELAY, "DELAY"),
|
||||
(RCPT_NOTIFY_FAILURE, "FAILURE"),
|
||||
] {
|
||||
if rcpt.flags & flag != 0 {
|
||||
notify.push(name);
|
||||
}
|
||||
}
|
||||
if notify.is_empty() && rcpt.flags & RCPT_NOTIFY_NEVER != 0 {
|
||||
notify.push("NEVER");
|
||||
}
|
||||
Address {
|
||||
email: rcpt.address().to_string(),
|
||||
parameters: (!notify.is_empty())
|
||||
.then(|| [("NOTIFY".to_string(), Some(notify.join(",")))].into()),
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
Envelope {
|
||||
mail_from: Address {
|
||||
email: self.message.return_path.to_string(),
|
||||
parameters: (!from.is_empty()).then_some(from),
|
||||
},
|
||||
rcpt_to,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `550 5.1.1 text` as a reply. A reply without an enhanced code gets its
|
||||
/// class's.
|
||||
pub fn parse_reply(reply: &str) -> Response<Box<str>> {
|
||||
let reply = reply.trim();
|
||||
let (code, rest) = reply.split_once(' ').unwrap_or((reply, ""));
|
||||
let code = code
|
||||
.parse::<u16>()
|
||||
.ok()
|
||||
.filter(|c| (200..600).contains(c))
|
||||
.unwrap_or(451);
|
||||
let class = (code / 100) as u8;
|
||||
let mut esc = [class, 0, 0];
|
||||
let mut message = rest;
|
||||
if let Some((first, after)) = rest.split_once(' ').or(Some((rest, "")))
|
||||
&& let [a, b, c] = first.split('.').collect::<Vec<_>>()[..]
|
||||
&& let (Ok(a), Ok(b), Ok(c)) = (a.parse::<u8>(), b.parse::<u8>(), c.parse::<u8>())
|
||||
&& a == class
|
||||
{
|
||||
esc = [a, b, c];
|
||||
message = after;
|
||||
}
|
||||
Response {
|
||||
code,
|
||||
esc,
|
||||
message: message.into(),
|
||||
}
|
||||
}
|
||||
|
||||
/// What a relay says became of one recipient handed to it.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum RelayedStatus {
|
||||
Delivered {
|
||||
reply: String,
|
||||
mx: Option<String>,
|
||||
},
|
||||
Refused {
|
||||
reply: String,
|
||||
mx: Option<String>,
|
||||
},
|
||||
/// Still queued at the relay, with its last reply, if any.
|
||||
Queued {
|
||||
reply: String,
|
||||
mx: Option<String>,
|
||||
},
|
||||
/// The relay no longer has the message: hand it over again.
|
||||
Lost,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum Applied {
|
||||
Done,
|
||||
/// The queue is working on the message: try again later.
|
||||
Locked,
|
||||
/// The message is no longer queued here.
|
||||
Gone,
|
||||
}
|
||||
|
||||
/// Records a relay's statuses for recipients it was handed, by address,
|
||||
/// and wakes the queue to send any DSN they call for.
|
||||
pub async fn apply_relayed(
|
||||
server: &Server,
|
||||
queue_id: u64,
|
||||
queue_name: &str,
|
||||
relay: &str,
|
||||
statuses: &[(String, RelayedStatus)],
|
||||
) -> Applied {
|
||||
let Some(queue_name) = QueueName::new(queue_name) else {
|
||||
return Applied::Gone;
|
||||
};
|
||||
if !server.try_lock_event(queue_id, queue_name).await {
|
||||
return Applied::Locked;
|
||||
}
|
||||
let Some(mut message) = server.read_message(queue_id, queue_name).await else {
|
||||
server.unlock_event(queue_id, queue_name).await;
|
||||
return Applied::Gone;
|
||||
};
|
||||
message.span_id = server.inner.data.span_id_gen.generate();
|
||||
let now = now();
|
||||
let mut changed = false;
|
||||
let mut wake = false;
|
||||
for rcpt in message.message.recipients.iter_mut() {
|
||||
// Expired here meanwhile: this server's word stands
|
||||
if rcpt.flags & RCPT_RELAYED == 0
|
||||
|| matches!(
|
||||
rcpt.status,
|
||||
Status::Completed(_) | Status::PermanentFailure(_)
|
||||
)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
let Some((_, status)) = statuses
|
||||
.iter()
|
||||
.find(|(address, _)| address.eq_ignore_ascii_case(rcpt.address()))
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
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 {
|
||||
hostname: entity(mx),
|
||||
response: parse_reply(reply),
|
||||
}),
|
||||
RelayedStatus::Refused { reply, mx } => Status::PermanentFailure(ErrorDetails {
|
||||
entity: entity(mx),
|
||||
details: Error::UnexpectedResponse(UnexpectedResponse {
|
||||
command: "".into(),
|
||||
response: parse_reply(reply),
|
||||
}),
|
||||
}),
|
||||
RelayedStatus::Queued { reply, mx } if !reply.is_empty() => {
|
||||
Status::TemporaryFailure(ErrorDetails {
|
||||
entity: entity(mx),
|
||||
details: Error::UnexpectedResponse(UnexpectedResponse {
|
||||
command: "".into(),
|
||||
response: parse_reply(reply),
|
||||
}),
|
||||
})
|
||||
}
|
||||
RelayedStatus::Queued { .. } => continue,
|
||||
RelayedStatus::Lost => Status::TemporaryFailure(ErrorDetails {
|
||||
entity: relay.into(),
|
||||
details: Error::Io("The relay no longer has the message.".into()),
|
||||
}),
|
||||
};
|
||||
if matches!(
|
||||
status,
|
||||
RelayedStatus::Delivered { .. } | RelayedStatus::Refused { .. } | RelayedStatus::Lost
|
||||
) {
|
||||
rcpt.flags &= !RCPT_RELAYED;
|
||||
rcpt.retry.due = now;
|
||||
wake = true;
|
||||
}
|
||||
if rcpt.status != new {
|
||||
rcpt.status = new;
|
||||
changed = true;
|
||||
}
|
||||
}
|
||||
if changed || wake {
|
||||
message
|
||||
.save_changes(server, None, wake.then_some(now))
|
||||
.await;
|
||||
}
|
||||
server.unlock_event(queue_id, queue_name).await;
|
||||
if wake {
|
||||
let _ = server.inner.ipc.queue_tx.send(QueueEvent::Refresh).await;
|
||||
}
|
||||
Applied::Done
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn replies() {
|
||||
let r = parse_reply("550 5.1.1 No such user here");
|
||||
assert_eq!(
|
||||
(r.code, r.esc, r.message.as_ref()),
|
||||
(550, [5, 1, 1], "No such user here")
|
||||
);
|
||||
let r = parse_reply("250 Ok");
|
||||
assert_eq!((r.code, r.esc, r.message.as_ref()), (250, [2, 0, 0], "Ok"));
|
||||
// An enhanced code of another class isn't one
|
||||
let r = parse_reply("451 5.1.1 odd");
|
||||
assert_eq!(
|
||||
(r.code, r.esc, r.message.as_ref()),
|
||||
(451, [4, 0, 0], "5.1.1 odd")
|
||||
);
|
||||
let r = parse_reply("");
|
||||
assert_eq!((r.code, r.esc), (451, [4, 0, 0]));
|
||||
let r = parse_reply("552 5.3.4");
|
||||
assert_eq!((r.code, r.esc, r.message.as_ref()), (552, [5, 3, 4], ""));
|
||||
}
|
||||
}
|
||||
@@ -158,6 +158,9 @@ pub const FROM_REPORT: u64 = 1 << 36;
|
||||
pub const FROM_AUTOGENERATED: u64 = 1 << 37;
|
||||
|
||||
pub const RCPT_DSN_SENT: u64 = 1 << 32;
|
||||
// inbuxa: handed to an inbuxa Relay, which hasn't reported a final status
|
||||
// (relay spec, RL-24)
|
||||
pub const RCPT_RELAYED: u64 = 1 << 40;
|
||||
|
||||
pub const RCPT_SPAM_SHIFT: u64 = 56;
|
||||
pub const RCPT_SPAM_MASK: u64 = 0xff << RCPT_SPAM_SHIFT;
|
||||
|
||||
@@ -307,7 +307,7 @@ so a reboot without the link comes back with the same settings (RL-22).
|
||||
| `blobId` | The message as the downstream signed it |
|
||||
| `envelope` | RFC 8621 `Envelope` |
|
||||
| `size` | Bytes |
|
||||
| `deliveryStatus` | RFC 8621 `DeliveryStatus` per recipient: `smtpReply` (empty before the first try), `delivered` (`queued`, `yes`, `no`, `unknown`) |
|
||||
| `deliveryStatus` | RFC 8621 `DeliveryStatus` per recipient: `smtpReply` (empty before the first try), `delivered` (`queued`, `yes`, `no`, `unknown`), and `mx`, the host the reply came from when one answered (not in RFC 8621) |
|
||||
| `nextAttemptAt`, `attempts`, `expiresAt` | Retry state. `nextAttemptAt` only while a recipient is queued |
|
||||
|
||||
A create takes `blobId` and `envelope` only; the relay sets the rest, and no
|
||||
@@ -552,9 +552,25 @@ allowing the edge's IP and disabling SPF, DMARC and reverse IP for it.
|
||||
- **RL-24.** A new route type, `inbuxaRelay`, alongside the existing route
|
||||
types, names one or more `inbuxa:Relay`s. With `sendThrough` on, it is the
|
||||
route for every non-local recipient.
|
||||
- `sendThrough` on any enabled relay sends every non-local recipient
|
||||
through relays: the enabled ones not marked `inboundOnly`, that the node
|
||||
delivering is linked to, each message starting with the next in turn.
|
||||
A relay that refuses a message for now (its spool full, its link
|
||||
failing) passes it to the next; one that refuses it for good
|
||||
(`tooLarge`, `forbiddenFrom`) fails those recipients. With no relay
|
||||
linked, the recipients wait in the queue and are retried as usual.
|
||||
- The downstream's queue keeps the message until a relay reports a final
|
||||
status. The item shows "handed to relay ‹name›", and then the remote
|
||||
server's reply.
|
||||
server's reply. A handed-over recipient isn't handed over again while it
|
||||
waits, except when the relay no longer has the message (a reset). A
|
||||
message cancelled from the downstream's queue is destroyed on the relay
|
||||
too, unless an attempt is under way, and then at the next status.
|
||||
- The relay's statuses are followed by one node, the lowest linked to that
|
||||
relay, for every message handed to it from any node. A final status is
|
||||
recorded in the downstream's queue, which sends any DSN as for its own
|
||||
deliveries, and the message is then destroyed on the relay.
|
||||
- Mail sent with REQUIRETLS (RFC 8689) isn't handed to a relay yet: it
|
||||
takes the route it would without relays.
|
||||
|
||||
### DNS and certificates
|
||||
|
||||
@@ -801,8 +817,18 @@ In progress. Done so far:
|
||||
recipient status from the remote server's replies, retries, expiry, and
|
||||
`outbound.*` events (RL-T12 tested against a scripted MX with fixed DNS
|
||||
answers). The release binary went from 5.4 MB to 8.1 MB with it
|
||||
(glibc, measured 2026-10-30). The downstream doesn't hand anything over
|
||||
yet: that is the rest of step 4.
|
||||
(glibc, measured 2026-10-30).
|
||||
- Outbound through relays on the downstream: `sendThrough` and
|
||||
`inboundOnly` decide the route (RL-20, RL-24); each message is uploaded
|
||||
and created with `Outbound/set` over the delivering node's link, its
|
||||
recipients are marked in the queue as handed to the relay, and the link
|
||||
service follows `Outbound` with push, records each reply, and destroys
|
||||
the relay's copy once every recipient is final. Tested end to end with
|
||||
the relay sending to a scripted MX (RL-T11): a test build of the relay,
|
||||
with the `test-mx` feature, sends every domain to one local port.
|
||||
`outboundTls` is set like any other setting, `settings/outboundTls`.
|
||||
- Still to come in step 4: the named `inbuxaRelay` route, TLS-RPT reports
|
||||
from the relay's `outbound.tls` events, and REQUIRETLS through relays.
|
||||
- `crates/outbound-tls` (RL-30): the TLSA and MTA-STS types, the DANE
|
||||
certificate check, MTA-STS policy parsing and matching, and the DNSSEC
|
||||
lookups (with the bogus-answer recheck), moved out of `crates/smtp` and
|
||||
|
||||
+201
-1
@@ -33,6 +33,8 @@ use std::str::FromStr;
|
||||
|
||||
const USING: &[&str] = &[
|
||||
"urn:ietf:params:jmap:core",
|
||||
"urn:ietf:params:jmap:mail",
|
||||
"urn:ietf:params:jmap:submission",
|
||||
"urn:inbuxa:jmap",
|
||||
"urn:inbuxa:jmap:registry",
|
||||
];
|
||||
@@ -542,6 +544,99 @@ async fn wait_for_mail(account: &Account, subject: &str, count: usize) {
|
||||
}
|
||||
}
|
||||
|
||||
/// Sends a message from `sender` to `to` by JMAP submission.
|
||||
async fn submit(sender: &Account, identity: &str, mailbox: &str, to: &[&str], subject: &str) {
|
||||
let (_, response) = call(
|
||||
sender,
|
||||
"Email/set",
|
||||
json!({"create": {"e": {
|
||||
"mailboxIds": {mailbox: true},
|
||||
"from": [{"email": sender.name()}],
|
||||
"to": to.iter().map(|a| json!({"email": a})).collect::<Vec<_>>(),
|
||||
"subject": subject,
|
||||
"bodyValues": {"b": {"value": "Hello from the far side."}},
|
||||
"textBody": [{"partId": "b", "type": "text/plain"}]
|
||||
}}}),
|
||||
)
|
||||
.await;
|
||||
let email = response["created"]["e"]["id"]
|
||||
.as_str()
|
||||
.unwrap_or_else(|| panic!("draft: {response}"))
|
||||
.to_string();
|
||||
let (_, response) = call(
|
||||
sender,
|
||||
"EmailSubmission/set",
|
||||
json!({"create": {"s": {"emailId": email, "identityId": identity}}}),
|
||||
)
|
||||
.await;
|
||||
assert!(response["created"].get("s").is_some(), "{response}");
|
||||
}
|
||||
|
||||
/// The destination the relay sends to (`INBUXA_RELAY_TEST_MX`): takes
|
||||
/// `bob`, refuses `nobody`, defers `later`. The messages it took.
|
||||
async fn scripted_mx() -> (u16, std::sync::Arc<std::sync::Mutex<Vec<String>>>) {
|
||||
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let port = listener.local_addr().unwrap().port();
|
||||
let seen = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
|
||||
let taken = seen.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Ok((stream, _)) = listener.accept().await {
|
||||
let taken = taken.clone();
|
||||
tokio::spawn(async move {
|
||||
let (read, mut write) = stream.into_split();
|
||||
let mut read = BufReader::new(read);
|
||||
let _ = write.write_all(b"220 remote.test ready\r\n").await;
|
||||
let mut accepted = 0;
|
||||
loop {
|
||||
let mut line = String::new();
|
||||
if read.read_line(&mut line).await.unwrap_or(0) == 0 {
|
||||
return;
|
||||
}
|
||||
let upper = line.to_ascii_uppercase();
|
||||
let answer = if upper.starts_with("EHLO") {
|
||||
"250-remote.test\r\n250 8BITMIME\r\n"
|
||||
} else if upper.starts_with("MAIL") {
|
||||
accepted = 0;
|
||||
"250 2.1.0 Ok\r\n"
|
||||
} else if upper.starts_with("RCPT") && upper.contains("<BOB@") {
|
||||
accepted += 1;
|
||||
"250 2.1.5 Ok\r\n"
|
||||
} else if upper.starts_with("RCPT") && upper.contains("<LATER@") {
|
||||
"451 4.7.1 Try again later\r\n"
|
||||
} else if upper.starts_with("RCPT") {
|
||||
"550 5.1.1 No such user here\r\n"
|
||||
} else if upper.starts_with("DATA") && accepted > 0 {
|
||||
let _ = write.write_all(b"354 Go ahead\r\n").await;
|
||||
let mut message = String::new();
|
||||
loop {
|
||||
let mut line = String::new();
|
||||
if read.read_line(&mut line).await.unwrap_or(0) == 0 {
|
||||
return;
|
||||
}
|
||||
if line == ".\r\n" {
|
||||
break;
|
||||
}
|
||||
message.push_str(&line);
|
||||
}
|
||||
taken.lock().unwrap().push(message);
|
||||
"250 2.0.0 Queued as 1\r\n"
|
||||
} else if upper.starts_with("DATA") {
|
||||
"554 5.5.1 No valid recipients\r\n"
|
||||
} else if upper.starts_with("QUIT") {
|
||||
let _ = write.write_all(b"221 Bye\r\n").await;
|
||||
return;
|
||||
} else {
|
||||
"250 Ok\r\n"
|
||||
};
|
||||
let _ = write.write_all(answer.as_bytes()).await;
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
(port, seen)
|
||||
}
|
||||
|
||||
fn message(subject: &str, extra_headers: &str) -> String {
|
||||
format!(
|
||||
"{extra_headers}From: Someone <[email protected]>\nTo: [email protected]\n\
|
||||
@@ -556,7 +651,7 @@ pub async fn link(test: &mut TestServer) {
|
||||
let binary = relay_binary();
|
||||
assert!(
|
||||
binary.exists(),
|
||||
"{} is missing: run cargo build -p inbuxa-relay",
|
||||
"{} is missing: run cargo build -p inbuxa-relay --features test-mx",
|
||||
binary.display()
|
||||
);
|
||||
let jane = admin
|
||||
@@ -570,6 +665,7 @@ pub async fn link(test: &mut TestServer) {
|
||||
.await;
|
||||
let link_port = free_port();
|
||||
let smtp_port = free_port();
|
||||
let (mx_port, mx_seen) = scripted_mx().await;
|
||||
|
||||
let (_, response) = call(
|
||||
admin,
|
||||
@@ -612,6 +708,7 @@ pub async fn link(test: &mut TestServer) {
|
||||
"--state",
|
||||
])
|
||||
.arg(&state)
|
||||
.env("INBUXA_RELAY_TEST_MX", mx_port.to_string())
|
||||
.stdout(std::process::Stdio::null())
|
||||
.stderr(std::process::Stdio::null())
|
||||
.spawn()
|
||||
@@ -757,6 +854,109 @@ pub async fn link(test: &mut TestServer) {
|
||||
2
|
||||
);
|
||||
|
||||
// Outbound through the relay (RL-24): handed over, sent by the relay,
|
||||
// and each recipient's reply recorded here, with one DSN (RL-T11)
|
||||
call(
|
||||
admin,
|
||||
"inbuxa:Relay/set",
|
||||
json!({"update": {&id: {"sendThrough": true, "settings/maxRecipients": 10}}}),
|
||||
)
|
||||
.await;
|
||||
let (_, response) = call(
|
||||
&jane,
|
||||
"Identity/set",
|
||||
json!({"create": {"i": {"name": "Jane", "email": "[email protected]"}}}),
|
||||
)
|
||||
.await;
|
||||
let identity = response["created"]["i"]["id"]
|
||||
.as_str()
|
||||
.unwrap_or_else(|| panic!("{response}"))
|
||||
.to_string();
|
||||
let (_, response) = call(
|
||||
&jane,
|
||||
"Mailbox/set",
|
||||
json!({"create": {"m": {"name": "Outbound drafts"}}}),
|
||||
)
|
||||
.await;
|
||||
let drafts = response["created"]["m"]["id"].as_str().unwrap().to_string();
|
||||
// The relay has the new maxRecipients before anything is handed to it
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30);
|
||||
loop {
|
||||
let replies = rcpt_replies(smtp_port, &["[email protected]", "[email protected]"]).await;
|
||||
if replies[1].starts_with("250") {
|
||||
break;
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"maxRecipients reached the relay: {replies:?}"
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
}
|
||||
submit(
|
||||
&jane,
|
||||
&identity,
|
||||
&drafts,
|
||||
&["[email protected]", "[email protected]"],
|
||||
"Through the relay",
|
||||
)
|
||||
.await;
|
||||
wait_for_mail(&jane, "Failed to deliver message", 1).await;
|
||||
let seen = mx_seen.lock().unwrap().clone();
|
||||
assert!(
|
||||
seen.iter()
|
||||
.any(|m| m.contains("Subject: Through the relay")),
|
||||
"the relay sent it on: {seen:?}"
|
||||
);
|
||||
// Gone from this server's queue, and from the relay's
|
||||
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 in_queue = queued.to_string().contains("[email protected]");
|
||||
let at_relay = std::fs::read_dir(state.join("spool").join("out"))
|
||||
.map(|dir| dir.count())
|
||||
.unwrap_or_default();
|
||||
if !in_queue && at_relay == 0 {
|
||||
break;
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"queued here: {in_queue}, files at the relay: {at_relay}"
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
}
|
||||
// A deferral stays queued here, with the remote server's reply
|
||||
submit(
|
||||
&jane,
|
||||
&identity,
|
||||
&drafts,
|
||||
&["[email protected]"],
|
||||
"Deferred by the remote",
|
||||
)
|
||||
.await;
|
||||
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 text = queued.to_string();
|
||||
if text.contains("[email protected]") && text.contains("Try again later") {
|
||||
break;
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"the deferral is shown: {queued}"
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
|
||||
assert_eq!(
|
||||
subjects(&jane)
|
||||
.await
|
||||
.iter()
|
||||
.filter(|s| s.starts_with("Failed") || s.starts_with("Warning"))
|
||||
.count(),
|
||||
1,
|
||||
"one DSN, for the refusal"
|
||||
);
|
||||
|
||||
// The relay's events were taken as this server's relay.* trace events,
|
||||
// and acknowledged, so the relay forgot them
|
||||
let events = state.join("events.jsonl");
|
||||
|
||||
Reference in new issue
Block a user