From bb7fd6f12229c06799188dd1b079c3bb6ef7e3eb Mon Sep 17 00:00:00 2001 From: Taksh Date: Thu, 23 Jul 2026 13:16:57 +0300 Subject: [PATCH 1/3] fix(workflow): route add_reaction through ActionSink Signed-off-by: Taksh Co-authored-by: Cursor --- crates/buzz-relay/src/workflow_sink.rs | 141 +++++++++++++++++++++++- crates/buzz-workflow/src/action_sink.rs | 18 +++ crates/buzz-workflow/src/executor.rs | 115 ++++++------------- 3 files changed, 193 insertions(+), 81 deletions(-) diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..c8077afc77 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -8,7 +8,7 @@ use std::future::Future; use std::pin::Pin; use std::sync::{Arc, Weak}; -use buzz_core::kind::KIND_STREAM_MESSAGE; +use buzz_core::kind::{KIND_REACTION, KIND_STREAM_MESSAGE}; use buzz_core::tenant::CommunityId; use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; use chrono::Utc; @@ -362,6 +362,145 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn add_reaction( + &self, + community_id: CommunityId, + message_id: &str, + emoji: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>> { + let message_id = message_id.to_owned(); + let emoji = emoji.to_owned(); + let author_pubkey = author_pubkey.to_owned(); + + Box::pin(async move { + // 0. Upgrade weak reference — fails only during shutdown. + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + // 1. Resolve community → TenantContext. + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + // 2. Parse author pubkey. + let author_pubkey_key = nostr::PublicKey::from_hex(&author_pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}")) + })?; + let author_pubkey_hex = author_pubkey_key.to_hex(); + let author_bytes = author_pubkey_key.to_bytes().to_vec(); + + // 3. Parse message_id as EventId and look up target event. + let target_event_id = nostr::EventId::from_hex(&message_id) + .map_err(|e| ActionSinkError::InvalidInput(format!("invalid message_id: {e}")))?; + let target_bytes = target_event_id.as_bytes().to_vec(); + + let target = state + .db + .get_event_by_id(tenant.community(), &target_bytes) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::InvalidInput(format!( + "reaction target event not found: {message_id}" + )) + })?; + + // 4. Normalise emoji: empty → "+"; trim; reject if too long. + let emoji_normalised = if emoji.is_empty() { + "+".to_owned() + } else { + emoji.trim().to_owned() + }; + if emoji_normalised.chars().count() > 64 { + return Err(ActionSinkError::InvalidInput( + "emoji exceeds 64 characters".into(), + )); + } + + // 5. Build kind:7 event. + // Tags: e (target), p (author), buzz:workflow true. + let tags = vec![ + Tag::parse(["e", &message_id]) + .map_err(|e| ActionSinkError::EventBuild(format!("e tag: {e}")))?, + Tag::parse(["p", &author_pubkey_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, + Tag::parse(["buzz:workflow", "true"]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + ]; + + let kind = Kind::from(KIND_REACTION as u16); + let event = EventBuilder::new(kind, &emoji_normalised) + .tags(tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; + + let event_id_hex = event.id.to_hex(); + + info!( + event_id = %event_id_hex, + target = %message_id, + author = %author_pubkey_hex, + emoji = %emoji_normalised, + "Workflow AddReaction: posting kind:7 event" + ); + + // 6. Persist via insert_reaction_event_with_thread_metadata. + let channel_id = target.channel_id; + match state + .db + .insert_reaction_event_with_thread_metadata( + tenant.community(), + &event, + channel_id, + None, + &target_bytes, + &author_bytes, + &emoji_normalised, + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + { + buzz_db::ReactionEventInsertOutcome::Inserted { + stored_event, + was_inserted, + } => { + if was_inserted { + let _ = dispatch_persistent_event( + &tenant, + &state, + &stored_event, + KIND_REACTION, + &author_pubkey_hex, + None, + ) + .await; + } + Ok(event_id_hex) + } + buzz_db::ReactionEventInsertOutcome::Duplicate => { + // Idempotent — return the original message_id. + Ok(message_id) + } + buzz_db::ReactionEventInsertOutcome::TargetMissing => { + Err(ActionSinkError::InvalidInput(format!( + "reaction target event not found: {message_id}" + ))) + } + } + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..056044b699 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -66,4 +66,22 @@ pub trait ActionSink: Send + Sync { text: &str, author_pubkey: &str, ) -> Pin> + Send + '_>>; + + /// Add a reaction to a message on behalf of a workflow owner. + /// + /// - `community_id`: the community that owns the workflow run + /// - `message_id`: hex event ID of the target message + /// - `emoji`: reaction emoji (empty → "+"; must be ≤ 64 chars) + /// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for + /// the `p` attribution tag; the relay keypair signs the kind:7 event) + /// + /// Returns the event ID hex string on success. Duplicate reactions are + /// treated as idempotent and return the original message_id. + fn add_reaction( + &self, + community_id: CommunityId, + message_id: &str, + emoji: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index a029b44622..985186c9cd 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -597,23 +597,42 @@ pub async fn dispatch_action( )); } - #[cfg(feature = "reqwest")] - { - let result = add_reaction_impl(&trigger_ctx.message_id, emoji).await?; - Ok(StepResult::Completed(result)) - } + let wf_run = engine + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "AddReaction: failed to load workflow run {run_id}: {e}" + )) + })?; + let workflow = engine + .db + .get_workflow(community_id, wf_run.workflow_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "AddReaction: failed to load workflow {}: {e}", + wf_run.workflow_id + )) + })?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); - #[cfg(not(feature = "reqwest"))] - { - warn!( - run_id = %run_id, - step = step_id, - "AddReaction: reqwest feature not enabled, skipping HTTP call" - ); - Ok(StepResult::Completed( - serde_json::json!({ "added": false, "skipped": true }), - )) - } + let event_id = engine + .action_sink()? + .add_reaction( + community_id, + &trigger_ctx.message_id, + emoji, + &owner_pubkey_hex, + ) + .await + .map_err(WorkflowError::from)?; + + Ok(StepResult::Completed(serde_json::json!({ + "added": true, + "event_id": event_id, + }))) } CallWebhook { @@ -865,70 +884,6 @@ async fn call_webhook_impl( })) } -/// Returns a shared `reqwest::Client` reused across all workflow HTTP calls. -/// Sharing a single client reuses the underlying connection pool. -#[cfg(feature = "reqwest")] -fn shared_http_client() -> &'static reqwest::Client { - use std::sync::LazyLock; - use std::time::Duration; - static CLIENT: LazyLock = LazyLock::new(|| { - reqwest::Client::builder() - .timeout(Duration::from_secs(10)) - .build() - .expect("HTTP client build must succeed") - }); - &CLIENT -} - -/// POST `{"emoji": emoji}` to `POST /api/messages/{message_id}/reactions`. -#[cfg(feature = "reqwest")] -async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result { - let base_url = - std::env::var("BUZZ_RELAY_BASE_URL").unwrap_or_else(|_| "http://localhost:3000".to_owned()); - - let url = format!("{base_url}/api/messages/{message_id}/reactions"); - - let client = shared_http_client(); - - let mut req = client - .post(&url) - .header("Content-Type", "application/json") - .json(&serde_json::json!({ "emoji": emoji })); - - if let Ok(token) = std::env::var("BUZZ_API_TOKEN") { - req = req.header("Authorization", format!("Bearer {token}")); - } else if let Ok(pubkey) = std::env::var("BUZZ_RELAY_PUBKEY") { - req = req.header("X-Pubkey", pubkey); - } - - let resp = req - .send() - .await - .map_err(|e| WorkflowError::WebhookError(format!("AddReaction HTTP error: {e}")))?; - - let status = resp.status(); - - if !status.is_success() { - let body = resp - .text() - .await - .unwrap_or_else(|_| "".to_owned()); - return Err(WorkflowError::WebhookError(format!( - "AddReaction: relay returned {status} for message {message_id}: {body}" - ))); - } - - let body_text = resp.text().await.unwrap_or_else(|_| String::new()); - let body_json: JsonValue = serde_json::from_str(&body_text) - .unwrap_or_else(|_| serde_json::json!({ "raw": body_text })); - - Ok(serde_json::json!({ - "added": true, - "status": status.as_u16(), - "response": body_json, - })) -} - /// Rich return type from `execute_run` / `execute_from_step`. /// /// Carries enough information for the caller to: From a6e6eb08515e97b7ab6faf2950ca6406896d974e Mon Sep 17 00:00:00 2001 From: Taksh Date: Thu, 23 Jul 2026 13:16:57 +0300 Subject: [PATCH 2/3] fix(desktop): skip .cmd Claude CLI shim on Windows Signed-off-by: Taksh Co-authored-by: Cursor --- .../src-tauri/src/managed_agents/runtime.rs | 17 +--- .../managed_agents/runtime/configure_cli.rs | 81 +++++++++++++++++++ .../src/managed_agents/runtime/tests.rs | 31 +++++++ 3 files changed, 115 insertions(+), 14 deletions(-) create mode 100644 desktop/src-tauri/src/managed_agents/runtime/configure_cli.rs diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index cef669ab25..fecb94dc50 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -14,6 +14,9 @@ use crate::{ util::now_iso, }; +mod configure_cli; +pub(crate) use configure_cli::configure_runtime_cli; + mod path; pub(in crate::managed_agents) use path::build_augmented_path; @@ -1591,20 +1594,6 @@ pub(crate) fn build_respond_to_env( Ok((set, remove)) } -pub(crate) fn configure_runtime_cli( - command: &mut std::process::Command, - runtime: Option<&KnownAcpRuntime>, -) { - let Some(runtime) = runtime else { - return; - }; - if runtime.id != "claude" { - return; - } - if let Some(cli_path) = runtime.underlying_cli.and_then(resolve_command) { - command.env("CLAUDE_CODE_EXECUTABLE", cli_path); - } -} /// Spawn an agent process without holding any locks on records or runtimes. /// Returns the child process and log path on success. The caller is responsible diff --git a/desktop/src-tauri/src/managed_agents/runtime/configure_cli.rs b/desktop/src-tauri/src/managed_agents/runtime/configure_cli.rs new file mode 100644 index 0000000000..79d580da04 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/runtime/configure_cli.rs @@ -0,0 +1,81 @@ +//! Configure the `CLAUDE_CODE_EXECUTABLE` environment variable for Claude +//! agent spawns. +//! +//! On Windows, `CLAUDE_CODE_EXECUTABLE` must point to an actual executable, +//! not a `.cmd` or `.bat` shim. Passing a shim path causes `CreateProcess` +//! to fail with `EINVAL` because the kernel cannot directly execute a batch +//! script. When the resolved CLI path has a `.cmd` / `.bat` extension we skip +//! setting the variable so the agent's own PATH lookup finds the real binary. + +use crate::managed_agents::{resolve_command, KnownAcpRuntime}; + +/// Set `CLAUDE_CODE_EXECUTABLE` on `command` when the resolved CLI path is +/// safe to pass directly to `CreateProcess` / `execve`. +/// +/// Skips the env-var on Windows when the resolved path ends in `.cmd` or +/// `.bat` (case-insensitive) — those are batch shims that spawn a new +/// `cmd.exe` process and cannot be exec'd directly, causing `EINVAL`. +pub(crate) fn configure_runtime_cli( + command: &mut std::process::Command, + runtime: Option<&KnownAcpRuntime>, +) { + let Some(runtime) = runtime else { + return; + }; + if runtime.id != "claude" { + return; + } + if let Some(cli_path) = runtime.underlying_cli.and_then(resolve_command) { + #[cfg(windows)] + { + if let Some(ext) = cli_path.extension() { + let ext_lower = ext.to_string_lossy().to_lowercase(); + if ext_lower == "cmd" || ext_lower == "bat" { + // Batch shim — skip CLAUDE_CODE_EXECUTABLE to avoid + // CreateProcess EINVAL; the agent will find the real + // binary via PATH. + return; + } + } + } + command.env("CLAUDE_CODE_EXECUTABLE", cli_path); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::managed_agents::known_acp_runtime; + + /// On Windows, a `.cmd` shim must NOT set CLAUDE_CODE_EXECUTABLE because + /// `CreateProcess` cannot exec batch scripts directly (EINVAL). + #[cfg(windows)] + #[test] + fn cmd_shim_does_not_set_claude_code_executable() { + let _guard = crate::managed_agents::lock_path_mutex(); + let temp = tempfile::tempdir().expect("temp dir"); + + // Write a .cmd shim — no execute bit needed on Windows. + let shim = temp.path().join("claude.cmd"); + std::fs::write(&shim, "@echo off\r\necho shim\r\n").expect("write shim"); + + let original_path = std::env::var_os("PATH"); + std::env::set_var("PATH", temp.path()); + + let mut command = std::process::Command::new("buzz-acp"); + configure_runtime_cli(&mut command, known_acp_runtime("claude-agent-acp")); + + if let Some(path) = original_path { + std::env::set_var("PATH", path); + } else { + std::env::remove_var("PATH"); + } + + assert!( + !command + .get_envs() + .any(|(key, _)| key == "CLAUDE_CODE_EXECUTABLE"), + ".cmd shim must not set CLAUDE_CODE_EXECUTABLE" + ); + } +} diff --git a/desktop/src-tauri/src/managed_agents/runtime/tests.rs b/desktop/src-tauri/src/managed_agents/runtime/tests.rs index d86b69938f..64d5cc3f93 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/tests.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/tests.rs @@ -618,6 +618,37 @@ fn codex_spawn_does_not_set_a_claude_executable() { .any(|(key, _)| key == "CLAUDE_CODE_EXECUTABLE")); } +/// On Windows, a `.cmd` shim must NOT set CLAUDE_CODE_EXECUTABLE because +/// `CreateProcess` cannot exec batch scripts directly (EINVAL). +#[cfg(windows)] +#[test] +fn cmd_shim_does_not_set_claude_code_executable() { + let _guard = crate::managed_agents::lock_path_mutex(); + let temp = tempfile::tempdir().expect("temp dir"); + + let shim = temp.path().join("claude.cmd"); + std::fs::write(&shim, "@echo off\r\necho shim\r\n").expect("write shim"); + + let original_path = std::env::var_os("PATH"); + std::env::set_var("PATH", temp.path()); + + let mut command = std::process::Command::new("buzz-acp"); + super::configure_runtime_cli(&mut command, super::known_acp_runtime("claude-agent-acp")); + + if let Some(path) = original_path { + std::env::set_var("PATH", path); + } else { + std::env::remove_var("PATH"); + } + + assert!( + !command + .get_envs() + .any(|(key, _)| key == "CLAUDE_CODE_EXECUTABLE"), + ".cmd shim must not set CLAUDE_CODE_EXECUTABLE" + ); +} + // ── PGID-based orphan sweep tests ─────────────────────────────────────── /// Validates the kernel invariant that the orphan sweep PGID fix relies on: From 9a5b22a403437e0544c0d526c27ac7373e49314f Mon Sep 17 00:00:00 2001 From: Taksh Date: Thu, 23 Jul 2026 13:16:57 +0300 Subject: [PATCH 3/3] fix(desktop): stop retrying oversized relay frames Signed-off-by: Taksh Co-authored-by: Cursor --- desktop/src/shared/api/relayClientSession.ts | 65 ++++++----- .../src/shared/api/relayClientTransport.ts | 108 ++++++++++++++++++ .../src/shared/api/relayFrameLimit.test.mjs | 89 +++++++++++++++ desktop/src/shared/api/relayFrameLimit.ts | 70 ++++++++++++ 4 files changed, 303 insertions(+), 29 deletions(-) create mode 100644 desktop/src/shared/api/relayClientTransport.ts create mode 100644 desktop/src/shared/api/relayFrameLimit.test.mjs create mode 100644 desktop/src/shared/api/relayFrameLimit.ts diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index 6dcc857156..cd2bba4e98 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -48,6 +48,12 @@ import { } from "@/shared/api/relayReconnectPolicy"; import { RelayStallWatchdog } from "@/shared/api/relayStallWatchdog"; import { closeWebSocket } from "@/shared/api/relayWebSocketClose"; +import { + noteFrameTooLargeLimit, + sendRelayTextFrame, + sendRelayTextFrameWithReconnectRetry, +} from "@/shared/api/relayClientTransport"; +import { isOversizedFrameError } from "@/shared/api/relayFrameLimit"; import { buildThreadReferenceTags } from "@/features/messages/lib/threading"; const RECONNECT_BASE_DELAY_MS = 1_000, RECONNECT_MAX_DELAY_MS = 30_000, @@ -101,6 +107,8 @@ export class RelayClient { private connectionGeneration = 0; private stabilityTimer: number | null = null; private visibleChannelId: string | null = null; + /** Server-advertised maximum frame size in bytes; null until first NOTICE. */ + private maxFrameBytes: number | null = null; /** * Sticky terminal flag. Set when `resetConnection` is called with @@ -158,6 +166,7 @@ export class RelayClient { this.notifyReconnectListeners = false; this.terminal = false; this.visibleChannelId = null; + this.maxFrameBytes = null; this.connectionStateEmitter.set("idle"); if (this.wsId !== null) { @@ -644,17 +653,7 @@ export class RelayClient { } private async sendRaw(payload: unknown[]) { - if (this.wsId === null) { - throw new Error("Relay socket is not connected."); - } - - await invoke("plugin:websocket|send", { - id: this.wsId, - message: { - type: "Text", - data: JSON.stringify(payload), - }, - }); + await sendRelayTextFrame(this.wsId, payload, this.maxFrameBytes); } private normalizeRelayError(error: unknown, fallbackMessage: string) { @@ -666,6 +665,11 @@ export class RelayClient { fallbackMessage: string, ): Error { const normalizedError = this.normalizeRelayError(error, fallbackMessage); + // Oversized frames are a permanent client-side error — resetting the + // connection would just trigger the same rejection on reconnect. + if (isOversizedFrameError(normalizedError)) { + return normalizedError; + } this.resetConnection(normalizedError); return normalizedError; } @@ -674,24 +678,13 @@ export class RelayClient { payload: unknown[], fallbackMessage: string, ) { - try { - await this.sendRaw(payload); - } catch (error) { - const normalizedError = this.recoverFromSocketFailure( - error, - fallbackMessage, - ); - - try { - await this.ensureConnected(); - await this.sendRaw(payload); - } catch (retryError) { - throw this.recoverFromSocketFailure( - retryError, - normalizedError.message, - ); - } - } + await sendRelayTextFrameWithReconnectRetry({ + payload, + fallbackMessage, + sendRaw: (p) => this.sendRaw(p), + recoverFromSocketFailure: (e, m) => this.recoverFromSocketFailure(e, m), + ensureConnected: () => this.ensureConnected(), + }); } private async closeSubscription(subId: string) { @@ -731,6 +724,13 @@ export class RelayClient { sendErrorMessage, ); + // Oversized frames are permanently rejected — never reconnect-retry. + if (isOversizedFrameError(normalizedError)) { + window.clearTimeout(timeout); + reject(normalizedError); + return; + } + try { await this.ensureConnected(); if (!pendingEvent) { @@ -839,6 +839,13 @@ export class RelayClient { if (notice.startsWith("rate-limited:")) { activateRateLimit(parseRateLimitHint(notice)); } + // Frame-size limit advertised by the relay — tighten our budget so + // future frames are pre-checked before sending. + const newMax = noteFrameTooLargeLimit(notice, this.maxFrameBytes); + if (newMax !== null) { + this.maxFrameBytes = newMax; + console.warn(`[relay] Frame size limit updated to ${newMax} bytes`); + } } } diff --git a/desktop/src/shared/api/relayClientTransport.ts b/desktop/src/shared/api/relayClientTransport.ts new file mode 100644 index 0000000000..43e10bdf98 --- /dev/null +++ b/desktop/src/shared/api/relayClientTransport.ts @@ -0,0 +1,108 @@ +/** + * Low-level WebSocket send helpers for the relay client. + * + * Extracted from RelayClient so relayClientSession.ts stays within its line + * budget. This module owns: + * - frame serialisation and byte-limit enforcement + * - the reconnect-retry wrapper (with OversizedFrameError short-circuit) + * - frame-size limit bookkeeping from relay NOTICE messages + */ + +import { invoke } from "@tauri-apps/api/core"; + +import { + assertWithinFrameLimit, + isOversizedFrameError, + parseFrameTooLargeNotice, +} from "@/shared/api/relayFrameLimit"; + +/** + * Serialise `payload` as JSON and send it as a WebSocket text frame. + * + * @param wsId - Active socket ID (null → throws "not connected"). + * @param payload - Array to serialise. Must be non-empty. + * @param maxFrameBytes - When non-null, asserts the serialised size is within + * the limit before sending; throws {@link OversizedFrameError} otherwise. + */ +export async function sendRelayTextFrame( + wsId: number | null, + payload: unknown[], + maxFrameBytes: number | null, +): Promise { + if (wsId === null) { + throw new Error("Relay socket is not connected."); + } + if (maxFrameBytes !== null) { + assertWithinFrameLimit(payload, maxFrameBytes); + } + await invoke("plugin:websocket|send", { + id: wsId, + message: { + type: "Text", + data: JSON.stringify(payload), + }, + }); +} + +/** + * Parse a relay NOTICE for a frame-size limit update. + * + * Returns the new limit (taken from the NOTICE) when the notice matches + * *and* the reported limit is smaller than `currentMax` (or `currentMax` is + * null), indicating the client should tighten its frame budget. Returns + * `null` when the notice is unrelated or the limit has not decreased. + */ +export function noteFrameTooLargeLimit( + notice: string, + currentMax: number | null, +): number | null { + const parsed = parseFrameTooLargeNotice(notice); + if (!parsed) return null; + const { limit } = parsed; + if (currentMax !== null && currentMax <= limit) return null; + return limit; +} + +/** + * Send a relay frame, retrying once after a reconnect on transient failures. + * + * {@link OversizedFrameError} is **never** retried — the same payload will be + * rejected every time, so retrying would only trigger another reconnect storm. + * + * @param options.payload - Frame to send. + * @param options.fallbackMessage - Error message used when the caught error + * cannot be normalised. + * @param options.sendRaw - Sends the frame on the current socket. + * @param options.recoverFromSocketFailure - Called on socket errors; resets + * the connection and returns a normalised Error. + * @param options.ensureConnected - Awaits (re)connection before the retry. + */ +export async function sendRelayTextFrameWithReconnectRetry({ + payload, + fallbackMessage, + sendRaw, + recoverFromSocketFailure, + ensureConnected, +}: { + payload: unknown[]; + fallbackMessage: string; + sendRaw: (payload: unknown[]) => Promise; + recoverFromSocketFailure: (error: unknown, message: string) => Error; + ensureConnected: () => Promise; +}): Promise { + try { + await sendRaw(payload); + } catch (error) { + // Never retry oversized frames — the same payload will always be rejected. + if (isOversizedFrameError(error)) { + throw error; + } + const normalizedError = recoverFromSocketFailure(error, fallbackMessage); + try { + await ensureConnected(); + await sendRaw(payload); + } catch (retryError) { + throw recoverFromSocketFailure(retryError, normalizedError.message); + } + } +} diff --git a/desktop/src/shared/api/relayFrameLimit.test.mjs b/desktop/src/shared/api/relayFrameLimit.test.mjs new file mode 100644 index 0000000000..6dff505769 --- /dev/null +++ b/desktop/src/shared/api/relayFrameLimit.test.mjs @@ -0,0 +1,89 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + OversizedFrameError, + assertWithinFrameLimit, + isOversizedFrameError, + parseFrameTooLargeNotice, + relayFrameByteLength, +} from "./relayFrameLimit.ts"; + +// ── parseFrameTooLargeNotice ───────────────────────────────────────────────── + +test("parseFrameTooLargeNotice: parses valid notice", () => { + const result = parseFrameTooLargeNotice( + "frame too large (65536 bytes, limit 32768)", + ); + assert.deepEqual(result, { bytes: 65536, limit: 32768 }); +}); + +test("parseFrameTooLargeNotice: returns null for unrelated notices", () => { + assert.equal(parseFrameTooLargeNotice("rate-limited: slow down"), null); + assert.equal(parseFrameTooLargeNotice(""), null); + assert.equal(parseFrameTooLargeNotice("frame too large"), null); +}); + +test("parseFrameTooLargeNotice: matches embedded in longer string", () => { + const result = parseFrameTooLargeNotice( + "error: frame too large (100 bytes, limit 50)", + ); + assert.deepEqual(result, { bytes: 100, limit: 50 }); +}); + +// ── relayFrameByteLength ───────────────────────────────────────────────────── + +test("relayFrameByteLength: ASCII payload gives same byte and char count", () => { + const payload = ["EVENT", { id: "abc" }]; + const json = JSON.stringify(payload); + assert.equal(relayFrameByteLength(payload), json.length); +}); + +test("relayFrameByteLength: multi-byte UTF-8 chars counted correctly", () => { + // "€" is 3 bytes in UTF-8 but 1 char + const payload = ["NOTE", "€"]; + const bytes = relayFrameByteLength(payload); + assert.ok(bytes > JSON.stringify(payload).length, "UTF-8 bytes > char count"); +}); + +// ── assertWithinFrameLimit ─────────────────────────────────────────────────── + +test("assertWithinFrameLimit: does not throw when within limit", () => { + const payload = ["REQ", "sub1", {}]; + assert.doesNotThrow(() => assertWithinFrameLimit(payload, 1_000_000)); +}); + +test("assertWithinFrameLimit: throws OversizedFrameError when over limit", () => { + const payload = ["EVENT", { content: "x".repeat(100) }]; + assert.throws(() => assertWithinFrameLimit(payload, 10), OversizedFrameError); +}); + +test("assertWithinFrameLimit: OversizedFrameError carries bytes and limit", () => { + const payload = ["NOTE", "hello"]; + const bytes = relayFrameByteLength(payload); + const limit = 1; + try { + assertWithinFrameLimit(payload, limit); + assert.fail("expected OversizedFrameError"); + } catch (err) { + assert.ok(err instanceof OversizedFrameError); + assert.equal(err.bytes, bytes); + assert.equal(err.limit, limit); + } +}); + +// ── isOversizedFrameError ──────────────────────────────────────────────────── + +test("isOversizedFrameError: true for OversizedFrameError", () => { + assert.ok(isOversizedFrameError(new OversizedFrameError(100, 50))); +}); + +test("isOversizedFrameError: false for plain Error", () => { + assert.equal(isOversizedFrameError(new Error("oops")), false); +}); + +test("isOversizedFrameError: false for non-error values", () => { + assert.equal(isOversizedFrameError(null), false); + assert.equal(isOversizedFrameError("string"), false); + assert.equal(isOversizedFrameError(42), false); +}); diff --git a/desktop/src/shared/api/relayFrameLimit.ts b/desktop/src/shared/api/relayFrameLimit.ts new file mode 100644 index 0000000000..95a510461d --- /dev/null +++ b/desktop/src/shared/api/relayFrameLimit.ts @@ -0,0 +1,70 @@ +/** + * Utilities for enforcing relay WebSocket frame size limits. + * + * The relay may reject oversized frames with a NOTICE message of the form: + * "frame too large (N bytes, limit M)" + * When that happens, the client should NOT reconnect and retry — the same + * payload will be rejected every time. Instead, surface an OversizedFrameError + * so callers can propagate it without triggering the reconnect loop. + */ + +/** Thrown when a serialised WebSocket frame exceeds the server's size limit. */ +export class OversizedFrameError extends Error { + readonly bytes: number; + readonly limit: number; + + constructor(bytes: number, limit: number) { + super( + `Relay frame too large: ${bytes} bytes exceeds limit of ${limit} bytes`, + ); + this.name = "OversizedFrameError"; + this.bytes = bytes; + this.limit = limit; + } +} + +/** Type-guard for OversizedFrameError. */ +export function isOversizedFrameError( + error: unknown, +): error is OversizedFrameError { + return error instanceof OversizedFrameError; +} + +/** + * Parse a relay NOTICE that signals a frame-size rejection. + * + * Matches: `frame too large (N bytes, limit M)` + * Returns `{ bytes, limit }` on match, or `null` otherwise. + */ +export function parseFrameTooLargeNotice( + notice: string, +): { bytes: number; limit: number } | null { + const match = /frame too large \((\d+) bytes, limit (\d+)\)/.exec(notice); + if (!match) return null; + return { bytes: Number(match[1]), limit: Number(match[2]) }; +} + +/** + * Compute the UTF-8 byte length of a serialised WebSocket payload. + * + * This mirrors the byte count the relay sees on the wire so the client can + * pre-flight frames before sending, avoiding a round-trip rejection. + */ +export function relayFrameByteLength(payload: unknown[]): number { + return new TextEncoder().encode(JSON.stringify(payload)).byteLength; +} + +/** + * Assert that the payload's wire size is within `maxBytes`. + * + * @throws {OversizedFrameError} when the frame would exceed the limit. + */ +export function assertWithinFrameLimit( + payload: unknown[], + maxBytes: number, +): void { + const bytes = relayFrameByteLength(payload); + if (bytes > maxBytes) { + throw new OversizedFrameError(bytes, maxBytes); + } +}