From ae8815263abe6a66b4d524af0f62b74277b9ac88 Mon Sep 17 00:00:00 2001 From: James Maes Date: Mon, 10 Aug 2026 14:57:58 -0500 Subject: [PATCH] fix(filter): make filter execution single-owner per account, default off Every daemon syncing an account decides for itself that a message is new, because "new" means "not in this daemon's store". So every daemon fires the same rules against the same message. MOVE, COPY and FLAG survive that -- ten moves converge on one outcome -- but FORWARD and CALL WEBHOOK land on a third party with no shared state to compare against. Ten daemons send ten forwards and page an on-call engineer ten times per alert. Nothing in the outbox constrains this either: pending_remote_mutations has no uniqueness of any kind. The engine cannot fix this by itself. The servers coordinate state, not actions, so there is no resource to compare-and-set against for an effect that leaves the system. Until a server-side claim is available, ownership has to be declared: filters_enabled on the account, and exactly one daemon per account turns it on. Default off, including for accounts that already exist. An account already configured on several machines would otherwise have all of them start forwarding the moment they upgrade. Off costs one setting to correct; a sent forward cannot be recalled. The salvage path deliberately does not recover the flag either, for the same reason. A non-owning daemon still syncs and stores everything -- it is a full mail client, it simply does not act. Triage is bound by the same rule since it produces the identical side effects, and refuses outright when this daemon owns no account rather than streaming a scan that reports progress and applies nothing, which would read as "no rules matched". Webhooks now carry an Idempotency-Key stable across retries of a rule/message pair, so a redelivery caused by a lost response can be discarded rather than acted on twice. It is not yet stable across daemons -- rule ids are random per daemon -- which single-owner execution is what covers today. Refs #389 --- crates/nuncio-core/src/config.rs | 18 +++ crates/nuncio-filter/src/webhook.rs | 17 +++ crates/nuncio-store/src/db.rs | 64 ++++++++- crates/nuncio-store/src/recovery.rs | 10 ++ crates/nunciod/src/grpc.rs | 63 +++++++- crates/nunciod/src/orchestrator.rs | 1 + crates/nunciod/src/send.rs | 1 + crates/nunciod/src/sync.rs | 134 ++++++++++++++++-- crates/nunciod/src/sync_dispatcher.rs | 1 + .../nunciod/tests/filters_triage_e2e_test.rs | 78 ++++++++++ .../nunciod/tests/outbox_executor_e2e_test.rs | 1 + 11 files changed, 369 insertions(+), 19 deletions(-) diff --git a/crates/nuncio-core/src/config.rs b/crates/nuncio-core/src/config.rs index 8c5fc99..245ad11 100644 --- a/crates/nuncio-core/src/config.rs +++ b/crates/nuncio-core/src/config.rs @@ -154,6 +154,21 @@ pub struct AccountConfig { pub keyring_secret_key: String, /// Background sync polling interval in seconds (minimum 10s). pub sync_interval_secs: u64, + /// Whether this daemon executes filter rules for this account. + /// + /// **Defaults to `false`**, which is the conservative answer to a problem + /// the engine cannot solve on its own. Every daemon syncing an account + /// decides independently that a message is new, so every daemon fires the + /// same rules against it. `MOVE`/`COPY`/`FLAG` survive that -- ten moves + /// converge on one outcome -- but `FORWARD` and `CALL WEBHOOK` land on a + /// third party with no shared state to compare against, so N daemons send + /// N forwards and N webhook calls. + /// + /// Enabling this on exactly one daemon per account makes filter execution + /// single-owner. It is a deliberate, documented limitation rather than a + /// silent duplication bug; a server-side claim is the eventual fix where + /// the server supports one. + pub filters_enabled: bool, /// The account's single transport. pub transport: Transport, } @@ -343,6 +358,7 @@ mod tests { email_address: "user@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-123".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: Transport::ImapSmtp(valid_imap_smtp_transport()), } } @@ -354,6 +370,7 @@ mod tests { email_address: "user@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-jmap-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: Transport::Jmap(JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), @@ -367,6 +384,7 @@ mod tests { email_address: "user@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-cal-1".to_string(), sync_interval_secs: 300, + filters_enabled: false, transport: Transport::Dav(DavTransport { collection_url: "https://caldav.example.com/calendars/user/work/".to_string(), }), diff --git a/crates/nuncio-filter/src/webhook.rs b/crates/nuncio-filter/src/webhook.rs index 123f82e..165ba0d 100644 --- a/crates/nuncio-filter/src/webhook.rs +++ b/crates/nuncio-filter/src/webhook.rs @@ -189,9 +189,26 @@ impl WebhookDispatcher { mac.update(format!("{timestamp}.{payload_str}").as_bytes()); let signature = hex::encode(mac.finalize().into_bytes()); + // Stable across every retry of this rule/message pair, so a + // receiver can discard a redelivery caused by a lost response + // rather than acting twice. It is NOT yet stable across daemons: + // `rule_id` is random per daemon, so two daemons executing the + // same rule produce different keys. Single-owner filter execution + // is what prevents that case today; a rule key derived from the + // normalized NSQL would make it dedupable directly. + let idempotency_key = { + let mut hasher = ::new(); + for field in [rule_id, message_id] { + sha2::Digest::update(&mut hasher, (field.len() as u64).to_le_bytes()); + sha2::Digest::update(&mut hasher, field.as_bytes()); + } + hex::encode(sha2::Digest::finalize(hasher)) + }; + let send_result = client .post(url) .header("Content-Type", "application/json") + .header("Idempotency-Key", &idempotency_key) .header( "X-Nuncio-Signature", format!("t={timestamp},v1={signature}"), diff --git a/crates/nuncio-store/src/db.rs b/crates/nuncio-store/src/db.rs index c22ba55..30d64b3 100644 --- a/crates/nuncio-store/src/db.rs +++ b/crates/nuncio-store/src/db.rs @@ -543,7 +543,13 @@ impl DatabaseEngine { smtp_port INTEGER, imap_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls', smtp_tls_mode TEXT NOT NULL DEFAULT 'implicit_tls', - collection_url TEXT + collection_url TEXT, + -- Defaults to 0 (off), including for accounts written before + -- the column existed. Filter execution is not safe to run on + -- more than one daemon per account, and defaulting a shared + -- account to "every daemon forwards" would be the wrong way + -- to be wrong. + filters_enabled INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS filter_rules ( @@ -750,12 +756,49 @@ impl DatabaseEngine { self.ensure_accounts_smtp_columns().await?; self.ensure_accounts_tls_mode_columns().await?; self.ensure_accounts_dav_columns().await?; + self.ensure_accounts_filters_enabled_column().await?; self.ensure_messages_identity_columns().await?; self.backfill_message_fts().await?; Ok(()) } + /// Additive migration adding the `filters_enabled` column to a + /// pre-existing `accounts` table. + /// + /// Defaults to `0` for every existing row. That is the deliberate choice: + /// an account already configured on several machines would otherwise have + /// each of them start executing its rules at once, and the observable + /// result is duplicated forwards and duplicated webhook calls to third + /// parties. Turning it back on is one setting; unsending a forward is not + /// possible. + /// + /// SQLite has no `ADD COLUMN IF NOT EXISTS`, so presence is checked first, + /// making this safe to run on every daemon startup. + async fn ensure_accounts_filters_enabled_column(&self) -> Result<(), DatabaseError> { + let existing_columns: Vec = sqlx::query("PRAGMA table_info(accounts)") + .fetch_all(&self.pool) + .await + .map_err(DatabaseError::Query)? + .iter() + .map(|row| row.get::("name")) + .collect(); + + if existing_columns.iter().any(|c| c == "filters_enabled") { + return Ok(()); + } + + sqlx::query("ALTER TABLE accounts ADD COLUMN filters_enabled INTEGER NOT NULL DEFAULT 0") + .execute(&self.pool) + .await + .map_err(DatabaseError::Query)?; + tracing::info!( + "migrated accounts table: added filters_enabled (defaulting existing accounts to off)" + ); + + Ok(()) + } + /// Additive, backfill-safe migration that adds the `remote_id` / /// `uid_validity` message-identity columns to a pre-existing `messages` /// table that predated the opaque surrogate id, and installs the @@ -1071,8 +1114,8 @@ impl DatabaseEngine { sqlx::query( r#" INSERT OR REPLACE INTO accounts - (id, name, email_address, protocol, server_host, server_port, use_tls, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + (id, name, email_address, protocol, server_host, server_port, use_tls, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url, filters_enabled) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "#, ) .bind(&config.id) @@ -1095,6 +1138,7 @@ impl DatabaseEngine { .bind(tls_mode_to_db(imap_tls_mode)) .bind(tls_mode_to_db(smtp_tls_mode)) .bind(&collection_url) + .bind(i64::from(config.filters_enabled)) .execute(&self.pool) .await .map_err(DatabaseError::Query)?; @@ -1140,9 +1184,10 @@ impl DatabaseEngine { String, String, Option, + i64, )> = sqlx::query_as( r#" - SELECT id, name, email_address, protocol, server_host, server_port, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url + SELECT id, name, email_address, protocol, server_host, server_port, keyring_secret_key, sync_interval_secs, smtp_host, smtp_port, imap_tls_mode, smtp_tls_mode, collection_url, filters_enabled FROM accounts "#, ) @@ -1167,6 +1212,7 @@ impl DatabaseEngine { imap_tls_mode, smtp_tls_mode, collection_url, + filters_enabled, )| { let protocol = serde_json::from_str(&protocol_str) .unwrap_or(nuncio_core::AccountProtocol::ImapSmtp); @@ -1213,6 +1259,7 @@ impl DatabaseEngine { email_address, keyring_secret_key, sync_interval_secs: sync_interval_secs as u64, + filters_enabled: filters_enabled != 0, transport, } }, @@ -2901,6 +2948,7 @@ mod tests { email_address: "contended@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-contended-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -3480,6 +3528,7 @@ mod tests { email_address: "work@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-test-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -3524,6 +3573,7 @@ mod tests { email_address: "mixed@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-tls-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 143, @@ -3566,6 +3616,7 @@ mod tests { email_address: "cal@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-caldav-store-1".to_string(), sync_interval_secs: 300, + filters_enabled: false, transport: nuncio_core::Transport::Dav(nuncio_core::DavTransport { collection_url: "https://dav.example.com/calendars/user/work/".to_string(), }), @@ -3598,6 +3649,7 @@ mod tests { email_address: "imap@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-mix-imap".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 143, @@ -3613,6 +3665,7 @@ mod tests { email_address: "jmap@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-mix-jmap".to_string(), sync_interval_secs: 120, + filters_enabled: false, transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), @@ -3623,6 +3676,7 @@ mod tests { email_address: "dav@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-mix-dav".to_string(), sync_interval_secs: 300, + filters_enabled: false, transport: nuncio_core::Transport::Dav(nuncio_core::DavTransport { collection_url: "https://dav.example.com/cal/".to_string(), }), @@ -3662,6 +3716,7 @@ mod tests { email_address: "del@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-del-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -3854,6 +3909,7 @@ mod tests { email_address: "post@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-post-migration".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, diff --git a/crates/nuncio-store/src/recovery.rs b/crates/nuncio-store/src/recovery.rs index 177793b..64d82a2 100644 --- a/crates/nuncio-store/src/recovery.rs +++ b/crates/nuncio-store/src/recovery.rs @@ -453,6 +453,13 @@ impl SqliteRecoveryEngine { email_address, keyring_secret_key, sync_interval_secs: sync_interval_secs as u64, + // Salvage deliberately does not recover this flag. + // The column may be among what the corruption took, + // and re-enabling filter execution on a guess would + // start forwarding mail from a daemon the user never + // chose as the owner. Off is recoverable by one + // setting; wrongly on is not recoverable at all. + filters_enabled: false, transport, } }, @@ -618,6 +625,7 @@ mod tests { email_address: "work@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-test-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -679,6 +687,7 @@ mod tests { email_address: "salvage@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-salvage-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -816,6 +825,7 @@ mod tests { email_address: "starttls@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-starttls-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 143, diff --git a/crates/nunciod/src/grpc.rs b/crates/nunciod/src/grpc.rs index 8d387bf..14425d0 100644 --- a/crates/nunciod/src/grpc.rs +++ b/crates/nunciod/src/grpc.rs @@ -490,6 +490,7 @@ fn map_account_config_from_proto( email_address: config.email_address, keyring_secret_key: config.keyring_secret_key, sync_interval_secs: nuncio_proto::time::duration_to_secs(&sync_interval), + filters_enabled: false, transport, }) } @@ -1879,12 +1880,29 @@ impl Mail for MailGrpcService { } let synced = if let Some(backend) = &self.overrides.mail_backend { + // Filter execution is owned by whichever daemon the account opts + // in (see `AccountConfig::filters_enabled`). Resolve it from the + // stored account rather than assuming: an injected backend must + // take the same path production does, or the gate would only ever + // be exercised in production. + let filters_enabled = match &account_id { + Some(id) => self + .db + .list_accounts() + .await + .map_err(|e| Status::internal(format!("sync failed: {e}")))? + .into_iter() + .find(|a| &a.id == id) + .is_some_and(|a| a.filters_enabled), + None => false, + }; crate::sync::sync_with_backend( &self.db, &self.event_bus, backend.as_ref(), &self.filter_engine, account_id, + filters_enabled, ) .await .map_err(|e| Status::internal(format!("sync failed: {e}")))? @@ -2388,7 +2406,34 @@ impl Filters for FiltersGrpcService { req.chunk_size as usize }; - tracing::info!(chunk_size, "Filters: triage rescan started"); + // Triage produces exactly the side effects a live sync produces, so it + // is bound by the same single-owner rule: only accounts this daemon + // owns filter execution for may be acted on. Resolved once here rather + // than per message. + let owning_accounts: std::collections::HashSet = self + .db + .list_accounts() + .await + .map_err(|e| Status::internal(format!("failed to read accounts for triage: {e}")))? + .into_iter() + .filter(|a| a.filters_enabled) + .map(|a| a.id) + .collect(); + + if owning_accounts.is_empty() { + // An honest refusal beats streaming a scan that reports progress + // and applies nothing, which would read as "no rules matched". + return Err(Status::failed_precondition( + "no account on this daemon has filter execution enabled; \ + enable filters_enabled on exactly one daemon per account", + )); + } + + tracing::info!( + chunk_size, + owning_accounts = owning_accounts.len(), + "Filters: triage rescan started" + ); let (tx, rx) = tokio::sync::mpsc::channel::>(16); let db = self.db.clone(); @@ -2419,8 +2464,14 @@ impl Filters for FiltersGrpcService { let is_last_page = batch.len() < chunk_size; for email in &batch { - let actions_for_email = - crate::sync::apply_filter_actions(&db, &engine, email).await; + // Messages belonging to an account this daemon does not + // own are counted as scanned but never acted on, so the + // progress total still reflects the whole store. + let actions_for_email = if owning_accounts.contains(&email.account_id) { + crate::sync::apply_filter_actions(&db, &engine, email).await + } else { + 0 + }; scanned += 1; applied += actions_for_email as u64; if actions_for_email > 0 { @@ -3644,6 +3695,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::Jmap(nuncio_core::JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), @@ -4405,6 +4457,7 @@ mod tests { email_address: "rotate@nuncio.mx".to_string(), keyring_secret_key: key.to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -4448,6 +4501,7 @@ mod tests { email_address: "rotate2@nuncio.mx".to_string(), keyring_secret_key: key.to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -5062,6 +5116,7 @@ mod tests { email_address: "sender@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-send-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -5230,6 +5285,7 @@ mod tests { email_address: "sender@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-injected-send-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -5346,6 +5402,7 @@ mod tests { email_address: "sender@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-injected-send-fail-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, diff --git a/crates/nunciod/src/orchestrator.rs b/crates/nunciod/src/orchestrator.rs index 24af9aa..9b5bf5d 100644 --- a/crates/nunciod/src/orchestrator.rs +++ b/crates/nunciod/src/orchestrator.rs @@ -154,6 +154,7 @@ mod tests { email_address: "orch@nuncio.mx".to_string(), keyring_secret_key: "nuncio/acct-orch-1".to_string(), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, diff --git a/crates/nunciod/src/send.rs b/crates/nunciod/src/send.rs index de07d15..27a24d2 100644 --- a/crates/nunciod/src/send.rs +++ b/crates/nunciod/src/send.rs @@ -292,6 +292,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, diff --git a/crates/nunciod/src/sync.rs b/crates/nunciod/src/sync.rs index 0c5036a..514158b 100644 --- a/crates/nunciod/src/sync.rs +++ b/crates/nunciod/src/sync.rs @@ -232,6 +232,7 @@ async fn fetch_and_persist( backend: &dyn MailBackend, filter_engine: &FilterEngine, account_id: Option<&str>, + filters_enabled: bool, ) -> Result { let folders = backend.sync_folders().await?; let mut synced = 0usize; @@ -263,7 +264,12 @@ async fn fetch_and_persist( let is_new = !already_present.contains(&email.id) && seen_this_pass.insert(email.id.clone()); db.save_email(&email).await?; - if is_new { + // Filter execution is single-owner per account (see + // `AccountConfig::filters_enabled`). A daemon that is not the owner + // still syncs and stores the message; it just does not act on it, + // because the side-effecting actions -- FORWARD, CALL WEBHOOK -- + // are not idempotent across daemons. + if is_new && filters_enabled { apply_filter_actions(db, filter_engine, &email).await; } synced += 1; @@ -305,6 +311,7 @@ pub async fn sync_with_backend( backend: &dyn MailBackend, filter_engine: &FilterEngine, account_id: Option, + filters_enabled: bool, ) -> Result { match &account_id { Some(id) => event_bus.process_command(CoreCommand::SyncAccount { @@ -313,8 +320,15 @@ pub async fn sync_with_backend( None => event_bus.process_command(CoreCommand::SyncAll), } - let result = - fetch_and_persist(db, event_bus, backend, filter_engine, account_id.as_deref()).await; + let result = fetch_and_persist( + db, + event_bus, + backend, + filter_engine, + account_id.as_deref(), + filters_enabled, + ) + .await; if let Err(e) = &result { event_bus.process_command(CoreCommand::ReportError { @@ -386,18 +400,20 @@ pub async fn run_account_sync( let setup = async { let config = find_account(db, account_id).await?; let password = secrets.get_secret(&config.keyring_secret_key)?; - build_mail_backend(&config, &password) + let filters_enabled = config.filters_enabled; + build_mail_backend(&config, &password).map(|backend| (backend, filters_enabled)) } .await; match setup { - Ok(backend) => { + Ok((backend, filters_enabled)) => { sync_with_backend( db, event_bus, backend.as_ref(), filter_engine, Some(account_id.to_string()), + filters_enabled, ) .await } @@ -438,6 +454,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: Transport::Jmap(nuncio_core::JmapTransport { endpoint_host: host.to_string(), }), @@ -524,6 +541,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, @@ -578,6 +596,7 @@ mod tests { &mock, &filter_engine, Some("acct-mock-1".to_string()), + true, ) .await .expect("sync succeeds"); @@ -634,7 +653,7 @@ mod tests { }); let filter_engine = empty_filter_engine(); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 0); @@ -677,6 +696,7 @@ mod tests { &mock, &filter_engine, Some("acct-fail".to_string()), + true, ) .await .expect_err("sync fails"); @@ -896,7 +916,7 @@ mod tests { }); mock.add_message(mock_email("m-urgent", "inbox", "Urgent: server down")); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 1); @@ -924,6 +944,94 @@ mod tests { ); } + #[tokio::test] + async fn a_daemon_that_does_not_own_filters_stores_the_mail_but_fires_nothing() { + // The single-owner rule. A non-owning daemon must still sync and store + // -- it is a full mail client -- while producing none of the side + // effects that would be duplicated across the fleet. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + let event_bus = EventBus::new(); + + let rule = nuncio_filter::NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse rule"); + let filter_engine = FilterEngine::new(vec![rule]).expect("compile rule"); + + let mock = MockMailBackend::new(); + mock.add_folder(Folder { + id: "inbox".to_string(), + name: "Inbox".to_string(), + total_messages: 1, + unread_messages: 1, + }); + mock.add_message(mock_email("m-urgent", "inbox", "Urgent: server down")); + + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, false) + .await + .expect("sync succeeds"); + + assert_eq!(synced, 1, "the message must still be synced and stored"); + let stored = db.get_message("m-urgent").await.expect("message persisted"); + assert!(!stored.read, "a non-owning daemon must not apply MARK READ"); + assert!( + db.list_filter_execution_logs(10) + .await + .expect("list logs") + .is_empty(), + "no rule ran, so nothing may be logged as executed" + ); + assert!( + db.list_pending_mutations(10) + .await + .expect("list mutations") + .is_empty(), + "a non-owning daemon must enqueue no remote mutation" + ); + } + + #[tokio::test] + async fn run_account_sync_takes_filter_ownership_from_the_stored_account() { + // The flag has to come from the account row, not a caller's guess -- + // otherwise the gate is only ever exercised by tests that pass it. + let (db, _dir) = DatabaseEngine::connect_ephemeral() + .await + .expect("ephemeral db"); + + let mut account = sample_imap_account("acct-owner"); + assert!( + !account.filters_enabled, + "a freshly constructed account must default to not owning filters" + ); + db.save_account(&account).await.expect("save account"); + let reloaded = db + .list_accounts() + .await + .expect("list accounts") + .into_iter() + .find(|a| a.id == "acct-owner") + .expect("account round-trips"); + assert!(!reloaded.filters_enabled); + + account.filters_enabled = true; + db.save_account(&account).await.expect("update account"); + let reloaded = db + .list_accounts() + .await + .expect("list accounts") + .into_iter() + .find(|a| a.id == "acct-owner") + .expect("account round-trips"); + assert!( + reloaded.filters_enabled, + "opting one daemon in must persist" + ); + } + #[tokio::test] async fn sync_with_backend_enqueues_outbox_mutation_for_remote_action_on_match() { let (db, _dir) = DatabaseEngine::connect_ephemeral() @@ -948,7 +1056,7 @@ mod tests { }); mock.add_message(mock_email("m-archive", "inbox", "Please Archive Me")); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 1); @@ -1002,7 +1110,7 @@ mod tests { }); mock.add_message(mock_email("m-plain", "inbox", "Just a normal update")); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 1); @@ -1062,7 +1170,7 @@ mod tests { mock.add_message(mock_email("m-repeat", "inbox", "Urgent: server down")); // First sync: the message is genuinely new, so both rules must fire. - let synced_first = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced_first = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("first sync succeeds"); assert_eq!(synced_first, 1); @@ -1094,7 +1202,7 @@ mod tests { // Second sync: the mock backend reports the SAME message again // (unchanged id), simulating a non-incremental re-sync. - let synced_second = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced_second = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("second sync succeeds"); assert_eq!(synced_second, 1); @@ -1171,7 +1279,7 @@ mod tests { mock.add_message(mock_email("m-new-1", "inbox", "Urgent: brand new")); mock.add_message(mock_email("m-new-2", "inbox", "Urgent: also new")); - let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None) + let synced = sync_with_backend(&db, &event_bus, &mock, &filter_engine, None, true) .await .expect("sync succeeds"); assert_eq!(synced, 3, "every fetched message is still persisted"); @@ -1235,6 +1343,7 @@ mod tests { &mock, &filter_engine, Some("acct-inc-1".to_string()), + true, ) .await .expect("first sync succeeds"); @@ -1256,6 +1365,7 @@ mod tests { &mock, &filter_engine, Some("acct-inc-1".to_string()), + true, ) .await .expect("second sync succeeds"); diff --git a/crates/nunciod/src/sync_dispatcher.rs b/crates/nunciod/src/sync_dispatcher.rs index 4526da0..52bdd6c 100644 --- a/crates/nunciod/src/sync_dispatcher.rs +++ b/crates/nunciod/src/sync_dispatcher.rs @@ -806,6 +806,7 @@ mod tests { email_address: format!("{id}@nuncio.mx"), keyring_secret_key: format!("nuncio/{id}"), sync_interval_secs: 60, + filters_enabled: false, transport: Transport::Jmap(JmapTransport { endpoint_host: "jmap.nuncio.mx".to_string(), }), diff --git a/crates/nunciod/tests/filters_triage_e2e_test.rs b/crates/nunciod/tests/filters_triage_e2e_test.rs index 60689d3..48d1038 100644 --- a/crates/nunciod/tests/filters_triage_e2e_test.rs +++ b/crates/nunciod/tests/filters_triage_e2e_test.rs @@ -56,6 +56,27 @@ fn sample_email(id: &str, subject: &str) -> Email { } } +/// The account the seeded messages belong to, opted in as this daemon's +/// filter-execution owner. +fn owning_account() -> nuncio_core::AccountConfig { + nuncio_core::AccountConfig { + id: ACCOUNT_ID.to_string(), + name: "Triage E2E".to_string(), + email_address: "triage-e2e@nuncio.mx".to_string(), + keyring_secret_key: format!("nuncio/{ACCOUNT_ID}"), + sync_interval_secs: 60, + filters_enabled: true, + transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { + imap_host: "imap.nuncio.mx".to_string(), + imap_port: 993, + imap_tls_mode: nuncio_core::TlsMode::ImplicitTls, + smtp_host: "smtp.nuncio.mx".to_string(), + smtp_port: 465, + smtp_tls_mode: nuncio_core::TlsMode::ImplicitTls, + }), + } +} + /// Boots the real daemon gRPC server (bearer-token authenticated) on an /// ephemeral loopback port, backed by a real temp file-backed /// `DatabaseEngine` and the given live `FilterEngine`. Returns the server @@ -79,6 +100,13 @@ async fn boot_daemon( .expect("open temp file-backed database"), ); + // Triage only acts on accounts this daemon owns filter execution for, so + // the account must exist and be opted in -- otherwise the RPC honestly + // refuses rather than scanning and applying nothing. + db.save_account(&owning_account()) + .await + .expect("seed the triage account"); + let token_bytes = secrets .get_or_create_key_bytes(GRPC_TOKEN_ACCOUNT, 32) .expect("mint gRPC bearer token from mock vault"); @@ -267,3 +295,53 @@ async fn triage_streams_multiple_chunks_for_a_store_bigger_than_one_chunk() { prev_scanned = update.scanned_count; } } + +/// A daemon that owns filter execution for no account must refuse `Triage` +/// outright. Streaming a scan that reports progress and applies nothing would +/// be indistinguishable, from the caller's side, from "no rules matched" -- +/// and the whole point of single-owner execution is that the non-owner stays +/// visibly inert rather than quietly so. +#[tokio::test] +async fn triage_refuses_when_this_daemon_owns_no_account() { + let secrets = Arc::new(SecretManager::mock()); + let rule = NsqlParser::parse_rule( + "Urgent Auto-Read", + 1, + "WHERE subject CONTAINS 'Urgent' ACTION MARK READ", + ) + .expect("parse rule"); + let filter_engine = Arc::new(FilterEngine::new(vec![rule]).expect("compile rule")); + + let (addr, token, db, _dir) = boot_daemon(secrets, filter_engine).await; + + // Take ownership away: the account exists, but this daemon is not its + // filter owner. + let mut account = owning_account(); + account.filters_enabled = false; + db.save_account(&account) + .await + .expect("opt the account out"); + + db.save_email(&sample_email("m-1", "Urgent: server down")) + .await + .expect("seed a matching message"); + + let mut client = FiltersClient::connect(format!("http://{addr}")) + .await + .expect("connect to the daemon"); + let status = client + .triage(authed_request(&token, TriageRequest::default())) + .await + .expect_err("triage must refuse without an owning account"); + + assert_eq!(status.code(), tonic::Code::FailedPrecondition); + assert!( + status.message().contains("filter execution"), + "the refusal must say why, got: {}", + status.message() + ); + + // And it must genuinely have applied nothing. + let stored = db.get_message("m-1").await.expect("message persisted"); + assert!(!stored.read, "a refused triage must not apply any action"); +} diff --git a/crates/nunciod/tests/outbox_executor_e2e_test.rs b/crates/nunciod/tests/outbox_executor_e2e_test.rs index 5fccdcd..7388dd6 100644 --- a/crates/nunciod/tests/outbox_executor_e2e_test.rs +++ b/crates/nunciod/tests/outbox_executor_e2e_test.rs @@ -101,6 +101,7 @@ fn sample_account() -> nuncio_core::AccountConfig { email_address: "owner@nuncio.mx".to_string(), keyring_secret_key: format!("nuncio/{ACCOUNT_ID}"), sync_interval_secs: 60, + filters_enabled: false, transport: nuncio_core::Transport::ImapSmtp(nuncio_core::ImapSmtpTransport { imap_host: "imap.nuncio.mx".to_string(), imap_port: 993, -- 2.54.0