laantungir/ngit-grasp
1
0
Fork 0
mirror of https://relay.ngit.dev/npub15qydau2hjma6ngxkl2cyar74wzyjshvl65za5k5rl69264ar2exs5cyejr/ngit-grasp.git synced 2026-10-05 15:08:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: implement announcement purgatory core (breaks archive sync test)

Browse Source 
Route new announcements to purgatory instead of accepting immediately.
Announcements are promoted to the database when git data arrives,
ensuring we only serve announcements for repos with actual content.

Implemented:
- AnnouncementPurgatoryEntry type and DashMap store
- Route new announcements to purgatory (replacement announcements skip)
- Promote announcements on git data arrival (process_purgatory_announcements)
- Authorization checks purgatory announcements (fetch_repository_data_with_purgatory)
- State policy uses purgatory announcements for maintainer validation
- Cleanup task handles announcement expiry
- Updated count()/cleanup() to 3-tuples

Known broken:
- test_archive_read_only_creates_bare_repo fails: sync module does not
  treat purgatory announcements as confirmed repos, so per-repo sync
  (state events, PRs) is never triggered for purgatory announcements
- Announcement persistence (save/restore) not implemented
- SyncLevel (StateOnly vs Full) not implemented
- Soft expiry two-phase not implemented
- Expiry extension on state event / git auth not wired up
This commit is contained in:
DanConwayDev
2026-02-13 17:29:23 +00:00
parent a2a99d5a41
commit 1d09e4bdea
13 changed files with 691 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/git/authorization.rs
+36 -2
View File
@@ -287,6 +287,39 @@ pub async fn fetch_repository_data(
})
}
/// Fetch repository data including announcements from purgatory
///
/// This combines database announcements with purgatory announcements,
/// which is needed for authorization when the announcement hasn't been
/// promoted yet (no git data has arrived).
pub async fn fetch_repository_data_with_purgatory(
database: &SharedDatabase,
purgatory: &crate::purgatory::Purgatory,
identifier: &str,
) -> Result<RepositoryData> {
// First, fetch from database
let mut repo_data = fetch_repository_data(database, identifier).await?;
// Then, add announcements from purgatory
let purgatory_announcements = purgatory.get_announcements_by_identifier(identifier);
let purgatory_count = purgatory_announcements.len();
for entry in purgatory_announcements {
if let Ok(announcement) = RepositoryAnnouncement::from_event(entry.event) {
repo_data.announcements.push(announcement);
}
}
debug!(
"Fetched repository data with purgatory: {} announcements ({} from purgatory), {} states",
repo_data.announcements.len(),
purgatory_count,
repo_data.states.len()
);
Ok(repo_data)
}
pub fn pubkey_authorised_for_repo_owners(
pubkey: &PublicKey,
db_repo_data: &RepositoryData,
@@ -539,8 +572,9 @@ pub async fn get_state_authorization_for_specific_owner_repo(
use crate::git::list_refs;
use crate::purgatory::RefUpdate;
// Fetch announcements only - we don't need database states
let repo_data = fetch_repository_data(database, identifier).await?;
// Fetch announcements from database AND purgatory - needed for authorization
// when the announcement hasn't been promoted yet (no git data has arrived)
let repo_data = fetch_repository_data_with_purgatory(database, purgatory, identifier).await?;
if repo_data.announcements.is_empty() {
return Ok(AuthorizationResult::denied(
src/git/sync.rs
+109 -1
View File
@@ -51,6 +51,8 @@ use crate::purgatory::{can_apply_state, Purgatory};
/// or from purgatory sync fetching OIDs from remote servers).
#[derive(Debug, Default, Clone)]
pub struct ProcessResult {
/// Number of announcements released from purgatory
pub announcements_released: usize,
/// Number of state events released from purgatory
pub states_released: usize,
/// Number of PR events released from purgatory
@@ -70,11 +72,12 @@ pub struct ProcessResult {
impl ProcessResult {
/// Check if any events were released
pub fn released_any(&self) -> bool {
self.states_released > 0 || self.prs_released > 0
self.announcements_released > 0 || self.states_released > 0 || self.prs_released > 0
}
/// Merge another ProcessResult into this one
pub fn merge(&mut self, other: ProcessResult) {
self.announcements_released += other.announcements_released;
self.states_released += other.states_released;
self.prs_released += other.prs_released;
self.repos_synced += other.repos_synced;
@@ -836,6 +839,18 @@ pub async fn process_newly_available_git_data(
"Processing newly available git data"
);
// Process announcements from purgatory
let announcement_result = process_purgatory_announcements(
&identifier,
source_repo_path,
database,
local_relay,
purgatory,
git_data_path,
)
.await;
result.merge(announcement_result);
// Process state events from purgatory
let state_result = process_purgatory_state_events(
&identifier,
@@ -863,6 +878,7 @@ pub async fn process_newly_available_git_data(
if result.released_any() {
info!(
identifier = %identifier,
announcements_released = result.announcements_released,
states_released = result.states_released,
prs_released = result.prs_released,
repos_synced = result.repos_synced,
@@ -1250,6 +1266,90 @@ async fn process_purgatory_pr_events(
result
}
/// Process announcements from purgatory that can now be promoted.
///
/// When git data arrives for a repository, any announcements in purgatory
/// for that repository should be promoted to the database and served to clients.
async fn process_purgatory_announcements(
identifier: &str,
source_repo_path: &Path,
database: &SharedDatabase,
local_relay: Option<&nostr_relay_builder::LocalRelay>,
purgatory: &Purgatory,
git_data_path: &Path,
) -> ProcessResult {
let mut result = ProcessResult::default();
// Extract owner pubkey from the source repo path
let owner_pubkey = match extract_owner_from_repo_path(source_repo_path, git_data_path) {
Some(npub) => npub,
None => {
debug!(
identifier = %identifier,
"Could not extract owner from repo path"
);
return result;
}
};
// Parse the npub back to PublicKey
let owner = match nostr_sdk::PublicKey::parse(&owner_pubkey) {
Ok(pk) => pk,
Err(e) => {
warn!(
identifier = %identifier,
owner_pubkey = %owner_pubkey,
error = %e,
"Failed to parse owner pubkey"
);
result.errors.push(format!("Failed to parse owner pubkey: {}", e));
return result;
}
};
// Check if there's an announcement in purgatory for this owner and identifier
let announcement_event = purgatory.promote_announcement(&owner, identifier);
if let Some(event) = announcement_event {
// Save to database
match database.save_event(&event).await {
Ok(_) => {
info!(
identifier = %identifier,
event_id = %event.id,
"Promoted announcement from purgatory to database"
);
// Notify WebSocket subscribers
if let Some(relay) = local_relay {
if relay.notify_event(event.clone()) {
debug!(
identifier = %identifier,
event_id = %event.id,
"Broadcast announcement event to WebSocket listeners"
);
}
}
result.announcements_released += 1;
}
Err(e) => {
warn!(
identifier = %identifier,
event_id = %event.id,
error = %e,
"Failed to save announcement to database"
);
result
.errors
.push(format!("Failed to save announcement: {}", e));
}
}
}
result
}
/// Extract owner pubkey from a repository path.
///
/// Given a path like `{git_data_path}/{npub}/{identifier}.git`, extracts the npub.
@@ -1271,6 +1371,7 @@ mod tests {
#[test]
fn test_process_result_default() {
let result = ProcessResult::default();
assert_eq!(result.announcements_released, 0);
assert_eq!(result.states_released, 0);
assert_eq!(result.prs_released, 0);
assert_eq!(result.repos_synced, 0);
@@ -1282,6 +1383,10 @@ mod tests {
let mut result = ProcessResult::default();
assert!(!result.released_any());
result.announcements_released = 1;
assert!(result.released_any());
result.announcements_released = 0;
result.states_released = 1;
assert!(result.released_any());
@@ -1293,6 +1398,7 @@ mod tests {
#[test]
fn test_process_result_merge() {
let mut result1 = ProcessResult {
announcements_released: 0,
states_released: 1,
prs_released: 2,
repos_synced: 3,
@@ -1303,6 +1409,7 @@ mod tests {
};
let result2 = ProcessResult {
announcements_released: 5,
states_released: 10,
prs_released: 20,
repos_synced: 30,
@@ -1314,6 +1421,7 @@ mod tests {
result1.merge(result2);
assert_eq!(result1.announcements_released, 5);
assert_eq!(result1.states_released, 11);
assert_eq!(result1.prs_released, 22);
assert_eq!(result1.repos_synced, 33);
src/main.rs
+4 -4
View File
@@ -142,11 +142,11 @@ async fn main() -> Result<()> {
let mut interval = tokio::time::interval(Duration::from_secs(60));
loop {
interval.tick().await;
let (state_removed, pr_removed) = cleanup_purgatory.cleanup();
if state_removed > 0 || pr_removed > 0 {
let (announcement_removed, state_removed, pr_removed) = cleanup_purgatory.cleanup();
if announcement_removed > 0 || state_removed > 0 || pr_removed > 0 {
info!(
"Purgatory cleanup: removed {} state events, {} PR events",
state_removed, pr_removed
"Purgatory cleanup: removed {} announcements, {} state events, {} PR events",
announcement_removed, state_removed, pr_removed
);
}
}
src/nostr/builder.rs
+23
View File
@@ -138,6 +138,29 @@ impl Nip34WritePolicy {
}
}
}
AnnouncementResult::AcceptPurgatory => {
// New announcement - add to purgatory
match self.announcement_policy.add_to_purgatory(event) {
Ok(()) => {
tracing::info!(
"Accepted announcement to purgatory: {} (waiting for git data)",
event_id_str
);
WritePolicyResult::Reject {
status: true, // Client sees OK
message: "purgatory: won't be served until git data arrives".into(),
}
}
Err(e) => {
tracing::warn!(
"Failed to add announcement to purgatory {}: {}",
event_id_str,
e
);
WritePolicyResult::reject(e)
}
}
}
AnnouncementResult::AcceptMaintainer => {
// Parse announcement to get details for logging
match RepositoryAnnouncement::from_event(event.clone()) {
src/nostr/policy/announcement.rs
+111 -6
View File
@@ -3,6 +3,7 @@
/// Handles validation of NIP-34 repository announcements (kind 30617)
/// according to GRASP-01 specification.
use nostr_relay_builder::prelude::{Alphabet, Event, Filter, Kind, PublicKey, SingleLetterTag};
use std::collections::HashSet;
use super::PolicyContext;
use crate::config::Config;
@@ -11,12 +12,14 @@ use crate::nostr::events::{validate_announcement, RepositoryAnnouncement};
/// Result of announcement policy evaluation
#[derive(Debug, Clone, PartialEq)]
pub enum AnnouncementResult {
/// Accept: Event lists our service (GRASP-01 compliant)
/// Accept: Event lists our service (GRASP-01 compliant) - replacement announcement
Accept,
/// Accept as maintainer: Event accepted via maintainer exception (multi-maintainer)
AcceptMaintainer,
/// Accept as archive: Event accepted via GRASP-05 archive whitelist (read-only)
AcceptArchive,
/// Accept to purgatory: New announcement, waiting for git data
AcceptPurgatory,
/// Reject: Event fails validation with reason
Reject(String),
}
@@ -35,10 +38,12 @@ impl AnnouncementPolicy {
/// Validate a repository announcement event
///
/// Returns `Accept` if the announcement lists the service properly,
/// `AcceptMaintainer` if accepted via maintainer exception,
/// `AcceptArchive` if accepted via GRASP-05 archive config,
/// or `Reject` with reason.
/// Returns:
/// - `Accept` if this is a replacement announcement (active announcement exists)
/// - `AcceptPurgatory` if this is a new announcement (no active announcement exists)
/// - `AcceptMaintainer` if accepted via maintainer exception
/// - `AcceptArchive` if accepted via GRASP-05 archive config
/// - `Reject` with reason if validation fails
pub async fn validate(&self, event: &Event) -> AnnouncementResult {
// First, try validation (GRASP-01 + GRASP-05)
let validation_result = validate_announcement(event, &self.config);
@@ -67,11 +72,111 @@ impl AnnouncementPolicy {
Err(_) => AnnouncementResult::Reject(reason),
}
}
// Accept, AcceptArchive, or AcceptMaintainer - return as-is
AnnouncementResult::Accept | AnnouncementResult::AcceptArchive => {
// Parse announcement to check for existing active announcement
match RepositoryAnnouncement::from_event(event.clone()) {
Ok(announcement) => {
// Check if there's already an active announcement for this (pubkey, identifier)
match self
.has_active_announcement(&event.pubkey, &announcement.identifier)
.await
{
Ok(true) => {
// Replacement announcement - accept immediately
tracing::debug!(
identifier = %announcement.identifier,
"Replacement announcement - accepting immediately"
);
validation_result
}
Ok(false) => {
// New announcement - route to purgatory
tracing::debug!(
identifier = %announcement.identifier,
"New announcement - routing to purgatory"
);
AnnouncementResult::AcceptPurgatory
}
Err(e) => {
tracing::warn!(
error = %e,
"Failed to check for existing announcement - rejecting"
);
AnnouncementResult::Reject(format!(
"Database error checking existing announcement: {}",
e
))
}
}
}
Err(e) => AnnouncementResult::Reject(format!(
"Failed to parse announcement: {}",
e
)),
}
}
// AcceptPurgatory shouldn't come from validate_announcement, but handle it
result => result,
}
}
/// Check if there's an active announcement in the database for this (pubkey, identifier)
async fn has_active_announcement(
&self,
pubkey: &PublicKey,
identifier: &str,
) -> Result<bool, String> {
let filter = Filter::new()
.kind(Kind::GitRepoAnnouncement)
.author(*pubkey)
.custom_tag(
SingleLetterTag::lowercase(Alphabet::D),
identifier.to_string(),
);
let events: Vec<Event> = match self.ctx.database.query(filter).await {
Ok(events) => events.into_iter().collect(),
Err(e) => return Err(format!("Database query failed: {}", e)),
};
Ok(!events.is_empty())
}
/// Add an announcement to purgatory
///
/// Creates the bare repository and stores the announcement in purgatory
/// until git data arrives.
pub fn add_to_purgatory(&self, event: &Event) -> Result<(), String> {
let announcement = RepositoryAnnouncement::from_event(event.clone())
.map_err(|e| format!("Failed to parse announcement: {}", e))?;
// Create bare repository
self.ensure_bare_repository(&announcement)?;
// Build repo path
let repo_path = self.ctx.git_data_path.join(announcement.repo_path());
// Extract relays from announcement
let relays: HashSet<String> = announcement.relays.iter().cloned().collect();
// Add to purgatory
self.ctx.purgatory.add_announcement(
event.clone(),
announcement.identifier.clone(),
event.pubkey,
repo_path,
relays,
);
tracing::info!(
identifier = %announcement.identifier,
event_id = %event.id,
"Added announcement to purgatory"
);
Ok(())
}
/// Create a bare git repository if it doesn't exist
/// Path format: <git_data_path>/<npub>/<identifier>.git
pub fn ensure_bare_repository(
src/nostr/policy/state.rs
+8 -2
View File
@@ -10,7 +10,7 @@ use nostr_relay_builder::prelude::Event;
use super::PolicyContext;
use crate::git;
use crate::git::authorization::fetch_repository_data;
use crate::git::authorization::fetch_repository_data_with_purgatory;
use crate::nostr::events::{validate_state, RepositoryAnnouncement, RepositoryState};
/// Result of state policy evaluation
@@ -76,7 +76,13 @@ impl StatePolicy {
}
// Get all repositories and state events from db with identifier
let db_repo_data = fetch_repository_data(&self.ctx.database, &state.identifier).await?;
// Include purgatory announcements for authorization
let db_repo_data = fetch_repository_data_with_purgatory(
&self.ctx.database,
&self.ctx.purgatory,
&state.identifier,
)
.await?;
// CRITICAL: Check if author is authorized via maintainer set
// State events MUST be rejected if author is not in maintainer set of any accepted announcement
src/purgatory/mod.rs
+228 -32
View File
@@ -17,7 +17,7 @@ pub mod sync;
mod types;
pub use helpers::{can_apply_state, can_satisfy_state, extract_refs_from_state, get_unpushed_refs};
pub use types::{PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry};
pub use types::{AnnouncementPurgatoryEntry, PrPurgatoryEntry, RefPair, RefUpdate, StatePurgatoryEntry};
use dashmap::DashMap;
use nostr_sdk::prelude::*;
@@ -100,7 +100,8 @@ struct PurgatoryState {
/// Main purgatory structure holding events awaiting git data.
///
/// Provides thread-safe concurrent access to two separate stores:
/// Provides thread-safe concurrent access to three separate stores:
/// - Announcements indexed by (pubkey, identifier)
/// - State events indexed by repository identifier
/// - PR events indexed by event ID
///
@@ -121,6 +122,10 @@ struct PurgatoryState {
/// that we've already determined have no git data available.
#[derive(Clone)]
pub struct Purgatory {
/// Repository announcements (kind 30617) indexed by (owner pubkey, identifier).
/// Key: (PublicKey, String) where String is the repository identifier.
announcement_purgatory: Arc<DashMap<(PublicKey, String), AnnouncementPurgatoryEntry>>,
/// State events (kind 30618) indexed by repository identifier.
/// Multiple state events can wait for the same identifier (different maintainers).
state_events: Arc<DashMap<String, Vec<StatePurgatoryEntry>>>,
@@ -145,6 +150,7 @@ impl Purgatory {
/// Create a new empty purgatory.
pub fn new(git_data_path: impl Into<PathBuf>) -> Self {
Self {
announcement_purgatory: Arc::new(DashMap::new()),
state_events: Arc::new(DashMap::new()),
pr_events: Arc::new(DashMap::new()),
sync_queue: Arc::new(DashMap::new()),
@@ -513,9 +519,171 @@ impl Purgatory {
self.pr_events.remove(event_id);
}
// =========================================================================
// Announcement Purgatory Methods
// =========================================================================
/// Add a repository announcement to purgatory.
///
/// The announcement will be held until git data arrives, at which point
/// it will be promoted to the database and served to clients.
///
/// # Arguments
/// * `event` - The announcement event (kind 30617)
/// * `identifier` - The repository identifier from the 'd' tag
/// * `owner` - The owner pubkey (event author)
/// * `repo_path` - Path to the bare git repository
/// * `relays` - Relay URLs from the announcement (for sync registration)
pub fn add_announcement(
&self,
event: Event,
identifier: String,
owner: PublicKey,
repo_path: PathBuf,
relays: HashSet<String>,
) {
let now = Instant::now();
let entry = AnnouncementPurgatoryEntry {
event,
identifier: identifier.clone(),
owner,
repo_path,
relays,
created_at: now,
expires_at: now + DEFAULT_EXPIRY,
soft_expired: false,
};
let key = (owner, identifier);
self.announcement_purgatory.insert(key.clone(), entry);
tracing::debug!(
owner = %key.0,
identifier = %key.1,
"Added announcement to purgatory"
);
}
/// Find an announcement in purgatory by owner and identifier.
///
/// # Arguments
/// * `owner` - The owner pubkey
/// * `identifier` - The repository identifier
///
/// # Returns
/// The announcement entry if found, None otherwise
pub fn find_announcement(&self, owner: &PublicKey, identifier: &str) -> Option<AnnouncementPurgatoryEntry> {
let key = (*owner, identifier.to_string());
self.announcement_purgatory.get(&key).map(|entry| entry.clone())
}
/// Get all announcements in purgatory for a given identifier.
///
/// This is used for authorization - state events and git pushes need to
/// check purgatory announcements for maintainer validation.
///
/// # Arguments
/// * `identifier` - The repository identifier
///
/// # Returns
/// Vector of announcement entries for this identifier
pub fn get_announcements_by_identifier(&self, identifier: &str) -> Vec<AnnouncementPurgatoryEntry> {
self.announcement_purgatory
.iter()
.filter(|entry| entry.key().1 == identifier)
.map(|entry| entry.value().clone())
.collect()
}
/// Remove an announcement from purgatory.
///
/// # Arguments
/// * `owner` - The owner pubkey
/// * `identifier` - The repository identifier
pub fn remove_announcement(&self, owner: &PublicKey, identifier: &str) {
let key = (*owner, identifier.to_string());
self.announcement_purgatory.remove(&key);
tracing::debug!(
owner = %owner,
identifier = %identifier,
"Removed announcement from purgatory"
);
}
/// Promote an announcement from purgatory to active status.
///
/// This is called when git data arrives. The announcement event is returned
/// so it can be saved to the database.
///
/// # Arguments
/// * `owner` - The owner pubkey
/// * `identifier` - The repository identifier
///
/// # Returns
/// The announcement event if found, None otherwise
pub fn promote_announcement(&self, owner: &PublicKey, identifier: &str) -> Option<Event> {
let key = (*owner, identifier.to_string());
self.announcement_purgatory.remove(&key).map(|(_, entry)| {
tracing::info!(
owner = %owner,
identifier = %identifier,
"Promoted announcement from purgatory to database"
);
entry.event
})
}
/// Check if there's an announcement in purgatory for the given owner and identifier.
///
/// # Arguments
/// * `owner` - The owner pubkey
/// * `identifier` - The repository identifier
///
/// # Returns
/// true if an announcement exists in purgatory, false otherwise
pub fn has_purgatory_announcement(&self, owner: &PublicKey, identifier: &str) -> bool {
let key = (*owner, identifier.to_string());
self.announcement_purgatory.contains_key(&key)
}
/// Extend the expiry for an announcement in purgatory.
///
/// This is called when state events arrive for a purgatory announcement,
/// indicating the repository is actively receiving metadata.
///
/// # Arguments
/// * `owner` - The owner pubkey
/// * `identifier` - The repository identifier
/// * `duration` - Minimum duration to guarantee from now
pub fn extend_announcement_expiry(&self, owner: &PublicKey, identifier: &str, duration: Duration) {
let key = (*owner, identifier.to_string());
if let Some(mut entry) = self.announcement_purgatory.get_mut(&key) {
let now = Instant::now();
let new_expiry = now + duration;
if entry.expires_at < new_expiry {
entry.expires_at = new_expiry;
// If soft-expired, revive it
if entry.soft_expired {
entry.soft_expired = false;
tracing::debug!(
owner = %owner,
identifier = %identifier,
"Revived soft-expired announcement"
);
}
}
}
}
/// Get count of announcements in purgatory.
pub fn announcement_count(&self) -> usize {
self.announcement_purgatory.len()
}
/// Get all event IDs currently stored in purgatory AND previously expired events.
///
/// Returns a HashSet of all event IDs for:
/// - Announcements currently held in purgatory
/// - State events currently held in purgatory
/// - PR events currently held in purgatory
/// - Events that previously expired from purgatory without finding git data
@@ -530,6 +698,11 @@ impl Purgatory {
pub fn event_ids(&self) -> HashSet<EventId> {
let mut ids = HashSet::new();
// Collect announcement event IDs
for entry in self.announcement_purgatory.iter() {
ids.insert(entry.value().event.id);
}
// Collect state event IDs
for entry in self.state_events.iter() {
for state_entry in entry.value().iter() {
@@ -609,9 +782,28 @@ impl Purgatory {
/// will be filtered out during future negentropy/REQ sync operations.
///
/// # Returns
/// Tuple of (num_state_removed, num_pr_removed)
pub fn cleanup(&self) -> (usize, usize) {
/// Tuple of (num_announcement_removed, num_state_removed, num_pr_removed)
pub fn cleanup(&self) -> (usize, usize, usize) {
let now = Instant::now();
// Remove expired announcements and mark them as expired
let expired_announcements: Vec<(PublicKey, String, EventId)> = self
.announcement_purgatory
.iter()
.filter(|entry| entry.value().expires_at <= now)
.map(|entry| {
let key = entry.key();
let event_id = entry.value().event.id;
(key.0.clone(), key.1.clone(), event_id)
})
.collect();
let announcement_removed = expired_announcements.len();
for (owner, identifier, event_id) in expired_announcements {
self.mark_expired(event_id);
self.announcement_purgatory.remove(&(owner, identifier));
}
let mut state_removed = 0;
// Remove expired state events and mark them as expired
@@ -655,17 +847,17 @@ impl Purgatory {
self.pr_events.remove(&event_id_str);
}
(state_removed, pr_removed)
(announcement_removed, state_removed, pr_removed)
}
/// Remove expired entries from purgatory (legacy method).
///
/// # Returns
/// Total number of entries removed (state + PR events)
/// Total number of entries removed (announcement + state + PR events)
#[deprecated(since = "0.1.0", note = "Use cleanup() instead for separate counts")]
pub fn remove_expired(&self) -> usize {
let (state, pr) = self.cleanup();
state + pr
let (announcement, state, pr) = self.cleanup();
announcement + state + pr
}
/// Remove old expired event records.
@@ -699,11 +891,12 @@ impl Purgatory {
/// Get current count of entries in purgatory.
///
/// # Returns
/// Tuple of (state_event_count, pr_event_count)
pub fn count(&self) -> (usize, usize) {
/// Tuple of (announcement_count, state_event_count, pr_event_count)
pub fn count(&self) -> (usize, usize, usize) {
let announcement_count = self.announcement_purgatory.len();
let state_count: usize = self.state_events.iter().map(|e| e.value().len()).sum();
let pr_count = self.pr_events.len();
(state_count, pr_count)
(announcement_count, state_count, pr_count)
}
/// Get count of expired events being tracked.
@@ -717,6 +910,7 @@ impl Purgatory {
/// Clear all entries from purgatory (for testing).
#[cfg(test)]
pub fn clear(&self) {
self.announcement_purgatory.clear();
self.state_events.clear();
self.pr_events.clear();
self.sync_queue.clear();
@@ -990,7 +1184,8 @@ mod tests {
#[test]
fn test_purgatory_creation() {
let purgatory = Purgatory::new(PathBuf::new());
let (state_count, pr_count) = purgatory.count();
let (announcement_count, state_count, pr_count) = purgatory.count();
assert_eq!(announcement_count, 0);
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
}
@@ -1008,7 +1203,8 @@ mod tests {
purgatory.add_state(event.clone(), "test-repo".to_string(), keys.public_key());
purgatory.add_pr(event, "test-event-id".to_string(), "abc123".to_string());
let (state_count, pr_count) = purgatory.count();
let (announcement_count, state_count, pr_count) = purgatory.count();
assert_eq!(announcement_count, 0);
assert_eq!(state_count, 1);
assert_eq!(pr_count, 1);
}
@@ -1213,7 +1409,7 @@ fn test_cleanup_removes_expired_entries() {
purgatory.add_pr_placeholder("pr-456".to_string(), "commit-def".to_string());
// Verify entries are there
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 1);
assert_eq!(pr_count, 2);
@@ -1231,14 +1427,14 @@ fn test_cleanup_removes_expired_entries() {
}
// Run cleanup
let (state_removed, pr_removed) = purgatory.cleanup();
let (_, state_removed, pr_removed) = purgatory.cleanup();
// Verify counts
assert_eq!(state_removed, 1);
assert_eq!(pr_removed, 2);
// Verify entries are gone
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
}
@@ -1260,14 +1456,14 @@ fn test_cleanup_preserves_non_expired_entries() {
purgatory.add_pr(pr_event, "pr-123".to_string(), "commit-abc".to_string());
// Run cleanup
let (state_removed, pr_removed) = purgatory.cleanup();
let (_, state_removed, pr_removed) = purgatory.cleanup();
// Nothing should be removed
assert_eq!(state_removed, 0);
assert_eq!(pr_removed, 0);
// Verify entries are still there
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 1);
assert_eq!(pr_count, 1);
}
@@ -1314,14 +1510,14 @@ fn test_cleanup_mixed_expired_and_fresh() {
}
// Run cleanup
let (state_removed, pr_removed) = purgatory.cleanup();
let (_, state_removed, pr_removed) = purgatory.cleanup();
// One of each should be removed
assert_eq!(state_removed, 1);
assert_eq!(pr_removed, 1);
// Verify remaining counts
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 1); // One state event remains
assert_eq!(pr_count, 1); // One PR event remains
}
@@ -1391,7 +1587,7 @@ fn test_expired_event_tracking() {
}
// Run cleanup
let (state_removed, pr_removed) = purgatory.cleanup();
let (_, state_removed, pr_removed) = purgatory.cleanup();
assert_eq!(state_removed, 1);
assert_eq!(pr_removed, 1);
@@ -1501,7 +1697,7 @@ fn test_expired_events_prevent_readdition() {
}
// Event should NOT be re-added
let (state_count, _) = purgatory.count();
let (_, state_count, _) = purgatory.count();
assert_eq!(state_count, 0, "Event should not be re-added to purgatory");
}
@@ -1520,7 +1716,7 @@ fn test_pr_placeholder_not_marked_expired() {
}
// Run cleanup
let (_, pr_removed) = purgatory.cleanup();
let (_, _, pr_removed) = purgatory.cleanup();
assert_eq!(pr_removed, 1);
// Expired count should be 0 (placeholders don't have event IDs to track)
@@ -1606,7 +1802,7 @@ async fn test_save_and_restore_state_events() {
assert!(!state_file.exists());
// Verify state events were restored
let (state_count, _) = purgatory2.count();
let (_, state_count, _) = purgatory2.count();
assert_eq!(state_count, 2);
let restored_entries = purgatory2.find_state("test-repo");
@@ -1662,7 +1858,7 @@ async fn test_save_and_restore_pr_events() {
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify PR event was restored
let (_, pr_count) = purgatory2.count();
let (_, _, pr_count) = purgatory2.count();
assert_eq!(pr_count, 1);
let restored_entry = purgatory2.find_pr("pr-event-id").unwrap();
@@ -1691,7 +1887,7 @@ async fn test_save_and_restore_pr_placeholders() {
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify placeholder was restored
let (_, pr_count) = purgatory2.count();
let (_, _, pr_count) = purgatory2.count();
assert_eq!(pr_count, 1);
let restored_entry = purgatory2.find_pr("placeholder-id").unwrap();
@@ -1769,7 +1965,7 @@ async fn test_save_and_restore_empty_purgatory() {
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify purgatory is still empty
let (state_count, pr_count) = purgatory2.count();
let (_, state_count, pr_count) = purgatory2.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
assert_eq!(purgatory2.expired_count(), 0);
@@ -1789,7 +1985,7 @@ async fn test_restore_missing_file() {
assert!(result.is_err());
// Purgatory should remain empty
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
}
@@ -1811,7 +2007,7 @@ async fn test_restore_corrupted_json() {
assert!(result.is_err());
// Purgatory should remain empty
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
}
@@ -2044,7 +2240,7 @@ async fn test_mixed_pr_events_and_placeholders() {
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify both were restored correctly
let (_, pr_count) = purgatory2.count();
let (_, _, pr_count) = purgatory2.count();
assert_eq!(pr_count, 2);
// Verify PR event
@@ -2141,7 +2337,7 @@ async fn test_comprehensive_roundtrip() {
purgatory.cleanup();
// Verify initial state
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 2); // state1, state2 (expired_event was cleaned up)
assert_eq!(pr_count, 2); // pr-1, pr-2
assert_eq!(purgatory.expired_count(), 1); // expired_event
@@ -2154,7 +2350,7 @@ async fn test_comprehensive_roundtrip() {
purgatory2.restore_from_disk(&state_file).unwrap();
// Verify all data was restored correctly
let (state_count2, pr_count2) = purgatory2.count();
let (_, state_count2, pr_count2) = purgatory2.count();
assert_eq!(state_count2, 2);
assert_eq!(pr_count2, 2);
assert_eq!(purgatory2.expired_count(), 1);
src/purgatory/sync/context.rs
+6 -1
View File
@@ -279,7 +279,12 @@ impl SyncContext for RealSyncContext {
}
async fn fetch_repository_data(&self, identifier: &str) -> Result<RepositoryData> {
crate::git::authorization::fetch_repository_data(&self.database, identifier).await
crate::git::authorization::fetch_repository_data_with_purgatory(
&self.database,
&self.purgatory,
identifier,
)
.await
}
fn collect_needed_oids(&self, identifier: &str) -> HashSet<String> {
src/purgatory/types.rs
+39
View File
@@ -6,6 +6,8 @@
use nostr_sdk::prelude::*;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::path::PathBuf;
use std::time::Instant;
/// Default value for Instant fields during deserialization
@@ -113,3 +115,40 @@ pub struct PrPurgatoryEntry {
#[serde(skip, default = "instant_now")]
pub expires_at: Instant,
}
/// Entry for a repository announcement (kind 30617) waiting in purgatory.
///
/// Announcements are held in purgatory until git data arrives, proving
/// the repository has actual content. This prevents serving announcements
/// for empty repositories.
///
/// Note: `Instant` fields cannot be serialized directly. Use the `persistence`
/// module to convert to/from serializable wrapper types.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AnnouncementPurgatoryEntry {
/// The nostr announcement event (kind 30617)
pub event: Event,
/// The repository identifier from the event's 'd' tag
pub identifier: String,
/// The owner pubkey (event author)
pub owner: PublicKey,
/// Path to the bare git repository
pub repo_path: PathBuf,
/// Relay URLs from the announcement (for sync registration)
pub relays: HashSet<String>,
/// When this entry was added to purgatory
#[serde(skip, default = "instant_now")]
pub created_at: Instant,
/// Expiry deadline (30 min from creation, may be extended)
#[serde(skip, default = "instant_now")]
pub expires_at: Instant,
/// Whether the bare repo has been deleted (soft expiry)
pub soft_expired: bool,
}
src/sync/mod.rs
+66 -2
View File
@@ -1719,8 +1719,50 @@ impl SyncManager {
// For sync-triggered events that go to purgatory, trigger immediate sync
// (instead of the default 3-minute delay for user-submitted events)
if result == ProcessResult::Purgatory {
// Announcements (kind 30617) - re-process rejected state events
// When an announcement goes to purgatory, state events that were
// previously rejected ("no announcement exists") can now be authorized
// via fetch_repository_data_with_purgatory.
if event.kind == Kind::GitRepoAnnouncement {
use crate::nostr::events::RepositoryAnnouncement;
if let Ok(announcement) = RepositoryAnnouncement::from_event((*event).clone()) {
// Re-process rejected state events for this announcement
let (removed, hot_events) = rejected_events_index.invalidate_and_get(
&event.pubkey,
&announcement.identifier,
Some(rejected_index::EventType::State),
);
if removed > 0 {
tracing::info!(
pubkey = %event.pubkey,
identifier = %announcement.identifier,
removed_from_cold_index = removed,
hot_cache_events = hot_events.len(),
"Invalidated rejected state events (announcement now in purgatory)"
);
}
// Re-process state events from hot cache immediately
if !hot_events.is_empty() {
let _stats = Self::reprocess_events_from_hot_cache(
hot_events,
"state event (announcement in purgatory)",
&event.pubkey,
&announcement.identifier,
&relay_url_clone,
&database,
&write_policy,
&local_relay,
&rejected_events_index,
)
.await;
}
}
}
// State events (kind 30618) - extract identifier and trigger immediate sync
if event.kind.as_u16() == 30618 {
else if event.kind.as_u16() == 30618 {
if let Some(identifier) = event.tags.iter().find_map(|tag| {
let tag_vec = tag.clone().to_vec();
if tag_vec.len() >= 2 && tag_vec[0] == "d" {
@@ -1754,7 +1796,9 @@ impl SyncManager {
// Track pagination state for this subscription (REQ+EOSE)
// and received event IDs for negentropy batches
if result == ProcessResult::Saved || result == ProcessResult::Duplicate {
// Include Purgatory results so announcements in purgatory still trigger
// per-repo sync (state events, PR events) from the source relay.
if result == ProcessResult::Saved || result == ProcessResult::Duplicate || result == ProcessResult::Purgatory {
let mut pending = pending_sync_index.write().await;
if let Some(batches) = pending.get_mut(&relay_url_clone) {
for batch in batches.iter_mut() {
@@ -2506,6 +2550,26 @@ impl SyncManager {
"{} added to purgatory (waiting for git data)",
context
);
// Trigger immediate sync for re-processed events that go to purgatory
// (same as sync-triggered events in the main event loop)
if event.kind.as_u16() == 30618 {
// State event - extract identifier from 'd' tag
if let Some(id) = event.tags.iter().find_map(|tag| {
let tag_vec = tag.clone().to_vec();
if tag_vec.len() >= 2 && tag_vec[0] == "d" {
Some(tag_vec[1].clone())
} else {
None
}
}) {
write_policy.purgatory().enqueue_sync_immediate(&id);
}
} else if event.kind.as_u16() == 1617 || event.kind.as_u16() == 1618 {
// PR event - extract identifier from 'a' tag
if let Some(id) = crate::git::sync::extract_identifier_from_pr_event(&event) {
write_policy.purgatory().enqueue_sync_immediate(&id);
}
}
}
ProcessResult::Rejected => {
stats.rejected += 1;
tests/archive_read_only.rs
+45 -14
View File
@@ -165,6 +165,7 @@ async fn test_archive_read_only_creates_bare_repo() {
// c) Put state event in purgatory (git data missing on archive relay)
// d) Fetch git data from source relay's clone URL
// e) Release the state event from purgatory
let found = wait_for_event_served(
archive_relay.url(),
&state_event_id,
@@ -267,11 +268,13 @@ async fn test_archive_read_only_creates_bare_repo() {
/// This verifies the security model: archive mode only syncs git data
/// when there are state events to validate against.
///
/// Scenario:
/// 1. Start source relay with announcement only (no state events)
/// 2. Start archive relay syncing from source
/// 3. Archive relay syncs announcement (creates bare repo)
/// 4. Verify git data is NOT synced (no state events to trigger purgatory sync)
/// With announcement purgatory, the flow is:
/// 1. Send announcement to source relay (goes to purgatory)
/// 2. Send state event to source relay (goes to purgatory)
/// 3. Push git data to source relay (promotes announcement and state event)
/// 4. Start archive relay with sync from source
/// 5. Archive relay syncs the promoted announcement
/// 6. Verify git data is NOT synced (archive has no state event to authorize git fetch)
#[tokio::test]
async fn test_archive_without_state_events_does_not_sync_git() {
// 1. Start source relay
@@ -290,7 +293,7 @@ async fn test_archive_without_state_events_does_not_sync_git() {
let npub = keys.public_key().to_bech32().expect("Failed to get npub");
// 3. Create and send announcement listing BOTH relays (but NO state event)
// 3. Create and send announcement listing BOTH relays
let announcement = create_repo_announcement(
&keys,
&[&source_relay.domain(), &archive_domain],
@@ -306,7 +309,7 @@ async fn test_archive_without_state_events_does_not_sync_git() {
tokio::time::sleep(Duration::from_millis(500)).await;
// Send announcement to source relay
// Send announcement to source relay (goes to purgatory)
source_client
.send_event(&announcement)
.await
@@ -314,11 +317,39 @@ async fn test_archive_without_state_events_does_not_sync_git() {
tokio::time::sleep(Duration::from_millis(200)).await;
// 4. Push git data to source relay (but no state event to authorize it)
// This push will fail because there's no state event in purgatory
// That's expected - we're testing that archive mode doesn't blindly fetch git data
// 4. Create and send state event to source relay (goes to purgatory)
let clone_url = format!(
"http://{}/{}/{}.git",
source_relay.domain(),
npub,
identifier
);
let relay_url = source_relay.url().to_string();
// 5. Start archive relay
let state_event = create_state_event(
&keys,
identifier,
&[("main", &commit_hash)],
&[],
&[&clone_url],
&[&relay_url],
)
.expect("Failed to create state event");
source_client
.send_event(&state_event)
.await
.expect("Failed to send state event to source");
tokio::time::sleep(Duration::from_millis(200)).await;
// 5. Push git data to source relay (promotes announcement and state event)
push_to_relay(temp_dir.path(), &source_relay.domain(), &npub, identifier)
.expect("Push to source should succeed");
tokio::time::sleep(Duration::from_millis(500)).await;
// 6. Start archive relay (without state event - we don't send state event to archive)
let archive_relay = TestRelay::start_with_archive_and_sync(
archive_port,
Some(source_relay.url().to_string()),
@@ -333,10 +364,10 @@ async fn test_archive_without_state_events_does_not_sync_git() {
.await
.expect("Sync connection should establish");
// Give time for any potential git sync to happen
// Give time for sync to fetch announcement
tokio::time::sleep(Duration::from_secs(3)).await;
// 6. Verify bare repository was created (announcement was accepted)
// 7. Verify bare repository was created (announcement was synced and accepted to purgatory)
let repo_path = archive_relay
.git_data_path()
.join(format!("{}/{}.git", npub, identifier));
@@ -346,7 +377,7 @@ async fn test_archive_without_state_events_does_not_sync_git() {
"Bare repository should be created for archive announcement"
);
// 7. Verify git data was NOT synced (no state events to trigger purgatory sync)
// 8. Verify git data was NOT synced (no state events on archive to trigger git fetch)
// Check that the commit does NOT exist in the archive relay's repo
let output = tokio::process::Command::new("git")
.args(["cat-file", "-t", &commit_hash])
tests/purgatory.rs
+2 -2
View File
@@ -58,10 +58,10 @@ macro_rules! isolated_purgatory_test {
}
// ============================================================
// Announcement Purgatory Tests (commented out - feature not yet implemented)
// Announcement Purgatory Tests
// ============================================================
// isolated_purgatory_test!(test_announcement_not_served_before_git_data);
isolated_purgatory_test!(test_announcement_not_served_before_git_data);
isolated_purgatory_test!(test_announcement_served_after_git_push);
isolated_purgatory_test!(test_bare_repo_exists_for_purgatory_announcement);
isolated_purgatory_test!(test_state_event_accepted_for_purgatory_announcement);
tests/purgatory_persistence.rs
+14 -12
View File
@@ -120,7 +120,8 @@ async fn test_full_purgatory_save_restore_cycle() {
// so we'll focus on testing state and PR events persistence
// Verify initial counts
let (state_count, pr_count) = purgatory.count();
let (announcement_count, state_count, pr_count) = purgatory.count();
assert_eq!(announcement_count, 0, "Should have 0 announcements");
assert_eq!(state_count, 2, "Should have 2 state events");
assert_eq!(
pr_count, 3,
@@ -142,7 +143,8 @@ async fn test_full_purgatory_save_restore_cycle() {
);
// Verify all data was restored
let (state_count2, pr_count2) = purgatory2.count();
let (announcement_count2, state_count2, pr_count2) = purgatory2.count();
assert_eq!(announcement_count2, 0, "Should have 0 announcements after restore");
assert_eq!(state_count2, 2, "Should have 2 state events after restore");
assert_eq!(
pr_count2, 3,
@@ -275,7 +277,7 @@ async fn test_purgatory_downtime_adjustment() {
purgatory2.restore_from_disk(&state_path).unwrap();
// Verify event is still there (downtime was accounted for)
let (state_count, _) = purgatory2.count();
let (_, state_count, _) = purgatory2.count();
assert_eq!(state_count, 1);
let repo1_states = purgatory2.find_state("repo1");
@@ -401,7 +403,7 @@ async fn test_purgatory_restore_missing_file() {
assert!(result.is_err(), "Should error on missing file");
// Purgatory should still be usable (empty state)
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
@@ -410,7 +412,7 @@ async fn test_purgatory_restore_missing_file() {
let event = create_test_event(&keys, "test").await;
purgatory.add_state(event, "repo1".to_string(), keys.public_key());
let (state_count, _) = purgatory.count();
let (_, state_count, _) = purgatory.count();
assert_eq!(state_count, 1);
}
@@ -461,7 +463,7 @@ async fn test_purgatory_restore_corrupted_file() {
assert!(result.is_err(), "Should error on corrupted file");
// Purgatory should still be usable
let (state_count, pr_count) = purgatory.count();
let (_, state_count, pr_count) = purgatory.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
}
@@ -504,7 +506,7 @@ async fn test_empty_purgatory_save_restore() {
purgatory2.restore_from_disk(&state_path).unwrap();
// Verify empty state
let (state_count, pr_count) = purgatory2.count();
let (_, state_count, pr_count) = purgatory2.count();
assert_eq!(state_count, 0);
assert_eq!(pr_count, 0);
assert_eq!(purgatory2.expired_count(), 0);
@@ -591,7 +593,7 @@ async fn test_purgatory_continues_working_after_restore() {
purgatory2.add_state(event2.clone(), "repo2".to_string(), keys.public_key());
// Verify both old and new events work
let (state_count, _) = purgatory2.count();
let (_, state_count, _) = purgatory2.count();
assert_eq!(state_count, 2);
let repo1_states = purgatory2.find_state("repo1");
@@ -603,7 +605,7 @@ async fn test_purgatory_continues_working_after_restore() {
assert_eq!(repo2_states[0].event.id, event2.id);
// Verify cleanup still works
let (state_removed, pr_removed) = purgatory2.cleanup();
let (_, state_removed, pr_removed) = purgatory2.cleanup();
// Nothing should be expired yet
assert_eq!(state_removed, 0);
assert_eq!(pr_removed, 0);
@@ -684,15 +686,15 @@ async fn test_purgatory_entries_expired_during_downtime() {
purgatory2.restore_from_disk(&state_path).unwrap();
// Event should be restored
let (state_count, _) = purgatory2.count();
let (_, state_count, _) = purgatory2.count();
assert_eq!(state_count, 1);
// Cleanup should work (even if nothing is expired yet)
let (state_removed, _) = purgatory2.cleanup();
let (_, state_removed, _) = purgatory2.cleanup();
// Nothing expired yet since we didn't wait 30 minutes
assert_eq!(state_removed, 0);
let (state_count, _) = purgatory2.count();
let (_, state_count, _) = purgatory2.count();
assert_eq!(state_count, 1);
}