From 6a86531ac9d27b5123cca14636ced581578975df Mon Sep 17 00:00:00 2001 From: Wes Date: Mon, 21 Sep 2026 11:30:03 -0700 Subject: [PATCH] feat(relay): add atomic complete read-state snapshots (#7572) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Opened by Brain on behalf of Wes (GitHub: wesbillman). ## Summary Add an opt-in, host-scoped completeness primitive for NIP-RS read-state clients. Ordinary query arrays and WebSocket behavior remain unchanged. - Advertise a version-1 `read_state_snapshot` capability in NIP-11 only when the request resolves a community. - Accept exactly one self-author kind-30078 snapshot filter on authenticated `POST /query`, after the existing admission, replay and membership gates. - Read all retained, nondeleted self-author kind-30078 coordinates from the writer at one SQL-statement MVCC cut—including unrelated application coordinates and events without a read-state tag. - Return a complete versioned envelope bound to community, signer and a content-identity hash. Reject malformed contracts, reconstruction/signature/storage failures and whole-request overflow; never return a truncated complete response. - Bound snapshots at 4,096 events and 8 MiB of compact encoded event-array bytes, advertised as `max_event_array_bytes`; envelope metadata is additional. Retain the separate stored-payload preflight budget. ## Safety and scope This is point-in-time enumeration, **not** a live freshness guarantee, cursor, CAS token, cross-relay transaction, or unread total. NIP-RS now explicitly requires fresh complete enumeration from every write relay before each override action, canonical publication, carry-forward or deletion. It does not enable synchronized manual-unread overrides in any client. No schema migration or deployment is included. ## Validation Executed against the source committed as `1fd5a5e0c0d893c1259386885517c0c44ae7630f`: - Four real-router/PostgreSQL snapshot cases: beyond-page-cap enumeration and overflow; corruption/storage failure; production payload-bound NIP-98, Redis replay guard and membership enforcement; host discovery/scope/strict request shape. - Two database cases: replacement during preflight preserves a single statement cut; encoded-byte overflow and writer-not-replica routing. The cluster-wide probe uses the required `cluster_global_` serialization prefix. - Production-call-site mutations killed for dispatch, strict filter validation, discovery tenant, snapshot tenant, encoded-byte guard, membership enforcement and replay enforcement; isolated mutation tree restored. - Both affected crates pass all-target Clippy; formatting, differential file-size and repository PostgreSQL test-discovery checks pass. - Normal pre-push gates pass without bypass: branch skew, push-head scope, file size, Rust unit lanes and desktop Tauri checks. - Independent read-only review by Pinky found no remaining relay blocker at this commit. The execution/mutation results above were run by Brain, not duplicated by the reviewer. Hosted CI and maintainer review remain separate gates. No production traffic or deployment was used for these fixture tests. Originating Buzz channel: `a394ecdb-9c67-4c4e-a3c6-4a95f69e8cc8` (unread-but-good), thread `2413d0ba7fa5b2ac5c777df6b3754e2099b042ec8fb96ca64302a0919a0550b3`. ## Review repair (2026-09-21) Carl updated this existing branch to `2f9c9b11266c24947412cf28cefae4e96e7167b7`, merging main `5079c770fe30bb3d8204822ce6c2431eacac6d4b` without rewriting published history. - Resolved the DB export conflict, preserving both snapshot and storage-accounting exports. - Addressed the blocking review using its accepted alternative: rename `max_bytes` to `max_event_array_bytes` and explicitly document that success-envelope metadata is outside the array budget. No auth, query, runtime-limit, or client-behavior redesign. - Added a PostgreSQL/router regression proving discovery, exact 8,388,608-byte array success with a larger envelope, and one-byte-over rejection (413, no partial complete response). Exact-head validation: all seven snapshot PostgreSQL/router tests passed in an isolated local database; both affected crates passed all-target Clippy; mandatory pre-commit/pre-push gates passed without bypass (Rust unit lanes, Tauri default/mesh Clippy and tests, differential file size, branch-skew/ref checks). Reverting only the descriptor rename made the new regression fail before the final commit. Independent bounded source-delta review by Mordecai found no blockers; execution was Carl's. A broader `cargo test -p buzz-db -p buzz-relay --all-targets` run on the merged candidate passed 131 DB tests and 1,057 relay tests, then encountered the same unrelated `mesh_demo::demo_join_forwarded_arm_round_trips_echo` 504 recorded in the prior review. Full repository `just ci` / `just test` were not rerun locally; hosted CI remains a separate gate. No deployment or production traffic was exercised. --------- Signed-off-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz> Signed-off-by: Carl <32a2e2c9d428ee08902cab75d956da2c1d235a22d4766b0dd4138bf6e2e5db1d@buzz.block.builderlab.xyz> Co-authored-by: Brain <1a02c72794dcd0f07058a353bc3a81f4028b8c77c92c87fce6d5c8b85970a20b@buzz.block.builderlab.xyz> Co-authored-by: Carl <32a2e2c9d428ee08902cab75d956da2c1d235a22d4766b0dd4138bf6e2e5db1d@buzz.block.builderlab.xyz> --- crates/buzz-db/src/error.rs | 4 + crates/buzz-db/src/lib.rs | 4 +- crates/buzz-db/src/store/mod.rs | 2 + crates/buzz-db/src/store/read_state.rs | 276 +++++++++++ crates/buzz-relay/src/api/bridge.rs | 8 +- .../src/api/bridge/read_state_snapshot.rs | 442 ++++++++++++++++++ crates/buzz-relay/src/nip11.rs | 12 + docs/nips/NIP-RS.md | 95 +++- 8 files changed, 838 insertions(+), 5 deletions(-) create mode 100644 crates/buzz-db/src/store/read_state.rs create mode 100644 crates/buzz-relay/src/api/bridge/read_state_snapshot.rs diff --git a/crates/buzz-db/src/error.rs b/crates/buzz-db/src/error.rs index 4f4e6b105c..79a371acb4 100644 --- a/crates/buzz-db/src/error.rs +++ b/crates/buzz-db/src/error.rs @@ -64,6 +64,10 @@ pub enum DbError { #[error("deletion safety error: {0}")] DeletionSafety(String), + /// A complete read-state snapshot exceeds its bounded resource budget. + #[error("read-state snapshot exceeds event or byte limit")] + ReadStateSnapshotTooLarge, + /// A stored timestamp value could not be interpreted. #[error("invalid timestamp: {0}")] InvalidTimestamp(i64), diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index d444d9ce52..f749d1a625 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -64,8 +64,8 @@ pub(crate) use runtime::{ pub use store::{ admin_moderation, allowlist, api_token, archived_identities, channel, channel_members, community, deletion, dm, event, feed, git_repo, moderation, partition, product_feedback, push, - reaction, relay_admin_actions, relay_invite, relay_members, relay_operators, reminder, - replaceable, storage_accounting, thread, usage, user, workflow, + reaction, read_state, relay_admin_actions, relay_invite, relay_members, relay_operators, + reminder, replaceable, storage_accounting, thread, usage, user, workflow, }; pub use allowlist::AllowlistEntry; diff --git a/crates/buzz-db/src/store/mod.rs b/crates/buzz-db/src/store/mod.rs index 79739373d0..3aeb2257aa 100644 --- a/crates/buzz-db/src/store/mod.rs +++ b/crates/buzz-db/src/store/mod.rs @@ -34,6 +34,8 @@ pub mod product_feedback; pub mod push; /// Reaction persistence. pub mod reaction; +/// Complete own-author read-state snapshots. +pub mod read_state; /// HTTP report-resolution enforcement state machine persistence. pub mod relay_admin_actions; /// Use-limited relay invite persistence (v2 opaque tokens). diff --git a/crates/buzz-db/src/store/read_state.rs b/crates/buzz-db/src/store/read_state.rs new file mode 100644 index 0000000000..5a8a2a2c6c --- /dev/null +++ b/crates/buzz-db/src/store/read_state.rs @@ -0,0 +1,276 @@ +//! Atomic, bounded, writer-only snapshots of an author's kind-30078 state. + +use buzz_core::{kind::KIND_READ_STATE, CommunityId}; +use serde_json::Value; +use sha2::{Digest, Sha256}; +use sqlx::Row; + +use crate::{observability, Db, DbError, Result}; + +/// Maximum number of events in a complete snapshot; overflow is an error. +pub const MAX_SNAPSHOT_EVENTS: usize = 4096; +/// Maximum stored payload or encoded event-array bytes; overflow is an error. +pub const MAX_SNAPSHOT_BYTES: usize = 8 * 1024 * 1024; + +/// A complete view at one writer statement's MVCC cut, not a live-delivery fence. +#[derive(Debug)] +pub struct ReadStateSnapshot { + /// Content identifier bound to community, author, and ordered event ids. + /// Not a monotone revision, CAS token, or live-stream cursor. + pub snapshot_id: String, + /// Every current own kind-30078 event, including unrelated application data. + pub events: Vec, +} + +impl Db { + /// Load all current kind-30078 events for exactly one author and community. + /// + /// A single writer statement binds selection, resource preflight, and row + /// retrieval to the same MVCC snapshot. No tag, horizon, replica, or ordinary + /// query-page cap is involved. Any corrupt row or budget overflow fails the + /// whole operation before a caller can assert completeness. + pub async fn read_state_snapshot( + &self, + community: CommunityId, + author: &nostr::PublicKey, + ) -> Result { + let mut conn = observability::acquire_writer( + &self.pool, + observability::WriterOperation::SubscriptionHistory, + ) + .await?; + // Select only bounded metadata first. A left join retains the budget + // sentinel even for empty/oversized sets; oversized payloads never leave + // Postgres. Both CTEs and the payload join share ONE statement snapshot, + // including when a coordinate is hard-deleted/replaced concurrently. + let rows = sqlx::query( + "WITH candidates AS MATERIALIZED ( + SELECT id, created_at, + octet_length(content)::bigint + octet_length(tags::text)::bigint AS bytes + FROM events + WHERE community_id = $1 AND pubkey = $2 AND kind = $3 + AND deleted_at IS NULL + ORDER BY created_at DESC, id ASC LIMIT $4 + ), budget AS MATERIALIZED ( + SELECT count(*) AS n, COALESCE(sum(bytes), 0)::bigint AS bytes FROM candidates + ) + SELECT budget.n AS snapshot_count, budget.bytes AS snapshot_bytes, + e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, e.sig, + e.received_at, e.channel_id + FROM budget + LEFT JOIN candidates c ON budget.n <= $5 AND budget.bytes <= $6 + LEFT JOIN events e ON e.community_id = $1 AND e.id = c.id + AND e.created_at = c.created_at + ORDER BY e.created_at DESC, e.id ASC", + ) + .bind(community.as_uuid()) + .bind(author.to_bytes().to_vec()) + .bind(KIND_READ_STATE as i32) + .bind((MAX_SNAPSHOT_EVENTS + 1) as i64) + .bind(MAX_SNAPSHOT_EVENTS as i64) + .bind(MAX_SNAPSHOT_BYTES as i64) + .fetch_all(&mut *conn) + .await?; + + let first = rows.first().ok_or_else(|| { + DbError::InvalidData("read-state snapshot missing budget sentinel".into()) + })?; + let count: i64 = first.try_get("snapshot_count")?; + let stored_bytes: i64 = first.try_get("snapshot_bytes")?; + if count > MAX_SNAPSHOT_EVENTS as i64 || stored_bytes > MAX_SNAPSHOT_BYTES as i64 { + return Err(DbError::ReadStateSnapshotTooLarge); + } + let mut events = Vec::with_capacity(count as usize); + let mut encoded_bytes = 2usize; // JSON array brackets + let mut hash = Sha256::new(); + hash.update(b"buzz-read-state-snapshot-v1\0"); + hash.update(community.as_uuid().as_bytes()); + hash.update(author.to_bytes()); + for row in rows { + if count == 0 { + break; + } + let stored = super::event::row_to_stored_event(row)?.ok_or_else(|| { + DbError::InvalidData("unreadable read-state snapshot event".into()) + })?; + let event = stored.event; + if event.pubkey != *author || event.kind.as_u16() as u32 != KIND_READ_STATE { + return Err(DbError::InvalidData( + "read-state snapshot scope mismatch".into(), + )); + } + event.verify().map_err(|_| { + DbError::InvalidData("invalid signed read-state snapshot event".into()) + })?; + let value = serde_json::to_value(&event)?; + encoded_bytes += serde_json::to_vec(&value)?.len() + usize::from(!events.is_empty()); + if encoded_bytes > MAX_SNAPSHOT_BYTES { + return Err(DbError::ReadStateSnapshotTooLarge); + } + hash.update(event.id.as_bytes()); + events.push(value); + } + if events.len() != count as usize { + return Err(DbError::InvalidData( + "read-state snapshot row count mismatch".into(), + )); + } + Ok(ReadStateSnapshot { + snapshot_id: hex::encode(hash.finalize()), + events, + }) + } +} + +#[cfg(test)] +mod postgres_tests { + use super::*; + use nostr::{EventBuilder, Keys, Kind, Tag, Timestamp}; + use sqlx::{postgres::PgPoolOptions, PgPool}; + use std::time::Duration; + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn cluster_global_read_state_snapshot_replacement_during_preflight_keeps_one_mvcc_cut() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let db = Db::from_pool(pool.clone()); + let community = db + .ensure_configured_community(&format!("mvcc-{}.local", uuid::Uuid::new_v4().simple())) + .await + .unwrap() + .id; + let key = Keys::generate(); + let slot = format!("read-state:{}", uuid::Uuid::new_v4().simple()); + let make = |ts, content| { + EventBuilder::new(Kind::Custom(30078), content) + .tags([Tag::identifier(&slot), Tag::hashtag("read-state")]) + .custom_created_at(Timestamp::from(ts)) + .sign_with_keys(&key) + .unwrap() + }; + let before = make(1789090000, "before"); + let after = make(1789090001, "after"); + db.replace_parameterized_event(community, &before, &slot, None) + .await + .unwrap(); + + // Gate the real production SQL at octet_length(content), after it has + // taken its statement snapshot. Only this connection's private schema + // shadows the builtin; the normal writer path is completely unchanged. + let schema = format!("snapshot_gate_{}", uuid::Uuid::new_v4().simple()); + sqlx::query(sqlx::AssertSqlSafe(format!("CREATE SCHEMA {schema}"))) + .execute(&pool) + .await + .unwrap(); + sqlx::query(sqlx::AssertSqlSafe(format!( + "CREATE FUNCTION {schema}.octet_length(text) RETURNS integer + LANGUAGE plpgsql VOLATILE AS $$ BEGIN + PERFORM pg_catalog.pg_advisory_xact_lock(81193042); + RETURN pg_catalog.octet_length($1); END $$" + ))) + .execute(&pool) + .await + .unwrap(); + let application = schema.clone(); + let gate_schema = schema.clone(); + let snapshot_pool = PgPoolOptions::new().max_connections(1).after_connect(move |conn, _| { + let setup = format!("SET search_path TO {gate_schema},pg_catalog,public; SET application_name='{gate_schema}'"); + Box::pin(async move { sqlx::raw_sql(sqlx::AssertSqlSafe(setup)).execute(conn).await?; Ok(()) }) + }).connect(&crate::test_support::database_url()).await.unwrap(); + let snapshot_db = Db::from_pool(snapshot_pool); + let mut blocker = pool.acquire().await.unwrap(); + sqlx::query("SELECT pg_advisory_lock(81193042)") + .execute(&mut *blocker) + .await + .unwrap(); + let reader = snapshot_db.clone(); + let pubkey = key.public_key(); + let pending = + tokio::spawn(async move { reader.read_state_snapshot(community, &pubkey).await }); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let waiting: bool = sqlx::query_scalar( + "SELECT EXISTS(SELECT 1 FROM pg_stat_activity + WHERE application_name=$1 AND wait_event='advisory')", + ) + .bind(&application) + .fetch_one(&pool) + .await + .unwrap(); + if waiting { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("snapshot must reach the real metadata read"); + assert!( + db.replace_parameterized_event(community, &after, &slot, None) + .await + .unwrap() + .1 + ); + sqlx::query("SELECT pg_advisory_unlock(81193042)") + .execute(&mut *blocker) + .await + .unwrap(); + let cut = pending.await.unwrap().unwrap(); + assert_eq!(cut.events, vec![serde_json::to_value(&before).unwrap()]); + let next = snapshot_db + .read_state_snapshot(community, &key.public_key()) + .await + .unwrap(); + assert_eq!(next.events, vec![serde_json::to_value(&after).unwrap()]); + assert_ne!(cut.snapshot_id, next.snapshot_id); + sqlx::query(sqlx::AssertSqlSafe(format!("DROP SCHEMA {schema} CASCADE"))) + .execute(&pool) + .await + .unwrap(); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn read_state_snapshot_encoded_bytes_bound_and_writer_not_replica() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let db = Db::from_pool(pool.clone()); + let community = db + .ensure_configured_community(&format!("bytes-{}.local", uuid::Uuid::new_v4().simple())) + .await + .unwrap() + .id; + let key = Keys::generate(); + // Stored bytes fit but escaped JSON event bytes do not. + let event = EventBuilder::new(Kind::Custom(30078), "\u{0001}".repeat(2 * 1024 * 1024)) + .tags([Tag::identifier("escape-fixture")]) + .sign_with_keys(&key) + .unwrap(); + db.insert_event(community, &event, None).await.unwrap(); + let dead_replica = PgPoolOptions::new() + .connect_lazy("postgres://unused@127.0.0.1:1/absent") + .unwrap(); + dead_replica.close().await; + let routed = Db::from_pools(pool.clone(), dead_replica); + assert!(matches!( + routed + .read_state_snapshot(community, &key.public_key()) + .await, + Err(DbError::ReadStateSnapshotTooLarge) + )); + sqlx::query("DELETE FROM events WHERE community_id=$1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .unwrap(); + let empty = routed + .read_state_snapshot(community, &key.public_key()) + .await + .unwrap(); + assert!(empty.events.is_empty()); + assert_eq!(empty.snapshot_id.len(), 64); + } +} diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 93e4f1769f..e7433d43b1 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -3,6 +3,8 @@ //! These endpoints provide HTTP access to the relay's Nostr protocol, //! authenticated via NIP-98 signed events. +mod read_state_snapshot; + use std::sync::Arc; use axum::{ @@ -1145,6 +1147,10 @@ async fn query_events_authed( )); } + if read_state_snapshot::requested(&raw_filters) { + return read_state_snapshot::query(state, tenant, &pubkey, &raw_filters).await; + } + // Get channels this user can access — same enforcement as WS REQ handler. let mut accessible_channels = state .get_accessible_channel_ids_cached(tenant.community(), &pubkey_bytes) @@ -3861,7 +3867,7 @@ mod postgres_tests { /// - Redis pool points at the local dev instance for the admission check. /// /// Returns `None` when local Postgres is not reachable. - async fn bridge_handler_test_state() -> Option> { + pub(super) async fn bridge_handler_test_state() -> Option> { let mut config = crate::config::Config::from_env().ok()?; config.database_url = crate::test_support::database_url(); // Use the real local Redis so enforce_http_admission can pass. diff --git a/crates/buzz-relay/src/api/bridge/read_state_snapshot.rs b/crates/buzz-relay/src/api/bridge/read_state_snapshot.rs new file mode 100644 index 0000000000..178c327eaa --- /dev/null +++ b/crates/buzz-relay/src/api/bridge/read_state_snapshot.rs @@ -0,0 +1,442 @@ +//! Opt-in atomic full-state query extension. Ordinary query arrays cannot +//! accidentally satisfy this envelope, even on relays that ignore extensions. + +use axum::{http::StatusCode, Json}; +use buzz_core::TenantContext; +use serde_json::{json, Value}; + +use crate::{api::api_error, state::AppState}; + +pub(super) fn requested(filters: &[Value]) -> bool { + filters + .iter() + .any(|f| f.get("read_state_snapshot").is_some()) +} + +pub(super) async fn query( + state: &AppState, + tenant: &TenantContext, + author: &nostr::PublicKey, + filters: &[Value], +) -> Result, (StatusCode, Json)> { + // Exact raw shape: never silently drop constraints or unknown versions. + let expected = json!({ + "kinds": [buzz_core::kind::KIND_READ_STATE], + "authors": [author.to_hex()], + "read_state_snapshot": 1, + }); + if filters != [expected] { + return Err(api_error( + StatusCode::BAD_REQUEST, + "read_state_snapshot requires one exact version-1 own-author kind-30078 filter", + )); + } + let snapshot = state + .db + .read_state_snapshot(tenant.community(), author) + .await + .map_err(|error| match error { + buzz_db::DbError::ReadStateSnapshotTooLarge => api_error( + StatusCode::PAYLOAD_TOO_LARGE, + "read_state_snapshot exceeds event or byte limit; cannot prove complete", + ), + _ => { + tracing::error!(%error, "read-state snapshot failed"); + api_error( + StatusCode::SERVICE_UNAVAILABLE, + "read_state_snapshot unavailable; cannot prove complete", + ) + } + })?; + Ok(Json(json!({ + "read_state_snapshot": 1, + "complete": true, + "community_id": tenant.community().as_uuid(), + "pubkey": author.to_hex(), + "snapshot_id": snapshot.snapshot_id, + "events": snapshot.events, + }))) +} + +#[cfg(test)] +mod postgres_tests { + use super::*; + use axum::{ + body::{to_bytes, Body}, + http::Request, + }; + use nostr::{EventBuilder, Keys, Kind, Tag}; + use std::sync::Arc; + use tower::ServiceExt; + + async fn request( + state: Arc, + host: &str, + key: Option<&Keys>, + body: Value, + ) -> (StatusCode, Value) { + let mut req = Request::builder() + .method("POST") + .uri("/query") + .header("host", host); + if let Some(key) = key { + req = req.header("x-pubkey", key.public_key().to_hex()); + } + let response = crate::router::build_router(state) + .oneshot( + req.body(Body::from(serde_json::to_vec(&body).unwrap())) + .unwrap(), + ) + .await + .unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 10 * 1024 * 1024) + .await + .unwrap(); + (status, serde_json::from_slice(&bytes).unwrap()) + } + + fn filter(key: &Keys) -> Value { + json!([{"kinds":[30078], "authors":[key.public_key().to_hex()], "read_state_snapshot":1}]) + } + + async fn setup() -> (Arc, String, buzz_core::CommunityId, Keys) { + let state = super::super::postgres_tests::bridge_handler_test_state() + .await + .expect("isolated Postgres/Redis fixture"); + let host = format!("snapshot-{}.local", uuid::Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .unwrap() + .id; + (state, host, community, Keys::generate()) + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn read_state_snapshot_router_real_auth_membership_and_replay() { + use base64::Engine; + use sha2::{Digest, Sha256}; + + let (fixture, host, community, key) = setup().await; + let mut state = (*fixture).clone(); + let config = Arc::make_mut(&mut state.config); + config.require_auth_token = true; + config.require_relay_membership = true; + // Restore the production Redis guard; never accept every replay in this test. + state.nip98_replay = Arc::new(buzz_pubsub::RedisNip98ReplayGuard::new( + state.redis_pool.clone(), + )); + let state = Arc::new(state); + let body = serde_json::to_vec(&filter(&key)).unwrap(); + let proof = |signed_host: &str| { + let auth = EventBuilder::new(Kind::Custom(27235), "") + .tags([ + Tag::parse(["u", &format!("https://{signed_host}/query")]).unwrap(), + Tag::parse(["method", "POST"]).unwrap(), + Tag::parse(["payload", &hex::encode(Sha256::digest(&body))]).unwrap(), + Tag::parse(["nonce", &uuid::Uuid::new_v4().to_string()]).unwrap(), + ]) + .sign_with_keys(&key) + .unwrap(); + format!( + "Nostr {}", + base64::engine::general_purpose::STANDARD + .encode(serde_json::to_vec(&auth).unwrap()) + ) + }; + let post = |authorization: String, bytes: Vec| { + let router = crate::router::build_router(state.clone()); + let req = Request::builder() + .method("POST") + .uri("/query") + .header("host", &host) + .header("authorization", authorization) + .body(Body::from(bytes)) + .unwrap(); + async move { + let response = router.oneshot(req).await.unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 1024 * 1024).await.unwrap(); + (status, serde_json::from_slice::(&bytes).unwrap()) + } + }; + let denied = post(proof(&host), body.clone()).await; + assert_eq!(denied.0, StatusCode::FORBIDDEN, "{}", denied.1); + assert_eq!(denied.1["error"], "relay_membership_required"); + assert_ne!(denied.1["complete"], true); + + state + .db + .add_relay_member(community, &key.public_key().to_hex(), "member", None) + .await + .unwrap(); + // Dev headers must not bypass real auth even for an admitted member. + assert_eq!( + request(state.clone(), &host, Some(&key), filter(&key)) + .await + .0, + StatusCode::UNAUTHORIZED + ); + assert_eq!( + post(proof("wrong-host.invalid"), body.clone()).await.0, + StatusCode::UNAUTHORIZED + ); + assert_eq!( + post(proof(&host), b"[]".to_vec()).await.0, + StatusCode::UNAUTHORIZED + ); + + let auth = proof(&host); + let accepted = post(auth.clone(), body.clone()).await; + assert_eq!(accepted.0, StatusCode::OK, "{}", accepted.1); + assert_eq!(accepted.1["complete"], true); + assert_eq!(accepted.1["community_id"], community.as_uuid().to_string()); + let replay = post(auth, body.clone()).await; + assert_eq!(replay.0, StatusCode::UNAUTHORIZED, "{}", replay.1); + assert_ne!(replay.1["complete"], true); + assert_eq!(post(proof(&host), body).await.0, StatusCode::OK); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn read_state_snapshot_router_scope_discovery_and_strict_contract() { + let (state, host, community, key) = setup().await; + let other = Keys::generate(); + let other_host = format!("other-{}.local", uuid::Uuid::new_v4().simple()); + let other_community = state + .db + .ensure_configured_community(&other_host) + .await + .unwrap() + .id; + let event = EventBuilder::new(Kind::Custom(30078), "encrypted") + .tags([Tag::identifier("unrelated-app-without-t-tag")]) + .sign_with_keys(&key) + .unwrap(); + state + .db + .replace_parameterized_event(community, &event, "unrelated-app-without-t-tag", None) + .await + .unwrap(); + let other_event = EventBuilder::new(Kind::Custom(30078), "other-author") + .tags([Tag::identifier("other")]) + .sign_with_keys(&other) + .unwrap(); + state + .db + .replace_parameterized_event(community, &other_event, "other", None) + .await + .unwrap(); + state + .db + .replace_parameterized_event( + other_community, + &event, + "unrelated-app-without-t-tag", + None, + ) + .await + .unwrap(); + + let (status, body) = request(state.clone(), &host, Some(&key), filter(&key)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["complete"], true); + assert_eq!(body["read_state_snapshot"], 1); + assert_eq!(body["community_id"], community.as_uuid().to_string()); + assert_eq!(body["pubkey"], key.public_key().to_hex()); + assert_eq!(body["events"], json!([event])); + let info = crate::nip11::nip11_document(&state, &host).await; + assert_eq!( + info.read_state_snapshot.as_ref().unwrap()["community_id"], + body["community_id"] + ); + assert!(crate::nip11::nip11_document(&state, "unmapped.invalid") + .await + .read_state_snapshot + .is_none()); + let (_, other_body) = request(state.clone(), &other_host, Some(&key), filter(&key)).await; + assert_eq!( + other_body["community_id"], + other_community.as_uuid().to_string() + ); + assert_ne!(other_body["snapshot_id"], body["snapshot_id"]); + let (status, _) = request(state.clone(), &host, None, filter(&key)).await; + assert_eq!(status, StatusCode::UNAUTHORIZED); + let (status, _) = + request(state.clone(), "unmapped.invalid", Some(&key), filter(&key)).await; + assert_eq!(status, StatusCode::NOT_FOUND); + let (status, _) = request(state.clone(), &host, Some(&other), filter(&key)).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + for (name, value) in [ + ("since", json!(0)), + ("limit", json!(1)), + ("#t", json!(["read-state"])), + ("read_state_snapshot", json!(2)), + ("read_state_snapshot", json!(false)), + ("read_state_snapshot", Value::Null), + ] { + let mut bad = filter(&key); + bad[0][name] = value; + let (status, body) = request(state.clone(), &host, Some(&key), bad).await; + assert_eq!(status, StatusCode::BAD_REQUEST, "{name}: {body}"); + assert_ne!(body["complete"], true); + } + let mut mixed = filter(&key); + mixed.as_array_mut().unwrap().push(json!({"kinds":[0]})); + assert_eq!( + request(state.clone(), &host, Some(&key), mixed).await.0, + StatusCode::BAD_REQUEST + ); + let ordinary = json!([{"kinds":[30078], "authors":[key.public_key().to_hex()]}]); + let (status, body) = request(state, &host, Some(&key), ordinary).await; + assert_eq!(status, StatusCode::OK); + assert!(body.is_array()); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn read_state_snapshot_router_corruption_and_storage_failure_are_not_complete() { + let (state, host, community, key) = setup().await; + let event = EventBuilder::new(Kind::Custom(30078), "payload") + .tags([Tag::identifier("fixture")]) + .sign_with_keys(&key) + .unwrap(); + state + .db + .replace_parameterized_event(community, &event, "fixture", None) + .await + .unwrap(); + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + for malformed in [json!(42), json!([["d", "fixture"]])] { + sqlx::query( + "UPDATE events SET tags=$1, content='corrupt' WHERE community_id=$2 AND id=$3", + ) + .bind(malformed) + .bind(community.as_uuid()) + .bind(event.id.as_bytes().as_slice()) + .execute(&pool) + .await + .unwrap(); + let (status, body) = request(state.clone(), &host, Some(&key), filter(&key)).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{body}"); + assert_ne!(body["complete"], true); + assert!(body.get("events").is_none()); + } + // Real DB error, not a mocked empty query: lock out the relation in this + // isolated per-test DB, while tenant/auth lookups remain usable. + sqlx::query("ALTER TABLE events RENAME TO snapshot_unavailable_events") + .execute(&pool) + .await + .unwrap(); + let (status, body) = request(state, &host, Some(&key), filter(&key)).await; + sqlx::query("ALTER TABLE snapshot_unavailable_events RENAME TO events") + .execute(&pool) + .await + .unwrap(); + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{body}"); + assert_ne!(body["complete"], true); + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn read_state_snapshot_router_advertised_event_array_byte_boundary() { + let (state, host, community, key) = setup().await; + let info = crate::nip11::nip11_document(&state, &host).await; + let descriptor = info.read_state_snapshot.unwrap(); + let budget = descriptor["max_event_array_bytes"].as_u64().unwrap() as usize; + assert_eq!(budget, buzz_db::read_state::MAX_SNAPSHOT_BYTES); + assert!(descriptor.get("max_bytes").is_none()); + + // ASCII content adds exactly one encoded byte per character; timestamp, + // event id and signature widths stay fixed across these signed events. + let make = |content: String, timestamp| { + EventBuilder::new(Kind::Custom(30078), content) + .tags([Tag::identifier("byte-boundary")]) + .custom_created_at(nostr::Timestamp::from(timestamp)) + .sign_with_keys(&key) + .unwrap() + }; + let overhead = serde_json::to_vec(&json!([make(String::new(), 1789090000)])) + .unwrap() + .len(); + for extra in 0..=1 { + let event = make( + "x".repeat(budget - overhead + extra), + 1789090001 + extra as u64, + ); + assert_eq!( + serde_json::to_vec(&json!([event])).unwrap().len(), + budget + extra + ); + state + .db + .replace_parameterized_event(community, &event, "byte-boundary", None) + .await + .unwrap(); + let (status, body) = request(state.clone(), &host, Some(&key), filter(&key)).await; + if extra == 0 { + assert_eq!(status, StatusCode::OK); + assert_eq!(body["complete"], true); + assert_eq!(serde_json::to_vec(&body["events"]).unwrap().len(), budget); + // Discovery bounds the array, not the larger success envelope. + assert!(serde_json::to_vec(&body).unwrap().len() > budget); + } else { + assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE); + assert_ne!(body["complete"], true); + assert!(body.get("events").is_none()); + } + } + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn read_state_snapshot_router_beyond_page_cap_and_overflow() { + let (state, host, community, key) = setup().await; + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + // Same-second plateau > ordinary 1,000 cap, including unrelated data. + for n in 0..1001 { + let event = EventBuilder::new(Kind::Custom(30078), "payload") + .custom_created_at(nostr::Timestamp::from(1789090000)) + .tags([Tag::identifier(format!("fixture-{n}"))]) + .sign_with_keys(&key) + .unwrap(); + state + .db + .insert_event(community, &event, None) + .await + .unwrap(); + } + let (status, body) = request(state.clone(), &host, Some(&key), filter(&key)).await; + assert_eq!(status, StatusCode::OK, "{body}"); + assert_eq!(body["events"].as_array().unwrap().len(), 1001); + // Stored byte preflight must reject before decoding these corrupt rows. + sqlx::query("UPDATE events SET content=repeat('x', 9000) WHERE community_id=$1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .unwrap(); + let (status, body) = request(state.clone(), &host, Some(&key), filter(&key)).await; + assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE, "{body}"); + assert_ne!(body["complete"], true); + sqlx::query("DELETE FROM events WHERE community_id=$1") + .bind(community.as_uuid()) + .execute(&pool) + .await + .unwrap(); + // Count overflow must not masquerade as a capped complete response. + sqlx::query("INSERT INTO events (community_id,id,pubkey,created_at,kind,tags,content,sig) + SELECT $1, decode(md5(n::text)||md5(n::text),'hex'),$2,to_timestamp(1789090000),30078,'[]','',decode(repeat('00',64),'hex') + FROM generate_series(1,4097) n") + .bind(community.as_uuid()).bind(key.public_key().to_bytes().to_vec()).execute(&pool).await.unwrap(); + let (status, body) = request(state, &host, Some(&key), filter(&key)).await; + assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE, "{body}"); + assert_ne!(body["complete"], true); + } +} diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index e6b18cdd0f..043458b4bc 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -31,6 +31,9 @@ pub struct RelayInfo { /// admins/owners via the kind:9033 command. Omitted when no icon is set. #[serde(skip_serializing_if = "Option::is_none")] pub icon: Option, + /// Host-bound atomic read-state snapshot capability; absent on unresolved hosts. + #[serde(skip_serializing_if = "Option::is_none")] + pub read_state_snapshot: Option, /// Relay operator's public key (hex), if published. pub pubkey: Option, /// Contact address for the relay operator. @@ -203,6 +206,7 @@ impl RelayInfo { name: "Buzz Relay".to_string(), description: "Buzz — private team communication relay".to_string(), icon: icon.filter(|s| !s.is_empty()).map(|s| s.to_string()), + read_state_snapshot: None, pubkey: None, contact: None, supported_nips, @@ -293,6 +297,14 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st admin_api.as_deref(), state.config.klipy.as_ref().map(|_| "klipy"), ); + if let Ok(tenant) = crate::tenant::bind_community(&state.db, raw_host).await { + info.read_state_snapshot = Some(serde_json::json!({ + "version": 1, + "community_id": tenant.community().as_uuid(), + "max_events": buzz_db::read_state::MAX_SNAPSHOT_EVENTS, + "max_event_array_bytes": buzz_db::read_state::MAX_SNAPSHOT_BYTES, + })); + } let tenant_host = if state.config.push_enabled { crate::tenant::bind_community(&state.db, raw_host) .await diff --git a/docs/nips/NIP-RS.md b/docs/nips/NIP-RS.md index 6a095df60a..071fc82f47 100644 --- a/docs/nips/NIP-RS.md +++ b/docs/nips/NIP-RS.md @@ -378,6 +378,97 @@ Until a complete load succeeds, the client MUST evaluate unread state from its o The number of coordinates a full-state load must retrieve is bounded by the number of installations that have ever used the override layer, plus their not-yet-deleted rotation predecessors. It grows with the user's device history, not with elapsed time, and — because a coordinate carrying `ov_*` entries may not be deleted until it has been carried forward (see Client-ID Rotation) — it does not shrink on its own. Clients SHOULD carry forward and delete rotation predecessors promptly so the count stays near one coordinate per live installation. +#### Buzz Atomic Snapshot Extension (version 1) + +A relay MAY offer a complete **point-in-time** alternative to the paginated +full-state load above through the existing authenticated HTTP Nostr query bridge. +This extension does not change kind 30078, encryption, register algebra, or the +ordinary `POST /query` array response. It does not assert the five WebSocket +conformance properties above. + +Discovery: obtain NIP-11 from the configured, trusted relay origin. Its optional +`read_state_snapshot` descriptor contains `version: 1`, `community_id` (canonical +UUID string), `max_events: 4096`, and `max_event_array_bytes: 8388608`. The byte +budget applies only to the compact JSON `events` array, not the complete HTTP +response; clients MUST allow additional space for the success envelope metadata. +The community identifier is resolved from that request's Host through the active +community map; it is not supplied by the client or derived from its key. An +unresolved host omits the descriptor. This is TLS/origin-bound discovery, not an +independently signed global community identity. Clients MUST bind discovery and +queries to the same trusted relay origin and community session, and fence +asynchronous results on identity or community changes. + +Request: NIP-98 authenticated `POST /query` with exactly one raw filter: + +```json +[{"kinds":[30078],"authors":[""],"read_state_snapshot":1}] +``` + +No extra keys, tags, time bounds, pagination, other kinds/authors, or mixed filters +are accepted. Admission, replay protection, and relay membership gates apply as +for ordinary bridge reads. The relay never substitutes an agent owner's identity +for the authenticated signing identity. + +Success is a versioned envelope, **not an event array**: + +```json +{ + "read_state_snapshot": 1, + "complete": true, + "community_id": "", + "pubkey": "", + "snapshot_id": "<64-lowercase-hex>", + "events": [] +} +``` + +`events` contains every non-deleted own-author kind-30078 event retained at one +writer-database MVCC statement cut, including unrelated application coordinates +and events without a `t` tag. There is no ordinary query cap, age window, replica +routing, or multi-request pagination. A coordinate replaced during the statement +is consistently represented by the version visible at its cut; a later snapshot +sees the replacement. This guarantee covers retained current state, not purged +history, uncommitted writes, or future commits. + +`snapshot_id` is SHA-256 of the concatenation of UTF-8 +`buzz-read-state-snapshot-v1` plus a zero byte, the 16 raw community UUID bytes, +the 32 author bytes, and each 32-byte event id in `created_at DESC, id ASC` order. +It is a content identifier, **not** a monotone write revision, CAS precondition, +or live-delivery cursor. An unchanged view may have the same identifier. + +Both event count and bytes are hard bounded. More than 4096 events, more than +8 MiB of stored content plus PostgreSQL JSON tag text, or more than 8 MiB of +compact encoded event-array bytes fails the whole request (HTTP 413). No partial +envelope is returned. Row reconstruction, signature verification, serialization, +or database failure produces an explicit non-success response (HTTP 503 after +authentication). Invalid extension requests produce HTTP 400; normal auth, +unknown-host, membership, replay and admission errors retain their bridge status. +Clients MUST treat **all** such errors as `cannot prove complete`. There is no +permission to delete old coordinates to get under the snapshot cap. + +Clients MUST reject absent discovery, version mismatch, ordinary arrays (including +an old relay silently ignoring the extension), missing/false completeness, +community/pubkey mismatch, invalid signatures, duplicate/conflicting coordinates, +or invalid event envelopes. Read-state selection/decryption is client-side. An +unreadable or unsupported recognized read-state coordinate MUST block override +actions/publication; it MUST NOT be interpreted as absent state. + +A validated snapshot substitutes for the paginated enumeration and mutation fence +**only for its point-in-time load**. Before **each** override action, canonical +publication, or carry-forward/deletion decision, obtain a fresh snapshot from +**every** write relay and merge those snapshots with durable local state. Failure +on any required relay blocks that operation. Do not publish own coordinates while +loading. Ordinary read-before-write, primary co-location, permanent clear floors, +uint32/budget exhaustion, and per-relay acceptance-before-deletion rules remain. +A failed/ambiguous publish does not report synchronized action success. + +Actions after different snapshot cuts are concurrent CRDT actions, not serialized +transactions. A snapshot does not reserve a coordinate, prevent another writer, +or make subsequent live delivery lossless. Clients MUST NOT cache `complete:true` +as a session-wide authority or use it to label live freshness. On reconnect, +identity changes, or later canonical publication, acquire a new snapshot; merge +arriving live state as evidence but do not infer a complete load from it. + ### Merge Rule After decrypting all fetched blobs, the effective read timestamp for each context is: @@ -783,9 +874,9 @@ that expose read activity to other users MUST require explicit user consent. ## Backwards Compatibility -This NIP introduces no changes to existing event kinds and adds no new kind, wire message, or relay-stored read-state logic. It uses only standard NIP-01 event storage, NIP-33 addressable event semantics, NIP-44 encryption, and NIP-78 application data conventions. Clients that do not implement this NIP are unaffected, as are clients that implement everything but the manual-unread override layer. +This NIP introduces no changes to existing event kinds or relay-stored read-state logic. The baseline uses only standard NIP-01 event storage, NIP-33 addressable event semantics, NIP-44 encryption, and NIP-78 application data conventions. The optional Buzz Atomic Snapshot Extension adds explicitly discovered, versioned query/discovery envelopes; ordinary query arrays and WebSocket messages remain unchanged. Clients that do not implement this NIP are unaffected, as are clients that implement everything but the manual-unread override layer. -The override layer is the exception, and it is a relay-compatibility one rather than a client one. Its full-state load carries the completeness guarantee only against a relay that satisfies the ordering, capacity, floor, push, and barrier requirements enumerated in Full-State Load. Against a relay known or evidenced not to conform, every load resolves to *cannot prove complete* and the actions that depend on a complete load report as failed; against an undetectably nonconforming relay, a load may still return *complete*, and the completeness guarantee does not apply to that verdict. In either case the layer still runs and still merges, and frontier sync is unaffected. +The override layer is the exception, and it is a relay-compatibility one rather than a client one. Its paginated full-state load carries the completeness guarantee only against a relay that satisfies the ordering, capacity, floor, push, and barrier requirements enumerated in Full-State Load. A validated atomic snapshot supplies the alternative point-in-time guarantee described in Buzz Atomic Snapshot Extension, without asserting those WebSocket properties. Without either proof, every load resolves to *cannot prove complete* and the actions that depend on a complete load report as failed; against an undetectably nonconforming relay, a load may still return *complete*, and the completeness guarantee does not apply to that verdict. In either case the layer still runs and still merges, and frontier sync is unaffected. ## References