From 7a57640312bc19599b52017f3c331e4f96dfd8b8 Mon Sep 17 00:00:00 2001 From: John Tennant Date: Tue, 1 Sep 2026 12:38:42 -0400 Subject: [PATCH] Publish collaborative project state Signed-off-by: John Tennant --- crates/buzz-db/src/store/project_state.rs | 10 +- crates/buzz-relay/src/handlers/mod.rs | 2 + .../buzz-relay/src/handlers/project_change.rs | 15 ++ .../src/handlers/project_state_projection.rs | 195 ++++++++++++++++++ crates/buzz-relay/src/main.rs | 22 ++ 5 files changed, 242 insertions(+), 2 deletions(-) create mode 100644 crates/buzz-relay/src/handlers/project_state_projection.rs diff --git a/crates/buzz-db/src/store/project_state.rs b/crates/buzz-db/src/store/project_state.rs index c428069544..f277bba874 100644 --- a/crates/buzz-db/src/store/project_state.rs +++ b/crates/buzz-db/src/store/project_state.rs @@ -490,7 +490,7 @@ impl Db { let project_owner: Vec = coordinate.try_get("project_owner")?; let project_d_tag: String = coordinate.try_get("project_d_tag")?; if let Some(candidate) = self - .load_project_state_projection_candidate( + .load_pending_project_state_projection( community_id, &project_owner, &project_d_tag, @@ -504,7 +504,13 @@ impl Db { Ok(candidates) } - async fn load_project_state_projection_candidate( + /// Load one coherent Project state when it requires publication by + /// `projection_pubkey`. + /// + /// The returned candidate is safe to sign outside the transaction because + /// [`Self::commit_project_state_projection`] revalidates every observed + /// head field while holding the same Project coordinate lock. + pub async fn load_pending_project_state_projection( &self, community_id: CommunityId, project_owner: &[u8], diff --git a/crates/buzz-relay/src/handlers/mod.rs b/crates/buzz-relay/src/handlers/mod.rs index 8ac38cbafb..e8ff9ec4b7 100644 --- a/crates/buzz-relay/src/handlers/mod.rs +++ b/crates/buzz-relay/src/handlers/mod.rs @@ -28,6 +28,8 @@ pub mod moderation_notices; pub mod product_feedback; /// NIP-PC collaborative Project change validation and execution. pub mod project_change; +/// Relay signing and durable repair for NIP-PC Project State projections. +pub mod project_state_projection; #[allow(dead_code, missing_docs)] pub mod push_lease; /// NIP-43 relay membership admin command handler (kinds 9030–9032). diff --git a/crates/buzz-relay/src/handlers/project_change.rs b/crates/buzz-relay/src/handlers/project_change.rs index 7c0b176b15..8cedd3f965 100644 --- a/crates/buzz-relay/src/handlers/project_change.rs +++ b/crates/buzz-relay/src/handlers/project_change.rs @@ -207,6 +207,21 @@ pub(crate) async fn handle( ProjectChangeApplyResult::InvalidMutation(message) | ProjectChangeApplyResult::InvalidBase(message) => return Err(invalid(message)), }; + if let Err(error) = super::project_state_projection::publish_project_state_for_coordinate( + tenant, + state, + &parsed.owner, + &parsed.d_tag, + ) + .await + { + tracing::warn!( + project_owner = %hex::encode(&parsed.owner), + project_d_tag = %parsed.d_tag, + %error, + "accepted Project change awaits projection repair" + ); + } Ok(IngestResult { event_id: event.id.to_hex(), accepted: true, diff --git a/crates/buzz-relay/src/handlers/project_state_projection.rs b/crates/buzz-relay/src/handlers/project_state_projection.rs new file mode 100644 index 0000000000..e715454fe7 --- /dev/null +++ b/crates/buzz-relay/src/handlers/project_state_projection.rs @@ -0,0 +1,195 @@ +//! Relay signing and durable repair for NIP-PC Project State projections. + +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{anyhow, Context}; +use buzz_core::event::StoredEvent; +use buzz_core::kind::KIND_PROJECT_STATE; +use buzz_core::tenant::TenantContext; +use buzz_db::project_state::{ProjectStateProjectionCandidate, ProjectStateProjectionCommitResult}; +use nostr::{Event, EventBuilder, Timestamp}; +use tokio_util::sync::CancellationToken; + +use crate::state::AppState; + +use super::event::dispatch_persistent_event; + +const RECONCILE_BATCH_SIZE: i64 = 100; +const RECONCILE_INTERVAL: Duration = Duration::from_secs(60); + +fn projection_created_at(now: u64, previous: Option) -> anyhow::Result { + let after_previous = previous + .map(|value| { + value + .checked_add(1) + .ok_or_else(|| anyhow!("Project projection timestamp overflow")) + }) + .transpose()?; + Ok(after_previous.map_or(now, |value| now.max(value))) +} + +fn sign_candidate( + candidate: &ProjectStateProjectionCandidate, + relay_keypair: &nostr::Keys, +) -> anyhow::Result { + let template = candidate.template(); + let created_at = + projection_created_at(Timestamp::now().as_secs(), candidate.previous_created_at())?; + EventBuilder::new(template.kind, template.content.clone()) + .tags(template.tags.clone()) + .custom_created_at(Timestamp::from(created_at)) + .sign_with_keys(relay_keypair) + .context("sign Project State projection") +} + +async fn publish_candidate( + tenant: &TenantContext, + state: &Arc, + candidate: ProjectStateProjectionCandidate, +) -> anyhow::Result { + if candidate.community_id() != tenant.community() { + return Err(anyhow!("Project projection candidate crossed communities")); + } + let event = sign_candidate(&candidate, &state.relay_keypair)?; + let result = state + .db + .commit_project_state_projection(&candidate, &event) + .await + .context("commit Project State projection")?; + if result == ProjectStateProjectionCommitResult::Stale { + return Ok(false); + } + + let stored = StoredEvent::new(event, None); + let relay_pubkey = state.relay_keypair.public_key().to_hex(); + dispatch_persistent_event( + tenant, + state, + &stored, + KIND_PROJECT_STATE, + &relay_pubkey, + None, + ) + .await; + Ok(true) +} + +/// Publish the current projection for one accepted Project command, if pending. +pub(crate) async fn publish_project_state_for_coordinate( + tenant: &TenantContext, + state: &Arc, + project_owner: &[u8], + project_d_tag: &str, +) -> anyhow::Result { + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + let Some(candidate) = state + .db + .load_pending_project_state_projection( + tenant.community(), + project_owner, + project_d_tag, + &relay_pubkey, + ) + .await + .context("load pending Project State projection")? + else { + return Ok(false); + }; + publish_candidate(tenant, state, candidate).await +} + +/// Repair a bounded batch of durable Project State publication markers. +pub async fn reconcile_project_state_projections(state: &Arc) -> anyhow::Result { + let relay_pubkey = state.relay_keypair.public_key().to_bytes(); + let candidates = state + .db + .load_pending_project_state_projections(&relay_pubkey, RECONCILE_BATCH_SIZE) + .await + .context("load pending Project State projections")?; + let mut committed = 0; + for candidate in candidates { + let community_id = candidate.community_id(); + let result = async { + let host = state + .db + .lookup_community_host(community_id) + .await? + .ok_or_else(|| anyhow!("Project projection community has no active host"))?; + let tenant = TenantContext::resolved(community_id, host); + publish_candidate(&tenant, state, candidate).await + } + .await; + match result { + Ok(true) => committed += 1, + Ok(false) => {} + Err(error) => { + tracing::warn!(%community_id, %error, "Project State projection repair failed") + } + } + } + Ok(committed) +} + +/// Run periodic bounded repair until graceful shutdown is requested. +pub fn spawn_project_state_projection_reconciler( + state: Arc, + cancel: CancellationToken, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + match reconcile_project_state_projections(&state).await { + Ok(count) if count > 0 => { + tracing::info!(count, "Project State projections repaired on startup") + } + Ok(_) => {} + Err(error) => { + tracing::warn!(%error, "Project State startup reconciliation failed") + } + } + + let mut interval = tokio::time::interval(RECONCILE_INTERVAL); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval.tick().await; + loop { + tokio::select! { + _ = cancel.cancelled() => break, + _ = interval.tick() => { + match reconcile_project_state_projections(&state).await { + Ok(count) if count > 0 => { + tracing::info!(count, "Project State projections repaired") + } + Ok(_) => {} + Err(error) => tracing::warn!(%error, "Project State projection reconciliation failed"), + } + } + } + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn projection_timestamp_is_strictly_monotonic() { + assert_eq!(projection_created_at(100, None).expect("timestamp"), 100); + assert_eq!( + projection_created_at(100, Some(50)).expect("timestamp"), + 100 + ); + assert_eq!( + projection_created_at(100, Some(100)).expect("timestamp"), + 101 + ); + assert_eq!( + projection_created_at(100, Some(200)).expect("timestamp"), + 201 + ); + } + + #[test] + fn projection_timestamp_rejects_overflow() { + assert!(projection_created_at(100, Some(u64::MAX)).is_err()); + } +} diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index bb8715508e..03db167f32 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -467,6 +467,16 @@ async fn main() -> anyhow::Result<()> { ); let state = Arc::new(app_state); + // Project changes commit independently of their relay-signed read model. + // Start best-effort repair immediately without making listener availability + // depend on it, then retry a bounded batch periodically. + let project_state_projection_cancel = CancellationToken::new(); + let mut project_state_projection_task = + buzz_relay::handlers::project_state_projection::spawn_project_state_projection_reconciler( + Arc::clone(&state), + project_state_projection_cancel.clone(), + ); + // Inter-relay mesh (BUZZ_MESH seam). `boot_mesh` returns None when the // kill switch is off — nothing is bound, published, or spawned, so the // relay behaves byte-identically to a build without the mesh. When @@ -1151,6 +1161,18 @@ async fn main() -> anyhow::Result<()> { serve(router, health_router, Arc::clone(&state)).await?; state.community_revalidator_cancel.cancel(); + project_state_projection_cancel.cancel(); + if tokio::time::timeout( + std::time::Duration::from_secs(5), + &mut project_state_projection_task, + ) + .await + .is_err() + { + project_state_projection_task.abort(); + let _ = project_state_projection_task.await; + tracing::warn!("Project State projection reconciler did not stop within the drain window"); + } // Signal the audit worker to stop accepting, flush buffered entries, and // exit. Uses a CancellationToken so it works regardless of how many