feat(core): clock_driver — unified per-frame time-slice fan-out; drive the conference pump through it

inc/cvc/core/clock_driver.h

@@ -0,0 +1,174 @@
+/*
+  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.
+*/
+
+#ifndef __CVC_CLOCK_DRIVER_H__
+#define __CVC_CLOCK_DRIVER_H__
+
+#include <chrono>
+#include <cstdint>
+#include <cvc/core/namespace.h>
+#include <cvc/core/world_clock.h>
+#include <functional>
+#include <vector>
+
+namespace cvc {
+
+class world_clock_binding;
+
+// ---------------
+// cvc::clock_driver
+// ---------------
+// The one place a host advances a world_clock per frame and fans the result out
+// to everything that follows it, in a fixed order, sharing a single reading of
+// time.
+//
+// A world_clock says "one driver per clock; everything else follows" but leaves
+// the driving to the host. In practice every real host grew the same loop by
+// hand: measure a wall delta, call advance() exactly once, then -- in a
+// load-bearing order -- step the simulations whose quanta fell due and present
+// what the frame should show at the returned alpha. The Ariadne scene driver
+// (tick_scene) is this loop; the conference host is this loop open-coded as an
+// eleven-step pump; a holon will be this loop again. clock_driver is that loop
+// factored out so there is exactly ONE advance per clock per frame and one place
+// that owns the order.
+//
+// Two kinds of observer, mirroring the advance()/for-each-step/present shape the
+// world_clock header sketches:
+//
+//   step observers  - fire first, in ascending `order`. Semantically "advance
+//     your own state by the quanta that fell due this frame" (a physics world's
+//     follow(), a navigator, a holon's simulation). They read frame.step
+//     (first_tick + 1 .. first_tick + steps) or keep a cvc::clock_follower; they
+//     must not advance the clock.
+//   frame observers - fire after every step observer, in ascending `order`.
+//     Semantically "do the once-per-frame work / present at frame.alpha" (drain
+//     a transport, forward media stamped with frame.t, interpolate a pose,
+//     publish telemetry). The conference's whole pump is frame observers.
+//
+// Both kinds receive the SAME clock_frame, computed once at the top of advance()
+// -- so a media timestamp (frame.t) read by one observer is bit-identical to the
+// one read by another, which is what lets a receiver align two streams that were
+// stamped in the same frame.
+//
+// Threading: a clock_driver is single-threaded. advance() and the observers it
+// calls run on whatever thread drives the frame (the SDL/render main thread for
+// a window host). It takes no locks and must not be driven from two threads; a
+// follower on another thread waits on the clock's wait_for_tick_change() and
+// keeps its own cursor, it does not share a driver. Re-entering advance() from
+// within an observer throws.
+class clock_driver {
+public:
+  // A frame of time, computed once per advance() and handed to every observer.
+  struct clock_frame {
+    // What this advance decided: steps (whole quanta to run now), alpha
+    // ([0,1) into the next quantum), dropped_steps, first_tick (the tick BEFORE
+    // this advance; the quanta to run are first_tick + 1 .. first_tick + steps).
+    world_clock::step_result step;
+    std::uint64_t tick = 0; // clock.tick() AFTER this advance (first_tick + steps)
+    double t = 0.0;         // world seconds, clock.t() (tick * fixed_dt)
+    double alpha = 0.0;     // step.alpha; interpolation fraction into the next quantum
+    double fixed_dt = 0.0;  // the clock's quantum
+    // The raw wall delta this frame, in seconds, BEFORE any time scaling or
+    // pause -- for work that must track real time no matter the sim rate
+    // (a talk-hold decay, a camera, an fps readout). Never the scaled world dt.
+    double wall_dt = 0.0;
+    bool first = false;      // true only on the very first advance (step.steps == 0)
+    std::uint64_t index = 0; // monotonic frame counter, 0 on the first advance
+  };
+
+  using observer_fn = std::function<void(const clock_frame &)>;
+
+  // A registration token. Keep it to remove() the observer later; a default
+  // (id == 0) handle names nothing.
+  struct handle {
+    std::uint64_t id = 0;
+    explicit operator bool() const noexcept { return id != 0; }
+    bool operator==(const handle &o) const noexcept { return id == o.id; }
+  };
+
+  explicit clock_driver(world_clock &clock);
+
+  world_clock &clock() noexcept { return _clock; }
+  const world_clock &clock() const noexcept { return _clock; }
+
+  // Route advance() through a world_clock_binding instead of the clock directly,
+  // so the sim.* control keys (speed/paused/mode/step) steer this driver and the
+  // telemetry keys (tick/time/dropped) are published each frame. nullptr (the
+  // default) drives the clock raw -- used where the clock is a private media
+  // timebase with no state face. The binding is not owned.
+  void set_binding(world_clock_binding *b) noexcept { _binding = b; }
+  world_clock_binding *binding() const noexcept { return _binding; }
+
+  // Register an observer. `order` sorts observers of the same kind ascending;
+  // ties keep registration order (stable). Returns a handle for remove().
+  // Registering from inside an observer is allowed; the new observer does not
+  // fire until the next advance().
+  handle add_step_observer(int order, observer_fn fn);
+  handle add_frame_observer(int order, observer_fn fn);
+
+  // Remove a previously registered observer. Safe to call from within an
+  // observer, including an observer removing itself (a one-shot). Returns true
+  // if a live observer with that handle was found.
+  bool remove(handle h);
+
+  // Advance the clock once and fire observers: every step observer in order,
+  // then every frame observer in order, all with the same clock_frame.
+  //
+  // wall_dt < 0 (the default) measures a steady_clock delta since the previous
+  // advance -- real-time animation that does not crawl under a slow frame. The
+  // FIRST such call books no time (delta 0) and only seeds the baseline, so a
+  // host that constructs the driver long before its first frame does not inject
+  // a huge startup delta. wall_dt >= 0 uses the given delta verbatim (a fixed dt
+  // for deterministic capture, or a test), on the first call too.
+  //
+  // Returns the frame it built; the same reference frame() returns until the
+  // next advance.
+  const clock_frame &advance(double wall_dt = -1.0);
+
+  const clock_frame &frame() const noexcept { return _frame; }
+
+private:
+  struct entry {
+    std::uint64_t id = 0;
+    int order = 0;
+    std::uint64_t seq = 0; // insertion order, breaks `order` ties stably
+    observer_fn fn;
+    bool dead = false;      // removed while firing; compacted after the fan-out
+    bool want_step = false; // target list, used only for pending adds
+  };
+
+  handle add(std::vector<entry> &list, bool &dirty, int order, observer_fn fn);
+  static void sort_if_dirty(std::vector<entry> &list, bool &dirty);
+  void fire(std::vector<entry> &list);
+  void flush_pending();
+
+  world_clock &_clock;
+  world_clock_binding *_binding = nullptr;
+
+  std::vector<entry> _step_obs;
+  std::vector<entry> _frame_obs;
+  std::vector<entry> _pending; // observers added during a fan-out
+  bool _step_dirty = false;
+  bool _frame_dirty = false;
+
+  std::uint64_t _next_id = 1;
+  std::uint64_t _next_seq = 0;
+  std::uint64_t _index = 0;
+
+  bool _started = false;
+  bool _in_advance = false;
+  std::chrono::steady_clock::time_point _last_tp;
+
+  clock_frame _frame;
+};
+
+} // namespace cvc
+
+#endif // __CVC_CLOCK_DRIVER_H__

src/arigl/main.cpp

@@ -68,6 +68,7 @@
 #include <cvc/ariadne/stream/stream.h>         // stream::open — per-peer decode proxies
 #include <cvc/ariadne/stream/stream_channel.h> // subscribe / publish_external
 #include <cvc/ariadne/stream/stream_registry.h> // look up the local camera/mic streams by token
+#include <cvc/core/clock_driver.h>        // one-advance-per-frame fan-out over the world_clock
 #include <cvc/core/world_clock.h>         // app.world_clock().t() — the common media timestamp
 #include <cvc/gl/capture/audio_source.h>  // list_audio_capture_devices — mic picker
 #include <cvc/gl/capture/camera_source.h> // list_cameras — camera picker
@@ -355,21 +356,31 @@ struct ConferenceSession {
     return "media." + origin + "." + kind;
   }
 
-  // The common media timestamp: cvc::world_clock seconds (the canonical time base; frame pts is
-  // meant to be world time). The conference .ari declares no `clock:` block, so nothing else
-  // advances the per-app clock — drive it from wall time here, once per pump, and stamp BOTH our
-  // camera and mic frames with the returned t() so a receiver can align them (their sources' own
-  // pts are on unrelated clocks). t() is tick*fixed_dt (drift-free, 1/120 s quanta — sub-frame for
-  // media).
-  double last_wall_ = -1.0;
-  double media_clock_now() {
-    const double wall =
-        std::chrono::duration<double>(std::chrono::steady_clock::now().time_since_epoch()).count();
-    if (last_wall_ >= 0.0)
-      app_->world_clock().advance(wall - last_wall_);
-    last_wall_ = wall;
-    return app_->world_clock().t();
-  }
+  // The per-frame clock driver: advances the per-app cvc::world_clock from wall time ONCE per frame
+  // (the conference .ari declares no `clock:` block, so nothing else drives this app's clock) and
+  // fans the conference's per-frame steps out as ordered observers. driver_->frame().t is the ONE
+  // media timestamp per frame — world_clock seconds, tick*fixed_dt (drift-free, 1/120 s quanta;
+  // sub-frame for media) — stamped on BOTH our camera and mic frames so a receiver can align them
+  // (their sources' own pts are on unrelated clocks). Built in setup() once the app is known; null
+  // for a plain local document (no relay), where pump() no-ops. install_observers() defines the
+  // step order below.
+  std::unique_ptr<cvc::clock_driver> driver_;
+  cvc::clock_driver::handle dev_enum_h_; // handle of the self-removing device-enum one-shot
+
+  // The order each conference step fires in, each frame (ascending). Gaps leave room to slot a
+  // future step between these without renumbering.
+  enum : int {
+    kOrderPumpAll = 10,   // drain/flush the state transport (replicated room tree) — first
+    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)
+    kOrderApplyDev = 50,  // map the device combos' picks to cam/mic.want_device
+    kOrderForward = 60,   // forward our camera+mic, each stamped with frame.t
+    kOrderDrain = 70,     // route peers' inbound frames into per-peer buffers
+    kOrderMix = 80,       // sum peers' queued audio into the single mixed output
+    kOrderShow = 90,      // advance each peer's video to its audio clock
+    kOrderActivity = 100, // decay + publish talk activity for the roster — last
+  };
 
   // Call AFTER the GL bundles and BEFORE load(): register the room verbs, seed conf.room/conf.self,
   // and dial the relay. Returns false (a plain local document) when no relay is configured.
@@ -405,9 +416,12 @@ struct ConferenceSession {
     app_ = &app;
     pid_ = nodeId;
     prefix_ = prefix;
-    // Drive the world clock from wall time (media_clock_now): live mode so advance() banks wall
-    // time.
+    // 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.
     app.world_clock().set_mode(cvc::world_clock::mode::live);
+    driver_ = std::make_unique<cvc::clock_driver>(app.world_clock());
+    install_observers();
     // The streams we forward: the camera (latest — freshest frame) and the mic (ring — every chunk
     // in order, since dropped audio is a glitch). The .ari publishes their tokens in
     // cam_token/mic_token.
@@ -492,30 +506,61 @@ struct ConferenceSession {
     active = true;
     return true;
   }
+  // 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
+  // (snapshot mode); a plain local document has no driver and no-ops.
   void pump() {
-    if (!tp_)
+    if (!tp_ || !driver_)
       return;
-    tp_->pump_all();
-    const double stamp =
-        media_clock_now(); // one world-clock reading per pump stamps all we forward
-    now_wall_ =            // wall-clock seconds for the talk-hold (frame-rate indep.)
-        std::chrono::duration<double>(std::chrono::steady_clock::now().time_since_epoch()).count();
-    read_peer_controls(); // pull per-peer volume/mute the user-list window set
+    driver_->advance(); // measured wall delta; fires kOrderPumpAll .. kOrderActivity in order
+  }
+
+  // Register the per-frame steps as ordered observers on the driver, once, in setup(). They fire
+  // in ascending kOrder* each advance(), all BEFORE the document's on:tick (conf.pump runs before
+  // app_rt_.drain()), preserving the deliberate one-frame lag read_peer_controls relies on.
+  void install_observers() {
+    using cf = cvc::clock_driver::clock_frame;
+    // 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(); });
+    // 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 &) {
+      now_wall_ = std::chrono::duration<double>(std::chrono::steady_clock::now().time_since_epoch())
+                      .count();
+    });
+    // Pull the user-list window's per-peer volume/mute (local only, never sent).
+    driver_->add_frame_observer(kOrderControls, [this](const cf &) { read_peer_controls(); });
     // Enumerate the camera/mic devices for the pickers LAZILY — once the camera stream is open, so
     // the SDL audio/camera subsystems are already initialized (enumeration is then a cheap refcount
     // bump) and the Asyncify event loop is running. Doing this at setup instead STALLS startup on
-    // single-threaded wasm (SDL device queries before the loop runs block / can't resolve).
-    if (!devices_enumerated_ && app_ &&
-        !cvc::state::instance(*app_)(prefix_ + ".cam_token").value().empty()) {
-      enumerate_devices();
-      devices_enumerated_ = true;
-    }
-    apply_device_selection(); // map the device combos' picks to cam/mic.want_device
-    forward_streams(stamp);   // (also measures our own mic level for self "talking")
-    drain_inbox();            // (also measures each peer's mic level)
-    mix_audio();              // sum all peers' queued audio (per-peer gain, skip muted)
-    sync_and_show();    // advance each peer's video to its audio clock (skip locally video-muted)
-    publish_activity(); // decay talk levels + publish media.<pid>.talking for the name highlight
+    // single-threaded wasm (SDL device queries before the loop runs block / can't resolve). A
+    // self-removing one-shot: it checks every frame until the camera token appears, enumerates
+    // once, then unregisters so the per-frame check stops.
+    dev_enum_h_ = driver_->add_frame_observer(kOrderDevEnum, [this](const cf &) {
+      if (!devices_enumerated_ && app_ &&
+          !cvc::state::instance(*app_)(prefix_ + ".cam_token").value().empty()) {
+        enumerate_devices();
+        devices_enumerated_ = true;
+        driver_->remove(dev_enum_h_);
+      }
+    });
+    // Map the device combos' picks to cam/mic.want_device for the .ari to re-capture against.
+    driver_->add_frame_observer(kOrderApplyDev, [this](const cf &) { apply_device_selection(); });
+    // Forward our camera (latest) + mic (ring), each re-stamped with THIS frame's single
+    // world-clock reading (frame.t) so a receiver can A/V-align them; also measures our own mic
+    // level.
+    driver_->add_frame_observer(kOrderForward, [this](const cf &f) { forward_streams(f.t); });
+    // Route peers' inbound frames (queued by the reader thread) into per-peer buffers; also
+    // measures each peer's mic level.
+    driver_->add_frame_observer(kOrderDrain, [this](const cf &) { drain_inbox(); });
+    // Sum all peers' queued audio into the single mixed output (per-peer gain, skip muted).
+    driver_->add_frame_observer(kOrderMix, [this](const cf &) { mix_audio(); });
+    // Advance each peer's video to its audio clock (skip locally video-muted).
+    driver_->add_frame_observer(kOrderShow, [this](const cf &) { sync_and_show(); });
+    // Decay talk levels + publish media.<pid>.talking for the name highlight.
+    driver_->add_frame_observer(kOrderActivity, [this](const cf &) { publish_activity(); });
   }
 
   // Pull the per-peer LOCAL controls the user-list window writes — keyed by roster INDEX, which the
@@ -624,9 +669,19 @@ struct ConferenceSession {
     for (publisher &p : pubs_)
       p.sub.reset();
     if (stop_)
-      stop_();
+      stop_(); // joins the transport reader thread (idempotent: _running.exchange(false))
   }
 
+  // Join the transport reader BEFORE any member is destroyed. stop() runs in the destructor body,
+  // ahead of reverse-order member destruction, so the reader (which the transport destructor would
+  // otherwise join only AFTER media_mu_/inbox_ are already gone — they are declared after
+  // ws_/grpc_) can no longer fire on_media into a destroyed mutex/vector. This covers every
+  // teardown path: snapshot/check/info and a document-load failure return without the run loop's
+  // explicit stop(), and an exception unwinding the run loop skips it too. stop() is idempotent, so
+  // the run path's explicit call followed by this one is safe. (Single-threaded wasm has no reader
+  // thread.)
+  ~ConferenceSession() { stop(); }
+
   // Mix: pop one chunk from each peer that has audio queued, sum them (silence for an empty peer),
   // clamp, and play to the single output. Produce mixed chunks while the output is below ~100 ms so
   // it stays fed without running far ahead; stop when no peer has a chunk (output drains to

src/cvc/CMakeLists.txt

@@ -54,6 +54,8 @@ set(INCLUDE_FILES
   ../../inc/cvc/state/state_cluster_membership.h
   ../../inc/cvc/state/state_hybrid_time.h
   ../../inc/cvc/core/world_clock.h
+  ../../inc/cvc/core/clock_follower.h
+  ../../inc/cvc/core/clock_driver.h
   ../../inc/cvc/core/world_units.h
   ../../inc/cvc/core/unit_registry.h
   ../../inc/cvc/physics/types.h
@@ -253,6 +255,7 @@ set(SOURCE_FILES
   state/state_exec/exec_coordinator.cpp
   core/world_clock.cpp
   core/world_clock_binding.cpp
+  core/clock_driver.cpp
   core/world_units.cpp
   core/unit_registry.cpp
   physics/world.cpp