joule-code/joule-code-rs/crates/joule-moq-room/README.md

@@ -18,7 +18,9 @@ authority and clock keys and:
   watches each publisher's video track (a SUBSCRIBE that asks for no objects): when the
   track appears the clock signs `join` and `publish` for the publisher on its events track
   (live-events §8.1), and when the relay ends the track (the publisher's session closed) it
-  signs `leave`. Pages author none of it.
+  signs `leave`. Pages author none of it. The events track is published with the room's
+  first grant, before anything is signed on it, so a viewer who enters a room nobody has
+  joined subscribes to a track that is there.
 - **serves the room.** `GET /room` gives the signed descriptor, the composition descriptor
   and the publishers' grants, which a page verifies the room with (live-rooms §9).
 

joule-code/joule-code-rs/crates/joule-moq-room/src/http.rs

@@ -375,6 +375,14 @@ pub fn grant(svc: &Arc<Service>, client: &str, host: Option<&str>, authorization
     if role == Role::Publisher || issued.audience {
         tokio::spawn(svc.clone().watch(generation, principal, until));
     }
+    // The clock's events track is published with the grant, before the room's first event:
+    // a viewer who enters a room nobody has joined finds it.
+    let s = svc.clone();
+    tokio::spawn(async move {
+        if let Err(e) = s.open_events(generation).await {
+            tracing::warn!(error = %e, "the clock's events track is not published");
+        }
+    });
     tracing::info!(role = ?role, room = %hex::encode(issued.room), principal = %hex::encode(&principal[..6]), "a grant");
     reply(
         StatusCode::OK,

joule-code/joule-code-rs/crates/joule-moq-room/src/lib.rs

@@ -562,9 +562,17 @@ impl Service {
         self.emit_items_after(generation, items, 0).await
     }
 
-    /// [`Service::emit_items`], each record at room-clock `min_t` or later.
-    pub async fn emit_items_after(&self, generation: u64, items: &[Item], min_t: u64) -> Result<Vec<(Vec<u8>, u64)>, String> {
+    /// Publish the clock's events track in the room of `generation` now, unless it is
+    /// published: before the room's first event, so that a viewer who enters a room nobody has
+    /// joined subscribes to a track that is there (the relay refuses a subscription no
+    /// publisher answers within its rendezvous wait). Each grant asks for it.
+    pub async fn open_events(&self, generation: u64) -> Result<(), String> {
         let mut clock = self.clock.lock().await;
+        self.events_track(&mut clock, generation).await.map(|_| ())
+    }
+
+    /// The clock's session, with its events track in the room of `generation` published.
+    async fn events_track(&self, clock: &mut Clock, generation: u64) -> Result<Arc<Client>, String> {
         let client = clock.client.clone().ok_or("the clock has no session")?;
         let (room, t0_us) = {
             let st = self.state.lock().unwrap();
@@ -588,6 +596,14 @@ impl Service {
             clock.track = Some(Track { generation, room, t0_us, alias, _publish: publish, chain: Chain::default(), group: None, stream: None, next_object: 0 });
             self.co.lock().unwrap().room(generation).clock_chain.clear();
         }
+        Ok(client)
+    }
+
+    /// [`Service::emit_items`], each record at room-clock `min_t` or later.
+    pub async fn emit_items_after(&self, generation: u64, items: &[Item], min_t: u64) -> Result<Vec<(Vec<u8>, u64)>, String> {
+        let mut clock = self.clock.lock().await;
+        let client = self.events_track(&mut clock, generation).await?;
+        let room = clock.track.as_ref().expect("made above").room;
         let mut out = Vec::new();
         for what in items {
             let track = clock.track.as_mut().expect("made above");

joule-code/joule-code-rs/crates/joule-moq-room/tests/room_service.rs

@@ -9,6 +9,9 @@
 //! - the grant alone, without the proof, opens no session;
 //! - publishers are bounded, a viewer is not counted among them, grants are bounded per
 //!   client, and a new room starts once the last publisher has left;
+//! - a viewer may enter a room nobody has joined: the clock's events track is published with
+//!   the grant, so the viewer's subscription is answered, and the clock's next record (the
+//!   first publisher's `join`) reaches it;
 //! - an end-to-end encrypted room (WAI confidentiality): its record checks against its
 //!   descriptor, members key through the service's log as a page does (KeyPackages, the
 //!   clock's proposals, a GroupInfo, commits and Welcomes, each checked by the clock before
@@ -244,6 +247,58 @@ async fn run() {
     server.shutdown().await;
 }
 
+/// The relay answers a subscription only once its track has a publisher (or refuses it at
+/// its rendezvous wait), and the clock signs nothing in a room nobody has joined: the clock's
+/// events track has to be published with the grant, or a viewer of an empty room is refused.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn a_viewer_enters_a_room_nobody_has_joined() {
+    timeout(Duration::from_secs(120), run_empty()).await.expect("the test finishes in time");
+}
+
+async fn run_empty() {
+    let keys = Keys::generate();
+    let server = relay(&keys).await;
+    let cfg = Config::from_toml(&format!(
+        "listen = \"127.0.0.1:0\"\nkeys = \"/nonexistent\"\nrelay_url = \"ws://{}{}\"\n[room]\nmax_publishers = 2\njoin_wait_s = 30\n",
+        server.websocket_addr.unwrap(),
+        server.relay().config().path
+    ))
+    .unwrap();
+    let svc = Service::new(Keys::from_toml(&keys.to_toml()).unwrap(), &cfg);
+    let listener = tokio::net::TcpListener::bind(cfg.listen).await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(svc.clone().run_clock());
+    tokio::spawn(http::serve(svc.clone(), listener));
+    timeout(STEP, async {
+        while !svc.connected.load(Ordering::Acquire) {
+            tokio::time::sleep(Duration::from_millis(50)).await;
+        }
+    })
+    .await
+    .expect("the clock's session opens");
+
+    // A viewer first, in a room nobody has joined: its subscription to the clock's events
+    // track is answered.
+    let viewer = key(0x51);
+    let (st, v) = ask(addr, &viewer, "viewer", "10.0.0.5").await;
+    assert_eq!((st, v["role"].as_str()), (200, Some("viewer")), "{v}");
+    let room = v["room"]["room"].as_str().unwrap().to_string();
+    let cv = session(&server, with_grant(&server, &hex::decode(v["grant"].as_str().unwrap()).unwrap(), &viewer)).await;
+    let clock = pubhex(&keys.clock);
+    let ns = TrackNamespace::from_strs([room.as_str(), clock.as_str(), "events"]);
+    cv.subscribe(ns, TrackName::from_str("0"), Parameters::new()).await.expect("a viewer of an empty room subscribes to the clock's events track");
+
+    // Then a publisher enters; the clock signs her `join`, and it reaches the viewer on that
+    // subscription.
+    let alice = key(0x52);
+    let (st, a) = ask(addr, &alice, "publisher", "10.0.0.6").await;
+    assert_eq!(st, 200, "{a}");
+    assert_eq!(a["room"]["room"].as_str(), Some(room.as_str()), "the publisher enters the viewer's room");
+    let ca = session(&server, with_grant(&server, &hex::decode(a["grant"].as_str().unwrap()).unwrap(), &alice)).await;
+    let _held = publish_media(&ca, &room, &alice).await;
+    assert!(cv.next_data(STEP).await.is_some(), "the clock's record reaches the viewer on its subscription");
+}
+
 // ------------------------------------------------------------------- end-to-end encryption
 
 /// A member as a page is one: it reads the room's log in order and creates or commits