diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/dashboard/dashboard.html | 91 | ||||
| -rw-r--r-- | src/api/dashboard/mod.rs | 75 |
2 files changed, 132 insertions, 34 deletions
diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html index 53a8282..0b15450 100644 --- a/src/api/dashboard/dashboard.html +++ b/src/api/dashboard/dashboard.html @@ -79,9 +79,6 @@ border: 2px solid var(--self); opacity: .5; } .scroll { overflow: auto; min-height: 0; } #figures-box .scroll, #document-box .scroll { flex: 1; } - h2 button { float: right; font: inherit; text-transform: none; letter-spacing: 0; - color: var(--muted); background: none; border: 1px solid var(--line); border-radius: 4px; - padding: 0 6px; cursor: pointer; } .tree pre { margin: 2px 0 4px 1em; white-space: pre-wrap; word-break: break-all; max-height: 20em; overflow: auto; } .note { color: var(--muted); font-style: italic; } @@ -128,7 +125,7 @@ <div class="scroll"><div id="stats" class="tree"></div></div> </section> <section id="document-box"> - <h2>Document <button id="document-reload" title="Read again what is open">reload</button></h2> + <h2>Document</h2> <div class="scroll"><div id="document" class="tree"></div></div> </section> </main> @@ -538,16 +535,21 @@ function renderGraph() { root.replaceChildren(...children); } -// --- Document: read a level at a time (`/store?shallow=true`) as it is opened --- -// What is open, by path (as JSON), kept across reloads of the tree: the first level -// to begin with. +// --- Document: read a level at a time (`/store?shallow=true`) as it is opened, and +// read again where the node says it changed (`document` events) --- +// What is open, by path (as JSON): the first level to begin with. const documentOpen = new Set([JSON.stringify([])]); +// The open branches read, by path (as JSON): to read again on a change. +const documentBranches = new Map(); function storeUrl(path, shallow) { const where = path.length ? "../store/" + path.map(encodeURIComponent).join("/") : "../store"; return where + (shallow ? "?shallow=true" : ""); } // Store paths split at "/": a key with one (or an empty key) can't be read on its own. const addressable = path => path.every(segment => segment !== "" && !segment.includes("/")); +// Either is within the other. +const related = (a, b) => a.length <= b.length + ? a.every((segment, i) => segment === b[i]) : related(b, a); function describe(summary) { switch (summary.type) { case "map": return "{" + summary.size + (summary.size === 1 ? " key}" : " keys}"); @@ -561,51 +563,85 @@ async function readStore(path, shallow) { if (!response.ok) throw new Error("HTTP " + response.status); return response.json(); } -// One entry: a short scalar as it is; anything else a branch read when opened. +// One entry: a short scalar as it is; anything else a branch read when opened. A +// branch's entries are kept across reads, so what's open inside stays open. function documentEntry(key, summary, path) { if ("value" in summary) { return el("div", {}, el("span", { class: "key" }, key + ": "), JSON.stringify(summary.value)); } const details = el("details", {}); - details.append(el("summary", {}, el("span", { class: "key" }, key), " ", describe(summary))); + const described = document.createTextNode(describe(summary)); + details.append(el("summary", {}, el("span", { class: "key" }, key), " ", described)); + // A newer summary from the parent: a branch of the same kind stays as it is. + details.updated = newer => { + summary = newer; + described.textContent = describe(newer); + }; + details.sameKind = newer => !("value" in newer) && newer.type === summary.type; if (!addressable(path)) { details.append(el("div", { class: "note" }, "Its path can't be read on its own (a key with \"/\").")); return details; } const id = JSON.stringify(path); + const children = new Map(); let read = false; - const load = async () => { - if (read) return; - read = true; + const fill = async () => { try { if (summary.type === "map") { const shallow = await readStore(path, true); - // As it is now (the root's is unknown until read). - details.querySelector("summary").lastChild.textContent = describe(shallow); - const entries = Object.entries(shallow.entries || {}); - if (!entries.length) details.append(el("div", { class: "note" }, "empty")); - for (const [k, child] of entries) details.append(documentEntry(k, child, [...path, k])); + details.updated(shallow); + const entries = shallow.entries || {}; + for (const [k, child] of children) { + if (!(k in entries)) { child.element.remove(); children.delete(k); } + } + for (const [k, newer] of Object.entries(entries)) { + const known = children.get(k); + if (known && JSON.stringify(known.summary) === JSON.stringify(newer)) continue; + if (known && known.element.sameKind && known.element.sameKind(newer)) { + known.element.updated(newer); + known.summary = newer; + continue; + } + const element = documentEntry(k, newer, [...path, k]); + if (known) known.element.replaceWith(element); + children.set(k, { element, summary: newer }); + } + // In the node's order (sorted); moving keeps their state. + details.querySelectorAll(":scope > .note").forEach(note => note.remove()); + if (!children.size) details.append(el("div", { class: "note" }, "empty")); + for (const k of Object.keys(entries)) details.append(children.get(k).element); } else { const value = await readStore(path, false); - details.append(el("pre", {}, - typeof value === "string" ? value : JSON.stringify(value, null, 2))); + const shown = el("pre", {}, typeof value === "string" ? value : JSON.stringify(value, null, 2)); + const before = details.querySelector(":scope > pre"); + if (before) before.replaceWith(shown); else details.append(shown); } } catch (error) { + details.querySelectorAll(":scope > .note").forEach(note => note.remove()); details.append(el("div", { class: "note" }, "Not read: " + error.message)); } }; + const open = () => { + documentOpen.add(id); + documentBranches.set(id, { path, element: details, refresh: fill }); + if (!read) { read = true; fill(); } + }; details.addEventListener("toggle", () => { - if (details.open) { documentOpen.add(id); load(); } else documentOpen.delete(id); + if (details.open) open(); + else { documentOpen.delete(id); documentBranches.delete(id); } }); - if (documentOpen.has(id)) { details.open = true; load(); } + if (documentOpen.has(id)) { details.open = true; open(); } return details; } -function renderDocument() { - const root = documentEntry("store", { type: "map", size: "…" }, []); - document.getElementById("document").replaceChildren(root); +// The node says what changed (paths), or that anything may have (`all`). +function documentChanged(change) { + for (const [id, branch] of documentBranches) { + if (!branch.element.isConnected) { documentBranches.delete(id); continue; } + if (change.all || change.paths.some(path => related(path, branch.path))) branch.refresh(); + } } -document.getElementById("document-reload").addEventListener("click", renderDocument); -renderDocument(); +document.getElementById("document").replaceChildren( + documentEntry("store", { type: "map", size: "…" }, [])); function render() { if (!state) return; @@ -650,6 +686,8 @@ function connect() { location.reload(); return; } + // Changes while not connected went unseen: what is open is read again. + if (state !== null) documentChanged({ all: true }); state = snapshot; stateAt = Date.now(); live = true; status("live", true); pulse(); render(); }); @@ -657,6 +695,7 @@ function connect() { if (!state) return; state = mergePatch(state, JSON.parse(event.data)); stateAt = Date.now(); pulse(); render(); }); + events.addEventListener("document", event => { pulse(); documentChanged(JSON.parse(event.data)); }); // 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 99a0b83..7679787 100644 --- a/src/api/dashboard/mod.rs +++ b/src/api/dashboard/mod.rs @@ -56,6 +56,9 @@ struct Hub { latest: watch::Sender<Option<Arc<Published>>>, /// Each patch, to the clients connected. patches: broadcast::Sender<Arc<Published>>, + /// Where the document changed since the last tick, to the clients connected: + /// `{"paths": [...]}`, or `{"all": true}` if too many or not known. + documents: broadcast::Sender<Arc<String>>, control: Mutex<Control>, /// True once the node stops: every client gets an `offline` event, and its stream /// ends (so the HTTP server needn't wait for clients to go). @@ -72,10 +75,12 @@ impl Hub { fn new(state: AppState, closing: watch::Receiver<bool>) -> Arc<Self> { let (latest, _) = watch::channel(None); let (patches, _) = broadcast::channel(64); + let (documents, _) = broadcast::channel(64); Arc::new(Self { state, latest, patches, + documents, control: Mutex::new(Control { running: false, wanted_until: Instant::now(), @@ -120,6 +125,7 @@ impl Hub { let mut seq = 0u64; let mut graph = Graph::default(); let mut graph_due = Instant::now(); + let mut changes = self.state.change_events.subscribe(); loop { if Instant::now() >= graph_due { graph.refresh(&self.state).await; @@ -146,6 +152,9 @@ impl Hub { last = Some(snapshot); } tokio::time::sleep(TICK).await; + if let Some(changed) = document_changes(&mut changes) { + let _ = self.documents.send(Arc::new(changed.to_string())); + } { let mut control = self.control.lock_or_recover(); let unwanted = @@ -159,6 +168,41 @@ impl Hub { } } +/// Paths changed in a tick listed at most; more: `all`. +const MAX_CHANGED: usize = 256; + +/// Where the document changed since last asked (`Hub::documents`), if anywhere. +fn document_changes( + changes: &mut broadcast::Receiver<Arc<crate::swarm::sync::ChangeSet>>, +) -> Option<JsonValue> { + let mut paths = BTreeSet::new(); + let mut all = false; + loop { + match changes.try_recv() { + Ok(set) => { + paths.extend(set.paths.iter().cloned()); + all |= paths.len() > MAX_CHANGED; + } + Err(broadcast::error::TryRecvError::Lagged(_)) => all = true, + Err(_) => break, + } + } + if all { + Some(json!({ "all": true })) + } else if paths.is_empty() { + None + } else { + Some(json!({ "paths": paths })) + } +} + +/// What a client's stream waits for. +enum Next { + Patch(Result<Arc<Published>, broadcast::error::RecvError>), + Document(Result<Arc<String>, broadcast::error::RecvError>), + Closing, +} + /// The intended graph, recomputed every `LINKS_EVERY` (it costs about `members²`), when /// members are asked for their links too. #[derive(Default)] @@ -360,11 +404,13 @@ async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response { /// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a /// `patch` whenever something changed (each with its number as the event's ID; a -/// client that fell behind gets a `snapshot` again), and `offline` when the node -/// stops, which ends the stream. +/// client that fell behind gets a `snapshot` again), a `document` where the document +/// changed (`Hub::documents`, at most every tick), and `offline` when the node stops, +/// which ends the stream. async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse { // Subscribed before the snapshot is read: no patch after it is missed. let mut patches = hub.patches.subscribe(); + let mut documents = hub.documents.subscribe(); hub.want(); let first = hub.current().await; let stream = async_stream::stream! { @@ -375,13 +421,26 @@ async fn events(Extension(hub): Extension<Arc<Hub>>) -> impl IntoResponse { let mut seen = first.seq; yield Ok(snapshot_event(&first)); loop { - let received = tokio::select! { - received = patches.recv() => Some(received), - () = hub.closed() => None, + let next = tokio::select! { + received = patches.recv() => Next::Patch(received), + changed = documents.recv() => Next::Document(changed), + () = hub.closed() => Next::Closing, }; - let Some(received) = received else { - yield Ok(offline_event()); - break; + let received = match next { + Next::Patch(received) => received, + Next::Document(Ok(changed)) => { + yield Ok(Event::default().event("document").data(changed.as_str())); + continue; + } + Next::Document(Err(broadcast::error::RecvError::Lagged(_))) => { + yield Ok(Event::default().event("document").data(r#"{"all":true}"#)); + continue; + } + Next::Document(Err(broadcast::error::RecvError::Closed)) => break, + Next::Closing => { + yield Ok(offline_event()); + break; + } }; match received { Ok(patch) if patch.seq <= seen => {} |