From 80dbdec1074f1464abb4cae7674f54034753b66e Mon Sep 17 00:00:00 2001 From: Gabriel Kaszewski Date: Mon, 26 Oct 2026 02:39:06 +0100 Subject: [PATCH] purge: skip broadcasts of deleted authors, purge outbox log, guards, 410 inbox test --- .../adapters/activitypub/src/event_handler.rs | 46 ++++++++- crates/adapters/activitypub/src/lib.rs | 1 + .../activitypub/tests/activity_ids.rs | 79 ++++++++++++++- crates/adapters/sqlite/src/lib.rs | 6 +- .../src/repositories/account_purge/mod.rs | 48 ++++++++- .../adapters/sqlite/src/repositories/mod.rs | 3 +- .../sqlite/tests/account_purge_test.rs | 98 ++++++++++++++++--- .../tests/users/purge_account_test.rs | 35 ++++++- crates/bootstrap/tests/bootstrap_test.rs | 29 ++++++ 9 files changed, 318 insertions(+), 27 deletions(-) diff --git a/crates/adapters/activitypub/src/event_handler.rs b/crates/adapters/activitypub/src/event_handler.rs index 4ac4a962..c08af08a 100644 --- a/crates/adapters/activitypub/src/event_handler.rs +++ b/crates/adapters/activitypub/src/event_handler.rs @@ -6,7 +6,7 @@ use domain::{ events::DomainEvent, ports::{ GoalQuery, LocalApContentQuery, MovieQuery, ReviewRepository, StatsRepository, - UserFederationSettingsQuery, + UserFederationSettingsQuery, UserRepository, }, value_objects::{InstanceIdentity, MovieId, ReviewId, UserId}, }; @@ -25,6 +25,7 @@ pub struct ActivityPubEventHandler { goal_repo: Arc, stats_repo: Arc, federation_settings: Arc, + user_repo: Arc, instance: InstanceIdentity, } @@ -38,6 +39,7 @@ impl ActivityPubEventHandler { goal_repo: Arc, stats_repo: Arc, federation_settings: Arc, + user_repo: Arc, instance: InstanceIdentity, ) -> Self { Self { @@ -48,6 +50,7 @@ impl ActivityPubEventHandler { goal_repo, stats_repo, federation_settings, + user_repo, instance, } } @@ -60,6 +63,15 @@ impl EventHandler for ActivityPubEventHandler { } async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { + // A deleted account must not speak again: its queued events would + // otherwise sign and send from the tombstone after the Delete. Only + // the Delete itself (sent by the deletion notifier) goes out. + if let Some(author) = self.outbound_author(event) + && self.is_tombstoned(&author).await + { + tracing::debug!(user = %author.value(), "skipping federation event of a deleted account"); + return Ok(()); + } match event { DomainEvent::ReviewLogged { review_id, user_id, .. @@ -161,6 +173,38 @@ impl EventHandler for ActivityPubEventHandler { } impl ActivityPubEventHandler { + /// The local account an outbound activity would be sent as. A queued + /// delivery counts unless it is the account's `Delete`. + fn outbound_author(&self, event: &DomainEvent) -> Option { + match event { + DomainEvent::ReviewLogged { user_id, .. } + | DomainEvent::ReviewUpdated { user_id, .. } + | DomainEvent::ReviewDeleted { user_id, .. } + | DomainEvent::UserUpdated { user_id } + | DomainEvent::WatchlistEntryAdded { user_id, .. } + | DomainEvent::WatchlistEntryRemoved { user_id, .. } + | DomainEvent::GoalCreated { user_id, .. } + | DomainEvent::GoalUpdated { user_id, .. } + | DomainEvent::GoalDeleted { user_id, .. } + | DomainEvent::UserAccountMoved { user_id, .. } => Some(user_id.clone()), + DomainEvent::FederationDeliveryRequested { + activity_json, + signing_actor_id, + .. + } if activity_json.get("type").and_then(|t| t.as_str()) != Some("Delete") => { + Some(UserId::from_uuid(*signing_actor_id)) + } + _ => None, + } + } + + async fn is_tombstoned(&self, user: &UserId) -> bool { + matches!( + self.user_repo.find_tombstoned_by_id(user).await, + Ok(Some(_)) + ) + } + async fn on_review_logged(&self, user_id: &UserId, review_id: &ReviewId) -> anyhow::Result<()> { let flags = self .federation_settings diff --git a/crates/adapters/activitypub/src/lib.rs b/crates/adapters/activitypub/src/lib.rs index 62b834a3..50ad31a9 100644 --- a/crates/adapters/activitypub/src/lib.rs +++ b/crates/adapters/activitypub/src/lib.rs @@ -179,6 +179,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result { goal_repo, stats_repo, federation_settings, + std::sync::Arc::clone(&user_repo), instance, )) as std::sync::Arc; diff --git a/crates/adapters/activitypub/tests/activity_ids.rs b/crates/adapters/activitypub/tests/activity_ids.rs index 4a52d46e..d160376f 100644 --- a/crates/adapters/activitypub/tests/activity_ids.rs +++ b/crates/adapters/activitypub/tests/activity_ids.rs @@ -16,7 +16,9 @@ use domain::{ errors::DomainError, events::DomainEvent, models::{DiaryEntry, WatchlistWithMovie}, - ports::{EventHandler, LocalApContentQuery, MovieCommand}, + ports::{ + EventHandler, FederationDeletionNotifier, LocalApContentQuery, MovieCommand, UserRepository, + }, testing::{ InMemoryStore, PanicGoalQuery, PanicReviewRepository, PanicStatsRepository, test_movie, }, @@ -71,6 +73,8 @@ struct Harness { delivered: mpsc::UnboundedReceiver, user: UserId, movie_id: MovieId, + store: Arc, + service: Arc, } async fn harness() -> Harness { @@ -119,14 +123,23 @@ async fn harness() -> Harness { let store = InMemoryStore::new(); let movie = test_movie("Heat"); store.upsert_movie(&movie, &[]).await.unwrap(); + let service = Arc::new(service); + store + .save(&domain::testing::test_user_with_role( + user.clone(), + domain::models::UserRole::Standard, + )) + .await + .unwrap(); let handler = ActivityPubEventHandler::new( - Arc::new(service), + Arc::clone(&service), Arc::new(NoContent), Arc::new(PanicReviewRepository), store.clone(), Arc::new(PanicGoalQuery), Arc::new(PanicStatsRepository), - store, + store.clone(), + store.clone(), InstanceIdentity::new("https://diary.example"), ); Harness { @@ -134,6 +147,8 @@ async fn harness() -> Harness { delivered: rx, user, movie_id: movie.id().clone(), + store, + service, } } @@ -191,3 +206,61 @@ async fn redelivered_delete_gets_a_fresh_activity_id_known_k_ap_limitation() { assert_ne!(first, second); } + +/// Queued events of a deleted account must not sign and send from its +/// tombstone; the account's own Delete still goes out. +#[tokio::test] +async fn a_deleted_account_sends_nothing_but_its_delete() { + let mut h = harness().await; + h.store.mark_deleted(&h.user, &[]).await.unwrap(); + let logged = DomainEvent::ReviewLogged { + review_id: domain::value_objects::ReviewId::generate(), + user_id: h.user.clone(), + movie_id: h.movie_id.clone(), + rating: domain::value_objects::Rating::new(4).unwrap(), + watched_at: NaiveDate::from_ymd_opt(2026, 1, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(), + }; + let added = DomainEvent::WatchlistEntryAdded { + user_id: h.user.clone(), + movie_id: h.movie_id.clone(), + movie_title: "Heat".into(), + release_year: 1995, + external_metadata_id: None, + added_at: NaiveDate::from_ymd_opt(2026, 1, 1) + .unwrap() + .and_hms_opt(0, 0, 0) + .unwrap(), + }; + let queued = DomainEvent::FederationDeliveryRequested { + inbox_url: "https://remote.example/inbox".into(), + activity_json: serde_json::json!({"type": "Create"}), + signing_actor_id: h.user.value(), + }; + + // The review repository panics if the handler gets past the guard. + for event in [&logged, &added, &queued] { + h.handler.handle(event).await.unwrap(); + } + assert!( + tokio::time::timeout(Duration::from_millis(300), h.delivered.recv()) + .await + .is_err(), + "nothing is delivered for a deleted account" + ); + + activitypub::ApDeletionNotifier::new( + Arc::clone(&h.service), + InstanceIdentity::new("https://diary.example"), + ) + .notify_deleted(&h.user) + .await + .unwrap(); + let delete = tokio::time::timeout(Duration::from_secs(5), h.delivered.recv()) + .await + .expect("the Delete is delivered") + .unwrap(); + assert_eq!(delete["type"], "Delete"); +} diff --git a/crates/adapters/sqlite/src/lib.rs b/crates/adapters/sqlite/src/lib.rs index 82bfac2e..46e00d49 100644 --- a/crates/adapters/sqlite/src/lib.rs +++ b/crates/adapters/sqlite/src/lib.rs @@ -8,15 +8,15 @@ mod shared; pub use event_queue::{DbEventQueueConfig, SqliteEventQueue}; use repositories::SqliteUserSettingsRepository; pub use repositories::{ - PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, SqliteApContentQuery, - SqliteDiaryRepository, SqliteGoalRepository, SqliteImageRefAdapter, + PAYLOAD_KEYED_USER_COLUMNS, PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, + SqliteApContentQuery, SqliteDiaryRepository, SqliteGoalRepository, SqliteImageRefAdapter, SqliteImportProfileRepository, SqliteImportSessionRepository, SqliteInsightsQuery, SqliteMovieDeduplicator, SqliteMovieProfileRepository, SqliteMovieRepository, SqlitePersonAdapter, SqliteProcessedEventLedger, SqliteProfileFieldsRepository, SqliteRefreshSessionAdapter, SqliteRemoteGoalRepository, SqliteReviewRepository, SqliteSocialRepository, SqliteStatsRepository, SqliteUserRepository, SqliteWatchEventRepository, SqliteWatchlistRepository, SqliteWebhookTokenRepository, - SqliteWrapUpRepository, SqliteWrapUpStatsQuery, URL_KEYED_USER_TABLES, + SqliteWrapUpRepository, SqliteWrapUpStatsQuery, URL_KEYED_USER_COLUMNS, create_federated_profile_query, create_image_ref, create_person_adapter, create_search_adapter, }; diff --git a/crates/adapters/sqlite/src/repositories/account_purge/mod.rs b/crates/adapters/sqlite/src/repositories/account_purge/mod.rs index ff48f97a..389667ee 100644 --- a/crates/adapters/sqlite/src/repositories/account_purge/mod.rs +++ b/crates/adapters/sqlite/src/repositories/account_purge/mod.rs @@ -34,9 +34,19 @@ pub const PURGED_USER_TABLES: &[(&str, &str)] = &[ /// the keypair is needed to sign queued federation deliveries. pub const RETAINED_USER_TABLES: &[&str] = &["ap_local_actors"]; -/// Tables with no user column whose rows hang off the user's content or actor -/// URL; the purge clears them by URL. -pub const URL_KEYED_USER_TABLES: &[&str] = &["ap_announces", "ap_followers", "ap_following"]; +/// Columns that identify the user by actor or content URL rather than by id. +/// The purge matches rows on them; the guard test asserts each still exists so +/// a rename cannot silently stop the purge from clearing them. +pub const URL_KEYED_USER_COLUMNS: &[(&str, &str)] = &[ + ("ap_announces", "object_url"), + ("ap_announces", "actor_url"), + ("ap_followers", "remote_actor_url"), + ("ap_following", "remote_actor_url"), +]; + +/// Columns holding serialized events and activities that carry the user's id +/// (the outbox log keeps finished rows, including authored activity JSON). +pub const PAYLOAD_KEYED_USER_COLUMNS: &[(&str, &str)] = &[("event_queue", "payload")]; pub struct SqliteAccountPurger { pool: SqlitePool, @@ -116,6 +126,22 @@ impl AccountPurger for SqliteAccountPurger { .await .map_err(map_sqlx_error)?; + // The outbox log: events and signed activities authored as this user. + // The `UserDeleted` row (which triggered this run) and the pending + // `Delete` deliveries must stay; both go with the tombstone. + sqlx::query( + "DELETE FROM event_queue + WHERE instr(payload, ?1) > 0 + AND event_type != 'UserDeleted' + AND NOT (event_type = 'FederationDeliveryRequested' + AND status IN ('pending', 'processing') + AND instr(payload, '\"type\":\"Delete\"') > 0)", + ) + .bind(&id) + .execute(&mut *tx) + .await + .map_err(map_sqlx_error)?; + // Other accounts' follows of this one. for table in ["ap_followers", "ap_following"] { sqlx::query(&format!("DELETE FROM {table} WHERE remote_actor_url = ?")) @@ -158,6 +184,22 @@ impl AccountPurger for SqliteAccountPurger { ) -> Result { let cutoff = cutoff.format(TIMESTAMP_FORMAT).to_string(); let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; + let expiring: Vec = sqlx::query_scalar( + "SELECT id FROM users WHERE purged_at IS NOT NULL AND purged_at < ?", + ) + .bind(&cutoff) + .fetch_all(&mut *tx) + .await + .map_err(map_sqlx_error)?; + for id in &expiring { + // What the purge had to leave: the `UserDeleted` row and the + // `Delete` deliveries. + sqlx::query("DELETE FROM event_queue WHERE instr(payload, ?) > 0") + .bind(id) + .execute(&mut *tx) + .await + .map_err(map_sqlx_error)?; + } for table in RETAINED_USER_TABLES { sqlx::query(&format!( "DELETE FROM {table} WHERE user_id IN ( diff --git a/crates/adapters/sqlite/src/repositories/mod.rs b/crates/adapters/sqlite/src/repositories/mod.rs index 5d67dadf..f51576b4 100644 --- a/crates/adapters/sqlite/src/repositories/mod.rs +++ b/crates/adapters/sqlite/src/repositories/mod.rs @@ -25,7 +25,8 @@ mod watchlist; mod wrapup; pub use account_purge::{ - PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, URL_KEYED_USER_TABLES, + PAYLOAD_KEYED_USER_COLUMNS, PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, + URL_KEYED_USER_COLUMNS, }; pub use ap_content::SqliteApContentQuery; pub use diary::SqliteDiaryRepository; diff --git a/crates/adapters/sqlite/tests/account_purge_test.rs b/crates/adapters/sqlite/tests/account_purge_test.rs index df0b72d9..59e91cda 100644 --- a/crates/adapters/sqlite/tests/account_purge_test.rs +++ b/crates/adapters/sqlite/tests/account_purge_test.rs @@ -14,9 +14,9 @@ use domain::ports::{ use domain::testing::{test_goal, test_movie, test_review, test_user, test_watched_at}; use domain::value_objects::{FollowStatus, InstanceIdentity, UserId}; use sqlite::{ - PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, SqliteProfileFieldsRepository, - SqliteSocialRepository, SqliteWatchEventRepository, SqliteWebhookTokenRepository, - URL_KEYED_USER_TABLES, + PAYLOAD_KEYED_USER_COLUMNS, PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, + SqliteProfileFieldsRepository, SqliteSocialRepository, SqliteWatchEventRepository, + SqliteWebhookTokenRepository, URL_KEYED_USER_COLUMNS, }; use sqlx::SqlitePool; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; @@ -266,6 +266,45 @@ async fn seed(pool: &SqlitePool, parts: &sqlite::SqliteWireOutput) -> Seeded { ) .await; } + for (event_type, status, payload) in [ + ( + "ReviewLogged", + "done", + format!(r#"{{"user_id":"{a_str}","comment":"secret"}}"#), + ), + ( + "FederationDeliveryRequested", + "done", + format!( + r#"{{"signing_actor_id":"{a_str}","activity_json":{{"type":"Create","content":"secret"}}}}"# + ), + ), + ( + "FederationDeliveryRequested", + "pending", + format!( + r#"{{"signing_actor_id":"{a_str}","activity_json":{{"type":"Delete","id":"x"}}}}"# + ), + ), + ( + "UserDeleted", + "processing", + format!(r#"{{"user_id":"{a_str}"}}"#), + ), + ( + "ReviewLogged", + "done", + format!(r#"{{"user_id":"{b_str}"}}"#), + ), + ] { + sqlx::query("INSERT INTO event_queue (event_type, payload, status) VALUES (?, ?, ?)") + .bind(event_type) + .bind(payload) + .bind(status) + .execute(pool) + .await + .unwrap(); + } Seeded { user: a, other: b, @@ -332,6 +371,24 @@ async fn purge_removes_every_user_owned_row() { .unwrap(); assert_eq!(announces, vec!["an3"], "only bob's announce survives"); + let queue: Vec<(String, String)> = + sqlx::query_as("SELECT event_type, status FROM event_queue ORDER BY id") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!( + queue, + vec![ + ( + "FederationDeliveryRequested".to_string(), + "pending".to_string() + ), + ("UserDeleted".to_string(), "processing".to_string()), + ("ReviewLogged".to_string(), "done".to_string()), + ], + "the user's authored events and activities are gone; the pending Delete, the trigger and bob's event stay" + ); + // The other account is untouched. for (table, column) in PURGED_USER_TABLES { assert!( @@ -384,17 +441,36 @@ async fn the_purge_table_lists_cover_every_user_referencing_column_in_the_schema "{t} has a foreign key to users but is not in the purge lists" ); } - // URL-keyed tables must exist, so a rename cannot silently drop them. - for t in URL_KEYED_USER_TABLES { - let n: i64 = count( - &pool, - &format!("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='{t}'"), - ) - .await; - assert_eq!(n, 1, "{t} is listed as URL-keyed but missing"); + // Columns the purge matches by URL or payload must still exist, so a + // rename fails here instead of silently leaving rows behind. + for (table, column) in URL_KEYED_USER_COLUMNS + .iter() + .chain(PAYLOAD_KEYED_USER_COLUMNS) + { + let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pragma_table_info(?) WHERE name = ?") + .bind(table) + .bind(column) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(n, 1, "{table}.{column} is matched by the purge but missing"); } } +/// `ap_activities` (k-ap's inbound dedup) holds only remote activity ids and a +/// timestamp: nothing in it is attributable to a local user. If it ever grows a +/// column, the purge must start clearing it. +#[tokio::test] +async fn ap_activities_holds_nothing_attributable_to_a_local_user() { + let pool = migrated_pool().await; + let columns: Vec = + sqlx::query_scalar("SELECT name FROM pragma_table_info('ap_activities') ORDER BY name") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(columns, vec!["id", "processed_at"]); +} + #[tokio::test] async fn purge_keeps_movies_and_metadata() { let pool = migrated_pool().await; diff --git a/crates/application/tests/users/purge_account_test.rs b/crates/application/tests/users/purge_account_test.rs index 5c2103c2..e29f6b6c 100644 --- a/crates/application/tests/users/purge_account_test.rs +++ b/crates/application/tests/users/purge_account_test.rs @@ -90,6 +90,7 @@ impl ObjectStorage for Storage { struct Fixture { handler: Arc, + purger: Arc, log: Log, store: Arc, user: UserId, @@ -103,21 +104,23 @@ async fn fixture(notify_fails: bool, tombstoned: bool) -> Fixture { store.mark_deleted(alice.id(), &[]).await.unwrap(); } let log: Log = Arc::default(); + let purger = Arc::new(Purger { + log: log.clone(), + purged: Mutex::new(false), + keys: vec!["avatars/a.webp".into(), "banners/a.webp".into()], + }); let handler = Arc::new(AccountPurgeHandler::new( store.clone(), Arc::new(Notifier { log: log.clone(), fail: notify_fails, }), - Arc::new(Purger { - log: log.clone(), - purged: Mutex::new(false), - keys: vec!["avatars/a.webp".into(), "banners/a.webp".into()], - }), + purger.clone(), Arc::new(Storage { log: log.clone() }), )); Fixture { handler, + purger, log, store, user: alice.id().clone(), @@ -251,3 +254,25 @@ async fn a_redelivered_event_is_skipped_by_the_ledger() { let purges = entries(&f.log).iter().filter(|e| *e == "purge").count(); assert_eq!(purges, 1); } + +/// The rows were committed but the worker died before deleting the image +/// objects: the redelivery finds the account purged, still sees the image +/// references, deletes the objects and clears the references. +#[tokio::test] +async fn a_rerun_after_a_crash_following_the_commit_deletes_the_images() { + let f = fixture(false, true).await; + *f.purger.purged.lock().unwrap() = true; + + f.handler.handle(&deleted(&f.user)).await.unwrap(); + + assert_eq!( + entries(&f.log), + [ + "purge", + "delete avatars/a.webp", + "delete banners/a.webp", + "forget" + ], + "no second federation notice, images removed and forgotten" + ); +} diff --git a/crates/bootstrap/tests/bootstrap_test.rs b/crates/bootstrap/tests/bootstrap_test.rs index 8bcd9ef5..175f672c 100644 --- a/crates/bootstrap/tests/bootstrap_test.rs +++ b/crates/bootstrap/tests/bootstrap_test.rs @@ -300,3 +300,32 @@ async fn actor_document_for_a_tombstone_is_gone() { .await; assert_eq!(status, 404, "the classic profile page stays 404"); } + +/// Remote servers posting to a deleted account's inbox are told it is gone, +/// before any signature check. +#[cfg(feature = "federation")] +#[tokio::test] +async fn posting_to_a_tombstones_inbox_is_gone() { + let (router, pool) = support::server().await; + let alice = support::register(&router, "alice", "alice@example.com").await; + let post = |path: String| { + axum::http::Request::builder() + .method("POST") + .uri(path) + .header("content-type", "application/activity+json") + .body(axum::body::Body::from("{}")) + .unwrap() + }; + let path = format!("/users/{}/inbox", alice.user_id); + let (live, _, _) = support::send(&router, post(path.clone())).await; + assert_ne!(live, 410, "a live user's inbox is not gone"); + + sqlx::query("UPDATE users SET deleted_at = '2026-01-01T00:00:00Z' WHERE id = ?") + .bind(&alice.user_id) + .execute(&pool) + .await + .unwrap(); + + let (status, _, body) = support::send(&router, post(path)).await; + assert_eq!(status, 410, "{body}"); +}