crates/relay-proto/src/jmap.rs

@@ -28,6 +28,9 @@ pub mod method {
     pub const INBOUND_QUERY: &str = "Inbound/query";
     pub const INBOUND_CLAIM: &str = "Inbound/claim";
     pub const INBOUND_SET: &str = "Inbound/set";
+    pub const OUTBOUND_GET: &str = "Outbound/get";
+    pub const OUTBOUND_CHANGES: &str = "Outbound/changes";
+    pub const OUTBOUND_SET: &str = "Outbound/set";
     pub const RELAY_EVENT_GET: &str = "RelayEvent/get";
     pub const RELAY_EVENT_CHANGES: &str = "RelayEvent/changes";
     pub const RELAY_EVENT_SET: &str = "RelayEvent/set";

crates/relay-proto/src/types.rs

@@ -349,6 +349,70 @@ impl Verdict {
     }
 }
 
+/// A message the downstream handed to the relay to send (RL-19).
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Outbound {
+    pub id: String,
+    pub created_at: String,
+    /// The upload the message was made from, as the downstream signed it.
+    pub blob_id: String,
+    pub envelope: Envelope,
+    pub size: u64,
+    /// Per recipient, by address.
+    pub delivery_status: BTreeMap<String, DeliveryStatus>,
+    /// The next try, while any recipient is still queued.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub next_attempt_at: Option<String>,
+    pub attempts: u32,
+    pub expires_at: String,
+}
+
+/// What `Outbound/set` create takes.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub struct NewOutbound {
+    pub blob_id: String,
+    pub envelope: Envelope,
+}
+
+/// RFC 8621 `DeliveryStatus`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DeliveryStatus {
+    /// The remote server's last reply, empty before the first try.
+    pub smtp_reply: String,
+    pub delivered: Delivered,
+    /// The relay never learns this.
+    pub displayed: Displayed,
+}
+
+impl DeliveryStatus {
+    pub fn queued() -> Self {
+        DeliveryStatus {
+            smtp_reply: String::new(),
+            delivered: Delivered::Queued,
+            displayed: Displayed::Unknown,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Delivered {
+    Queued,
+    Yes,
+    No,
+    Unknown,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Displayed {
+    Unknown,
+    Yes,
+}
+
 /// A relay event, for the downstream's history and tracer.
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -407,6 +471,30 @@ mod tests {
         assert!(s.serves("EXAMPLE.com"));
     }
 
+    #[test]
+    fn outbound_wire_form() {
+        let status = serde_json::to_value(DeliveryStatus::queued()).unwrap();
+        assert_eq!(
+            status,
+            json!({"smtpReply": "", "delivered": "queued", "displayed": "unknown"})
+        );
+        let new: NewOutbound = serde_json::from_value(json!({
+            "blobId": "b1",
+            "envelope": {"mailFrom": {"email": "[email protected]"}, "rcptTo": [{"email": "[email protected]"}]},
+        }))
+        .unwrap();
+        assert_eq!(new.envelope.rcpt_to.len(), 1);
+        assert!(
+            serde_json::from_value::<NewOutbound>(json!({
+                "blobId": "b1",
+                "envelope": {"mailFrom": {"email": ""}, "rcptTo": []},
+                "deliveryStatus": {},
+            }))
+            .is_err(),
+            "the relay sets the status"
+        );
+    }
+
     #[test]
     fn certificate_key_stays_out_of_debug() {
         let cert = SmtpCertificate {

crates/relay/src/app.rs

@@ -8,6 +8,7 @@
 
 use crate::{
     events::Events,
+    outbound::Outbox,
     recipients::Recipients,
     spool::Spool,
     state::{self, Identity, Paths},
@@ -43,6 +44,7 @@ pub struct App {
     settings_state: AtomicU64,
     pub recipients: RwLock<Recipients>,
     pub spool: Mutex<Spool>,
+    pub outbox: Mutex<Outbox>,
     pub events: Mutex<Events>,
     pub push: broadcast::Sender<Changed>,
     /// Link connections, and those with push on for `Inbound`.
@@ -60,6 +62,7 @@ impl App {
         let settings = state::load_settings(&paths)?.map(Arc::new);
         let recipients = Recipients::load(&paths)?;
         let spool = Spool::open(&paths.spool_in())?;
+        let outbox = Outbox::open(&paths.spool_out())?;
         let events = Events::open(paths.events())?;
         // Uploads don't survive a restart.
         if let Ok(dir) = std::fs::read_dir(paths.blobs()) {
@@ -75,6 +78,7 @@ impl App {
             settings_state: AtomicU64::new(1),
             recipients: RwLock::new(recipients),
             spool: Mutex::new(spool),
+            outbox: Mutex::new(outbox),
             events: Mutex::new(events),
             push,
             link_clients: AtomicU32::new(0),
@@ -145,10 +149,14 @@ impl App {
     }
 
     pub fn status(&self) -> RelayStatus {
-        let (inbound, bytes) = {
+        let (inbound, inbound_bytes) = {
             let spool = self.spool.lock().expect("spool");
             (spool.count(), spool.bytes())
         };
+        let (outbound, outbound_bytes) = {
+            let outbox = self.outbox.lock().expect("outbox");
+            (outbox.count(), outbox.bytes())
+        };
         RelayStatus {
             started_at: time::utc_date(self.started_at),
             relay_version: VERSION.to_string(),
@@ -156,8 +164,8 @@ impl App {
             mode: self.mode(),
             spool: SpoolStatus {
                 inbound,
-                outbound: 0,
-                bytes,
+                outbound,
+                bytes: inbound_bytes + outbound_bytes,
             },
             recipient_state: self.recipients.read().expect("recipients").state(),
             certificate_not_after: None,
@@ -166,9 +174,11 @@ impl App {
         }
     }
 
-    /// Whether the spool is too full for new mail: past 90% of the cap.
+    /// Whether the spool is too full for new mail, either way: past 90% of
+    /// the cap.
     pub fn spool_full(&self, settings: &RelaySettings) -> bool {
-        let bytes = self.spool.lock().expect("spool").bytes();
+        let bytes =
+            self.spool.lock().expect("spool").bytes() + self.outbox.lock().expect("outbox").bytes();
         bytes.saturating_mul(10) >= settings.spool_max_bytes.saturating_mul(9)
             || disk_nearly_full(&self.paths.spool_in())
     }

crates/relay/src/changelog.rs

@@ -0,0 +1,103 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! State strings and `/changes` for the spools: a counter, and the ids
+//! destroyed lately. Runtime only; a restart starts every client over.
+
+use std::collections::VecDeque;
+
+/// How many destroyed ids `/changes` can still report.
+const TOMBSTONES: usize = 10_000;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Changes {
+    pub old_state: String,
+    pub new_state: String,
+    pub created: Vec<String>,
+    pub updated: Vec<String>,
+    pub destroyed: Vec<String>,
+    pub has_more: bool,
+}
+
+pub struct Changelog {
+    modseq: u64,
+    tombstones: VecDeque<(u64, String)>,
+    /// `/changes` can't go back past this.
+    floor: u64,
+}
+
+impl Default for Changelog {
+    fn default() -> Self {
+        Changelog {
+            modseq: 1,
+            tombstones: VecDeque::new(),
+            floor: 1,
+        }
+    }
+}
+
+impl Changelog {
+    pub fn state(&self) -> String {
+        self.modseq.to_string()
+    }
+
+    /// The next modseq, for an object created or changed.
+    pub fn bump(&mut self) -> u64 {
+        self.modseq += 1;
+        self.modseq
+    }
+
+    pub fn destroyed(&mut self, id: &str) {
+        let modseq = self.bump();
+        self.tombstones.push_back((modseq, id.to_string()));
+        while self.tombstones.len() > TOMBSTONES {
+            if let Some((seq, _)) = self.tombstones.pop_front() {
+                self.floor = seq;
+            }
+        }
+    }
+
+    /// What changed since `since`, given each live object's id and its
+    /// created and last-changed modseqs. `None` when the client must start
+    /// over: an unknown state, or more than `max` changes.
+    pub fn changes<'a>(
+        &self,
+        since: &str,
+        max: usize,
+        objects: impl Iterator<Item = (&'a String, u64, u64)>,
+    ) -> Option<Changes> {
+        let since: u64 = since.parse().ok()?;
+        if since < self.floor || since > self.modseq {
+            return None;
+        }
+        let mut created = Vec::new();
+        let mut updated = Vec::new();
+        for (id, created_at, modseq) in objects {
+            if created_at > since {
+                created.push(id.clone());
+            } else if modseq > since {
+                updated.push(id.clone());
+            }
+        }
+        let destroyed: Vec<String> = self
+            .tombstones
+            .iter()
+            .filter(|(seq, _)| *seq > since)
+            .map(|(_, id)| id.clone())
+            .collect();
+        if created.len() + updated.len() + destroyed.len() > max {
+            return None;
+        }
+        Some(Changes {
+            old_state: since.to_string(),
+            new_state: self.state(),
+            created,
+            updated,
+            destroyed,
+            has_more: false,
+        })
+    }
+}

crates/relay/src/link/jmap.rs

@@ -16,7 +16,7 @@ use inbuxa_relay_proto::{
     jmap::{Invocation, MethodError, Request, Response, method},
     recipient::{RecipientDomain, RecipientHash},
     time, token,
-    types::{RELAY_SINGLETON, RelaySettings, Verdict, type_name},
+    types::{NewOutbound, RELAY_SINGLETON, RelaySettings, Verdict, type_name},
 };
 use rustls_pki_types::{CertificateDer, pem::PemObject};
 use serde_json::{Map, Value, json};
@@ -115,6 +115,9 @@ fn call_one(app: &App, caller: &Caller, call: &Invocation) -> MethodResult {
         method::INBOUND_CHANGES => inbound_changes(app, args),
         method::INBOUND_CLAIM => inbound_claim(app, caller, args),
         method::INBOUND_SET => inbound_set(app, args),
+        method::OUTBOUND_GET => outbound_get(app, args),
+        method::OUTBOUND_CHANGES => outbound_changes(app, args),
+        method::OUTBOUND_SET => outbound_set(app, args),
         method::RELAY_EVENT_GET => event_get(app, args),
         method::RELAY_EVENT_CHANGES => event_changes(app, args),
         method::RELAY_EVENT_SET => event_set(app, args),
@@ -747,6 +750,148 @@ fn inbound_set(app: &App, args: &Map<String, Value>) -> MethodResult {
     }))
 }
 
+// ---- Outbound -------------------------------------------------------------
+
+fn outbound_get(app: &App, args: &Map<String, Value>) -> MethodResult {
+    let ids = string_list(args, "ids")?;
+    let outbox = app.outbox.lock().expect("outbox");
+    let ids = ids.unwrap_or_else(|| outbox.ids());
+    let mut list = Vec::new();
+    let mut not_found = Vec::new();
+    for id in ids {
+        match outbox.get(&id) {
+            Some(outbound) => {
+                list.push(serde_json::to_value(outbound).expect("outbound serializes"))
+            }
+            None => not_found.push(id),
+        }
+    }
+    Ok(json!({
+        "accountId": ACCOUNT_ID,
+        "state": outbox.state(),
+        "list": list,
+        "notFound": not_found,
+    }))
+}
+
+fn outbound_changes(app: &App, args: &Map<String, Value>) -> MethodResult {
+    let since = args
+        .get("sinceState")
+        .and_then(Value::as_str)
+        .ok_or_else(|| err(MethodError::InvalidArguments, "sinceState is required"))?;
+    let max = args
+        .get("maxChanges")
+        .and_then(Value::as_u64)
+        .map_or(MAX_CHANGES, |m| (m as usize).clamp(1, MAX_CHANGES));
+    let outbox = app.outbox.lock().expect("outbox");
+    let changes = outbox
+        .changes(since, max)
+        .ok_or((MethodError::CannotCalculateChanges, None))?;
+    Ok(json!({
+        "accountId": ACCOUNT_ID,
+        "oldState": changes.old_state,
+        "newState": changes.new_state,
+        "hasMoreChanges": changes.has_more,
+        "created": changes.created,
+        "updated": changes.updated,
+        "destroyed": changes.destroyed,
+    }))
+}
+
+/// RL-19: the only way mail leaves through the relay.
+fn outbound_set(app: &App, args: &Map<String, Value>) -> MethodResult {
+    let mut created = Map::new();
+    let mut not_created = Map::new();
+    let mut not_updated = Map::new();
+    let mut destroyed = Vec::new();
+    let mut not_destroyed = Map::new();
+    let settings = app.settings();
+    let create = object(args, "create")?;
+    let full = match (&settings, create) {
+        (Some(settings), Some(create)) if !create.is_empty() => app.spool_full(settings),
+        _ => false,
+    };
+    let mut outbox = app.outbox.lock().expect("outbox");
+    let old_state = outbox.state();
+    for (key, value) in create.into_iter().flatten() {
+        let new = match serde_json::from_value::<NewOutbound>(value.clone()) {
+            Ok(new) if crate::link::is_blob_id(&new.blob_id) => new,
+            Ok(_) => {
+                not_created.insert(
+                    key.clone(),
+                    set_error("invalidProperties", "blobId: an uploaded blob"),
+                );
+                continue;
+            }
+            Err(e) => {
+                not_created.insert(key.clone(), set_error("invalidProperties", e.to_string()));
+                continue;
+            }
+        };
+        let Some(settings) = &settings else {
+            not_created.insert(
+                key.clone(),
+                set_error("forbidden", "the relay has no settings yet"),
+            );
+            continue;
+        };
+        let blob = app.paths.blobs().join(&new.blob_id);
+        match outbox.create(
+            &blob,
+            &new.blob_id,
+            new.envelope,
+            settings,
+            full,
+            time::now(),
+        ) {
+            Ok(id) => {
+                let object = outbox.get(&id).expect("just created");
+                created.insert(
+                    key.clone(),
+                    serde_json::to_value(object).expect("outbound serializes"),
+                );
+            }
+            Err(refusal) => {
+                not_created.insert(key.clone(), set_error(refusal.kind, refusal.description));
+            }
+        }
+    }
+    for id in object(args, "update")?
+        .into_iter()
+        .flatten()
+        .map(|(id, _)| id)
+    {
+        not_updated.insert(
+            id.clone(),
+            set_error("forbidden", "only the relay changes an Outbound"),
+        );
+    }
+    for id in string_list(args, "destroy")?.unwrap_or_default() {
+        if outbox.destroy(&id) {
+            destroyed.push(id);
+        } else {
+            not_destroyed.insert(id, set_error("notFound", "no such message"));
+        }
+    }
+    let new_state = outbox.state();
+    drop(outbox);
+    if new_state != old_state {
+        app.notify(type_name::OUTBOUND, new_state.clone());
+        app.relay_changed();
+    }
+    Ok(json!({
+        "accountId": ACCOUNT_ID,
+        "oldState": old_state,
+        "newState": new_state,
+        "created": created,
+        "notCreated": not_created,
+        "updated": {},
+        "notUpdated": not_updated,
+        "destroyed": destroyed,
+        "notDestroyed": not_destroyed,
+    }))
+}
+
 // ---- RelayEvent -----------------------------------------------------------
 
 fn event_get(app: &App, args: &Map<String, Value>) -> MethodResult {
@@ -823,6 +968,10 @@ pub fn states(app: &App) -> Vec<(&'static str, String)> {
         (type_name::RECIPIENT, recipient.clone()),
         (type_name::RECIPIENT_DOMAIN, recipient),
         (type_name::INBOUND, app.spool.lock().expect("spool").state()),
+        (
+            type_name::OUTBOUND,
+            app.outbox.lock().expect("outbox").state(),
+        ),
         (
             type_name::RELAY_EVENT,
             app.events.lock().expect("events").state(),

crates/relay/src/main.rs

@@ -12,8 +12,10 @@
 //! command line can join, run, show status, and reset; nothing else.
 
 mod app;
+mod changelog;
 mod events;
 mod link;
+mod outbound;
 mod recipients;
 mod smtp;
 mod spool;
@@ -286,10 +288,9 @@ fn status(paths: &Paths) -> Result<(), String> {
             println!("Running:    yes, since {}", s.started_at);
             println!("Mode:       {:?}", s.mode);
             println!("Linked:     {} connection(s)", s.link_clients);
-            println!(
-                "Held mail:  {} message(s), {} bytes",
-                s.spool.inbound, s.spool.bytes
-            );
+            println!("Held mail:  {} message(s)", s.spool.inbound);
+            println!("Outbound:   {} message(s)", s.spool.outbound);
+            println!("Spool:      {} bytes", s.spool.bytes);
             if let Some(e) = s.last_error {
                 println!("Last error: {e}");
             }

crates/relay/src/outbound.rs

@@ -0,0 +1,550 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! The outbound spool (spec "Outbound", "The spool", RL-19).
+//!
+//! Each message is two files in `spool/out/`: `<id>.eml`, the message as
+//! the downstream handed it over, moved in from its upload and fsynced
+//! first; then `<id>.json`, its properties and delivery status, written
+//! atomically. The `.json` is the commit point, as in the inbound spool.
+//! Outbound mail isn't sealed: the relay has to read it to send it.
+
+use crate::{
+    changelog::{Changelog, Changes},
+    state::{set_private, sync_dir, write_atomic},
+};
+use inbuxa_relay_proto::{
+    time,
+    types::{Address, DeliveryStatus, Envelope, Outbound, RelaySettings},
+};
+use serde::{Deserialize, Serialize};
+use std::{
+    collections::{BTreeMap, HashSet},
+    fs, io,
+    path::{Path, PathBuf},
+};
+
+/// Longest address, as RFC 5321 §4.5.3.1 allows for a path.
+const ADDRESS_MAX: usize = 256;
+const PARAMETER_NAME_MAX: usize = 64;
+const PARAMETER_VALUE_MAX: usize = 500;
+
+/// A message's stored properties.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Meta {
+    pub id: String,
+    pub created_at: u64,
+    pub blob_id: String,
+    pub envelope: Envelope,
+    pub size: u64,
+    pub delivery_status: BTreeMap<String, DeliveryStatus>,
+    pub next_attempt_at: u64,
+    pub attempts: u32,
+    pub expires_at: u64,
+}
+
+struct Entry {
+    meta: Meta,
+    created: u64,
+    modseq: u64,
+}
+
+/// Why a create was refused: a JMAP `SetError` type and a description.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Refusal {
+    pub kind: &'static str,
+    pub description: String,
+}
+
+fn refuse(kind: &'static str, description: impl Into<String>) -> Refusal {
+    Refusal {
+        kind,
+        description: description.into(),
+    }
+}
+
+pub struct Outbox {
+    dir: PathBuf,
+    entries: BTreeMap<String, Entry>,
+    log: Changelog,
+    bytes: u64,
+}
+
+pub fn new_id() -> String {
+    let mut id = crate::spool::new_id();
+    id.replace_range(0..1, "o");
+    id
+}
+
+/// Whether `id` is one of ours: never a path.
+pub fn is_valid_id(id: &str) -> bool {
+    id.len() == 25 && id.starts_with('o') && id[1..].bytes().all(|b| b.is_ascii_hexdigit())
+}
+
+impl Outbox {
+    /// Opens the spool, deleting what was never committed.
+    pub fn open(dir: &Path) -> io::Result<Outbox> {
+        fs::create_dir_all(dir)?;
+        let mut outbox = Outbox {
+            dir: dir.to_path_buf(),
+            entries: BTreeMap::new(),
+            log: Changelog::default(),
+            bytes: 0,
+        };
+        let mut messages = Vec::new();
+        for entry in fs::read_dir(dir)? {
+            let entry = entry?;
+            let name = entry.file_name().to_string_lossy().into_owned();
+            if let Some(id) = name.strip_suffix(".json") {
+                if !is_valid_id(id) {
+                    continue;
+                }
+                let meta: Meta = match fs::read(entry.path())
+                    .ok()
+                    .and_then(|b| serde_json::from_slice(&b).ok())
+                {
+                    Some(meta) => meta,
+                    None => {
+                        eprintln!("inbuxa-relay: unreadable outbound entry {id}, skipped");
+                        continue;
+                    }
+                };
+                if !outbox.eml_path(id).exists() {
+                    let _ = fs::remove_file(entry.path());
+                    continue;
+                }
+                outbox.bytes += meta.size;
+                outbox.entries.insert(
+                    id.to_string(),
+                    Entry {
+                        meta,
+                        created: 1,
+                        modseq: 1,
+                    },
+                );
+            } else if name.ends_with(".eml") || name.ends_with(".tmp") {
+                messages.push((name, entry.path()));
+            }
+        }
+        for (name, path) in messages {
+            let id = name.split('.').next().unwrap_or("");
+            if name.ends_with(".tmp") || !outbox.entries.contains_key(id) {
+                let _ = fs::remove_file(path);
+            }
+        }
+        Ok(outbox)
+    }
+
+    pub fn eml_path(&self, id: &str) -> PathBuf {
+        self.dir.join(format!("{id}.eml"))
+    }
+
+    fn meta_path(&self, id: &str) -> PathBuf {
+        self.dir.join(format!("{id}.json"))
+    }
+
+    pub fn state(&self) -> String {
+        self.log.state()
+    }
+
+    pub fn count(&self) -> u64 {
+        self.entries.len() as u64
+    }
+
+    pub fn bytes(&self) -> u64 {
+        self.bytes
+    }
+
+    /// RL-19: takes the uploaded message at `blob` for `envelope`. The
+    /// upload is moved into the spool, so it can't be used twice. `full`
+    /// is whether the spool is past its limit.
+    pub fn create(
+        &mut self,
+        blob: &Path,
+        blob_id: &str,
+        envelope: Envelope,
+        settings: &RelaySettings,
+        full: bool,
+        now: u64,
+    ) -> Result<String, Refusal> {
+        check_envelope(&envelope, settings)?;
+        let size = fs::metadata(blob)
+            .map_err(|_| refuse("invalidProperties", "blobId: no such upload"))?
+            .len();
+        if size == 0 {
+            return Err(refuse("invalidProperties", "the message is empty"));
+        }
+        if size > settings.max_message_size {
+            return Err(refuse(
+                "tooLarge",
+                format!("the message is over {} bytes", settings.max_message_size),
+            ));
+        }
+        if full {
+            return Err(refuse("overQuota", "the relay's spool is full"));
+        }
+        let id = new_id();
+        let meta = Meta {
+            id: id.clone(),
+            created_at: now,
+            blob_id: blob_id.to_string(),
+            delivery_status: envelope
+                .rcpt_to
+                .iter()
+                .map(|r| (r.email.clone(), DeliveryStatus::queued()))
+                .collect(),
+            envelope,
+            size,
+            next_attempt_at: now,
+            attempts: 0,
+            expires_at: now + u64::from(settings.outbound_expiry),
+        };
+        self.commit(blob, &meta)
+            .map_err(|e| refuse("serverFail", format!("could not spool the message: {e}")))?;
+        let modseq = self.log.bump();
+        self.bytes += size;
+        self.entries.insert(
+            id.clone(),
+            Entry {
+                meta,
+                created: modseq,
+                modseq,
+            },
+        );
+        Ok(id)
+    }
+
+    fn commit(&self, blob: &Path, meta: &Meta) -> io::Result<()> {
+        let eml = self.eml_path(&meta.id);
+        fs::File::open(blob)?.sync_all()?;
+        fs::rename(blob, &eml)?;
+        set_private(&eml, false)?;
+        sync_dir(&self.dir)?;
+        let json = serde_json::to_vec(meta).map_err(io::Error::other)?;
+        if let Err(e) = write_atomic(&self.meta_path(&meta.id), &json) {
+            let _ = fs::remove_file(&eml);
+            return Err(e);
+        }
+        Ok(())
+    }
+
+    /// The JMAP object for one message.
+    pub fn get(&self, id: &str) -> Option<Outbound> {
+        let m = &self.entries.get(id)?.meta;
+        let queued = m
+            .delivery_status
+            .values()
+            .any(|s| s.delivered == inbuxa_relay_proto::types::Delivered::Queued);
+        Some(Outbound {
+            id: m.id.clone(),
+            created_at: time::utc_date(m.created_at),
+            blob_id: m.blob_id.clone(),
+            envelope: m.envelope.clone(),
+            size: m.size,
+            delivery_status: m.delivery_status.clone(),
+            next_attempt_at: queued.then(|| time::utc_date(m.next_attempt_at)),
+            attempts: m.attempts,
+            expires_at: time::utc_date(m.expires_at),
+        })
+    }
+
+    /// Ids, oldest first.
+    pub fn ids(&self) -> Vec<String> {
+        let mut ids: Vec<&Meta> = self.entries.values().map(|e| &e.meta).collect();
+        ids.sort_by_key(|m| (m.created_at, m.id.clone()));
+        ids.into_iter().map(|m| m.id.clone()).collect()
+    }
+
+    pub fn changes(&self, since: &str, max: usize) -> Option<Changes> {
+        self.log.changes(
+            since,
+            max,
+            self.entries.iter().map(|(id, e)| (id, e.created, e.modseq)),
+        )
+    }
+
+    /// Cancels a queued message, or clears a finished one.
+    pub fn destroy(&mut self, id: &str) -> bool {
+        let Some(e) = self.entries.remove(id) else {
+            return false;
+        };
+        self.bytes = self.bytes.saturating_sub(e.meta.size);
+        let _ = fs::remove_file(self.meta_path(id));
+        let _ = fs::remove_file(self.eml_path(id));
+        let _ = sync_dir(&self.dir);
+        self.log.destroyed(id);
+        true
+    }
+}
+
+/// RL-19: a sender at one of the relay's domains, or none; real recipients;
+/// SMTP parameters that can't break the command line they go on.
+pub fn check_envelope(envelope: &Envelope, settings: &RelaySettings) -> Result<(), Refusal> {
+    let from = &envelope.mail_from;
+    if !from.email.is_empty() {
+        let domain = domain_of(&from.email)
+            .ok_or_else(|| refuse("invalidProperties", "mailFrom isn't an address"))?;
+        if !settings.serves(domain) {
+            return Err(refuse(
+                "forbiddenFrom",
+                "mailFrom isn't at one of the relay's domains",
+            ));
+        }
+    }
+    check_parameters(from, "mailFrom")?;
+    if envelope.rcpt_to.is_empty() {
+        return Err(refuse("invalidProperties", "rcptTo needs a recipient"));
+    }
+    if envelope.rcpt_to.len() > settings.max_recipients as usize {
+        return Err(refuse(
+            "tooLarge",
+            format!("at most {} recipients", settings.max_recipients),
+        ));
+    }
+    let mut seen = HashSet::new();
+    for rcpt in &envelope.rcpt_to {
+        if domain_of(&rcpt.email).is_none() {
+            return Err(refuse(
+                "invalidProperties",
+                format!("rcptTo: {} isn't an address", rcpt.email),
+            ));
+        }
+        if !seen.insert(rcpt.email.to_lowercase()) {
+            return Err(refuse(
+                "invalidProperties",
+                format!("rcptTo: {} is listed twice", rcpt.email),
+            ));
+        }
+        check_parameters(rcpt, "rcptTo")?;
+    }
+    Ok(())
+}
+
+/// The domain of a plain address, if it is one.
+fn domain_of(email: &str) -> Option<&str> {
+    if email.len() > ADDRESS_MAX
+        || email
+            .chars()
+            .any(|c| c.is_control() || c.is_whitespace() || matches!(c, '<' | '>'))
+    {
+        return None;
+    }
+    let (local, domain) = email.rsplit_once('@')?;
+    let quoted = local.len() > 1 && local.starts_with('"') && local.ends_with('"');
+    let domain = domain.trim_end_matches('.');
+    (!local.is_empty() && !domain.is_empty() && (quoted || !local.contains('@'))).then_some(domain)
+}
+
+fn check_parameters(address: &Address, field: &str) -> Result<(), Refusal> {
+    for (name, value) in address.parameters.iter().flatten() {
+        let name_ok = !name.is_empty()
+            && name.len() <= PARAMETER_NAME_MAX
+            && name.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-');
+        let value_ok = value.as_ref().is_none_or(|v| {
+            !v.is_empty()
+                && v.len() <= PARAMETER_VALUE_MAX
+                && !v
+                    .chars()
+                    .any(|c| c.is_control() || c.is_whitespace() || c == '=')
+        });
+        if !name_ok || !value_ok {
+            return Err(refuse(
+                "invalidProperties",
+                format!("{field}: the parameter {name} isn't valid"),
+            ));
+        }
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use inbuxa_relay_proto::types::Delivered;
+
+    fn settings() -> RelaySettings {
+        RelaySettings {
+            domains: vec!["example.com".into()],
+            max_recipients: 3,
+            max_message_size: 100,
+            ..Default::default()
+        }
+    }
+
+    fn envelope(from: &str, to: &[&str]) -> Envelope {
+        Envelope {
+            mail_from: Address::new(from),
+            rcpt_to: to.iter().map(|t| Address::new(*t)).collect(),
+        }
+    }
+
+    fn dir(name: &str) -> PathBuf {
+        let d = std::env::temp_dir().join(format!(
+            "inbuxa-relay-outbound-{name}-{}",
+            std::process::id()
+        ));
+        let _ = fs::remove_dir_all(&d);
+        d
+    }
+
+    fn upload(dir: &Path, body: &[u8]) -> PathBuf {
+        fs::create_dir_all(dir).unwrap();
+        let path = dir.join(crate::spool::new_id());
+        fs::write(&path, body).unwrap();
+        path
+    }
+
+    #[test]
+    fn senders_only_from_the_relays_domains() {
+        let s = settings();
+        assert!(check_envelope(&envelope("[email protected]", &["[email protected]"]), &s).is_ok());
+        assert!(check_envelope(&envelope("[email protected].", &["[email protected]"]), &s).is_ok());
+        assert!(
+            check_envelope(&envelope("", &["[email protected]"]), &s).is_ok(),
+            "a bounce"
+        );
+        for from in [
+            "[email protected]",
+            "[email protected]",
+            "[email protected]@evil.org",
+        ] {
+            assert_eq!(
+                check_envelope(&envelope(from, &["[email protected]"]), &s)
+                    .unwrap_err()
+                    .kind,
+                if from.contains("@evil") {
+                    "invalidProperties"
+                } else {
+                    "forbiddenFrom"
+                },
+                "{from}"
+            );
+        }
+    }
+
+    #[test]
+    fn recipients_and_parameters_are_checked() {
+        let s = settings();
+        let bad = [
+            envelope("[email protected]", &[]),
+            envelope(
+                "[email protected]",
+                &["[email protected]", "[email protected]", "[email protected]", "[email protected]"],
+            ),
+            envelope("[email protected]", &["[email protected]", "[email protected]"]),
+            envelope("[email protected]", &["[email protected]\r\nRCPT TO:<[email protected]>"]),
+            envelope("[email protected]", &["no-at-sign"]),
+            envelope("[email protected]", &["<[email protected]>"]),
+        ];
+        for e in &bad {
+            assert!(check_envelope(e, &s).is_err(), "{e:?}");
+        }
+        let mut e = envelope("[email protected]", &["[email protected]"]);
+        e.rcpt_to[0].parameters = Some(BTreeMap::from([("NOTIFY".into(), Some("NEVER".into()))]));
+        e.mail_from.parameters = Some(BTreeMap::from([
+            ("BODY".into(), Some("8BITMIME".into())),
+            ("SMTPUTF8".into(), None),
+        ]));
+        assert!(check_envelope(&e, &s).is_ok());
+        e.rcpt_to[0].parameters = Some(BTreeMap::from([(
+            "NOTIFY".into(),
+            Some("NEVER AUTH=x".into()),
+        )]));
+        assert!(check_envelope(&e, &s).is_err());
+    }
+
+    #[test]
+    fn created_survives_restart_and_destroys() {
+        let d = dir("restart");
+        let blobs = d.join("blobs");
+        let out = d.join("out");
+        let s = settings();
+        let mut outbox = Outbox::open(&out).unwrap();
+        let s0 = outbox.state();
+
+        let blob = upload(&blobs, b"Subject: hi\r\n\r\nhi\r\n");
+        let id = outbox
+            .create(
+                &blob,
+                "b1",
+                envelope("[email protected]", &["[email protected]", "[email protected]"]),
+                &s,
+                false,
+                100,
+            )
+            .unwrap();
+        assert!(!blob.exists(), "the upload moved into the spool");
+        assert_eq!(
+            fs::read(outbox.eml_path(&id)).unwrap(),
+            b"Subject: hi\r\n\r\nhi\r\n"
+        );
+        let got = outbox.get(&id).unwrap();
+        assert_eq!(got.blob_id, "b1");
+        assert_eq!(got.size, 19);
+        assert_eq!(got.delivery_status.len(), 2);
+        assert!(
+            got.delivery_status
+                .values()
+                .all(|s| s.delivered == Delivered::Queued)
+        );
+        assert_eq!(got.next_attempt_at.as_deref(), Some(&*time::utc_date(100)));
+        assert_eq!(got.expires_at, time::utc_date(100 + 3 * 86_400));
+        assert_eq!(outbox.changes(&s0, 10).unwrap().created, vec![id.clone()]);
+
+        // A crash left an uncommitted message.
+        fs::write(out.join(format!("{}.eml", new_id())), b"junk").unwrap();
+        let mut outbox = Outbox::open(&out).unwrap();
+        assert_eq!(outbox.count(), 1);
+        assert_eq!(outbox.bytes(), 19);
+        assert_eq!(fs::read_dir(&out).unwrap().count(), 2);
+
+        let s1 = outbox.state();
+        assert!(outbox.destroy(&id));
+        assert!(!outbox.destroy(&id));
+        assert_eq!(outbox.changes(&s1, 10).unwrap().destroyed, vec![id]);
+        assert_eq!(fs::read_dir(&out).unwrap().count(), 0);
+        fs::remove_dir_all(d).unwrap();
+    }
+
+    #[test]
+    fn create_refusals() {
+        let d = dir("refusals");
+        let blobs = d.join("blobs");
+        let s = settings();
+        let mut outbox = Outbox::open(&d.join("out")).unwrap();
+        let ok = envelope("[email protected]", &["[email protected]"]);
+
+        let big = upload(&blobs, &[b'x'; 101]);
+        let r = outbox.create(&big, "b", ok.clone(), &s, false, 0);
+        assert_eq!(r.unwrap_err().kind, "tooLarge");
+        assert!(big.exists(), "a refused upload stays where it was");
+
+        let blob = upload(&blobs, b"x");
+        let r = outbox.create(&blob, "b", ok.clone(), &s, true, 0);
+        assert_eq!(r.unwrap_err().kind, "overQuota");
+        let r = outbox.create(
+            &blob,
+            "b",
+            envelope("[email protected]", &["[email protected]"]),
+            &s,
+            false,
+            0,
+        );
+        assert_eq!(r.unwrap_err().kind, "forbiddenFrom");
+        let r = outbox.create(&blobs.join("gone"), "b", ok, &s, false, 0);
+        assert_eq!(r.unwrap_err().kind, "invalidProperties");
+        assert_eq!(outbox.count(), 0);
+        fs::remove_dir_all(d).unwrap();
+    }
+
+    #[test]
+    fn ids_are_never_paths() {
+        assert!(is_valid_id(&new_id()));
+        assert!(!crate::spool::is_valid_id(&new_id()));
+        assert!(!is_valid_id("o../../etc/passwd"));
+    }
+}

crates/relay/src/spool.rs

@@ -15,7 +15,10 @@
 //! Leases and live verdict waiters are runtime only. After a restart every
 //! message is held and unleased; the downstream deduplicates by id (RL-16).
 
-use crate::state::{set_private, sync_dir, write_atomic};
+use crate::{
+    changelog::Changelog,
+    state::{set_private, sync_dir, write_atomic},
+};
 use inbuxa_relay_proto::{
     seal::Sealer,
     time,
@@ -23,14 +26,13 @@ use inbuxa_relay_proto::{
 };
 use serde::{Deserialize, Serialize};
 use std::{
-    collections::{BTreeMap, VecDeque},
+    collections::BTreeMap,
     fs, io,
     path::{Path, PathBuf},
 };
 use tokio::{io::AsyncWriteExt, sync::oneshot};
 
-/// How many destroyed ids `/changes` can still report.
-const TOMBSTONES: usize = 10_000;
+pub use crate::changelog::Changes;
 
 /// A message's stored properties.
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -74,16 +76,6 @@ impl Entry {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct Changes {
-    pub old_state: String,
-    pub new_state: String,
-    pub created: Vec<String>,
-    pub updated: Vec<String>,
-    pub destroyed: Vec<String>,
-    pub has_more: bool,
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum VerdictError {
     NotFound,
@@ -96,10 +88,7 @@ pub enum VerdictError {
 pub struct Spool {
     dir: PathBuf,
     entries: BTreeMap<String, Entry>,
-    modseq: u64,
-    tombstones: VecDeque<(u64, String)>,
-    /// `/changes` can't go back past this.
-    floor: u64,
+    log: Changelog,
     bytes: u64,
 }
 
@@ -129,9 +118,7 @@ impl Spool {
         let mut spool = Spool {
             dir: dir.to_path_buf(),
             entries: BTreeMap::new(),
-            modseq: 1,
-            tombstones: VecDeque::new(),
-            floor: 1,
+            log: Changelog::default(),
             bytes: 0,
         };
         let mut msgs = Vec::new();
@@ -191,7 +178,7 @@ impl Spool {
     }
 
     pub fn state(&self) -> String {
-        self.modseq.to_string()
+        self.log.state()
     }
 
     pub fn count(&self) -> u64 {
@@ -207,8 +194,7 @@ impl Spool {
     }
 
     fn bump(&mut self) -> u64 {
-        self.modseq += 1;
-        self.modseq
+        self.log.bump()
     }
 
     /// Starts writing a new message in the spool directory `dir`, sealed
@@ -298,38 +284,11 @@ impl Spool {
     }
 
     pub fn changes(&self, since: &str, max: usize) -> Option<Changes> {
-        let since: u64 = since.parse().ok()?;
-        if since < self.floor || since > self.modseq {
-            return None;
-        }
-        let mut created = Vec::new();
-        let mut updated = Vec::new();
-        for (id, e) in &self.entries {
-            if e.created > since {
-                created.push(id.clone());
-            } else if e.modseq > since {
-                updated.push(id.clone());
-            }
-        }
-        let destroyed: Vec<String> = self
-            .tombstones
-            .iter()
-            .filter(|(seq, _)| *seq > since)
-            .map(|(_, id)| id.clone())
-            .collect();
-        let total = created.len() + updated.len() + destroyed.len();
-        if total > max {
-            // Too much at once: the client starts over with /query.
-            return None;
-        }
-        Some(Changes {
-            old_state: since.to_string(),
-            new_state: self.state(),
-            created,
-            updated,
-            destroyed,
-            has_more: false,
-        })
+        self.log.changes(
+            since,
+            max,
+            self.entries.iter().map(|(id, e)| (id, e.created, e.modseq)),
+        )
     }
 
     /// RL-15: leases up to `limit` unleased messages to `conn`.
@@ -438,13 +397,7 @@ impl Spool {
         let _ = fs::remove_file(self.meta_path(id));
         let _ = fs::remove_file(self.msg_path(id));
         let _ = sync_dir(&self.dir);
-        let modseq = self.bump();
-        self.tombstones.push_back((modseq, id.to_string()));
-        while self.tombstones.len() > TOMBSTONES {
-            if let Some((seq, _)) = self.tombstones.pop_front() {
-                self.floor = seq;
-            }
-        }
+        self.log.destroyed(id);
         true
     }
 

crates/relay/src/state.rs

@@ -55,12 +55,21 @@ impl Paths {
         self.root.join("spool").join("in")
     }
 
+    pub fn spool_out(&self) -> PathBuf {
+        self.root.join("spool").join("out")
+    }
+
     pub fn blobs(&self) -> PathBuf {
         self.root.join("blobs")
     }
 
     pub fn create_dirs(&self) -> io::Result<()> {
-        for dir in [self.root.clone(), self.spool_in(), self.blobs()] {
+        for dir in [
+            self.root.clone(),
+            self.spool_in(),
+            self.spool_out(),
+            self.blobs(),
+        ] {
             fs::create_dir_all(&dir)?;
             set_private(&dir, true)?;
         }

crates/relay/src/tests_e2e.rs

@@ -622,6 +622,58 @@ async fn relay_end_to_end() {
     let got = ws.ok("Inbound/get", json!({ "ids": [held_id] })).await;
     assert_eq!(got["list"][0]["state"], "held");
 
+    // Outbound: handed over as an upload (RL-19), never from a sender
+    // outside the relay's domains (RL-T10), and cancelled while queued.
+    let (status, body) = down
+        .http(
+            "POST",
+            "/jmap/upload/relay",
+            b"Subject: out\r\n\r\nhello\r\n".to_vec(),
+        )
+        .await;
+    assert_eq!(status, 201);
+    let blob: Value = serde_json::from_slice(&body).unwrap();
+    let envelope = |from: &str| json!({ "mailFrom": { "email": from }, "rcptTo": [{ "email": "[email protected]" }] });
+    let set = ws
+        .ok(
+            "Outbound/set",
+            json!({ "create": {
+                "a": { "blobId": blob["blobId"], "envelope": envelope("[email protected]") },
+                "b": { "blobId": blob["blobId"], "envelope": envelope("[email protected]") },
+            }}),
+        )
+        .await;
+    assert_eq!(set["notCreated"]["b"]["type"], "forbiddenFrom", "{set}");
+    let out_id = set["created"]["a"]["id"].as_str().unwrap().to_string();
+    assert_eq!(
+        set["created"]["a"]["deliveryStatus"]["[email protected]"]["delivered"],
+        "queued"
+    );
+    ws.wait_change("Outbound").await;
+    let got = ws.ok("Outbound/get", json!({ "ids": [out_id] })).await;
+    assert_eq!(got["list"][0]["size"], 23);
+    let eml = app.paths.spool_out().join(format!("{out_id}.eml"));
+    assert!(eml.exists());
+    let relay = ws.ok("Relay/get", json!({})).await;
+    assert_eq!(relay["list"][0]["status"]["spool"]["outbound"], 1);
+    let again = ws
+        .ok(
+            "Outbound/set",
+            json!({
+                "create": { "c": { "blobId": blob["blobId"], "envelope": envelope("[email protected]") }},
+                "update": { out_id.clone(): { "attempts": 9 }},
+            }),
+        )
+        .await;
+    assert_eq!(
+        again["notCreated"]["c"]["type"], "invalidProperties",
+        "an upload is used once"
+    );
+    assert_eq!(again["notUpdated"][&out_id]["type"], "forbidden");
+    let gone = ws.ok("Outbound/set", json!({ "destroy": [out_id] })).await;
+    assert_eq!(gone["destroyed"], json!([out_id]));
+    assert!(!eml.exists());
+
     // Events went up the link.
     let events = ws.ok("RelayEvent/get", json!({})).await;
     let kinds: Vec<&str> = events["list"]

docs/spec/features/relay.md

@@ -306,8 +306,17 @@ so a reboot without the link comes back with the same settings (RL-22).
 | `id`, `createdAt` | |
 | `blobId` | The message as the downstream signed it |
 | `envelope` | RFC 8621 `Envelope` |
-| `deliveryStatus` | RFC 8621 `DeliveryStatus` per recipient: `smtpReply`, `delivered` (`queued`, `yes`, `no`, `unknown`) |
-| `nextAttemptAt`, `attempts`, `expiresAt` | Retry state |
+| `size` | Bytes |
+| `deliveryStatus` | RFC 8621 `DeliveryStatus` per recipient: `smtpReply` (empty before the first try), `delivered` (`queued`, `yes`, `no`, `unknown`) |
+| `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
+update is allowed. The upload moves into the spool, so it can't be used
+twice. Besides `forbiddenFrom` (RL-19), a create is refused as `tooLarge`
+over `maxMessageSize` or `maxRecipients`, `overQuota` when the spool is
+full, and `invalidProperties` for an empty `rcptTo`, a recipient listed
+twice, or an address or SMTP parameter that couldn't go on a command line
+as it is.
 
 **`RelayEvent`**: read-only, `{id, at, type, data}`. The relay's sessions,
 deliveries, rejections and errors, for the downstream's history and tracer.
@@ -761,6 +770,10 @@ In progress. Done so far:
   link to a joined relay for an hour raises `relay.link-down` (a warning),
   once, and `relay.link-up` when it is back (RL-T23). Both reach the log,
   webhooks and OpenTelemetry.
+- `Outbound` on the relay: `Outbound/set` create and destroy, `/get` and
+  `/changes`, with push, in `spool/out/` (RL-19, RL-T10). The relay doesn't
+  deliver it yet, and the downstream doesn't hand anything over yet; both
+  are the rest of step 4.
 - Not done yet: `inbuxa:RelayMessage`, which comes with inbuxa-admin's
   Relays page (step 5), once `Outbound` exists to show beside `Inbound`.