feat(cli): --verbose drain logging; e2e runs against production

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`) |

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,29 @@ 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,
+		// 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",
 	};
 	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 +422,56 @@ 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 });
+	}
+	// 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) {
+		// 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`,
+		);
 	}
-	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();
@@ -493,8 +538,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
@@ -543,6 +589,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 {}

cli/src/forward.rs

@@ -76,6 +76,27 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result<i32>
     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<i32>
         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));
@@ -134,7 +136,21 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result<i32>
         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 +165,13 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result<i32>
         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,23 +220,71 @@ pub async fn run(session: &Session, out: Out, args: ForwardArgs) -> Result<i32>
         }
         // 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)
 }
 
+/// 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<Self> {
+        #[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);

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) => {

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 {