diff --git a/crates/buzz-agent/src/auth.rs b/crates/buzz-agent/src/auth.rs index 34974fbf0b..5de3bcd877 100644 --- a/crates/buzz-agent/src/auth.rs +++ b/crates/buzz-agent/src/auth.rs @@ -1,9 +1,9 @@ //! Token sources for the LLM transport layer. //! -//! [`TokenSource`] decouples request auth from `Config::api_key`: providers -//! can supply a static string ([`StaticTokenSource`]) or a refreshable OAuth -//! 2.0 PKCE engine ([`PkceOAuthTokenSource`]). Engines own their own cache -//! and refresh logic; the [`Llm`] just asks for a bearer per request. +//! `TokenSource` decouples request auth from `Config::api_key`: providers +//! can supply a static string (`StaticTokenSource`) or a refreshable OAuth +//! 2.0 PKCE engine (`PkceOAuthTokenSource`). Engines own their own cache +//! and refresh logic; the LLM layer just asks for a bearer per request. //! //! The PKCE engine implements RFC 6749 + RFC 7636 with on-disk token //! caching keyed by `sha256(discovery_url|client_id|scopes)`. It's the @@ -38,10 +38,11 @@ const TOKEN_REFRESH_LEEWAY: Duration = Duration::from_secs(60); /// We match: any longer and the user has gone to lunch. const BROWSER_AUTH_TIMEOUT: Duration = Duration::from_secs(60); -/// Asynchronous source of a bearer token. The [`Llm`] calls this per +/// Asynchronous source of a bearer token. The LLM layer calls this per /// request, so impls are expected to be cheap on the cache-hit path. #[async_trait] pub trait TokenSource: Send + Sync { + /// Return a bearer token, refreshing or launching an interactive flow if needed. async fn bearer(&self) -> Result; /// Return a bearer token from cache or refresh, **never** opening a browser. @@ -76,6 +77,7 @@ pub trait TokenSource: Send + Sync { pub struct StaticTokenSource(String); impl StaticTokenSource { + /// Create a static token source that always returns the given token. pub fn new(token: impl Into) -> Self { Self(token.into()) } @@ -96,9 +98,13 @@ impl TokenSource for StaticTokenSource { /// the token JSON lives in — separates providers' caches cleanly. #[derive(Debug, Clone)] pub struct PkceOAuthConfig { + /// RFC 8414 well-known discovery URL for the OAuth authorization server. pub discovery_url: String, + /// OAuth client id used in authorization and token requests. pub client_id: String, + /// OAuth scopes to request for the token. pub scopes: Vec, + /// Cache sub-directory under `~/.config/buzz-agent/oauth/` for this provider's tokens. pub cache_namespace: String, /// When `Some`, the engine writes tokens here instead of /// `~/.config/buzz-agent/oauth//`. Production code @@ -141,6 +147,7 @@ pub struct PkceOAuthTokenSource { } impl PkceOAuthTokenSource { + /// Construct a new PKCE token source, loading any cached token from disk. pub fn new(cfg: PkceOAuthConfig) -> Result, AgentError> { let cache_path = cache_path_for(&cfg)?; if let Some(parent) = cache_path.parent() { diff --git a/crates/buzz-agent/src/catalog.rs b/crates/buzz-agent/src/catalog.rs index aa2a121c99..c42469a4c7 100644 --- a/crates/buzz-agent/src/catalog.rs +++ b/crates/buzz-agent/src/catalog.rs @@ -3,7 +3,7 @@ //! Exposes [`discover_databricks_models`] — an async helper that lists //! available models for the `databricks` and `databricks_v2` providers //! without triggering a browser OAuth flow. Auth is acquired in-process via -//! [`build_token_source`](crate::llm::build_token_source): +//! `build_token_source`: //! //! - Static bearer (`DATABRICKS_TOKEN`): returned immediately. //! - PKCE cache hit: returned from disk without a network round-trip. @@ -22,7 +22,9 @@ use crate::{ /// label (same as `id` for Databricks — the API has no separate display name). #[derive(Debug, Clone, PartialEq, Eq)] pub struct ModelEntry { + /// Model identifier used as the picker value. pub id: String, + /// Human-readable display label. pub name: String, } diff --git a/crates/buzz-agent/src/config.rs b/crates/buzz-agent/src/config.rs index 864fa7d237..cc64bf1f3e 100644 --- a/crates/buzz-agent/src/config.rs +++ b/crates/buzz-agent/src/config.rs @@ -1,5 +1,6 @@ use std::time::Duration; +/// ACP protocol version advertised to clients during `initialize`. pub const PROTOCOL_VERSION: u32 = 2; /// Reasoning/thinking effort level for providers that support it. @@ -17,12 +18,19 @@ pub const PROTOCOL_VERSION: u32 = 2; /// - **Databricks**: routed by model family (Claude → Anthropic mapping, GPT-5 → Responses, MLflow → Chat). #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub enum ThinkingEffort { + /// Disable thinking entirely. None, + /// Minimal thinking effort. Minimal, + /// Low thinking effort. Low, + /// Medium thinking effort. Medium, + /// High thinking effort. High, + /// Extra-high thinking effort (above `High`). XHigh, + /// Maximum thinking effort supported by the provider. Max, } @@ -99,7 +107,7 @@ fn strip_catalog_prefix(model: &str) -> &str { /// Build the Anthropic thinking/effort request fields for the given model and effort level. /// /// API shape selection (per Anthropic extended-thinking support table, -/// https://platform.claude.com/docs/en/build-with-claude/extended-thinking, July 2025): +/// , July 2025): /// /// **Adaptive families** — `thinking: {type:"adaptive"}` + `output_config: {effort}`. /// These models use adaptive thinking; `thinking:{type:"adaptive"}` is required to enable @@ -637,7 +645,9 @@ pub fn parse_thinking_effort(raw: Option<&str>) -> Result } } +/// Maximum size, in bytes, of a single `session/prompt` user prompt. pub const MAX_PROMPT_BYTES: usize = 1024 * 1024; +/// Maximum combined size, in bytes, of the system prompt sent at `session/new`. pub const MAX_SYSTEM_PROMPT_BYTES: usize = 512 * 1024; /// Total per-result byte ceiling (text + images). Sized for image-bearing /// results — view_image can legitimately return multi-MiB base64 payloads. @@ -649,20 +659,27 @@ pub const MAX_TOOL_RESULT_BYTES: usize = 8 * 1024 * 1024; /// shell-output caps in sprout-dev-mcp, goose, and pi; codex defaults to /// 10 KB. Tunable via `BUZZ_AGENT_MAX_TOOL_RESULT_TEXT_BYTES`. pub const DEFAULT_TOOL_RESULT_TEXT_BYTES: usize = 50 * 1024; +/// Maximum number of tool calls allowed in a single prompt turn. pub const MAX_TOOL_CALLS_PER_TURN: usize = 64; +/// Maximum output tokens requested from the summarizing model during a context handoff. pub const HANDOFF_MAX_OUTPUT_TOKENS: u32 = 8192; +/// Maximum byte size of the preserved original-task text carried across a handoff. pub const HANDOFF_ORIGINAL_TASK_MAX_BYTES: usize = 16 * 1024; +/// Maximum number of tool names retained in the post-handoff tool summary. pub const HANDOFF_MAX_TOOL_NAMES: usize = 20; const DEFAULT_SYSTEM_PROMPT: &str = "You are buzz-agent. Use the provided tools to act. Tool calls are your only output."; #[derive(Debug, Clone, Copy, PartialEq)] +/// LLM provider to route requests to. pub enum Provider { + /// Anthropic (Claude) via the Messages API. Anthropic, + /// OpenAI or an OpenAI-compatible endpoint (Chat Completions or Responses). OpenAi, /// Databricks model serving. Routes to `{base_url}/serving-endpoints/{model}/invocations` /// with a dynamically-acquired bearer (OAuth 2.0 PKCE, or static `DATABRICKS_TOKEN`). @@ -680,25 +697,42 @@ pub enum Provider { /// provider error. #[derive(Debug, Clone, Copy, PartialEq)] pub enum OpenAiApi { + /// Use the Chat Completions API (`/v1/chat/completions`). Chat, + /// Use the Responses API (`/v1/responses`). Responses, + /// Pick the API based on the host (Responses for `*.openai.com`, Chat otherwise). Auto, } #[derive(Debug, Clone)] +/// Process-wide agent configuration, loaded from environment variables. pub struct Config { + /// LLM provider requests are routed to. pub provider: Provider, + /// System prompt prepended to every prompt turn (before hints). pub system_prompt: String, + /// Maximum number of model round-trips per prompt turn (0 = unlimited). pub max_rounds: u32, + /// Maximum output tokens requested from the model per request. pub max_output_tokens: u32, + /// Timeout for a single LLM completion request. pub llm_timeout: Duration, + /// Timeout for a single tool (MCP) call. pub tool_timeout: Duration, + /// Timeout for an MCP server's initialization handshake. pub mcp_init_timeout: Duration, + /// Maximum number of restart attempts when an MCP server exits unexpectedly. pub mcp_max_restart_attempts: u32, + /// Base delay, in milliseconds, for the exponential MCP restart backoff. pub mcp_restart_base_ms: u64, + /// Upper cap, in milliseconds, for the exponential MCP restart backoff. pub mcp_restart_max_ms: u64, + /// Maximum number of concurrent sessions the agent will accept. pub max_sessions: usize, + /// Maximum size, in bytes, of a single stdio line read from the client. pub max_line_bytes: usize, + /// Maximum total byte size of conversation history retained per session. pub max_history_bytes: usize, /// Per-tool-result cap on text content. Oversized text is middle-elided /// (head + tail kept) before entering history. Images are exempt — they @@ -712,8 +746,11 @@ pub struct Config { /// operators lower/raise it for other models. Set via /// `BUZZ_AGENT_MAX_CONTEXT_TOKENS`. pub max_context_tokens: u64, + /// Maximum number of context-summarizing handoffs allowed per prompt turn. pub max_handoffs: usize, + /// Maximum number of tool calls dispatched in parallel within a round. pub max_parallel_tools: usize, + /// Timeout for a single hook server invocation. pub hook_timeout: Duration, /// Maximum `_Stop` rejections per prompt. Default 3. Set to 0 to /// disable `_Stop` hooks entirely (agent always honors end_turn). @@ -722,9 +759,13 @@ pub struct Config { /// Default (env unset/empty) is `None` — hooks are off unless the /// operator explicitly opts in. pub hook_servers: HookServers, + /// API key or static bearer token used to authenticate with the provider. pub api_key: String, + /// Configured model identifier. pub model: String, + /// Base URL the provider API is served from. pub base_url: String, + /// `anthropic-version` header value sent on Anthropic requests. pub anthropic_api_version: String, /// OpenAI endpoint selection. See [`OpenAiApi`]. pub openai_api: OpenAiApi, @@ -734,6 +775,7 @@ pub struct Config { /// `BUZZ_AGENT_PREFER_MESH_FOR_AUTO=1`; other providers keep their /// existing `auto` semantics. pub prefer_mesh_for_auto: bool, + /// Whether to inject discovered skill/hint docs into the system prompt. pub hints_enabled: bool, /// Thinking/reasoning effort level. `None` = use provider default (no /// thinking config sent). Set via `BUZZ_AGENT_THINKING_EFFORT`. @@ -741,6 +783,7 @@ pub struct Config { } impl Config { + /// Build a [`Config`] by reading the process environment. pub fn from_env() -> Result { let databricks_host = env("DATABRICKS_HOST"); let databricks_model = env("DATABRICKS_MODEL"); @@ -840,7 +883,7 @@ impl Config { /// Construct a minimal `Config` for model-catalog discovery. /// - /// Only the fields used by [`build_token_source`](crate::llm::build_token_source) + /// Only the fields used by `build_token_source` and the catalog HTTP /// and the catalog HTTP helpers are meaningful; all others are set to /// inert defaults. Never call `from_env` for discovery — it requires /// `DATABRICKS_MODEL` and other fields that are irrelevant here. @@ -1063,8 +1106,11 @@ where /// - `a,b,c` → `Only(["a","b","c"])` #[derive(Debug, Clone)] pub enum HookServers { + /// No hook servers are eligible (hooks disabled). None, + /// Every server is eligible to receive hook calls. All, + /// Only the listed servers may receive hook calls. Only(Vec), } diff --git a/crates/buzz-agent/src/lib.rs b/crates/buzz-agent/src/lib.rs index e141b9860f..e277273020 100644 --- a/crates/buzz-agent/src/lib.rs +++ b/crates/buzz-agent/src/lib.rs @@ -1,20 +1,30 @@ #![forbid(unsafe_code)] +#![warn(missing_docs)] +//! Minimal ACP-compliant agent. +//! +//! `buzz-agent` is a non-streaming, tool-calls-as-output agent that speaks the +//! Agent Client Protocol (ACP) over stdio. It wires an LLM provider to MCP tool +//! servers and exposes session management, model switching, mid-turn steering, +//! and context-summarizing handoffs. mod agent; +/// Provider-agnostic OAuth/PKCE and static token sources. pub mod auth; mod builtin; +/// Databricks model-catalog discovery. pub mod catalog; +/// Process-wide configuration loaded from environment variables. pub mod config; mod handoff; mod hints; mod llm; mod mcp; +/// Wire protocol, error, and message types shared across modules. pub mod types; mod wire; pub use catalog::{discover_databricks_models, ModelEntry, DATABRICKS_V2_KNOWN_MODELS}; pub use config::Provider; pub use types::AgentError; - /// Environment keys the Windows Git Bash resolver may inspect. `spawn_one()` /// forwards every key in this list into its otherwise-cleared MCP child; Doctor /// uses the same contract so a ready agent can always start its shell tool. @@ -107,6 +117,10 @@ fn die(msg: String) -> ! { std::process::exit(2); } +/// Entry point for the `buzz-agent` binary. +/// +/// Dispatches the `auth` subcommand (interactive provider login) when invoked +/// as `buzz-agent auth `; otherwise runs the ACP stdio server. pub fn run() -> Result<(), Box> { let args: Vec = std::env::args().collect(); if matches!(args.get(1).map(String::as_str), Some("auth")) { diff --git a/crates/buzz-agent/src/types.rs b/crates/buzz-agent/src/types.rs index d29e975e03..446dc96210 100644 --- a/crates/buzz-agent/src/types.rs +++ b/crates/buzz-agent/src/types.rs @@ -14,9 +14,17 @@ use serde_json::Value; const IMAGE_CONTEXT_TOKEN_EQUIV: usize = 16 * 1024; #[derive(Debug, Clone)] +/// A single content part returned by a tool call. pub enum ToolResultContent { + /// Plain-text content. Text(String), - Image { data: String, mime_type: String }, + /// Base64-encoded image data plus its MIME type. + Image { + /// Base64-encoded image payload. + data: String, + /// MIME type of the image (e.g. `image/png`). + mime_type: String, + }, } impl ToolResultContent { @@ -35,10 +43,10 @@ impl ToolResultContent { /// Token-equivalent context-window pressure, in bytes (the handoff gate /// maps bytes→tokens at 1:1). Identical to [`Self::estimated_bytes`] for - /// text, but an image is charged a flat [`IMAGE_CONTEXT_TOKEN_EQUIV`] - /// budget rather than its base64 length — providers bill it as visual - /// tiles (~2K tokens), so counting `data.len()` over-counts by ~1500× and - /// forces a handoff on a single image. + /// text, but an image is charged a flat per-image token budget + /// (`IMAGE_CONTEXT_TOKEN_EQUIV`) rather than its base64 length — providers + /// bill it as visual tiles (~2K tokens), so counting `data.len()` over- + /// counts by ~1500× and forces a handoff on a single image. pub fn context_pressure_bytes(&self) -> usize { match self { Self::Text(s) => s.len(), @@ -46,6 +54,8 @@ impl ToolResultContent { } } + /// Return the content as plain text. Image content is rendered as a + /// size summary placeholder since it has no textual form. pub fn as_text_lossy(&self) -> String { match self { Self::Text(s) => s.clone(), @@ -57,16 +67,23 @@ impl ToolResultContent { } #[derive(Debug, Clone)] +/// One item in a session's conversation history. pub enum HistoryItem { + /// A user turn. User(String), + /// An assistant turn with its text and any tool calls it made. Assistant { + /// Assistant text output for this turn. text: String, + /// Tool calls requested by the assistant in this turn. tool_calls: Vec, }, + /// A completed tool call result fed back to the model. ToolResult(ToolResult), } impl HistoryItem { + /// Estimated serialized byte size of this item. pub fn estimated_bytes(&self) -> usize { self.size_with(ToolResultContent::estimated_bytes) } @@ -104,20 +121,29 @@ impl HistoryItem { } #[derive(Debug, Clone)] +/// A tool call requested by the model. pub struct ToolCall { + /// Provider-assigned call id used to correlate results with this call. pub provider_id: String, + /// Name of the tool to invoke. pub name: String, + /// JSON arguments to pass to the tool. pub arguments: Value, } #[derive(Debug, Clone)] +/// The outcome of executing a tool call. pub struct ToolResult { + /// Provider-assigned call id this result corresponds to. pub provider_id: String, + /// Content blocks returned by the tool. pub content: Vec, + /// Whether the tool execution reported an error. pub is_error: bool, } impl ToolResult { + /// Join all content blocks into a single text representation. pub fn text(&self) -> String { self.content .iter() @@ -128,9 +154,13 @@ impl ToolResult { } #[derive(Debug, Clone)] +/// The parsed response from an LLM completion request. pub struct LlmResponse { + /// Assistant text output. pub text: String, + /// Tool calls the model requested, if any. pub tool_calls: Vec, + /// Why the model stopped generating. pub stop: ProviderStop, /// Total input tokens the provider reported for this request, or `None` /// if the response carried no usage. For Anthropic/Databricks this is the @@ -155,31 +185,48 @@ pub struct LlmResponse { } #[derive(Debug, Clone, Copy, PartialEq)] +/// Stop reason reported by the upstream provider. pub enum ProviderStop { + /// The model finished its turn naturally. EndTurn, + /// The model stopped to emit tool calls. ToolUse, + /// The `max_tokens` limit was reached. MaxTokens, + /// The model refused the request. Refusal, + /// Any other provider-reported stop reason. Other, } #[derive(Debug, Clone)] +/// Definition of a tool exposed to the model. pub struct ToolDef { + /// Tool name as seen by the model. pub name: String, + /// Human-readable description of what the tool does. pub description: String, + /// JSON schema describing the tool's arguments. pub input_schema: Value, } #[derive(Debug, Clone, Copy, PartialEq)] +/// Why an agent prompt turn ended, as surfaced to the client. pub enum StopReason { + /// The model finished its turn naturally. EndTurn, + /// The turn was cancelled by the client. Cancelled, + /// The `max_tokens` limit was reached. MaxTokens, + /// The configured maximum number of turn requests was reached. MaxTurnRequests, + /// The model refused the request. Refusal, } impl StopReason { + /// Return the ACP wire string for this stop reason. pub fn as_wire(self) -> &'static str { match self { Self::EndTurn => "end_turn", @@ -192,41 +239,62 @@ impl StopReason { } #[derive(Debug, Deserialize, Clone)] +/// Specification of an stdio MCP server to spawn. pub struct McpServerStdio { + /// Logical name of the MCP server. pub name: String, + /// Executable command to run. pub command: String, + /// Command-line arguments to pass to the command. #[serde(default)] pub args: Vec, + /// Environment variables to set on the child process. #[serde(default)] pub env: Vec, } #[derive(Debug, Deserialize, Clone)] +/// A name/value pair for a child process environment variable. pub struct EnvVar { + /// Environment variable name. pub name: String, + /// Environment variable value. pub value: String, } #[derive(Debug, Deserialize, Clone)] #[serde(tag = "type", rename_all = "snake_case")] +/// A content block from a user prompt. pub enum ContentBlock { + /// A text block. Text { + /// The text content. text: String, }, + /// A reference to an external resource. ResourceLink { + /// URI of the referenced resource. uri: String, }, + /// A content block type this agent does not handle. #[serde(other)] Unsupported, } #[derive(Debug)] +/// Errors that can occur while driving an agent prompt turn. pub enum AgentError { + /// The client sent malformed or invalid request parameters. InvalidParams(String), + /// A generic LLM transport or parsing error. Llm(String), + /// An LLM authentication or authorization failure. LlmAuth(String), + /// The requested model was not found or not available. LlmModelNotFound(String), + /// An MCP server transport or protocol error. Mcp(String), + /// The turn was cancelled. Cancelled, } @@ -246,6 +314,7 @@ impl std::fmt::Display for AgentError { impl std::error::Error for AgentError {} impl AgentError { + /// Return the JSON-RPC error code this error maps to. pub fn json_rpc_code(&self) -> i32 { match self { Self::InvalidParams(_) => -32602, @@ -256,6 +325,8 @@ impl AgentError { } } +/// Truncate `s` to at most `max` bytes on a UTF-8 char boundary, appending a +/// `[truncated]` marker when truncation occurs. pub fn clamp(mut s: String, max: usize) -> String { if s.len() <= max { return s; diff --git a/crates/buzz-conformance/src/lib.rs b/crates/buzz-conformance/src/lib.rs index 3e1cfe13e3..1b376d084d 100644 --- a/crates/buzz-conformance/src/lib.rs +++ b/crates/buzz-conformance/src/lib.rs @@ -1,4 +1,5 @@ //! Runtime trace schema + independent replay checker for +#![warn(missing_docs)] //! `docs/spec/MultiTenantRelay.tla`. //! //! North star (from the runtime-formal-compliance skill): don't ask "did the diff --git a/crates/buzz-media/src/auth.rs b/crates/buzz-media/src/auth.rs index c6fff2be47..848545e219 100644 --- a/crates/buzz-media/src/auth.rs +++ b/crates/buzz-media/src/auth.rs @@ -5,7 +5,9 @@ use crate::error::MediaError; /// Blossom kind:24242 verbs Buzz currently accepts. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum BlossomVerb { + /// Upload (write) a blob to the server. Upload, + /// Get (read) a blob from the server. Get, } diff --git a/crates/buzz-media/src/bucket_index.rs b/crates/buzz-media/src/bucket_index.rs index bb83dc517f..4a03876b2a 100644 --- a/crates/buzz-media/src/bucket_index.rs +++ b/crates/buzz-media/src/bucket_index.rs @@ -33,15 +33,31 @@ use crate::error::MediaError; #[derive(Debug, Clone, PartialEq, Eq)] pub enum KeyClass { /// `{sha256}.thumb.jpg` — attributed to the blob's sha. - Thumb { sha256: String }, + Thumb { + /// The 64-char lowercase-hex SHA-256 digest from the key stem. + sha256: String, + }, /// `{sha256}.{ext}` — physical bytes, logical join key. - Blob { sha256: String, ext: String }, + Blob { + /// The 64-char lowercase-hex SHA-256 digest from the key stem. + sha256: String, + /// The blob's file extension (1-8 mixed-case alphanumeric chars). + ext: String, + }, /// `_meta/{community}/{sha256}.json` — the (community, sha) binding. - Sidecar { community: Uuid, sha256: String }, + Sidecar { + /// The canonical lowercase community UUID parsed from the key path. + community: Uuid, + /// The 64-char lowercase-hex SHA-256 digest bound to this community. + sha256: String, + }, /// `_uploads/{community}/{sha256}/{event_id}.json` — fleet physical only. Auxiliary { + /// The canonical lowercase community UUID parsed from the key path. community: Uuid, + /// The 64-char lowercase-hex SHA-256 digest of the uploaded blob. sha256: String, + /// The Crockford-base32 ULID event id from the key's last segment. event_id: String, }, /// Anything that doesn't match one of the four strict shapes above. @@ -195,7 +211,9 @@ fn parse_auxiliary_key(key: &str) -> Option<(Uuid, String, String)> { /// Per-community logical storage: bytes and object count of bound shas. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct CommunityStorage { + /// Total logical bytes attributed to this community. pub bytes: u64, + /// Number of logical objects attributed to this community. pub objects: u64, } @@ -204,16 +222,20 @@ pub struct CommunityStorage { /// data — no I/O, cheap to clone into a cached snapshot. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct BucketSnapshot { - /// Every listed object, every class (kind=physical). + /// Total bytes of every listed object, every class (kind=physical). pub physical_bytes: u64, + /// Total count of every listed object, every class (kind=physical). pub physical_objects: u64, - /// Sum of per-community logical bytes/objects (kind=logical). Bills a + /// Sum of per-community logical bytes (kind=logical). Bills a /// blob bound to N communities N times — intentional (D-EXT/D-COUNT). pub logical_bytes: u64, + /// Sum of per-community logical object counts (kind=logical). pub logical_objects: u64, + /// Per-community logical breakdown keyed by community UUID. pub per_community: HashMap, - /// Blob shas with zero sidecar binding in any community. + /// Total bytes of blob shas with zero sidecar binding in any community. pub orphan_blob_bytes: u64, + /// Count of blob shas with zero sidecar binding in any community. pub orphan_blob_count: u64, /// Sidecar bindings whose sha has no blob bytes at all. pub orphan_sidecar_count: u64, @@ -221,7 +243,9 @@ pub struct BucketSnapshot { pub multi_variant_shas: u64, /// Total bytes of ALL blob variants belonging to anomalous shas. pub multi_variant_bytes: u64, + /// Total bytes of keys that did not match any known class. pub unknown_key_bytes: u64, + /// Count of keys that did not match any known class. pub unknown_key_objects: u64, } @@ -341,7 +365,12 @@ impl BucketAggregate { pub enum SweepError { /// Cumulative listed-object count exceeded `cap` mid-listing. #[error("object cap exceeded: {seen} listed objects > cap {cap}")] - CapExceeded { seen: u64, cap: u64 }, + CapExceeded { + /// Number of objects listed when the cap was breached. + seen: u64, + /// The configured maximum object count for a single sweep. + cap: u64, + }, /// The page source (S3, or a test double) failed. #[error("storage error during listing: {0}")] Storage(#[from] MediaError), @@ -362,8 +391,11 @@ pub enum SweepError { /// fold below can be driven by a synthetic closure in tests. #[derive(Debug, Clone, Default)] pub struct Page { + /// The `(key, size)` pairs returned by this listing page. pub objects: Vec<(String, u64)>, + /// S3 continuation token for the next page, if the listing is truncated. pub next_continuation_token: Option, + /// Whether more pages remain to be fetched. pub is_truncated: bool, } diff --git a/crates/buzz-media/src/error.rs b/crates/buzz-media/src/error.rs index c3d180402f..ba873f1ca5 100644 --- a/crates/buzz-media/src/error.rs +++ b/crates/buzz-media/src/error.rs @@ -6,60 +6,93 @@ use axum::response::{IntoResponse, Response}; /// Errors from media operations. #[derive(Debug, thiserror::Error)] pub enum MediaError { + /// The uploaded bytes did not match any known magic-number signature. #[error("unknown content type")] UnknownContentType, + /// The detected content type is not on the allowlist. #[error("disallowed content type: {0}")] DisallowedContentType(String), + /// The uploaded file exceeds the configured maximum size. #[error("file too large: {size} bytes (max {max})")] - FileTooLarge { size: u64, max: u64 }, + FileTooLarge { + /// Actual size of the uploaded file in bytes. + size: u64, + /// Maximum allowed size in bytes. + max: u64, + }, + /// The decoded image exceeds the configured pixel-dimension cap. #[error("image dimensions too large")] ImageTooLarge, + /// The image bytes could not be decoded. #[error("invalid image data")] InvalidImage, + /// The media embeds forbidden metadata or a non-canonical metadata channel. #[error("media contains metadata or a non-canonical metadata channel")] MetadataForbidden, + /// The provided signature failed verification. #[error("invalid signature")] InvalidSignature, + /// The auth event has the wrong Nostr kind. #[error("invalid auth event kind")] InvalidAuthKind, + /// The auth event carries an unsupported Blossom verb. #[error("invalid auth verb")] InvalidAuthVerb, + /// A required Nostr tag is absent from the auth event. #[error("missing required tag: {0}")] MissingTag(&'static str), + /// The computed SHA-256 does not match the claimed hash. #[error("hash mismatch")] HashMismatch, + /// The auth event references a different server URL. #[error("server mismatch")] ServerMismatch, + /// The auth event's expiration timestamp has passed. #[error("token expired")] TokenExpired, + /// The auth event's `created_at` falls outside the acceptance window. #[error("timestamp out of window")] TimestampOutOfWindow, + /// An underlying storage backend (S3) operation failed. #[error("storage error: {0}")] StorageError(String), + /// A generic internal error not covered by a more specific variant. #[error("internal error")] Internal, + /// The requested blob does not exist in storage. #[error("not found")] NotFound, + /// The request carried no authorization header. #[error("missing authorization header")] MissingAuth, + /// The authorization scheme is not supported. #[error("invalid authorization scheme")] InvalidAuthScheme, + /// The authorization header contained invalid base64. #[error("invalid base64 encoding")] InvalidBase64, + /// The decoded authorization event is malformed. #[error("invalid auth event")] InvalidAuthEvent, + /// The authenticated principal is not permitted to perform this action. #[error("unauthorized")] Unauthorized, + /// The auth event lacks the required Blossom scope for this operation. #[error("insufficient scope")] InsufficientScope, + /// The pubkey is not a member of the required relay. #[error("relay membership required")] RelayMembershipRequired, + /// The auth token has been revoked. #[error("token revoked")] TokenRevoked, + /// The auth event's pubkey does not match the requested resource's owner. #[error("pubkey mismatch")] PubkeyMismatch, + /// The uploader has exceeded the per-window upload rate limit. #[error("upload rate limit exceeded")] UploadRateLimitExceeded, + /// The uploader has reached the concurrent-upload limit. #[error("upload concurrency limit reached")] UploadConcurrencyLimitReached, /// A video/audio track does not use the canonical H.264/AAC codecs. diff --git a/crates/buzz-media/src/lib.rs b/crates/buzz-media/src/lib.rs index ac05ea6d51..4fdc1015e2 100644 --- a/crates/buzz-media/src/lib.rs +++ b/crates/buzz-media/src/lib.rs @@ -1,3 +1,5 @@ +#![warn(missing_docs)] + //! Media storage, validation, and thumbnail generation for Buzz. //! //! Library crate — no Axum dependency for handlers. Axum handlers live in `buzz-relay`. diff --git a/crates/buzz-media/src/storage.rs b/crates/buzz-media/src/storage.rs index 0e9809af2f..f566bf136b 100644 --- a/crates/buzz-media/src/storage.rs +++ b/crates/buzz-media/src/storage.rs @@ -69,7 +69,7 @@ impl MediaStorage { /// Store an object from a byte slice. /// /// Used for images, sidecars, and thumbnails. For large video files use - /// [`put_file`] to avoid loading the entire blob into RAM. + /// [`MediaStorage::put_file`] to avoid loading the entire blob into RAM. pub async fn put(&self, key: &str, bytes: &[u8], content_type: &str) -> Result<(), MediaError> { self.bucket .put_object_with_content_type(key, bytes, content_type) @@ -377,6 +377,7 @@ mod tests { /// Metadata returned by HEAD — just enough for BUD-01 response headers. pub struct BlobHeadMeta { + /// Object size in bytes. pub size: u64, } diff --git a/crates/buzz-media/src/validation.rs b/crates/buzz-media/src/validation.rs index ee940dfb24..f457f19202 100644 --- a/crates/buzz-media/src/validation.rs +++ b/crates/buzz-media/src/validation.rs @@ -233,7 +233,7 @@ pub struct VideoMeta { /// Validate uploaded bytes for the **image** upload path. /// /// Checks magic bytes, MIME allowlist (images only), size, and pixel dimensions. -/// Rejects `video/mp4` — video uploads must use [`process_video_upload`] which +/// Rejects `video/mp4` — video uploads must use [`crate::process_video_upload`] which /// has its own magic-byte check and full MP4 validation pipeline. pub fn validate_content(bytes: &[u8], config: &MediaConfig) -> Result { // 1. Magic bytes — never trust Content-Type header diff --git a/crates/buzz-pair-relay/src/lib.rs b/crates/buzz-pair-relay/src/lib.rs index 8d30a4b1ec..0919354a69 100644 --- a/crates/buzz-pair-relay/src/lib.rs +++ b/crates/buzz-pair-relay/src/lib.rs @@ -1,3 +1,5 @@ +#![warn(missing_docs)] + //! Ephemeral sidecar relay for NIP-AB device pairing handshakes. //! //! Accepts WebSocket connections, matches incoming kind:24134 events against @@ -101,6 +103,7 @@ struct Sub { writer_tx: mpsc::Sender, } +/// Core relay state: live subscriptions, connection counter, and bounded dedup/delivery maps. pub struct Relay { subs: Mutex>, conn_count: AtomicU32, @@ -118,6 +121,7 @@ impl Default for Relay { } impl Relay { + /// Create a new empty relay with no subscriptions or tracked state. pub fn new() -> Self { Self { subs: Mutex::new(Vec::new()), diff --git a/crates/buzz-push-gateway/src/apns.rs b/crates/buzz-push-gateway/src/apns.rs index 8f6f182000..f6849bfc71 100644 --- a/crates/buzz-push-gateway/src/apns.rs +++ b/crates/buzz-push-gateway/src/apns.rs @@ -77,7 +77,9 @@ pub fn classify(code: u16, reason: Option<&str>, timestamp: Option) -> Deli /// content; the concrete transport always uses `APNS_RECONNECT_PAYLOAD`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct DeliveryAttempt { + /// Unique id of the delivery request. pub request_id: uuid::Uuid, + /// Unix timestamp at which the request expires. pub expires_at: i64, } diff --git a/crates/buzz-push-gateway/src/app_attest.rs b/crates/buzz-push-gateway/src/app_attest.rs index ebb1fc56bc..95fdfce103 100644 --- a/crates/buzz-push-gateway/src/app_attest.rs +++ b/crates/buzz-push-gateway/src/app_attest.rs @@ -13,27 +13,36 @@ const APPLE_APP_ATTEST_ROOT_PEM_SHA256: [u8; 32] = [ 0x6a, 0xed, 0x4a, 0xd4, 0x49, 0x0e, 0x1e, 0x92, 0xc0, 0x5c, 0xb3, 0x55, 0x21, 0x2a, 0x50, 0x13, ]; +/// Result of a successful attestation verification: the attested key material. #[derive(Debug, Clone, PartialEq, Eq)] pub struct VerifiedAttestation { + /// The attested key id (raw 32 bytes). pub key_id: Vec, + /// The attested P-256 public key (raw X9.62 bytes). pub public_key: Vec, } +/// Result of a successful assertion verification: the monotonic counter. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct VerifiedAssertion { + /// The assertion's monotonic counter (must strictly exceed the previous). pub counter: u32, } +/// Errors raised by App Attest verification. #[derive(Debug, Error)] pub enum AppAttestError { + /// The attestation/assertion was malformed, unverifiable, or replayed. #[error("invalid app attestation or assertion")] Invalid, } +/// Apple App Attest verifier bound to an app id and pinned root certificate. #[derive(Clone)] pub struct AppAttestVerifier { app_id: String, apple_root_cert_pem: Vec, } impl AppAttestVerifier { + /// Create a verifier for `app_id`, pinning the Apple App Attest root cert. pub fn new(app_id: String, apple_root_cert_pem: Vec) -> Result { if app_id.is_empty() || Sha256::digest(&apple_root_cert_pem).as_slice() != APPLE_APP_ATTEST_ROOT_PEM_SHA256 @@ -81,6 +90,7 @@ impl AppAttestVerifier { public_key: public_key.to_vec(), }) } + /// Verify an App Attest assertion, enforcing counter advancement and challenge match. pub fn verify_assertion( &self, assertion_b64: &str, diff --git a/crates/buzz-push-gateway/src/authority.rs b/crates/buzz-push-gateway/src/authority.rs index 36c220885c..62ab3d1b69 100644 --- a/crates/buzz-push-gateway/src/authority.rs +++ b/crates/buzz-push-gateway/src/authority.rs @@ -11,72 +11,120 @@ use std::sync::Mutex; use thiserror::Error; use uuid::Uuid; +/// Single-use enrollment challenge issued to a client. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Challenge { + /// Unique challenge id. pub id: Uuid, + /// 32-byte random challenge value. pub value: [u8; 32], + /// Unix timestamp at which the challenge expires. pub expires_at: i64, } +/// Installation record at creation time, before it has been persisted. #[derive(Debug, Clone, PartialEq, Eq)] pub struct NewInstallation { + /// Unique installation id. pub id: Uuid, + /// App Attest key id (raw 32 bytes). pub app_attest_key_id: Vec, + /// App Attest P-256 public key (raw bytes). pub app_attest_public_key: Vec, + /// Current assertion counter. pub assertion_counter: u32, + /// App profile bound to this installation. pub profile: AppProfile, + /// Encrypted APNs token ciphertext. pub token_ciphertext: Vec, + /// SHA-256 fingerprint of the plaintext token. pub token_fingerprint: [u8; 32], + /// Endpoint epoch for this token generation. pub endpoint_epoch: i64, + /// Unix timestamp at which the installation expires. pub expires_at: i64, } +/// Durable installation authority record. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Installation { + /// Unique installation id. pub id: Uuid, + /// App Attest key id (raw 32 bytes). pub app_attest_key_id: Vec, + /// App Attest P-256 public key (raw bytes). pub app_attest_public_key: Vec, + /// Current assertion counter. pub assertion_counter: u32, + /// App profile bound to this installation. pub profile: AppProfile, + /// Encrypted APNs token ciphertext. pub token_ciphertext: Vec, + /// SHA-256 fingerprint of the plaintext token. pub token_fingerprint: [u8; 32], + /// Endpoint epoch for this token generation. pub endpoint_epoch: i64, + /// Unix timestamp at which the installation expires. pub expires_at: i64, + /// Whether the installation has been revoked. pub revoked: bool, } +/// Relay delegation granting a relay the right to deliver for an installation. #[derive(Debug, Clone, PartialEq, Eq)] pub struct Delegation { + /// Unique delegation id. pub id: Uuid, + /// The installation being delegated. pub installation_id: Uuid, + /// Nostr relay public key (hex) receiving the delegation. pub relay_pubkey: String, + /// Endpoint epoch the delegation is bound to. pub endpoint_epoch: i64, + /// Monotonic delegation generation (revocation bumps it). pub generation: i64, + /// Unix timestamp from which the delegation is valid. pub not_before: i64, + /// Unix timestamp at which the delegation expires. pub expires_at: i64, + /// Whether the delegation has been revoked. pub revoked: bool, } +/// Locked authority required to perform one delivery. #[derive(Debug, Clone, PartialEq, Eq)] pub struct DeliveryAuthority { + /// The delegation authorizing delivery. pub delegation_id: Uuid, + /// The installation the delivery targets. pub installation_id: Uuid, + /// Nostr relay public key (hex) performing delivery. pub relay_pubkey: String, + /// App profile of the installation. pub profile: AppProfile, + /// Encrypted APNs token ciphertext. pub token_ciphertext: Vec, + /// Endpoint epoch of the token. pub endpoint_epoch: i64, + /// Delegation generation at authorization time. pub generation: i64, + /// Unix timestamp at which the authority expires. pub expires_at: i64, } +/// One-use authority to perform a delivery, returned by `authorize_delivery`. #[derive(Debug)] pub struct DeliveryPermit { + /// The locked delivery authority. pub authority: DeliveryAuthority, + /// Nostr relay public key (hex) performing the delivery. pub relay_pubkey: String, + /// Unique id of the delivery request. pub request_id: Uuid, } impl DeliveryPermit { + /// Assemble a permit from its authority, relay, and request id. pub fn new(authority: DeliveryAuthority, relay_pubkey: String, request_id: Uuid) -> Self { Self { authority, @@ -86,16 +134,22 @@ impl DeliveryPermit { } } +/// Outcome of a delivery as reported to `finish_delivery`. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DeliveryDisposition { + /// The delivery is complete; no retry. Terminal, + /// The delivery may be retried. Retryable, } +/// Errors raised by the authority store. #[derive(Debug, Error, Clone, Copy, PartialEq, Eq)] pub enum AuthorityError { + /// The authority state rejected the request. #[error("authority state rejected the request")] Rejected, + /// The authority store was unavailable. #[error("authority store unavailable")] Unavailable, } @@ -106,25 +160,32 @@ pub enum AuthorityError { pub trait AuthorityStore: Send + Sync { /// Readiness must fail closed when durable authority cannot participate. async fn ready(&self) -> Result<(), AuthorityError>; + /// Persist a freshly issued challenge. async fn put_challenge(&self, challenge: Challenge) -> Result<(), AuthorityError>; + /// Consume a challenge by id+value if it has not expired. async fn consume_challenge( &self, id: Uuid, value: [u8; 32], now: i64, ) -> Result<(), AuthorityError>; + /// Persist a new installation. async fn create_installation( &self, installation: NewInstallation, ) -> Result<(), AuthorityError>; + /// Load an installation by id (fails if expired). async fn installation(&self, id: Uuid, now: i64) -> Result; + /// Advance an installation's assertion counter from `previous` to `next`. async fn advance_assertion_counter( &self, installation_id: Uuid, previous: u32, next: u32, ) -> Result<(), AuthorityError>; + /// Insert or update a delegation (bumping generation on change). async fn upsert_delegation(&self, delegation: Delegation) -> Result<(), AuthorityError>; + /// Rotate an installation's endpoint/token, expecting `expected_epoch`. async fn rotate_endpoint( &self, installation_id: Uuid, @@ -133,12 +194,14 @@ pub trait AuthorityStore: Send + Sync { token_ciphertext: Vec, token_fingerprint: [u8; 32], ) -> Result<(), AuthorityError>; + /// Revoke a relay's delegation, bumping it to `new_generation`. async fn revoke_delegation( &self, installation_id: Uuid, relay_pubkey: &str, new_generation: i64, ) -> Result<(), AuthorityError>; + /// Revoke an installation, expecting `expected_epoch` and rotating to `new_epoch`. async fn revoke_installation( &self, installation_id: Uuid, diff --git a/crates/buzz-push-gateway/src/config.rs b/crates/buzz-push-gateway/src/config.rs index c6194edbcb..1b8f214c96 100644 --- a/crates/buzz-push-gateway/src/config.rs +++ b/crates/buzz-push-gateway/src/config.rs @@ -6,39 +6,61 @@ use std::{ }; use thiserror::Error; +/// A named 256-bit key entry in a keyring. #[derive(Debug, Clone, PartialEq, Eq)] pub struct KeyConfig { + /// Stable identifier for the key (used for KID selection). pub id: String, + /// Raw 32-byte key material. pub key: Vec, } +/// Resolved gateway configuration, loaded from environment variables. #[derive(Debug, Clone)] pub struct Config { + /// Address the delivery HTTP server binds to. pub bind_addr: SocketAddr, + /// Address the health/metrics server binds to. pub health_addr: SocketAddr, + /// Public https URL relays address deliveries to. pub public_delivery_url: url::Url, + /// Maximum lifetime (seconds) of a delivery grant. pub max_grant_lifetime_seconds: i64, + /// Maximum lifetime (seconds) of an installation. pub max_installation_lifetime_seconds: i64, + /// Per-endpoint quota window length (seconds). pub endpoint_quota_window_seconds: i64, + /// Maximum deliveries per endpoint per quota window. pub endpoint_quota_max_deliveries: i64, + /// App profiles the gateway will accept. pub enabled_profiles: HashSet, + /// Postgres connection string. pub database_url: String, + /// App Attest app id (TeamID.bundleID). pub app_attest_app_id: String, + /// Path to the pinned Apple App Attest root certificate. pub app_attest_root_cert_path: PathBuf, /// Ordered current key first, followed by decrypt-only predecessors. pub grant_keys: Vec, /// Independent token-custody keyring. These keys MUST NOT be reused for /// externally presented delivery capabilities. pub token_keys: Vec, + /// Path to the APNs provider signing key (.p8). pub apns_key_path: PathBuf, + /// APNs provider key id. pub apns_key_id: String, + /// APNs team id. pub apns_team_id: String, + /// APNs topic (bundle id). pub apns_topic: String, } +/// Errors raised while resolving configuration. #[derive(Debug, Error)] pub enum ConfigError { + /// A required environment variable was missing. #[error("missing required environment variable {0}")] Missing(&'static str), + /// An environment variable had an invalid value. #[error("invalid environment variable {0}")] Invalid(&'static str), } @@ -76,9 +98,11 @@ fn parse_keyring( Ok(keys) } impl Config { + /// Resolve configuration from the process environment. pub fn from_env() -> Result { Self::from_map(&std::env::vars().collect()) } + /// Resolve configuration from an explicit key/value map (tests). pub fn from_map(e: &HashMap) -> Result { fn req<'a>( e: &'a HashMap, diff --git a/crates/buzz-push-gateway/src/grant.rs b/crates/buzz-push-gateway/src/grant.rs index 54a29bac3d..5b1030b615 100644 --- a/crates/buzz-push-gateway/src/grant.rs +++ b/crates/buzz-push-gateway/src/grant.rs @@ -13,29 +13,36 @@ use thiserror::Error; const AAD_PREFIX: &[u8] = b"buzz-stateful-delivery-capability-v1:"; const MAX_KEY_ID_BYTES: usize = 32; +/// A single named AES-256-GCM key used to seal/open endpoint grants. #[derive(Clone)] pub struct GrantKey { id: String, cipher: Aes256Gcm, } +/// Ordered keyring: one current seal/open key plus decrypt-only predecessors. #[derive(Clone)] pub struct GrantKeyring { current: GrantKey, predecessors: HashMap, } +/// Errors raised while minting or opening endpoint grants. #[derive(Debug, Error)] pub enum GrantError { + /// The grant or key material was malformed. #[error("invalid endpoint grant")] Invalid, + /// The keyring contained two keys with the same id. #[error("duplicate grant key id")] DuplicateKeyId, + /// The keyring contained no keys. #[error("grant keyring is empty")] EmptyKeyring, } impl GrantKey { + /// Construct a grant key from its id and 32-byte raw material. pub fn new(id: impl Into, key: &[u8]) -> Result { let id = id.into(); if id.is_empty() diff --git a/crates/buzz-push-gateway/src/http.rs b/crates/buzz-push-gateway/src/http.rs index 0564972c07..def203f304 100644 --- a/crates/buzz-push-gateway/src/http.rs +++ b/crates/buzz-push-gateway/src/http.rs @@ -33,20 +33,34 @@ use std::{ use tower::limit::ConcurrencyLimitLayer; use tower_http::{limit::RequestBodyLimitLayer, timeout::TimeoutLayer}; +/// Shared application state threaded through every handler. #[derive(Clone)] pub struct AppState { + /// Keyring used to mint/open endpoint grants. pub grant_keyring: Arc, + /// App Attest verifier bound to the configured app id. pub app_attest: Arc, + /// Durable authority store. pub authority: Arc, + /// Keyring used to seal/open APNs tokens. pub token_keyring: Arc, + /// APNs transport used to send deliveries. pub transport: Arc, + /// Public https delivery URL relays address. pub delivery_url: url::Url, + /// Maximum lifetime (seconds) of a delivery grant. pub max_grant_lifetime_seconds: i64, + /// Maximum lifetime (seconds) of an installation. pub max_installation_lifetime_seconds: i64, + /// Per-endpoint quota window length (seconds). pub endpoint_quota_window_seconds: i64, + /// Maximum deliveries per endpoint per quota window. pub endpoint_quota_max_deliveries: i64, + /// App profiles the gateway accepts. pub enabled_profiles: HashSet, + /// Wall-clock source (injectable for tests). pub now: fn() -> i64, + /// Gate flipped to false during shutdown to refuse new work. pub accepting: Arc, } fn error(status: StatusCode, code: &'static str) -> Response { @@ -724,6 +738,7 @@ async fn ready(State(s): State) -> Response { } Json(serde_json::json!({"status":"ready"})).into_response() } +/// Build the public and private routers without a metrics endpoint. pub fn router(state: AppState) -> (Router, Router) { router_with_metrics(state, None) } diff --git a/crates/buzz-push-gateway/src/lib.rs b/crates/buzz-push-gateway/src/lib.rs index 563d725db9..e6f5dc3a28 100644 --- a/crates/buzz-push-gateway/src/lib.rs +++ b/crates/buzz-push-gateway/src/lib.rs @@ -1,13 +1,23 @@ +#![warn(missing_docs)] //! Stateful, capability-gated APNs last hop for NIP-PL. pub mod apns; +/// App Attest enrollment verification. pub mod app_attest; +/// Durable installation authority and relay delegation state. pub mod authority; +/// Environment-driven configuration. pub mod config; +/// Authenticated, expiring endpoint grants. pub mod grant; +/// Stateful installation, delegation, delivery, and health HTTP API. pub mod http; +/// Metrics counters and labels. pub mod metrics; +/// Closed wire types for the stateful gateway. pub mod model; +/// PostgreSQL authority store. pub mod postgres; pub(crate) mod strict_json; +/// APNs token custody. pub mod token; pub use http::{router, router_with_metrics, AppState}; diff --git a/crates/buzz-push-gateway/src/model.rs b/crates/buzz-push-gateway/src/model.rs index 23f8015fe0..5886815548 100644 --- a/crates/buzz-push-gateway/src/model.rs +++ b/crates/buzz-push-gateway/src/model.rs @@ -2,20 +2,29 @@ use serde::{Deserialize, Serialize}; +/// Maximum accepted size of a delivery request body in bytes. pub const MAX_REQUEST_BYTES: usize = 8 * 1024; +/// Maximum encoded length of an endpoint grant envelope in bytes. pub const MAX_GRANT_BYTES: usize = 4096; +/// Maximum length of a hex-encoded APNs endpoint token in bytes. pub const MAX_ENDPOINT_HEX_BYTES: usize = 512; +/// Compiled-in APNs reconnect payload sent for every delivery. pub const APNS_RECONNECT_PAYLOAD: &[u8] = br#"{"aps":{"alert":{"body":"Reconnect to your relay now"},"mutable-content":1}}"#; +/// Wire protocol version for gateway request/response bodies. pub const WIRE_VERSION: u8 = 1; +/// App profile identifying an iOS app/release track (production vs sandbox). #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "kebab-case")] pub enum AppProfile { + /// Buzz iOS production track. BuzzIosProduction, + /// Buzz iOS sandbox track. BuzzIosSandbox, } impl AppProfile { + /// Lowercase kebab-case string form of the profile. pub const fn as_str(self) -> &'static str { match self { Self::BuzzIosProduction => "buzz-ios-production", @@ -31,9 +40,13 @@ impl AppProfile { #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct DeliveryRequest { + /// Wire version. pub v: u8, + /// Opaque authenticated endpoint-grant ciphertext. pub endpoint_grant: String, + /// Unique id of the delivery request. pub request_id: uuid::Uuid, + /// Unix timestamp at which the request expires. pub expires_at: i64, } @@ -43,25 +56,38 @@ pub struct DeliveryRequest { #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct EndpointGrant { + /// Wire version. pub v: u8, + /// The durable delegation id backing the grant. pub delegation_id: uuid::Uuid, + /// Nostr relay public key (hex) the grant is bound to. pub relay_pubkey: String, + /// App profile of the installation. pub app_profile: AppProfile, + /// Endpoint epoch the grant is bound to. pub endpoint_epoch: i64, + /// Delegation generation at mint time. pub generation: i64, + /// Unix timestamp at which the grant expires. pub expires_at: i64, } +/// Request body for an installation enrollment challenge. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct InstallationChallengeRequest { + /// Wire version. pub v: u8, } +/// Response body returning a freshly issued enrollment challenge. #[derive(Debug, Clone, Serialize)] pub struct InstallationChallengeResponse { + /// Unique id of the issued challenge. pub challenge_id: uuid::Uuid, + /// Base64-encoded challenge value. pub challenge: String, + /// Unix timestamp at which the challenge expires. pub expires_at: i64, } @@ -71,100 +97,162 @@ pub struct InstallationChallengeResponse { #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct InstallationEnrollRequest { + /// Wire version. pub v: u8, + /// Unique id of the challenge being answered. pub challenge_id: uuid::Uuid, + /// Base64-encoded challenge value. pub challenge: String, + /// Base64-encoded App Attest key id. pub key_id: String, + /// Base64-encoded Apple attestation CBOR. pub attestation: String, + /// App profile to enroll. pub app_profile: AppProfile, + /// Hex-encoded APNs endpoint token. pub endpoint: String, + /// Endpoint epoch for the submitted token. pub endpoint_epoch: i64, + /// Unix timestamp at which the installation should expire. pub expires_at: i64, } +/// Response body returning the created installation handle. #[derive(Debug, Clone, Serialize)] pub struct InstallationEnrollResponse { + /// Opaque handle identifying the installation. pub installation_handle: uuid::Uuid, + /// Endpoint epoch recorded for the installation. pub endpoint_epoch: i64, + /// Unix timestamp at which the installation expires. pub expires_at: i64, } +/// Request body delegating delivery authority to a relay. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct DelegationRequest { + /// Wire version. pub v: u8, + /// Unique id of the challenge being answered. pub challenge_id: uuid::Uuid, + /// Base64-encoded challenge value. pub challenge: String, + /// Handle of the installation delegating. pub installation_handle: uuid::Uuid, + /// Endpoint epoch the delegation is bound to. pub endpoint_epoch: i64, + /// Requested delegation generation. pub generation: i64, + /// Nostr relay public key (hex) receiving the delegation. pub relay_pubkey: String, + /// Unix timestamp from which the delegation is valid. pub not_before: i64, + /// Unix timestamp at which the delegation expires. pub expires_at: i64, + /// Base64-encoded App Attest assertion over the request transcript. pub assertion: String, } +/// Response body returning the minted endpoint grant. #[derive(Debug, Clone, Serialize)] pub struct DelegationResponse { + /// Opaque authenticated endpoint-grant ciphertext. pub endpoint_grant: String, } +/// Request body rotating an installation's APNs endpoint token. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct RotateEndpointRequest { + /// Wire version. pub v: u8, + /// Unique id of the challenge being answered. pub challenge_id: uuid::Uuid, + /// Base64-encoded challenge value. pub challenge: String, + /// Handle of the installation rotating its endpoint. pub installation_handle: uuid::Uuid, + /// Expected current endpoint epoch (compare-and-swap guard). pub endpoint_epoch: i64, + /// New endpoint epoch after rotation. pub new_endpoint_epoch: i64, + /// Hex-encoded new APNs endpoint token. pub endpoint: String, + /// Base64-encoded App Attest assertion over the request transcript. pub assertion: String, } +/// Request body revoking a relay's delegation. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct RevokeDelegationRequest { + /// Wire version. pub v: u8, + /// Unique id of the challenge being answered. pub challenge_id: uuid::Uuid, + /// Base64-encoded challenge value. pub challenge: String, + /// Handle of the installation whose delegation is revoked. pub installation_handle: uuid::Uuid, + /// Nostr relay public key (hex) losing the delegation. pub relay_pubkey: String, + /// Delegation generation to revoke. pub generation: i64, + /// Base64-encoded App Attest assertion over the request transcript. pub assertion: String, } +/// Request body revoking an installation. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct RevokeInstallationRequest { + /// Wire version. pub v: u8, + /// Unique id of the challenge being answered. pub challenge_id: uuid::Uuid, + /// Base64-encoded challenge value. pub challenge: String, + /// Handle of the installation to revoke. pub installation_handle: uuid::Uuid, + /// Expected current endpoint epoch (compare-and-swap guard). pub endpoint_epoch: i64, + /// New endpoint epoch after revocation. pub new_endpoint_epoch: i64, + /// Base64-encoded App Attest assertion over the request transcript. pub assertion: String, } +/// Generic mutation success response. #[derive(Debug, Clone, Serialize)] pub struct MutationResponse { + /// Human-readable status string. pub status: &'static str, } +/// Outcome of a delivery attempt, serialized as the response body. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "snake_case", tag = "status", deny_unknown_fields)] pub enum DeliveryResponse { + /// APNs accepted the delivery. Accepted, + /// The endpoint is permanently invalid. InvalidEndpoint { + /// Delegation generation at invalidation. generation: i64, + /// When the endpoint became invalid, if known. invalid_at: Option, }, + /// The delivery may be retried after an optional delay. Retry { + /// Suggested delay before retry, in seconds. retry_after_seconds: Option, }, } +/// Error response body. #[derive(Debug, Clone, Serialize)] pub struct ErrorBody { + /// Stable machine-readable error code. pub error: &'static str, } diff --git a/crates/buzz-push-gateway/src/postgres.rs b/crates/buzz-push-gateway/src/postgres.rs index bd69ec2564..87c3ec4a47 100644 --- a/crates/buzz-push-gateway/src/postgres.rs +++ b/crates/buzz-push-gateway/src/postgres.rs @@ -7,15 +7,18 @@ use chrono::{DateTime, Utc}; use sqlx::{AssertSqlSafe, PgPool, Row}; use uuid::Uuid; +/// Postgres-backed implementation of [`AuthorityStore`]. #[derive(Clone)] pub struct PostgresAuthorityStore { pool: PgPool, } impl PostgresAuthorityStore { + /// Wrap a connection pool as an authority store. pub fn new(pool: PgPool) -> Self { Self { pool } } + /// Run SQLx migrations and grant runtime role permissions on the pool. pub async fn apply_migrations_and_grants( pool: &PgPool, runtime_role: &str, diff --git a/crates/buzz-push-gateway/src/token.rs b/crates/buzz-push-gateway/src/token.rs index 74fe4ce131..b791f82c16 100644 --- a/crates/buzz-push-gateway/src/token.rs +++ b/crates/buzz-push-gateway/src/token.rs @@ -12,26 +12,32 @@ const AAD_PREFIX: &[u8] = b"buzz-apns-token-v1:"; const MAX_KEY_ID_BYTES: usize = 32; const MAX_CIPHERTEXT_BYTES: usize = 2048; +/// A single named AES-256-GCM key used to seal/open APNs tokens. #[derive(Clone)] pub struct TokenKey { id: String, cipher: Aes256Gcm, } +/// Ordered keyring: one current seal/open key plus decrypt-only predecessors. #[derive(Clone)] pub struct TokenKeyring { current: TokenKey, predecessors: HashMap, } +/// Errors raised while sealing or opening APNs token ciphertext. #[derive(Debug, Error)] pub enum TokenError { + /// The ciphertext, token, or key material was malformed. #[error("invalid token ciphertext")] Invalid, + /// The keyring was empty or contained duplicate ids. #[error("token keyring is empty or contains duplicate ids")] InvalidKeyring, } impl TokenKey { + /// Construct a token key from its id and 32-byte raw material. pub fn new(id: impl Into, key: &[u8]) -> Result { let id = id.into(); if id.is_empty() @@ -53,6 +59,7 @@ impl TokenKey { } impl TokenKeyring { + /// Build a keyring ordered current key first, then decrypt-only predecessors. pub fn new(keys: Vec) -> Result { let mut keys = keys.into_iter(); let current = keys.next().ok_or(TokenError::InvalidKeyring)?; @@ -66,6 +73,7 @@ impl TokenKeyring { predecessors: rest.into_iter().map(|k| (k.id.clone(), k)).collect(), }) } + /// Seal a raw APNs token under the current key, producing a self-describing ciphertext. pub fn seal(&self, token: &[u8]) -> Result, TokenError> { if token.is_empty() || token.len() > crate::model::MAX_ENDPOINT_HEX_BYTES { return Err(TokenError::Invalid); @@ -92,6 +100,7 @@ impl TokenKeyring { } Ok(encoded) } + /// Open a sealed token, selecting the key by its authenticated id prefix. pub fn open(&self, encoded: &[u8]) -> Result, TokenError> { if encoded.len() > MAX_CIPHERTEXT_BYTES { return Err(TokenError::Invalid); diff --git a/crates/buzz-relay-mesh/src/endpoint.rs b/crates/buzz-relay-mesh/src/endpoint.rs index f7bedd8789..6af7e012b4 100644 --- a/crates/buzz-relay-mesh/src/endpoint.rs +++ b/crates/buzz-relay-mesh/src/endpoint.rs @@ -44,14 +44,17 @@ impl MeshEndpoint { }) } + /// This endpoint's mesh identity (its public key). pub fn runtime_id(&self) -> RuntimeId { self.runtime_id } + /// Clone of the underlying iroh endpoint handle. pub fn endpoint(&self) -> Endpoint { self.endpoint.clone() } + /// The endpoint's dialable address (public key + transport addrs). pub fn addr(&self) -> EndpointAddr { self.endpoint.addr() } @@ -70,6 +73,7 @@ impl MeshEndpoint { .collect() } + /// Accept one inbound mesh connection, or `Ok(None)` if the endpoint is closed. pub async fn accept(&self) -> Result, MeshError> { let Some(incoming) = self.endpoint.accept().await else { return Ok(None); @@ -80,6 +84,7 @@ impl MeshEndpoint { crate::peer::MeshPeer::from_connection(self.endpoint.clone(), conn).map(Some) } + /// Dial a peer by its [`EndpointAddr`] and complete the mesh handshake. pub async fn connect( &self, peer_addr: EndpointAddr, @@ -93,14 +98,17 @@ impl MeshEndpoint { } } +/// Derive a [`RuntimeId`] from an iroh public key. pub fn runtime_id_from_public_key(public_key: PublicKey) -> RuntimeId { RuntimeId(*public_key.as_bytes()) } +/// Recover the iroh public key from a [`RuntimeId`]. pub fn public_key_from_runtime_id(runtime_id: RuntimeId) -> Result { PublicKey::from_bytes(&runtime_id.0).map_err(|err| MeshError::Transport(err.to_string())) } +/// Build a single-IP [`EndpointAddr`] for a peer (no relay/DNS paths). pub fn direct_addr(runtime_id: RuntimeId, addr: SocketAddr) -> Result { Ok(EndpointAddr::from_parts( public_key_from_runtime_id(runtime_id)?, diff --git a/crates/buzz-relay-mesh/src/gossip.rs b/crates/buzz-relay-mesh/src/gossip.rs index 5da5cbbe54..0d5329ebf5 100644 --- a/crates/buzz-relay-mesh/src/gossip.rs +++ b/crates/buzz-relay-mesh/src/gossip.rs @@ -10,23 +10,33 @@ use serde::{Deserialize, Serialize}; use crate::{MeshError, RuntimeId}; +/// Wire version of the gossip payload exchanged over the control stream. pub const GOSSIP_PAYLOAD_VERSION: u8 = 1; +/// A single runtime's gossiped state: identity, dialability, load, and version. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct GossipRecord { + /// The runtime this record describes. pub runtime_id: RuntimeId, + /// Dialable endpoint address strings for the runtime. pub endpoint_addrs: Vec, + /// Protocol version the runtime speaks. pub proto_version: u16, + /// Advisory load factor (0.0..) gossiped by the runtime. pub load: f32, + /// Whether the runtime is draining. pub draining: bool, + /// Capability strings advertised by the runtime. pub capabilities: Vec, /// Per-runtime monotonic version. Only the owning runtime may increment its /// own record; receivers apply last-version-wins. pub version: u64, + /// Wall-clock heartbeat timestamp (ms since UNIX_EPOCH). pub heartbeat_millis: u64, } impl GossipRecord { + /// Create a fresh local record with default load (0.0) and version 1. pub fn new(runtime_id: RuntimeId, endpoint_addrs: Vec, proto_version: u16) -> Self { Self { runtime_id, @@ -41,28 +51,40 @@ impl GossipRecord { } } +/// One entry in a gossip digest: a runtime id paired with its latest version. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct GossipDigestEntry { + /// The runtime the version refers to. pub runtime_id: RuntimeId, + /// The version known to the digest sender. pub version: u64, } +/// A gossip control message: either a digest request or a delta reply. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub enum GossipMessage { + /// Compact digest of the sender's known versions. Digest { + /// Gossip payload version. version: u8, + /// One entry per known runtime. entries: Vec, }, + /// Delta of records newer than the receiver's digest. Delta { + /// Gossip payload version. version: u8, + /// Records the receiver does not yet have at this version. records: Vec, }, } +/// Serialize a gossip message with postcard. pub fn encode_message(message: &GossipMessage) -> Result, MeshError> { postcard::to_extend(message, Vec::new()).map_err(MeshError::Encode) } +/// Deserialize a gossip message and validate its payload version. pub fn decode_message(bytes: &[u8]) -> Result { let message: GossipMessage = postcard::from_bytes(bytes).map_err(MeshError::Decode)?; let version = match &message { @@ -83,20 +105,26 @@ pub struct GossipState { } impl GossipState { + /// Create a state seeded with the local runtime's own record. pub fn new(local: GossipRecord) -> Self { let mut records = HashMap::new(); records.insert(local.runtime_id, local); Self { records } } + /// Iterate over all known records (including the local one). pub fn records(&self) -> impl Iterator { self.records.values() } + /// Look up the record for a runtime, if known. pub fn get(&self, runtime_id: RuntimeId) -> Option<&GossipRecord> { self.records.get(&runtime_id) } + /// Apply `update` to the local copy of a runtime's record, bumping its + /// version and heartbeat. Returns the updated record, or `None` if the + /// runtime is unknown. pub fn update_local(&mut self, runtime_id: RuntimeId, update: F) -> Option where F: FnOnce(&mut GossipRecord), @@ -108,6 +136,7 @@ impl GossipState { Some(record.clone()) } + /// Build a digest message covering every known runtime, sorted by id. pub fn digest(&self) -> GossipMessage { let mut entries: Vec<_> = self .records @@ -124,6 +153,7 @@ impl GossipState { } } + /// Build the delta of records newer than the peer's `digest`. pub fn delta_for(&self, digest: &[GossipDigestEntry]) -> GossipMessage { let remote_versions: HashMap<_, _> = digest .iter() @@ -164,6 +194,8 @@ impl GossipState { } } +/// Phi-accrual failure detector: estimates a suspicion score from heartbeat +/// inter-arrival intervals. Higher `phi` means the peer is more likely down. #[derive(Clone, Debug)] pub struct PhiAccrual { samples: Vec, @@ -178,6 +210,7 @@ impl Default for PhiAccrual { } impl PhiAccrual { + /// Create a detector that retains at most `max_samples` intervals. pub fn new(max_samples: usize) -> Self { Self { samples: Vec::new(), @@ -186,6 +219,7 @@ impl PhiAccrual { } } + /// Record a heartbeat observation at `at`, updating the interval window. pub fn observe(&mut self, at: SystemTime) { if let Some(prev) = self.last_heartbeat { if let Ok(interval) = at.duration_since(prev) { @@ -200,6 +234,7 @@ impl PhiAccrual { self.last_heartbeat = Some(at); } + /// Current suspicion score, or `None` until at least one interval is known. pub fn phi_at(&self, now: SystemTime) -> Option { let last = self.last_heartbeat?; if self.samples.is_empty() { @@ -214,12 +249,14 @@ impl PhiAccrual { Some((elapsed / mean) / std::f64::consts::LN_10) } + /// Mean heartbeat inter-arrival interval in seconds. pub fn mean_secs(&self) -> f64 { let total: f64 = self.samples.iter().map(Duration::as_secs_f64).sum(); total / self.samples.len() as f64 } } +/// Current wall clock as milliseconds since UNIX_EPOCH. pub fn now_millis() -> u64 { SystemTime::now() .duration_since(UNIX_EPOCH) @@ -228,6 +265,7 @@ pub fn now_millis() -> u64 { .min(u128::from(u64::MAX)) as u64 } +/// Convert a `now_millis`-style timestamp back into a [`SystemTime`]. pub fn system_time_from_millis(millis: u64) -> SystemTime { UNIX_EPOCH + Duration::from_millis(millis) } diff --git a/crates/buzz-relay-mesh/src/lib.rs b/crates/buzz-relay-mesh/src/lib.rs index be031cb371..d2fb1af292 100644 --- a/crates/buzz-relay-mesh/src/lib.rs +++ b/crates/buzz-relay-mesh/src/lib.rs @@ -1,3 +1,4 @@ +#![warn(missing_docs)] //! buzz-relay-mesh — the inter-relay QUIC mesh. //! //! One iroh endpoint per relay runtime (identity = a boot-unique mesh @@ -18,13 +19,21 @@ //! **The law:** mesh membership is a hint; the Redis fenced generation is the //! arbiter. Nothing in this crate grants ownership — see [`wire::FencedHeader`]. +/// QUIC endpoint wrapper and connection lifecycle. pub mod endpoint; +/// Scuttlebutt membership gossip state machine and digest exchange. pub mod gossip; +/// Phi-accrual membership tracker and live peer view. pub mod membership; +/// Per-peer connection state and framed transport halves. pub mod peer; +/// Ready registry: Redis-backed attestation/heartbeat directory of runtimes. pub mod registry; +/// [`MeshRuntime`] — the trait iroh-backed runtimes implement. pub mod runtime; +/// `/_mesh` status snapshot types surfaced to operators. pub mod status; +/// The fenced wire contract: framing, version, and headers. pub mod wire; // Lane modules — one owner per file (see the mesh thread for lane map): @@ -62,28 +71,51 @@ pub struct MeshConfig { pub registry_refresh: std::time::Duration, } +/// Errors raised by the mesh transport and wire codec. #[derive(Debug, thiserror::Error)] pub enum MeshError { + /// Failed to encode a wire frame. #[error("frame encode: {0}")] Encode(#[source] postcard::Error), + /// Failed to decode a wire frame. #[error("frame decode: {0}")] Decode(#[source] postcard::Error), + /// Encountered an unrecognized wire protocol version. #[error("unknown wire version {0}")] UnknownWireVersion(u8), + /// Received an empty (zero-length) frame. #[error("empty frame")] EmptyFrame, + /// A frame exceeded the configured maximum size. #[error("frame exceeds max size ({size} > {max})")] - FrameTooLarge { size: usize, max: usize }, + FrameTooLarge { + /// Actual frame size in bytes. + size: usize, + /// Configured maximum frame size in bytes. + max: usize, + }, + /// A datagram exceeded the connection's `max_datagram_size`. #[error("datagram exceeds connection max_datagram_size ({size} > {max})")] - DatagramTooLarge { size: usize, max: usize }, + DatagramTooLarge { + /// Actual datagram size in bytes. + size: usize, + /// Connection maximum datagram size in bytes. + max: usize, + }, + /// The target peer has no live mesh connection. #[error("peer {0} not connected")] PeerNotConnected(RuntimeId), + /// The target peer is draining and refuses new traffic. #[error("peer {0} is draining")] PeerDraining(RuntimeId), + /// The frame's generation is older than the known generation for the session. #[error("stale generation for session {session_id}: frame {frame_generation} < known {known_generation}")] StaleGeneration { + /// The session the stale frame targeted. session_id: uuid::Uuid, + /// Generation claimed by the rejected frame. frame_generation: u64, + /// Highest generation known for the session. known_generation: u64, }, // The three variants below complete the fence-rejection taxonomy alongside @@ -94,32 +126,48 @@ pub enum MeshError { // `stale_generation` | `no_active_lease` | `owner_mismatch` | // `future_generation`. None of these are serialized — the wire-level fence // signal remains `GoodbyeReason::StaleGeneration`. + /// A frame arrived for a session with no live lease on this runtime. #[error("no active lease for session {session_id}: frame generation {frame_generation}, known generation {known_generation}, claimed owner {frame_owner_runtime_id}")] NoActiveLease { + /// The session the frame targeted. session_id: uuid::Uuid, + /// Generation claimed by the frame. frame_generation: u64, + /// Highest generation known for the session. known_generation: u64, /// The owner the *frame* claimed — there is no current owner by /// definition when no live lease exists. frame_owner_runtime_id: RuntimeId, }, + /// The frame's claimed owner does not match the session's current owner. #[error("owner mismatch for session {session_id} generation {generation}: frame owner {frame_owner_runtime_id} != current owner {current_owner_runtime_id}")] OwnerMismatch { + /// The session whose owner was contested. session_id: uuid::Uuid, + /// Generation at which the mismatch was detected. generation: u64, + /// Owner claimed by the frame. frame_owner_runtime_id: RuntimeId, + /// The runtime that actually owns the session. current_owner_runtime_id: RuntimeId, }, + /// The frame carried a generation ahead of any known lease (replay/forge signal). #[error("future generation for session {session_id}: frame {frame_generation} > known {known_generation}")] FutureGeneration { + /// The session the frame targeted. session_id: uuid::Uuid, + /// Generation claimed by the frame. frame_generation: u64, + /// Highest generation known for the session. known_generation: u64, }, + /// The mesh is disabled via `BUZZ_MESH=off`. #[error("mesh is disabled (BUZZ_MESH=off)")] Disabled, + /// Lower-level transport failure not covered by a more specific variant. #[error("transport: {0}")] Transport(String), + /// A Redis operation failed. #[error("redis: {0}")] Redis(#[from] redis::RedisError), } @@ -127,7 +175,9 @@ pub enum MeshError { /// A peer as membership sees it. Everything here is a routing HINT. #[derive(Clone, Debug)] pub struct PeerInfo { + /// The peer's mesh identity. pub runtime_id: RuntimeId, + /// Whether the peer has begun draining and should shed load. pub draining: bool, /// Phi-accrual suspicion; `None` until enough heartbeats observed. pub phi: Option, @@ -175,7 +225,9 @@ pub trait RelayPeerTransport: Send + Sync + 'static { /// Inbound mesh traffic, delivered after wire decode + Hello validation. pub trait InboundHandler: Send + Sync + 'static { + /// Called for each inbound realtime datagram after wire decode. fn on_datagram(&self, from: RuntimeId, dgram: MeshDatagram); + /// Called for each inbound session stream after `Hello` validation. fn on_session_stream(&self, from: RuntimeId, hello: StreamHello, stream: MeshStream); } @@ -188,22 +240,30 @@ pub struct MeshStream { pub(crate) recv: Box, } +/// Write half of a framed mesh stream. pub trait StreamSendHalf: Send + 'static { + /// Send a single length-delimited frame. fn send_frame(&mut self, frame: MeshStreamFrame) -> BoxFuture<'_, Result<(), MeshError>>; + /// Finish the send half, signalling clean EOF to the peer. fn finish(&mut self) -> Result<(), MeshError>; } +/// Read half of a framed mesh stream. pub trait StreamRecvHalf: Send + 'static { + /// Receive the next frame, or `Ok(None)` at clean EOF. fn recv_frame(&mut self) -> BoxFuture<'_, Result, MeshError>>; } impl MeshStream { + /// Send a single length-delimited frame on the underlying stream. pub fn send_frame(&mut self, frame: MeshStreamFrame) -> BoxFuture<'_, Result<(), MeshError>> { self.send.send_frame(frame) } + /// Receive the next frame, or `Ok(None)` at clean EOF. pub fn recv_frame(&mut self) -> BoxFuture<'_, Result, MeshError>> { self.recv.recv_frame() } + /// Finish the send half, signalling clean EOF to the peer. pub fn finish(&mut self) -> Result<(), MeshError> { self.send.finish() } diff --git a/crates/buzz-relay-mesh/src/membership.rs b/crates/buzz-relay-mesh/src/membership.rs index 9efffb1d80..d7435bf8eb 100644 --- a/crates/buzz-relay-mesh/src/membership.rs +++ b/crates/buzz-relay-mesh/src/membership.rs @@ -14,6 +14,7 @@ use crate::registry::ReadyRecord; use crate::status::{ConnectionState, MeshCounters, MeshPeerCounters, MeshPeerStatus, MeshStatus}; use crate::{PeerInfo, RelayMeshMembership, RuntimeId}; +/// Default phi value at which a peer is considered suspect (excluded from routing). pub const DEFAULT_PHI_SUSPECT_THRESHOLD: f64 = 8.0; #[derive(Clone, Debug)] @@ -43,6 +44,7 @@ pub struct MeshMembership { } impl MeshMembership { + /// Create a membership view seeded with the local runtime's record. pub fn new(local_record: GossipRecord) -> Self { Self { local_runtime_id: local_record.runtime_id, @@ -63,11 +65,13 @@ impl MeshMembership { self } + /// Set the phi value above which a peer is treated as suspect. pub fn with_phi_suspect_threshold(mut self, threshold: f64) -> Self { self.phi_suspect_threshold = threshold; self } + /// Clone of this runtime's own gossip record. pub fn local_record(&self) -> GossipRecord { self.local_record .read() @@ -152,6 +156,7 @@ impl MeshMembership { } } + /// Override a peer's connection state (no-op if the peer is unknown). pub fn mark_connection_state(&self, runtime_id: RuntimeId, state: ConnectionState) { if let Some(peer) = self .peers @@ -163,6 +168,7 @@ impl MeshMembership { } } + /// Mutate the local record under `update`, bumping its version/heartbeat. pub fn update_local(&self, update: F) -> GossipRecord where F: FnOnce(&mut GossipRecord), @@ -177,6 +183,7 @@ impl MeshMembership { local.clone() } + /// Whether this runtime is draining. pub fn is_draining(&self) -> bool { self.draining.load(Ordering::Relaxed) } @@ -246,42 +253,49 @@ impl MeshMembership { } } + /// Increment a peer's `streams_opened` counter. pub fn record_stream_opened(&self, runtime_id: RuntimeId) { self.update_peer_counters(runtime_id, |c| { c.streams_opened = c.streams_opened.saturating_add(1) }); } + /// Increment a peer's `streams_received` counter. pub fn record_stream_received(&self, runtime_id: RuntimeId) { self.update_peer_counters(runtime_id, |c| { c.streams_received = c.streams_received.saturating_add(1) }); } + /// Increment a peer's `datagrams_sent` counter. pub fn record_datagram_sent(&self, runtime_id: RuntimeId) { self.update_peer_counters(runtime_id, |c| { c.datagrams_sent = c.datagrams_sent.saturating_add(1) }); } + /// Increment a peer's `datagrams_received` counter. pub fn record_datagram_received(&self, runtime_id: RuntimeId) { self.update_peer_counters(runtime_id, |c| { c.datagrams_received = c.datagrams_received.saturating_add(1) }); } + /// Increment a peer's `gossip_frames_sent` counter. pub fn record_gossip_frame_sent(&self, runtime_id: RuntimeId) { self.update_peer_counters(runtime_id, |c| { c.gossip_frames_sent = c.gossip_frames_sent.saturating_add(1) }); } + /// Increment a peer's `gossip_frames_received` counter. pub fn record_gossip_frame_received(&self, runtime_id: RuntimeId) { self.update_peer_counters(runtime_id, |c| { c.gossip_frames_received = c.gossip_frames_received.saturating_add(1) }); } + /// Record a stale-generation fence rejection, optionally against a peer. pub fn record_stale_generation_rejection(&self, runtime_id: Option) { self.stale_generation_rejections .fetch_add(1, Ordering::Relaxed); @@ -292,6 +306,7 @@ impl MeshMembership { } } + /// Snapshot the full `/_mesh` status for operators. pub fn status(&self) -> MeshStatus { let now = SystemTime::now(); let local = self.local_record(); diff --git a/crates/buzz-relay-mesh/src/peer.rs b/crates/buzz-relay-mesh/src/peer.rs index 59ee8308fc..b2d08302d6 100644 --- a/crates/buzz-relay-mesh/src/peer.rs +++ b/crates/buzz-relay-mesh/src/peer.rs @@ -6,11 +6,16 @@ use crate::{ StreamRecvHalf, StreamSendHalf, ALPN, }; +/// Snapshot of counters for a single peer connection. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct PeerCounters { + /// Reliable streams this side opened to the peer. pub streams_opened: u64, + /// Reliable streams accepted from the peer. pub streams_accepted: u64, + /// Datagrams sent to the peer. pub datagrams_sent: u64, + /// Datagrams received from the peer. pub datagrams_received: u64, } @@ -62,18 +67,22 @@ impl MeshPeer { }) } + /// The peer's mesh identity. pub fn runtime_id(&self) -> RuntimeId { self.runtime_id } + /// Max datagram size the connection accepts, if datagrams are enabled. pub fn max_datagram_size(&self) -> Option { self.conn.max_datagram_size() } + /// Snapshot this peer's per-connection counters. pub fn counters(&self) -> PeerCounters { self.counters.snapshot() } + /// Open a bidirectional framed stream to the peer. pub async fn open_bi(&self) -> Result { let (send, recv) = self .conn @@ -87,6 +96,7 @@ impl MeshPeer { )) } + /// Accept the next bidirectional framed stream from the peer. pub async fn accept_bi(&self) -> Result { let (send, recv) = self .conn @@ -102,6 +112,7 @@ impl MeshPeer { )) } + /// Send a realtime datagram, enforcing the connection's size limit. pub fn send_datagram(&self, dgram: &MeshDatagram) -> Result<(), MeshError> { let max = self .conn @@ -115,6 +126,7 @@ impl MeshPeer { Ok(()) } + /// Receive the next realtime datagram from the peer. pub async fn recv_datagram(&self) -> Result { let bytes = self .conn diff --git a/crates/buzz-relay-mesh/src/registry.rs b/crates/buzz-relay-mesh/src/registry.rs index 635dff6328..33147d9618 100644 --- a/crates/buzz-relay-mesh/src/registry.rs +++ b/crates/buzz-relay-mesh/src/registry.rs @@ -16,9 +16,13 @@ use sha2::{Digest, Sha256}; use crate::{MeshError, RuntimeId}; +/// Redis key prefix for ready records. pub const READY_KEY_PREFIX: &str = "mesh:ready:"; +/// Default refresh interval for the ready-registry heartbeat. pub const DEFAULT_REGISTRY_REFRESH: Duration = Duration::from_secs(15); +/// TTL multiplier applied to the refresh interval to get record expiry. pub const REGISTRY_EXPIRY_MULTIPLIER: u64 = 3; +/// Versioned domain-separation string for ready-record attestations. pub const ATTESTATION_CONTEXT: &str = "buzz-relay-mesh-ready-v1"; /// Relay-key-signed binding for a boot-unique runtime endpoint pubkey. @@ -35,6 +39,7 @@ pub struct RuntimeAttestation { } impl RuntimeAttestation { + /// Create and sign an attestation binding `runtime_id` to the relay key. pub fn new(relay_keys: &nostr::Keys, runtime_id: RuntimeId) -> Self { let relay_pubkey = relay_keys.public_key().to_hex(); let message = attestation_message(runtime_id, &relay_pubkey); @@ -45,6 +50,7 @@ impl RuntimeAttestation { } } + /// Verify the attestation against the given `runtime_id`. pub fn verify(&self, runtime_id: RuntimeId) -> Result<(), MeshError> { verify_attestation(runtime_id, &self.relay_pubkey, &self.relay_sig) } @@ -97,6 +103,7 @@ fn attestation_message(runtime_id: RuntimeId, relay_pubkey: &str) -> Message { /// Value stored at `mesh:ready:{runtime_id}`. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] pub struct ReadyRecord { + /// The boot-unique mesh identity being attested. pub runtime_id: RuntimeId, /// Explicit duplicate of `runtime_id` for the contract record shape: this /// is the boot-unique ed25519/iroh endpoint pubkey being attested. @@ -108,11 +115,14 @@ pub struct ReadyRecord { /// Dialable iroh endpoint addresses, serialized as strings so this layer /// does not depend on transport internals. pub endpoint_addrs: Vec, + /// Protocol version the runtime speaks. pub proto_version: u16, + /// Capability strings advertised by the runtime. pub capabilities: Vec, } impl ReadyRecord { + /// Create a signed record for `runtime_id` using the relay key. pub fn new( runtime_id: RuntimeId, relay_keys: &nostr::Keys, @@ -132,10 +142,12 @@ impl ReadyRecord { } } + /// The Redis key this record is stored under. pub fn key(&self) -> String { ready_key(self.runtime_id) } + /// Verify `runtime_pubkey`/`relay_sig` against the record's `runtime_id`. pub fn verify_attestation(&self) -> Result<(), MeshError> { if self.runtime_pubkey != self.runtime_id.to_hex() { return Err(MeshError::Transport(format!( @@ -147,10 +159,12 @@ impl ReadyRecord { } } +/// Build the Redis key for a runtime's ready record. pub fn ready_key(runtime_id: RuntimeId) -> String { format!("{READY_KEY_PREFIX}{runtime_id}") } +/// TTL to set for a record given the configured refresh interval. pub fn expiry_for(refresh: Duration) -> Duration { refresh.saturating_mul(REGISTRY_EXPIRY_MULTIPLIER as u32) } @@ -163,14 +177,17 @@ pub struct ReadyRegistry { } impl ReadyRegistry { + /// Create a registry over `pool` with the given heartbeat `refresh`. pub fn new(pool: deadpool_redis::Pool, refresh: Duration) -> Self { Self { pool, refresh } } + /// The configured heartbeat refresh interval. pub fn refresh_interval(&self) -> Duration { self.refresh } + /// The TTL applied to published records. pub fn expiry(&self) -> Duration { expiry_for(self.refresh) } @@ -257,6 +274,7 @@ impl ReadyRegistry { Ok(out) } + /// Build a readiness-gated heartbeat driver for `record`. pub fn heartbeat(&self, record: ReadyRecord) -> ReadyHeartbeat { ReadyHeartbeat { registry: self.clone(), @@ -284,14 +302,17 @@ pub struct ReadyHeartbeat { } impl ReadyHeartbeat { + /// The record being heartbeated. pub fn record(&self) -> &ReadyRecord { &self.record } + /// Whether the record is currently published. pub fn published(&self) -> bool { self.published } + /// Advance one tick: publish while `ready`, clear on the ready→not-ready edge. pub async fn tick(&mut self, ready: bool) -> Result<(), MeshError> { if ready { self.registry.publish_ready(&self.record).await?; @@ -303,6 +324,7 @@ impl ReadyHeartbeat { Ok(()) } + /// Clear the published record on clean shutdown (no-op if not published). pub async fn shutdown(&mut self) -> Result<(), MeshError> { if self.published { self.registry.clear_ready(self.record.runtime_id).await?; diff --git a/crates/buzz-relay-mesh/src/runtime.rs b/crates/buzz-relay-mesh/src/runtime.rs index 897de6ca8c..f12c58209e 100644 --- a/crates/buzz-relay-mesh/src/runtime.rs +++ b/crates/buzz-relay-mesh/src/runtime.rs @@ -99,6 +99,7 @@ impl MeshRuntime { ) } + /// Like [`Self::start`] but with explicit gossip/reconcile intervals (tests). pub fn start_with_intervals( endpoint: MeshEndpoint, membership: MeshMembership, @@ -126,10 +127,12 @@ impl MeshRuntime { } } + /// The membership table backing this mesh. pub fn membership(&self) -> &MeshMembership { &self.inner.membership } + /// This runtime's mesh identity. pub fn local_runtime_id(&self) -> RuntimeId { self.inner.endpoint.runtime_id() } diff --git a/crates/buzz-relay-mesh/src/status.rs b/crates/buzz-relay-mesh/src/status.rs index 895cb7bea5..062e86ce59 100644 --- a/crates/buzz-relay-mesh/src/status.rs +++ b/crates/buzz-relay-mesh/src/status.rs @@ -4,57 +4,92 @@ use serde::Serialize; +/// Top-level `/_mesh` status payload, serialized directly as JSON. #[derive(Clone, Debug, Default, Serialize)] pub struct MeshStatus { + /// Whether the mesh is enabled for this runtime. pub enabled: bool, + /// This runtime's mesh identity (hex). pub local_runtime_id: String, + /// Whether this runtime is draining. pub draining: bool, + /// Number of known peers. pub peer_count: usize, + /// Per-peer status entries. pub peers: Vec, + /// Aggregate counters. pub counters: MeshCounters, } +/// Status of a single peer as surfaced to operators. #[derive(Clone, Debug, Serialize)] pub struct MeshPeerStatus { + /// The peer's mesh identity (hex). pub runtime_id: String, + /// Dialable endpoint address strings. pub endpoint_addrs: Vec, + /// Protocol version the peer speaks. pub proto_version: u16, + /// Whether the peer is draining. pub draining: bool, + /// Last observed connection state. pub connection_state: ConnectionState, + /// Phi-accrual suspicion score, if enough heartbeats observed. pub phi: Option, + /// Advisory load factor gossiped by the peer. pub load: f32, + /// Last gossiped record version. pub record_version: u64, + /// Last heartbeat timestamp (ms since UNIX_EPOCH). pub last_heartbeat_millis: u64, + /// Per-peer counters. pub counters: MeshPeerCounters, } +/// Observed transport state of a peer connection. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] pub enum ConnectionState { + /// No live connection to the peer. #[default] Disconnected, + /// A connection attempt is in flight. Connecting, + /// Authenticated connection is established. Connected, + /// Peer is considered suspect by the phi-accrual detector. Suspect, } +/// Aggregate counters across all peers. #[derive(Clone, Debug, Default, Serialize)] pub struct MeshCounters { + /// Frames rejected because their generation was stale. pub stale_generation_rejections: u64, /// Ready-registry seeds rejected because their `relay_pubkey` did not /// match this deployment's relay identity (or no anchor was configured). pub foreign_relay_rejections: u64, + /// Per-peer counters, in the same order as `MeshStatus::peers`. pub peers: Vec, } +/// Counters for a single peer. #[derive(Clone, Debug, Default, Serialize)] pub struct MeshPeerCounters { + /// The peer's mesh identity (hex). pub runtime_id: String, + /// Reliable streams opened to the peer. pub streams_opened: u64, + /// Reliable streams received from the peer. pub streams_received: u64, + /// Datagrams sent to the peer. pub datagrams_sent: u64, + /// Datagrams received from the peer. pub datagrams_received: u64, + /// Gossip control frames sent to the peer. pub gossip_frames_sent: u64, + /// Gossip control frames received from the peer. pub gossip_frames_received: u64, + /// Stale-generation fence rejections attributed to the peer. pub stale_generation_rejections: u64, } diff --git a/crates/buzz-relay-mesh/src/wire.rs b/crates/buzz-relay-mesh/src/wire.rs index 73c8b56925..8415948162 100644 --- a/crates/buzz-relay-mesh/src/wire.rs +++ b/crates/buzz-relay-mesh/src/wire.rs @@ -62,6 +62,7 @@ pub const MAX_STREAM_FRAME: u32 = 16 * 1024 * 1024; pub struct RuntimeId(pub [u8; 32]); impl RuntimeId { + /// Lowercase hex encoding of the runtime id. pub fn to_hex(&self) -> String { hex::encode(self.0) } @@ -83,6 +84,7 @@ impl std::fmt::Display for RuntimeId { /// every hop against the Redis lease. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct FencedHeader { + /// The session this frame belongs to. pub session_id: Uuid, /// Monotonic lease generation from the Redis CAS. A receiver that has /// observed generation G for a session rejects any frame with < G. @@ -109,6 +111,7 @@ pub enum Profile { /// One QUIC datagram: realtime media only. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct MeshDatagram { + /// The fence header for the session this datagram carries. pub fenced: FencedHeader, /// Sender-scoped monotonic sequence for loss/reorder observability. /// Receivers tolerate gaps and reordering; they never wait. @@ -128,19 +131,26 @@ pub enum MeshStreamFrame { Hello(StreamHello), /// Opaque tunnel bytes for a reliable-stream session. Data { + /// The fence header for the session this data carries. fenced: FencedHeader, + /// Opaque reliable-stream payload. payload: Vec, }, /// Clean close: the sender will send no more `Data` for this session. /// Distinct from a QUIC reset — receivers treat reset as abnormal. Goodbye { + /// The fence header for the session being closed. fenced: FencedHeader, + /// Why the stream is closing. reason: GoodbyeReason, }, /// Membership gossip on the control stream (one per peer connection). /// Payload is the gossip lane's postcard-encoded digest/delta exchange — /// opaque at this layer so gossip can evolve without a wire bump here. - Gossip { payload: Vec }, + Gossip { + /// Gossip-lane postcard payload (opaque here). + payload: Vec, + }, } /// Stream role, declared in the Hello. @@ -151,17 +161,23 @@ pub enum StreamRole { Control, /// A reliable-stream tunnel session. Session { + /// The fence header establishing the session's ownership. fenced: FencedHeader, + /// The tunnel profile fixed for this session. profile: Profile, }, } +/// The first frame on every mesh stream, declaring sender and role. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct StreamHello { + /// The runtime id of the sender. pub sender: RuntimeId, + /// The role this stream plays (control or session). pub role: StreamRole, } +/// Why a stream is being closed cleanly. #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum GoodbyeReason { /// Client closed / session ended normally. diff --git a/crates/git-credential-nostr/src/lib.rs b/crates/git-credential-nostr/src/lib.rs index b51443600d..a0b216854f 100644 --- a/crates/git-credential-nostr/src/lib.rs +++ b/crates/git-credential-nostr/src/lib.rs @@ -1,4 +1,5 @@ //! git-credential-nostr — NIP-98 git credential helper for Buzz. +#![warn(missing_docs)] //! //! Git calls this via the credential helper protocol (stdin/stdout). //! We read the request, sign a kind:27235 event, and return the base64-encoded diff --git a/crates/git-sign-nostr/src/lib.rs b/crates/git-sign-nostr/src/lib.rs index d316711200..2c3eb2cefe 100644 --- a/crates/git-sign-nostr/src/lib.rs +++ b/crates/git-sign-nostr/src/lib.rs @@ -1,4 +1,5 @@ //! git-sign-nostr — NIP-GS git object signing with Nostr keys. +#![warn(missing_docs)] //! //! A pluggable git signing program (`gpg.x509.program`) that signs commits //! and tags with BIP-340 Schnorr signatures using the signer's Nostr keypair.