aboutsummaryrefslogtreecommitdiffstats
path: root/src/api/dashboard/mod.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/api/dashboard/mod.rs')
-rw-r--r--src/api/dashboard/mod.rs667
1 files changed, 667 insertions, 0 deletions
diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs
new file mode 100644
index 0000000..2557906
--- /dev/null
+++ b/src/api/dashboard/mod.rs
@@ -0,0 +1,667 @@
+//! The dashboard (`--features dashboard`): a page showing the node's figures
+//! (`/admin/stats`), its peers (`/peers`) and a map of the swarm, kept up to date
+//! through Server-Sent Events.
+//!
+//! One producer per node serves every client: while at least one is connected (or
+//! asked for the state within `LINGER`), it builds the snapshot every `TICK` and, if
+//! something changed, publishes it, and a JSON merge patch (RFC 7386) from the one
+//! before to every client connected. Nothing changed, nothing is sent (but SSE's
+//! keep-alive comments). Ages (`…SecsAgo`, `uptimeSecs` and the like, `GROWING`) grow by
+//! themselves: they alone changing is no change, and the page counts them on.
+//!
+//! The map: the graph the member list makes (`topology::intended_graph`, as this node
+//! sees it), and the links every member measures: subscribed to with LINKS while anyone
+//! looks (renewed every `LINKS_EVERY`, ended when nobody does; `swarm::links`), so a
+//! member tells of a change as it happens; the same answers for all clients. Or, if
+//! asked (`POST /admin/dashboard/links?live=false`, for every client, until asked
+//! otherwise or the node restarts), only asked every `LINKS_EVERY`, without
+//! subscribing: fewer messages in the swarm, a map up to `LINKS_EVERY` behind. With
+//! the clocks the links carry, every member's clock against the swarm's (`clocks`).
+
+mod clocks;
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::convert::Infallible;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, Mutex, OnceLock};
+use std::time::Duration;
+
+use axum::extract::Query;
+use axum::http::{HeaderValue, StatusCode, header::CONTENT_TYPE};
+use axum::response::sse::{Event, KeepAlive, Sse};
+use axum::response::{Html, IntoResponse, Response};
+use axum::routing::{get, post};
+use axum::{Extension, Router};
+use serde_json::{Map, Value as JsonValue, json};
+use tokio::sync::{Notify, broadcast, watch};
+use tokio::time::Instant;
+
+use crate::locks::MutexExt;
+use crate::state::AppState;
+use crate::swarm::{links, members, topology};
+use crate::transport::{Hex, Link};
+
+/// How often the snapshot is built and compared, while anyone looks.
+const TICK: Duration = Duration::from_secs(2);
+/// How often the subscriptions to the members' links are renewed (each renewal is
+/// answered: a reply at least this often), and the graph recomputed; changes come in
+/// between, as they happen (`swarm::links`).
+const LINKS_EVERY: Duration = Duration::from_secs(30);
+/// How long a subscription lasts unless renewed: a node that is gone stops it.
+const LINKS_LEASE: Duration = Duration::from_secs(90);
+/// A member's links older than this are left out (it didn't answer since).
+const LINKS_KEPT_FOR: Duration = Duration::from_secs(90);
+/// The producer runs on this long after the last request for the state.
+const LINGER: Duration = Duration::from_secs(60);
+/// Members asked for their links at most, each round.
+const MAX_ASKED: usize = 512;
+
+/// A snapshot or a patch as published: its number, and its JSON.
+struct Published {
+ seq: u64,
+ json: String,
+}
+
+/// The producer and what it published, shared by all clients.
+struct Hub {
+ state: AppState,
+ /// The latest snapshot; None while the producer hasn't made one (since it started).
+ 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>>,
+ /// The messages exchanged since the last tick, one by one (`traffic::MessageLog`),
+ /// to the clients connected: a JSON list.
+ messages: broadcast::Sender<Arc<String>>,
+ control: Mutex<Control>,
+ /// Someone asked for it now (`refresh`): a tick at once, members asked again.
+ poke: Notify,
+ /// The members' links subscribed to (true, at first), or only asked for every
+ /// `LINKS_EVERY` (`set_links`).
+ live: AtomicBool,
+ /// 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).
+ closing: watch::Receiver<bool>,
+}
+
+struct Control {
+ running: bool,
+ /// The producer stops once no client is connected and this is past.
+ wanted_until: Instant,
+}
+
+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);
+ let (messages, _) = broadcast::channel(64);
+ Arc::new(Self {
+ state,
+ latest,
+ patches,
+ documents,
+ messages,
+ poke: Notify::new(),
+ live: AtomicBool::new(true),
+ control: Mutex::new(Control {
+ running: false,
+ wanted_until: Instant::now(),
+ }),
+ closing,
+ })
+ }
+
+ /// Someone looks: the producer runs (started if it wasn't), for `LINGER` at least.
+ fn want(self: &Arc<Self>) {
+ let mut control = self.control.lock_or_recover();
+ control.wanted_until = Instant::now() + LINGER;
+ if !control.running && !*self.closing.borrow() {
+ control.running = true;
+ // What it published before it stopped is stale.
+ self.latest.send_replace(None);
+ tokio::spawn(Arc::clone(self).run());
+ }
+ }
+
+ /// The latest snapshot, once there is one; None if the node stops first.
+ async fn current(&self) -> Option<Arc<Published>> {
+ let mut latest = self.latest.subscribe();
+ let mut closing = self.closing.clone();
+ // The borrows (lock guards, not Send) end within the arms.
+ tokio::select! {
+ published = latest.wait_for(Option::is_some) => {
+ published.ok().and_then(|current| current.clone())
+ }
+ _ = closing.wait_for(|closing| *closing) => None,
+ }
+ }
+
+ /// Resolves once the node stops (or its signal is gone).
+ async fn closed(&self) {
+ let mut closing = self.closing.clone();
+ let _ = closing.wait_for(|closing| *closing).await;
+ }
+
+ async fn run(self: Arc<Self>) {
+ let mut last: Option<JsonValue> = None;
+ let mut seq = 0u64;
+ let mut graph = Graph::default();
+ let mut graph_due = Instant::now();
+ let mut changes = self.state.change_events.subscribe();
+ // The messages one by one, from now on, while it runs.
+ let log = &self.state.traffic.log;
+ log.watch(true);
+ let mut logged = 0u64;
+ loop {
+ let live = self.live.load(Ordering::Relaxed);
+ if Instant::now() >= graph_due {
+ graph.refresh(&self.state, live).await;
+ graph_due = Instant::now() + LINKS_EVERY;
+ }
+ let snapshot = build(&self.state, &graph, live).await;
+ let changed = last
+ .as_ref()
+ .is_none_or(|old| without_ages(old) != without_ages(&snapshot));
+ if changed {
+ seq += 1;
+ if let Some(old) = &last
+ && let Some(patch) = merge_patch(old, &snapshot)
+ {
+ let _ = self.patches.send(Arc::new(Published {
+ seq,
+ json: patch.to_string(),
+ }));
+ }
+ self.latest.send_replace(Some(Arc::new(Published {
+ seq,
+ json: snapshot.to_string(),
+ })));
+ last = Some(snapshot);
+ }
+ let poked = tokio::select! {
+ () = tokio::time::sleep(TICK) => false,
+ () = self.poke.notified() => true,
+ };
+ if poked {
+ graph_due = Instant::now();
+ }
+ if let Some(changed) = document_changes(&mut changes) {
+ let _ = self.documents.send(Arc::new(changed.to_string()));
+ }
+ let mut messages = log.after(logged, MESSAGES_AT_MOST);
+ by_node(&self.state, &mut messages).await;
+ if let Some(last) = messages.last() {
+ logged = last.seq;
+ if let Ok(json) = serde_json::to_string(&messages) {
+ let _ = self.messages.send(Arc::new(json));
+ }
+ }
+ let unwanted = || {
+ self.patches.receiver_count() == 0
+ && Instant::now() >= self.control.lock_or_recover().wanted_until
+ };
+ if *self.closing.borrow() {
+ log.watch(false);
+ self.control.lock_or_recover().running = false;
+ return;
+ }
+ if unwanted() {
+ // Nobody looks: the members needn't tell us any more. Unless someone
+ // came meanwhile, then it goes on (and subscribes again).
+ for member in self.state.links.subscribed() {
+ links::unsubscribe(&self.state, &member).await;
+ }
+ let mut control = self.control.lock_or_recover();
+ if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until {
+ log.watch(false);
+ control.running = false;
+ return;
+ }
+ drop(control);
+ graph_due = Instant::now();
+ }
+ }
+ }
+}
+
+/// Paths changed in a tick listed at most; more: `all`.
+/// The messages the paths layer logged under an address (its own PINGs and PONGs) under
+/// the node there instead, where one is known: the address it proved itself at, or is
+/// said to be at (`Transport::identities_of`), named in `address`.
+async fn by_node(state: &AppState, messages: &mut [crate::transport::traffic::Logged]) {
+ let peers = state.peers_snapshot().await;
+ let known = |peer: &str| peers.iter().any(|(known, _)| known == peer);
+ if messages.iter().all(|message| known(&message.peer)) {
+ return;
+ }
+ let mut nodes: std::collections::HashMap<String, String> = std::collections::HashMap::new();
+ for (peer, _) in &peers {
+ for address in state.transport.identities_of(peer) {
+ nodes.entry(address).or_insert_with(|| peer.clone());
+ }
+ }
+ for message in messages {
+ if let Some(node) = nodes.get(&message.peer) {
+ message.address = Some(std::mem::replace(&mut message.peer, node.clone()));
+ }
+ }
+}
+
+const MAX_CHANGED: usize = 256;
+/// Messages sent to the clients per tick at most, the latest (the console keeps fewer).
+const MESSAGES_AT_MOST: usize = 500;
+
+/// Where the document changed since last asked (`Hub::documents`), if anywhere.
+fn document_changes(
+ changes: &mut broadcast::Receiver<Arc<crate::document::store::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>),
+ Messages(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: subscribed to, if `live`.
+#[derive(Default)]
+struct Graph {
+ intended: BTreeMap<String, BTreeSet<String>>,
+}
+
+impl Graph {
+ async fn refresh(&mut self, state: &AppState, live: bool) {
+ let lease = if live { LINKS_LEASE } else { Duration::ZERO };
+ self.intended = topology::intended_graph(state).unwrap_or_default();
+ let members: Vec<String> = {
+ let replica = state.doc.read().await;
+ members::members(&replica.doc)
+ };
+ for member in members
+ .into_iter()
+ .filter(|member| member != state.local_id())
+ .take(MAX_ASKED)
+ {
+ let state = state.clone();
+ tokio::spawn(async move { links::ask(&state, &member, lease).await });
+ }
+ }
+}
+
+/// The snapshot: the node, its figures, its peers, and the map (`live`: its links
+/// subscribed to).
+async fn build(state: &AppState, graph: &Graph, live: bool) -> JsonValue {
+ // The peers once, for their list and the figures' totals over them.
+ let list = super::peers::peer_list(state).await;
+ let stats = serde_json::to_value(super::stats::collect(state, &list).await).unwrap_or_default();
+ let peers = serde_json::to_value(&list).unwrap_or_default();
+ let aliases = {
+ let replica = state.doc.read().await;
+ members::aliases(&replica.doc)
+ };
+ let local = state.local_id();
+ let intended: Map<String, JsonValue> = graph
+ .intended
+ .iter()
+ .map(|(member, neighbours)| (member.clone(), json!(neighbours)))
+ .collect();
+ let own = links::own_links(state).await;
+ let replies: Vec<_> = state
+ .links
+ .replies()
+ .into_iter()
+ .filter(|(_, age, _)| *age <= LINKS_KEPT_FOR)
+ .collect();
+ let mut measured = Map::new();
+ measured.insert(
+ local.to_owned(),
+ json!({ "secsAgo": 0, "links": links_json(&own) }),
+ );
+ for (member, age, member_links) in &replies {
+ measured.insert(
+ member.clone(),
+ json!({ "secsAgo": age.as_secs(), "links": links_json(member_links) }),
+ );
+ }
+ let clocks = clocks::swarm(
+ local,
+ std::iter::once((local, &own[..])).chain(
+ replies
+ .iter()
+ .map(|(member, _, member_links)| (member.as_str(), &member_links[..])),
+ ),
+ u64::try_from(state.clock_skew_warn.as_millis()).unwrap_or(u64::MAX),
+ );
+ // Everyone on the map: in the member list's graph, or measured.
+ let mut members_json = Map::new();
+ for member in graph.intended.keys().chain(measured.keys()) {
+ members_json.insert(
+ member.clone(),
+ json!({ "alias": aliases.get(member), "self": member == local }),
+ );
+ }
+ json!({
+ "node": { "id": local, "alias": state.alias },
+ "page": page().0,
+ "stats": stats,
+ "peers": peers,
+ "graph": {
+ "members": members_json,
+ "intended": intended,
+ "measured": measured,
+ "clocks": clocks,
+ "live": live,
+ },
+ })
+}
+
+fn links_json(links: &[Link]) -> JsonValue {
+ links
+ .iter()
+ .map(|link| {
+ json!({
+ "peer": link.peer,
+ "transport": link.transport,
+ "relayed": link.relayed,
+ "down": link.down,
+ "rttMs": link.rtt_us.map(|us| f64::from(us) / 1000.0),
+ "dormant": link.dormant,
+ "clock": link.clock,
+ "others": link.others.iter().map(|other| json!({
+ "transport": other.transport,
+ "rttMs": other.rtt_us.map(|us| f64::from(us) / 1000.0),
+ "relayed": other.relayed,
+ })).collect::<Vec<_>>(),
+ })
+ })
+ .collect()
+}
+
+/// Fields that grow by themselves besides those named `…SecsAgo`; the page gets the
+/// list when served (`GROWING_MARK`).
+const GROWING: &[&str] = &[
+ "uptimeSecs",
+ "dormantForSecs",
+ "divergedForSecs",
+ "observedSecs",
+ "lastFromNeighbourSecs",
+ "oldestQuarantineSecs",
+];
+
+/// Whether the field `key` is an age, which grows by itself.
+fn is_age(key: &str) -> bool {
+ key == "secsAgo" || key.ends_with("SecsAgo") || GROWING.contains(&key)
+}
+
+/// `value` without its ages, to tell whether anything else changed.
+fn without_ages(value: &JsonValue) -> JsonValue {
+ match value {
+ JsonValue::Object(object) => JsonValue::Object(
+ object
+ .iter()
+ .filter(|(key, _)| !is_age(key))
+ .map(|(key, value)| (key.clone(), without_ages(value)))
+ .collect(),
+ ),
+ JsonValue::Array(items) => JsonValue::Array(items.iter().map(without_ages).collect()),
+ other => other.clone(),
+ }
+}
+
+/// The JSON merge patch (RFC 7386) from `old` to `new`, if they differ: objects field
+/// by field (a field gone is null), anything else replaced whole. A null in `new` reads
+/// as the field gone, as a merge patch can't tell them apart.
+fn merge_patch(old: &JsonValue, new: &JsonValue) -> Option<JsonValue> {
+ match (old, new) {
+ (JsonValue::Object(old), JsonValue::Object(new)) => {
+ let mut patch = Map::new();
+ for key in old.keys() {
+ if !new.contains_key(key) {
+ patch.insert(key.clone(), JsonValue::Null);
+ }
+ }
+ for (key, value) in new {
+ match old.get(key) {
+ Some(previous) => {
+ if let Some(change) = merge_patch(previous, value) {
+ patch.insert(key.clone(), change);
+ }
+ }
+ None => {
+ patch.insert(key.clone(), value.clone());
+ }
+ }
+ }
+ (!patch.is_empty()).then_some(JsonValue::Object(patch))
+ }
+ (old, new) if old == new => None,
+ (_, new) => Some(new.clone()),
+ }
+}
+
+/// The dashboard's routes (authenticated, as the rest of `/admin`), with their hub.
+pub(crate) fn router(state: AppState) -> Router<AppState> {
+ // Its streams end when the node shuts down (`AppState::closing`).
+ let closing = state.closing.subscribe();
+ Router::new()
+ .route("/admin/dashboard", get(page_html))
+ .route("/admin/dashboard/state", get(current_state))
+ .route("/admin/dashboard/events", get(events))
+ .route("/admin/dashboard/refresh", post(refresh))
+ .route("/admin/dashboard/links", post(set_links))
+ .layer(Extension(Hub::new(state, closing)))
+}
+
+/// The page as built into this binary.
+const PAGE: &str = include_str!("dashboard.html");
+/// Where the page says which version of it it is.
+const PAGE_VERSION_MARK: &str = "__PAGE_VERSION__";
+/// Where the page gets the age fields (`GROWING`), so the list is kept once.
+const GROWING_MARK: &str = "__GROWING__";
+
+/// The page's version (a hash of it), and the page saying so: a page whose version
+/// differs from the snapshot's `page` was served by another build, and reloads.
+fn page() -> &'static (String, String) {
+ static PAGE_SERVED: OnceLock<(String, String)> = OnceLock::new();
+ PAGE_SERVED.get_or_init(|| {
+ use sha2::{Digest, Sha256};
+ let growing = serde_json::to_string(GROWING).unwrap_or_else(|_| "[]".to_owned());
+ let page = PAGE.replace(GROWING_MARK, &growing);
+ // Of the page with its list: a new list is a new version too.
+ let version = Hex(&Sha256::digest(page.as_bytes())[..8]).to_string();
+ let served = page.replace(PAGE_VERSION_MARK, &version);
+ (version, served)
+ })
+}
+
+/// `GET /admin/dashboard`: the page.
+async fn page_html() -> Html<&'static str> {
+ Html(page().1.as_str())
+}
+
+/// `POST /admin/dashboard/refresh`: a tick now rather than at the next `TICK`, and the
+/// members asked for their links again; what changed comes as events, as ever.
+async fn refresh(Extension(hub): Extension<Arc<Hub>>) -> StatusCode {
+ hub.want();
+ hub.poke.notify_one();
+ StatusCode::NO_CONTENT
+}
+
+#[derive(serde::Deserialize)]
+struct LinksQuery {
+ live: bool,
+}
+
+/// `POST /admin/dashboard/links?live=…`: the members' links subscribed to (`true`), or
+/// only asked for every `LINKS_EVERY` (`false`), for every client; the subscriptions
+/// ended at once, and the members asked again now.
+async fn set_links(
+ Extension(hub): Extension<Arc<Hub>>,
+ Query(query): Query<LinksQuery>,
+) -> StatusCode {
+ hub.live.store(query.live, Ordering::Relaxed);
+ if !query.live {
+ for member in hub.state.links.subscribed() {
+ links::unsubscribe(&hub.state, &member).await;
+ }
+ }
+ hub.want();
+ hub.poke.notify_one();
+ StatusCode::NO_CONTENT
+}
+
+/// `GET /admin/dashboard/state`: the latest snapshot, for a client that wants it whole.
+async fn current_state(Extension(hub): Extension<Arc<Hub>>) -> Response {
+ hub.want();
+ let Some(current) = hub.current().await else {
+ return StatusCode::SERVICE_UNAVAILABLE.into_response();
+ };
+ (
+ [(CONTENT_TYPE, HeaderValue::from_static("application/json"))],
+ current.json.clone(),
+ )
+ .into_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), a `document` where the document
+/// changed (`Hub::documents`, at most every tick), `messages` with the messages since
+/// the last tick (`Hub::messages`), 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();
+ let mut messages = hub.messages.subscribe();
+ hub.want();
+ let first = hub.current().await;
+ let stream = async_stream::stream! {
+ let Some(first) = first else {
+ yield Ok::<_, Infallible>(offline_event());
+ return;
+ };
+ let mut seen = first.seq;
+ yield Ok(snapshot_event(&first));
+ loop {
+ let next = tokio::select! {
+ received = patches.recv() => Next::Patch(received),
+ changed = documents.recv() => Next::Document(changed),
+ logged = messages.recv() => Next::Messages(logged),
+ () = hub.closed() => Next::Closing,
+ };
+ 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::Messages(Ok(logged)) => {
+ yield Ok(Event::default().event("messages").data(logged.as_str()));
+ continue;
+ }
+ // Missed ones are only missed in the console.
+ Next::Messages(Err(broadcast::error::RecvError::Lagged(_))) => continue,
+ Next::Messages(Err(broadcast::error::RecvError::Closed)) => break,
+ Next::Closing => {
+ yield Ok(offline_event());
+ break;
+ }
+ };
+ match received {
+ Ok(patch) if patch.seq <= seen => {}
+ Ok(patch) => {
+ seen = patch.seq;
+ yield Ok(Event::default()
+ .event("patch")
+ .id(patch.seq.to_string())
+ .data(patch.json.clone()));
+ }
+ Err(broadcast::error::RecvError::Lagged(_)) => {
+ let Some(current) = hub.current().await else {
+ yield Ok(offline_event());
+ break;
+ };
+ seen = current.seq;
+ yield Ok(snapshot_event(&current));
+ }
+ Err(broadcast::error::RecvError::Closed) => break,
+ }
+ }
+ };
+ Sse::new(stream).keep_alive(KeepAlive::default())
+}
+
+/// The node stops: the client may reconnect once it is back.
+fn offline_event() -> Event {
+ Event::default().event("offline").data("{}")
+}
+
+fn snapshot_event(published: &Published) -> Event {
+ Event::default()
+ .event("snapshot")
+ .id(published.seq.to_string())
+ .data(published.json.clone())
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn merge_patches() {
+ let old = json!({ "a": 1, "b": { "c": 2, "d": [1, 2] }, "gone": true });
+ let new = json!({ "a": 1, "b": { "c": 3, "d": [1, 2] }, "added": "x" });
+ assert_eq!(
+ merge_patch(&old, &new),
+ Some(json!({ "b": { "c": 3 }, "gone": null, "added": "x" }))
+ );
+ assert_eq!(merge_patch(&old, &old), None);
+ assert_eq!(
+ merge_patch(&json!([1]), &json!([1, 2])),
+ Some(json!([1, 2]))
+ );
+ }
+
+ #[test]
+ fn ages_alone_are_no_change() {
+ let old =
+ json!({ "uptimeSecs": 1, "peers": { "p": { "lastSeenSecsAgo": 3, "rttMs": 2 } } });
+ let older =
+ json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 2 } } });
+ assert_eq!(without_ages(&old), without_ages(&older));
+ let moved =
+ json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 5 } } });
+ assert_ne!(without_ages(&old), without_ages(&moved));
+ }
+}