diff --git a/crates/adapters/sqlite/src/repositories/account_purge/mod.rs b/crates/adapters/sqlite/src/repositories/account_purge/mod.rs index 389667ee..ee0fd676 100644 --- a/crates/adapters/sqlite/src/repositories/account_purge/mod.rs +++ b/crates/adapters/sqlite/src/repositories/account_purge/mod.rs @@ -126,16 +126,16 @@ 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. + // The outbox log: finished events and activities authored as, or + // mentioning, this user. Unfinished rows are never touched (they + // belong to other users' work too and finish or dead-letter on their + // own); the tombstone cleanup removes their history later. The + // `UserDeleted` row that triggered this run stays until then as well. 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)", + AND status IN ('done', 'dead_lettered')", ) .bind(&id) .execute(&mut *tx) @@ -184,40 +184,42 @@ 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)?; + // A tombstone whose id is still named by an unfinished event waits for + // the next run: that event may yet be handled, and its history row + // must be erased with the rest. let expiring: Vec = sqlx::query_scalar( - "SELECT id FROM users WHERE purged_at IS NOT NULL AND purged_at < ?", + "SELECT id FROM users u + WHERE purged_at IS NOT NULL AND purged_at < ? + AND NOT EXISTS ( + SELECT 1 FROM event_queue q + WHERE q.status NOT IN ('done', 'dead_lettered') + AND instr(q.payload, u.id) > 0)", ) .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. + // Includes 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 ( - SELECT id FROM users WHERE purged_at IS NOT NULL AND purged_at < ?)" - )) - .bind(&cutoff) - .execute(&mut *tx) - .await - .map_err(map_sqlx_error)?; - } - let removed = - sqlx::query("DELETE FROM users WHERE purged_at IS NOT NULL AND purged_at < ?") - .bind(&cutoff) + for table in RETAINED_USER_TABLES { + sqlx::query(&format!("DELETE FROM {table} WHERE user_id = ?")) + .bind(id) + .execute(&mut *tx) + .await + .map_err(map_sqlx_error)?; + } + sqlx::query("DELETE FROM users WHERE id = ?") + .bind(id) .execute(&mut *tx) .await - .map_err(map_sqlx_error)? - .rows_affected(); + .map_err(map_sqlx_error)?; + } tx.commit().await.map_err(map_sqlx_error)?; - Ok(removed) + Ok(expiring.len() as u64) } } diff --git a/crates/adapters/sqlite/tests/account_purge_test.rs b/crates/adapters/sqlite/tests/account_purge_test.rs index 59e91cda..0e5a40fa 100644 --- a/crates/adapters/sqlite/tests/account_purge_test.rs +++ b/crates/adapters/sqlite/tests/account_purge_test.rs @@ -296,6 +296,21 @@ async fn seed(pool: &SqlitePool, parts: &sqlite::SqliteWireOutput) -> Seeded { "done", format!(r#"{{"user_id":"{b_str}"}}"#), ), + ( + "FollowRequested", + "pending", + format!(r#"{{"follower_id":"{b_str}","target":"{a_str}"}}"#), + ), + ( + "ActorBlocked", + "processing", + format!(r#"{{"user_id":"{b_str}","actor":"{a_str}"}}"#), + ), + ( + "Unfollowed", + "done", + format!(r#"{{"follower_id":"{b_str}","target":"{a_str}"}}"#), + ), ] { sqlx::query("INSERT INTO event_queue (event_type, payload, status) VALUES (?, ?, ?)") .bind(event_type) @@ -385,8 +400,10 @@ async fn purge_removes_every_user_owned_row() { ), ("UserDeleted".to_string(), "processing".to_string()), ("ReviewLogged".to_string(), "done".to_string()), + ("FollowRequested".to_string(), "pending".to_string()), + ("ActorBlocked".to_string(), "processing".to_string()), ], - "the user's authored events and activities are gone; the pending Delete, the trigger and bob's event stay" + "finished rows mentioning the user are gone; pending/processing rows (incl. other users' events about her) and bob's own rows stay" ); // The other account is untouched. @@ -657,12 +674,44 @@ async fn tombstone_cleanup_removes_rows_older_than_retention_and_keeps_younger_o assert_eq!(removed, 0); assert_eq!(count_for(&pool, "users", "id", &s.user).await, 1); + // Unfinished events still name the user (the pending Delete, bob's + // pending FollowRequested, a processing row): the tombstone waits and no + // unfinished row is ever deleted. + let alice = s.user.value().to_string(); + let unfinished = || async { + sqlx::query_scalar::<_, i64>( + "SELECT COUNT(*) FROM event_queue WHERE status IN ('pending', 'processing')", + ) + .fetch_one(&pool) + .await + .unwrap() + }; + let before = unfinished().await; + assert!(before >= 3); + let removed = p + .purge_expired_tombstones(now - chrono::Duration::days(7)) + .await + .unwrap(); + assert_eq!(removed, 0, "an unfinished event still names the user"); + assert_eq!(count_for(&pool, "users", "id", &s.user).await, 1); + assert_eq!(unfinished().await, before); + + // Once they finish, the cleanup takes the tombstone and the history. + exec(&pool, "UPDATE event_queue SET status = 'done'", &[]).await; let removed = p .purge_expired_tombstones(now - chrono::Duration::days(7)) .await .unwrap(); assert_eq!(removed, 1); assert_eq!(count_for(&pool, "users", "id", &s.user).await, 0); + let left: Vec = sqlx::query_scalar("SELECT payload FROM event_queue") + .fetch_all(&pool) + .await + .unwrap(); + assert!( + left.iter().all(|p| !p.contains(&alice)), + "no history of the user remains: {left:?}" + ); assert_eq!( count_for(&pool, "ap_local_actors", "user_id", &s.user).await, 0