Writing your own mail transport

Rullst Mail ships four delivery transports: Resend, AWS SES, SendPulse and SMTP, plus the log, memory and offline mock_* drivers for development and tests. Every other provider is integrated by the application through the same MailDriver trait the built-in transports implement. Version 13 removed the SendGrid, Postmark, Mailjet, Mailtrap and Azure Communication Services transports; see the migration guide row “Mail providers removed”.

This guide walks through a complete transport for a fictitious HTTP email API, AcmeMail. The code below is included from rullst-mail/examples/custom_transport.rs, which CI compiles and which runs offline from a clone of the Rullst repository:

cargo run -p rullst-mail --example custom_transport

In an application, the same types are available through the facade as rullst::mail::* with the mail feature. The example also uses reqwest, secrecy, serde_json, chrono and async-trait (MailDriver is an #[async_trait] trait; the facade’s rullst::async_trait works too); add equivalent crates to your own Cargo.toml.

What the framework does and what the transport does

The transport is trusted server code. A transport that cannot express part of a message (a schedule, an inline image, an unsubscribe header) must return an error instead of sending the message without it.

1. Implement MailDriver

/// Stable low-cardinality label used in errors and telemetry.
const PROVIDER: &str = "acmemail";
/// Largest delivery receipt read from AcmeMail.
const MAX_RECEIPT_BYTES: usize = 16 * 1024;

/// Messages the offline fixture captured instead of sending.
pub type OfflineOutbox = Arc<Mutex<Vec<Message>>>;

/// Transport for the fictitious AcmeMail HTTP API.
pub struct AcmeMailDriver {
    api_key: SecretString,
    endpoint: String,
    client: reqwest::Client,
    outbox: OfflineOutbox,
}

impl AcmeMailDriver {
    /// Creates the transport. Empty or `mock_*` keys select the offline fixture.
    pub fn try_new(api_key: impl Into<String>) -> Result<Self, MailError> {
        let api_key = api_key.into();
        if api_key.contains(['\r', '\n']) || api_key.len() > 4096 {
            return Err(MailError::ConfigError(
                "AcmeMail API key is malformed".into(),
            ));
        }
        // No redirects (they would forward the key and the message), and
        // deadlines for connecting and for the whole exchange.
        let client = reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .connect_timeout(Duration::from_secs(5))
            .timeout(Duration::from_secs(30))
            .build()
            .map_err(|_| MailError::ConfigError("cannot build the AcmeMail client".into()))?;
        Ok(Self {
            api_key: SecretString::from(api_key),
            endpoint: "https://api.acmemail.invalid/v1/messages".into(),
            client,
            outbox: OfflineOutbox::default(),
        })
    }

    /// Uses another endpoint: HTTPS, or plain HTTP to a loopback test fixture.
    pub fn try_with_endpoint(mut self, endpoint: impl Into<String>) -> Result<Self, MailError> {
        let endpoint = endpoint.into();
        let loopback = endpoint.starts_with("http://127.0.0.1:");
        if !endpoint.starts_with("https://") && !loopback {
            return Err(MailError::ConfigError(
                "AcmeMail endpoint must use HTTPS".into(),
            ));
        }
        self.endpoint = endpoint;
        Ok(self)
    }

    /// Whether the key selects the real API or the offline fixture.
    pub fn delivery_mode(&self) -> DeliveryMode {
        credential_mode(self.api_key.expose_secret())
    }

    /// The offline fixture's captured messages, shared with every clone.
    pub fn offline_outbox(&self) -> OfflineOutbox {
        Arc::clone(&self.outbox)
    }

    /// The API request body. Content the API cannot express is an error,
    /// never silently dropped.
    fn payload(message: &Message) -> Result<Value, MailError> {
        let from = message.from.as_deref().ok_or_else(|| {
            MailError::ConfigError(
                "AcmeMail requires a verified sender: set `from`, or MAIL_FROM for the Mail facade"
                    .into(),
            )
        })?;
        if message
            .send_at
            .is_some_and(|send_at| send_at > chrono::Utc::now())
        {
            return Err(MailError::ConfigError(
                "AcmeMail cannot schedule mail; use Mail::enqueue with a durable queue".into(),
            ));
        }
        if message.attachments.iter().any(|item| item.is_inline()) {
            return Err(MailError::ValidationError(
                "AcmeMail has no inline images; use a transport that supports CIDs".into(),
            ));
        }
        let mut headers = serde_json::Map::new();
        if let Some(value) = message.list_unsubscribe_header() {
            headers.insert("List-Unsubscribe".into(), Value::String(value));
            // RFC 8058 one-click unsubscription requires an HTTPS URI.
            if message
                .unsubscribe_url
                .as_deref()
                .is_some_and(|url| url.starts_with("https://"))
            {
                headers.insert(
                    "List-Unsubscribe-Post".into(),
                    Value::String("List-Unsubscribe=One-Click".into()),
                );
            }
        }
        let attachments: Vec<Value> = message
            .attachments
            .iter()
            .map(|item| {
                json!({
                    "filename": item.filename,
                    "content_type": item.mime_type,
                    "content": item.to_base64(),
                })
            })
            .collect();
        Ok(json!({
            "from": from,
            "to": message.to,
            "subject": message.subject,
            "text": message.body_text,
            "html": message.body_html,
            "headers": headers,
            "attachments": attachments,
        }))
    }

    /// Reads at most `MAX_RECEIPT_BYTES` and requires a message ID.
    async fn receipt(mut response: reqwest::Response) -> Result<(), MailError> {
        let invalid = || MailError::SendError("AcmeMail returned an invalid receipt".into());
        let mut body = Vec::new();
        while let Some(chunk) = response
            .chunk()
            .await
            .map_err(|_| MailError::transport(PROVIDER, "receipt read failed"))?
        {
            if body.len() + chunk.len() > MAX_RECEIPT_BYTES {
                return Err(invalid());
            }
            body.extend_from_slice(&chunk);
        }
        let receipt: Value = serde_json::from_slice(&body).map_err(|_| invalid())?;
        match receipt.get("id").and_then(Value::as_str) {
            Some(id) if !id.is_empty() && id.len() <= 128 => Ok(()),
            _ => Err(invalid()),
        }
    }
}

#[async_trait]
impl MailDriver for AcmeMailDriver {
    async fn send(&self, message: &Message) -> Result<(), MailError> {
        // The shared pre-flight checks: header injection, recipients,
        // outbound secrets, links and attachment bounds.
        let prepared = DeliveryPipeline::prepare(message)?;
        let message = prepared.message();
        let body = Self::payload(message)?;
        if self.delivery_mode() == DeliveryMode::OfflineMock {
            self.outbox
                .lock()
                .map_err(|_| MailError::DriverError("AcmeMail outbox is unavailable".into()))?
                .push(message.clone());
            return Ok(());
        }
        let response = self
            .client
            .post(&self.endpoint)
            .bearer_auth(self.api_key.expose_secret())
            .json(&body)
            .send()
            .await
            .map_err(|_| MailError::transport(PROVIDER, "request failed before a response"))?;
        if response.status().is_success() {
            return Self::receipt(response).await;
        }
        let retry_after = response
            .headers()
            .get(reqwest::header::RETRY_AFTER)
            .and_then(|value| value.to_str().ok())
            .and_then(|value| value.parse::<u64>().ok())
            .map(|seconds| Duration::from_secs(seconds.min(86_400)));
        // Error bodies can echo addresses and content: keep only the status
        // and the retry delay. 429 becomes `RateLimited`, 5xx is transient.
        Err(MailError::from_provider_response(
            PROVIDER,
            response.status().as_u16(),
            "provider response body omitted",
            retry_after,
        ))
    }
}

Points that matter for security and testing:

• Run the pipeline first. DeliveryPipeline::prepare applies the same pre-flight checks as the built-in transports; send prepared.message(), never the original.
• Never invent a sender. A message without from fails with MailError::ConfigError before any request. The Mail facade fills it from MAIL_FROM or [mail] from before the transport sees it.
• Reject what the API cannot express. A future send_at without provider scheduling, inline CID images or headers the API has no field for are errors, not silent drops. Durable scheduling belongs to Mail::enqueue.
• Offline fallback. An empty or mock_* key selects DeliveryMode::OfflineMock through credential_mode, and the transport captures the message without network I/O, as section 3.5 of Rullst’s AGENTS.md requires. Validate the payload before that check so the fixture enforces the live contract.
• Bound the HTTP exchange. Disable redirects (they would forward the key and the message), set connect and request deadlines, require HTTPS except for a loopback test fixture and read at most a bounded receipt.
• Classify failures. MailError::from_provider_response turns HTTP 429 into RateLimited with its Retry-After and 5xx into a transient error; MailError::transport marks a failure before any response. FailoverDriver moves only transient and rate-limited failures to a fallback. Keep provider error bodies out of the error: they can echo addresses and content.
• Accept only a real receipt. Success requires the provider’s message ID; a 2xx without one is a SendError. Acceptance is not proof of inbox delivery.
• Idempotency. send_with_delivery_id defaults to at-least-once. Override it only when the provider deduplicates by a key you forward.

2. Install it

    // 4. Install the transport for the `Mail` facade (and its queue worker)...
    let transport = AcmeMailDriver::try_new("mock_acme")?;
    let facade_outbox = transport.offline_outbox();
    Mail::set_driver(Box::new(transport));
    Mail::send_now(welcome("[email protected]")).await?;
    Mail::reset_driver();

    // ...or for one tenant, with the global default for everyone else.
    let (default_driver, default_store) = MemoryDriver::isolated();
    let resolver = TenantMailResolver::with_default(default_driver);
    resolver.register("tenant_acme", AcmeMailDriver::try_new("mock_tenant")?)?;
    resolver
        .send_for_tenant("tenant_acme", &welcome("[email protected]"))
        .await?;
    resolver
        .send_for_tenant("tenant_other", &welcome("[email protected]"))
        .await?;

    // Shared guards wrap a custom transport like a built-in one.
    let store = InMemorySuppressionStore::new(16, 16)?;
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)?
        .as_secs();
    store
        .record(SuppressionEvent::try_new(
            PROVIDER,
            "evt_bounce_1",
            "[email protected]",
            SuppressionReason::HardBounce,
            now,
        )?)
        .await?;
    let guarded = SuppressionGuard::new(AcmeMailDriver::try_new("mock_acme")?, store);
    assert!(matches!(
        guarded.send(&welcome("[email protected]")).await,
        Err(MailError::SuppressedRecipient { .. })
    ));

Mail::set_driver replaces the facade’s configured driver for every send and for the queue worker (register_mail_handler) in that process, so install it at start-up in every process that sends or works the mail queue. Mail::reset_driver returns to MAIL_DRIVER. A TenantMailResolver selects a transport per tenant, with a default for the others. Suppression, attachment inspection and observation wrap the transport with static dispatch, as they wrap the built-in ones.

A MAIL_DRIVER value cannot name a custom transport: the facade builds only the kept drivers from configuration, and a removed driver name fails with MailError::ConfigError instead of falling back.

3. Test the transport

The example’s main exercises the contract offline:

• a mock_* key captures the message and never reaches the network;
• a message without a sender, and one with an inline image, fail before any request;
• against a loopback HTTP fixture, a 202 receipt succeeds, a 429 becomes RateLimited with a seven-second Retry-After, and a 503 fails over to the fallback driver;
• the facade, a tenant resolver and a suppression guard use the transport.

Add the provider’s documented request and response examples as fixtures, and validate the live transport against a sandbox account and a verified sender before enabling it in production. Tests with mocks establish local contracts, not provider interoperability.