fix(watcher): gate delete side effects with a SETNX tombstone that read-through cannot resurrect

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<bool> {
+    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<String> = 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(())
+}

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;

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;

nexus-watcher/src/events/handlers/follow.rs

@@ -1,14 +1,14 @@
 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;
 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(())
 }
 

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(())
 }

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).
 ///