feat: attribute deletion gate winners

src/nostr/lifecycle/deletion/service.rs

@@ -1,9 +1,11 @@
 use anyhow::Result;
 use nostr_relay_builder::prelude::{
-    nip62, Alphabet, Event, Filter, Kind, PublicKey, RelayUrl, SingleLetterTag, WritePolicyResult,
+    nip62, Alphabet, Event, Filter, Kind, PublicKey, RelayUrl, SingleLetterTag, Timestamp,
+    WritePolicyResult,
 };
 
 use crate::nostr::events::RepositoryAnnouncement;
+use crate::nostr::lifecycle::RequestLifecycleRecord;
 use crate::nostr::policy::{reject_error, reject_invalid, AnnouncementResult};
 
 use super::{DeletionContext, DeletionPolicy};
@@ -21,56 +23,180 @@ impl DeletionService {
     }
 
     pub async fn gate(&self, event: &Event) -> Option<WritePolicyResult> {
-        // 1. Vanished pubkey
-        if self
-            .ctx
-            .tombstones()
-            .is_pubkey_vanished(&event.pubkey)
-            .await
-        {
-            tracing::debug!(
-                event_id = %event.id.to_hex(),
-                author = %event.pubkey.to_hex(),
-                "Rejected event from vanished pubkey"
-            );
-            return Some(reject_invalid("this pubkey has requested to vanish"));
+        // The operator's current archival-mode policy governs every retained
+        // request, regardless of how it was classified when received.
+        if self.ctx.config.deletion_request_disrespector {
+            return None;
         }
 
-        // 2. Deleted event id (author-bound: only the event's own author may
-        //    have deleted it)
-        if self
-            .ctx
-            .tombstones()
-            .is_event_deleted(&event.id, &event.pubkey)
+        let mut candidates = Vec::new();
+        let tombstones = self.ctx.tombstones();
+        match tombstones.vanish_candidates(&event.pubkey).await {
+            Ok(records) => candidates.extend(records.into_iter().map(|record| (record, "vanish"))),
+            Err(error) => return Some(Self::gate_error("querying vanish candidates", error)),
+        }
+        match tombstones
+            .event_deletion_candidates(&event.id, &event.pubkey)
             .await
         {
-            tracing::debug!(
-                event_id = %event.id.to_hex(),
-                "Rejected re-submission of deleted event"
-            );
-            return Some(reject_invalid("this event is deleted"));
+            Ok(records) => {
+                candidates.extend(records.into_iter().map(|record| (record, "event-id")))
+            }
+            Err(error) => {
+                return Some(Self::gate_error(
+                    "querying event deletion candidates",
+                    error,
+                ))
+            }
         }
 
-        // 3. Deleted coordinate (replaceable / addressable events only)
+        // Deleted coordinate (replaceable / addressable events only).
         if event.kind.is_replaceable() || event.kind.is_addressable() {
             if let Some(coord) = Self::event_coordinate(event) {
-                if self
-                    .ctx
-                    .tombstones()
-                    .is_coordinate_deleted(&coord, event.created_at)
+                match tombstones
+                    .coordinate_deletion_candidates(&coord, event.created_at)
                     .await
                 {
-                    tracing::debug!(
-                        event_id = %event.id.to_hex(),
-                        coordinate = %coord,
-                        "Rejected event whose coordinate was deleted"
-                    );
-                    return Some(reject_invalid("this event is deleted"));
+                    Ok(records) => {
+                        candidates.extend(records.into_iter().map(|record| (record, "coordinate")))
+                    }
+                    Err(error) => {
+                        return Some(Self::gate_error(
+                            "querying coordinate deletion candidates",
+                            error,
+                        ));
+                    }
                 }
             }
         }
 
-        None
+        let now = Timestamp::now();
+        let mut eligible = std::collections::HashMap::new();
+        for (record, category) in candidates {
+            match self.request_is_expired(&record, now) {
+                Ok(true) => continue,
+                Ok(false) => {}
+                Err(error) => {
+                    return Some(Self::gate_error(
+                        "evaluating deletion-request retention",
+                        error,
+                    ));
+                }
+            }
+            eligible
+                .entry(record.request.id)
+                .or_insert((record, category));
+        }
+
+        let candidate_count = eligible.len();
+        let (winner, category) = eligible
+            .into_values()
+            .min_by(|(left, _), (right, _)| Self::winner_order(left, right))?;
+
+        let previous_last_used_at = winner.last_used_at;
+        if let Err(error) = self.ctx.database().save_event(&winner.request).await {
+            return Some(Self::gate_error("promoting deletion request", error));
+        }
+        let used_at = Timestamp::now();
+        let updated = match tombstones
+            .mark_request_used(&winner.request.id, used_at)
+            .await
+        {
+            Ok(Some(record)) if record.last_used_at >= Some(used_at) => record,
+            Ok(Some(record)) => {
+                return Some(Self::gate_error(
+                    "verifying deletion-request lifecycle update",
+                    anyhow::anyhow!(
+                        "request {} retained an older last_used_at {:?}",
+                        record.request.id,
+                        record.last_used_at
+                    ),
+                ));
+            }
+            Ok(None) => {
+                return Some(Self::gate_error(
+                    "updating deletion-request lifecycle",
+                    anyhow::anyhow!(
+                        "request {} no longer has lifecycle metadata",
+                        winner.request.id
+                    ),
+                ));
+            }
+            Err(error) => {
+                return Some(Self::gate_error(
+                    "updating deletion-request lifecycle",
+                    error,
+                ))
+            }
+        };
+
+        tracing::debug!(
+            incoming_event_id = %event.id.to_hex(),
+            winning_request_id = %winner.request.id.to_hex(),
+            request_kind = winner.request.kind.as_u16(),
+            first_seen_at = winner.first_seen_at.as_secs(),
+            previous_last_used_at = ?previous_last_used_at.map(|timestamp| timestamp.as_secs()),
+            new_last_used_at = ?updated.last_used_at.map(|timestamp| timestamp.as_secs()),
+            candidate_count,
+            match_category = category,
+            "Rejected event through attributed deletion-request gate"
+        );
+
+        Some(if winner.request.kind == Kind::RequestToVanish {
+            reject_invalid("this pubkey has requested to vanish")
+        } else {
+            reject_invalid("this event is deleted")
+        })
+    }
+
+    fn request_is_expired(&self, record: &RequestLifecycleRecord, now: Timestamp) -> Result<bool> {
+        let (anchor, served, additional) = match record.last_used_at {
+            Some(last_used_at) => (
+                last_used_at,
+                self.ctx
+                    .config
+                    .deletion_request_retention_used_served_after_last_used(),
+                self.ctx
+                    .config
+                    .deletion_request_retention_used_unserved_gating_additional(),
+            ),
+            None => (
+                record.first_seen_at,
+                self.ctx.config.deletion_request_retention_unused_served(),
+                self.ctx
+                    .config
+                    .deletion_request_retention_unused_unserved_gating_additional(),
+            ),
+        };
+        let lifetime = served
+            .as_secs()
+            .checked_add(additional.as_secs())
+            .ok_or_else(|| anyhow::anyhow!("deletion-request retention duration overflow"))?;
+        let deadline = anchor
+            .as_secs()
+            .checked_add(lifetime)
+            .ok_or_else(|| anyhow::anyhow!("deletion-request retention deadline overflow"))?;
+        if now.as_secs() >= deadline {
+            return Ok(true);
+        }
+        Ok(false)
+    }
+
+    fn winner_order(
+        left: &RequestLifecycleRecord,
+        right: &RequestLifecycleRecord,
+    ) -> std::cmp::Ordering {
+        right
+            .last_used_at
+            .is_some()
+            .cmp(&left.last_used_at.is_some())
+            .then_with(|| left.first_seen_at.cmp(&right.first_seen_at))
+            .then_with(|| left.request.id.to_hex().cmp(&right.request.id.to_hex()))
+    }
+
+    fn gate_error(context: &str, error: impl std::fmt::Display) -> WritePolicyResult {
+        tracing::error!(error = %error, "Deletion-request admission gate failed while {context}");
+        reject_error(format!("internal error {context}: {error}"))
     }
 
     /// Remove stale out-of-order kind-5 requests that targeted this event ID but
@@ -598,3 +724,203 @@ impl DeletionAdmissionHooks<'_> {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::grasp06::receive::new_repo_init_locks;
+    use crate::nostr::lifecycle::{
+        HoldingStore, ReplaceableHistoryStore, RepositoryLifecycle, Tombstones,
+    };
+    use crate::purgatory::Purgatory;
+    use nostr_relay_builder::prelude::{EventBuilder, EventId, FinalizeEvent, Keys, Tag};
+    use std::path::PathBuf;
+    use std::sync::Arc;
+
+    fn context(config: crate::config::Config) -> DeletionContext {
+        DeletionContext::new(
+            "test.example.com",
+            Arc::new(nostr_memory::MemoryDatabase::unbounded()),
+            Tombstones::in_memory(),
+            HoldingStore::in_memory(),
+            RepositoryLifecycle::in_memory(),
+            ReplaceableHistoryStore::in_memory(),
+            PathBuf::new(),
+            Arc::new(Purgatory::new(PathBuf::new())),
+            config,
+            new_repo_init_locks(),
+        )
+    }
+
+    fn deletion(keys: &Keys, target: EventId) -> Event {
+        EventBuilder::new(Kind::EventDeletion, "")
+            .tags(vec![Tag::event(target)])
+            .finalize(keys)
+            .unwrap()
+    }
+
+    #[test]
+    fn retention_expiry_uses_exact_deadline_boundary() {
+        let service = DeletionService::new(context(crate::config::Config {
+            deletion_request_retention_unused_served_secs: 2,
+            deletion_request_retention_unused_unserved_gating_additional_secs: 3,
+            deletion_request_retention_used_served_after_last_used_secs: 2,
+            deletion_request_retention_used_unserved_gating_additional_secs: 3,
+            ..crate::config::Config::for_testing()
+        }));
+        let request = deletion(&Keys::generate(), EventId::all_zeros());
+        let unused = RequestLifecycleRecord {
+            metadata_event_id: EventId::all_zeros(),
+            request: request.clone(),
+            first_seen_at: Timestamp::from_secs(10),
+            last_used_at: None,
+            classification: crate::nostr::lifecycle::RequestClassification::LocallyActionable,
+        };
+        assert!(!service
+            .request_is_expired(&unused, Timestamp::from_secs(14))
+            .unwrap());
+        assert!(service
+            .request_is_expired(&unused, Timestamp::from_secs(15))
+            .unwrap());
+        let used = RequestLifecycleRecord {
+            last_used_at: Some(Timestamp::from_secs(20)),
+            ..unused
+        };
+        assert!(!service
+            .request_is_expired(&used, Timestamp::from_secs(24))
+            .unwrap());
+        assert!(service
+            .request_is_expired(&used, Timestamp::from_secs(25))
+            .unwrap());
+    }
+
+    #[test]
+    fn winner_order_prefers_used_then_first_seen_then_event_id() {
+        let keys = Keys::generate();
+        let first_request = EventBuilder::new(Kind::EventDeletion, "first")
+            .tags(vec![Tag::event(EventId::all_zeros())])
+            .finalize(&keys)
+            .unwrap();
+        let second_request = EventBuilder::new(Kind::EventDeletion, "second")
+            .tags(vec![Tag::event(EventId::all_zeros())])
+            .finalize(&keys)
+            .unwrap();
+        let record = |request: Event, first_seen_at: u64, last_used_at: Option<u64>| {
+            RequestLifecycleRecord {
+                metadata_event_id: EventId::all_zeros(),
+                request,
+                first_seen_at: Timestamp::from_secs(first_seen_at),
+                last_used_at: last_used_at.map(Timestamp::from_secs),
+                classification: crate::nostr::lifecycle::RequestClassification::LocallyActionable,
+            }
+        };
+        let unused_early = record(first_request.clone(), 10, None);
+        let unused_late = record(second_request.clone(), 20, None);
+        let used_late = record(second_request.clone(), 20, Some(21));
+        assert_eq!(
+            DeletionService::winner_order(&used_late, &unused_early),
+            std::cmp::Ordering::Less
+        );
+        assert_eq!(
+            DeletionService::winner_order(&unused_early, &unused_late),
+            std::cmp::Ordering::Less
+        );
+        let same_time_left = record(first_request, 10, None);
+        let same_time_right = record(second_request, 10, None);
+        assert_eq!(
+            DeletionService::winner_order(&same_time_left, &same_time_right),
+            same_time_left
+                .request
+                .id
+                .to_hex()
+                .cmp(&same_time_right.request.id.to_hex())
+        );
+    }
+
+    #[tokio::test]
+    async fn gate_deduplicates_matches_promotes_winner_and_updates_only_it() {
+        let ctx = context(crate::config::Config::for_testing());
+        let service = DeletionService::new(ctx.clone());
+        let keys = Keys::generate();
+        let incoming = EventBuilder::new(Kind::TextNote, "target")
+            .finalize(&keys)
+            .unwrap();
+        let older = deletion(&keys, incoming.id);
+        let newer = EventBuilder::new(Kind::EventDeletion, "newer")
+            .tags(vec![Tag::event(incoming.id)])
+            .finalize(&keys)
+            .unwrap();
+        let now = Timestamp::now().as_secs();
+        ctx.tombstones
+            .record_request(
+                &older,
+                Timestamp::from_secs(now - 2),
+                crate::nostr::lifecycle::RequestClassification::LocallyActionable,
+            )
+            .await
+            .unwrap();
+        ctx.tombstones
+            .record_request(
+                &newer,
+                Timestamp::from_secs(now - 1),
+                crate::nostr::lifecycle::RequestClassification::LocallyActionable,
+            )
+            .await
+            .unwrap();
+
+        assert!(service.gate(&incoming).await.is_some());
+        assert!(ctx
+            .database
+            .query(Filter::new().id(older.id))
+            .await
+            .unwrap()
+            .iter()
+            .any(|event| event.id == older.id));
+        assert!(ctx
+            .tombstones
+            .lifecycle_for_request(&older.id)
+            .await
+            .unwrap()
+            .last_used_at
+            .is_some());
+        assert_eq!(
+            ctx.tombstones
+                .lifecycle_for_request(&newer.id)
+                .await
+                .unwrap()
+                .last_used_at,
+            None
+        );
+    }
+
+    #[tokio::test]
+    async fn disrespector_mode_bypasses_all_deletion_gates() {
+        let ctx = context(crate::config::Config {
+            deletion_request_disrespector: true,
+            ..crate::config::Config::for_testing()
+        });
+        let service = DeletionService::new(ctx.clone());
+        let keys = Keys::generate();
+        let event = EventBuilder::new(Kind::TextNote, "target")
+            .finalize(&keys)
+            .unwrap();
+        let request = deletion(&keys, event.id);
+        ctx.tombstones
+            .record_request(
+                &request,
+                Timestamp::now(),
+                crate::nostr::lifecycle::RequestClassification::LocallyActionable,
+            )
+            .await
+            .unwrap();
+        assert!(service.gate(&event).await.is_none());
+        assert_eq!(
+            ctx.tombstones
+                .lifecycle_for_request(&request.id)
+                .await
+                .unwrap()
+                .last_used_at,
+            None
+        );
+    }
+}

src/nostr/lifecycle/tombstones.rs

@@ -249,14 +249,29 @@ impl Tombstones {
         &self,
         request_id: &EventId,
     ) -> Option<RequestLifecycleRecord> {
-        let request = match self.request_payload(request_id).await {
-            Ok(Some(event)) if Self::is_request_kind(event.kind) => event,
-            Ok(Some(event)) => {
-                tracing::warn!(request_id = %request_id, kind = event.kind.as_u16(), "Tombstone lifecycle payload has invalid kind");
-                return None;
+        match self.lifecycle_for_request_result(request_id).await {
+            Ok(record) => record,
+            Err(error) => {
+                tracing::error!(error = %error, request_id = %request_id, "Failed to read tombstone lifecycle record");
+                None
             }
-            Ok(None) => {
-                let orphaned_metadata = self.metadata_events_for_request(request_id).await;
+        }
+    }
+
+    /// Fallible counterpart to [`Self::lifecycle_for_request`], for policy paths
+    /// where an unreadable lifecycle must not be mistaken for an absent one.
+    pub async fn lifecycle_for_request_result(
+        &self,
+        request_id: &EventId,
+    ) -> anyhow::Result<Option<RequestLifecycleRecord>> {
+        let request = match self.request_payload(request_id).await? {
+            Some(event) if Self::is_request_kind(event.kind) => event,
+            Some(event) => {
+                tracing::warn!(request_id = %request_id, kind = event.kind.as_u16(), "Tombstone lifecycle payload has invalid kind");
+                return Ok(None);
+            }
+            None => {
+                let orphaned_metadata = self.metadata_events_for_request_result(request_id).await?;
                 if !orphaned_metadata.is_empty() {
                     tracing::warn!(
                         request_id = %request_id,
@@ -264,21 +279,17 @@ impl Tombstones {
                         "Ignoring orphan tombstone lifecycle metadata without a request payload"
                     );
                 }
-                return None;
-            }
-            Err(error) => {
-                tracing::error!(error = %error, request_id = %request_id, "Failed to read tombstone lifecycle payload");
-                return None;
+                return Ok(None);
             }
         };
 
-        let metadata = self.metadata_events_for_request(request_id).await;
+        let metadata = self.metadata_events_for_request_result(request_id).await?;
         let mut valid: Vec<_> = metadata
             .into_iter()
             .filter_map(|event| self.parse_metadata(&event, request_id))
             .collect();
         if valid.is_empty() {
-            return None;
+            return Ok(None);
         }
 
         let first_seen_at = valid
@@ -296,13 +307,13 @@ impl Tombstones {
                 .then_with(|| left.id.to_hex().cmp(&right.id.to_hex()))
         });
         let canonical = valid.remove(0);
-        Some(RequestLifecycleRecord {
+        Ok(Some(RequestLifecycleRecord {
             metadata_event_id: canonical.id,
             request,
             first_seen_at,
             last_used_at,
             classification: canonical.classification,
-        })
+        }))
     }
 
     /// Set `last_used_at`, never moving it backwards, then compact superseded
@@ -371,7 +382,7 @@ impl Tombstones {
                 records.push(record);
             }
         }
-        records.sort_by(|left, right| left.request.id.to_hex().cmp(&right.request.id.to_hex()));
+        records.sort_by_key(|record| record.request.id.to_hex());
         records
     }
 
@@ -422,11 +433,8 @@ impl Tombstones {
     }
 
     async fn metadata_events_for_request(&self, request_id: &EventId) -> Vec<Event> {
-        let filter = Filter::new()
-            .kind(Kind::from(TOMBSTONE_REQUEST_METADATA_KIND))
-            .custom_tag(SingleLetterTag::lowercase(Alphabet::E), request_id.to_hex());
-        match self.db.query(filter).await {
-            Ok(events) => events.into_iter().collect(),
+        match self.metadata_events_for_request_result(request_id).await {
+            Ok(events) => events,
             Err(error) => {
                 tracing::error!(error = %error, request_id = %request_id, "Failed to query tombstone lifecycle metadata");
                 Vec::new()
@@ -434,6 +442,24 @@ impl Tombstones {
         }
     }
 
+    async fn metadata_events_for_request_result(
+        &self,
+        request_id: &EventId,
+    ) -> anyhow::Result<Vec<Event>> {
+        let filter = Filter::new()
+            .kind(Kind::from(TOMBSTONE_REQUEST_METADATA_KIND))
+            .custom_tag(SingleLetterTag::lowercase(Alphabet::E), request_id.to_hex());
+        self.db
+            .query(filter)
+            .await
+            .map(|events| events.into_iter().collect())
+            .map_err(|error| {
+                anyhow::anyhow!(
+                    "Failed to query tombstone lifecycle metadata for {request_id}: {error}"
+                )
+            })
+    }
+
     fn parse_metadata(&self, event: &Event, request_id: &EventId) -> Option<ParsedMetadata> {
         let linked = Self::tag_value(event, "e")
             .and_then(|value| EventId::from_hex(value).ok())
@@ -718,22 +744,36 @@ impl Tombstones {
     /// otherwise any pubkey could pre-emptively censor another's events by
     /// racing a kind-5 ahead of the target (out-of-order delete).
     pub async fn is_event_deleted(&self, id: &EventId, author: &PublicKey) -> bool {
-        // A stored kind-5 authored by `author` carrying this id in an `e` tag
-        // means `author` deleted their own event.
+        match self.event_deletion_candidates(id, author).await {
+            Ok(candidates) => !candidates.is_empty(),
+            Err(error) => {
+                tracing::error!(error = %error, "Tombstone query failed for is_event_deleted");
+                false
+            }
+        }
+    }
+
+    /// Return locally actionable, lifecycle-backed kind-5 requests that delete
+    /// `id` on behalf of `author`.
+    pub async fn event_deletion_candidates(
+        &self,
+        id: &EventId,
+        author: &PublicKey,
+    ) -> anyhow::Result<Vec<RequestLifecycleRecord>> {
         let filter = Filter::new()
             .kind(Kind::EventDeletion)
             .author(*author)
             .custom_tag(SingleLetterTag::lowercase(Alphabet::E), id.to_hex());
-        match self.db.query(filter).await {
-            Ok(events) => !events.is_empty(),
-            Err(e) => {
-                // Fail secure: if we cannot determine deletion status, do not
-                // claim the event is deleted (so we don't reject legitimate
-                // events), but log loudly.
-                tracing::error!(error = %e, "Tombstone query failed for is_event_deleted");
-                false
-            }
-        }
+        self.lifecycle_candidates(filter, |request| {
+            request.kind == Kind::EventDeletion
+                && request.pubkey == *author
+                && request.tags.iter().any(|tag| {
+                    tag.as_slice()
+                        .get(0..2)
+                        .is_some_and(|values| values[0] == "e" && values[1] == id.to_hex())
+                })
+        })
+        .await
     }
 
     /// Has this addressable/replaceable coordinate been deleted at or after
@@ -753,6 +793,25 @@ impl Tombstones {
         coordinate: &str,
         event_created_at: Timestamp,
     ) -> bool {
+        match self
+            .coordinate_deletion_candidates(coordinate, event_created_at)
+            .await
+        {
+            Ok(candidates) => !candidates.is_empty(),
+            Err(error) => {
+                tracing::error!(error = %error, "Tombstone query failed for is_coordinate_deleted");
+                false
+            }
+        }
+    }
+
+    /// Return locally actionable, lifecycle-backed kind-5 requests that delete
+    /// the coordinate at or after `event_created_at`.
+    pub async fn coordinate_deletion_candidates(
+        &self,
+        coordinate: &str,
+        event_created_at: Timestamp,
+    ) -> anyhow::Result<Vec<RequestLifecycleRecord>> {
         // The coordinate owner (pubkey hex) is the only party allowed to delete
         // it. Extract it so we can require the kind-5 author to match.
         let coord_owner_hex = coordinate.split(':').nth(1);
@@ -761,18 +820,19 @@ impl Tombstones {
             SingleLetterTag::lowercase(Alphabet::A),
             coordinate.to_string(),
         );
-        match self.db.query(filter).await {
-            Ok(events) => events.iter().any(|deletion| {
-                deletion.created_at >= event_created_at
-                    && coord_owner_hex
-                        .map(|owner| owner == deletion.pubkey.to_hex())
-                        .unwrap_or(false)
-            }),
-            Err(e) => {
-                tracing::error!(error = %e, "Tombstone query failed for is_coordinate_deleted");
-                false
-            }
-        }
+        self.lifecycle_candidates(filter, |request| {
+            request.kind == Kind::EventDeletion
+                && request.created_at >= event_created_at
+                && coord_owner_hex
+                    .map(|owner| owner == request.pubkey.to_hex())
+                    .unwrap_or(false)
+                && request.tags.iter().any(|tag| {
+                    tag.as_slice()
+                        .get(0..2)
+                        .is_some_and(|values| values[0] == "a" && values[1] == coordinate)
+                })
+        })
+        .await
     }
 
     /// Has this pubkey requested to vanish from this relay (recorded kind-62)?
@@ -780,14 +840,63 @@ impl Tombstones {
     /// Mirrors the backend's `is_pubkey_vanished` check.
     pub async fn is_pubkey_vanished(&self, pubkey: &PublicKey) -> bool {
         let filter = Filter::new().kind(Kind::RequestToVanish).author(*pubkey);
-        match self.db.query(filter).await {
-            Ok(events) => !events.is_empty(),
-            Err(e) => {
-                tracing::error!(error = %e, "Tombstone query failed for is_pubkey_vanished");
+        match self
+            .lifecycle_candidates(filter, |request| {
+                request.kind == Kind::RequestToVanish && request.pubkey == *pubkey
+            })
+            .await
+        {
+            Ok(candidates) => !candidates.is_empty(),
+            Err(error) => {
+                tracing::error!(error = %error, "Tombstone query failed for is_pubkey_vanished");
                 false
             }
         }
     }
+
+    /// Return locally actionable, lifecycle-backed kind-62 requests made by
+    /// `pubkey`. Non-targeting and disrespector classifications never gate.
+    pub async fn vanish_candidates(
+        &self,
+        pubkey: &PublicKey,
+    ) -> anyhow::Result<Vec<RequestLifecycleRecord>> {
+        let filter = Filter::new().kind(Kind::RequestToVanish).author(*pubkey);
+        self.lifecycle_candidates(filter, |request| {
+            request.kind == Kind::RequestToVanish && request.pubkey == *pubkey
+        })
+        .await
+    }
+
+    async fn lifecycle_candidates<F>(
+        &self,
+        filter: Filter,
+        matches_request: F,
+    ) -> anyhow::Result<Vec<RequestLifecycleRecord>>
+    where
+        F: Fn(&Event) -> bool,
+    {
+        let requests =
+            self.db.query(filter).await.map_err(|error| {
+                anyhow::anyhow!("Failed to query tombstone candidates: {error}")
+            })?;
+        let mut candidates = Vec::new();
+        for request in requests {
+            // Queries are intentionally restricted to original kind-5/kind-62
+            // payloads; private metadata can never enter this path.
+            if !Self::is_request_kind(request.kind) || !matches_request(&request) {
+                continue;
+            }
+            let Some(record) = self.lifecycle_for_request_result(&request.id).await? else {
+                // Requests without valid metadata await startup migration and do
+                // not yet carry enough relay-observed state to gate admission.
+                continue;
+            };
+            if record.classification == RequestClassification::LocallyActionable {
+                candidates.push(record);
+            }
+        }
+        Ok(candidates)
+    }
 }
 
 #[cfg(test)]
@@ -1014,6 +1123,127 @@ mod tests {
         assert!(!store.is_pubkey_vanished(&keys.public_key()).await);
     }
 
+    #[tokio::test]
+    async fn event_deletion_candidates_are_author_bound_and_lifecycle_backed() {
+        let store = Tombstones::in_memory();
+        let owner = Keys::generate();
+        let other = Keys::generate();
+        let target = EventId::all_zeros();
+        let request = deletion_by_event(&owner, target);
+        store
+            .record_request(
+                &request,
+                Timestamp::from_secs(10),
+                RequestClassification::LocallyActionable,
+            )
+            .await
+            .unwrap();
+
+        let candidates = store
+            .event_deletion_candidates(&target, &owner.public_key())
+            .await
+            .unwrap();
+        assert_eq!(candidates.len(), 1);
+        assert_eq!(candidates[0].request.id, request.id);
+        assert!(store
+            .event_deletion_candidates(&target, &other.public_key())
+            .await
+            .unwrap()
+            .is_empty());
+    }
+
+    #[tokio::test]
+    async fn coordinate_candidates_enforce_cutoff_and_coordinate_owner() {
+        let store = Tombstones::in_memory();
+        let owner = Keys::generate();
+        let attacker = Keys::generate();
+        let coordinate = format!("30618:{}:repo", owner.public_key().to_hex());
+        let valid = EventBuilder::new(Kind::EventDeletion, "")
+            .tags(vec![Tag::custom("a", vec![coordinate.clone()])])
+            .custom_created_at(Timestamp::from_secs(100))
+            .finalize(&owner)
+            .unwrap();
+        let invalid = EventBuilder::new(Kind::EventDeletion, "")
+            .tags(vec![Tag::custom("a", vec![coordinate.clone()])])
+            .custom_created_at(Timestamp::from_secs(100))
+            .finalize(&attacker)
+            .unwrap();
+        for request in [&valid, &invalid] {
+            store
+                .record_request(
+                    request,
+                    Timestamp::from_secs(10),
+                    RequestClassification::LocallyActionable,
+                )
+                .await
+                .unwrap();
+        }
+
+        assert_eq!(
+            store
+                .coordinate_deletion_candidates(&coordinate, Timestamp::from_secs(100))
+                .await
+                .unwrap()
+                .iter()
+                .map(|record| record.request.id)
+                .collect::<Vec<_>>(),
+            vec![valid.id]
+        );
+        assert_eq!(
+            store
+                .coordinate_deletion_candidates(&coordinate, Timestamp::from_secs(99))
+                .await
+                .unwrap()
+                .len(),
+            1
+        );
+        assert!(store
+            .coordinate_deletion_candidates(&coordinate, Timestamp::from_secs(101))
+            .await
+            .unwrap()
+            .is_empty());
+    }
+
+    #[tokio::test]
+    async fn vanish_candidates_require_locally_actionable_classification() {
+        let store = Tombstones::in_memory();
+        let keys = Keys::generate();
+        let actionable = vanish(&keys);
+        store
+            .record_request(
+                &actionable,
+                Timestamp::from_secs(10),
+                RequestClassification::LocallyActionable,
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            store
+                .vanish_candidates(&keys.public_key())
+                .await
+                .unwrap()
+                .len(),
+            1
+        );
+
+        for classification in [
+            RequestClassification::NonTargetingNip62,
+            RequestClassification::Disrespector,
+        ] {
+            let store = Tombstones::in_memory();
+            let request = vanish(&keys);
+            store
+                .record_request(&request, Timestamp::from_secs(10), classification)
+                .await
+                .unwrap();
+            assert!(store
+                .vanish_candidates(&keys.public_key())
+                .await
+                .unwrap()
+                .is_empty());
+        }
+    }
+
     #[tokio::test]
     async fn lifecycle_metadata_survives_lmdb_reopen() {
         let directory = tempfile::tempdir().unwrap();