mirror of
https://github.com/block/buzz.git
synced 2026-10-31 22:41:12 +00:00
refactor(serverless): back agent relay layer with nostr-relay-pool
The serverless agent path hand-rolled multi-relay subscribe/reconnect/ resubscribe/dedup in relay.rs — reimplementing what nostr-relay-pool (rust-nostr) already does, with gaps that caused the agent to miss messages when a single relay 502'd mid-session. Replace the serverless backend with nostr-relay-pool's RelayPool, which provides the standard Nostr client model used by damus/0xchat/nostr-tools: - persistent REQ per relay, fanned out across all relays - auto-reconnect (default) with jittered backoff - auto-resubscribe on reconnect (relays replay stored events on REQ) - merge + dedup across relays (Event notification = first-seen only) HarnessRelay now dispatches to ServerlessRelay when serverless; server mode (NIP-42 AUTH + HTTP bridge) is unchanged. RestClient and the typing/ presence publisher share the pool. Subscription IDs match the server-mode scheme so event routing is identical. Proven: e2e_agent_responds passes 3x consecutively (~3s each) against live public relays, surviving damus 503s via nos.lol. - 272 unit tests pass, clippy clean
This commit is contained in:
Michael Neale
164
Cargo.lock
generated
164
Cargo.lock
generated
@@ -153,6 +153,37 @@ dependencies = [
|
||||
"syn",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-utility"
|
||||
version = "0.3.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a34a3b57207a7a1007832416c3e4862378c8451b4e8e093e436f48c2d3d2c151"
|
||||
dependencies = [
|
||||
"futures-util",
|
||||
"gloo-timers",
|
||||
"tokio",
|
||||
"wasm-bindgen-futures",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "async-wsocket"
|
||||
version = "0.13.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1c92385c7c8b3eb2de1b78aeca225212e4c9a69a78b802832759b108681a5069"
|
||||
dependencies = [
|
||||
"async-utility",
|
||||
"futures",
|
||||
"futures-util",
|
||||
"js-sys",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
"tokio-socks",
|
||||
"tokio-tungstenite 0.26.2",
|
||||
"url",
|
||||
"wasm-bindgen",
|
||||
"web-sys",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "atoi"
|
||||
version = "2.0.0"
|
||||
@@ -162,6 +193,12 @@ dependencies = [
|
||||
"num-traits",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "atomic-destructor"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "ef49f5882e4b6afaac09ad239a4f8c70a24b8f2b0897edb1f706008efd109cf4"
|
||||
|
||||
[[package]]
|
||||
name = "atomic-waker"
|
||||
version = "1.1.2"
|
||||
@@ -268,7 +305,7 @@ dependencies = [
|
||||
"sha1 0.10.6",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"tower",
|
||||
"tower-layer",
|
||||
"tower-service",
|
||||
@@ -727,7 +764,7 @@ dependencies = [
|
||||
"serde_json",
|
||||
"sprout-sdk",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"url",
|
||||
]
|
||||
|
||||
@@ -1386,6 +1423,18 @@ dependencies = [
|
||||
"regex-syntax",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "gloo-timers"
|
||||
version = "0.3.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "bbb143cf96099802033e0d4f4963b19fd2e0b728bcf076cd9cf7f6634f092994"
|
||||
dependencies = [
|
||||
"futures-channel",
|
||||
"futures-core",
|
||||
"js-sys",
|
||||
"wasm-bindgen",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "h2"
|
||||
version = "0.4.14"
|
||||
@@ -2040,6 +2089,12 @@ dependencies = [
|
||||
"tracing-subscriber",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "lru"
|
||||
version = "0.16.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7f66e8d5d03f609abc3a39e6f08e4164ebf1447a732906d39eb9b99b7919ef39"
|
||||
|
||||
[[package]]
|
||||
name = "lru-slab"
|
||||
version = "0.1.2"
|
||||
@@ -2237,6 +2292,12 @@ version = "0.1.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b"
|
||||
|
||||
[[package]]
|
||||
name = "negentropy"
|
||||
version = "0.5.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "f0efe882e02d206d8d279c20eb40e03baf7cb5136a1476dc084a324fbc3ec42d"
|
||||
|
||||
[[package]]
|
||||
name = "nix"
|
||||
version = "0.31.3"
|
||||
@@ -2273,6 +2334,35 @@ dependencies = [
|
||||
"url",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nostr-database"
|
||||
version = "0.44.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7462c9d8ae5ef6a28d66a192d399ad2530f1f2130b13186296dbb11bdef5b3d1"
|
||||
dependencies = [
|
||||
"lru",
|
||||
"nostr",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "nostr-relay-pool"
|
||||
version = "0.44.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "91b2c039df4f96c4bf7dae52a74fd5516ad6dda83a11c0c69dea91b5255a4f37"
|
||||
dependencies = [
|
||||
"async-utility",
|
||||
"async-wsocket",
|
||||
"atomic-destructor",
|
||||
"hex",
|
||||
"lru",
|
||||
"negentropy",
|
||||
"nostr",
|
||||
"nostr-database",
|
||||
"tokio",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "ntapi"
|
||||
version = "0.4.3"
|
||||
@@ -3649,6 +3739,7 @@ dependencies = [
|
||||
"hex",
|
||||
"nix",
|
||||
"nostr",
|
||||
"nostr-relay-pool",
|
||||
"reqwest 0.13.3",
|
||||
"rustls",
|
||||
"serde",
|
||||
@@ -3659,7 +3750,7 @@ dependencies = [
|
||||
"sprout-sdk",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"tokio-util",
|
||||
"toml",
|
||||
"tracing",
|
||||
@@ -3765,7 +3856,7 @@ dependencies = [
|
||||
"sprout-sdk",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"url",
|
||||
"uuid",
|
||||
]
|
||||
@@ -3856,7 +3947,7 @@ dependencies = [
|
||||
"sprout-sdk",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"tracing",
|
||||
"tracing-subscriber",
|
||||
"url",
|
||||
@@ -3904,7 +3995,7 @@ dependencies = [
|
||||
"serde_json",
|
||||
"sha2 0.11.0",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"tokio-util",
|
||||
]
|
||||
|
||||
@@ -3920,7 +4011,7 @@ dependencies = [
|
||||
"sprout-core",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"url",
|
||||
"zeroize",
|
||||
]
|
||||
@@ -3956,7 +4047,7 @@ dependencies = [
|
||||
"sprout-core",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"tower-http",
|
||||
"tracing",
|
||||
"tracing-subscriber",
|
||||
@@ -4081,7 +4172,7 @@ dependencies = [
|
||||
"sprout-mcp",
|
||||
"thiserror 2.0.18",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tokio-tungstenite 0.29.0",
|
||||
"tracing",
|
||||
"tracing-subscriber",
|
||||
"url",
|
||||
@@ -4541,6 +4632,18 @@ dependencies = [
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-socks"
|
||||
version = "0.5.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a7e2948f60dbe26b35f2c7fb74ac2854c1fddded0fe9d7548fcc674a246f7615"
|
||||
dependencies = [
|
||||
"either",
|
||||
"futures-util",
|
||||
"thiserror 1.0.69",
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-stream"
|
||||
version = "0.1.18"
|
||||
@@ -4552,6 +4655,22 @@ dependencies = [
|
||||
"tokio",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-tungstenite"
|
||||
version = "0.26.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "7a9daff607c6d2bf6c16fd681ccb7eecc83e4e2cdc1ca067ffaadfca5de7f084"
|
||||
dependencies = [
|
||||
"futures-util",
|
||||
"log",
|
||||
"rustls",
|
||||
"rustls-pki-types",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
"tungstenite 0.26.2",
|
||||
"webpki-roots 0.26.11",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tokio-tungstenite"
|
||||
version = "0.29.0"
|
||||
@@ -4564,7 +4683,7 @@ dependencies = [
|
||||
"rustls-pki-types",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
"tungstenite",
|
||||
"tungstenite 0.29.0",
|
||||
"webpki-roots 0.26.11",
|
||||
]
|
||||
|
||||
@@ -4759,6 +4878,25 @@ version = "0.2.5"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
|
||||
|
||||
[[package]]
|
||||
name = "tungstenite"
|
||||
version = "0.26.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4793cb5e56680ecbb1d843515b23b6de9a75eb04b66643e256a396d43be33c13"
|
||||
dependencies = [
|
||||
"bytes",
|
||||
"data-encoding",
|
||||
"http",
|
||||
"httparse",
|
||||
"log",
|
||||
"rand 0.9.4",
|
||||
"rustls",
|
||||
"rustls-pki-types",
|
||||
"sha1 0.10.6",
|
||||
"thiserror 2.0.18",
|
||||
"utf-8",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "tungstenite"
|
||||
version = "0.29.0"
|
||||
@@ -4863,6 +5001,12 @@ version = "2.1.3"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
|
||||
|
||||
[[package]]
|
||||
name = "utf-8"
|
||||
version = "0.7.6"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9"
|
||||
|
||||
[[package]]
|
||||
name = "utf8_iter"
|
||||
version = "1.0.4"
|
||||
|
||||
@@ -57,6 +57,10 @@ deadpool-redis = { version = "0.23", features = ["rt_tokio_1"] }
|
||||
|
||||
# Nostr
|
||||
nostr = { version = "0.44", features = ["nip44", "nip59", "nip98"] }
|
||||
# High-level relay pool: persistent multi-relay subscriptions with built-in
|
||||
# auto-reconnect + auto-resubscribe (relays replay stored events on REQ). Used
|
||||
# for the serverless agent path instead of hand-rolling reconnect/resubscribe.
|
||||
nostr-relay-pool = { version = "0.44", default-features = false }
|
||||
|
||||
# Serialization
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
|
||||
@@ -23,6 +23,7 @@ sprout-persona = { path = "../sprout-persona" }
|
||||
|
||||
# Nostr
|
||||
nostr = { workspace = true }
|
||||
nostr-relay-pool = { workspace = true }
|
||||
|
||||
# Async runtime
|
||||
tokio = { workspace = true }
|
||||
|
||||
@@ -8,6 +8,7 @@ mod observer;
|
||||
mod pool;
|
||||
mod queue;
|
||||
mod relay;
|
||||
mod serverless_relay;
|
||||
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Arc;
|
||||
|
||||
@@ -114,6 +114,10 @@ pub struct RestClient {
|
||||
/// down or rate-limiting — the standard Nostr client model (damus
|
||||
/// `RelayPool`, nostr-tools `SimplePool`).
|
||||
pub ws_urls: Vec<String>,
|
||||
/// Shared serverless relay pool. When `Some`, `query`/`submit_event` use the
|
||||
/// pool (auto-reconnect, multi-relay merge/dedup) instead of the hand-rolled
|
||||
/// `query_ws`/`submit_event_ws`. `None` in server mode.
|
||||
pub pool: Option<nostr_relay_pool::RelayPool>,
|
||||
}
|
||||
|
||||
/// Whether an HTTP status code is retriable (transient server/rate-limit errors).
|
||||
@@ -274,6 +278,19 @@ impl RestClient {
|
||||
/// Accepts a slice of `nostr::Filter` (serialized as JSON array).
|
||||
/// Returns the events as a `serde_json::Value` (JSON array of event objects).
|
||||
pub async fn query(&self, filters: &[nostr::Filter]) -> Result<Value, RelayError> {
|
||||
if let Some(pool) = &self.pool {
|
||||
let events = pool
|
||||
.fetch_events(
|
||||
filters.to_vec(),
|
||||
std::time::Duration::from_secs(8),
|
||||
nostr_relay_pool::relay::ReqExitPolicy::ExitOnEOSE,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| RelayError::Http(format!("fetch failed: {e}")))?;
|
||||
let events: Vec<Event> = events.into_iter().collect();
|
||||
return serde_json::to_value(events)
|
||||
.map_err(|e| RelayError::Http(format!("event serialize error: {e}")));
|
||||
}
|
||||
if self.serverless {
|
||||
return self.query_ws(filters).await;
|
||||
}
|
||||
@@ -289,6 +306,13 @@ impl RestClient {
|
||||
///
|
||||
/// The event must already be signed. Returns the relay response JSON.
|
||||
pub async fn submit_event(&self, event: &Event) -> Result<Value, RelayError> {
|
||||
if let Some(pool) = &self.pool {
|
||||
let output = pool
|
||||
.send_event(event)
|
||||
.await
|
||||
.map_err(|e| RelayError::Http(format!("publish failed: {e}")))?;
|
||||
return Ok(serde_json::json!({"accepted": !output.success.is_empty()}));
|
||||
}
|
||||
if self.serverless {
|
||||
return self.submit_event_ws(event).await;
|
||||
}
|
||||
@@ -686,13 +710,13 @@ enum RelayMessage {
|
||||
// ── Commands sent from HarnessRelay to the background task ───────────────────
|
||||
|
||||
/// Subscription ID for the global membership notification subscription.
|
||||
const MEMBERSHIP_NOTIF_SUB_ID: &str = "membership-notif";
|
||||
pub(crate) const MEMBERSHIP_NOTIF_SUB_ID: &str = "membership-notif";
|
||||
/// Subscription ID for encrypted owner-to-agent observer control frames.
|
||||
const OBSERVER_CONTROL_SUB_ID: &str = "agent-observer-control";
|
||||
pub(crate) const OBSERVER_CONTROL_SUB_ID: &str = "agent-observer-control";
|
||||
/// Subscription ID for the agent's NIP-17 gift-wrap inbox (encrypted serverless
|
||||
/// private channels / DMs). The relay only stores `kind 1059` blobs addressed
|
||||
/// by `#p`; the agent unwraps each and routes the inner message by its `h` tag.
|
||||
const GIFT_WRAP_SUB_ID: &str = "agent-gift-wrap-inbox";
|
||||
pub(crate) const GIFT_WRAP_SUB_ID: &str = "agent-gift-wrap-inbox";
|
||||
/// NIP-59 gift wrap kind.
|
||||
const KIND_GIFT_WRAP: u16 = 1059;
|
||||
|
||||
@@ -764,17 +788,32 @@ pub struct HarnessRelay {
|
||||
/// Handles to all background tasks (one per relay + the command fan-out),
|
||||
/// for clean shutdown.
|
||||
bg_handles: Vec<tokio::task::JoinHandle<()>>,
|
||||
/// Serverless backend (nostr-relay-pool). When `Some`, all subscribe /
|
||||
/// query / publish / next_event calls dispatch here instead of the
|
||||
/// hand-rolled background-task path above. Built only in serverless mode;
|
||||
/// server mode leaves this `None` and uses the original code unchanged.
|
||||
serverless_backend: Option<crate::serverless_relay::ServerlessRelay>,
|
||||
}
|
||||
|
||||
/// Cloneable publisher handle for signed events on the relay background socket.
|
||||
/// Cloneable publisher handle for signed events. In server mode it sends to the
|
||||
/// relay background task via `cmd_tx`; in serverless it publishes through the
|
||||
/// shared `RelayPool` (clone-cheap, Arc inside).
|
||||
#[derive(Clone)]
|
||||
pub struct RelayEventPublisher {
|
||||
cmd_tx: mpsc::Sender<RelayCommand>,
|
||||
pool: Option<nostr_relay_pool::RelayPool>,
|
||||
}
|
||||
|
||||
impl RelayEventPublisher {
|
||||
/// Publish a signed event through the relay background task.
|
||||
/// Publish a signed event through the relay background task (server) or the
|
||||
/// pool (serverless).
|
||||
pub async fn publish_event(&self, event: Event) -> Result<(), RelayError> {
|
||||
if let Some(pool) = &self.pool {
|
||||
pool.send_event(&event)
|
||||
.await
|
||||
.map_err(|e| RelayError::Http(format!("publish failed: {e}")))?;
|
||||
return Ok(());
|
||||
}
|
||||
self.cmd_tx
|
||||
.send(RelayCommand::PublishEvent {
|
||||
event: Box::new(event),
|
||||
@@ -800,6 +839,42 @@ impl HarnessRelay {
|
||||
auth_tag: Option<nostr::Tag>,
|
||||
serverless: bool,
|
||||
) -> Result<Self, RelayError> {
|
||||
// Serverless: delegate the entire relay layer to nostr-relay-pool
|
||||
// (persistent multi-relay REQ + auto-reconnect + auto-resubscribe +
|
||||
// merge/dedup). The hand-rolled background-task path below is used ONLY
|
||||
// for server mode (NIP-42 AUTH against the Sprout relay).
|
||||
if serverless {
|
||||
let backend = crate::serverless_relay::ServerlessRelay::connect(
|
||||
relay_url,
|
||||
keys,
|
||||
agent_pubkey_hex,
|
||||
)
|
||||
.await?;
|
||||
let relay_urls = backend.relay_urls().to_vec();
|
||||
let primary = relay_urls
|
||||
.first()
|
||||
.cloned()
|
||||
.unwrap_or_else(|| relay_url.to_string());
|
||||
// Unused channels/clients in serverless mode — the backend owns the
|
||||
// real streams. We keep the struct shape uniform.
|
||||
let (_dead_event_tx, event_rx) = mpsc::channel::<Option<SproutEvent>>(1);
|
||||
let (cmd_tx, _dead_cmd_rx) = mpsc::channel::<RelayCommand>(1);
|
||||
return Ok(Self {
|
||||
event_rx,
|
||||
observer_control_rx: None,
|
||||
cmd_tx,
|
||||
http: reqwest::Client::new(),
|
||||
relay_url: primary,
|
||||
relay_urls,
|
||||
keys: keys.clone(),
|
||||
auth_tag,
|
||||
serverless,
|
||||
seen_event_ids: std::collections::VecDeque::new(),
|
||||
bg_handles: Vec::new(),
|
||||
serverless_backend: Some(backend),
|
||||
});
|
||||
}
|
||||
|
||||
// Serverless workspaces pass a comma-separated relay LIST
|
||||
// (`wss://a,wss://b,...`). Connect to ALL of them and merge — this is
|
||||
// the standard Nostr client pattern (damus `RelayPool`, nostr-tools
|
||||
@@ -931,6 +1006,7 @@ impl HarnessRelay {
|
||||
serverless,
|
||||
seen_event_ids: std::collections::VecDeque::new(),
|
||||
bg_handles,
|
||||
serverless_backend: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1063,6 +1139,7 @@ impl HarnessRelay {
|
||||
} else {
|
||||
self.relay_urls.clone()
|
||||
},
|
||||
pool: self.serverless_backend.as_ref().map(|b| b.pool()),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1075,6 +1152,9 @@ impl HarnessRelay {
|
||||
channel_id: Uuid,
|
||||
filter: ChannelFilter,
|
||||
) -> Result<(), RelayError> {
|
||||
if let Some(backend) = &self.serverless_backend {
|
||||
return backend.subscribe_channel(channel_id, filter).await;
|
||||
}
|
||||
self.cmd_tx
|
||||
.send(RelayCommand::Subscribe { channel_id, filter })
|
||||
.await
|
||||
@@ -1085,6 +1165,13 @@ impl HarnessRelay {
|
||||
|
||||
/// Subscribe to membership notifications for this agent.
|
||||
pub async fn subscribe_membership_notifications(&mut self) -> Result<(), RelayError> {
|
||||
if let Some(backend) = &self.serverless_backend {
|
||||
// Generic relays don't emit Sprout's kind:44100 membership push
|
||||
// (that's a server side effect). In serverless the harness relies
|
||||
// on the periodic re-discovery loop instead — nothing to subscribe.
|
||||
// Still subscribe the gift-wrap inbox so encrypted DMs/privates work.
|
||||
return backend.subscribe_gift_wrap().await;
|
||||
}
|
||||
self.cmd_tx
|
||||
.send(RelayCommand::SubscribeMembership)
|
||||
.await
|
||||
@@ -1094,6 +1181,10 @@ impl HarnessRelay {
|
||||
|
||||
/// Subscribe to encrypted observer control frames addressed to this agent.
|
||||
pub async fn subscribe_observer_controls(&mut self) -> Result<(), RelayError> {
|
||||
if self.serverless_backend.is_some() {
|
||||
// Observer controls are a Sprout-relay feature; no-op in serverless.
|
||||
return Ok(());
|
||||
}
|
||||
self.cmd_tx
|
||||
.send(RelayCommand::SubscribeObserverControls)
|
||||
.await
|
||||
@@ -1110,11 +1201,16 @@ impl HarnessRelay {
|
||||
pub fn event_publisher(&self) -> RelayEventPublisher {
|
||||
RelayEventPublisher {
|
||||
cmd_tx: self.cmd_tx.clone(),
|
||||
pool: self.serverless_backend.as_ref().map(|b| b.pool()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Unsubscribe from a channel.
|
||||
pub async fn unsubscribe_channel(&mut self, channel_id: Uuid) -> Result<(), RelayError> {
|
||||
if let Some(backend) = &self.serverless_backend {
|
||||
backend.unsubscribe_channel(channel_id).await;
|
||||
return Ok(());
|
||||
}
|
||||
self.cmd_tx
|
||||
.send(RelayCommand::Unsubscribe { channel_id })
|
||||
.await
|
||||
@@ -1128,6 +1224,10 @@ impl HarnessRelay {
|
||||
/// Reads from the background task's event channel. Returns `None` on
|
||||
/// connection loss — the caller should call [`reconnect`](Self::reconnect).
|
||||
pub async fn next_event(&mut self) -> Option<SproutEvent> {
|
||||
if let Some(backend) = &mut self.serverless_backend {
|
||||
// The pool already dedups across relays — no local ring needed.
|
||||
return backend.next_event().await;
|
||||
}
|
||||
// The background tasks send `None` to signal connection loss. With
|
||||
// multiple relays the same event arrives from each relay it's stored on,
|
||||
// so dedup by event id (bounded ring, like damus's `seenEvents`).
|
||||
@@ -1152,6 +1252,9 @@ impl HarnessRelay {
|
||||
/// (typing indicators) prefer [`try_publish_event`] which never blocks.
|
||||
#[allow(dead_code)] // Public API — callers outside the harness may use this
|
||||
pub async fn publish_event(&self, event: Event) -> Result<(), RelayError> {
|
||||
if let Some(backend) = &self.serverless_backend {
|
||||
return backend.publish(&event).await;
|
||||
}
|
||||
self.cmd_tx
|
||||
.send(RelayCommand::PublishEvent {
|
||||
event: Box::new(event),
|
||||
@@ -1165,6 +1268,15 @@ impl HarnessRelay {
|
||||
/// Suitable for ephemeral commands like typing indicators where dropping
|
||||
/// the event on a full command channel is acceptable.
|
||||
pub fn try_publish_event(&self, event: Event) -> Result<(), RelayError> {
|
||||
if let Some(backend) = &self.serverless_backend {
|
||||
// Ephemeral (typing): fire-and-forget on the pool. Spawn so we don't
|
||||
// block; dropping on error is acceptable for typing indicators.
|
||||
let pool = backend.pool();
|
||||
tokio::spawn(async move {
|
||||
let _ = pool.send_event(&event).await;
|
||||
});
|
||||
return Ok(());
|
||||
}
|
||||
self.cmd_tx
|
||||
.try_send(RelayCommand::PublishEvent {
|
||||
event: Box::new(event),
|
||||
@@ -1218,6 +1330,10 @@ impl HarnessRelay {
|
||||
/// Reconnect after connection loss. Instructs the background task to
|
||||
/// re-authenticate and resubscribe to all previously active channels.
|
||||
pub async fn reconnect(&mut self) -> Result<(), RelayError> {
|
||||
if self.serverless_backend.is_some() {
|
||||
// The pool auto-reconnects + auto-resubscribes; nothing to do.
|
||||
return Ok(());
|
||||
}
|
||||
warn!("relay connection lost — reconnecting…");
|
||||
self.cmd_tx
|
||||
.send(RelayCommand::Reconnect)
|
||||
@@ -1232,6 +1348,10 @@ impl HarnessRelay {
|
||||
/// the background task to finish. Use this from async contexts instead of
|
||||
/// relying on `Drop` (which aborts immediately).
|
||||
pub async fn shutdown(mut self) {
|
||||
if let Some(backend) = &self.serverless_backend {
|
||||
backend.shutdown().await;
|
||||
return;
|
||||
}
|
||||
let _ = self.cmd_tx.send(RelayCommand::Shutdown).await;
|
||||
let handles = std::mem::take(&mut self.bg_handles);
|
||||
for handle in handles {
|
||||
@@ -3038,7 +3158,10 @@ fn jittered_duration(base: Duration) -> Duration {
|
||||
/// cryptographically verified `sender`, so we trust the rumor's author; the
|
||||
/// downstream pipeline reads only id/pubkey/kind/created_at/tags/content (never
|
||||
/// `.sig`). We materialize an `Event` via JSON with an empty signature.
|
||||
fn rumor_to_event(rumor: &nostr::UnsignedEvent, sender: nostr::PublicKey) -> Option<nostr::Event> {
|
||||
pub(crate) fn rumor_to_event(
|
||||
rumor: &nostr::UnsignedEvent,
|
||||
sender: nostr::PublicKey,
|
||||
) -> Option<nostr::Event> {
|
||||
let id = rumor.id?;
|
||||
// Rumors carry no signature (NIP-17). The seal already verified `sender`,
|
||||
// and the downstream pipeline never inspects `.sig`, so we attach an
|
||||
@@ -3055,7 +3178,7 @@ fn rumor_to_event(rumor: &nostr::UnsignedEvent, sender: nostr::PublicKey) -> Opt
|
||||
))
|
||||
}
|
||||
|
||||
fn extract_h_tag_uuid(event: &nostr::Event) -> Option<Uuid> {
|
||||
pub(crate) fn extract_h_tag_uuid(event: &nostr::Event) -> Option<Uuid> {
|
||||
event.tags.iter().find_map(|tag| {
|
||||
let tag_vec = tag.as_slice();
|
||||
if tag_vec.len() >= 2 && tag_vec[0] == "h" {
|
||||
|
||||
279
crates/sprout-acp/src/serverless_relay.rs
Normal file
279
crates/sprout-acp/src/serverless_relay.rs
Normal file
@@ -0,0 +1,279 @@
|
||||
//! Serverless relay backend built on `nostr-relay-pool`.
|
||||
//!
|
||||
//! In serverless mode the agent is an ordinary Nostr client talking to generic
|
||||
//! public relays (no Sprout server, no NIP-42 AUTH, no HTTP bridge). Rather
|
||||
//! than hand-roll multi-relay subscribe / reconnect / resubscribe / dedup (the
|
||||
//! `relay.rs` background-task machinery used for server mode), we delegate to
|
||||
//! [`RelayPool`], which provides exactly the standard Nostr client behaviour:
|
||||
//!
|
||||
//! - **Persistent REQ per relay**, fanned out across every relay in the pool.
|
||||
//! - **Auto-reconnect** (on by default) with jittered backoff.
|
||||
//! - **Auto-resubscribe** on reconnect — the pool re-sends every active REQ, and
|
||||
//! relays replay their stored events matching the filter (NIP-01), so events
|
||||
//! that arrived during a disconnect are recovered.
|
||||
//! - **Merge + dedup** across relays: `RelayPoolNotification::Event` fires only
|
||||
//! the first time an event id is seen, and excludes events we published.
|
||||
//!
|
||||
//! This mirrors how damus (`RelayPool`), 0xchat, and nostr-tools (`SimplePool`)
|
||||
//! receive messages reliably across flaky relays.
|
||||
//!
|
||||
//! Subscription IDs match the server-mode scheme so the harness routes events
|
||||
//! identically: `ch-<uuid>` for channels, plus the fixed membership / observer
|
||||
//! / gift-wrap inbox ids.
|
||||
|
||||
use nostr::{Event, Filter, Keys, Kind, PublicKey, SingleLetterTag, SubscriptionId};
|
||||
use nostr_relay_pool::{RelayOptions, RelayPool, RelayPoolNotification, SubscribeOptions};
|
||||
use tokio::sync::mpsc;
|
||||
use tracing::{debug, info, warn};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::config::ChannelFilter;
|
||||
use crate::relay::{
|
||||
channel_sub_id, extract_h_tag_uuid, rumor_to_event, RelayError, SproutEvent, GIFT_WRAP_SUB_ID,
|
||||
MEMBERSHIP_NOTIF_SUB_ID, OBSERVER_CONTROL_SUB_ID,
|
||||
};
|
||||
|
||||
/// Serverless relay backend. Owns a [`RelayPool`] and a background task that
|
||||
/// converts pool notifications into [`SproutEvent`]s.
|
||||
///
|
||||
/// One-shot queries (channel discovery) go through `RestClient`, which shares
|
||||
/// this pool via [`ServerlessRelay::pool`] — so there's no separate fetch path
|
||||
/// here.
|
||||
pub struct ServerlessRelay {
|
||||
pool: RelayPool,
|
||||
agent_pubkey: PublicKey,
|
||||
/// Merged, classified channel/membership/gift-wrap events.
|
||||
event_rx: mpsc::Receiver<Option<SproutEvent>>,
|
||||
/// Relay URLs the pool is configured with (for diagnostics).
|
||||
relay_urls: Vec<String>,
|
||||
/// Background notification-pump handle.
|
||||
_pump: tokio::task::JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl ServerlessRelay {
|
||||
/// Build the pool, add all relays, connect, and spawn the notification pump.
|
||||
pub async fn connect(
|
||||
relay_url: &str,
|
||||
keys: &Keys,
|
||||
agent_pubkey_hex: &str,
|
||||
) -> Result<Self, RelayError> {
|
||||
let relay_urls: Vec<String> = relay_url
|
||||
.split(',')
|
||||
.map(|s| s.trim().to_string())
|
||||
.filter(|s| !s.is_empty())
|
||||
.collect();
|
||||
if relay_urls.is_empty() {
|
||||
return Err(RelayError::Http("no relay URLs provided".into()));
|
||||
}
|
||||
|
||||
let agent_pubkey = PublicKey::from_hex(agent_pubkey_hex)
|
||||
.map_err(|e| RelayError::Http(format!("invalid agent pubkey: {e}")))?;
|
||||
|
||||
let pool = RelayPool::new();
|
||||
let mut added = 0usize;
|
||||
for url in &relay_urls {
|
||||
// RelayOptions::default() has reconnect=true + adjusted (jittered)
|
||||
// retry interval — exactly the auto-reconnect we want.
|
||||
match pool.add_relay(url.as_str(), RelayOptions::default()).await {
|
||||
Ok(_) => added += 1,
|
||||
Err(e) => warn!("serverless: failed to add relay {url}: {e}"),
|
||||
}
|
||||
}
|
||||
if added == 0 {
|
||||
return Err(RelayError::Http("no relays could be added".into()));
|
||||
}
|
||||
|
||||
// Connects all relays in the background; reconnection is automatic.
|
||||
pool.connect().await;
|
||||
info!(
|
||||
"serverless: relay pool connecting to {}/{} relay(s): {}",
|
||||
added,
|
||||
relay_urls.len(),
|
||||
relay_urls.join(", ")
|
||||
);
|
||||
|
||||
let (event_tx, event_rx) = mpsc::channel::<Option<SproutEvent>>(1024);
|
||||
|
||||
let pump_keys = keys.clone();
|
||||
let notifications = pool.notifications();
|
||||
let pump = tokio::spawn(notification_pump(notifications, pump_keys, event_tx));
|
||||
|
||||
Ok(Self {
|
||||
pool,
|
||||
agent_pubkey,
|
||||
event_rx,
|
||||
relay_urls,
|
||||
_pump: pump,
|
||||
})
|
||||
}
|
||||
|
||||
/// Relay URLs the pool is using.
|
||||
pub fn relay_urls(&self) -> &[String] {
|
||||
&self.relay_urls
|
||||
}
|
||||
|
||||
/// Cheap clone of the underlying pool (Arc inside) so detached tasks
|
||||
/// (presence, typing) can publish ephemeral events without holding the
|
||||
/// whole `ServerlessRelay`.
|
||||
pub fn pool(&self) -> RelayPool {
|
||||
self.pool.clone()
|
||||
}
|
||||
|
||||
/// Next merged event (channel / membership / decrypted gift-wrap). Returns
|
||||
/// `None` only if the pump task ended (pool shut down).
|
||||
pub async fn next_event(&mut self) -> Option<SproutEvent> {
|
||||
// The pump sends `Some(event)`; it never sends `None` unless closing.
|
||||
match self.event_rx.recv().await {
|
||||
Some(Some(ev)) => Some(ev),
|
||||
Some(None) | None => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Publish a signed event to all relays (succeeds if any accepts).
|
||||
pub async fn publish(&self, event: &Event) -> Result<(), RelayError> {
|
||||
let output = self
|
||||
.pool
|
||||
.send_event(event)
|
||||
.await
|
||||
.map_err(|e| RelayError::Http(format!("publish failed: {e}")))?;
|
||||
if output.success.is_empty() {
|
||||
return Err(RelayError::Http(format!(
|
||||
"no relay accepted event (failed on {} relay(s))",
|
||||
output.failed.len()
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Subscribe to a channel with the standard `ch-<uuid>` id. The pool keeps
|
||||
/// the REQ open and resubscribes on reconnect automatically.
|
||||
pub async fn subscribe_channel(
|
||||
&self,
|
||||
channel_id: Uuid,
|
||||
filter: ChannelFilter,
|
||||
) -> Result<(), RelayError> {
|
||||
let id = SubscriptionId::new(channel_sub_id(channel_id));
|
||||
let mut f = Filter::new().custom_tag(
|
||||
SingleLetterTag::lowercase(nostr::Alphabet::H),
|
||||
channel_id.to_string(),
|
||||
);
|
||||
if let Some(kinds) = filter.kinds {
|
||||
f = f.kinds(kinds.into_iter().map(|k| Kind::from(k as u16)));
|
||||
}
|
||||
if filter.require_mention {
|
||||
f = f.pubkey(self.agent_pubkey);
|
||||
}
|
||||
self.pool
|
||||
.subscribe_with_id(id, f, SubscribeOptions::default())
|
||||
.await
|
||||
.map_err(|e| RelayError::Http(format!("subscribe failed: {e}")))?;
|
||||
debug!("serverless: subscribed to channel {channel_id}");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Unsubscribe a channel.
|
||||
pub async fn unsubscribe_channel(&self, channel_id: Uuid) {
|
||||
let id = SubscriptionId::new(channel_sub_id(channel_id));
|
||||
self.pool.unsubscribe(&id).await;
|
||||
}
|
||||
|
||||
/// Subscribe to the agent's NIP-17 gift-wrap inbox (kind 1059, #p = agent).
|
||||
pub async fn subscribe_gift_wrap(&self) -> Result<(), RelayError> {
|
||||
let id = SubscriptionId::new(GIFT_WRAP_SUB_ID);
|
||||
let f = Filter::new().kind(Kind::GiftWrap).pubkey(self.agent_pubkey);
|
||||
self.pool
|
||||
.subscribe_with_id(id, f, SubscribeOptions::default())
|
||||
.await
|
||||
.map_err(|e| RelayError::Http(format!("gift-wrap subscribe failed: {e}")))?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Disconnect / shut down the pool.
|
||||
pub async fn shutdown(&self) {
|
||||
self.pool.shutdown().await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Classify pool notifications into SproutEvents / observer frames, mirroring
|
||||
/// the server-mode bg-task routing — but with no manual dedup/reconnect (the
|
||||
/// pool handles both).
|
||||
async fn notification_pump(
|
||||
mut notifications: tokio::sync::broadcast::Receiver<RelayPoolNotification>,
|
||||
keys: Keys,
|
||||
event_tx: mpsc::Sender<Option<SproutEvent>>,
|
||||
) {
|
||||
loop {
|
||||
let notif = match notifications.recv().await {
|
||||
Ok(n) => n,
|
||||
// Lagged: we missed some — keep going (relays will replay on any
|
||||
// resubscribe; dedup is by id so no harm).
|
||||
Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
|
||||
warn!("serverless: notification stream lagged by {n}");
|
||||
continue;
|
||||
}
|
||||
Err(tokio::sync::broadcast::error::RecvError::Closed) => {
|
||||
let _ = event_tx.send(None).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
match notif {
|
||||
RelayPoolNotification::Event {
|
||||
subscription_id,
|
||||
event,
|
||||
..
|
||||
} => {
|
||||
let sub = subscription_id.as_str();
|
||||
if sub == GIFT_WRAP_SUB_ID {
|
||||
// NIP-17 encrypted message: unwrap to the inner rumor.
|
||||
match nostr::nips::nip59::UnwrappedGift::from_gift_wrap(&keys, &event).await {
|
||||
Ok(unwrapped) => {
|
||||
if let Some(inner) = rumor_to_event(&unwrapped.rumor, unwrapped.sender)
|
||||
{
|
||||
if let Some(channel_uuid) = extract_h_tag_uuid(&inner) {
|
||||
let _ = event_tx
|
||||
.send(Some(SproutEvent {
|
||||
channel_id: channel_uuid,
|
||||
event: inner,
|
||||
}))
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => debug!("serverless: gift wrap not for us: {e}"),
|
||||
}
|
||||
} else if sub == OBSERVER_CONTROL_SUB_ID {
|
||||
// Observer controls are a Sprout-relay feature; never
|
||||
// subscribed in serverless, so this branch is unreachable.
|
||||
} else if sub == MEMBERSHIP_NOTIF_SUB_ID {
|
||||
if let Some(channel_uuid) = extract_h_tag_uuid(&event) {
|
||||
let _ = event_tx
|
||||
.send(Some(SproutEvent {
|
||||
channel_id: channel_uuid,
|
||||
event: *event,
|
||||
}))
|
||||
.await;
|
||||
}
|
||||
} else if let Some(channel_uuid) = channel_uuid_from_sub(sub) {
|
||||
let _ = event_tx
|
||||
.send(Some(SproutEvent {
|
||||
channel_id: channel_uuid,
|
||||
event: *event,
|
||||
}))
|
||||
.await;
|
||||
}
|
||||
}
|
||||
RelayPoolNotification::Message { .. } => {}
|
||||
RelayPoolNotification::Shutdown => {
|
||||
let _ = event_tx.send(None).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse `ch-<uuid>` subscription ids.
|
||||
fn channel_uuid_from_sub(sub: &str) -> Option<Uuid> {
|
||||
sub.strip_prefix("ch-")
|
||||
.and_then(|s| Uuid::parse_str(s).ok())
|
||||
}
|
||||
Reference in New Issue
Block a user