diff --git a/nexus-common/src/db/kv/index/guards.rs b/nexus-common/src/db/kv/index/guards.rs new file mode 100644 index 000000000..78940c39d --- /dev/null +++ b/nexus-common/src/db/kv/index/guards.rs @@ -0,0 +1,60 @@ +use crate::db::get_redis_conn; +use crate::db::kv::RedisResult; + +/// Attempts to acquire a guard key using `SET key 1 NX EX ttl`. +/// +/// The key is only written if it does not already exist (SETNX semantics), so +/// concurrent or retried callers observe exactly one successful acquisition +/// until the key is released or its TTL expires. Unlike regular index entries, +/// a guard key is never written by read-through cache population, which makes +/// it a reliable "already ran" marker for non-idempotent side effects. +/// +/// # Arguments +/// +/// * `key` - The full Redis key for the guard (no prefix is added). +/// * `ttl_secs` - The TTL (in seconds) after which the guard expires on its own. +/// +/// # Returns +/// +/// Returns `Ok(true)` if the guard was acquired (the key did not exist), +/// `Ok(false)` if the guard is already held. +/// +/// # Errors +/// +/// Returns an error if the operation fails. +pub async fn try_acquire(key: &str, ttl_secs: u64) -> RedisResult { + let mut redis_conn = get_redis_conn().await?; + + // SET with NX returns "OK" when the key was set, nil when it already exists. + let outcome: Option = redis::cmd("SET") + .arg(key) + .arg(1) + .arg("NX") + .arg("EX") + .arg(ttl_secs) + .query_async(&mut redis_conn) + .await?; + + Ok(outcome.is_some()) +} + +/// Releases a guard key previously acquired with [`try_acquire`]. +/// +/// Deleting a non-existent key is a no-op, so releasing twice is safe. +/// +/// # Arguments +/// +/// * `key` - The full Redis key for the guard (no prefix is added). +/// +/// # Errors +/// +/// Returns an error if the operation fails. +pub async fn release(key: &str) -> RedisResult<()> { + let mut redis_conn = get_redis_conn().await?; + + let _: () = redis::cmd("DEL") + .arg(key) + .query_async(&mut redis_conn) + .await?; + Ok(()) +} diff --git a/nexus-common/src/db/kv/index/mod.rs b/nexus-common/src/db/kv/index/mod.rs index 0d64d28ab..e97d69378 100644 --- a/nexus-common/src/db/kv/index/mod.rs +++ b/nexus-common/src/db/kv/index/mod.rs @@ -1,4 +1,5 @@ /// Module for redis Indexing operations split into modules by Redis types +pub mod guards; pub mod json; pub mod lists; pub mod search; diff --git a/nexus-common/src/db/kv/mod.rs b/nexus-common/src/db/kv/mod.rs index a61222e3c..518f86b63 100644 --- a/nexus-common/src/db/kv/mod.rs +++ b/nexus-common/src/db/kv/mod.rs @@ -6,6 +6,7 @@ mod traits; pub use error::{RedisError, RedisResult}; pub use flush::clear_redis; +pub use index::guards; pub use index::json::JsonAction; pub(crate) use index::search; pub use index::sets; diff --git a/nexus-watcher/src/events/handlers/follow.rs b/nexus-watcher/src/events/handlers/follow.rs index ffe5b0a05..3d5af4009 100644 --- a/nexus-watcher/src/events/handlers/follow.rs +++ b/nexus-watcher/src/events/handlers/follow.rs @@ -1,6 +1,6 @@ use crate::events::EventProcessorError; -use nexus_common::db::kv::JsonAction; +use nexus_common::db::kv::{guards, JsonAction}; use nexus_common::db::OperationOutcome; use nexus_common::models::follow::{Followers, Following, Friends, UserFollows}; use nexus_common::models::notification::Notification; @@ -8,7 +8,7 @@ use nexus_common::models::user::{UserCounts, UserIngestor}; use pubky_app_specs::PubkyId; use tracing::debug; -use super::utils::fail_on_blacklisted_hs; +use super::utils::{fail_on_blacklisted_hs, follow_deletion_guard_key, DELETION_GUARD_TTL_SECS}; #[tracing::instrument(name = "follow.put", skip_all, fields(follower_id = %follower_id, followee_id = %followee_id))] pub async fn sync_put( @@ -80,6 +80,14 @@ pub async fn sync_put( } }; + // A new PUT for this edge proves any earlier delete cycle is finished, so + // its tombstone is stale; drop it (best-effort) so it cannot suppress the + // side effects of a later legitimate delete of the re-created follow. + let deletion_guard_key = follow_deletion_guard_key(&follower_id, &followee_id); + if let Err(e) = guards::release(&deletion_guard_key).await { + tracing::warn!("failed to release stale deletion guard {deletion_guard_key}: {e}"); + } + Ok(()) } @@ -101,20 +109,48 @@ pub async fn sync_del( // On retry (Redis already cleaned, graph edge still present), skip non-idempotent ops. let still_indexed = Followers::check_in_index(&followee_id, &follower_id).await?; + // SETNX tombstone: `still_indexed` alone is not retry-safe, read-through can + // resurrect the follow sets (see `follow_deletion_guard_key`). Acquired after + // all reads and only when the gate is present, so read failures stay + // retryable and a no-op delete leaves no tombstone behind. + let deletion_guard_key = follow_deletion_guard_key(&follower_id, &followee_id); + let first_attempt = if still_indexed { + guards::try_acquire(&deletion_guard_key, DELETION_GUARD_TTL_SECS).await? + } else { + false + }; + + // Non-idempotent side effects run only when both gates agree this is the + // first attempt. + let run_side_effects = still_indexed && first_attempt; + let followers = Followers(vec![follower_id.to_string()]); let following = Following(vec![followee_id.to_string()]); - // Redis cleanup first — SREM is idempotent + // Redis cleanup first, SREM is idempotent let indexing_results = nexus_common::traced_join!( tracing::info_span!("index.delete"); followers.del_from_index(&followee_id), following.del_from_index(&follower_id) ); - indexing_results.0?; - indexing_results.1?; + // A guard acquired by THIS attempt must not survive a failure here: no side + // effects have run yet, and a stranded tombstone would make the retry skip + // them. A guard held by a previous attempt is kept, that attempt may + // already have run them; once the counts update below starts, partial + // completion must not re-run either. + if let Err(e) = indexing_results.0.and(indexing_results.1) { + if first_attempt { + if let Err(release_err) = guards::release(&deletion_guard_key).await { + tracing::warn!( + "failed to release deletion guard {deletion_guard_key}: {release_err}" + ); + } + } + return Err(e.into()); + } // Only after indexes are confirmed clean: non-idempotent ops - if still_indexed { + if run_side_effects { update_follow_counts( &follower_id, &followee_id, @@ -128,6 +164,13 @@ pub async fn sync_del( // Graph deletion LAST — on retry, we re-enter here with indexes already clean. // MissingDependency means the resource is already gone — deletion is complete. Followers::del_from_graph(&follower_id, &followee_id).await?; + + // The delete completed: drop the tombstone. Best-effort: the TTL backstops + // a leaked key, and failing the event after a successful graph delete would + // only force a useless retry cycle. + if let Err(e) = guards::release(&deletion_guard_key).await { + tracing::warn!("failed to release deletion guard {deletion_guard_key}: {e}"); + } Ok(()) } diff --git a/nexus-watcher/src/events/handlers/post.rs b/nexus-watcher/src/events/handlers/post.rs index 48d2d5f06..896f57a8e 100644 --- a/nexus-watcher/src/events/handlers/post.rs +++ b/nexus-watcher/src/events/handlers/post.rs @@ -1,5 +1,6 @@ use crate::events::EventProcessorError; +use nexus_common::db::kv::guards; use nexus_common::db::queries::get::post_is_safe_to_delete; use nexus_common::db::{exec_single_row, execute_graph_operation, OperationOutcome}; use nexus_common::db::{queries, RedisOps}; @@ -14,7 +15,10 @@ use pubky_app_specs::{ }; use tracing::{debug, Instrument}; -use super::utils::{fail_on_blacklisted_hs, post_is_collection, post_relationships_is_reply}; +use super::utils::{ + fail_on_blacklisted_hs, post_deletion_guard_key, post_is_collection, + post_relationships_is_reply, DELETION_GUARD_TTL_SECS, +}; #[tracing::instrument(name = "post.put", skip_all, fields(user_id = %author_id, post_id = %post_id))] pub async fn sync_put( @@ -277,6 +281,14 @@ pub async fn sync_put( indexing_results.0?; indexing_results.1?; + // A new PUT for this key proves any earlier delete cycle is finished, so + // its tombstone is stale; drop it (best-effort) so it cannot suppress the + // side effects of a later legitimate delete of the re-created post. + let deletion_guard_key = post_deletion_guard_key(&author_id, &post_id); + if let Err(e) = guards::release(&deletion_guard_key).await { + tracing::warn!("failed to release stale deletion guard {deletion_guard_key}: {e}"); + } + Ok(()) } @@ -529,11 +541,40 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr // failed lookup strand the decrements on retry (gate gone, post_in_index false). let is_collection = post_in_index && post_is_collection(&author_id, &post_id).await?; + // SETNX tombstone: the index gate alone is not retry-safe, read-through can + // resurrect it (see `post_deletion_guard_key`). Acquired after all reads and + // only when the gate is present, so read failures stay retryable and a + // no-op delete leaves no tombstone behind. + let deletion_guard_key = post_deletion_guard_key(&author_id, &post_id); + let first_attempt = if post_in_index { + guards::try_acquire(&deletion_guard_key, DELETION_GUARD_TTL_SECS).await? + } else { + false + }; + + // Non-idempotent side effects (counter decrements, engagement score updates, + // notifications) run only when both gates agree this is the first attempt. + let run_side_effects = post_in_index && first_attempt; + // 2. Atomically commit the cleanup decision: remove the gate as the very // first mutation. Subsequent retries will observe `post_in_index = false` // and skip non-idempotent ops (counters, scores, notifications). if post_in_index { - PostRelationships::delete(&author_id, &post_id).await?; + // A guard acquired by THIS attempt must not survive a failure here: no + // side effects have run yet, and a stranded tombstone would make the + // retry skip them. A guard held by a previous attempt is kept, that + // attempt may already have run them; once the joins below start, + // partial completion must not re-run either. + if let Err(e) = PostRelationships::delete(&author_id, &post_id).await { + if first_attempt { + if let Err(release_err) = guards::release(&deletion_guard_key).await { + tracing::warn!( + "failed to release deletion guard {deletion_guard_key}: {release_err}" + ); + } + } + return Err(e.into()); + } } // 3. On retry (gate already gone), fall back to the graph for parent info @@ -558,16 +599,16 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr PostCounts::delete(&author_id, &post_id, !is_reply), // Guarded: skip on retry to avoid double-decrement. async { - if post_in_index { + if run_side_effects { UserCounts::decrement(&author_id, "posts", None).await?; } Ok::<(), EventProcessorError>(()) }, async { // reply XOR collection; never both fire. - if post_in_index && is_reply { + if run_side_effects && is_reply { UserCounts::decrement(&author_id, "replies", None).await?; - } else if is_collection { + } else if run_side_effects && is_collection { UserCounts::decrement(&author_id, "collections", None).await?; }; Ok::<(), EventProcessorError>(()) @@ -605,7 +646,7 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr let indexing_results = nexus_common::traced_join!( tracing::info_span!("index.delete", phase = "reply_parent"); async { - if post_in_index { + if run_side_effects { PostCounts::decrement_index_field(&parent_post_key_parts, "replies", None).await?; } Ok::<(), EventProcessorError>(()) @@ -614,7 +655,7 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr // Symmetric DEL gate: ZINCRBY -1 would create the member // if absent, leaking a reply parent into POST_TOTAL_ENGAGEMENT // with a negative score. - if post_in_index + if run_side_effects && !post_relationships_is_reply(&parent_user_id, &parent_post_id).await? { PostStream::decrement_score_index_sorted_set( @@ -628,7 +669,7 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr // Notification: "A reply to your post was deleted" — guarded to // prevent duplicate notifications on retry. async { - if post_in_index { + if run_side_effects { Notification::post_children_changed( &author_id, &replied_uri_str, @@ -667,7 +708,7 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr let indexing_results = nexus_common::traced_join!( tracing::info_span!("index.delete", phase = "repost_parent"); async { - if post_in_index { + if run_side_effects { PostCounts::decrement_index_field(parent_post_key_parts, "reposts", None).await?; } Ok::<(), EventProcessorError>(()) @@ -676,7 +717,7 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr // Symmetric DEL gate: ZINCRBY -1 would create the member // if absent, leaking a reply parent into POST_TOTAL_ENGAGEMENT // with a negative score. - if post_in_index + if run_side_effects && !post_relationships_is_reply(&reposted_uri.user_id, &parent_post_id).await? { PostStream::decrement_score_index_sorted_set( @@ -689,7 +730,7 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr }, // Notification: "A repost of your post was deleted" — guarded. async { - if post_in_index { + if run_side_effects { Notification::post_children_changed( &author_id, &reposted_uri_str, @@ -722,5 +763,12 @@ pub async fn sync_del(author_id: PubkyId, post_id: String) -> Result<(), EventPr .instrument(tracing::info_span!("graph.delete", phase = "post_graph")) .await?; + // The delete completed: drop the tombstone. Best-effort: the TTL backstops + // a leaked key, and failing the event after a successful graph delete would + // only force a useless retry cycle. + if let Err(e) = guards::release(&deletion_guard_key).await { + tracing::warn!("failed to release deletion guard {deletion_guard_key}: {e}"); + } + Ok(()) } diff --git a/nexus-watcher/src/events/handlers/utils.rs b/nexus-watcher/src/events/handlers/utils.rs index 9b7ee416d..10f93476f 100644 --- a/nexus-watcher/src/events/handlers/utils.rs +++ b/nexus-watcher/src/events/handlers/utils.rs @@ -5,6 +5,34 @@ use nexus_common::models::{ }; use pubky_app_specs::PubkyAppPostKind; +/// TTL (in seconds) for delete-tombstone guard keys: 6 hours. +/// +/// The tombstone must outlive the event retry backoff window so that a retried +/// delete still observes the guard acquired by the first attempt. If the delete +/// dead-letters and never completes, the key expires on its own and leaves no +/// permanent garbage behind. This value must exceed the worst-case retry window +/// derived from the `EventRetryConfig` `max_retries`/`max_backoff_secs` settings +/// (roughly 2.5 hours with defaults) and should be revisited if those are raised. +pub const DELETION_GUARD_TTL_SECS: u64 = 21600; + +/// Redis key of the SETNX tombstone marking an in-flight post deletion. +/// +/// Unlike the `PostRelationships` index gate, this key cannot be recreated by +/// read-through cache population, so it survives the retry window even if a +/// concurrent read resurrects the index entry from the still-present graph node. +pub fn post_deletion_guard_key(author_id: &str, post_id: &str) -> String { + format!("Deleting:Post:{author_id}:{post_id}") +} + +/// Redis key of the SETNX tombstone marking an in-flight follow deletion. +/// +/// Unlike the `Followers` index gate, this key cannot be recreated by +/// read-through cache population, so it survives the retry window even if a +/// concurrent read resurrects the follow sets from the still-present graph edge. +pub fn follow_deletion_guard_key(follower_id: &str, followee_id: &str) -> String { + format!("Deleting:Follow:{follower_id}:{followee_id}") +} + /// Classifies the outcome of a best-effort user ingestion attempted while /// handling an [`OperationOutcome::MissingDependency`](nexus_common::db::OperationOutcome::MissingDependency). /// diff --git a/nexus-watcher/tests/event_processor/follows/del_idempotent.rs b/nexus-watcher/tests/event_processor/follows/del_idempotent.rs index 91b942ad1..51a595065 100644 --- a/nexus-watcher/tests/event_processor/follows/del_idempotent.rs +++ b/nexus-watcher/tests/event_processor/follows/del_idempotent.rs @@ -3,7 +3,7 @@ use crate::event_processor::users::utils::find_user_counts; use crate::event_processor::utils::watcher::WatcherTest; use anyhow::Result; use nexus_common::{ - db::kv::JsonAction, + db::kv::{guards, JsonAction}, db::RedisOps, models::{ follow::{Followers, Following, UserFollows}, @@ -11,6 +11,7 @@ use nexus_common::{ }, }; use nexus_watcher::events::handlers::follow; +use nexus_watcher::events::handlers::utils::{follow_deletion_guard_key, DELETION_GUARD_TTL_SECS}; use pubky::Keypair; use pubky_app_specs::{PubkyAppUser, PubkyId}; @@ -211,6 +212,152 @@ async fn test_follow_del_recovers_stale_indexes() -> Result<()> { Ok(()) } +/// Tombstone gate vs read-through resurrection, follow flavor: attempt 1 +/// of deleting the F->X follow completed every Redis step (both follow sets +/// SREMed, both counters decremented once) but failed at the final graph +/// delete. Between attempts, `Followers::get_by_id(X)` read-through +/// re-populates `Followers:{X}` from the still-present graph edge, which is +/// the exact set the retry uses as its `still_indexed` gate. With the +/// tombstone held by attempt 1, the retry must skip the non-idempotent +/// decrements: F's `following` count is decremented exactly once across both +/// attempts. +/// +/// F follows a SECOND user (Y) so `following` starts at 2: the counter +/// decrement is floored at 0, so starting from 1 a double decrement would be +/// invisible (0 stays 0) and the headline assertion could never fail on the +/// unguarded code path. +#[tokio_shared_rt::test(shared)] +async fn test_follow_del_retry_skips_side_effects_after_readthrough_resurrection() -> Result<()> { + let mut test = WatcherTest::setup(None).await?; + + // Follower F + let f_kp = Keypair::random(); + let f_user = PubkyAppUser { + bio: Some( + "test_follow_del_retry_skips_side_effects_after_readthrough_resurrection".to_string(), + ), + image: None, + links: None, + name: "Watcher:DelResurrection:Follower".to_string(), + status: None, + }; + let f_id = test.create_user(&f_kp, &f_user).await?; + + // Followee X (the follow under deletion; F is X's only follower) + let x_kp = Keypair::random(); + let x_user = PubkyAppUser { + bio: Some( + "test_follow_del_retry_skips_side_effects_after_readthrough_resurrection".to_string(), + ), + image: None, + links: None, + name: "Watcher:DelResurrection:FolloweeX".to_string(), + status: None, + }; + let x_id = test.create_user(&x_kp, &x_user).await?; + + // Followee Y (keeps F's `following` count above the floor) + let y_kp = Keypair::random(); + let y_user = PubkyAppUser { + bio: Some( + "test_follow_del_retry_skips_side_effects_after_readthrough_resurrection".to_string(), + ), + image: None, + links: None, + name: "Watcher:DelResurrection:FolloweeY".to_string(), + status: None, + }; + let y_id = test.create_user(&y_kp, &y_user).await?; + + // F follows both X and Y. + test.create_follow(&f_kp, &x_id).await?; + test.create_follow(&f_kp, &y_id).await?; + + // Sanity: F is following 2 users, X has exactly 1 follower (F). + assert_eq!(find_user_counts(&f_id).await.following, 2); + assert_eq!(find_user_counts(&x_id).await.followers, 1); + + // Simulate attempt 1 of the tombstone-gated sync_del for F->X: the guard + // was acquired, both follow sets were SREMed, both counters were + // decremented once (F and X are not friends, so no friends update), and + // only the final graph delete failed. + let guard_key = follow_deletion_guard_key(&f_id, &x_id); + assert!( + guards::try_acquire(&guard_key, DELETION_GUARD_TTL_SECS).await?, + "fresh deletion guard should be acquirable" + ); + Followers(vec![f_id.to_string()]) + .del_from_index(&x_id) + .await?; + Following(vec![x_id.to_string()]) + .del_from_index(&f_id) + .await?; + UserCounts::update_index_field(&f_id, "following", JsonAction::Decrement(1)).await?; + UserCounts::update(&x_id, "followers", JsonAction::Decrement(1), None).await?; + + // Graph edge still present: the graph delete runs LAST and it failed. + assert!( + find_follow_relationship(&f_id, &x_id).await?, + "graph edge should survive the failed first attempt" + ); + + // Between attempts, a read-through resurrects the gate: `Followers:{X}` is + // re-populated from the still-present graph edge. + assert!( + Followers::get_by_id(&x_id, None, None).await?.is_some(), + "read-through should find X's followers in the graph" + ); + assert!( + Followers::check_in_index(&x_id, &f_id).await?, + "read-through must have resurrected the still_indexed gate" + ); + + // Retry. Without the tombstone this would observe the resurrected gate and + // decrement `following`/`followers` a second time. + follow::sync_del( + PubkyId::from(f_kp.public_key()), + PubkyId::from(x_kp.public_key()), + ) + .await?; + + // Graph edge gone; F's `following` decremented exactly once across both + // attempts (2 -> 1). An unguarded retry would have decremented again (1 -> 0). + assert!( + !find_follow_relationship(&f_id, &x_id).await?, + "graph edge should be gone after the retry" + ); + assert_eq!( + find_user_counts(&f_id).await.following, + 1, + "following count must not be double-decremented after read-through resurrection" + ); + assert_eq!( + find_user_counts(&x_id).await.followers, + 0, + "X should have 0 followers after the delete" + ); + + // The F->Y follow is untouched. + assert!( + Following::check_in_index(&f_id, &y_id).await?, + "F->Y follow should be unaffected" + ); + + // Successful completion released the tombstone: it must be acquirable again. + assert!( + guards::try_acquire(&guard_key, DELETION_GUARD_TTL_SECS).await?, + "deletion guard should have been released after the successful delete" + ); + guards::release(&guard_key).await?; + + // Cleanup + test.cleanup_user(&f_kp).await?; + test.cleanup_user(&x_kp).await?; + test.cleanup_user(&y_kp).await?; + + Ok(()) +} + /// Test that retrying an unfollow between friends does not double-decrement /// the friends counter for either user. #[tokio_shared_rt::test(shared)] diff --git a/nexus-watcher/tests/event_processor/posts/idempotent/del.rs b/nexus-watcher/tests/event_processor/posts/idempotent/del.rs index ec98bab00..6e8b887f0 100644 --- a/nexus-watcher/tests/event_processor/posts/idempotent/del.rs +++ b/nexus-watcher/tests/event_processor/posts/idempotent/del.rs @@ -5,9 +5,12 @@ use crate::event_processor::posts::utils::{ use crate::event_processor::users::utils::find_user_counts; use crate::event_processor::utils::watcher::WatcherTest; use anyhow::Result; +use nexus_common::db::kv::guards; +use nexus_common::models::post::PostRelationships; use nexus_common::utils::test_utils::default_ingestor_tests; use nexus_watcher::errors::EventProcessorError; use nexus_watcher::events::handlers; +use nexus_watcher::events::handlers::utils::{post_deletion_guard_key, DELETION_GUARD_TTL_SECS}; use pubky::Keypair; use pubky_app_specs::post_uri_builder; @@ -68,6 +71,104 @@ async fn test_post_del_recovers_after_partial_redis_cleanup() -> Result<()> { Ok(()) } +/// Tombstone gate vs read-through resurrection: a previous attempt +/// completed every Redis cleanup step (gate consumed, counters decremented) +/// but failed at the final graph delete. Between attempts, a public read +/// (`PostRelationships::get_by_id`, reachable from GET /v0/post) re-populates +/// the index gate from the still-present graph node. Without the SETNX +/// tombstone, the retry would see `post_in_index = true` again and re-run the +/// non-idempotent decrements. With the tombstone held by the first attempt, +/// the retry must skip them: the author's `posts` count is decremented +/// exactly once across both attempts. +/// +/// The author owns a SECOND post so `posts` starts at 2: the counter decrement +/// is floored at 0 by the Lua script, so starting from 1 a double decrement +/// would be invisible (0 stays 0) and the headline assertion could never fail +/// on the unguarded code path. +#[tokio_shared_rt::test(shared)] +async fn test_post_del_retry_skips_side_effects_after_readthrough_resurrection() -> Result<()> { + let mut test = WatcherTest::setup(None).await?; + + let user_kp = Keypair::random(); + let user_id = test + .create_user( + &user_kp, + &test_user( + "Watcher:Post:DelResurrection:User", + "test_post_del_retry_skips_side_effects_after_readthrough_resurrection", + ), + ) + .await?; + + let post = short_post("Watcher:Post:DelResurrection:Post"); + let (post_id, _post_path) = test.create_post(&user_kp, &post).await?; + + // Second post: keeps the author's `posts` count above the Lua floor across + // the scenario so a double decrement is observable (2 -> 1 -> 0, not 1 -> 0 -> 0). + let second_post = short_post("Watcher:Post:DelResurrection:SecondPost"); + let (_second_post_id, _second_post_path) = test.create_post(&user_kp, &second_post).await?; + + // Sanity: fully indexed. + assert!(find_post_details(&user_id, &post_id).await.is_ok()); + assert_eq!(find_user_counts(&user_id).await.posts, 2); + + // Simulate attempt 1 of the tombstone-gated sync_del against the FIRST + // post: the guard was acquired, every Redis cleanup step ran (gate + // consumed, UserCounts decremented), and only the final graph delete + // failed. + let guard_key = post_deletion_guard_key(&user_id, &post_id); + assert!( + guards::try_acquire(&guard_key, DELETION_GUARD_TTL_SECS).await?, + "fresh deletion guard should be acquirable" + ); + simulate_partial_del_cleanup_root(&user_id, &post_id).await?; + + // Between attempts, a read-through resurrects the index gate: the graph + // node is still present (graph delete runs LAST), so `get_by_id` misses + // the index, falls back to the graph, and re-populates the exact key the + // retry uses as its `post_in_index` gate. + assert!( + PostRelationships::get_by_id(&user_id, &post_id) + .await? + .is_some(), + "read-through should find the post in the graph" + ); + assert!( + PostRelationships::get_from_index(&user_id, &post_id) + .await? + .is_some(), + "read-through must have resurrected the index gate" + ); + + // Retry through the public entry point. Without the tombstone this would + // observe the resurrected gate and decrement `posts` a second time. + handlers::post::del( + pubky_id(&user_id)?, + post_id.clone(), + &default_ingestor_tests(), + ) + .await?; + + // Graph node gone; `posts` decremented exactly once across both attempts + // (2 -> 1). An unguarded retry would have decremented again (1 -> 0). + assert!(find_post_details(&user_id, &post_id).await.is_err()); + assert_eq!( + find_user_counts(&user_id).await.posts, + 1, + "posts count must not be double-decremented after read-through resurrection" + ); + + // Successful completion released the tombstone: it must be acquirable again. + assert!( + guards::try_acquire(&guard_key, DELETION_GUARD_TTL_SECS).await?, + "deletion guard should have been released after the successful delete" + ); + guards::release(&guard_key).await?; + + test.cleanup_user(&user_kp).await?; + Ok(()) +} + /// Replay sync_del on a fully deleted post: post::del should report /// MissingDependency (mapped to SkipIndexing) without corrupting state. #[tokio_shared_rt::test(shared)]