diff --git a/examples/ariadne/documents/conference_roster.ari b/examples/ariadne/documents/conference_roster.ari index eea9a1855..57a367dc1 100644 --- a/examples/ariadne/documents/conference_roster.ari +++ b/examples/ariadne/documents/conference_roster.ari @@ -87,30 +87,58 @@ windows: bind: (if (= (state-get "mic_on") "1") "● on" "muted") # ── Users: the roster with per-peer local controls ── + # One compact ROW per participant, as a table of unit widgets: every row is its own grid with the + # SAME fixed column widths, so the rows line up as columns. ping / down / up are each person's + # link to the relay AS THEY MEASURE IT (published to the room once a second; "--" = not measured + # yet). cam / mic / vol are LOCAL controls (what YOU see and hear of them), never sent. - window: Users id: users pos: [390, 40] - size: [380, 540] + size: [420, 260] closable: true children: - text: "in the room" bind: self.count - separator - # You (first row): your name lights up GREEN while YOU talk. - - text: "you" - bind: conf.self - visible_when: (not (= (state-get "self.talking") "1")) - - text: "you" - bind: conf.self - color: [0.30, 1.0, 0.40] - visible_when: (= (state-get "self.talking") "1") - - separator - # Everyone else: one row per peer (peer.). The on_tick fills peer..{pid,name,cam,talking}. + - type: group + id: users_head + layout: { kind: grid, col_widths: [118, 64, 50, 50, 34, 34, 0] } + children: + - text: "name" + - text: "ping" + - text: "down" + - text: "up" + - text: "cam" + - text: "mic" + - text: "vol" + # You: your name lights up GREEN while YOU talk (text colour is static, so it is a visibility + # swap: only one of the two is visible, so they share one cell). + - type: group + id: users_you + layout: { kind: grid, col_widths: [118, 64, 50, 50, 34, 34, 0] } + children: + - text: "" + bind: conf.self + visible_when: (not (= (state-get "self.talking") "1")) + - text: "" + bind: conf.self + color: [0.30, 1.0, 0.40] + visible_when: (= (state-get "self.talking") "1") + - text: "" + bind: self.ping + - text: "" + bind: self.down + - text: "" + bind: self.up + - text: "" + - text: "" + - text: "(you)" + # Everyone else: one row per peer (peer.), filled by on_tick. - type: group repeat: (state-get "peer.count") + layout: { kind: grid, col_widths: [118, 64, 50, 50, 34, 34, 0] } children: - # Name — click to open + raise this peer's camera window (cam_). Two texts: the GREEN one - # shows only while they are talking (text colour is static, so "light up" is a visibility swap). + # Name: click to open + raise this peer's camera window (cam_); GREEN while talking. - text: "" bind: peer.{i}.name on_click: (begin (state-set "tree.cam_{i}.open" "1") (state-set "tree.cam_{i}.raise" "1")) @@ -120,24 +148,30 @@ windows: color: [0.30, 1.0, 0.40] on_click: (begin (state-set "tree.cam_{i}.open" "1") (state-set "tree.cam_{i}.raise" "1")) visible_when: (= (state-get "peer.{i}.talking") "1") + - text: "" + bind: peer.{i}.ping + - text: "" + bind: peer.{i}.down + - text: "" + bind: peer.{i}.up # Local mute toggles (checked = on). Default on: the host treats von/aon as muted only when - # explicitly "0", so an untouched peer is shown + heard. + # explicitly "0", so an untouched peer is shown + heard. "##" hides the label (the header + # names the column). - type: checkbox - label: "cam" + label: "##cam" bind: peer.{i}.von def: true - type: checkbox - label: "mic" + label: "##mic" bind: peer.{i}.aon def: true # Local volume in our mix (1 = unity, up to 2x boost). - type: slider_float - label: "vol" + label: "##vol" bind: peer.{i}.vol lo: 0 hi: 2 def: 1 - - separator # ── One camera window per OTHER participant, titled by their name ── - window: Peer @@ -163,16 +197,16 @@ on_tick: | ; every cam_ declares the SAME authored pos:, so without this the per-peer windows land ; exactly on top of one another and look like a single window — the #1 reason a user reports ; "I can't see the other cameras". Tile them in a 2-column grid to the RIGHT of the Users - ; window (which sits at x≈390–610): the roster is NEVER covered, and up to four cameras show + ; window (which sits at x≈390–800): the roster is NEVER covered, and up to four cameras show ; side-by-side without overlapping on a laptop screen. Seeded ONCE here, before any peer window ; exists, so each window's first-use geometry seed lands at its own spot; the user's drag then ; persists and wins. Covers 6 peers; a 7th+ falls back to the authored pos. - (state-set "tree.cam_0.geometry" "612,66,330,290") - (state-set "tree.cam_1.geometry" "952,66,330,290") - (state-set "tree.cam_2.geometry" "612,364,330,290") - (state-set "tree.cam_3.geometry" "952,364,330,290") - (state-set "tree.cam_4.geometry" "612,662,330,290") - (state-set "tree.cam_5.geometry" "952,662,330,290") + (state-set "tree.cam_0.geometry" "806,66,330,290") + (state-set "tree.cam_1.geometry" "1146,66,330,290") + (state-set "tree.cam_2.geometry" "806,364,330,290") + (state-set "tree.cam_3.geometry" "1146,364,330,290") + (state-set "tree.cam_4.geometry" "806,662,330,290") + (state-set "tree.cam_5.geometry" "1146,662,330,290") (state-set "cam_on" "1") ; on/off toggles start ON (streams open unmuted) (state-set "mic_on" "1") (room-join "" (state-get "conf.room") (dict "name" (state-get "conf.self"))) @@ -251,6 +285,10 @@ on_tick: | (state-get (str-concat "media." pid ".camera"))) (state-set (str-concat "peer." idx ".talking") (state-get (str-concat "media." pid ".talking"))) + ; their link quality, as THEY measure it (ping to the relay, bits/s down + up) + (state-set (str-concat "peer." idx ".ping") (get-attr p "ping")) + (state-set (str-concat "peer." idx ".down") (get-attr p "down")) + (state-set (str-concat "peer." idx ".up") (get-attr p "up")) (set idx (+ idx 1)) (set total (+ total 1)))))) (state-set "peer.count" (str idx)) diff --git a/inc/cvc/ariadne/conference/room_net.h b/inc/cvc/ariadne/conference/room_net.h new file mode 100644 index 000000000..7d280853b --- /dev/null +++ b/inc/cvc/ariadne/conference/room_net.h @@ -0,0 +1,80 @@ +/* + Copyright 2026 The University of Texas at Austin + + This file is part of libcvc. + + libcvc is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License version 2.1 as published by the Free Software Foundation. + + libcvc is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. +*/ + +#ifndef __CVC_ARIADNE_CONFERENCE_ROOM_NET_H__ +#define __CVC_ARIADNE_CONFERENCE_ROOM_NET_H__ + +#include + +namespace cvc { +class app; + +namespace ariadne { +namespace conference { + +// ---------------- +// Network quality in the shared room tree +// ---------------- +// What each end measures about its links (state_transport::link_reports: the Ping/Pong round trip +// and traffic both ways), published as scalar leaves beside the rest of the room so every +// participant can see everyone's connection quality: +// +// conf.rooms..participants..net.{rtt_ms, in_kbps, out_kbps} +// the participant's OWN measurement of its link to the relay -- written by that client (like +// its .intent), shown next to its name in everyone's roster +// conf.rooms..participants..net.relay.{rtt_ms, in_kbps, out_kbps} +// the RELAY's measurement of its link to that participant (in = what the relay receives from +// it) -- written by the relay, like .pose +// conf.rooms..server.net.{in_kbps, out_kbps, links} +// the relay's totals over all of its links +// +// Values are kilobits/second and milliseconds; negative = not measured yet. + +struct net_sample { + double rtt_ms = -1.0; + double in_kbps = -1.0; + double out_kbps = -1.0; +}; + +// A participant's own measurement (client side). +void write_participant_net(cvc::app &app, const std::string &room, const std::string &pid, + const net_sample &s); +bool read_participant_net(cvc::app &app, const std::string &room, const std::string &pid, + net_sample &out); + +// The relay's measurement of its link to a participant (server side). +void write_relay_net(cvc::app &app, const std::string &room, const std::string &pid, + const net_sample &s); +bool read_relay_net(cvc::app &app, const std::string &room, const std::string &pid, + net_sample &out); + +// The relay's totals. +struct server_net { + double in_kbps = 0.0; + double out_kbps = 0.0; + int links = 0; +}; +void write_server_net(cvc::app &app, const std::string &room, const server_net &s); +bool read_server_net(cvc::app &app, const std::string &room, server_net &out); + +// Compact display text for a roster cell: "42 ms" / "--"; "850k" / "1.2M" / "--" (bits/s). +std::string format_rtt(double rtt_ms); +std::string format_rate(double kbps); + +} // namespace conference +} // namespace ariadne +} // namespace cvc + +#endif // __CVC_ARIADNE_CONFERENCE_ROOM_NET_H__ diff --git a/inc/cvc/ariadne/conference/room_scene.h b/inc/cvc/ariadne/conference/room_scene.h index 7c9ede1b0..22d067769 100644 --- a/inc/cvc/ariadne/conference/room_scene.h +++ b/inc/cvc/ariadne/conference/room_scene.h @@ -37,7 +37,7 @@ namespace conference { // // conf.rooms..scene.rev bumped by every scene write // conf.rooms..scene.env.background.{gradient, top.{r,g,b}, bottom.{r,g,b}} -// conf.rooms..scene.env.chrome the SceneGraph grid/axis chrome (1/0) +// conf.rooms..scene.env.chrome the SceneGraph diagnostic chrome (1/0) // conf.rooms..scene.env.lights..{present, kind, rig, azimuth, elevation, intensity, // cone, color.{r,g,b}, pos.{x,y,z}, target.{x,y,z}} // conf.rooms..scene.objects..{present, source.{uri, primitive, plane_size, @@ -97,7 +97,7 @@ std::uint64_t scene_revision(cvc::app &app, const std::string &room_id); cvc::ariadne::Scene read_room_scene(cvc::app &app, const std::string &room_id); // The MVP environment the server seeds a room with (§5g "ground + grid + a reference object"): a -// gradient background, the grid chrome on, a directional sun, a 20 m ground plane and a 1 m +// gradient background, the diagnostic chrome off, a directional sun, a 20 m ground plane and a 1 m // reference box standing on it at the origin. Z is up, as in every .ari scene. room_scene_env default_scene_env(); std::vector default_scene_objects(); diff --git a/inc/cvc/ariadne/holon.h b/inc/cvc/ariadne/holon.h index b83fac1be..9b3e918f0 100644 --- a/inc/cvc/ariadne/holon.h +++ b/inc/cvc/ariadne/holon.h @@ -25,6 +25,7 @@ #include #include #include +#include namespace cvc { namespace ariadne { @@ -57,8 +58,18 @@ namespace ariadne { enum class holon_role { server, client }; struct holon_config { + // A holon is advanced at its HOST's cadence, not a render loop's: arisrv pumps its server holon + // at ~5 Hz (24 quanta due per advance at 120 Hz) and a throttled browser tab may run at 10 fps. + // world_clock's default catch-up cap (8 quanta, sized for a 60+ Hz frame loop) would silently + // drop the rest -- running the room's avatars, and its tick, at a third of real time -- so a + // holon's clock allows 0.5 s of catch-up per advance. + holon_config() { clock.max_steps_per_advance = 60; } + cvc::world_clock::config clock; // the holon's own quantum / scale / mode double reconcile_tol = 0.05; // metres; the client's snap threshold (slice 2) + // CLIENT: how far (in ticks) the replica's clock may drift from the server's tick before + // apply_authoritative re-aligns it. 60 = 0.5 s at the default 120 Hz. See apply_authoritative. + std::uint64_t max_skew_ticks = 60; }; class holon { @@ -88,6 +99,8 @@ class holon { const cvc::physics::body_state &start = cvc::physics::body_state{}); void remove_participant(const std::string &pid); bool has_participant(const std::string &pid) const; + // Every registered pid, in pid order. + std::vector participants() const; // Set a participant's latest intent. Server: the validated input it integrates // into the authoritative pose. Client: only the LOCAL avatar's own input (a @@ -98,6 +111,15 @@ class holon { // arrived via replication. The local avatar reconciles to it (snap + replay on // divergence); a peer takes it as its newest interpolation sample. No-op on a // server holon or an unknown pid. + // + // Reconcile compares ticks, so the replica's clock must count the server's ticks. + // The two clocks start independently (a client joins a server that has run for + // hours), so a sample more than max_skew_ticks from the replica's tick -- the first + // one after joining, or one after a backgrounded browser tab stalled the frame + // loop -- re-aligns the replica: its clock seeks to server_tick and the local + // avatar's unacknowledged input log is dropped (its ticks no longer mean anything; + // its state is kept). Between re-alignments both clocks run from wall time at the + // same rate. void apply_authoritative(const std::string &pid, const cvc::physics::body_state &auth, std::uint64_t server_tick); @@ -112,23 +134,36 @@ class holon { // std::out_of_range for an unknown pid. cvc::physics::body_state state(const std::string &pid) const; - // A participant's DISPLAY pose at the current frame's interpolation alpha: - // predicted for the local avatar, interpolated between the last two samples for a - // peer (never extrapolated), the integrated pose on a server. + // A participant's DISPLAY pose this frame: predicted for the local avatar, the + // integrated pose on a server, and for a peer an interpolation between its last + // two samples (never extrapolated). Samples arrive far less often than quanta (the + // relay publishes ~5 Hz against a 120 Hz clock), so a peer is shown ONE SAMPLE + // INTERVAL behind: the newest sample's arrival starts a walk from `prev` to + // `newest` that takes as long as the server took between the two (their + // server_tick gap), measured on this holon's own clock (tick + alpha). cvc::physics::pose display_pose(const std::string &pid) const; + // CLIENT: the position jumps reconcile has applied to the local avatar `pid` + // (snap + replay on divergence) since the last call, summed, then cleared. A + // renderer eases this out so a correction reads as a glide rather than a jump; the + // holon itself stays exact. Zero for a peer, on a server, or for an unknown pid. + cvc::physics::vec3 take_correction(const std::string &pid); + private: struct participant { bool local = false; avatar_sim sim; // server authoritative integrator avatar_intent intent; // latest intent (server input / client-local input) predicted_avatar predicted; // client-local prediction + reconcile + cvc::physics::vec3 correction{0.0, 0.0, 0.0}; // reconcile jumps not yet taken // peer interpolation: the last two authoritative samples + a monotonic guard // (mirroring predicted_avatar's stale/duplicate-ack rejection) so a reordered // or duplicate peer sample never rewinds or freezes the interpolation window. bool have_sample = false; std::uint64_t last_sample_tick = 0; cvc::physics::body_state prev, newest; + std::uint64_t newest_arrival_tick = 0; // this clock's tick when `newest` arrived + std::uint64_t sample_span = 0; // server ticks from prev to newest (0 = one sample) }; void on_step(const cvc::clock_driver::clock_frame &f); diff --git a/inc/cvc/ariadne/holon_sync.h b/inc/cvc/ariadne/holon_sync.h index c11723250..5ca3f375a 100644 --- a/inc/cvc/ariadne/holon_sync.h +++ b/inc/cvc/ariadne/holon_sync.h @@ -84,12 +84,28 @@ bool read_intent(cvc::app &app, const std::string &room, const std::string &pid, // read its proposed .intent, feed it to the holon; advance the holon one frame; // then write every participant's authoritative .pose + .server_tick back to the // tree (which replicates to the clients). The caller supplies the server holon -// and calls this once per relay tick. A departed participant simply stops -// receiving intent/pose updates (reaping its holon slot is a follow-up). wall_dt +// and calls this once per relay tick. A participant who left (room_leave) is dropped +// from the holon, so a rejoin respawns on the spawn ring. wall_dt // is forwarded to holon::advance: the default (-1) measures real elapsed wall time // (what the relay wants); a test injects a fixed dt for a deterministic step count. void server_pump(holon &h, cvc::app &app, const std::string &room, double wall_dt = -1.0); +// Drive one frame of a CLIENT holon against the replicated room tree (H1 slice 6), the mirror of +// server_pump: register every present participant (room_roster) -- SELF_PID as the local, +// predicted avatar, everyone else as an interpolated peer, each starting at its replicated pose +// when one exists -- drop the ones that left, apply each replicated authoritative .pose (the holon +// ignores a sample it has already seen, so re-reading an unchanged pose every frame is harmless), +// then advance the holon one frame. Read the result with h.display_pose(pid). wall_dt as in +// server_pump. +void client_pump(holon &h, cvc::app &app, const std::string &room, const std::string &self_pid, + double wall_dt = -1.0); + +// Propose the local avatar's intent (client side): the client holon predicts with it at once and +// it is written to .intent for the server to validate. Call it when the intent CHANGES (each +// call is a replicated write), not every frame. +void propose_intent(holon &h, cvc::app &app, const std::string &room, const std::string &self_pid, + const avatar_intent &intent); + } // namespace ariadne } // namespace cvc diff --git a/inc/cvc/gl/CameraController.h b/inc/cvc/gl/CameraController.h index d226ef078..008167e55 100644 --- a/inc/cvc/gl/CameraController.h +++ b/inc/cvc/gl/CameraController.h @@ -226,6 +226,12 @@ class CameraController : public cvc::state_object { // camera would drift on its own. void releaseHeldKeys(); + // Whether `keySym` (a VTK key sym: "w", "Up", ...; a single letter matches either + // case) is held right now -- the same held set Fly navigation reads, so a host + // driving something else from the keys (an avatar in Orbit mode) sees exactly what + // the camera does, including the drop when a UI overlay takes the keyboard. + bool keyHeld(const std::string &keySym) const; + // Rebind a movement key (also settable via state "keys."). Actions: // "forward","backward","strafe_left","strafe_right","up","down","toggle_mode", // "sprint". Values are VTK key syms ("w","space","Control_L","Tab","Shift_L"). diff --git a/inc/cvc/gl/ariadne/room_view.h b/inc/cvc/gl/ariadne/room_view.h new file mode 100644 index 000000000..038ba4a61 --- /dev/null +++ b/inc/cvc/gl/ariadne/room_view.h @@ -0,0 +1,104 @@ +#ifndef CVC_GL_ARIADNE_ROOM_VIEW_H +#define CVC_GL_ARIADNE_ROOM_VIEW_H + +// The conference room in 3D (ARIADNE_CONFERENCE_ROADMAP.md §5g, H1 slice 6): the client side of +// the server-authoritative shared scene and the avatars. +// +// * The room scene (conf.rooms..scene.*, cvc::ariadne::conference::room_scene) is realized +// into the viewer's SceneGraph with realize_scene, and re-realized whenever its revision +// changes. Its node and light ids are prefixed "room_" so they never collide with the hosting +// document's own scene. +// * One marker per participant of a CLIENT holon (cvc::ariadne::client_pump keeps the holon in +// step with the replicated room), posed every frame at holon::display_pose: predicted for the +// local avatar, a smooth walk between samples for a peer. A reconcile jump on the local avatar +// (holon::take_correction) is eased out over `correction_tau`, so a server correction reads +// as a glide rather than a jump. +// * Third-person controls: intent_from_keys turns held WASD / arrow keys into an intent relative +// to the camera's heading, and self_position is where the host keeps the camera's orbit centre. +// +// Owner/render thread, like the SceneGraph it draws into. + +#include +#include // avatar_intent +#include // vec3 +#include +#include +#include +#include + +namespace cvc { +class app; +namespace ariadne { +class holon; +} // namespace ariadne +namespace gl { +class SceneRenderer; +namespace ariadne { +struct RealizedScene; + +// The local avatar's intent from the held movement keys, relative to the camera's heading on a +// Z-up ground: forward is the view direction (eye -> focal) flattened onto the ground, right is +// forward x up. Diagonals are normalized (never faster than `speed`); opposing keys cancel. A +// camera looking straight down has no heading, so forward is then +Y. +cvc::ariadne::avatar_intent intent_from_keys(bool forward, bool back, bool left, bool right, + const double eye[3], const double focal[3], + double speed); + +class room_view { +public: + // Draws ROOM's scene and SELF_PID's view of the avatars into `view`'s SceneGraph. BIND_PREFIX is + // the state prefix realize_scene resolves binds against (the host's, e.g. sg.getStatePrefix()). + room_view(SceneRenderer &view, cvc::app &app, std::string room, std::string self_pid, + std::string bind_prefix); + ~room_view(); // removes every node it created + room_view(const room_view &) = delete; + room_view &operator=(const room_view &) = delete; + + // Once per frame, after client_pump advanced the holon: re-realize the room scene if its + // revision changed, add/remove avatar markers to match the holon's participants, and pose them. + // wall_dt (seconds) paces the correction easing. + void update(cvc::ariadne::holon &h, double wall_dt); + + // Where the local avatar is shown (the eased display position, at the marker's feet). False + // until the holon has the local avatar. + bool self_position(double out[3]) const; + + // The room scene revision currently realized (0 = none yet). + std::uint64_t realized_revision() const { return _realized_rev; } + + // Non-fatal realize warnings since the last call (a missing mesh, ...), then cleared. + std::vector take_warnings(); + + // The pids that have an avatar marker right now, in pid order. + std::vector avatar_pids() const; + + // The SceneGraph node ids room_view uses, for hosts and tests. + static std::string object_node_id(const std::string &id) { return "room_" + id; } + static std::string avatar_node_id(const std::string &pid) { return "avatar_" + pid; } + + // Seconds over which a local reconcile correction is eased out (1/e). + double correction_tau = 0.12; + +private: + void sync_scene(); + void clear_scene(); + void sync_avatars(cvc::ariadne::holon &h, double wall_dt); + + SceneRenderer &_view; + cvc::app &_app; + std::string _room, _self, _prefix; + std::uint64_t _realized_rev = 0; + std::unique_ptr _realized; // owns the scene's light rigs while shown + std::vector _scene_ids; // the realized room nodes + lights, to remove + std::map _avatars; // pid -> marker created + cvc::physics::vec3 _self_offset{0.0, 0.0, 0.0}; // easing offset on the local avatar + bool _have_self = false; + double _self_pos[3] = {0.0, 0.0, 0.0}; + std::vector _warnings; +}; + +} // namespace ariadne +} // namespace gl +} // namespace cvc + +#endif // CVC_GL_ARIADNE_ROOM_VIEW_H diff --git a/inc/cvc/state/state_link_stats.h b/inc/cvc/state/state_link_stats.h new file mode 100644 index 000000000..1be9aa63a --- /dev/null +++ b/inc/cvc/state/state_link_stats.h @@ -0,0 +1,106 @@ +/* + Copyright 2026 The University of Texas at Austin + + This file is part of libcvc. + + libcvc is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License version 2.1 as published by the Free Software Foundation. + + libcvc is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. +*/ + +#ifndef __CVC_STATE_LINK_STATS_H__ +#define __CVC_STATE_LINK_STATS_H__ + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace cvc { + +// ---------------- +// cvc::state_link_stats +// ---------------- +// The quality of ONE transport connection (a link): traffic both ways and the round-trip time +// measured by the transport's Ping/Pong probe (state_transport.proto). Every connection of +// state_transport_ws and state_transport_grpc owns one, so a client's link to its relay, a relay's +// link to each client, and a server's links to other servers (delegation peers included) are all +// measured the same way. Counters are lock-free: the reader thread notes inbound bytes, any writer +// notes outbound bytes, and the pinger notes round trips concurrently. +// +// RTT: rtt_ms is the MEDIAN of the last kRttWindow round trips, so a one-off stall (a probe that +// waited while a browser tab was frozen) does not distort it for long -- an average would carry a +// multi-second outlier for dozens of probes. Last and min are kept beside it. All negative until +// the first Pong. +class state_link_stats { +public: + void note_rx(std::size_t bytes) noexcept; + void note_tx(std::size_t bytes) noexcept; + // The next Ping's sequence number (and counts the Ping as sent). + std::uint64_t next_ping_seq() noexcept; + // A Pong arrived: rtt_ns is now minus the echoed sent_ns, on the sender's own clock. + void note_rtt(std::int64_t rtt_ns) noexcept; + + struct counters { + std::uint64_t bytes_in = 0, bytes_out = 0; + std::uint64_t frames_in = 0, frames_out = 0; + std::uint64_t pings_sent = 0, pongs_received = 0; + double rtt_ms = -1.0, rtt_last_ms = -1.0, rtt_min_ms = -1.0; + }; + counters read() const noexcept; + + static constexpr int kRttWindow = 8; + +private: + std::atomic _bytes_in{0}, _bytes_out{0}; + std::atomic _frames_in{0}, _frames_out{0}; + std::atomic _pings{0}, _pongs{0}; + // RTTs in microseconds (-1 = none yet): the recent window (ring) for the median, last, and min. + mutable std::mutex _rtt_mu; + std::array _rtt_ring{}; + int _rtt_count = 0, _rtt_next = 0; + std::atomic _rtt_last_us{-1}, _rtt_min_us{-1}; +}; + +// One link as a transport reports it (state_transport::link_reports()). +struct state_link_report { + std::string transport; // "ws" | "grpc" + std::string remote_node_id; // the peer's node id (from its Hello); "" before the Hello + bool accepted = false; // this end ACCEPTED the connection (a relay's link to a client) + double age_s = 0.0; // since the connection opened + double idle_s = 0.0; // since the last inbound frame + state_link_stats::counters stats; +}; + +// Turns successive cumulative link reports into rates. update() is called periodically (say once a +// second) with the transports' current reports; it returns each link's bytes-per-second since the +// previous call, keyed like key_of(). A link seen for the first time reports 0 (no interval yet). +class state_link_rate_meter { +public: + struct rate { + double in_bps = 0.0, out_bps = 0.0; + }; + std::map update(const std::vector &reports, double now_s); + // ":" (+ ":in" for an accepted link), unique per live link. + static std::string key_of(const state_link_report &r); + +private: + struct sample { + std::uint64_t bytes_in = 0, bytes_out = 0; + double t = 0.0; + }; + std::map _last; +}; + +} // namespace cvc + +#endif // __CVC_STATE_LINK_STATS_H__ diff --git a/inc/cvc/state/state_transport.h b/inc/cvc/state/state_transport.h index cacee0375..a0598a1bd 100644 --- a/inc/cvc/state/state_transport.h +++ b/inc/cvc/state/state_transport.h @@ -11,10 +11,13 @@ #ifndef __CVC_STATE_TRANSPORT_H__ #define __CVC_STATE_TRANSPORT_H__ +#include +#include #include #include #include #include +#include #include #include #include @@ -110,6 +113,23 @@ class state_transport { state_peer_registry &peers() noexcept { return _peers; } const state_peer_registry &peers() const noexcept { return _peers; } + // -------- Link quality (Ping/Pong probes, per-connection traffic) -------- + // + // One report per live connection: bytes and frames both ways, and the round-trip time the + // transport's Ping/Pong probe measures (state_link_stats). A relay sees one link per client; a + // client sees its link to the relay; a server also sees its links to other servers. Transports + // without connections report none (the default). + virtual std::vector link_reports() const { return {}; } + + // How often each connection is pinged (default 2 s; 0 turns the probe off). The probe also keeps + // a quiet-but-connected peer from looking idle to a relay's reaper. + void set_ping_interval(std::chrono::milliseconds interval) noexcept { + _ping_interval_ms.store(interval.count(), std::memory_order_relaxed); + } + std::chrono::milliseconds ping_interval() const noexcept { + return std::chrono::milliseconds(_ping_interval_ms.load(std::memory_order_relaxed)); + } + // -------- Chunk fetch (volume/geometry streaming) -------- // // Callback-based chunk retrieval from remote peers. The @@ -202,6 +222,7 @@ class state_transport { protected: state_peer_registry _peers; state_transport *_bridge = nullptr; + std::atomic _ping_interval_ms{2000}; }; } // namespace cvc diff --git a/inc/cvc/state/state_transport_grpc.h b/inc/cvc/state/state_transport_grpc.h index 79def7cf9..4c26fc836 100644 --- a/inc/cvc/state/state_transport_grpc.h +++ b/inc/cvc/state/state_transport_grpc.h @@ -18,6 +18,7 @@ #include // media_frame + the media_router plug-in surface #include #include +#include #include #include #include @@ -26,6 +27,11 @@ #include namespace cvc { +namespace transport { +namespace v1 { +class Frame; // the generated envelope (proto/state_transport.proto), for on_link_frame +} // namespace v1 +} // namespace transport // ---------------- // cvc::state_transport_grpc @@ -162,6 +168,14 @@ class state_transport_grpc final : public state_transport, public media_endpoint void set_relay(bool enable) noexcept { _relay.store(enable, std::memory_order_relaxed); } bool relay() const noexcept { return _relay.load(std::memory_order_relaxed); } + // Invoked (on a server RPC thread) when a peer's stream ENDS -- it closed or dropped its + // connection -- with that peer's node id (from its Hello). The same contract as + // state_transport_ws::set_on_connection_closed, so a relay retracts a departed native client from + // the room exactly as it does a closed browser tab. Not called during stop(), nor for a stream + // that never sent a Hello. The handler must be thread-safe. + using connection_closed_fn = std::function; + void set_on_connection_closed(connection_closed_fn cb) { _on_conn_closed = std::move(cb); } + // ── Media plane (carried as the Frame envelope's MediaFrame, same as the WS transport) ── // publish_media_frame sends a frame to this transport's connections; a relay's media_router fans // each received frame back out via fan_media(). set_on_media is the leaf subscribe sugar. Media @@ -195,6 +209,8 @@ class state_transport_grpc final : public state_transport, public media_endpoint // Diagnostics. std::size_t shard_count() const; std::size_t connection_count() const; + // One report per live connection (bytes/frames both ways, Ping/Pong round-trip time). + std::vector link_reports() const override; std::uint64_t total_published() const noexcept { return _published.load(); } std::uint64_t total_sent_frames() const noexcept { return _sent_frames.load(); } std::uint64_t total_received_frames() const noexcept { return _recv_frames.load(); } @@ -242,6 +258,12 @@ class state_transport_grpc final : public state_transport, public media_endpoint // older ones being replayed. void admit_connection(const std::shared_ptr &conn); void unregister_connection(connection *conn); + // A server-side stream ended: fire the connection-closed handler for its peer (see + // set_on_connection_closed). + void notify_connection_closed(const std::string &node_id); + // Note an inbound frame on `c` (link counters + liveness), and answer / measure a link probe. + // Returns true when `in` was a Ping or Pong (fully handled: never dispatched further). + bool on_link_frame(connection &c, const ::cvc::transport::v1::Frame &in); void increment_recv_frames() noexcept { _recv_frames.fetch_add(1, std::memory_order_relaxed); } void increment_recv_mutations() noexcept { _recv_mutations.fetch_add(1, std::memory_order_relaxed); @@ -280,6 +302,7 @@ class state_transport_grpc final : public state_transport, public media_endpoint std::atomic _backfill_on_connect{true}; std::atomic _relay{false}; + connection_closed_fn _on_conn_closed; // set before start(); read on RPC threads std::atomic _backfilled{0}; std::atomic _published{0}; @@ -321,6 +344,12 @@ class state_transport_grpc final : public state_transport, public media_endpoint std::chrono::milliseconds _heartbeat_interval{0}; std::thread _heartbeat_thread; void heartbeat_loop(); + + // Link probe: every ping_interval(), send each live connection a Ping (state_transport.proto); + // the peer's Pong gives the round trip. Runs on its own thread from start() to stop(). + std::thread _ping_thread; + void ping_loop(); + void send_pings(); }; } // namespace cvc diff --git a/inc/cvc/state/state_transport_ws.h b/inc/cvc/state/state_transport_ws.h index 8fa0fb068..17bc2a284 100644 --- a/inc/cvc/state/state_transport_ws.h +++ b/inc/cvc/state/state_transport_ws.h @@ -181,6 +181,8 @@ class state_transport_ws final : public state_transport, public media_endpoint { // ── Diagnostics ── std::size_t shard_count() const; std::size_t connection_count() const; + // One report per live connection (bytes/frames both ways, Ping/Pong round-trip time). + std::vector link_reports() const override; std::uint64_t total_published() const noexcept { return _published.load(); } std::uint64_t total_sent_frames() const noexcept { return _sent_frames.load(); } std::uint64_t total_received_frames() const noexcept { return _recv_frames.load(); } @@ -231,6 +233,15 @@ class state_transport_ws final : public state_transport, public media_endpoint { void accept_loop(); void read_loop(std::shared_ptr conn); + // Link probe: Ping every live connection once ping_interval() has passed since the last round. + // Native: a thread started with the first start() / connect_to_peer(). Browser (no threads): + // pump_all() calls it every frame. A Ping that arrives is answered at once in deliver_inbound -- + // in a browser that is the WebSocket onmessage event, which keeps firing in a backgrounded tab. + void send_pings_if_due(); + void ping_loop(); + std::atomic _next_ping_ns{0}; + std::atomic _ping_started{false}; + struct impl; std::unique_ptr _impl; diff --git a/proto/state_transport.proto b/proto/state_transport.proto index 727f3416c..2f6406262 100644 --- a/proto/state_transport.proto +++ b/proto/state_transport.proto @@ -150,6 +150,21 @@ message MediaFrame { uint32 channels = 10; // audio: channel count (0 for video) } +// Link-quality probe. Each end of a connection sends a Ping on a timer; the other end answers at +// once with a Pong echoing seq + sent_ns, and the sender measures the round trip on its OWN +// monotonic clock (no clock sync). A Ping/Pong is link-level: never replicated, never relayed or +// bridged. Answering also proves liveness, so a relay's idle reaper keeps a client that is quiet +// but still connected (a backgrounded browser tab answers from its WebSocket event, which is not +// timer-throttled) while a ghost that cannot answer is still reaped. +message Ping { + uint64 seq = 1; + int64 sent_ns = 2; // sender's steady-clock ns +} +message Pong { + uint64 seq = 1; + int64 sent_ns = 2; // the Ping's sent_ns, echoed +} + // Single bidirectional frame envelope. Both client and server write // frames; same-cluster routing happens above this layer. message Frame { @@ -163,6 +178,8 @@ message Frame { SnapshotResponse snapshot_response = 7; Heartbeat heartbeat = 8; MediaFrame media = 9; + Ping ping = 10; + Pong pong = 11; } } diff --git a/src/arigl/main.cpp b/src/arigl/main.cpp index c7eef3de1..4179e5fce 100644 --- a/src/arigl/main.cpp +++ b/src/arigl/main.cpp @@ -58,18 +58,24 @@ #include #include #include +#include // std::random_device — the conference node id's per-launch suffix #include // arigl info: the GL report, line by line #if CVC_ENABLE_WS || CVC_ENABLE_GRPC // Conference mode: a state transport joins a hosted room, chosen by the relay URL scheme — // ws://|wss:// (WebSocket, what browsers dial; the host page sets wss:///…) or grpc:// // (native gRPC). register_conference supplies the room verbs and the shard/transport are wired here // in the host; both transports carry the canonical media_frame and interop via the relay's router. +#include // room_join (re-assert after a reconnect) +#include // publish our link quality to the room +#include // the client replica of the room holon (avatars, H1) +#include // client_pump / propose_intent over the room tree #include // frame_ptr — forward the local camera/mic streams #include // stream::open — per-peer decode proxies #include // subscribe / publish_external #include // look up the local camera/mic streams by token #include // one-advance-per-frame fan-out over the world_clock #include // app.world_clock().t() — the common media timestamp +#include // the room scene + avatar markers in the 3D view #include // list_audio_capture_devices — mic picker #include // list_cameras — camera picker #include // open_audio_output — play each peer's mic @@ -281,8 +287,39 @@ struct ConferenceSession { cvc::app *app_ = nullptr; std::string pid_; // this participant's node id (== the room-join pid; stamps our outgoing frames) std::string - prefix_; // the document chroot prefix: host state access must prepend it so it reaches - // the SAME keys the .ari's (state-get/-set) do (which are prefix-relative) + prefix_; // the document chroot prefix: host state access must prepend it so it reaches + // the SAME keys the .ari's (state-get/-set) do (which are prefix-relative) + std::string room_; // the joined room id (conf.rooms.) + + // ── link quality + reconnect ── + // Once a second: read our link to the relay (state_transport::link_reports -- the Ping/Pong + // round trip and traffic both ways), publish it to the room as participants..net.* so every + // roster shows it next to our name, and show it in our own row (self.ping / self.down / self.up). + // If the link is gone (the relay restarted, the network dropped, or a long-backgrounded tab was + // reaped) dial the relay again every few seconds, and after reconnecting re-assert our presence + // for a few seconds (the relay marked us as left; our write must land after its snapshot). + std::string relay_url_; + bool relay_grpc_ = false; + bool ever_linked_ = false; + double next_net_s_ = 0.0, next_reconnect_s_ = 0.0, rejoin_until_s_ = 0.0; + cvc::state_link_rate_meter link_meter_; + + // ── the room in 3D (H1 slice 6): the shared scene + avatars, third-person controls ── + // A client replica of the room holon, kept in step with the replicated room tree by client_pump + // each frame: our avatar predicted from our own intent, peers walked between the relay's pose + // samples. room_view draws the server's scene + one marker per participant into the window's + // SceneGraph; the camera orbits our avatar. Made by attach_view (window mode only); room_view_ + // is declared after holon_ and torn down first (stop()). + std::unique_ptr holon_; + std::unique_ptr room_view_; + cvc::gl::CameraController *cam_ = nullptr; + cvc::ariadne::avatar_intent last_intent_; // the last intent proposed (propose only on change) + bool intent_proposed_ = false; // proposed once yet (the first frame always proposes) + std::vector drawn_pids_; // avatars drawn at the last log (log on change) + bool framed_ = false; // camera framed on our avatar once + double orbit_center_[3] = {0.0, 0.0, 0.0}; + static constexpr double kWalkSpeed = 2.0; // m/s + static constexpr double kEyeHeight = 1.2; // orbit centre above the avatar's feet (m) // Publish: forward the local camera (latest) and mic (ring) streams. The .ari puts their tokens // in "cam_token" / "mic_token". We re-stamp each outgoing frame with a COMMON host clock (one @@ -371,6 +408,8 @@ struct ConferenceSession { // future step between these without renumbering. enum : int { kOrderPumpAll = 10, // drain/flush the state transport (replicated room tree) — first + kOrderNet = 12, // ~1 Hz: our link quality -> the room; reconnect a lost relay link + kOrderRoom = 15, // the room in 3D: keys -> intent, client holon, scene + avatars, camera kOrderWall = 20, // refresh now_wall_ for the wall-time talk-hold kOrderControls = 30, // read the user-list window's per-peer local controls kOrderDevEnum = 40, // lazily enumerate camera/mic devices (one-shot) @@ -397,11 +436,24 @@ struct ConferenceSession { // shown in the roster) may be any text — e.g. a browser's random "guest-ab12" — and room-join // already carries it separately as the "name" attr, so derive a safe node id here rather than // passing the display name straight through (a hyphen in it otherwise throws). + // + // The id also gets a random per-launch suffix. Replication drops a write whose (origin node id, + // sequence) it has already seen, so a client that restarts under the SAME id (a reloaded tab, + // a relaunched native client, two people who both kept the name "guest") would have every + // write ignored until the relay restarts: it would join a room nobody can see it in. A fresh + // id per session makes each launch a new participant; the name in the roster is unchanged. std::string nodeId; for (char c : self) nodeId += std::isalnum(static_cast(c)) ? c : '_'; if (nodeId.empty() || std::isdigit(static_cast(nodeId.front()))) nodeId.insert(0, "n_"); + { + std::random_device rd; + const char *digits = "0123456789abcdefghijklmnopqrstuvwxyz"; + nodeId += '_'; + for (int i = 0; i < 6; ++i) + nodeId += digits[rd() % 36]; + } gla::register_conference(app, backend); // room-join/roster + streaming + stream_view shard = std::make_unique(app, "conf", nodeId); // Replicate ONLY the shared room tree (conf.rooms.*), never this client's per-document UI state @@ -416,6 +468,7 @@ struct ConferenceSession { app_ = &app; pid_ = nodeId; prefix_ = prefix; + room_ = room; // Drive the world clock from wall time via the clock driver: live mode so advance() banks wall // time. The driver owns the single advance of this app's clock and fires the conference's // per-frame steps in order (install_observers); nothing else advances it. @@ -440,6 +493,8 @@ struct ConferenceSession { // the shared ops and send_media_/stop_ wrap the concrete's publish/stop. const std::string relayUrl(relay); const bool want_grpc = (relayUrl.rfind("grpc://", 0) == 0); + relay_url_ = relayUrl; + relay_grpc_ = want_grpc; bool ok = false; if (want_grpc) { #if CVC_ENABLE_GRPC @@ -506,6 +561,136 @@ struct ConferenceSession { active = true; return true; } + static double steady_s() { + return std::chrono::duration(std::chrono::steady_clock::now().time_since_epoch()) + .count(); + } + + void net_step() { + if (!app_ || !tp_) + return; + const double now = steady_s(); + if (now < next_net_s_) + return; + next_net_s_ = now + 1.0; + const std::vector links = tp_->link_reports(); + const cvc::state_link_report *relay = nullptr; + for (const cvc::state_link_report &r : links) + if (!r.accepted) { // the link WE dialed: the relay + relay = &r; + break; + } + auto &st = cvc::state::instance(*app_); + if (!relay) { + st(prefix_ + ".self.ping").value(std::string("--")); + if (ever_linked_ && now >= next_reconnect_s_) + reconnect(now); + return; + } + ever_linked_ = true; + const auto rates = link_meter_.update(links, now); + const auto it = rates.find(cvc::state_link_rate_meter::key_of(*relay)); + cvc::ariadne::conference::net_sample s; + s.rtt_ms = relay->stats.rtt_ms; + s.in_kbps = it == rates.end() ? 0.0 : it->second.in_bps * 8.0 / 1000.0; + s.out_kbps = it == rates.end() ? 0.0 : it->second.out_bps * 8.0 / 1000.0; + cvc::ariadne::conference::write_participant_net(*app_, room_, pid_, s); + st(prefix_ + ".self.ping").value(cvc::ariadne::conference::format_rtt(s.rtt_ms)); + st(prefix_ + ".self.down").value(cvc::ariadne::conference::format_rate(s.in_kbps)); + st(prefix_ + ".self.up").value(cvc::ariadne::conference::format_rate(s.out_kbps)); + if (now < rejoin_until_s_) + cvc::ariadne::conference::room_join(*app_, room_, pid_, st(prefix_ + ".conf.self").value()); + } + + void reconnect(double now) { + next_reconnect_s_ = now + 3.0; + std::fprintf(stderr, "arigl: conference: lost the relay link; reconnecting to %s\n", + relay_url_.c_str()); + bool ok = false; +#if CVC_ENABLE_GRPC + if (relay_grpc_ && grpc_) + ok = grpc_->connect_to_peer(relay_url_.substr(7), std::chrono::milliseconds(3000)); +#endif +#if CVC_ENABLE_WS + if (!relay_grpc_ && ws_) + ok = ws_->connect_to_peer(relay_url_, std::chrono::milliseconds(3000)); +#endif + if (ok) + rejoin_until_s_ = now + 6.0; // re-assert presence until the relay's snapshot has landed + } + + // Window mode: show the room in 3D in `view` — the server's shared scene and an avatar per + // participant — with third-person controls: WASD / arrow keys walk our avatar relative to the + // camera's heading, and `cam` (Orbit) keeps orbiting it. A no-op for a document with no relay. + void attach_view(cvc::gl::SceneRenderer &view, cvc::gl::CameraController &cam) { + if (!active || !driver_ || !app_) + return; + holon_ = std::make_unique(cvc::ariadne::holon_role::client); + room_view_ = + std::make_unique(view, *app_, room_, pid_, view.scene().getStatePrefix()); + cam_ = &cam; + driver_->add_frame_observer(kOrderRoom, + [this](const cvc::clock_driver::clock_frame &f) { step_room(f); }); + } + + void step_room(const cvc::clock_driver::clock_frame &f) { + if (!holon_ || !room_view_ || !cam_) + return; + // Keys -> intent, only in Orbit: in Fly the same keys move the camera instead. The camera's + // held set is dropped while the UI has the keyboard, so typing a name never walks. + cvc::ariadne::avatar_intent in; + if (cam_->mode() == cvc::gl::CameraController::Mode::Orbit) { + double eye[3], focal[3], up[3]; + cam_->getPose(eye, focal, up); + const auto held = [this](const char *a, const char *b) { + return cam_->keyHeld(a) || cam_->keyHeld(b); + }; + in = gla::intent_from_keys(held("w", "Up"), held("s", "Down"), held("a", "Left"), + held("d", "Right"), eye, focal, kWalkSpeed); + } + // Propose on the first frame even when standing still, so the tree never holds a walk an + // earlier session under this id proposed; after that, only on change (a replicated write). + if (!intent_proposed_ || in.move.x != last_intent_.move.x || in.move.y != last_intent_.move.y || + in.move.z != last_intent_.move.z) { + cvc::ariadne::propose_intent(*holon_, *app_, room_, pid_, in); + last_intent_ = in; + intent_proposed_ = true; + } + // Re-assert it locally every frame: our avatar joins the holon only once our room-join is in + // the tree, so an intent proposed before then must still reach the prediction. + holon_->set_intent(pid_, last_intent_); + cvc::ariadne::client_pump(*holon_, *app_, room_, pid_, f.wall_dt); + room_view_->update(*holon_, f.wall_dt); + // Log who is drawn whenever that changes (the browser page shows stderr in its footer and the + // console): the quickest check that a client sees the room the relay holds. + const std::vector drawn = room_view_->avatar_pids(); + if (drawn != drawn_pids_) { + std::string who; + for (const std::string &pid : drawn) + who += (who.empty() ? "" : ", ") + pid; + std::fprintf(stderr, "arigl: room: %zu avatar(s): %s\n", drawn.size(), who.c_str()); + drawn_pids_ = drawn; + } + for (const std::string &w : room_view_->take_warnings()) + std::fprintf(stderr, "arigl: room: %s\n", w.c_str()); + // Third person: frame our avatar once, then keep the orbit centre on it. + double p[3]; + if (room_view_->self_position(p)) { + if (!framed_) { + // Frame the room around you, not just you: arriving at a party, you want to see who's here. + cam_->frameBounds(p[0] - 8.0, p[1] - 8.0, p[2], p[0] + 8.0, p[1] + 8.0, p[2] + 2.0); + framed_ = true; + } + const double c[3] = {p[0], p[1], p[2] + kEyeHeight}; + if (std::fabs(c[0] - orbit_center_[0]) + std::fabs(c[1] - orbit_center_[1]) + + std::fabs(c[2] - orbit_center_[2]) > + 1e-4) { + cam_->setOrbitCenter(c[0], c[1], c[2]); + orbit_center_[0] = c[0], orbit_center_[1] = c[1], orbit_center_[2] = c[2]; + } + } + } + // One advance per frame. The driver advances the per-app world_clock from wall time (the single // media stamp, frame().t) and fires the conference's per-frame steps as the ordered observers // install_observers() registered. Called from the input pump (run mode) and directly per frame @@ -524,6 +709,8 @@ struct ConferenceSession { // Drive the state transport first: flush replicated room-tree writes and service queued reads // so room-join/roster are in place before anything this frame reads them. driver_->add_frame_observer(kOrderPumpAll, [this](const cf &) { tp_->pump_all(); }); + // Our link quality to the room, and reconnecting a lost relay link (~1 Hz). + driver_->add_frame_observer(kOrderNet, [this](const cf &) { net_step(); }); // Wall-clock seconds for the talk-hold — deliberately wall time, not world time, so the hold is // frame-rate independent and does not freeze when the sim clock is paused or scaled. driver_->add_frame_observer(kOrderWall, [this](const cf &) { @@ -664,7 +851,8 @@ struct ConferenceSession { } void stop() { - peers_.clear(); // close decode proxies before the transport/shard go away + room_view_.reset(); // remove the room's nodes while the SceneGraph is still alive + peers_.clear(); // close decode proxies before the transport/shard go away mix_out_.reset(); for (publisher &p : pubs_) p.sub.reset(); @@ -903,6 +1091,7 @@ struct ConferenceSession { "document locally.\n"); return false; } + void attach_view(cvc::gl::SceneRenderer &, cvc::gl::CameraController &) {} void pump() {} void stop() {} #endif @@ -976,7 +1165,8 @@ int run_window(const ari::HostOptions &o) { // A conference-setup failure must not take down the runtime — fall back to running the document // locally (no relay) and say why, rather than letting the exception abort main. try { - conf.setup(app, sg.getStatePrefix(), rt.backend()); + if (conf.setup(app, sg.getStatePrefix(), rt.backend())) + conf.attach_view(view, cam); // the room in 3D behind the conference windows } catch (const std::exception &e) { std::fprintf(stderr, "arigl: conference setup failed (%s); running the document locally\n", e.what()); diff --git a/src/arigl/web/cyprty.html b/src/arigl/web/cyprty.html index 9a9fd414f..b26f90e21 100644 --- a/src/arigl/web/cyprty.html +++ b/src/arigl/web/cyprty.html @@ -80,12 +80,18 @@

cyberpcangel · cyprty

// The relay URL is derived from the PAGE ORIGIN — a wss:// upgrade of this https page's host, at // /cyprty-ws (Apache reverse-proxies that to the relay). So the deployed client connects to its - // own server automatically; a future build could let the user pick a different one. - const RELAY = (location.protocol === 'https:' ? 'wss://' : 'ws://') + location.host + '/cyprty-ws'; - const NAME = 'guest-' + Math.random().toString(36).slice(2, 7); + // own server automatically. ?relay=ws://host:port (or wss://) points the page at another relay — + // a staging or local arisrv — and ?name= presets the display name (otherwise a random guest-xxxxx). + const PARAMS = new URLSearchParams(location.search); + const RELAY_PARAM = PARAMS.get('relay'); + const RELAY = (RELAY_PARAM && /^wss?:\/\/[^\s]+$/.test(RELAY_PARAM)) + ? RELAY_PARAM + : (location.protocol === 'https:' ? 'wss://' : 'ws://') + location.host + '/cyprty-ws'; + const NAME = (PARAMS.get('name') || '').trim().slice(0, 32) || + 'guest-' + Math.random().toString(36).slice(2, 7); // ?synthetic=1 drives the self-cam from a hardware-free MJPEG test pattern instead of the real // camera, so the conference (peer video included) can be tried in a browser with no camera. - const SYNTH = new URLSearchParams(location.search).get('synthetic'); + const SYNTH = PARAMS.get('synthetic'); // Browsers suspend an AudioContext created without a user gesture (autoplay policy). SDL's audio // output (how we play each peer's mic) is one such context, so without this a peer's audio never diff --git a/src/arisrv/main.cpp b/src/arisrv/main.cpp index b6ba15bed..3fa45e01c 100644 --- a/src/arisrv/main.cpp +++ b/src/arisrv/main.cpp @@ -23,9 +23,11 @@ #include #include #include +#include // std::getenv (CVC_ARISRV_ROOM_DEBUG) #include #include // room_leave (retract a departed peer) #include // room_relay_tick +#include // per-participant + room link-quality stats #include // seed_default_scene (the shared 3D scene) #include // server-authoritative avatar holon (H1) #include // server_pump: intent -> holon -> authoritative pose @@ -44,9 +46,40 @@ #include #include +#if defined(__linux__) && defined(__GLIBC__) +#include // backtrace -- a fatal signal logs the relay's stack +#include // write, STDERR_FILENO +#endif + namespace { std::atomic g_stop{false}; void on_signal(int) { g_stop.store(true); } + +#if defined(__linux__) && defined(__GLIBC__) +// A relay runs unattended, so a crash must explain itself in its log: on SIGSEGV / SIGABRT (glibc's +// "malloc(): ... corrupted" aborts) / SIGBUS / SIGFPE / SIGILL, write the stack to stderr, then +// re-raise with the default action so the process still dies (and exits non-zero). Only +// async-signal-safe calls: backtrace_symbols_fd writes straight to the fd without allocating, and +// install_crash_trace() calls backtrace() once up front so its lazy libgcc load never happens +// inside the handler on a corrupted heap. +void on_fatal(int sig) { + static const char head[] = "\n[arisrv] FATAL signal; stack:\n"; + (void)!write(STDERR_FILENO, head, sizeof(head) - 1); + void *frames[64]; + const int n = backtrace(frames, 64); + backtrace_symbols_fd(frames, n, STDERR_FILENO); + std::signal(sig, SIG_DFL); + std::raise(sig); +} +void install_crash_trace() { + void *warm[1]; + (void)backtrace(warm, 1); + for (int sig : {SIGSEGV, SIGABRT, SIGBUS, SIGFPE, SIGILL}) + std::signal(sig, on_fatal); +} +#else +void install_crash_trace() {} +#endif } // namespace int main(int argc, char **argv) { @@ -74,9 +107,22 @@ int main(int argc, char **argv) { cvc::app app; + // Presence: when a client's connection ends -- a native (gRPC) client exits, or a browser (WS) + // tab closes -- retract it from the room so the roster and the avatar holon do not accumulate + // ghosts. Both transports report the departed node id the same way, on their own threads, so the + // id is only queued here; the main loop below applies room_leave (where every other room-tree + // write happens) and the retraction rides the normal pump + relay forwarding to the others. + std::mutex departed_mu; + std::vector departed; + const auto on_departed = [&](const std::string &node_id) { + std::lock_guard lk(departed_mu); + departed.push_back(node_id); + }; + // Room-tree (state) plane: serve + RELAY (forward frames between clients). The shard ingests the // room tree into app.root() so the media relay below can watch it. cvc::state_transport_grpc state_tp; + state_tp.set_on_connection_closed(on_departed); state_tp.start(stateListen, nodeId, "conf"); state_tp.set_relay(true); cvc::state_cluster_shard shard(app, "conf", nodeId); @@ -98,23 +144,12 @@ int main(int argc, char **argv) { // stream-descriptor mutations cross both ways now; request_snapshot over WS for a late joiner's // full bootstrap is Phase 2 (a browser joining a populated room sees peers that join AFTER it). cvc::state_transport_ws ws_tp; + ws_tp.set_on_connection_closed(on_departed); // a closed browser tab leaves like a native client ws_tp.start(wsListen, nodeId, "conf"); ws_tp.set_relay(true); ws_tp.register_shard(&shard); state_tp.set_bridge(&ws_tp); ws_tp.set_bridge(&state_tp); - - // Presence: when a browser (WS) client closes its tab, retract it from the room so the roster - // does not accumulate ghosts of closed tabs. The close fires on a WS reader thread, so just queue - // the departed node id here and apply room_leave on the main loop below (where every other - // room-tree write happens), then the retraction rides the normal pump + relay forwarding to the - // others. - std::mutex departed_mu; - std::vector departed; - ws_tp.set_on_connection_closed([&](const std::string &node_id) { - std::lock_guard lk(departed_mu); - departed.push_back(node_id); - }); #endif // Media plane: a stream_relay. Constructed BEFORE media_tp.start() (it installs the frame @@ -135,6 +170,7 @@ int main(int argc, char **argv) { mrouter.add_endpoint(&ws_tp); // browser (wasm) WebSocket clients #endif + install_crash_trace(); std::signal(SIGINT, on_signal); std::signal(SIGTERM, on_signal); std::printf("[arisrv] relay up | room='%s' node='%s' | state=%s | media=%s\n", room.c_str(), @@ -160,9 +196,21 @@ int main(int argc, char **argv) { // several quanta per tick and clients interpolate between the ~5 Hz pose samples; // a smoother authoritative rate is a later tuning (tick the holon on its own). cvc::ariadne::holon server_holon(cvc::ariadne::holon_role::server); + const char *room_debug_env = std::getenv("CVC_ARISRV_ROOM_DEBUG"); + const bool room_debug = + room_debug_env && *room_debug_env && std::strcmp(room_debug_env, "0") != 0; + int room_debug_ticks = 0; + // Link stats (~1 s): every connection on both planes reports its traffic and Ping/Pong round + // trip (state_transport::link_reports); the rate meter turns the counters into rates. The relay + // publishes its view of each participant (net.relay.*) and its totals (server.net.*) into the + // room tree, and logs them: every link each ~5 s with CVC_ARISRV_ROOM_DEBUG, a totals line each + // minute otherwise. Links that are not participants (peer servers, delegation) are logged too. + cvc::state_link_rate_meter link_meter; + int net_ticks = 0; + const auto net_start = std::chrono::steady_clock::now(); // The shared 3D scene (H1 slice 5, roadmap §5g): the relay is its sole writer and seeds the MVP - // environment (ground + grid + a reference box) once. The writes are local shard mutations, so - // the first pass of the loop below fans them to the clients, and late joiners get them in the + // environment (ground + a reference box under a sun) once. The writes are local shard mutations, + // so the first pass of the loop below fans them to the clients, and late joiners get them in the // snapshot. if (cvc::ariadne::conference::seed_default_scene(app, room)) { const auto rev = cvc::ariadne::conference::scene_revision(app, room); @@ -185,8 +233,12 @@ int main(int argc, char **argv) { // server-authoritative room+scene tree rides on. ws_tp.broadcast_snapshot(); } - // Apply any queued departures (a browser tab closed) before pumping, so each retraction goes - // out to the remaining clients this tick. +#endif + // Apply any queued departures (a native client exited or a browser tab closed) before pumping, + // so each retraction goes out to the remaining clients this tick. The departed participant's + // proposed intent is zeroed too: the server holon drops them now (server_pump follows the + // roster), and a client that rejoins under the same id must not resume a walk it proposed + // before it left. { std::vector gone; { @@ -195,16 +247,77 @@ int main(int argc, char **argv) { } for (const std::string &nid : gone) { cvc::ariadne::conference::room_leave(app, room, nid); + cvc::ariadne::write_intent(app, room, nid, cvc::ariadne::avatar_intent{}); std::printf("[arisrv] participant left: %s\n", nid.c_str()); } if (!gone.empty()) std::fflush(stdout); } -#endif cvc::ariadne::conference::room_relay_tick(app, room, relay, relayed); // Advance the authoritative avatar holon and write each participant's pose. The // pose writes are local shard mutations, fanned to clients by the drain below. cvc::ariadne::server_pump(server_holon, app, room); + // CVC_ARISRV_ROOM_DEBUG=1: every ~5 s, log the room as the server holds it -- each present + // participant's authoritative position and its proposed intent -- so replication across the + // gRPC and WebSocket planes can be checked from the relay without a client. + if (++net_ticks % 5 == 0) { + std::vector links = state_tp.link_reports(); +#if CVC_ENABLE_WS + for (cvc::state_link_report &r : ws_tp.link_reports()) + links.push_back(std::move(r)); +#endif + const double now_s = + std::chrono::duration(std::chrono::steady_clock::now() - net_start).count(); + const auto rates = link_meter.update(links, now_s); + cvc::ariadne::conference::server_net total; + std::set present; + for (const auto &p : cvc::ariadne::conference::room_roster(app, room)) + present.insert(p.pid); + for (const cvc::state_link_report &r : links) { + const auto it = rates.find(cvc::state_link_rate_meter::key_of(r)); + const double in_kbps = it == rates.end() ? 0.0 : it->second.in_bps * 8.0 / 1000.0; + const double out_kbps = it == rates.end() ? 0.0 : it->second.out_bps * 8.0 / 1000.0; + total.in_kbps += in_kbps; + total.out_kbps += out_kbps; + ++total.links; + if (present.count(r.remote_node_id)) + cvc::ariadne::conference::write_relay_net(app, room, r.remote_node_id, + {r.stats.rtt_ms, in_kbps, out_kbps}); + } + cvc::ariadne::conference::write_server_net(app, room, total); + const bool detail = room_debug && net_ticks % 25 == 0; + if (detail || net_ticks % 300 == 0) { + std::printf("[arisrv] net: %d link(s) | in %.1f kb/s | out %.1f kb/s\n", total.links, + total.in_kbps, total.out_kbps); + if (detail) + for (const cvc::state_link_report &r : links) { + const auto it = rates.find(cvc::state_link_rate_meter::key_of(r)); + std::printf("[arisrv] %-4s %-16s %-7s rtt %6.1f ms (min %.1f) | in %7.1f kb/s | out " + "%7.1f kb/s | idle %.1fs\n", + r.transport.c_str(), r.remote_node_id.c_str(), + r.accepted ? "client" : "peer", r.stats.rtt_ms, r.stats.rtt_min_ms, + it == rates.end() ? 0.0 : it->second.in_bps * 8.0 / 1000.0, + it == rates.end() ? 0.0 : it->second.out_bps * 8.0 / 1000.0, r.idle_s); + } + std::fflush(stdout); + } + } + if (room_debug && ++room_debug_ticks % 25 == 0) { + const auto roster = cvc::ariadne::conference::room_roster(app, room); + std::printf("[arisrv] room '%s': %zu present, holon tick %llu\n", room.c_str(), roster.size(), + static_cast(server_holon.tick())); + for (const auto &p : roster) { + cvc::physics::body_state s; + std::uint64_t t = 0; + cvc::ariadne::avatar_intent in; + const bool has_pose = cvc::ariadne::read_pose(app, room, p.pid, s, t); + const bool has_intent = cvc::ariadne::read_intent(app, room, p.pid, in); + std::printf("[arisrv] %-16s pos %s (%.2f, %.2f) intent %s (%.2f, %.2f)\n", p.pid.c_str(), + has_pose ? "" : "none", s.position.x, s.position.y, has_intent ? "" : "none", + in.move.x, in.move.y); + } + std::fflush(stdout); + } #if CVC_ENABLE_WS // Fan the relay's OWN room-tree writes — the room_leave retractions queued above and // room_relay_tick's per-stream "ready" markers — out to EVERY client on BOTH transports. The diff --git a/src/cvc/CMakeLists.txt b/src/cvc/CMakeLists.txt index f78bc700e..a32cdb68c 100644 --- a/src/cvc/CMakeLists.txt +++ b/src/cvc/CMakeLists.txt @@ -42,6 +42,7 @@ set(INCLUDE_FILES ../../inc/cvc/state/state_message.h ../../inc/cvc/state/state_message_bus.h ../../inc/cvc/state/state_peer_registry.h + ../../inc/cvc/state/state_link_stats.h ../../inc/cvc/state/state_write_policy.h ../../inc/cvc/state/state_distributed_metrics.h ../../inc/cvc/state/state_delegation_manager.h @@ -293,6 +294,7 @@ set(SOURCE_FILES media/media_router.cpp state/state_message_bus.cpp state/state_peer_registry.cpp + state/state_link_stats.cpp state/state_write_policy.cpp state/state_distributed_metrics.cpp state/state_delegation_manager.cpp @@ -1310,7 +1312,7 @@ list(APPEND INCLUDE_FILES ../../inc/cvc/net/http_client.h) # the retained Widget tree, the Runtime + reconcile boundary, and direct # cvc::state binding, walking a pluggable Backend. It touches NO VTK/ImGui/GL — # the ImGui-over-VTK backend lives in cvcGL (src/cvcGL/ariadne). Roadmap §16.1. -list(APPEND SOURCE_FILES ariadne/ariadne.cpp ariadne/loader.cpp ariadne/uri.cpp ariadne/uri_state.cpp ariadne/uri_http.cpp ariadne/uri_http_cache.cpp ariadne/net_intrinsics.cpp ariadne/stream_intrinsics.cpp ariadne/conference/room_intrinsics.cpp ariadne/conference/room_scene.cpp ariadne/state_io.cpp ariadne/ftxui_backend.cpp ariadne/term_graphics.cpp ariadne/tvision_backend.cpp ariadne/app_runtime.cpp ariadne/doc_args.cpp ariadne/host_cli.cpp ariadne/host_signals.cpp ariadne/requires.cpp ariadne/host_info.cpp ariadne/avatar_sim.cpp ariadne/holon.cpp ariadne/holon_sync.cpp) +list(APPEND SOURCE_FILES ariadne/ariadne.cpp ariadne/loader.cpp ariadne/uri.cpp ariadne/uri_state.cpp ariadne/uri_http.cpp ariadne/uri_http_cache.cpp ariadne/net_intrinsics.cpp ariadne/stream_intrinsics.cpp ariadne/conference/room_intrinsics.cpp ariadne/conference/room_scene.cpp ariadne/conference/room_net.cpp ariadne/state_io.cpp ariadne/ftxui_backend.cpp ariadne/term_graphics.cpp ariadne/tvision_backend.cpp ariadne/app_runtime.cpp ariadne/doc_args.cpp ariadne/host_cli.cpp ariadne/host_signals.cpp ariadne/requires.cpp ariadne/host_info.cpp ariadne/avatar_sim.cpp ariadne/holon.cpp ariadne/holon_sync.cpp) list(APPEND INCLUDE_FILES ../../inc/cvc/ariadne/widget.h ../../inc/cvc/ariadne/backend.h @@ -1334,6 +1336,7 @@ list(APPEND INCLUDE_FILES ../../inc/cvc/ariadne/conference/room_intrinsics.h ../../inc/cvc/ariadne/conference/room_media_intrinsics.h ../../inc/cvc/ariadne/conference/room_scene.h + ../../inc/cvc/ariadne/conference/room_net.h ../../inc/cvc/ariadne/state_io.h ../../inc/cvc/ariadne/scene.h ../../inc/cvc/ariadne/value.h diff --git a/src/cvc/ariadne/conference/room_intrinsics.cpp b/src/cvc/ariadne/conference/room_intrinsics.cpp index d8a013e42..361a2fb81 100644 --- a/src/cvc/ariadne/conference/room_intrinsics.cpp +++ b/src/cvc/ariadne/conference/room_intrinsics.cpp @@ -12,6 +12,7 @@ #include // strtod (room-set-intent numeric args) #include // register_action_intrinsics #include +#include // the roster entries carry each link's quality #include // write_intent (room-set-intent verb) #include // app.root() #include // cvc::state, SEPARATOR, findDescendant, children, value @@ -271,12 +272,23 @@ void register_room_intrinsics(cvc::app &app) { const std::string room_id = rh ? rh->room_id : handle_token(args[0]); if (room_id.empty()) return err("room-roster: ROOM must be a room-join handle or room-id"); + // Each entry also carries the participant's self-reported link quality (room_net.h): + // raw numbers (negative = not measured yet) and roster-cell text. std::vector list; - for (const room_participant &p : room_roster(app, room_id)) + for (const room_participant &p : room_roster(app, room_id)) { + net_sample net; + read_participant_net(app, room_id, p.pid, net); list.push_back(se::make_dict({ {"pid", se::value_t(p.pid)}, {"name", se::value_t(p.name)}, + {"rtt_ms", se::value_t(net.rtt_ms)}, + {"in_kbps", se::value_t(net.in_kbps)}, + {"out_kbps", se::value_t(net.out_kbps)}, + {"ping", se::value_t(format_rtt(net.rtt_ms))}, + {"down", se::value_t(format_rate(net.in_kbps))}, + {"up", se::value_t(format_rate(net.out_kbps))}, })); + } return se::make_list(std::move(list)); }); diff --git a/src/cvc/ariadne/conference/room_net.cpp b/src/cvc/ariadne/conference/room_net.cpp new file mode 100644 index 000000000..114d0841b --- /dev/null +++ b/src/cvc/ariadne/conference/room_net.cpp @@ -0,0 +1,146 @@ +/* + Copyright 2026 The University of Texas at Austin + + This file is part of libcvc. + + libcvc is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License version 2.1 as published by the Free Software Foundation. + + libcvc is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. +*/ + +#include +#include +#include // room_participant_path +#include +#include +#include +#include +#include + +namespace cvc { +namespace ariadne { +namespace conference { + +namespace { + +std::string sub(const std::string &base, std::initializer_list segs) { + std::string s = base; + for (const char *seg : segs) { + s += cvc::state::SEPARATOR; + s += seg; + } + return s; +} + +// One decimal is plenty for display, and an unchanged value writes nothing new to replicate. +std::string fmt(double v) { + char buf[32]; + std::snprintf(buf, sizeof(buf), "%.1f", v); + return std::string(buf); +} + +// touch-before-value, so a just-created node's first write is not lost to the distributed-state +// adapter's attach race (the room tree's set_replicated idiom). A value that has not changed is not +// rewritten, so a steady link does not add replication traffic. +void set_replicated(cvc::app &app, const std::string &path, const std::string &val) { + cvc::state &n = app.root()(path); + if (n.value() == val) + return; + n.touch(); + n.value(val); +} + +double read_double(cvc::app &app, const std::string &path, double def) { + cvc::state *n = app.root().findDescendant(path); + return n && !n->value().empty() ? std::strtod(n->value().c_str(), nullptr) : def; +} + +void write_sample(cvc::app &app, const std::string &base, const net_sample &s) { + set_replicated(app, sub(base, {"rtt_ms"}), fmt(s.rtt_ms)); + set_replicated(app, sub(base, {"in_kbps"}), fmt(s.in_kbps)); + set_replicated(app, sub(base, {"out_kbps"}), fmt(s.out_kbps)); +} + +bool read_sample(cvc::app &app, const std::string &base, net_sample &out) { + if (!app.root().findDescendant(sub(base, {"rtt_ms"}))) + return false; + out.rtt_ms = read_double(app, sub(base, {"rtt_ms"}), -1.0); + out.in_kbps = read_double(app, sub(base, {"in_kbps"}), -1.0); + out.out_kbps = read_double(app, sub(base, {"out_kbps"}), -1.0); + return true; +} + +std::string participant_net(const std::string &room, const std::string &pid) { + return sub(room_participant_path(room, pid), {"net"}); +} + +std::string server_net_path(const std::string &room) { + return sub("conf", {"rooms", room.c_str(), "server", "net"}); +} + +} // namespace + +void write_participant_net(cvc::app &app, const std::string &room, const std::string &pid, + const net_sample &s) { + write_sample(app, participant_net(room, pid), s); +} + +bool read_participant_net(cvc::app &app, const std::string &room, const std::string &pid, + net_sample &out) { + return read_sample(app, participant_net(room, pid), out); +} + +void write_relay_net(cvc::app &app, const std::string &room, const std::string &pid, + const net_sample &s) { + write_sample(app, sub(participant_net(room, pid), {"relay"}), s); +} + +bool read_relay_net(cvc::app &app, const std::string &room, const std::string &pid, + net_sample &out) { + return read_sample(app, sub(participant_net(room, pid), {"relay"}), out); +} + +void write_server_net(cvc::app &app, const std::string &room, const server_net &s) { + const std::string base = server_net_path(room); + set_replicated(app, sub(base, {"in_kbps"}), fmt(s.in_kbps)); + set_replicated(app, sub(base, {"out_kbps"}), fmt(s.out_kbps)); + set_replicated(app, sub(base, {"links"}), std::to_string(s.links)); +} + +bool read_server_net(cvc::app &app, const std::string &room, server_net &out) { + const std::string base = server_net_path(room); + if (!app.root().findDescendant(sub(base, {"links"}))) + return false; + out.in_kbps = read_double(app, sub(base, {"in_kbps"}), 0.0); + out.out_kbps = read_double(app, sub(base, {"out_kbps"}), 0.0); + out.links = static_cast(read_double(app, sub(base, {"links"}), 0.0)); + return true; +} + +std::string format_rtt(double rtt_ms) { + if (rtt_ms < 0.0) + return "--"; + char buf[32]; + std::snprintf(buf, sizeof(buf), "%.0f ms", rtt_ms); + return std::string(buf); +} + +std::string format_rate(double kbps) { + if (kbps < 0.0) + return "--"; + char buf[32]; + if (kbps >= 1000.0) + std::snprintf(buf, sizeof(buf), "%.1fM", kbps / 1000.0); + else + std::snprintf(buf, sizeof(buf), "%.0fk", kbps); + return std::string(buf); +} + +} // namespace conference +} // namespace ariadne +} // namespace cvc diff --git a/src/cvc/ariadne/conference/room_scene.cpp b/src/cvc/ariadne/conference/room_scene.cpp index c2768c0c4..ad2398850 100644 --- a/src/cvc/ariadne/conference/room_scene.cpp +++ b/src/cvc/ariadne/conference/room_scene.cpp @@ -295,7 +295,7 @@ room_scene_env default_scene_env() { env.background_top[0] = 0.20f, env.background_top[1] = 0.24f, env.background_top[2] = 0.30f; env.background_bottom[0] = 0.05f, env.background_bottom[1] = 0.06f, env.background_bottom[2] = 0.08f; - env.chrome = true; + env.chrome = false; // the SceneGraph chrome is a debug overlay (bbox, axes, labels), not decor cvc::ariadne::SceneLight sun; sun.id = "sun"; sun.kind = "directional"; diff --git a/src/cvc/ariadne/holon.cpp b/src/cvc/ariadne/holon.cpp index 05bae69ca..7ca50f2f2 100644 --- a/src/cvc/ariadne/holon.cpp +++ b/src/cvc/ariadne/holon.cpp @@ -53,6 +53,14 @@ bool holon::has_participant(const std::string &pid) const { return _participants.find(pid) != _participants.end(); } +std::vector holon::participants() const { + std::vector out; + out.reserve(_participants.size()); + for (const auto &kv : _participants) + out.push_back(kv.first); + return out; +} + void holon::set_intent(const std::string &pid, const avatar_intent &in) { auto it = _participants.find(pid); if (it == _participants.end()) @@ -70,9 +78,36 @@ void holon::apply_authoritative(const std::string &pid, const cvc::physics::body auto it = _participants.find(pid); if (it == _participants.end()) return; + // Count the server's ticks (see the header): re-align whenever the replica is more + // than max_skew_ticks away -- on the first sample when the client joined a running + // server, and after a stall. A replica already within the skew is left alone, so its + // logged inputs still line up with the server's ticks and replay correctly. + const std::uint64_t now = _clock.tick(); + const std::uint64_t skew = now > server_tick ? now - server_tick : server_tick - now; + if (skew > _cfg.max_skew_ticks) { + _clock.seek_tick(server_tick); + for (auto &kv : _participants) { + participant &q = kv.second; + if (q.local) { + // The logged inputs' ticks are on the old timeline: keep the state only. + q.predicted = predicted_avatar(q.predicted.state()); + } else { + // Shift the walk's start by the same jump, so the peer carries on from where + // it is shown instead of restarting the walk (a visible step back). + q.newest_arrival_tick = server_tick > now + ? q.newest_arrival_tick + (server_tick - now) + : (q.newest_arrival_tick > now - server_tick + ? q.newest_arrival_tick - (now - server_tick) + : 0); + } + } + } + participant &p = it->second; if (p.local) { - p.predicted.reconcile(auth, server_tick, _clock.fixed_dt(), _cfg.reconcile_tol); + const cvc::physics::vec3 before = p.predicted.state().position; + if (p.predicted.reconcile(auth, server_tick, _clock.fixed_dt(), _cfg.reconcile_tol)) + p.correction = p.correction + (p.predicted.state().position - before); } else { // A peer. Reject a stale/duplicate/reordered sample (server_tick <= the newest // one applied), exactly as predicted_avatar does for the local path: otherwise @@ -84,12 +119,21 @@ void holon::apply_authoritative(const std::string &pid, const cvc::physics::body if (!p.have_sample) { p.newest = auth; p.prev = auth; + p.sample_span = 0; p.have_sample = true; } else { - p.prev = p.newest; // shift the window: prev <- old newest, newest <- auth + // Walk on from wherever the peer is SHOWN now (normally the old newest; earlier + // when this sample came before the last walk finished), so a sample that + // arrives early never makes the peer skip forward. + const cvc::physics::pose shown = display_pose(pid); + p.prev = p.newest; + p.prev.position = shown.position; + p.prev.rotation = shown.rotation; p.newest = auth; + p.sample_span = server_tick - p.last_sample_tick; } p.last_sample_tick = server_tick; + p.newest_arrival_tick = _clock.tick(); } } @@ -138,9 +182,25 @@ cvc::physics::pose holon::display_pose(const std::string &pid) const { const cvc::physics::body_state &s = p.predicted.state(); return cvc::physics::pose{s.position, s.rotation}; } - // Peer: interpolate between the last two authoritative samples at this frame's - // alpha (interpolate_peer clamps alpha to [0,1), so it never overshoots). - return interpolate_peer(p.prev, p.newest, _alpha); + // Peer: walk from prev to newest over the server's own interval between them, + // timed on this clock from the newest sample's arrival (interpolate_peer clamps + // the fraction to [0,1), so it never overshoots and holds at newest once done). + if (p.sample_span == 0) + return cvc::physics::pose{p.newest.position, p.newest.rotation}; + const std::uint64_t tick = _clock.tick(); + const double elapsed = + (tick >= p.newest_arrival_tick ? static_cast(tick - p.newest_arrival_tick) : 0.0) + + _alpha; + return interpolate_peer(p.prev, p.newest, elapsed / static_cast(p.sample_span)); +} + +cvc::physics::vec3 holon::take_correction(const std::string &pid) { + auto it = _participants.find(pid); + if (it == _participants.end()) + return cvc::physics::vec3{0.0, 0.0, 0.0}; + const cvc::physics::vec3 c = it->second.correction; + it->second.correction = cvc::physics::vec3{0.0, 0.0, 0.0}; + return c; } } // namespace ariadne diff --git a/src/cvc/ariadne/holon_sync.cpp b/src/cvc/ariadne/holon_sync.cpp index f29479317..bbb91603d 100644 --- a/src/cvc/ariadne/holon_sync.cpp +++ b/src/cvc/ariadne/holon_sync.cpp @@ -13,6 +13,7 @@ Lesser General Public License for more details. */ +#include #include #include #include // room_participant_path, room_roster @@ -22,7 +23,9 @@ #include // vec3, quat #include // cvc::state, SEPARATOR, findDescendant #include +#include #include +#include namespace cvc { namespace ariadne { @@ -130,14 +133,30 @@ bool read_intent(cvc::app &app, const std::string &room, const std::string &pid, void server_pump(holon &h, cvc::app &app, const std::string &room, double wall_dt) { // The present participants, straight from the replicated room tree. const std::vector roster = conference::room_roster(app, room); - // Add any newcomer, then feed each one's latest proposed intent to the holon. + // Add any newcomer -- spawned on a 4 m ring around the room's origin (where the seeded + // reference object stands), each at the next golden-angle step so joiners never stack + // -- then feed each one's latest proposed intent to the holon. for (const conference::room_participant &p : roster) { - if (!h.has_participant(p.pid)) - h.add_participant(p.pid); + if (!h.has_participant(p.pid)) { + constexpr double kSpawnRadius = 4.0; + constexpr double kGoldenAngle = 2.39996322972865332; // radians + const double a = kGoldenAngle * static_cast(h.participants().size()); + cvc::physics::body_state start; + start.position = {kSpawnRadius * std::cos(a), kSpawnRadius * std::sin(a), 0.0}; + h.add_participant(p.pid, /*local=*/false, start); + } avatar_intent in; if (read_intent(app, room, p.pid, in)) h.set_intent(p.pid, in); } + // Drop anyone who left (room_leave): the holon stops integrating them, and a later rejoin under + // the same id respawns them on the ring rather than wherever a stale intent carried them. + std::set present; + for (const conference::room_participant &p : roster) + present.insert(p.pid); + for (const std::string &pid : h.participants()) + if (!present.count(pid)) + h.remove_participant(pid); // One authoritative frame. h.advance(wall_dt); // Publish each participant's authoritative pose (these are local writes on the @@ -146,5 +165,31 @@ void server_pump(holon &h, cvc::app &app, const std::string &room, double wall_d write_pose(app, room, p.pid, h.state(p.pid), h.tick()); } +void client_pump(holon &h, cvc::app &app, const std::string &room, const std::string &self_pid, + double wall_dt) { + const std::vector roster = conference::room_roster(app, room); + std::set present; + for (const conference::room_participant &p : roster) { + present.insert(p.pid); + cvc::physics::body_state s; + std::uint64_t server_tick = 0; + const bool have = read_pose(app, room, p.pid, s, server_tick); + if (!h.has_participant(p.pid)) + h.add_participant(p.pid, /*local=*/p.pid == self_pid, have ? s : cvc::physics::body_state{}); + if (have) + h.apply_authoritative(p.pid, s, server_tick); + } + for (const std::string &pid : h.participants()) + if (!present.count(pid)) + h.remove_participant(pid); + h.advance(wall_dt); +} + +void propose_intent(holon &h, cvc::app &app, const std::string &room, const std::string &self_pid, + const avatar_intent &intent) { + h.set_intent(self_pid, intent); + write_intent(app, room, self_pid, intent); +} + } // namespace ariadne } // namespace cvc diff --git a/src/cvc/state/state_link_stats.cpp b/src/cvc/state/state_link_stats.cpp new file mode 100644 index 000000000..85dc152ae --- /dev/null +++ b/src/cvc/state/state_link_stats.cpp @@ -0,0 +1,102 @@ +/* + Copyright 2026 The University of Texas at Austin + + This file is part of libcvc. + + libcvc is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License version 2.1 as published by the Free Software Foundation. + + libcvc is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. +*/ + +#include +#include + +namespace cvc { + +void state_link_stats::note_rx(std::size_t bytes) noexcept { + _bytes_in.fetch_add(bytes, std::memory_order_relaxed); + _frames_in.fetch_add(1, std::memory_order_relaxed); +} + +void state_link_stats::note_tx(std::size_t bytes) noexcept { + _bytes_out.fetch_add(bytes, std::memory_order_relaxed); + _frames_out.fetch_add(1, std::memory_order_relaxed); +} + +std::uint64_t state_link_stats::next_ping_seq() noexcept { + return _pings.fetch_add(1, std::memory_order_relaxed) + 1; +} + +void state_link_stats::note_rtt(std::int64_t rtt_ns) noexcept { + if (rtt_ns < 0) + return; // a clock that went backwards: not a measurement + const std::int64_t us = rtt_ns / 1000; + _pongs.fetch_add(1, std::memory_order_relaxed); + _rtt_last_us.store(us, std::memory_order_relaxed); + std::int64_t mn = _rtt_min_us.load(std::memory_order_relaxed); + while ((mn < 0 || us < mn) && + !_rtt_min_us.compare_exchange_weak(mn, us, std::memory_order_relaxed)) { + } + std::lock_guard lk(_rtt_mu); + _rtt_ring[_rtt_next] = us; + _rtt_next = (_rtt_next + 1) % kRttWindow; + if (_rtt_count < kRttWindow) + ++_rtt_count; +} + +state_link_stats::counters state_link_stats::read() const noexcept { + counters c; + c.bytes_in = _bytes_in.load(std::memory_order_relaxed); + c.bytes_out = _bytes_out.load(std::memory_order_relaxed); + c.frames_in = _frames_in.load(std::memory_order_relaxed); + c.frames_out = _frames_out.load(std::memory_order_relaxed); + c.pings_sent = _pings.load(std::memory_order_relaxed); + c.pongs_received = _pongs.load(std::memory_order_relaxed); + const auto ms = [](std::int64_t us) { return us < 0 ? -1.0 : static_cast(us) / 1000.0; }; + { + std::lock_guard lk(_rtt_mu); + if (_rtt_count > 0) { + std::array w = _rtt_ring; + std::sort(w.begin(), w.begin() + _rtt_count); + const int n = _rtt_count; + const std::int64_t med = (n % 2) ? w[n / 2] : (w[n / 2 - 1] + w[n / 2]) / 2; + c.rtt_ms = ms(med); + } + } + c.rtt_last_ms = ms(_rtt_last_us.load(std::memory_order_relaxed)); + c.rtt_min_ms = ms(_rtt_min_us.load(std::memory_order_relaxed)); + return c; +} + +std::string state_link_rate_meter::key_of(const state_link_report &r) { + return r.transport + ":" + r.remote_node_id + (r.accepted ? ":in" : ""); +} + +std::map +state_link_rate_meter::update(const std::vector &reports, double now_s) { + std::map out; + std::map next; + for (const state_link_report &r : reports) { + const std::string key = key_of(r); + rate rt; + auto it = _last.find(key); + // A link whose counters went DOWN is a new connection under the same key: start over. + if (it != _last.end() && now_s > it->second.t && r.stats.bytes_in >= it->second.bytes_in && + r.stats.bytes_out >= it->second.bytes_out) { + const double dt = now_s - it->second.t; + rt.in_bps = static_cast(r.stats.bytes_in - it->second.bytes_in) / dt; + rt.out_bps = static_cast(r.stats.bytes_out - it->second.bytes_out) / dt; + } + out[key] = rt; + next[key] = sample{r.stats.bytes_in, r.stats.bytes_out, now_s}; + } + _last.swap(next); // links no longer reported are forgotten + return out; +} + +} // namespace cvc diff --git a/src/cvc/state/state_transport_grpc.cpp b/src/cvc/state/state_transport_grpc.cpp index 8c32fb463..28f814847 100644 --- a/src/cvc/state/state_transport_grpc.cpp +++ b/src/cvc/state/state_transport_grpc.cpp @@ -13,6 +13,7 @@ #include #include +#include // the write watchdog's report #include #include #include @@ -27,6 +28,14 @@ namespace { namespace pb = ::cvc::transport::v1; +// Steady-clock nanoseconds: the link probe's clock (a Ping carries it; the Pong echoes it back to +// the same process, so no cross-host clock sync is involved) and the link report's age / idle base. +inline std::int64_t link_now_ns() { + return std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(); +} + // Largest gRPC message we accept/send, in bytes. The gRPC default receive cap is ~4 MiB, too small // for a large binary message payload (a 1080p RGBA frame is ~8 MiB); match the IPC transport's // 64 MiB frame cap so both wire paths agree on the ceiling. @@ -223,13 +232,48 @@ class state_transport_grpc::connection { std::atomic alive{true}; std::string remote_node_id; std::string remote_cluster_id; + // Link quality (state_link_stats): traffic both ways + the Ping/Pong round trip. `accepted` marks + // the server end of the stream; opened/last-rx are steady-clock ns for the report's age / idle. + state_link_stats link; + bool accepted = false; + std::int64_t opened_ns = 0; + std::atomic last_rx_ns{0}; + // Steady-clock ns when the write in progress started; 0 when none is. A peer that stops reading + // makes a stream Write block on flow control while holding write_mu, so every other writer to it + // would wait forever; the probe thread's watchdog cancels a stream whose write has been blocked + // too long (state_transport_grpc::ping_loop). `cancel` cancels the stream; the server handler + // clears it (under cancel_mu) before its ServerContext goes away. + std::atomic write_started_ns{0}; + std::mutex cancel_mu; + std::function cancel; bool can_write() const { return static_cast(write_raw); } bool write_locked(const pb::Frame &f) { if (!alive.load() || !write_raw) return false; - return write_raw(f); + const std::size_t bytes = f.ByteSizeLong(); + write_started_ns.store(link_now_ns(), std::memory_order_relaxed); + const bool ok = write_raw(f); + write_started_ns.store(0, std::memory_order_relaxed); + if (!ok) + return false; + link.note_tx(bytes); + return true; + } + + // Write only if no other write holds the stream right now (the link probe never waits). + bool try_write(const pb::Frame &f) { + std::unique_lock lk(write_mu, std::try_to_lock); + return lk.owns_lock() && write_locked(f); + } + + // Cancel the stream: a blocked Write fails and frees write_mu (the write watchdog). + void abort_io() { + alive.store(false); + std::lock_guard lk(cancel_mu); + if (cancel) + cancel(); } bool write(const pb::Frame &f) { @@ -270,6 +314,10 @@ class StateTransportServiceImpl final : public pb::StateTransport::Service { } auto conn = std::make_shared(); + conn->accepted = true; + conn->opened_ns = link_now_ns(); + conn->last_rx_ns.store(conn->opened_ns); + conn->cancel = [ctx] { ctx->TryCancel(); }; auto *raw_stream = stream; conn->write_raw = [raw_stream](const pb::Frame &f) -> bool { return raw_stream->Write(f); }; @@ -280,6 +328,8 @@ class StateTransportServiceImpl final : public pb::StateTransport::Service { pb::Frame in; while (!ctx->IsCancelled() && stream->Read(&in)) { _owner->increment_recv_frames(); + if (_owner->on_link_frame(*conn, in)) + continue; if (in.has_hello()) { conn->remote_node_id = in.hello().node_id(); conn->remote_cluster_id = in.hello().cluster_id(); @@ -331,7 +381,12 @@ class StateTransportServiceImpl final : public pb::StateTransport::Service { } conn->alive.store(false); + { + std::lock_guard lk(conn->cancel_mu); + conn->cancel = nullptr; // ctx dies with this handler + } _owner->unregister_connection(conn.get()); + _owner->notify_connection_closed(conn->remote_node_id); return grpc::Status::OK; } @@ -413,6 +468,7 @@ void state_transport_grpc::start(const std::string &listen_addr, const std::stri if (_heartbeat_interval.count() > 0) { _heartbeat_thread = std::thread([this]() { heartbeat_loop(); }); } + _ping_thread = std::thread([this]() { ping_loop(); }); } void state_transport_grpc::set_heartbeat_interval(std::chrono::milliseconds interval) noexcept { @@ -473,10 +529,18 @@ bool state_transport_grpc::connect_to_peer(const std::string &target, // makes it visible to stop(), which joins client_reader — so the thread // has to already be assigned or stop() could miss it and leave a // joinable std::thread to destruct. + conn->opened_ns = link_now_ns(); + conn->last_rx_ns.store(conn->opened_ns); + conn->cancel = [ctx = conn->client_ctx] { + if (ctx) + ctx->TryCancel(); + }; conn->client_reader = std::thread([this, conn]() { pb::Frame in; while (conn->alive.load() && conn->client_stream->Read(&in)) { increment_recv_frames(); + if (on_link_frame(*conn, in)) + continue; if (in.has_hello()) { conn->remote_node_id = in.hello().node_id(); conn->remote_cluster_id = in.hello().cluster_id(); @@ -535,6 +599,12 @@ bool state_transport_grpc::connect_to_peer(const std::string &target, return true; } +void state_transport_grpc::notify_connection_closed(const std::string &node_id) { + // stop() clears _running before it cancels the streams, so a shutdown never reports departures. + if (_running.load() && _on_conn_closed && !node_id.empty()) + _on_conn_closed(node_id); +} + void state_transport_grpc::stop() { if (!_running.exchange(false)) return; @@ -560,6 +630,8 @@ void state_transport_grpc::stop() { if (_heartbeat_thread.joinable()) _heartbeat_thread.join(); + if (_ping_thread.joinable()) + _ping_thread.join(); for (auto &c : conns) { if (c->client_reader.joinable()) @@ -993,6 +1065,92 @@ std::size_t state_transport_grpc::shard_count() const { return _shards.size(); } +std::vector state_transport_grpc::link_reports() const { + std::vector> conns; + { + std::lock_guard lk(_conns_mu); + conns = _conns; + } + const std::int64_t now = link_now_ns(); + std::vector out; + for (const auto &c : conns) { + if (!c || !c->alive.load()) + continue; + state_link_report r; + r.transport = "grpc"; + r.remote_node_id = c->remote_node_id; + r.accepted = c->accepted; + r.age_s = static_cast(now - c->opened_ns) * 1e-9; + r.idle_s = static_cast(now - c->last_rx_ns.load()) * 1e-9; + r.stats = c->link.read(); + out.push_back(std::move(r)); + } + return out; +} + +bool state_transport_grpc::on_link_frame(connection &c, const pb::Frame &in) { + c.link.note_rx(in.ByteSizeLong()); + c.last_rx_ns.store(link_now_ns(), std::memory_order_relaxed); + if (in.has_ping()) { + pb::Frame f; + f.mutable_pong()->set_seq(in.ping().seq()); + f.mutable_pong()->set_sent_ns(in.ping().sent_ns()); + if (c.write(f)) + _sent_frames.fetch_add(1, std::memory_order_relaxed); + return true; + } + if (in.has_pong()) { + c.link.note_rtt(link_now_ns() - in.pong().sent_ns()); + return true; + } + return false; +} + +void state_transport_grpc::send_pings() { + std::vector> conns; + { + std::lock_guard lk(_conns_mu); + conns = _conns; + } + for (auto &c : conns) { + if (!c || !c->alive.load() || !c->can_write()) + continue; + pb::Frame f; + f.mutable_ping()->set_seq(c->link.next_ping_seq()); + f.mutable_ping()->set_sent_ns(link_now_ns()); + if (c->try_write(f)) + _sent_frames.fetch_add(1, std::memory_order_relaxed); + } +} + +void state_transport_grpc::ping_loop() { + // How long one write may block before its stream is cancelled (connection::write_started_ns). + constexpr std::int64_t kWriteStallNs = 3'000'000'000; + auto next = std::chrono::steady_clock::now(); + while (_running.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + std::vector> conns; + { + std::lock_guard lk(_conns_mu); + conns = _conns; + } + const std::int64_t now = link_now_ns(); + for (auto &c : conns) { + const std::int64_t started = c ? c->write_started_ns.load(std::memory_order_relaxed) : 0; + if (started != 0 && now - started > kWriteStallNs && c->alive.load()) { + std::fprintf(stderr, "[grpc] write to '%s' blocked > 3 s (peer not reading): closing\n", + c->remote_node_id.c_str()); + c->abort_io(); + } + } + const auto interval = ping_interval(); + if (interval.count() <= 0 || std::chrono::steady_clock::now() < next) + continue; + next = std::chrono::steady_clock::now() + interval; + send_pings(); + } +} + std::size_t state_transport_grpc::connection_count() const { std::lock_guard lk(_conns_mu); std::size_t n = 0; diff --git a/src/cvc/state/state_transport_ws.cpp b/src/cvc/state/state_transport_ws.cpp index dddaf4d35..1bb191225 100644 --- a/src/cvc/state/state_transport_ws.cpp +++ b/src/cvc/state/state_transport_ws.cpp @@ -236,6 +236,17 @@ class state_transport_ws::connection { // reaper uses it to drop the ghost. 0 until the first frame; set to "now" when a connection opens // so a brand-new, not-yet-speaking client is not reaped. std::atomic last_rx_ns{0}; + // Link quality (state_link_stats): traffic both ways + the Ping/Pong round trip. `accepted` marks + // the server end; opened_ns (steady-clock) is the report's age base, set at registration. + state_link_stats link; + bool accepted = false; + std::int64_t opened_ns = 0; + // Steady-clock ns when the write in progress started; 0 when none is. A peer that stops reading + // (a frozen tab, a dead network) makes a write block once its TCP buffer is full -- while holding + // write_mu, so every other writer to it (the relay's main loop included) would wait forever. The + // probe thread's watchdog closes a connection whose write has been blocked too long, which fails + // the write and frees the lock (state_transport_ws::ping_loop). + std::atomic write_started_ns{0}; #ifndef __EMSCRIPTEN__ std::unique_ptr> ws; @@ -253,13 +264,28 @@ class state_transport_ws::connection { return false; beast::error_code ec; ws->binary(true); + write_started_ns.store(std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()) + .count(), + std::memory_order_relaxed); ws->write(net::buffer(bytes), ec); + write_started_ns.store(0, std::memory_order_relaxed); if (ec) { alive.store(false); return false; } + link.note_tx(bytes.size()); return true; } + + // Unblock any read or write in progress and mark the connection dead (the write watchdog). + void abort_io() { + alive.store(false); + if (ws) { + beast::error_code ec; + ws->next_layer().shutdown(tcp::socket::shutdown_both, ec); + } + } #else // Browser WebSocket: the handle, a back-pointer to the owning transport (for the C callbacks), // and an `open` flag — the browser cannot send until onopen has fired. There is no reader thread; @@ -276,8 +302,11 @@ class state_transport_ws::connection { std::string bytes; if (!f.SerializeToString(&bytes)) return false; - return emscripten_websocket_send_binary(em_socket, const_cast(bytes.data()), - static_cast(bytes.size())) >= 0; + if (emscripten_websocket_send_binary(em_socket, const_cast(bytes.data()), + static_cast(bytes.size())) < 0) + return false; + link.note_tx(bytes.size()); + return true; } #endif @@ -286,6 +315,12 @@ class state_transport_ws::connection { return write_locked(f); } + // Write only if no other write holds the connection right now (the link probe never waits). + bool try_write(const pb::Frame &f) { + std::unique_lock lk(write_mu, std::try_to_lock); + return lk.owns_lock() && write_locked(f); + } + // Bytes still unsent in this connection's OS send buffer (Linux only; 0 elsewhere). A large value // means the peer is not draining fast enough and the next write would BLOCK. The fan paths check // this before writing and skip a backed-up connection — dropping a loss-tolerant media frame, or @@ -311,6 +346,7 @@ struct state_transport_ws::impl { net::io_context ioc; std::unique_ptr acceptor; std::thread accept_thread; + std::thread ping_thread; // link probe (ping_loop), joined by stop() std::string listen_addr_resolved; }; #else @@ -396,6 +432,8 @@ void state_transport_ws::start(const std::string &listen_addr, const std::string _running.store(true); _impl->accept_thread = std::thread([this]() { accept_loop(); }); + if (!_ping_started.exchange(true)) + _impl->ping_thread = std::thread([this]() { ping_loop(); }); } // Server accept loop: synchronous accept → WebSocket handshake → admit + spawn reader. Breaks when @@ -411,6 +449,7 @@ void state_transport_ws::accept_loop() { break; // woken by stop()'s loopback self-connect auto conn = std::make_shared(); + conn->accepted = true; conn->ws = std::make_unique>(std::move(sock)); conn->ws->read_message_max(kMaxWsMessageBytes); conn->ws->accept(ec); // server-side WebSocket handshake @@ -457,6 +496,8 @@ bool state_transport_ws::connect_to_peer(const std::string &url, conn->reader = std::thread([this, conn]() { read_loop(conn); }); _running.store(true); // so stop() tears the client connection down admit_connection(conn); + if (!_ping_started.exchange(true)) + _impl->ping_thread = std::thread([this]() { ping_loop(); }); return true; } @@ -524,6 +565,9 @@ void state_transport_ws::stop() { if (_impl->accept_thread.joinable()) _impl->accept_thread.join(); + if (_impl->ping_thread.joinable()) + _impl->ping_thread.join(); + _ping_started.store(false); for (auto &c : conns) if (c->reader.joinable()) c->reader.join(); @@ -651,9 +695,24 @@ static std::int64_t steady_ns() { void state_transport_ws::deliver_inbound(connection *c, const void *data, std::size_t len) { increment_recv_frames(); c->last_rx_ns.store(steady_ns(), std::memory_order_relaxed); // liveness for the reaper + c->link.note_rx(len); pb::Frame in; if (!in.ParseFromArray(data, static_cast(len))) return; + // Link probe: answer a Ping at once (in a browser this runs in the WebSocket onmessage event, + // which a backgrounded tab still receives), and measure a Pong against our own clock. + if (in.has_ping()) { + pb::Frame f; + f.mutable_pong()->set_seq(in.ping().seq()); + f.mutable_pong()->set_sent_ns(in.ping().sent_ns()); + if (c->write(f)) + _sent_frames.fetch_add(1, std::memory_order_relaxed); + return; + } + if (in.has_pong()) { + c->link.note_rtt(steady_ns() - in.pong().sent_ns()); + return; + } if (in.has_hello()) { c->remote_node_id = in.hello().node_id(); c->remote_cluster_id = in.hello().cluster_id(); @@ -955,6 +1014,9 @@ std::size_t state_transport_ws::pump_shard(state_cluster_shard &shard) { } std::size_t state_transport_ws::pump_all() { +#ifdef __EMSCRIPTEN__ + send_pings_if_due(); // no ping thread in a browser: the frame pump drives the probe +#endif std::vector snap; { std::lock_guard lk(_shards_mu); @@ -1069,7 +1131,86 @@ void state_transport_ws::on_inbound_message(const state_message &m, connection * _bridge->publish_message(m); } +std::vector state_transport_ws::link_reports() const { + std::vector> conns; + { + std::lock_guard lk(_conns_mu); + conns = _conns; + } + const std::int64_t now = steady_ns(); + std::vector out; + for (const auto &c : conns) { + if (!c || !c->alive.load() || !c->can_write()) + continue; + state_link_report r; + r.transport = "ws"; + r.remote_node_id = c->remote_node_id; + r.accepted = c->accepted; + r.age_s = static_cast(now - c->opened_ns) * 1e-9; + r.idle_s = static_cast(now - c->last_rx_ns.load()) * 1e-9; + r.stats = c->link.read(); + out.push_back(std::move(r)); + } + return out; +} + +void state_transport_ws::send_pings_if_due() { + const auto interval = ping_interval(); + if (interval.count() <= 0) + return; + const std::int64_t now = steady_ns(); + std::int64_t due = _next_ping_ns.load(std::memory_order_relaxed); + if (now < due || + !_next_ping_ns.compare_exchange_strong( + due, now + std::chrono::duration_cast(interval).count())) + return; + std::vector> conns; + { + std::lock_guard lk(_conns_mu); + conns = _conns; + } + for (auto &c : conns) { + // Skip a connection that cannot send yet, and one whose send buffer is badly backed up (the + // write would block the prober behind it; its RTT would only measure the backlog anyway). + if (!c || !c->alive.load() || !c->can_write() || c->send_backlog() > (64u << 10)) + continue; + pb::Frame f; + f.mutable_ping()->set_seq(c->link.next_ping_seq()); + f.mutable_ping()->set_sent_ns(steady_ns()); + if (c->try_write(f)) + _sent_frames.fetch_add(1, std::memory_order_relaxed); + } +} + +void state_transport_ws::ping_loop() { +#ifndef __EMSCRIPTEN__ + // How long one write may block before its connection is closed (see + // connection::write_started_ns). + constexpr std::int64_t kWriteStallNs = 3'000'000'000; + while (_running.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + send_pings_if_due(); + std::vector> conns; + { + std::lock_guard lk(_conns_mu); + conns = _conns; + } + const std::int64_t now = steady_ns(); + for (auto &c : conns) { + const std::int64_t started = c ? c->write_started_ns.load(std::memory_order_relaxed) : 0; + if (started != 0 && now - started > kWriteStallNs && c->alive.load()) { + std::fprintf(stderr, "[ws] write to '%s' blocked > 3 s (peer not reading): closing\n", + c->remote_node_id.c_str()); + c->abort_io(); + } + } + } +#endif +} + void state_transport_ws::register_connection(std::shared_ptr conn) { + if (conn->opened_ns == 0) + conn->opened_ns = steady_ns(); std::lock_guard lk(_conns_mu); _conns.push_back(std::move(conn)); } diff --git a/src/cvc/tests/CMakeLists.txt b/src/cvc/tests/CMakeLists.txt index ff3204d7a..2b1e42f2b 100644 --- a/src/cvc/tests/CMakeLists.txt +++ b/src/cvc/tests/CMakeLists.txt @@ -110,6 +110,7 @@ add_executable(physics_world_test physics_world_test.cpp) add_executable(avatar_sim_test avatar_sim_test.cpp) add_executable(holon_test holon_test.cpp) add_executable(holon_sync_test holon_sync_test.cpp) +add_executable(state_link_stats_test state_link_stats_test.cpp) add_executable(room_scene_test room_scene_test.cpp) add_executable(physics_codec_test physics_codec_test.cpp) add_executable(physics_field_test physics_field_test.cpp) @@ -313,6 +314,7 @@ set(TEST_TARGETS avatar_sim_test holon_test holon_sync_test + state_link_stats_test room_scene_test physics_codec_test physics_field_test @@ -1689,6 +1691,13 @@ target_link_libraries(holon_sync_test GTest::gtest_main ) +target_link_libraries(state_link_stats_test + PRIVATE + cvc + GTest::gtest + GTest::gtest_main +) + target_link_libraries(room_scene_test PRIVATE cvc @@ -2106,6 +2115,7 @@ target_compile_features(physics_world_test PRIVATE cxx_std_17) target_compile_features(avatar_sim_test PRIVATE cxx_std_17) target_compile_features(holon_test PRIVATE cxx_std_17) target_compile_features(holon_sync_test PRIVATE cxx_std_17) +target_compile_features(state_link_stats_test PRIVATE cxx_std_17) target_compile_features(room_scene_test PRIVATE cxx_std_17) target_compile_features(physics_codec_test PRIVATE cxx_std_17) target_compile_features(physics_field_test PRIVATE cxx_std_17) @@ -2568,6 +2578,7 @@ cvc_discover_tests(physics_world_test) cvc_discover_tests(avatar_sim_test) cvc_discover_tests(holon_test) cvc_discover_tests(holon_sync_test) +cvc_discover_tests(state_link_stats_test) cvc_discover_tests(room_scene_test) cvc_discover_tests(physics_codec_test) cvc_discover_tests(physics_field_test) diff --git a/src/cvc/tests/holon_sync_test.cpp b/src/cvc/tests/holon_sync_test.cpp index 8caadd29d..ff0268473 100644 --- a/src/cvc/tests/holon_sync_test.cpp +++ b/src/cvc/tests/holon_sync_test.cpp @@ -16,6 +16,7 @@ // C++ verb + arisrv hosting are the live-integration follow-on): just the mapping // and that it rides the existing conf.rooms.* replication. +#include #include #include #include @@ -166,7 +167,15 @@ TEST(HolonSyncTest, ServerPumpIntegratesRosterFromIntents) { holon server(holon_role::server); const double dt = server.clock().fixed_dt(); - for (int frame = 0; frame < 5; ++frame) { + // The first pump registers both newcomers at their spawn points (then steps one quantum). + cvc::ariadne::server_pump(server, server_app, room, dt); + tp.pump_all(); + const vec3 a0 = server.state("A").position, b0 = server.state("B").position; + // Spawned on the 4 m ring around the origin, apart from each other. + EXPECT_NEAR(std::hypot(a0.x, a0.y), 4.0, 1.0 * dt + 1e-9); + EXPECT_NEAR(std::hypot(b0.x, b0.y), 4.0, 2.0 * dt + 1e-9); + EXPECT_GT(std::hypot(a0.x - b0.x, a0.y - b0.y), 1.0) << "joiners never stack"; + for (int frame = 1; frame < 5; ++frame) { cvc::ariadne::server_pump(server, server_app, room, dt); tp.pump_all(); } @@ -177,10 +186,10 @@ TEST(HolonSyncTest, ServerPumpIntegratesRosterFromIntents) { std::uint64_t ta = 0, tb = 0; ASSERT_TRUE(cvc::ariadne::read_pose(server_app, room, "A", pa, ta)); ASSERT_TRUE(cvc::ariadne::read_pose(server_app, room, "B", pb, tb)); - EXPECT_GT(pa.position.x, 0.0) << "A moved +x per its intent"; - EXPECT_DOUBLE_EQ(pa.position.y, 0.0); - EXPECT_GT(pb.position.y, 0.0) << "B moved +y per its intent"; - EXPECT_DOUBLE_EQ(pb.position.x, 0.0); + EXPECT_GT(pa.position.x, a0.x) << "A moved +x per its intent"; + EXPECT_DOUBLE_EQ(pa.position.y, a0.y); + EXPECT_GT(pb.position.y, b0.y) << "B moved +y per its intent"; + EXPECT_DOUBLE_EQ(pb.position.x, b0.x); EXPECT_EQ(ta, server.tick()); // And the authoritative poses replicate back to the client. @@ -195,3 +204,72 @@ TEST(HolonSyncTest, ServerPumpIntegratesRosterFromIntents) { ss.detach(); cs.detach(); } + +// client_pump end to end, timed like the live relay: clients pump every quantum, the +// server every 24 (~5 Hz). Client A predicts its own avatar and stays near the +// authoritative one; client B sees A as a peer that moves forward smoothly. +TEST(HolonSyncTest, ClientPumpPredictsSelfAndWalksPeersSmoothly) { + cvc::app server_app, a_app, b_app; + cvc::state_transport_inproc tp; + cvc::state_cluster_shard ss(server_app, "conf", "server"); + cvc::state_cluster_shard as(a_app, "conf", "A"); + cvc::state_cluster_shard bs(b_app, "conf", "B"); + for (cvc::state_cluster_shard *s : {&ss, &as, &bs}) { + s->attach(); + tp.register_shard(s); + } + const std::string room = "r"; + cvc::ariadne::conference::room_join(a_app, room, "A"); + cvc::ariadne::conference::room_join(b_app, room, "B"); + tp.pump_all(); + + holon server(holon_role::server); + holon a(holon_role::client), b(holon_role::client); + const double dt = server.clock().fixed_dt(); + constexpr int kServerEvery = 24; + + cvc::ariadne::client_pump(a, a_app, room, "A", dt); // registers A (local) + B (peer) + cvc::ariadne::propose_intent(a, a_app, room, "A", avatar_intent{vec3{2.0, 0.0, 0.0}}); + tp.pump_all(); + + double last_peer_x = -1.0; + for (int frame = 1; frame <= 10 * kServerEvery; ++frame) { + if (frame % kServerEvery == 0) + cvc::ariadne::server_pump(server, server_app, room, kServerEvery * dt); + tp.pump_all(); + cvc::ariadne::client_pump(a, a_app, room, "A", dt); + cvc::ariadne::client_pump(b, b_app, room, "B", dt); + ASSERT_TRUE(b.has_participant("A")); + const double peer_x = b.display_pose("A").position.x; + EXPECT_GE(peer_x, last_peer_x) << "frame " << frame << ": B never sees A step back"; + last_peer_x = peer_x; + } + + const double server_x = server.state("A").position.x; + EXPECT_GT(server_x, 3.9) + << "the server moved A per its intent (2 m/s for 2 s), no dropped quanta"; + // A's own prediction stays within a relay interval of the authority. + EXPECT_NEAR(a.display_pose("A").position.x, server_x, 2.0 * kServerEvery * dt + 1e-9); + // B shows A at most one sample interval behind the authority. + EXPECT_NEAR(last_peer_x, server_x, 2.0 * kServerEvery * dt + 1e-9); + // Both replicas count the server's ticks. + EXPECT_LE(a.tick() > server.tick() ? a.tick() - server.tick() : server.tick() - a.tick(), + static_cast(kServerEvery)); + EXPECT_LE(b.tick() > server.tick() ? b.tick() - server.tick() : server.tick() - b.tick(), + static_cast(kServerEvery)); + + // A participant who leaves is dropped from the replicas, and from the server holon. + ASSERT_TRUE(server.has_participant("B")); + cvc::ariadne::conference::room_leave(b_app, room, "B"); + tp.pump_all(); + cvc::ariadne::client_pump(a, a_app, room, "A", dt); + EXPECT_FALSE(a.has_participant("B")); + cvc::ariadne::server_pump(server, server_app, room, dt); + EXPECT_FALSE(server.has_participant("B")) << "the server stops integrating a departed avatar"; + EXPECT_TRUE(server.has_participant("A")); + + for (cvc::state_cluster_shard *s : {&ss, &as, &bs}) { + tp.unregister_shard(s); + s->detach(); + } +} diff --git a/src/cvc/tests/holon_test.cpp b/src/cvc/tests/holon_test.cpp index d6c176cdc..8409eb3c0 100644 --- a/src/cvc/tests/holon_test.cpp +++ b/src/cvc/tests/holon_test.cpp @@ -27,6 +27,7 @@ using cvc::ariadne::avatar_intent; using cvc::ariadne::avatar_sim; using cvc::ariadne::holon; +using cvc::ariadne::holon_config; using cvc::ariadne::holon_role; using cvc::physics::body_state; using cvc::physics::vec3; @@ -197,3 +198,109 @@ TEST(HolonTest, PeerInterpolatesThroughFrameAlpha) { EXPECT_LT(x_near, 10.0) << "peer interpolation never overshoots newest"; EXPECT_GT(x_near, 9.0) << "and it did advance most of the way (not frozen)"; } + +// ── a peer walks smoothly across SPARSE samples (the live ~5 Hz relay) ────── +// Samples arrive every 24 quanta (5 Hz on a 120 Hz clock). The peer must walk from +// one to the next over those 24 quanta: monotonic, never past the newest sample, +// and actually moving (not held at the sample or flicking between the two). + +TEST(HolonTest, PeerWalksAcrossSparseSamples) { + holon client(holon_role::client); + client.add_participant("B", /*local=*/false); + const double dt = client.clock().fixed_dt(); + constexpr int kSpan = 24; + + double last_x = -1.0; + int moved_frames = 0; + for (int k = 0; k <= 4; ++k) { + body_state s; + s.position = vec3{static_cast(k), 0.0, 0.0}; // 1 m per sample + client.apply_authoritative("B", s, 100 + static_cast(k * kSpan)); + for (int f = 0; f < kSpan; ++f) { + client.advance(dt); + const double x = client.display_pose("B").position.x; + EXPECT_GE(x, last_x) << "sample " << k << " frame " << f << ": the peer never steps back"; + EXPECT_LE(x, static_cast(k)) << "never past the newest sample"; + if (x > last_x) + ++moved_frames; + last_x = x; + } + } + EXPECT_GT(last_x, 3.9) << "it walked (nearly) all the way to the last sample"; + EXPECT_GT(moved_frames, 4 * (kSpan - 2)) << "it moved on (nearly) every frame of each walk"; +} + +// ── a client replica counts the server's ticks ────────────────────────────── + +TEST(HolonTest, ClientRealignsItsClockWhenSkewed) { + holon_config cfg; + cfg.max_skew_ticks = 60; + holon client(holon_role::client, cfg); + client.add_participant("A", /*local=*/true); + client.add_participant("B", /*local=*/false); + const double dt = client.clock().fixed_dt(); + for (int i = 0; i < 500; ++i) // the replica has run on its own for a while + client.advance(dt); + ASSERT_EQ(client.tick(), 500u); + + // Joining a server whose clock is elsewhere: the first sample aligns to its tick. + body_state s; + s.position = vec3{3.0, 0.0, 0.0}; + client.apply_authoritative("A", s, 12); + EXPECT_EQ(client.tick(), 12u); + EXPECT_DOUBLE_EQ(client.state("A").position.x, 3.0) << "the local avatar adopts the sample"; + + // Within max_skew_ticks the clocks are left alone (they run at the same rate). + body_state b0, b1; + b1.position = vec3{1.0, 0.0, 0.0}; + client.apply_authoritative("B", b0, 12); + client.apply_authoritative("B", b1, 36); // B walks 0 -> 1 over 24 ticks + for (int i = 0; i < 10; ++i) + client.advance(dt); + client.apply_authoritative("A", s, 40); // 18 ticks away: no re-alignment + EXPECT_EQ(client.tick(), 22u); + const double b_mid = client.display_pose("B").position.x; + EXPECT_GT(b_mid, 0.3); + EXPECT_LT(b_mid, 0.6); + + // Past it (a stalled frame loop), the clock re-aligns -- and the peer's walk carries on + // from where it was shown rather than restarting. + client.apply_authoritative("A", s, 400); + EXPECT_EQ(client.tick(), 400u); + EXPECT_NEAR(client.display_pose("B").position.x, b_mid, 1e-12); +} + +// ── a reconcile jump is reported once, for the renderer to ease out ───────── + +TEST(HolonTest, TakeCorrectionReportsReconcileJumpsOnce) { + holon client(holon_role::client); + client.add_participant("A", /*local=*/true); + client.add_participant("B", /*local=*/false); + const double dt = client.clock().fixed_dt(); + for (int i = 0; i < 3; ++i) { + client.set_intent("A", avatar_intent{vec3{1.0, 0.0, 0.0}}); + client.advance(dt); + } + const vec3 before = client.state("A").position; + + // The prediction holds: no correction. + body_state held; + held.position = vec3{2.0 * dt, 0.0, 0.0}; // what A predicted at tick 2 + held.linear_velocity = vec3{1.0, 0.0, 0.0}; + client.apply_authoritative("A", held, 2); + const vec3 none = client.take_correction("A"); + EXPECT_DOUBLE_EQ(none.x, 0.0); + EXPECT_DOUBLE_EQ(none.y, 0.0); + + // The server overrides: the jump is reported, exactly, once. + body_state moved; + moved.position = vec3{0.0, 5.0, 0.0}; + client.apply_authoritative("A", moved, 3); + const vec3 after = client.state("A").position; + const vec3 c = client.take_correction("A"); + EXPECT_DOUBLE_EQ(c.x, after.x - before.x); + EXPECT_DOUBLE_EQ(c.y, after.y - before.y); + EXPECT_DOUBLE_EQ(client.take_correction("A").y, 0.0) << "taken, so cleared"; + EXPECT_DOUBLE_EQ(client.take_correction("B").y, 0.0) << "a peer has no corrections"; + EXPECT_DOUBLE_EQ(client.take_correction("nobody").y, 0.0); +} diff --git a/src/cvc/tests/room_scene_test.cpp b/src/cvc/tests/room_scene_test.cpp index bb0b98307..8fef743a1 100644 --- a/src/cvc/tests/room_scene_test.cpp +++ b/src/cvc/tests/room_scene_test.cpp @@ -238,7 +238,7 @@ TEST(RoomSceneTest, ReadRoomSceneIsARealizableAriadneScene) { ASSERT_EQ(s.lights.size(), 1u); EXPECT_EQ(s.lights[0].id, "sun"); EXPECT_TRUE(s.has_chrome); - EXPECT_TRUE(s.chrome_visible); + EXPECT_FALSE(s.chrome_visible) << "the debug overlay is off in a conference room"; EXPECT_TRUE(s.has_background); EXPECT_TRUE(s.background_gradient); } diff --git a/src/cvc/tests/state_link_stats_test.cpp b/src/cvc/tests/state_link_stats_test.cpp new file mode 100644 index 000000000..207c9116e --- /dev/null +++ b/src/cvc/tests/state_link_stats_test.cpp @@ -0,0 +1,98 @@ +/* + Copyright 2026 The University of Texas at Austin + + This file is part of libcvc. + + libcvc is free software; you can redistribute it and/or + modify it under the terms of the GNU Lesser General Public + License version 2.1 as published by the Free Software Foundation. + + libcvc is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + Lesser General Public License for more details. +*/ + +// cvc::state_link_stats / state_link_rate_meter: the per-connection counters and round-trip math +// behind the transports' link reports, without a network. + +#include +#include +#include + +using cvc::state_link_rate_meter; +using cvc::state_link_report; +using cvc::state_link_stats; + +TEST(StateLinkStatsTest, CountsTrafficBothWays) { + state_link_stats s; + s.note_rx(100); + s.note_rx(50); + s.note_tx(7); + const auto c = s.read(); + EXPECT_EQ(c.bytes_in, 150u); + EXPECT_EQ(c.frames_in, 2u); + EXPECT_EQ(c.bytes_out, 7u); + EXPECT_EQ(c.frames_out, 1u); +} + +TEST(StateLinkStatsTest, RoundTripMedianMinAndLast) { + state_link_stats s; + EXPECT_LT(s.read().rtt_ms, 0.0) << "no measurement before the first Pong"; + EXPECT_EQ(s.next_ping_seq(), 1u); + EXPECT_EQ(s.next_ping_seq(), 2u); + s.note_rtt(40'000'000); // 40 ms + s.note_rtt(80'000'000); // 80 ms: median of two = their mean, 60 ms + s.note_rtt(-5); // a clock that went backwards is ignored + auto c = s.read(); + EXPECT_EQ(c.pings_sent, 2u); + EXPECT_EQ(c.pongs_received, 2u); + EXPECT_DOUBLE_EQ(c.rtt_ms, 60.0); + EXPECT_DOUBLE_EQ(c.rtt_last_ms, 80.0); + EXPECT_DOUBLE_EQ(c.rtt_min_ms, 40.0); +} + +// One stall (a probe answered only after a frozen tab woke) barely moves the median, and leaves the +// window entirely after kRttWindow more samples. +TEST(StateLinkStatsTest, MedianShrugsOffAStall) { + state_link_stats s; + for (int i = 0; i < 4; ++i) + s.note_rtt(50'000'000); + s.note_rtt(14'800'000'000); // a 14.8 s stall + for (int i = 0; i < 3; ++i) + s.note_rtt(50'000'000); + EXPECT_DOUBLE_EQ(s.read().rtt_ms, 50.0) << "7 of 8 recent samples are 50 ms"; + EXPECT_DOUBLE_EQ(s.read().rtt_last_ms, 50.0); +} + +TEST(StateLinkStatsTest, RateMeterTurnsCountersIntoRates) { + state_link_rate_meter m; + state_link_report r; + r.transport = "ws"; + r.remote_node_id = "alice"; + r.accepted = true; + r.stats.bytes_in = 1000; + r.stats.bytes_out = 4000; + const std::string key = state_link_rate_meter::key_of(r); + EXPECT_EQ(key, "ws:alice:in"); + + auto first = m.update({r}, 10.0); + EXPECT_DOUBLE_EQ(first[key].in_bps, 0.0) << "no interval yet"; + + r.stats.bytes_in = 3000; // +2000 in 2 s + r.stats.bytes_out = 4400; // +400 in 2 s + auto second = m.update({r}, 12.0); + EXPECT_DOUBLE_EQ(second[key].in_bps, 1000.0); + EXPECT_DOUBLE_EQ(second[key].out_bps, 200.0); + + // A reconnect under the same key restarts its counters: no negative or bogus rate. + r.stats.bytes_in = 10; + r.stats.bytes_out = 10; + auto third = m.update({r}, 13.0); + EXPECT_DOUBLE_EQ(third[key].in_bps, 0.0); + EXPECT_DOUBLE_EQ(third[key].out_bps, 0.0); + + // A link that is no longer reported is forgotten. + auto gone = m.update({}, 14.0); + EXPECT_TRUE(gone.empty()); +} diff --git a/src/cvc/tests/state_transport_grpc_test.cpp b/src/cvc/tests/state_transport_grpc_test.cpp index 4a279d002..32bc51009 100644 --- a/src/cvc/tests/state_transport_grpc_test.cpp +++ b/src/cvc/tests/state_transport_grpc_test.cpp @@ -529,6 +529,84 @@ TEST(StateTransportGrpcTest, BlobPayloadRoundTrip) { tB.stop(); } +// A relay learns when a peer's stream ends, with the peer's node id (the WS transport's contract), +// so it can retract a departed native client from the room; its own shutdown reports nothing. +TEST(StateTransportGrpcTest, ConnectionClosedReportsThePeersNodeId) { + cvc::state_transport_grpc relay, client; + std::mutex mu; + std::vector closed; + relay.set_on_connection_closed([&](const std::string &node_id) { + std::lock_guard lk(mu); + closed.push_back(node_id); + }); + relay.start("127.0.0.1:0", "relay", "C"); + client.start("127.0.0.1:0", "alice", "C"); + ASSERT_TRUE(client.connect_to_peer(relay.listen_address(), std::chrono::milliseconds(2000))); + ASSERT_TRUE(wait_connected(client, relay, std::chrono::milliseconds(2000))); + // Leave only once the relay has READ alice's Hello: a cancelled stream drops frames the relay has + // not read yet, and a stream that never identified itself is (by contract) not reported. + const auto hello_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + for (;;) { + const auto links = relay.link_reports(); + if (!links.empty() && links[0].remote_node_id == "alice") + break; + ASSERT_LT(std::chrono::steady_clock::now(), hello_deadline) << "the relay never saw alice"; + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } + + client.stop(); // alice leaves + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + for (;;) { + { + std::lock_guard lk(mu); + if (!closed.empty()) + break; + } + ASSERT_LT(std::chrono::steady_clock::now(), deadline) << "no connection-closed report"; + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + relay.stop(); + std::lock_guard lk(mu); + ASSERT_EQ(closed.size(), 1u) << "reported once, and not again on the relay's own stop()"; + EXPECT_EQ(closed[0], "alice"); +} + +// The link probe: both ends Ping, the other end Pongs, and each measures the round trip on its own +// clock. Each side reports one link with its peer's node id, traffic both ways, and an RTT -- the +// same report a relay gets for a client and a server gets for a peer (delegation) server. +TEST(StateTransportGrpcTest, LinkProbeMeasuresRoundTripAtBothEnds) { + cvc::state_transport_grpc relay, client; + relay.set_ping_interval(std::chrono::milliseconds(30)); + client.set_ping_interval(std::chrono::milliseconds(30)); + relay.start("127.0.0.1:0", "relay", "C"); + client.start("127.0.0.1:0", "alice", "C"); + ASSERT_TRUE(client.connect_to_peer(relay.listen_address(), std::chrono::milliseconds(2000))); + ASSERT_TRUE(wait_connected(client, relay, std::chrono::milliseconds(2000))); + const auto measured = [](const cvc::state_transport_grpc &t) { + const auto links = t.link_reports(); + return links.size() == 1 && links[0].stats.pongs_received >= 2 && links[0].stats.rtt_ms >= 0.0; + }; + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5); + while (!(measured(relay) && measured(client))) { + ASSERT_LT(std::chrono::steady_clock::now(), deadline) << "no round trip measured"; + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + const auto server_end = relay.link_reports().at(0); + const auto client_end = client.link_reports().at(0); + EXPECT_EQ(server_end.transport, "grpc"); + EXPECT_TRUE(server_end.accepted); + EXPECT_EQ(server_end.remote_node_id, "alice"); + EXPECT_FALSE(client_end.accepted); + EXPECT_EQ(client_end.remote_node_id, "relay"); + for (const auto &r : {server_end, client_end}) { + EXPECT_GT(r.stats.bytes_in, 0u); + EXPECT_GT(r.stats.bytes_out, 0u); + EXPECT_LT(r.stats.rtt_ms, 1000.0) << "loopback round trip"; + } + client.stop(); + relay.stop(); +} + TEST(StateTransportGrpcTest, DisconnectAfterStop) { cvc::state_transport_grpc tA, tB; tA.start("127.0.0.1:0", "A", "C"); diff --git a/src/cvc/tests/state_transport_ws_test.cpp b/src/cvc/tests/state_transport_ws_test.cpp index b2b8d6433..03863305b 100644 --- a/src/cvc/tests/state_transport_ws_test.cpp +++ b/src/cvc/tests/state_transport_ws_test.cpp @@ -13,6 +13,10 @@ // (the browser topology — clients that dial without a local acceptor, as a wasm peer will). #include +#include +#include +#include +#include #include #include #include @@ -254,6 +258,9 @@ TEST(StateTransportWsTest, RelayFansMediaFrameToOtherClient) { TEST(StateTransportWsTest, RelayReapsIdleConnection) { cvc::app aA; cvc::state_transport_ws S, tA; + // No link probe here: a client that answers Pings is NOT idle (see QuietClientAnsweringPings...). + S.set_ping_interval(std::chrono::milliseconds(0)); + tA.set_ping_interval(std::chrono::milliseconds(0)); S.start("127.0.0.1:0", "S", "C"); S.set_relay(true); std::mutex mu; @@ -295,6 +302,117 @@ TEST(StateTransportWsTest, RelayReapsIdleConnection) { S.stop(); } +// The link probe: both ends Ping, the other end Pongs, and each measures the round trip on its own +// clock. Each side reports one link with its peer's node id, traffic both ways, and an RTT. +TEST(StateTransportWsTest, LinkProbeMeasuresRoundTripAtBothEnds) { + cvc::app aA; + cvc::state_transport_ws S, tA; + S.set_ping_interval(std::chrono::milliseconds(30)); + tA.set_ping_interval(std::chrono::milliseconds(30)); + S.start("127.0.0.1:0", "S", "C"); + cvc::state_cluster_shard sA(aA, "C", "A"); + sA.attach(); + tA.register_shard(&sA); + ASSERT_TRUE(tA.connect_to_peer(ws(S.listen_address()), std::chrono::milliseconds(2000))); + const auto measured = [](const cvc::state_transport_ws &t) { + const auto links = t.link_reports(); + return links.size() == 1 && links[0].stats.pongs_received >= 2 && links[0].stats.rtt_ms >= 0.0; + }; + ASSERT_TRUE( + wait_until([&] { return measured(S) && measured(tA); }, std::chrono::milliseconds(3000))); + const auto server = S.link_reports().at(0); + const auto client = tA.link_reports().at(0); + EXPECT_EQ(server.transport, "ws"); + EXPECT_TRUE(server.accepted); + EXPECT_EQ(server.remote_node_id, "A"); + EXPECT_FALSE(client.accepted); + EXPECT_EQ(client.remote_node_id, "S"); + for (const auto &r : {server, client}) { + EXPECT_GT(r.stats.bytes_in, 0u); + EXPECT_GT(r.stats.bytes_out, 0u); + EXPECT_GE(r.stats.rtt_min_ms, 0.0); + EXPECT_LT(r.stats.rtt_ms, 1000.0) << "loopback round trip"; + EXPECT_GE(r.age_s, 0.0); + } + tA.stop(); + S.stop(); +} + +// A client that sends nothing of its own but still answers the relay's Pings is alive, so the idle +// reaper keeps it -- the backgrounded-browser-tab case (its WebSocket onmessage still fires and +// answers, while its timers are throttled). A client that cannot answer is still reaped +// (RelayReapsIdleConnection). +TEST(StateTransportWsTest, QuietClientAnsweringPingsIsNotReaped) { + cvc::app aA; + cvc::state_transport_ws S, tA; + S.set_ping_interval(std::chrono::milliseconds(20)); + tA.set_ping_interval(std::chrono::milliseconds(0)); // the client never initiates anything + S.start("127.0.0.1:0", "S", "C"); + S.set_relay(true); + std::atomic reaped{0}; + S.set_on_connection_closed([&](const std::string &) { reaped.fetch_add(1); }); + cvc::state_cluster_shard sA(aA, "C", "A"); + sA.attach(); + tA.register_shard(&sA); + ASSERT_TRUE(tA.connect_to_peer(ws(S.listen_address()), std::chrono::milliseconds(2000))); + ASSERT_TRUE( + wait_until([&] { return S.connection_count() >= 1; }, std::chrono::milliseconds(2000))); + for (int i = 0; i < 20; ++i) { // ~600 ms of reaping with a 150 ms idle limit + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + S.reap_idle(std::chrono::milliseconds(150)); + } + EXPECT_EQ(reaped.load(), 0) << "a client answering Pings is not a ghost"; + EXPECT_EQ(S.connection_count(), 1u); + tA.stop(); + S.stop(); +} + +// A peer that stops reading (a frozen browser tab, a dead network) must not hang the relay: a write +// to it blocks once its TCP buffers fill, holding the connection's write lock, so without the write +// watchdog every later writer to it -- the relay's main loop included -- waits forever (seen in +// production). The watchdog closes a connection whose write has been blocked > 3 s: the write +// returns and the connection is gone. +TEST(StateTransportWsTest, WriteToAPeerThatStopsReadingIsAbortedNotHung) { + namespace beast = boost::beast; + namespace asio = boost::asio; + cvc::state_transport_ws S; + S.set_ping_interval(std::chrono::milliseconds(0)); // the watchdog still runs; no probe traffic + S.start("127.0.0.1:0", "S", "C"); + const std::string addr = S.listen_address(); + const std::string port = addr.substr(addr.rfind(':') + 1); + + // A raw WebSocket peer: completes the handshake, then never reads. + asio::io_context ioc; + asio::ip::tcp::resolver resolver(ioc); + asio::ip::tcp::socket sock(ioc); + asio::connect(sock, resolver.resolve("127.0.0.1", port)); + beast::websocket::stream peer(std::move(sock)); + peer.handshake("127.0.0.1", "/"); + ASSERT_TRUE( + wait_until([&] { return S.connection_count() == 1; }, std::chrono::milliseconds(2000))); + + std::atomic returned{false}; + std::thread writer([&] { + cvc::ws_media_frame m; + m.origin_node_id = "S"; + m.stream_id = "camera"; + m.codec = "mjpeg"; + m.payload.assign(std::size_t(32) << 20, 0x5a); // far beyond any socket buffer + S.publish_media_frame(m); + S.publish_media_frame(m); + returned.store(true); + }); + EXPECT_TRUE(wait_until([&] { return returned.load(); }, std::chrono::milliseconds(10000))) + << "the write to a non-reading peer never returned (the relay would be hung)"; + EXPECT_TRUE( + wait_until([&] { return S.connection_count() == 0; }, std::chrono::milliseconds(5000))) + << "the stalled connection was not closed"; + writer.join(); + beast::error_code ec; + peer.next_layer().close(ec); + S.stop(); +} + // Late-joiner bootstrap: a relay that holds the room tree in its own shard pushes a full snapshot // to each newly admitted connection, so a client that joins AFTER others are present still sees // them (live set_relay forwarding only carries writes made after the client connects). This is what diff --git a/src/cvcGL/CMakeLists.txt b/src/cvcGL/CMakeLists.txt index cc597da3a..305d6a45a 100644 --- a/src/cvcGL/CMakeLists.txt +++ b/src/cvcGL/CMakeLists.txt @@ -811,6 +811,11 @@ add_executable(cvcgl_ariadne_realize test/cvcgl_ariadne_realize.cpp) target_link_libraries(cvcgl_ariadne_realize PRIVATE cvcGL) add_test(NAME cvcgl_ariadne_realize COMMAND cvcgl_ariadne_realize) +# H1 slice 6: the conference room in 3D (room scene realize + avatar markers + key -> intent). +add_executable(cvcgl_room_view test/cvcgl_room_view.cpp) +target_link_libraries(cvcgl_room_view PRIVATE cvcGL) +add_test(NAME cvcgl_room_view COMMAND cvcgl_room_view) + # The terminal extension (TerminalScene, roadmap §16.6): a scene rendered offscreen becomes the # terminal backend's background under the widgets, and a background drag turns the camera. add_executable(cvcgl_terminal_scene test/cvcgl_terminal_scene.cpp) diff --git a/src/cvcGL/CameraController.cpp b/src/cvcGL/CameraController.cpp index b58f18fb2..dafc00079 100644 --- a/src/cvcGL/CameraController.cpp +++ b/src/cvcGL/CameraController.cpp @@ -1099,6 +1099,10 @@ void CameraController::setPoseMirrorHz(double hz) { void CameraController::releaseHeldKeys() { m_impl->held.clear(); } +bool CameraController::keyHeld(const std::string &keySym) const { + return m_impl->held_has(normKey(keySym)); +} + void CameraController::setPointerCapture(bool on) { Impl &s = *m_impl; s.pointerCapture = on; diff --git a/src/cvcGL/ariadne/room_view.cpp b/src/cvcGL/ariadne/room_view.cpp new file mode 100644 index 000000000..f55191c70 --- /dev/null +++ b/src/cvcGL/ariadne/room_view.cpp @@ -0,0 +1,240 @@ +// The conference room in 3D (H1 slice 6) — see inc/cvc/gl/ariadne/room_view.h. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace cvc { +namespace gl { +namespace ariadne { + +namespace { + +namespace rs = cvc::ariadne::conference; + +// An avatar marker: a 0.5 x 0.5 x 1.7 m box standing on the pose (its base at the pose's z). +constexpr float kAvatarWidth = 0.5f; +constexpr float kAvatarHeight = 1.7f; + +// A stable, distinct colour per peer (a hue from the pid's hash); the local avatar is gold. +void avatar_color(const std::string &pid, bool self, float out[3]) { + if (self) { + out[0] = 0.95f, out[1] = 0.75f, out[2] = 0.20f; + return; + } + std::uint32_t h = 2166136261u; // FNV-1a: the same pid colours the same on every client + for (unsigned char c : pid) + h = (h ^ c) * 16777619u; + // A hue in [0, 360) that skips 30-75 degrees, the oranges and yellows near the local avatar's + // gold, so a peer is never mistaken for you. + const std::uint32_t deg = h % 315u; + const float hue = static_cast(deg < 30u ? deg : deg + 45u) / 60.0f; // [0, 6) + const float s = 0.55f, v = 0.85f, c = v * s; + const float x = c * (1.0f - std::fabs(std::fmod(hue, 2.0f) - 1.0f)); + float r = 0, g = 0, b = 0; + switch (static_cast(hue)) { + case 0: + r = c, g = x; + break; + case 1: + r = x, g = c; + break; + case 2: + g = c, b = x; + break; + case 3: + g = x, b = c; + break; + case 4: + r = x, b = c; + break; + default: + r = c, b = x; + break; + } + out[0] = r + (v - c), out[1] = g + (v - c), out[2] = b + (v - c); +} + +// Row-major 4x4 of a rotation quaternion and a translation. +void pose_matrix(const cvc::physics::quat &q, double tx, double ty, double tz, double m[16]) { + const double w = q.w, x = q.x, y = q.y, z = q.z; + m[0] = 1 - 2 * (y * y + z * z), m[1] = 2 * (x * y - w * z), m[2] = 2 * (x * z + w * y), m[3] = tx; + m[4] = 2 * (x * y + w * z), m[5] = 1 - 2 * (x * x + z * z), m[6] = 2 * (y * z - w * x), m[7] = ty; + m[8] = 2 * (x * z - w * y), m[9] = 2 * (y * z + w * x), m[10] = 1 - 2 * (x * x + y * y), + m[11] = tz; + m[12] = 0, m[13] = 0, m[14] = 0, m[15] = 1; +} + +} // namespace + +cvc::ariadne::avatar_intent intent_from_keys(bool forward, bool back, bool left, bool right, + const double eye[3], const double focal[3], + double speed) { + double fx = focal[0] - eye[0], fy = focal[1] - eye[1]; + const double fl = std::hypot(fx, fy); + if (fl < 1e-9) { // looking straight down (or up): no heading + fx = 0.0, fy = 1.0; + } else { + fx /= fl, fy /= fl; + } + const double rx = fy, ry = -fx; // forward x up(0,0,1) + const double f = (forward ? 1.0 : 0.0) - (back ? 1.0 : 0.0); + const double r = (right ? 1.0 : 0.0) - (left ? 1.0 : 0.0); + double mx = f * fx + r * rx, my = f * fy + r * ry; + const double ml = std::hypot(mx, my); + cvc::ariadne::avatar_intent in; + if (ml > 1e-9) { + mx *= speed / ml, my *= speed / ml; + in.move = cvc::physics::vec3{mx, my, 0.0}; + } + return in; +} + +room_view::room_view(SceneRenderer &view, cvc::app &app, std::string room, std::string self_pid, + std::string bind_prefix) + : _view(view), _app(app), _room(std::move(room)), _self(std::move(self_pid)), + _prefix(std::move(bind_prefix)) {} + +room_view::~room_view() { + clear_scene(); + SceneGraph &sg = _view.scene(); + for (const auto &kv : _avatars) + sg.removeGraphics(avatar_node_id(kv.first)); +} + +std::vector room_view::take_warnings() { + std::vector w; + w.swap(_warnings); + return w; +} + +void room_view::update(cvc::ariadne::holon &h, double wall_dt) { + sync_scene(); + sync_avatars(h, wall_dt); +} + +void room_view::clear_scene() { + SceneGraph &sg = _view.scene(); + for (const std::string &id : _scene_ids) + sg.removeGraphics(id); + _scene_ids.clear(); + _realized.reset(); // after the nodes: ~StageLighting removes a rig's own lights +} + +void room_view::sync_scene() { + const std::uint64_t rev = rs::scene_revision(_app, _room); + if (rev == _realized_rev) + return; + clear_scene(); + cvc::ariadne::Scene scene = rs::read_room_scene(_app, _room); + for (cvc::ariadne::SceneNode &n : scene.nodes) + n.id = object_node_id(n.id); + for (cvc::ariadne::SceneLight &l : scene.lights) + l.id = object_node_id(l.id); + SceneGraph &sg = _view.scene(); + _realized = std::make_unique(realize_scene(sg, scene, _prefix, &_warnings)); + _scene_ids = _realized->created; + for (const cvc::ariadne::SceneLight &l : scene.lights) + if (l.rig.empty()) // a rig's lights belong to (and leave with) its StageLighting + _scene_ids.push_back(l.id); + // The background is a view property (as AriRuntime applies a document's). + if (scene.has_background) { + const float *t = scene.background_top, *b = scene.background_bottom; + vtkRenderer *r = _view.renderer(); + if (scene.background_gradient && r) { + r->SetGradientBackground(true); + r->SetBackground(b[0], b[1], b[2]); // bottom + r->SetBackground2(t[0], t[1], t[2]); // top + } else { + if (r) + r->SetGradientBackground(false); + _view.setBackground(t[0], t[1], t[2]); + } + } + _realized_rev = rev; +} + +void room_view::sync_avatars(cvc::ariadne::holon &h, double wall_dt) { + SceneGraph &sg = _view.scene(); + const std::vector pids = h.participants(); + const std::set present(pids.begin(), pids.end()); + + // Departed participants: drop their markers. + for (auto it = _avatars.begin(); it != _avatars.end();) { + if (!present.count(it->first)) { + sg.removeGraphics(avatar_node_id(it->first)); + it = _avatars.erase(it); + } else { + ++it; + } + } + if (!present.count(_self)) + _have_self = false; + + for (const std::string &pid : pids) { + const bool self = (pid == _self); + const std::string id = avatar_node_id(pid); + if (!_avatars.count(pid)) { + cvc::ariadne::Scene one; + cvc::ariadne::SceneNode n; + n.id = id; + n.source_primitive = "box"; + n.box_size[0] = n.box_size[1] = kAvatarWidth; + n.box_size[2] = kAvatarHeight; + n.has_material = true; + avatar_color(pid, self, n.color); + n.ambient = 0.3f; + n.diffuse = 0.8f; + one.nodes.push_back(n); + realize_scene(sg, one, _prefix, &_warnings); // a plain geometry node: nothing to keep + _avatars[pid] = true; + } + + cvc::physics::pose p = h.display_pose(pid); + if (self) { + // Ease the local avatar's reconcile corrections: absorb each jump into an offset + // (so the shown position does not move this frame), then decay the offset. + const cvc::physics::vec3 c = h.take_correction(pid); + _self_offset = _self_offset - c; + const double k = wall_dt > 0.0 ? std::exp(-wall_dt / std::max(correction_tau, 1e-6)) : 1.0; + _self_offset = cvc::physics::vec3{_self_offset.x * k, _self_offset.y * k, _self_offset.z * k}; + p.position = p.position + _self_offset; + _self_pos[0] = p.position.x, _self_pos[1] = p.position.y, _self_pos[2] = p.position.z; + _have_self = true; + } + if (std::shared_ptr node = sg.getGraphics(id)) { + double m[16]; + // The box is centred on its origin: lift it by half its height to stand on the pose. + pose_matrix(p.rotation, p.position.x, p.position.y, p.position.z + 0.5 * kAvatarHeight, m); + node->setPoseMatrix(m); + } + } +} + +std::vector room_view::avatar_pids() const { + std::vector out; + for (const auto &kv : _avatars) + out.push_back(kv.first); + return out; +} + +bool room_view::self_position(double out[3]) const { + if (!_have_self) + return false; + out[0] = _self_pos[0], out[1] = _self_pos[1], out[2] = _self_pos[2]; + return true; +} + +} // namespace ariadne +} // namespace gl +} // namespace cvc diff --git a/src/cvcGL/test/cvcgl_room_view.cpp b/src/cvcGL/test/cvcgl_room_view.cpp new file mode 100644 index 000000000..d9596b51d --- /dev/null +++ b/src/cvcGL/test/cvcgl_room_view.cpp @@ -0,0 +1,161 @@ +// H1 slice 6: cvc::gl::ariadne::room_view — the conference room in 3D, offscreen. The room scene +// from the replicated tree is realized (and re-realized on a revision change), avatar markers +// follow a client holon's participants and poses, a local reconcile jump is eased (no visible +// step), the third-person key -> intent mapping is camera-relative, and teardown removes every +// node room_view made. +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using cvc::ariadne::avatar_intent; +using cvc::ariadne::holon; +using cvc::ariadne::holon_role; +using cvc::gl::SceneGraph; +using cvc::gl::SceneRenderer; +using cvc::gl::ariadne::intent_from_keys; +using cvc::gl::ariadne::room_view; +using cvc::physics::body_state; +using cvc::physics::vec3; +namespace rs = cvc::ariadne::conference; + +static int fails = 0; +static void chk(bool ok, const std::string &w) { + printf(" %s %s\n", ok ? "PASS" : "FAIL", w.c_str()); + if (!ok) + ++fails; +} +static bool near(double a, double b, double tol = 1e-6) { return std::fabs(a - b) < tol; } + +static bool node_xyz(SceneGraph &sg, const std::string &id, double out[3]) { + std::shared_ptr n = sg.getGraphics(id); + if (!n) + return false; + const vtkMatrix4x4 *m = n->getTransform(); + out[0] = m->GetElement(0, 3), out[1] = m->GetElement(1, 3), out[2] = m->GetElement(2, 3); + return true; +} + +int main() { + cvc::app app; + app.properties("system.log_verbosity", "0"); + + // ── keys -> intent, relative to the camera's heading ───────────────────── + { + const double eye[3] = {-10, 0, 5}, focal[3] = {0, 0, 0}; // looking along +x + avatar_intent w = intent_from_keys(true, false, false, false, eye, focal, 2.0); + chk(near(w.move.x, 2.0) && near(w.move.y, 0.0) && near(w.move.z, 0.0), + "W walks along the view"); + avatar_intent d = intent_from_keys(false, false, false, true, eye, focal, 2.0); + chk(near(d.move.x, 0.0) && near(d.move.y, -2.0), "D strafes right of the view (-y here)"); + avatar_intent wd = intent_from_keys(true, false, false, true, eye, focal, 2.0); + chk(near(std::hypot(wd.move.x, wd.move.y), 2.0) && wd.move.x > 0 && wd.move.y < 0, + "a diagonal is normalized, not faster"); + avatar_intent ws = intent_from_keys(true, true, false, false, eye, focal, 2.0); + chk(near(ws.move.x, 0.0) && near(ws.move.y, 0.0), "opposing keys cancel"); + const double top[3] = {0, 0, 10}, down[3] = {0, 0, 0}; + avatar_intent td = intent_from_keys(true, false, false, false, top, down, 1.0); + chk(near(td.move.y, 1.0) && near(td.move.x, 0.0), "straight down: forward is +Y"); + } + + // ── the camera's held keys are what the host reads for the avatar ───────── + { + cvc::gl::CameraController c(app, "room_keys.camera"); + c.keyDown("W"); + c.keyDown("Up"); + chk(c.keyHeld("w") && c.keyHeld("W") && c.keyHeld("Up"), "keyHeld sees held keys, any case"); + c.keyUp("w"); + chk(!c.keyHeld("w") && c.keyHeld("Up"), "and their release"); + c.releaseHeldKeys(); // what the UI overlay does when it takes the keyboard + chk(!c.keyHeld("Up"), "and the drop when a UI takes the keyboard"); + } + + // ── the room scene: realized, then re-realized on a revision change ─────── + { + SceneGraph sg(app, "room_scene"); + SceneRenderer view(sg, 128, 128, /*offscreen=*/true, "main"); + holon h(holon_role::client); + rs::seed_default_scene(app, "r"); + { + room_view rv(view, app, "r", "A", sg.getStatePrefix()); + rv.update(h, 1.0 / 60.0); + chk(rv.realized_revision() == rs::scene_revision(app, "r"), "realized the current revision"); + chk(sg.getGraphics("room_ground") != nullptr, "the ground is in the scene graph"); + chk(sg.getGraphics("room_ref") != nullptr, "the reference box is in the scene graph"); + chk(sg.getGraphics("room_sun") != nullptr, "the sun light is in the scene graph"); + double ref[3]; + chk(node_xyz(sg, "room_ref", ref) && near(ref[2], 0.5, 1e-4), "the box stands on the ground"); + + rs::remove_scene_object(app, "r", "ref"); + rv.update(h, 1.0 / 60.0); + chk(sg.getGraphics("room_ref") == nullptr, "a removed object leaves the graph"); + chk(sg.getGraphics("room_ground") != nullptr, "the rest is re-realized"); + chk(sg.getGraphics("room_sun") != nullptr, "the light too (not duplicated or lost)"); + } + chk(sg.getGraphics("room_ground") == nullptr && sg.getGraphics("room_sun") == nullptr, + "teardown removes the room's nodes and lights"); + } + + // ── avatars follow the holon; a local correction is eased ──────────────── + { + SceneGraph sg(app, "room_avatars"); + SceneRenderer view(sg, 128, 128, /*offscreen=*/true, "main"); + holon h(holon_role::client); + body_state a0, b0; + a0.position = vec3{4.0, 0.0, 0.0}; + b0.position = vec3{0.0, 4.0, 0.0}; + h.add_participant("A", /*local=*/true, a0); + h.add_participant("B", /*local=*/false, b0); + const double dt = h.clock().fixed_dt(); + { + room_view rv(view, app, "empty", "A", sg.getStatePrefix()); + h.advance(dt); + rv.update(h, dt); + double pa[3], pb[3], self[3]; + chk(node_xyz(sg, "avatar_A", pa) && near(pa[0], 4.0) && near(pa[2], 0.85, 1e-4), + "the local avatar's marker stands at its pose"); + chk(node_xyz(sg, "avatar_B", pb) && near(pb[1], 4.0), "a peer's marker stands at its pose"); + chk(rv.self_position(self) && near(self[0], 4.0), "self_position is the local avatar"); + + // A server correction teleports A by +3 m in y: the marker must not jump this frame, + // then glides to the corrected pose. + double before[3]; + node_xyz(sg, "avatar_A", before); + body_state moved = a0; + moved.position.y = 3.0; + h.apply_authoritative("A", moved, h.tick() + 1); + h.advance(dt); + rv.update(h, dt); + double now[3]; + node_xyz(sg, "avatar_A", now); + // One frame's share of the 3 m correction (3 * (1 - exp(-dt / tau)) ~ 0.2 m), not the jump. + const double share = 3.0 * (1.0 - std::exp(-dt / rv.correction_tau)); + chk(near(now[1] - before[1], share, 1e-6), "a correction does not jump the marker"); + for (int i = 0; i < 120; ++i) { // 1 s >> correction_tau + h.advance(dt); + rv.update(h, dt); + } + node_xyz(sg, "avatar_A", now); + chk(near(now[1], 3.0, 1e-3), "and the marker glides onto the corrected pose"); + + h.remove_participant("B"); + h.advance(dt); + rv.update(h, dt); + chk(sg.getGraphics("avatar_B") == nullptr, "a departed peer's marker is removed"); + chk(sg.getGraphics("avatar_A") != nullptr, "the others stay"); + } + chk(sg.getGraphics("avatar_A") == nullptr, "teardown removes the avatar markers"); + } + + printf("%s (%d failure%s)\n", fails ? "FAILED" : "OK", fails, fails == 1 ? "" : "s"); + return fails ? 1 : 0; +}