From dc118dcb2beb372b6914317238c5c81629124773 Mon Sep 17 00:00:00 2001 From: tv Date: Mon, 9 Nov 2026 01:00:28 +0000 Subject: dashboard: where a stream's lag comes from MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stream's lag in the tooltip of `behind: …` was one figure, from when a node logged a message to when the page had it: a second and a half for a node a LAN away said nothing of where it went. The node now stamps each `messages` event with when it sent it (`{"sent": …, "messages": […]}`), and the tooltip splits each stream's lag: how long its batches waited in the node, from logging the first message to sending the batch (the node's own clock: exact), and how long they took on the way, from sending to arriving. This node's own stream is always listed, beside the three slowest. --- src/api/dashboard/dashboard.html | 71 ++++++++++++++++++++++++++++------------ src/api/dashboard/mod.rs | 17 +++++++--- 2 files changed, 63 insertions(+), 25 deletions(-) (limited to 'src/api/dashboard') diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index 1d95db1..c99d773 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -2565,7 +2565,11 @@ function follow(id, delay = 2000) { const followMessages = (wait = 2000) => { if (controller.signal.aborted) return; read(messagesUrl, controller.signal, () => { entry.messagesLive = true; wait = 2000; }, - (kind, data) => { if (kind === "messages") remoteMessagesArrived(id, entry, JSON.parse(data)); }) + (kind, data) => { + if (kind !== "messages") return; + const batch = JSON.parse(data); + remoteMessagesArrived(id, entry, batch.messages, batch.sent); + }) .catch(() => { entry.messagesLive = false; if (!controller.signal.aborted) setTimeout(() => followMessages(Math.min(60000, wait * 2)), wait); @@ -3108,12 +3112,22 @@ function aheadOf(node) { const self = clocksKnown[state.node.id], it = clocksKnown[node]; return self && it ? it.offsetMs - self.offsetMs : 0; } -// `node`'s stream brought a batch whose earliest message lagged `lag` ms (by the -// browser's clock, against this node's): the most of it, which the delay must cover. -function noteLag(node, lag) { - const now = performance.now(); +// A batch of `node`'s stream came, sent by it at `sent` (its clock): how long its +// messages lagged, the most of them (by the browser's clock against this node's, +// `aheadOf`), which the delay must cover; and of that, how long they waited in the node +// (`sent` against when it logged them: its own clock, exact) and how long the batch took +// on the way (from `sent`, by the clocks as known). A batch of a gap (lagging more than +// `REPLAY_MOST_MS`) counts for nothing: it would hold the map that far behind for a while. +function noteStream(node, list, sent) { + if (!state || list.length === 0) return; + const ahead = aheadOf(node), wall = Date.now(), now = performance.now(); + const earliest = Math.min(...list.map(message => message.at)); + const lag = wall - (earliest - ahead); + if (lag > REPLAY_MOST_MS) return; + const inNode = typeof sent === "number" ? sent - earliest : null; + const way = typeof sent === "number" ? wall - (sent - ahead) : null; const kept = (lags.get(node) || []).filter(([at]) => now - at <= REPLAY_WINDOW_MS); - kept.push([now, lag]); + kept.push([now, lag, inNode, way]); lags.set(node, kept); for (const [other, samples] of lags) { if (!samples.some(([at]) => now - at <= REPLAY_WINDOW_MS)) lags.delete(other); @@ -3139,12 +3153,17 @@ function showReplay() { } function explainReplay() { const shown = document.getElementById("replay-delay"); - if (!shown) return; + if (!shown || !state) return; const s = ms => (ms / 1000).toFixed(2) + " s"; - const each = [...lags].map(([node, samples]) => ({ node, - least: Math.min(...samples.map(([, lag]) => lag)), - most: Math.max(...samples.map(([, lag]) => lag)) })) - .sort((x, y) => y.most - x.most); + const range = values => values.length === 0 ? "?" + : Math.min(...values) === Math.max(...values) ? s(values[0]) + : s(Math.min(...values)) + "–" + s(Math.max(...values)); + const self = state.node.id; + const each = [...lags].map(([node, samples]) => { + const of = i => samples.map(sample => sample[i]).filter(value => typeof value === "number"); + return { node, least: Math.min(...of(1)), most: Math.max(...of(1)), + inNode: of(2), way: of(3) }; + }).sort((x, y) => y.most - x.most); if (each.length === 0) { shown.title = "Messages run when they went, this far behind real time; no stream yet."; return; @@ -3152,13 +3171,21 @@ function explainReplay() { const slowest = each[0]; const bound = slowest.most + REPLAY_MARGIN_MS < REPLAY_LEAST_MS ? ", raised to the least, " + s(REPLAY_LEAST_MS) : slowest.most + REPLAY_MARGIN_MS > REPLAY_MOST_MS ? ", cut to the most, " + s(REPLAY_MOST_MS) : ""; + const line = ({ node, least, most, inNode, way }) => + (node === self ? "this node (" + name(node) + ")" : name(node)) + ": " + s(most) + + " (least " + s(least) + "); in the node " + range(inNode) + ", on the way " + range(way); + // The three slowest, and this node's own stream always. + const listed = each.slice(0, 3); + const own = each.find(stream => stream.node === self); + if (own && !listed.includes(own)) listed.push(own); shown.title = "Messages run when they went, this far behind real time, so the slowest" + " stream's are in too:\n" + s(replayDelay) + " = " + name(slowest.node) + "'s lag, the most" + " of any stream in the last 30 s (" + s(slowest.most) + "), + " + s(REPLAY_MARGIN_MS) + - bound + "\n\nThe slowest streams (the most and the least each lagged, last 30 s):\n" + - each.slice(0, 3).map(({ node, least, most }, i) => (i + 1) + ". " + name(node) + ": " + - s(most) + " (least " + s(least) + ")").join("\n") + - (each.length > 3 ? "\n(" + (each.length - 3) + " more)" : ""); + bound + "\n\nThe slowest streams, and this node's (the most and the least each lagged," + + " last 30 s; of that, how long a batch waited in its node, from logging its first" + + " message to sending it, and took on the way, from sending to arriving):\n" + + listed.map((stream, i) => (each.indexOf(stream) + 1) + ". " + line(stream)).join("\n") + + (each.length > listed.length ? "\n(" + (each.length - listed.length) + " more)" : ""); } // Messages as `node`'s stream brought them (each named by both ends, `from` and `to`; // `arrived`: logged by its receiver, the sender's not read, so it went its link's @@ -3172,9 +3199,6 @@ function runMessages(list, node) { // On the timeline: when the message was logged, as `performance.now()` has it, now // and behind. const placed = list.map(message => ({ ...message, lag: wall - (message.at - ahead) })); - // Not those of a gap, which would hold the map that far behind for a while. - const fresh = placed.filter(message => message.lag <= REPLAY_MOST_MS); - if (fresh.length > 0) noteLag(node, Math.max(...fresh.map(message => message.lag))); if (!flowsNow()) return; const links = new Map(); for (const message of placed) { @@ -3214,9 +3238,10 @@ function runMessages(list, node) { // This node's messages, as its stream brings them (`messages` events): listed in the // console, and run along its links, those it sent, and those it received from a node // whose messages the page doesn't read (a subscribed node's run from it, as they left). -function ownMessagesArrived(list) { +function ownMessagesArrived(list, sent) { messagesArrived(list); if (!state) return; + noteStream(state.node.id, list, sent); const self = state.node.id; runMessages(list.filter(message => message.way === "sent" || (message.way === "received" && !logging(message.peer))) @@ -3229,8 +3254,9 @@ function ownMessagesArrived(list) { // to show, and what it received from another subscribed node that one's (as what it // sent): each message once. Each named by both ends (`from`, `to`) and the node that // logged it (`node`), its number and exchange made its own (`node:seq`). -function remoteMessagesArrived(id, entry, list) { +function remoteMessagesArrived(id, entry, list, sent) { if (!state) return; + noteStream(id, list, sent); const self = state.node.id; // Its messages come whole, those since the last event, up to the latest 500: one // missing between (a busy moment) leaves what was delivered unknown for now. @@ -3744,7 +3770,10 @@ function connect() { remakesArrived(); }); events.addEventListener("document", event => { pulse(); documentChanged(JSON.parse(event.data)); }); - events.addEventListener("messages", event => ownMessagesArrived(JSON.parse(event.data))); + events.addEventListener("messages", event => { + const batch = JSON.parse(event.data); + ownMessagesArrived(batch.messages, batch.sent); + }); // The node stops: it says so, and ends the stream. Back once it is. events.addEventListener("offline", () => { events.close(); live = false; diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs index 1d63f38..7a643d0 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -250,13 +250,21 @@ impl Hub { }; logged = last.seq; by_node(&self.state, &mut messages).await; - if let Ok(json) = serde_json::to_string(&messages) { - let _ = self.messages.send(Arc::new(json)); - } + let batch = json!({ "sent": now_millis(), "messages": messages }); + let _ = self.messages.send(Arc::new(batch.to_string())); } } } +/// Unix milliseconds by the system's clock, as the message log stamps messages +/// (`Logged::at`): when a batch of them goes, so the page tells the time they waited +/// in the node from the time they took on the way. +fn now_millis() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |since| since.as_millis() as u64) +} + /// A task stopped when this is dropped. struct AbortOnDrop(tokio::task::JoinHandle<()>); @@ -605,7 +613,8 @@ async fn own_messages_events(Extension(hub): Extension>) -> impl IntoRe Some(json) }) .collect(); - yield Ok(Event::default().event("messages").data(JsonValue::Array(messages).to_string())); + let batch = json!({ "sent": now_millis(), "messages": messages }); + yield Ok(Event::default().event("messages").data(batch.to_string())); } }; ( -- cgit v1.3.1