From c4aff946ba0b9e9d2d21323765cf9d80e04a7612 Mon Sep 17 00:00:00 2001 From: hallelx2 Date: Mon, 12 Oct 2026 00:10:39 +0100 Subject: [PATCH 1/3] feat(cli): --verbose drain logging; run the e2e against production A live webhook once missed the 15 s window in production and nothing showed whether the stream wake was lost or the network stalled (stalls of 36 s and 86 s were measured the same night). --verbose logs stream frames, each pass with its duration, and what ended each wait, on stderr. The e2e now sends the dashboard origin (Better-Auth refuses Node's fetch without one), times live delivery from the relay's own receivedAt with the end-to-end figure beside it, and prints each forward's last events and verbose log when a check fails. --- cli/README.md | 1 + cli/e2e/cli.e2e.mjs | 49 ++++++++++++++++++++++++++++++++++++--------- cli/src/forward.rs | 40 +++++++++++++++++++++++++++--------- cli/src/main.rs | 6 +++++- cli/src/output.rs | 17 ++++++++++++++-- 5 files changed, 91 insertions(+), 22 deletions(-) diff --git a/cli/README.md b/cli/README.md index dcab0a3..afe8554 100644 --- a/cli/README.md +++ b/cli/README.md @@ -19,6 +19,7 @@ Logging in again (or with `--force`) revokes the token it replaces, so re-pairin | Flag / variable | | |---|---| | `--json` | one JSON object per line (`ready`, `delivered`, `skipped`, `target_down`, `target_up`, …) for scripts and agents | +| `--verbose`, `-v`, `BRIDGEHOOK_VERBOSE` | log stream frames, drain passes and waits to stderr, to see why a webhook is slow or waiting | | `--relay`, `BRIDGEHOOK_RELAY` | another relay (self-hosted); defaults to the one you signed in to | | `BRIDGEHOOK_TOKEN` | a device token, instead of the saved one (CI, containers) | | `--config-dir`, `BRIDGEHOOK_CONFIG_DIR` | where the token, channel keys and channel list live (default `~/.config/bridgehook`, `%APPDATA%\bridgehook`) | diff --git a/cli/e2e/cli.e2e.mjs b/cli/e2e/cli.e2e.mjs index 687fa39..2de511c 100644 --- a/cli/e2e/cli.e2e.mjs +++ b/cli/e2e/cli.e2e.mjs @@ -140,9 +140,9 @@ async function api(path, init = {}) { ...init, headers: { "content-type": "application/json", - // A browser on the dashboard sends its origin; locally the - // dashboard origin (WEB_URL) is the relay itself. - ...(LOCAL ? { origin: RELAY } : {}), + // The dashboard's origin, as a browser on it sends (Better-Auth + // refuses Node's fetch without one). Locally WEB_URL is the relay. + origin: LOCAL ? RELAY : (process.env.BRIDGEHOOK_WEB_URL ?? "https://app.bridgehook.dev"), ...(cookie ? { cookie } : {}), ...(init.headers ?? {}), }, @@ -214,20 +214,27 @@ async function resetAccount(label) { const CONFIG = mkdtempSync(join(tmpdir(), "bh-cli-e2e-config-")); cleanup.push(() => rmSync(CONFIG, { recursive: true, force: true })); +/** Every CLI run, so a failure can show what each one last reported. */ +const runs = []; + function cli(args) { const env = { ...process.env, BRIDGEHOOK_CONFIG_DIR: CONFIG, + BRIDGEHOOK_VERBOSE: "true", BRIDGEHOOK_RELAY: RELAY, NO_COLOR: "1", }; env.BRIDGEHOOK_TOKEN = ""; const child = spawn(BIN, ["--json", ...args], { env, stdio: ["ignore", "pipe", "pipe"] }); - const events = []; + const record = { args, events: [], stderr: "" }; + runs.push(record); + const events = record.events; const waiters = []; let stderr = ""; child.stderr.on("data", (d) => { stderr += d; + record.stderr += d; }); createInterface({ input: child.stdout }).on("line", (line) => { let e; @@ -413,20 +420,35 @@ try { 30_000, "stream connected", ); - const liveLatencies = []; + // The criterion is the relay-to-localhost leg: from the relay storing the + // webhook (its own receivedAt) to localhost getting it. A real sender + // (Stripe, GitHub) posts from a data centre near Cloudflare; this + // machine's own upload is reported beside it, not counted. Clocks are + // NTP-synced; skew is a few ms. + const live = []; for (let i = 0; i < 3; i++) { const before = received.length; const sent = Date.now(); await sendWebhook(`live-${i}`); await waitReceived(before + 1, 15_000); - liveLatencies.push(received[before].at - sent); + live.push({ tag: `live-${i}`, sent, at: received[before].at }); } - const best = Math.min(...liveLatencies); + const { events: history = [] } = await ( + await api(`/api/me/events?channel=${ready.channelId}&limit=50`) + ).json(); + const relayLeg = live.map((l) => { + const evt = history.find((e) => e.path.includes(`tag=${l.tag}`)); + check(evt, `the relay's record of ${l.tag}`); + return l.at - new Date(evt.receivedAt).getTime(); + }); + const endToEnd = live.map((l) => l.at - l.sent); check( - Math.max(...liveLatencies) < LIVE_LIMIT_MS, - `live delivery under ${LIVE_LIMIT_MS} ms: ${liveLatencies.join(", ")} ms`, + Math.max(...relayLeg) < LIVE_LIMIT_MS, + `live delivery under ${LIVE_LIMIT_MS} ms from the relay: ${relayLeg.join(", ")} ms`, + ); + step( + `live delivery: ${relayLeg.join(", ")} ms from the relay accepting it (${endToEnd.join(", ")} ms from this machine's send)`, ); - step(`live delivery: ${liveLatencies.join(", ")} ms (best ${best})`); // 5. target down: webhooks wait, then arrive in order. await stopServer(); @@ -543,6 +565,13 @@ try { } catch (err) { failed = true; console.error(`\nFAIL: ${err.message}`); + for (const r of runs.filter((r) => r.args[0] === "forward")) { + const tail = r.events.slice(-12).map(({ at, ...e }) => JSON.stringify(e)); + const log = r.stderr.trim().split("\n").slice(-30); + console.error( + ` forward ${r.args.slice(1).join(" ")}: last events\n ${tail.join("\n ")}\n verbose log\n ${log.join("\n ")}`, + ); + } try { await resetAccount("teardown"); } catch {} diff --git a/cli/src/forward.rs b/cli/src/forward.rs index 75a5ce2..0d48d92 100644 --- a/cli/src/forward.rs +++ b/cli/src/forward.rs @@ -134,7 +134,21 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result let ours = channel.id.clone(); tokio::spawn(async move { stream::run(relay, connected, stop, move |frame| { - if frame.kind == "webhook" && frame.channel_id.as_deref() == Some(ours.as_str()) { + let mine = frame.channel_id.as_deref() == Some(ours.as_str()); + out.debug(|| { + format!( + "stream frame {} channel={} id={}{}", + frame.kind, + frame.channel_id.as_deref().unwrap_or("-"), + frame.id.as_deref().unwrap_or("-"), + if frame.kind == "webhook" && mine { + " (wake)" + } else { + "" + } + ) + }); + if frame.kind == "webhook" && mine { wake.notify_one(); } }) @@ -149,7 +163,13 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result if stop.load(Ordering::SeqCst) { break; } - let wait = match drainer.pass(&deps, &mut notify).await { + let started = std::time::Instant::now(); + let result = drainer.pass(&deps, &mut notify).await; + out.debug(|| match &result { + Ok(r) => format!("pass {r:?} in {} ms", started.elapsed().as_millis()), + Err(e) => format!("pass failed in {} ms: {e:#}", started.elapsed().as_millis()), + }); + let wait = match result { Ok(PassResult::Idle) => { retry = RETRY_MIN; if connected.load(Ordering::Relaxed) { @@ -198,18 +218,20 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result } // Wait for the timer or Ctrl-C, or for the stream to report a // webhook (useless while the local server is down). - if drainer.target_down() { + out.debug(|| format!("waiting up to {} ms", wait.as_millis())); + let woke = if drainer.target_down() { tokio::select! { - _ = stopped.notified() => {} - _ = tokio::time::sleep(wait) => {} + _ = stopped.notified() => "stop", + _ = tokio::time::sleep(wait) => "timer", } } else { tokio::select! { - _ = stopped.notified() => {} - _ = wake.notified() => {} - _ = tokio::time::sleep(wait) => {} + _ = stopped.notified() => "stop", + _ = wake.notified() => "stream", + _ = tokio::time::sleep(wait) => "timer", } - } + }; + out.debug(|| format!("woken by {woke}")); } out.emit(json!({ "event": "stopped" }), || "Stopped.".into()); Ok(0) diff --git a/cli/src/main.rs b/cli/src/main.rs index 550b670..1949468 100644 --- a/cli/src/main.rs +++ b/cli/src/main.rs @@ -27,6 +27,10 @@ struct Cli { #[arg(long, global = true)] json: bool, + /// Log stream frames, drain passes and waits to stderr (for debugging). + #[arg(long, short = 'v', env = "BRIDGEHOOK_VERBOSE", global = true)] + verbose: bool, + /// Where the token, keys and channel list are kept. #[arg( long, @@ -62,7 +66,7 @@ enum Command { #[tokio::main] async fn main() { let cli = Cli::parse(); - let out = Out::new(cli.json); + let out = Out::new(cli.json, cli.verbose); let code = match run(cli, out).await { Ok(code) => code, Err(e) => { diff --git a/cli/src/output.rs b/cli/src/output.rs index 7f6e66a..4b0c71a 100644 --- a/cli/src/output.rs +++ b/cli/src/output.rs @@ -8,14 +8,27 @@ use std::io::{IsTerminal, Write}; #[derive(Clone, Copy)] pub struct Out { pub json: bool, + verbose: bool, color: bool, } impl Out { - pub fn new(json: bool) -> Self { + pub fn new(json: bool, verbose: bool) -> Self { let color = !json && std::io::stdout().is_terminal() && std::env::var_os("NO_COLOR").is_none(); - Self { json, color } + Self { + json, + verbose, + color, + } + } + + /// With `--verbose`: a timestamped line on stderr (in either mode, so + /// `--json` stdout stays machine-readable). + pub fn debug(&self, s: impl FnOnce() -> String) { + if self.verbose { + eprintln!("{} {}", chrono::Local::now().format("%H:%M:%S%.3f"), s()); + } } fn paint(&self, code: &str, s: &str) -> String { From 78fe405e3d1b348db60aaa55bf1a61233bac084c Mon Sep 17 00:00:00 2001 From: hallelx2 Date: Mon, 12 Oct 2026 00:22:57 +0100 Subject: [PATCH 2/3] fix(cli): register Ctrl-C and SIGTERM before printing ready The handler was installed by a spawned task after the ready line, so a Ctrl-C right after ready could reach the default handler and kill forward mid-delivery (the production e2e caught it). SIGTERM, which docker stop and systemd send, now stops as gracefully as Ctrl-C. e2e: verbose logging only for forward runs, a clock-skew guard on the relay-leg latency in production, and SIGTERM coverage. --- cli/e2e/cli.e2e.mjs | 24 +++++++++++-- cli/src/forward.rs | 86 +++++++++++++++++++++++++++++++++++---------- 2 files changed, 88 insertions(+), 22 deletions(-) diff --git a/cli/e2e/cli.e2e.mjs b/cli/e2e/cli.e2e.mjs index 2de511c..f3b7f37 100644 --- a/cli/e2e/cli.e2e.mjs +++ b/cli/e2e/cli.e2e.mjs @@ -221,7 +221,9 @@ function cli(args) { const env = { ...process.env, BRIDGEHOOK_CONFIG_DIR: CONFIG, - BRIDGEHOOK_VERBOSE: "true", + // Only forward logs its drain (for the failure dump); other commands' + // stderr stays their own error lines. + BRIDGEHOOK_VERBOSE: args[0] === "forward" ? "true" : "false", BRIDGEHOOK_RELAY: RELAY, NO_COLOR: "1", }; @@ -433,6 +435,21 @@ try { await waitReceived(before + 1, 15_000); live.push({ tag: `live-${i}`, sent, at: received[before].at }); } + // The relay leg compares the relay's clock with this machine's: refuse + // to measure across a skew the Date header can see (1 s resolution). + // A local relay shares this machine's clock; nothing to compare. + if (!LOCAL) { + const t0 = Date.now(); + const probe = await fetch(`${RELAY}/health`); + const t1 = Date.now(); + const date = probe.headers.get("date"); + check(date, "the relay sends a Date header to check clock skew against"); + const skew = new Date(date).getTime() + 500 - (t0 + t1) / 2; + check( + Math.abs(skew) < 1500, + `clock skew against the relay is about ${Math.round(skew)} ms; sync this clock (NTP) first`, + ); + } const { events: history = [] } = await ( await api(`/api/me/events?channel=${ready.channelId}&limit=50`) ).json(); @@ -515,8 +532,9 @@ try { check(allowed.status < 300, `a path inside --path is accepted (got ${allowed.status})`); await adopt.waitFor((e) => e.event === "delivered", 15_000, "delivery on the adopted channel"); check(hits.includes("/only/x"), "the adopted channel delivers to localhost"); - adopt.child.kill("SIGINT"); - check((await adopt.exited) === 0, "adopted forward stops cleanly"); + // SIGTERM, as `docker stop` or systemd sends: the same graceful stop as Ctrl-C. + adopt.child.kill("SIGTERM"); + check((await adopt.exited) === 0, "SIGTERM stops forward cleanly"); step("takeover: refused without --adopt; --adopt keeps the URL and applies --path"); // 7. Ctrl-C stops promptly, then a second login revokes the token it diff --git a/cli/src/forward.rs b/cli/src/forward.rs index 0d48d92..e0de87e 100644 --- a/cli/src/forward.rs +++ b/cli/src/forward.rs @@ -76,6 +76,27 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result apply_paths(session, &relay, &channel, &args, out).await?; let executor = config::executor_id(&session.paths)?; + // Ctrl-C (or SIGTERM from `docker stop` / systemd): finish the webhook in + // hand, then stop; a second one quits at once. The handlers are + // registered here, before `ready` is printed, so a signal right after it + // cannot reach the default handler and kill the process mid-delivery. + let stop = Arc::new(AtomicBool::new(false)); + let stopped = Arc::new(Notify::new()); + let mut signals = StopSignals::register()?; + { + let (stop, stopped) = (stop.clone(), stopped.clone()); + tokio::spawn(async move { + signals.next().await; + stop.store(true, Ordering::SeqCst); + stopped.notify_one(); + if !out.json { + eprintln!("\nStopping after the current webhook; Ctrl-C again to quit now."); + } + signals.next().await; + std::process::exit(130); + }); + } + out.emit( json!({ "event": "ready", @@ -105,25 +126,6 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result target: target.clone(), }; - // Ctrl-C: finish the webhook in hand, then stop. A second Ctrl-C quits. - let stop = Arc::new(AtomicBool::new(false)); - let stopped = Arc::new(Notify::new()); - { - let (stop, stopped) = (stop.clone(), stopped.clone()); - tokio::spawn(async move { - if tokio::signal::ctrl_c().await.is_ok() { - stop.store(true, Ordering::SeqCst); - stopped.notify_one(); - if !out.json { - eprintln!("\nStopping after the current webhook; Ctrl-C again to quit now."); - } - if tokio::signal::ctrl_c().await.is_ok() { - std::process::exit(130); - } - } - }); - } - // The live stream needs a signed-in token; signed out, the queue is polled. let wake = Arc::new(Notify::new()); let connected = Arc::new(AtomicBool::new(false)); @@ -237,6 +239,52 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result Ok(0) } +/// SIGINT and SIGTERM on Unix, Ctrl-C and Ctrl-Break on Windows, registered +/// when created. +struct StopSignals { + #[cfg(unix)] + int: tokio::signal::unix::Signal, + #[cfg(unix)] + term: tokio::signal::unix::Signal, + #[cfg(windows)] + c: tokio::signal::windows::CtrlC, + #[cfg(windows)] + brk: tokio::signal::windows::CtrlBreak, +} + +impl StopSignals { + fn register() -> Result { + #[cfg(unix)] + { + use tokio::signal::unix::{signal, SignalKind}; + Ok(Self { + int: signal(SignalKind::interrupt())?, + term: signal(SignalKind::terminate())?, + }) + } + #[cfg(windows)] + { + Ok(Self { + c: tokio::signal::windows::ctrl_c()?, + brk: tokio::signal::windows::ctrl_break()?, + }) + } + } + + async fn next(&mut self) { + #[cfg(unix)] + tokio::select! { + _ = self.int.recv() => {} + _ = self.term.recv() => {} + } + #[cfg(windows)] + tokio::select! { + _ = self.c.recv() => {} + _ = self.brk.recv() => {} + } + } +} + fn next_retry(retry: &mut Duration, max: Duration) -> Duration { let w = (*retry).min(max); *retry = (*retry * 2).min(max); From 898d7e8c0e2d7978c9b6c4d93fa92ed1926d2aa4 Mon Sep 17 00:00:00 2001 From: hallelx2 Date: Mon, 12 Oct 2026 09:19:33 +0100 Subject: [PATCH 3/3] test(cli): take the fastest of three probes for the clock-skew guard A network stall during a single probe shifts its midpoint by half the stall and fakes a skew; review finding on #19. --- cli/e2e/cli.e2e.mjs | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/cli/e2e/cli.e2e.mjs b/cli/e2e/cli.e2e.mjs index f3b7f37..2fb87e4 100644 --- a/cli/e2e/cli.e2e.mjs +++ b/cli/e2e/cli.e2e.mjs @@ -439,12 +439,18 @@ try { // to measure across a skew the Date header can see (1 s resolution). // A local relay shares this machine's clock; nothing to compare. if (!LOCAL) { - const t0 = Date.now(); - const probe = await fetch(`${RELAY}/health`); - const t1 = Date.now(); - const date = probe.headers.get("date"); - check(date, "the relay sends a Date header to check clock skew against"); - const skew = new Date(date).getTime() + 500 - (t0 + t1) / 2; + // The fastest of three probes: a network stall would shift a single + // sample's midpoint by half the stall and fake a skew. + let best = null; + for (let i = 0; i < 3; i++) { + const t0 = Date.now(); + const probe = await fetch(`${RELAY}/health`); + const t1 = Date.now(); + const date = probe.headers.get("date"); + check(date, "the relay sends a Date header to check clock skew against"); + if (!best || t1 - t0 < best.rtt) best = { rtt: t1 - t0, mid: (t0 + t1) / 2, date }; + } + const skew = new Date(best.date).getTime() + 500 - best.mid; check( Math.abs(skew) < 1500, `clock skew against the relay is about ${Math.round(skew)} ms; sync this clock (NTP) first`,