mirror of
https://github.com/block/buzz.git
synced 2026-10-28 04:51:40 +00:00
c4c86006f2
We are introducing a companion service to the relay that needs pubkey event mentions for pubkeys it manages. This PR introduces an outbox for pubkey mentions that will be sent to these "operator listener" services. ## Summary Deliver mentions of registered pubkeys to deployment-global operator listeners. - Configure listeners with `BUZZ_OPERATOR_LISTENERS`; listeners manage target pubkeys through NIP-98-authenticated `POST`/`DELETE /operator/listener/pubkeys` requests. - During supported event ingestion, match `p` tags to registrations and insert one outbox row per listener in the event transaction. - A background worker claims up to 10 rows, sends NIP-98-signed notifications, and retries transient failures with exponential backoff for up to 9 attempts (idle polling backs off from 250 ms to 2 s). - Outbox rows are ineligible for delivery after 15 minutes and are reaped every 5 minutes. Registered pubkeys live for 30 days; re-registering renews them, and a daily reaper removes expired registrations. - Keep persistence app-managed: no database triggers, functions, foreign keys, or separate match queue. ## Changed behavior Listener outbox rows are enqueued inside the shared event-insert and thread-metadata transactions; enqueue failure rolls back the event. `event_mentions` behavior is unchanged: ordinary event/reaction mention indexing remains post-commit best effort with warning-only failures. ## Testing - `cargo fmt --all` - `cargo test -p buzz-db --lib` - `cargo test -p buzz-relay --lib config::tests::operator_listener_routes` - `cargo check -p buzz-db -p buzz-relay`
777 lines
28 KiB
Rust
777 lines
28 KiB
Rust
//! Durable NIP-PL event matcher and gateway delivery worker.
|
|
|
|
use std::{
|
|
sync::Arc,
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
use buzz_core::filter::{filters_match, reader_authorized_for_event};
|
|
use chrono::{TimeDelta, Utc};
|
|
use nostr::Filter;
|
|
use serde::{Deserialize, Serialize};
|
|
use tracing::{error, warn};
|
|
|
|
use crate::{handlers::push_lease::Subscription, nip98::nip98_header, state::AppState};
|
|
|
|
const CLAIM_SECS: i64 = 30;
|
|
const EVENT_USEFUL_SECS: i64 = 3600;
|
|
const MAX_ATTEMPTS: i32 = 8;
|
|
/// Upper bound on one claimed matcher batch. Bounded well under
|
|
/// `get_events_by_ids`' 500-id batch-fetch contract.
|
|
const MATCH_BATCH_LIMIT: i64 = 64;
|
|
/// Idle poll floor and ceiling. An idle matcher previously issued a claim
|
|
/// transaction every 250ms forever; backing off to the ceiling folds that
|
|
/// steady-state cost while keeping first-message latency at the floor.
|
|
const IDLE_POLL_FLOOR: Duration = Duration::from_millis(250);
|
|
const IDLE_POLL_CEILING: Duration = Duration::from_secs(2);
|
|
/// Cadence of the poison-job sweep, which lives off the claim path: its scan
|
|
/// is not served by the due partial index, so running it inside every claim
|
|
/// made claims slower exactly when a backlog needed them fastest.
|
|
const REAP_INTERVAL: Duration = Duration::from_secs(30);
|
|
|
|
#[derive(Serialize)]
|
|
struct DeliveryRequest<'a> {
|
|
v: u8,
|
|
endpoint_grant: &'a str,
|
|
request_id: uuid::Uuid,
|
|
expires_at: i64,
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
#[serde(rename_all = "snake_case", tag = "status")]
|
|
enum DeliveryResponse {
|
|
Accepted,
|
|
InvalidEndpoint {
|
|
generation: i64,
|
|
invalid_at: Option<i64>,
|
|
},
|
|
Retry {
|
|
retry_after_seconds: Option<i64>,
|
|
},
|
|
}
|
|
|
|
/// Continuously claim accepted events in per-community batches and match
|
|
/// them against active leases (T2b). One batch costs one claim statement,
|
|
/// one event load, one lease scan, one membership scan, and one wake-enqueue
|
|
/// transaction — plus one complete/retry statement each — regardless of
|
|
/// batch size or how many (event, lease) pairs match.
|
|
pub async fn run_matcher(state: Arc<AppState>) {
|
|
let mut idle_delay = IDLE_POLL_FLOOR;
|
|
let mut last_reap = tokio::time::Instant::now();
|
|
loop {
|
|
if last_reap.elapsed() >= REAP_INTERVAL {
|
|
match state.db.reap_exhausted_push_matches().await {
|
|
Ok(reaped) if reaped > 0 => warn!(reaped, "reaped exhausted push match jobs"),
|
|
Ok(_) => {}
|
|
Err(e) => error!("push match reap failed: {e}"),
|
|
}
|
|
last_reap = tokio::time::Instant::now();
|
|
}
|
|
let until = Utc::now() + TimeDelta::seconds(CLAIM_SECS);
|
|
match state
|
|
.db
|
|
.claim_due_push_match_batch(MATCH_BATCH_LIMIT, until)
|
|
.await
|
|
{
|
|
Ok(Some(batch)) => {
|
|
idle_delay = IDLE_POLL_FLOOR;
|
|
process_match_batch(&state, batch).await;
|
|
}
|
|
Ok(None) => {
|
|
tokio::time::sleep(idle_delay).await;
|
|
idle_delay = (idle_delay * 2).min(IDLE_POLL_CEILING);
|
|
}
|
|
Err(e) => {
|
|
error!("push matcher claim failed: {e}");
|
|
tokio::time::sleep(Duration::from_secs(2)).await;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Per-batch state shared by every job: the community's active leases and
|
|
/// the exact (channel, lease author) membership pairs the jobs can consult.
|
|
struct MatchContext {
|
|
leases: Vec<buzz_db::push::MatchLease>,
|
|
memberships: std::collections::HashSet<(uuid::Uuid, Vec<u8>)>,
|
|
}
|
|
|
|
async fn load_match_context(
|
|
state: &AppState,
|
|
batch: &buzz_db::push::ClaimedMatchBatch,
|
|
) -> anyhow::Result<MatchContext> {
|
|
let leases = state.db.active_push_match_leases(batch.community).await?;
|
|
let mut channels: Vec<uuid::Uuid> = batch
|
|
.jobs
|
|
.iter()
|
|
.filter_map(|job| job.event.channel_id)
|
|
.collect();
|
|
channels.sort_unstable();
|
|
channels.dedup();
|
|
let mut authors: Vec<Vec<u8>> = leases.iter().map(|lease| lease.author.clone()).collect();
|
|
authors.sort_unstable();
|
|
authors.dedup();
|
|
let memberships = state
|
|
.db
|
|
.membership_pairs(batch.community, &channels, &authors)
|
|
.await?
|
|
.into_iter()
|
|
.collect();
|
|
Ok(MatchContext {
|
|
leases,
|
|
memberships,
|
|
})
|
|
}
|
|
|
|
async fn process_match_batch(state: &AppState, batch: buzz_db::push::ClaimedMatchBatch) {
|
|
let community = batch.community;
|
|
let context = match load_match_context(state, &batch).await {
|
|
Ok(context) => context,
|
|
Err(e) => {
|
|
// Shared context failed to load, so no job was evaluated: release
|
|
// the whole batch for retry. Jobs that keep failing are reaped by
|
|
// the periodic sweep once their attempts are exhausted.
|
|
warn!(%community, "push match context load failed: {e}");
|
|
metrics::counter!("buzz_push_match_jobs_total", "result" => "context_error")
|
|
.increment(batch.jobs.len() as u64);
|
|
let ids: Vec<Vec<u8>> = batch
|
|
.jobs
|
|
.iter()
|
|
.map(|job| job.event.event.id.as_bytes().to_vec())
|
|
.collect();
|
|
if let Err(e) = state
|
|
.db
|
|
.retry_push_match_batch(
|
|
community,
|
|
batch.claim_id,
|
|
&ids,
|
|
Utc::now() + TimeDelta::seconds(2),
|
|
)
|
|
.await
|
|
{
|
|
warn!(%community, "push match batch retry failed: {e}");
|
|
}
|
|
return;
|
|
}
|
|
};
|
|
let mut completed = Vec::new();
|
|
let mut retry = Vec::new();
|
|
// Jobs whose wakes are pending the set-wise enqueue below: their
|
|
// completion is decided by the flush, not by match evaluation.
|
|
let mut pending = Vec::new();
|
|
let mut wakes: Vec<buzz_db::push::WakeRequest> = Vec::new();
|
|
for job in &batch.jobs {
|
|
let match_queue_seconds = Utc::now()
|
|
.signed_duration_since(job.event.received_at)
|
|
.num_milliseconds()
|
|
.max(0) as f64
|
|
/ 1_000.0;
|
|
metrics::histogram!("buzz_push_match_queue_seconds").record(match_queue_seconds);
|
|
let event_id = job.event.event.id.as_bytes().to_vec();
|
|
match match_job(job, &context) {
|
|
Ok(job_wakes) if job_wakes.is_empty() => {
|
|
metrics::counter!("buzz_push_match_jobs_total", "result" => "unmatched")
|
|
.increment(1);
|
|
completed.push(event_id);
|
|
}
|
|
Ok(job_wakes) => {
|
|
metrics::counter!("buzz_push_match_jobs_total", "result" => "matched").increment(1);
|
|
pending.push((event_id, job.attempt));
|
|
wakes.extend(job_wakes);
|
|
}
|
|
Err(e) => {
|
|
metrics::counter!("buzz_push_match_jobs_total", "result" => "error").increment(1);
|
|
warn!(event_id=%job.event.event.id, attempt=job.attempt, "push match failed: {e}");
|
|
if job.attempt >= buzz_db::push::MAX_MATCH_ATTEMPTS {
|
|
// A poison event/lease must not retry forever or pin
|
|
// delivered outbox retention through the rematch guard.
|
|
completed.push(event_id);
|
|
} else {
|
|
retry.push(event_id);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// One transaction for every wake in the batch (T2b). InactiveLease and
|
|
// Duplicate outcomes are per-request non-errors; only a failed enqueue
|
|
// transaction sends the contributing jobs back for an idempotent rematch
|
|
// (the outbox dedup key absorbs any wakes that did commit elsewhere).
|
|
match state.db.enqueue_push_wakes(community, &wakes).await {
|
|
Ok(outcomes) => {
|
|
for outcome in outcomes {
|
|
let result = match outcome {
|
|
buzz_db::push::EnqueueWakeOutcome::Enqueued(_) => "enqueued",
|
|
buzz_db::push::EnqueueWakeOutcome::Duplicate(_) => "duplicate",
|
|
buzz_db::push::EnqueueWakeOutcome::InactiveLease => "inactive_lease",
|
|
};
|
|
metrics::counter!("buzz_push_wakes_total", "result" => result).increment(1);
|
|
}
|
|
completed.extend(pending.into_iter().map(|(event_id, _)| event_id));
|
|
}
|
|
Err(e) => {
|
|
metrics::counter!("buzz_push_wake_enqueue_errors_total").increment(1);
|
|
warn!(%community, "push wake batch enqueue failed: {e}");
|
|
for (event_id, attempt) in pending {
|
|
if attempt >= buzz_db::push::MAX_MATCH_ATTEMPTS {
|
|
completed.push(event_id);
|
|
} else {
|
|
retry.push(event_id);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if let Err(e) = state
|
|
.db
|
|
.complete_push_match_batch(community, batch.claim_id, &completed)
|
|
.await
|
|
{
|
|
warn!(%community, "push match batch completion failed: {e}");
|
|
}
|
|
if let Err(e) = state
|
|
.db
|
|
.retry_push_match_batch(
|
|
community,
|
|
batch.claim_id,
|
|
&retry,
|
|
Utc::now() + TimeDelta::seconds(2),
|
|
)
|
|
.await
|
|
{
|
|
warn!(%community, "push match batch retry failed: {e}");
|
|
}
|
|
}
|
|
|
|
/// Pure match evaluation: no DB access. Returns the wake requests this job
|
|
/// owes, which the caller flushes set-wise for the whole batch.
|
|
fn match_job(
|
|
job: &buzz_db::push::BatchedMatch,
|
|
context: &MatchContext,
|
|
) -> anyhow::Result<Vec<buzz_db::push::WakeRequest>> {
|
|
let mut wakes = Vec::new();
|
|
for lease in &context.leases {
|
|
let author_hex = hex::encode(&lease.author);
|
|
if !reader_authorized_for_event(&job.event.event, &author_hex) {
|
|
continue;
|
|
}
|
|
if let Some(channel) = job.event.channel_id {
|
|
if !context
|
|
.memberships
|
|
.contains(&(channel, lease.author.clone()))
|
|
{
|
|
continue;
|
|
}
|
|
}
|
|
let subscriptions: Vec<Subscription> = serde_json::from_value(lease.subscriptions.clone())?;
|
|
let mut class: Option<&str> = None;
|
|
for sub in &subscriptions {
|
|
let filter: Filter =
|
|
serde_json::from_value(serde_json::Value::Object(sub.filter.clone()))?;
|
|
if !push_filter_authorized_for_event(&filter, &job.event.event, &author_hex)
|
|
|| !filters_match(std::slice::from_ref(&filter), &job.event)
|
|
{
|
|
continue;
|
|
}
|
|
let ignored = sub.ignore.iter().any(|raw| {
|
|
serde_json::from_value::<Filter>(serde_json::Value::Object(raw.clone()))
|
|
.is_ok_and(|f| filters_match(&[f], &job.event))
|
|
});
|
|
let p_count = job
|
|
.event
|
|
.event
|
|
.tags
|
|
.iter()
|
|
.filter(|t| t.kind().to_string() == "p")
|
|
.count() as u64;
|
|
if ignored
|
|
|| sub
|
|
.suppress
|
|
.as_ref()
|
|
.is_some_and(|s| p_count > s.p_tags_max)
|
|
{
|
|
continue;
|
|
}
|
|
if class.is_none_or(|old| class_rank(&sub.class) > class_rank(old)) {
|
|
class = Some(&sub.class);
|
|
}
|
|
}
|
|
let Some(class) = class else { continue };
|
|
let event_deadline = job.event.event.created_at.as_secs() as i64 + EVENT_USEFUL_SECS;
|
|
let expires_at = lease.expires_at.min(event_deadline);
|
|
if expires_at <= Utc::now().timestamp() {
|
|
continue;
|
|
}
|
|
wakes.push(buzz_db::push::WakeRequest {
|
|
author: lease.author.clone(),
|
|
installation_id: lease.installation_id.clone(),
|
|
lease_generation: lease.generation,
|
|
event_id: job.event.event.id.as_bytes().to_vec(),
|
|
class: class.to_string(),
|
|
expires_at,
|
|
});
|
|
}
|
|
Ok(wakes)
|
|
}
|
|
|
|
/// Match-time counterpart of REQ's filter-level `#p` authorization gate.
|
|
/// Kind 1059 is globally stored and leaks recipient activity through wake
|
|
/// timing, so a lease may only match gift wraps addressed to its own author.
|
|
fn push_filter_authorized_for_event(
|
|
filter: &Filter,
|
|
event: &nostr::Event,
|
|
lease_author_hex: &str,
|
|
) -> bool {
|
|
if buzz_core::kind::event_kind_u32(event) != buzz_core::kind::KIND_GIFT_WRAP {
|
|
return true;
|
|
}
|
|
let p = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P);
|
|
filter.generic_tags.get(&p).is_some_and(|values| {
|
|
!values.is_empty()
|
|
&& values.iter().all(|value| value == lease_author_hex)
|
|
&& event
|
|
.tags
|
|
.filter(nostr::TagKind::SingleLetter(p))
|
|
.any(|tag| tag.content() == Some(lease_author_hex))
|
|
})
|
|
}
|
|
|
|
/// Continuously claim due wakes and deliver them through the push gateway.
|
|
pub async fn run_delivery_worker(state: Arc<AppState>) {
|
|
let http = match reqwest::Client::builder()
|
|
.timeout(state.config.push_gateway_timeout)
|
|
.build()
|
|
{
|
|
Ok(http) => http,
|
|
Err(error) => {
|
|
error!(%error, "push HTTP client initialization failed");
|
|
record_delivery("configuration_error");
|
|
return;
|
|
}
|
|
};
|
|
let mut idle_delay = Duration::from_millis(500);
|
|
loop {
|
|
let mut found = false;
|
|
match state.db.usage_community_hosts().await {
|
|
Ok(communities) => {
|
|
for community in communities {
|
|
let community = buzz_core::CommunityId::from_uuid(community.id);
|
|
let until = Utc::now() + TimeDelta::seconds(CLAIM_SECS);
|
|
match state.db.claim_due_push_wakes(community, 16, until).await {
|
|
Ok(wakes) => {
|
|
for wake in wakes {
|
|
found = true;
|
|
deliver_one(&state, &http, wake).await;
|
|
}
|
|
}
|
|
Err(e) => warn!(%community, "push wake claim failed: {e}"),
|
|
}
|
|
}
|
|
}
|
|
Err(e) => error!("push worker community scan failed: {e}"),
|
|
}
|
|
if found {
|
|
idle_delay = Duration::from_millis(500);
|
|
} else {
|
|
// Empty sweeps back off so an idle worker stops paying a full
|
|
// per-community claim scan every 500ms forever.
|
|
tokio::time::sleep(idle_delay).await;
|
|
idle_delay = (idle_delay * 2).min(IDLE_POLL_CEILING);
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn deliver_one(
|
|
state: &AppState,
|
|
http: &reqwest::Client,
|
|
claimed: buzz_db::push::ClaimedWake,
|
|
) {
|
|
let outcome = match state
|
|
.db
|
|
.revalidate_push_wake(claimed.community, claimed.id, claimed.claim_id)
|
|
.await
|
|
{
|
|
Ok(buzz_db::push::RevalidateWakeOutcome::Deliver(wake)) => wake,
|
|
Ok(buzz_db::push::RevalidateWakeOutcome::Suppressed) => {
|
|
let _ = state
|
|
.db
|
|
.fail_push_wake(claimed.community, claimed.id, claimed.claim_id)
|
|
.await;
|
|
record_delivery("suppressed");
|
|
return;
|
|
}
|
|
Err(e) => {
|
|
warn!(wake=%claimed.id, "push revalidation failed: {e}");
|
|
record_delivery("worker_error");
|
|
return;
|
|
}
|
|
};
|
|
if outcome.attempt == 1 {
|
|
let wake_queue_seconds = Utc::now()
|
|
.signed_duration_since(outcome.queued_at)
|
|
.num_milliseconds()
|
|
.max(0) as f64
|
|
/ 1_000.0;
|
|
metrics::histogram!("buzz_push_wake_queue_seconds").record(wake_queue_seconds);
|
|
}
|
|
if let Some(channel) = outcome.channel_id {
|
|
match state
|
|
.db
|
|
.is_member(outcome.community, channel, &outcome.author)
|
|
.await
|
|
{
|
|
Ok(true) => {}
|
|
Ok(false) => {
|
|
let _ = state
|
|
.db
|
|
.fail_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await;
|
|
record_delivery("suppressed");
|
|
return;
|
|
}
|
|
Err(e) => {
|
|
warn!(wake=%outcome.id, "push membership revalidation failed: {e}");
|
|
let _ = state
|
|
.db
|
|
.retry_push_wake(
|
|
outcome.community,
|
|
outcome.id,
|
|
outcome.claim_id,
|
|
Utc::now() + TimeDelta::seconds(2),
|
|
)
|
|
.await;
|
|
record_delivery("retry");
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
// Membership I/O above can race lease replacement. Re-run the generation
|
|
// fence as the final database operation before transport.
|
|
let outcome = match state
|
|
.db
|
|
.revalidate_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await
|
|
{
|
|
Ok(buzz_db::push::RevalidateWakeOutcome::Deliver(wake)) => wake,
|
|
Ok(buzz_db::push::RevalidateWakeOutcome::Suppressed) => {
|
|
let _ = state
|
|
.db
|
|
.fail_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await;
|
|
record_delivery("suppressed");
|
|
return;
|
|
}
|
|
Err(e) => {
|
|
warn!(wake=%outcome.id, "final push revalidation failed: {e}");
|
|
record_delivery("worker_error");
|
|
return;
|
|
}
|
|
};
|
|
let serving_write = match buzz_deletion::acquire_serving_write(
|
|
&state.db,
|
|
outcome.community,
|
|
"push_delivery",
|
|
)
|
|
.await
|
|
{
|
|
Ok(guard) => guard,
|
|
Err(error) => {
|
|
warn!(wake=%outcome.id, %error, "push delivery suppressed by community deletion fence");
|
|
let _ = state
|
|
.db
|
|
.fail_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await;
|
|
record_delivery("suppressed");
|
|
return;
|
|
}
|
|
};
|
|
let Some(url) = state.config.push_gateway_delivery_url.as_ref() else {
|
|
record_delivery("configuration_error");
|
|
return;
|
|
};
|
|
let body = match delivery_body(&outcome.endpoint_grant, outcome.id, outcome.expires_at) {
|
|
Ok(body) => body,
|
|
Err(error) => {
|
|
warn!(wake=%outcome.id, %error, "push delivery body encoding failed");
|
|
record_delivery("worker_error");
|
|
return;
|
|
}
|
|
};
|
|
let auth = match nip98_header(&state.relay_keypair, url.as_str(), &body) {
|
|
Ok(auth) => auth,
|
|
Err(e) => {
|
|
warn!(wake=%outcome.id, "push auth failed: {e}");
|
|
record_delivery("worker_error");
|
|
return;
|
|
}
|
|
};
|
|
if let Err(error) = serving_write.verify().await {
|
|
warn!(wake=%outcome.id, %error, "push serving lease lost before delivery");
|
|
record_delivery("suppressed");
|
|
return;
|
|
}
|
|
metrics::counter!("buzz_push_gateway_requests_total").increment(1);
|
|
let gateway_started = Instant::now();
|
|
let protected = serving_write
|
|
.protect(send_gateway_request(http, url, body, auth))
|
|
.await;
|
|
metrics::histogram!("buzz_push_gateway_request_seconds")
|
|
.record(gateway_started.elapsed().as_secs_f64());
|
|
let response = match protected {
|
|
Ok(response) => response,
|
|
Err(error) => {
|
|
warn!(wake=%outcome.id, %error, "push serving lease lost during delivery");
|
|
record_delivery("suppressed");
|
|
return;
|
|
}
|
|
};
|
|
match response {
|
|
Ok(r) if r.status().is_success() => match r.json::<DeliveryResponse>().await {
|
|
Ok(DeliveryResponse::Accepted) => {
|
|
let _ = state
|
|
.db
|
|
.complete_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await;
|
|
record_delivery("accepted");
|
|
}
|
|
_ => {
|
|
let _ = state
|
|
.db
|
|
.fail_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await;
|
|
record_delivery("failed");
|
|
}
|
|
},
|
|
Ok(r) if r.status() == reqwest::StatusCode::GONE => {
|
|
match r.json::<DeliveryResponse>().await {
|
|
Ok(DeliveryResponse::InvalidEndpoint {
|
|
generation,
|
|
invalid_at,
|
|
}) => {
|
|
if generation == outcome.lease_generation {
|
|
let _ = state
|
|
.db
|
|
.disable_push_endpoint(
|
|
outcome.community,
|
|
&outcome.author,
|
|
&outcome.installation_id,
|
|
generation,
|
|
)
|
|
.await;
|
|
}
|
|
warn!(wake=%outcome.id, ?invalid_at, "push endpoint permanently invalid");
|
|
}
|
|
_ => warn!(wake=%outcome.id, "invalid closed-protocol 410 response"),
|
|
}
|
|
let _ = state
|
|
.db
|
|
.fail_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await;
|
|
record_delivery("invalid_endpoint");
|
|
}
|
|
Ok(r) if r.status() == reqwest::StatusCode::SERVICE_UNAVAILABLE => {
|
|
let delay = match r.json::<DeliveryResponse>().await {
|
|
Ok(DeliveryResponse::Retry {
|
|
retry_after_seconds,
|
|
}) => retry_after_seconds
|
|
.filter(|seconds| *seconds > 0)
|
|
.unwrap_or(2),
|
|
_ => 2,
|
|
};
|
|
record_delivery(retry_or_fail(state, &outcome, delay).await);
|
|
}
|
|
Ok(r) if r.status() == reqwest::StatusCode::TOO_MANY_REQUESTS => {
|
|
record_delivery(retry_or_fail(state, &outcome, 2).await);
|
|
}
|
|
// A timed-out terminal attempt burns the stable request id. Its replay
|
|
// is indistinguishable from another invalid-grant 404, but sending a
|
|
// fresh id would double-deliver and defeat the gateway replay fence.
|
|
Ok(r) if r.status() == reqwest::StatusCode::NOT_FOUND && outcome.attempt > 1 => {
|
|
let _ = state
|
|
.db
|
|
.complete_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await;
|
|
record_delivery("replay_terminal");
|
|
}
|
|
Err(e) if e.is_timeout() || e.is_connect() => {
|
|
record_delivery(retry_or_fail(state, &outcome, 2).await);
|
|
}
|
|
_ => {
|
|
let _ = state
|
|
.db
|
|
.fail_push_wake(outcome.community, outcome.id, outcome.claim_id)
|
|
.await;
|
|
record_delivery("failed");
|
|
}
|
|
}
|
|
if let Err(error) = serving_write.finish().await {
|
|
warn!(wake=%outcome.id, %error, "failed to release community serving lease after push delivery");
|
|
}
|
|
}
|
|
|
|
fn delivery_body(
|
|
endpoint_grant: &str,
|
|
request_id: uuid::Uuid,
|
|
expires_at: i64,
|
|
) -> anyhow::Result<Vec<u8>> {
|
|
Ok(serde_json::to_vec(&DeliveryRequest {
|
|
v: 1,
|
|
endpoint_grant,
|
|
request_id,
|
|
expires_at,
|
|
})?)
|
|
}
|
|
|
|
async fn send_gateway_request(
|
|
http: &reqwest::Client,
|
|
url: &url::Url,
|
|
body: Vec<u8>,
|
|
auth: String,
|
|
) -> reqwest::Result<reqwest::Response> {
|
|
http.post(url.clone())
|
|
.header("Authorization", auth)
|
|
.header("Content-Type", "application/json")
|
|
.body(body)
|
|
.send()
|
|
.await
|
|
}
|
|
|
|
fn record_delivery(outcome: &'static str) {
|
|
metrics::counter!("buzz_push_deliveries_total", "outcome" => outcome).increment(1);
|
|
}
|
|
|
|
async fn retry_or_fail(
|
|
state: &AppState,
|
|
wake: &buzz_db::push::ClaimedWake,
|
|
delay: i64,
|
|
) -> &'static str {
|
|
if wake.attempt >= MAX_ATTEMPTS {
|
|
let _ = state
|
|
.db
|
|
.fail_push_wake(wake.community, wake.id, wake.claim_id)
|
|
.await;
|
|
"exhausted"
|
|
} else {
|
|
let secs = delay * (1_i64 << (wake.attempt - 1).clamp(0, 6));
|
|
let _ = state
|
|
.db
|
|
.retry_push_wake(
|
|
wake.community,
|
|
wake.id,
|
|
wake.claim_id,
|
|
Utc::now() + TimeDelta::seconds(secs),
|
|
)
|
|
.await;
|
|
"retry"
|
|
}
|
|
}
|
|
|
|
fn class_rank(_: &str) -> u8 {
|
|
1
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use axum::{
|
|
body::Bytes,
|
|
extract::{OriginalUri, State},
|
|
http::HeaderMap,
|
|
routing::post,
|
|
Json, Router,
|
|
};
|
|
use nostr::{EventBuilder, Kind, Tag};
|
|
use serde_json::Value;
|
|
use std::{future::IntoFuture, sync::Arc};
|
|
use tokio::sync::Mutex;
|
|
|
|
#[test]
|
|
fn gift_wrap_match_requires_self_p_filter_and_recipient() {
|
|
let recipient = nostr::Keys::generate();
|
|
let other = nostr::Keys::generate();
|
|
let sender = nostr::Keys::generate();
|
|
let recipient_hex = recipient.public_key().to_hex();
|
|
let event = EventBuilder::new(Kind::GiftWrap, "ciphertext")
|
|
.tag(Tag::public_key(other.public_key()))
|
|
.sign_with_keys(&sender)
|
|
.unwrap();
|
|
let self_filter = Filter::new().pubkey(recipient.public_key());
|
|
assert!(!push_filter_authorized_for_event(
|
|
&self_filter,
|
|
&event,
|
|
&recipient_hex
|
|
));
|
|
|
|
let event = EventBuilder::new(Kind::GiftWrap, "ciphertext")
|
|
.tag(Tag::public_key(recipient.public_key()))
|
|
.sign_with_keys(&sender)
|
|
.unwrap();
|
|
assert!(push_filter_authorized_for_event(
|
|
&self_filter,
|
|
&event,
|
|
&recipient_hex
|
|
));
|
|
assert!(!push_filter_authorized_for_event(
|
|
&Filter::new().author(sender.public_key()),
|
|
&event,
|
|
&recipient_hex
|
|
));
|
|
}
|
|
|
|
async fn capture(
|
|
State(seen): State<Arc<Mutex<Vec<Value>>>>,
|
|
OriginalUri(uri): OriginalUri,
|
|
headers: HeaderMap,
|
|
body: Bytes,
|
|
) -> Json<Value> {
|
|
let url: url::Url = format!("http://{}{}", headers["host"].to_str().unwrap(), uri)
|
|
.parse()
|
|
.unwrap();
|
|
assert!(!headers.contains_key("x-forwarded-proto"));
|
|
nostr::nips::nip98::verify_auth_header(
|
|
headers["authorization"].to_str().unwrap(),
|
|
&url,
|
|
nostr::nips::nip98::HttpMethod::POST,
|
|
nostr::Timestamp::now(),
|
|
Some(&body),
|
|
)
|
|
.unwrap();
|
|
seen.lock()
|
|
.await
|
|
.push(serde_json::from_slice(&body).unwrap());
|
|
Json(serde_json::json!({"status":"accepted"}))
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn gateway_retries_send_the_same_request_id_over_http() {
|
|
let seen = Arc::new(Mutex::new(Vec::new()));
|
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
|
let address = listener.local_addr().unwrap();
|
|
let server = tokio::spawn(
|
|
axum::serve(
|
|
listener,
|
|
Router::new()
|
|
.route("/v1/deliveries/apns", post(capture))
|
|
.with_state(seen.clone()),
|
|
)
|
|
.into_future(),
|
|
);
|
|
let url: url::Url = format!("http://{address}/v1/deliveries/apns")
|
|
.parse()
|
|
.unwrap();
|
|
let http = reqwest::Client::new();
|
|
let keys = nostr::Keys::generate();
|
|
let request_id = uuid::Uuid::new_v4();
|
|
for _ in 0..2 {
|
|
let body =
|
|
delivery_body("opaque-grant", request_id, Utc::now().timestamp() + 60).unwrap();
|
|
let auth = nip98_header(&keys, url.as_str(), &body).unwrap();
|
|
let response = send_gateway_request(&http, &url, body, auth).await.unwrap();
|
|
assert!(response.status().is_success());
|
|
}
|
|
server.abort();
|
|
let bodies = seen.lock().await;
|
|
assert_eq!(bodies.len(), 2);
|
|
assert_eq!(bodies[0]["request_id"], request_id.to_string());
|
|
assert_eq!(bodies[1]["request_id"], request_id.to_string());
|
|
}
|
|
}
|