From 3b23962cd398384db559296f4ddfe1eefcf3eceb Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 1 Nov 2026 11:15:37 -0500 Subject: [PATCH 1/6] refactor(relay): decode the refusal word exhaustively Refusal gains a None variant as byte 0, the value the first-cause-wins CAS already treats as "nothing recorded", and #[repr(u8)] with explicit discriminants on the three constants. A private word() encodes and a private from_word() decodes, so the two ends of the range are exhaustive together: adding a cause is now a non-exhaustive pattern in the decode rather than a word that reads as no refusal. Account::refusal and FrameSink::refusal return Refusal instead of Option, and every reader matches it: the connection loop and close_refused in relay.rs, the read's refusal guard in subscribe.rs, and send's two queue-full latches. A word outside {0, 1, 2} traps instead of reading as no refusal; Account::refuse is its only writer and writes a variant's own byte, so it cannot occur. No behaviour change: the queue-full latch, the budget refund, the close codes and the reason strings are untouched, and the 257 pre-existing crate tests pass unedited. --- .../nostream-relay/src/handlers/subscribe.rs | 14 ++- crates/nostream-relay/src/relay.rs | 11 +- crates/nostream-relay/src/sink.rs | 103 +++++++++++++----- 3 files changed, 91 insertions(+), 37 deletions(-) diff --git a/crates/nostream-relay/src/handlers/subscribe.rs b/crates/nostream-relay/src/handlers/subscribe.rs index 6644a8f3..6f24122f 100644 --- a/crates/nostream-relay/src/handlers/subscribe.rs +++ b/crates/nostream-relay/src/handlers/subscribe.rs @@ -29,6 +29,7 @@ use nostream_core::{Filter, Kind}; use crate::connection::InFlightRead; use crate::error::RelayError; use crate::handlers::{HandlerContext, HandlerFuture}; +use crate::sink::Refusal; /// `EventKinds.NIP43_INVITE_REQUEST`. const INVITE_REQUEST_KIND: Kind = Kind(28_935); @@ -282,10 +283,13 @@ async fn fetch_and_send(mut read: InFlightRead, filters: &[Filter], context: &Ha // every other exit of this function makes, and it is safe beside the loop's own // `close` because it removes the entry only while the generation it carries still // matches (`connection.rs:290-297`): whichever of the two runs second does nothing. - if context.sink.refusal().is_some() { - let _ = read.cancellation().changed().await; - context.connection.end_read(read); - return; + match context.sink.refusal() { + Refusal::None => {} + Refusal::QueueFull | Refusal::BudgetExhausted => { + let _ = read.cancellation().changed().await; + context.connection.end_read(read); + return; + } } let record = tokio::select! { @@ -509,7 +513,7 @@ mod tests { .is_err(), "the read parks once its frame is refused instead of draining the stream" ); - assert_eq!(refused.refusal(), Some(Refusal::BudgetExhausted)); + assert_eq!(refused.refusal(), Refusal::BudgetExhausted); assert_eq!( gate.available_permits(), 2, diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index f3ce3425..dda56a78 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -531,9 +531,12 @@ impl Relay { // Read at the top of every iteration, so a flag the producers set while the loop // was parked on a write is acted on as soon as that write returns. No `await` // sits between a write completing and this check. - if let Some(refusal) = sink.refusal() { - self.close_refused(&mut transport, peer, refusal).await; - break; + match sink.refusal() { + Refusal::None => {} + refusal => { + self.close_refused(&mut transport, peer, refusal).await; + break; + } } tokio::select! { @@ -653,6 +656,8 @@ impl Relay { refusal: Refusal, ) { match refusal { + // This socket was never refused, nothing to close. + Refusal::None => {} Refusal::QueueFull => { tracing::debug!(%peer, "closing a client that cannot keep up with its frame queue"); diff --git a/crates/nostream-relay/src/sink.rs b/crates/nostream-relay/src/sink.rs index 9cc26072..f39c8ec3 100644 --- a/crates/nostream-relay/src/sink.rs +++ b/crates/nostream-relay/src/sink.rs @@ -43,11 +43,14 @@ use nostream_core::messages::RelayMessage; /// Why a frame was refused: the bound that could not hold it. #[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[repr(u8)] pub enum Refusal { + /// No refusal recorded: this socket's frames are still being queued. + None = NO_REFUSAL, /// The frame did not fit this socket's `frame_queue_bytes`. - QueueFull, + QueueFull = QUEUE_FULL, /// The frame did not fit the relay's `frame_memory_budget_bytes`. - BudgetExhausted, + BudgetExhausted = BUDGET_EXHAUSTED, } /// No refusal recorded yet; the two values below are this file's only refusal causes. @@ -55,6 +58,27 @@ const NO_REFUSAL: u8 = 0; const QUEUE_FULL: u8 = 1; const BUDGET_EXHAUSTED: u8 = 2; +impl Refusal { + /// The byte this refusal is stored as. + fn word(self) -> u8 { + match self { + Self::None => NO_REFUSAL, + Self::QueueFull => QUEUE_FULL, + Self::BudgetExhausted => BUDGET_EXHAUSTED, + } + } + + /// The refusal a stored word stands for. + fn from_word(word: u8) -> Self { + match word { + NO_REFUSAL => Self::None, + QUEUE_FULL => Self::QueueFull, + BUDGET_EXHAUSTED => Self::BudgetExhausted, + other => unreachable!("refusal word {other} is not one of this file's causes"), + } + } +} + /// The process-wide bound on queued outbound payload bytes, shared by every socket of one /// relay (ADR 0008). /// @@ -159,25 +183,21 @@ struct Account { } impl Account { - /// The refusal recorded for this socket, if a sender has been refused. - fn refusal(&self) -> Option { - match self.refusal.load(Ordering::SeqCst) { - QUEUE_FULL => Some(Refusal::QueueFull), - BUDGET_EXHAUSTED => Some(Refusal::BudgetExhausted), - _ => None, - } + /// The refusal recorded for this socket: `Refusal::None` until a sender is refused. + fn refusal(&self) -> Refusal { + Refusal::from_word(self.refusal.load(Ordering::SeqCst)) } /// Records the first refusal for this socket, reporting whether this sender set it. fn refuse(&self, refusal: Refusal) -> bool { - let cause = match refusal { - Refusal::QueueFull => QUEUE_FULL, - Refusal::BudgetExhausted => BUDGET_EXHAUSTED, - }; - let first = self .refusal - .compare_exchange(NO_REFUSAL, cause, Ordering::SeqCst, Ordering::SeqCst) + .compare_exchange( + NO_REFUSAL, + refusal.word(), + Ordering::SeqCst, + Ordering::SeqCst, + ) .is_ok(); if first { @@ -262,7 +282,7 @@ impl FrameSink { return; }; - if self.account.refusal() == Some(Refusal::QueueFull) { + if self.account.refusal() == Refusal::QueueFull { return; } @@ -274,7 +294,7 @@ impl FrameSink { // in its own task and several readers share this sink, so two sends that interleaved // a load and a store would lose one increment and admit more than the bound. loop { - if self.account.refusal() == Some(Refusal::QueueFull) { + if self.account.refusal() == Refusal::QueueFull { return; } @@ -348,9 +368,9 @@ impl FrameSink { } } - /// Why a frame was refused, if the connection is over its bound. The connection task - /// closes the client once this is set. - pub fn refusal(&self) -> Option { + /// Why a frame was refused, if the connection is over its bound: `Refusal::None` until a + /// sender is refused. The connection task closes the client once this is a cause. + pub fn refusal(&self) -> Refusal { self.account.refusal() } @@ -390,7 +410,7 @@ mod tests { sink.send(&frame); sink.send(&frame); - assert_eq!(sink.refusal(), None); + assert_eq!(sink.refusal(), Refusal::None); assert_eq!(sink.queued_bytes(), 2 * len); assert!(frames.try_recv().is_ok()); assert!(frames.try_recv().is_ok()); @@ -398,7 +418,7 @@ mod tests { sink.send(&frame); - assert_eq!(sink.refusal(), Some(Refusal::QueueFull)); + assert_eq!(sink.refusal(), Refusal::QueueFull); assert_eq!( sink.queued_bytes(), 2 * len, @@ -412,7 +432,7 @@ mod tests { // Once the flag is set every later send is dropped as well. sink.send(&frame); - assert_eq!(sink.refusal(), Some(Refusal::QueueFull)); + assert_eq!(sink.refusal(), Refusal::QueueFull); assert_eq!(sink.queued_bytes(), 2 * len); // A frame larger than the bound still goes out while the queue is empty, so one @@ -421,13 +441,13 @@ mod tests { sink.send(&frame); - assert_eq!(sink.refusal(), None); + assert_eq!(sink.refusal(), Refusal::None); assert_eq!(sink.queued_bytes(), len); assert!(frames.try_recv().is_ok()); sink.send(&frame); - assert_eq!(sink.refusal(), Some(Refusal::QueueFull)); + assert_eq!(sink.refusal(), Refusal::QueueFull); assert_eq!(sink.queued_bytes(), len); assert!(frames.try_recv().is_err()); @@ -453,7 +473,7 @@ mod tests { } }); - assert_eq!(sink.refusal(), None, "every frame fits the bound"); + assert_eq!(sink.refusal(), Refusal::None, "every frame fits the bound"); assert_eq!(sink.queued_bytes(), SENDERS * len); for _ in 0..SENDERS { @@ -471,7 +491,7 @@ mod tests { sink.release(2 * len); assert_eq!(sink.queued_bytes(), 0); - assert_eq!(sink.refusal(), None); + assert_eq!(sink.refusal(), Refusal::None); sink.send(&frame); sink.send(&frame); @@ -503,7 +523,7 @@ mod tests { third.send(&frame); - assert_eq!(third.refusal(), Some(Refusal::BudgetExhausted)); + assert_eq!(third.refusal(), Refusal::BudgetExhausted); assert_eq!( third.queued_bytes(), 0, @@ -571,7 +591,7 @@ mod tests { third.send(&frame); - assert_eq!(third.refusal(), None); + assert_eq!(third.refusal(), Refusal::None); assert_eq!(budget.accounted(), len); assert!(third_frames.try_recv().is_ok()); } @@ -645,4 +665,29 @@ mod tests { assert_eq!(budget.accounted(), 0); } + + #[test] + fn every_refusal_word_is_pinned_to_its_byte() { + assert_eq!( + (NO_REFUSAL, QUEUE_FULL, BUDGET_EXHAUSTED), + (0, 1, 2), + "the bytes are the wire contract" + ); + + for refusal in [Refusal::None, Refusal::QueueFull, Refusal::BudgetExhausted] { + assert_eq!(Refusal::from_word(refusal.word()), refusal); + assert_eq!(refusal as u8, refusal.word()); + } + } + + #[test] + #[should_panic(expected = "not one of this file's")] + fn a_word_outside_the_three_causes_traps() { + // A word `refuse` cannot write: stored directly, as only a bug in this file could. + let (sink, _frames) = FrameSink::channel(1024, None); + + sink.account.refusal.store(3, Ordering::SeqCst); + + let _ = sink.refusal(); + } } From 36ec19c552c66cdbb57f9092c34480dbae55f7a3 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 1 Nov 2026 11:16:57 -0500 Subject: [PATCH 2/6] docs(relay): record the refusal word's convention where the next reader finds it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A module-doc paragraph names the one-byte convention and points at Refusal; the doc comment on Refusal (replacing the stale claim above the constants) carries the rule in three testable statements, and the decode's doc says why the trap is unreachable from the wire and what it costs if it fires — one connection's task, not the process. --- crates/nostream-relay/src/sink.rs | 29 +++++++++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/crates/nostream-relay/src/sink.rs b/crates/nostream-relay/src/sink.rs index f39c8ec3..d29d7dae 100644 --- a/crates/nostream-relay/src/sink.rs +++ b/crates/nostream-relay/src/sink.rs @@ -32,6 +32,14 @@ //! exception, so a frame that does not fit the budget is refused however empty its own //! queue. The socket's share of the budget is refunded when its last sink handle drops — //! see [`Account`] — because the next socket's admission depends on it. +//! +//! **A refusal is stored as one byte**, and [`Refusal`] owns both ends of it: [`Refusal`]'s +//! `word` encodes a variant and `from_word` decodes the byte, both exhaustively, so the byte's +//! meaning cannot drift from the enum. The type names its own "nothing refused" case, +//! [`Refusal::None`], as byte `0` — the value the first-cause-wins latch compares against — +//! instead of hiding it in an `Option`. That is what makes a new cause a compile error at +//! every reader that has to decide, rather than a word that silently reads as "no refusal"; +//! the rule is on [`Refusal`] itself and the trap beside the decode. use std::future::Future; use std::sync::Arc; @@ -41,7 +49,16 @@ use tokio::sync::{Notify, mpsc}; use nostream_core::messages::RelayMessage; -/// Why a frame was refused: the bound that could not hold it. +/// Why a frame was refused: the bound that could not hold it, or [`Refusal::None`] when no +/// sender has been refused. +/// +/// The variant is the byte it is stored as, and the whole of the contract is three statements. +/// First, `0` is no refusal, `1` is a full frame queue and `2` is an exhausted frame memory +/// budget, and no other byte is ever stored. Second, a cause is added by adding a variant +/// here, giving it a byte in `word` and answering the compiler at every reader of it — the +/// connection loop, the stored read's refusal guard and `close_refused`. Third, a word +/// outside `{0, 1, 2}` traps: it is neither a refusal nor no refusal, so no reader can read it +/// as either. #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(u8)] pub enum Refusal { @@ -53,7 +70,9 @@ pub enum Refusal { BudgetExhausted = BUDGET_EXHAUSTED, } -/// No refusal recorded yet; the two values below are this file's only refusal causes. +/// The bytes [`Refusal`] is stored as, and the sentinel the first cause wins against: `0` has +/// to stay "nothing recorded", because that is what `Account::refuse`'s compare-and-exchange +/// compares. const NO_REFUSAL: u8 = 0; const QUEUE_FULL: u8 = 1; const BUDGET_EXHAUSTED: u8 = 2; @@ -69,6 +88,12 @@ impl Refusal { } /// The refusal a stored word stands for. + /// + /// Not reachable from the wire; a word outside `{0, 1, 2}` is a bug here. The word is a + /// process-local atomic whose only writer is [`Account::refuse`], which stores a variant's + /// own byte, so nothing a client sends can reach the arm below. A trap unwinds the task + /// that read the word — one connection, not the process: no profile sets `panic = "abort"`, + /// and the connection loop observes the task's end through `JoinSet::join_next`. fn from_word(word: u8) -> Self { match word { NO_REFUSAL => Self::None, From c22b984fe93f5721f647a78e38c2b66258c851e4 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 1 Nov 2026 11:20:07 -0500 Subject: [PATCH 3/6] chore(changeset): record the exhaustive refusal decode --- .changeset/refusal-exhaustive-decode.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .changeset/refusal-exhaustive-decode.md diff --git a/.changeset/refusal-exhaustive-decode.md b/.changeset/refusal-exhaustive-decode.md new file mode 100644 index 00000000..a4987fd4 --- /dev/null +++ b/.changeset/refusal-exhaustive-decode.md @@ -0,0 +1,6 @@ +--- +--- + +refactor(relay): decode the refusal word exhaustively + +The refusal word's decode is now total and its bytes pinned: a cause the enum does not name is a compile error at every reader, and there is no behaviour change. From 2ae64bb5dcfa0a99f82061278eb23309643a23d9 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 1 Nov 2026 11:55:09 -0500 Subject: [PATCH 4/6] fix(relay): name the refusal causes in the connection loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The loop's second arm was the irrefutable binding `refusal =>`, so a fourth Refusal variant compiled there without a word and silently inherited close-and-break — while the type's doc named this loop as one of the readers the compiler asks. Measured at c22b984 by adding a fourth variant to a throwaway copy: E0004 at `word`, the stored read's refusal guard and `close_refused`, and nowhere at the loop. Naming the two causes with `refusal @ (Refusal::QueueFull | Refusal::BudgetExhausted)` keeps the binding that feeds `close_refused`, and the same experiment now reports E0004 at relay.rs:534:19. Refusal is Copy, and both causes still close exactly as they did. Corrects an overstatement round 1 found: 3b23962's body claimed the decode made adding a cause a non-exhaustive pattern "in the decode", and the loop was no pattern at all. --- crates/nostream-relay/src/relay.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/nostream-relay/src/relay.rs b/crates/nostream-relay/src/relay.rs index dda56a78..0cc88de2 100644 --- a/crates/nostream-relay/src/relay.rs +++ b/crates/nostream-relay/src/relay.rs @@ -533,7 +533,7 @@ impl Relay { // sits between a write completing and this check. match sink.refusal() { Refusal::None => {} - refusal => { + refusal @ (Refusal::QueueFull | Refusal::BudgetExhausted) => { self.close_refused(&mut transport, peer, refusal).await; break; } From 4b9a704ffdca73be802c06ff66865bf64dd97976 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 1 Nov 2026 11:55:14 -0500 Subject: [PATCH 5/6] docs(relay): say which readers the compiler asks when a cause is added MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The type's doc named the connection loop, the stored read's refusal guard and `close_refused` as the readers a new cause must answer, and the module doc called `word` and `from_word` "both exhaustively". Two clauses were false. `from_word` is total, not exhaustive — it ends `other => unreachable!` — so the compiler never asks there, and a cause added without its decode arm traps the first time that cause's byte is read instead of failing the build. With the loop now naming the two causes, the three forced readers are true of this tree, and the doc says the decode is the one site answered by hand. The module doc calls the encode exhaustive and the decode every-byte, which is what they are. Corrects the second overstatement round 1 found, and lands after the code it describes. --- crates/nostream-relay/src/sink.rs | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/crates/nostream-relay/src/sink.rs b/crates/nostream-relay/src/sink.rs index d29d7dae..7ef90848 100644 --- a/crates/nostream-relay/src/sink.rs +++ b/crates/nostream-relay/src/sink.rs @@ -34,11 +34,11 @@ //! see [`Account`] — because the next socket's admission depends on it. //! //! **A refusal is stored as one byte**, and [`Refusal`] owns both ends of it: [`Refusal`]'s -//! `word` encodes a variant and `from_word` decodes the byte, both exhaustively, so the byte's +//! `word` encodes a variant exhaustively and `from_word` decodes every byte, so the byte's //! meaning cannot drift from the enum. The type names its own "nothing refused" case, //! [`Refusal::None`], as byte `0` — the value the first-cause-wins latch compares against — -//! instead of hiding it in an `Option`. That is what makes a new cause a compile error at -//! every reader that has to decide, rather than a word that silently reads as "no refusal"; +//! instead of hiding it in an `Option`. A new cause is then a compile error at the encoder +//! and at every reader that matches, rather than a word that silently reads as "no refusal"; //! the rule is on [`Refusal`] itself and the trap beside the decode. use std::future::Future; @@ -55,10 +55,12 @@ use nostream_core::messages::RelayMessage; /// The variant is the byte it is stored as, and the whole of the contract is three statements. /// First, `0` is no refusal, `1` is a full frame queue and `2` is an exhausted frame memory /// budget, and no other byte is ever stored. Second, a cause is added by adding a variant -/// here, giving it a byte in `word` and answering the compiler at every reader of it — the -/// connection loop, the stored read's refusal guard and `close_refused`. Third, a word -/// outside `{0, 1, 2}` traps: it is neither a refusal nor no refusal, so no reader can read it -/// as either. +/// here and giving it a byte in `word`, which the compiler asks for, then answering it at +/// every reader that matches — the connection loop, the stored read's refusal guard and +/// `close_refused`. The decode is not asked: `from_word` ends in a wildcard, so a cause needs +/// its own arm there by hand, and skipping it traps the first time that cause's byte is read +/// instead of failing the build. Third, a word outside `{0, 1, 2}` traps: it is neither a +/// refusal nor no refusal, so no reader can read it as either. #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(u8)] pub enum Refusal { From 1718b15098cf4c4749c3c8aeac4758669661b892 Mon Sep 17 00:00:00 2001 From: chappie-daemon Date: Sun, 1 Nov 2026 11:55:18 -0500 Subject: [PATCH 6/6] chore(changeset): correct what the refusal decode is compile-checked by The sentence claimed a cause the enum does not name is "a compile error at every reader". The decoder is the exception to that, so it now says the compile error lands at the encoder and at every reader that matches, and that a byte `from_word` has no arm for traps instead of reading as no refusal. --- .changeset/refusal-exhaustive-decode.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.changeset/refusal-exhaustive-decode.md b/.changeset/refusal-exhaustive-decode.md index a4987fd4..d66b3d7c 100644 --- a/.changeset/refusal-exhaustive-decode.md +++ b/.changeset/refusal-exhaustive-decode.md @@ -3,4 +3,4 @@ refactor(relay): decode the refusal word exhaustively -The refusal word's decode is now total and its bytes pinned: a cause the enum does not name is a compile error at every reader, and there is no behaviour change. +The refusal word's decode is now total and its bytes pinned: a cause the enum does not name is a compile error at the encoder and at every reader that matches, a byte `from_word` has no arm for traps instead of reading as no refusal, and there is no behaviour change.