From 00d25c0ed2968928c2515ac7f508794f68a585c2 Mon Sep 17 00:00:00 2001 From: Kevin Ahrendt Date: Fri, 16 Oct 2026 12:51:12 -0400 Subject: [PATCH] Decide a client's availability and its join in the session A client's `available` and the join a turn to `available: true` makes for its stream roles were tracked in PlayerStream, whatever the client's roles, and ServerCore read them as the player's. Session now tracks them and sets the join for both stream roles; PlayerStream and VisualizerStream keep only their own copy for eligibility, and ServerCore reads the session's. A Session test covers which reports are a join, and a raw-client test has a visualizer-only client that starts unavailable in a playing group join it at the group lead once it reports available. --- src/player_stream.cpp | 7 --- src/player_stream.h | 9 +--- src/server_core.cpp | 2 +- src/session.cpp | 11 +++-- src/session.h | 6 +++ tests/test_server_visualizer.cpp | 75 ++++++++++++++++++++++---------- tests/test_session.cpp | 71 +++++++++++++++++++++++++++--- 7 files changed, 133 insertions(+), 48 deletions(-) diff --git a/src/player_stream.cpp b/src/player_stream.cpp index c52a733..0e8e238 100644 --- a/src/player_stream.cpp +++ b/src/player_stream.cpp @@ -65,15 +65,8 @@ std::optional PlayerStream::on_activation_written(bool playe } std::optional PlayerStream::on_client_state(const ClientState& state) { - const bool was_available = this->available_; - const bool first = !this->state_received_; this->available_ = state.available; - this->state_received_ = true; std::optional kind; - // The first client/state is an initial value; a report of true makes the player eligible - if (state.available && (first || !was_available)) { - kind = sched::TriggerKind::JOIN; - } // messaging.md "server/activate": an inactive role's object is ignored if (this->role_written_ && state.player) { if (!this->state_since_activation_) { diff --git a/src/player_stream.h b/src/player_stream.h index f3b4fb2..d4107f4 100644 --- a/src/player_stream.h +++ b/src/player_stream.h @@ -65,8 +65,7 @@ class PlayerStream { std::optional on_activation_written(bool player_active); /// @brief A validated client/state - /// @return a trigger kind when it made the player eligible: a report of `available: true` - /// that is the first or follows `false`, or the role's first object since its activation + /// @return a trigger kind when it carried the role's first object since its activation std::optional on_client_state(const ClientState& state); /// @brief Records the kind of the latest kind-setting event @@ -153,11 +152,6 @@ class PlayerStream { return this->role_written_ && this->available_ && this->state_since_activation_; } - /// @brief The latest client/state reported `available: true` - bool available() const { - return this->available_; - } - bool role_written() const { return this->role_written_; } @@ -245,7 +239,6 @@ class PlayerStream { bool role_written_{false}; bool state_since_activation_{false}; bool available_{false}; - bool state_received_{false}; sched::SchedTiming timing_; }; diff --git a/src/server_core.cpp b/src/server_core.cpp index d5abc4a..f316a86 100644 --- a/src/server_core.cpp +++ b/src/server_core.cpp @@ -710,7 +710,7 @@ void ServerCore::on_client_state(Client& client) { if (client.group == nullptr) { return; } - const bool available = client.driver->session().player().available(); + const bool available = client.driver->session().available(); // messaging.md "Server behavior when a client becomes unavailable": the first client/state is // an initial value, not a transition const bool turned_unavailable = client.available.value_or(false) && !available; diff --git a/src/session.cpp b/src/session.cpp index e09396b..002432d 100644 --- a/src/session.cpp +++ b/src/session.cpp @@ -400,14 +400,19 @@ void Session::on_client_state(const ClientState& state) { this->finish(SessionEnd::PROTOCOL_ERROR); return; } + // A report of `available: true` that is the first or follows `false` is a join for the + // client's stream roles + const bool turned_available = state.available && !this->available_; + this->available_ = state.available; this->artwork_.on_client_state(state); this->source_role_.on_client_state(state); // Queued by queue_state_marker(), behind what the owner queues in reaction (a group/update - // before the stream/end it leads to). The most recent kind-setting event wins, and both roles' - // events are JOIN. + // before the stream/end it leads to). The most recent kind-setting event wins, and every event + // here is a JOIN. const std::optional player = this->player_.on_client_state(state); const std::optional visualizer = this->visualizer_.on_client_state(state); - this->state_marker_kind_ = player ? player : visualizer; + this->state_marker_kind_ = + turned_available ? sched::TriggerKind::JOIN : (player ? player : visualizer); this->state_changed_ = true; } diff --git a/src/session.h b/src/session.h index a9f6a2d..6a11568 100644 --- a/src/session.h +++ b/src/session.h @@ -280,6 +280,11 @@ class Session { return this->queued_.empty() ? this->active_roles_ : this->queued_.back().active_roles; } + /// @brief The latest client/state reported `available: true`; false before the first + bool available() const { + return this->available_; + } + const PlayerStream& player() const { return this->player_; } @@ -488,6 +493,7 @@ class Session { bool leave_{false}; bool rekeyed_{false}; bool state_changed_{false}; + bool available_{false}; std::optional state_marker_kind_; SessionPhase phase_{SessionPhase::AWAIT_CLIENT_INIT}; }; diff --git a/tests/test_server_visualizer.cpp b/tests/test_server_visualizer.cpp index 98be3c6..1eb970c 100644 --- a/tests/test_server_visualizer.cpp +++ b/tests/test_server_visualizer.cpp @@ -645,6 +645,34 @@ TEST(ServerVisualizer, StaysOnTimeAcrossALiveStallWithNoPlayer) { sendspin::server::sched::timing_defaults::GROUP_DEFAULT_US); } +/// Stops the server, then checks client's first loudness message from the trace: at or past the +/// group lead g after its stream/start, on the first cell there unless that cell had passed when +/// the message was sent +void expect_first_message_group_lead_past_start(VisualizerServer& s, const std::string& client, + int64_t g) { + // The trace runs on the dispatch thread, which reports the client's disconnect at the stop + // after every trace call before it + s.server.stop(); + s.recorder.wait_for("disconnected " + client + " " + + std::to_string(static_cast(DisconnectReason::SERVER_STOPPED))); + std::optional started; + std::optional first; + for (const VisualizerTrace::Sent& sent : sent_to(s, client)) { + if (sent.start && !started) { + started = sent.stamp_us; + } else if (started && sent.id == wire::LOUDNESS_ID) { + first = sent; + break; + } + } + ASSERT_TRUE(started) << "no stream/start traced"; + ASSERT_TRUE(first) << "no message traced after the start"; + EXPECT_GE(first->ts - *started, g); + if (first->stamp_us < *started + g) { + EXPECT_LT(first->ts - *started, g + CELL_US); + } +} + // A visualizer that joins a playing group starts the group lead past its stream/start, as a joining // player starts its own lead past its trigger. The player here asks a 700 ms required lead, so `G` // is 710 ms with `M_pump`: the first message is at or past it. @@ -661,28 +689,31 @@ TEST(ServerVisualizer, StartsAJoinerTheGroupLeadPastItsStart) { ASSERT_TRUE(joiner.read_until_type("visualizer")); input->stop(); host.join(); - // The trace runs on the dispatch thread, which reports the joiner's disconnect at the stop - // after every trace call before it - s.server.stop(); - s.recorder.wait_for("disconnected " + joiner.id + " " + - std::to_string(static_cast(DisconnectReason::SERVER_STOPPED))); - std::optional started; - std::optional first; - for (const VisualizerTrace::Sent& sent : sent_to(s, joiner.id)) { - if (sent.start && !started) { - started = sent.stamp_us; - } else if (started && sent.id == wire::LOUDNESS_ID) { - first = sent; - break; - } - } - ASSERT_TRUE(started) << "no stream/start traced"; - ASSERT_TRUE(first) << "no message traced after the start"; - // The first cell at or past it, unless that cell had passed when the message was sent - EXPECT_GE(first->ts - *started, G); - if (first->stamp_us < *started + G) { - EXPECT_LT(first->ts - *started, G + CELL_US); - } + expect_first_message_group_lead_past_start(s, joiner.id, G); +} + +// A visualizer-only client whose first client/state reports `available: false` stays in the +// playing group it joined (spec messaging.md "Server behavior when a client becomes unavailable" +// moves only a client that turns unavailable); its report of `available: true` is a join, which +// starts its stream the group lead past its stream/start, `G` as above. The first state's own join +// may still be pending when the second arrives, so SessionTest.MakesATurnToAvailableAJoin holds the +// rule itself. +TEST(ServerVisualizer, JoinsAPlayingGroupOnceItTurnsAvailable) { + constexpr int64_t G = 710000; + VisualizerServer s; + const std::string loudness = R"({"types":["loudness"],"rate_max":50})"; + RawVisualizer both(s.server, "Both", 1000000, loudness, true); + const std::shared_ptr input = s.open({&both}); + ASSERT_TRUE(input); + std::thread host([&] { feed_tone(*input, 0, 3 * RATE, 0.5); }); + ASSERT_TRUE(both.read_until_type("visualizer")); + RawVisualizer joiner(s.server, "Joiner", 100000, loudness, false, false); + s.recorder.wait_for("connected " + joiner.id); + joiner.send_state(true, loudness); + ASSERT_TRUE(joiner.read_until_type("visualizer")); + input->stop(); + host.join(); + expect_first_message_group_lead_past_start(s, joiner.id, G); } // A group's drain completes once no stream is active, a visualizer's included, so a visualizer diff --git a/tests/test_session.cpp b/tests/test_session.cpp index 3a92e01..296337a 100644 --- a/tests/test_session.cpp +++ b/tests/test_session.cpp @@ -1109,15 +1109,9 @@ TEST_P(SessionTest, FailsARekeySilently) { } // ============================================================================== -// The source role +// Availability // ============================================================================== -SessionSettings source_settings() { - SessionSettings settings = test_settings(); - settings.hello.source_codecs = {sendspin::server::Codec::PCM, sendspin::server::Codec::FLAC}; - return settings; -} - /// A client/state as the driver takes it in: the marker it calls for follows void send_state(Harness& h, const std::string& payload) { h.send(h.client.encrypt_json(R"({"type":"client/state","payload":)" + payload + "}")); @@ -1126,6 +1120,69 @@ void send_state(Harness& h, const std::string& payload) { } } +/// Through an activation with the visualizer role alone, its markers taken +void activate_visualizer(Harness& h) { + ASSERT_NO_FATAL_FAILURE(h.complete_handshake()); + h.send(h.client.encrypt_json( + R"({"type":"client/hello","payload":{"name":"Display","supported_roles":["visualizer@v1"],)" + R"("visualizer@v1_support":{"buffer_capacity":100000},)" + R"("supported_pair_methods":{"pairing_psk":{}},"unpaired_access":{"enabled":true}}})")); + h.session.activate( + ServerActivate{true, std::vector{"visualizer@v1"}, std::nullopt}); + const std::vector sent = h.take_all(); + ASSERT_EQ(sent.size(), 1U); + // Read, so the client's nonce follows the server's + EXPECT_NE(h.decrypt_json(sent[0]).find("server/activate"), std::string::npos); +} + +// A report of `available: true` that is the first or follows `false` is a join for the client's +// stream roles, whose start needs the client available (spec messaging.md "stream/start"); another +// report is not. The client has the visualizer role alone. Each row's last client/state carries no +// visualizer object, so any join it makes is availability's; the client/states before it carry +// the object, and the first-report rows send none. With no group bound the trigger has no +// observable outcome, so the visualizer stream's preconditions show it. +TEST_P(SessionTest, MakesATurnToAvailableAJoin) { + constexpr const char* VISUALIZER = R"("visualizer":{"types":["loudness"],"rate_max":50})"; + struct Row { + const char* name; + /// The client/states before, each with the visualizer object + std::vector before; + bool available; + bool join; + }; + const Row rows[] = { + {"Control: true after false", {false}, true, true}, + {"the first report of true", {}, true, true}, + {"true after true", {true}, true, false}, + {"false after true", {true}, false, false}, + {"the first report of false", {}, false, false}, + }; + for (const Row& row : rows) { + SCOPED_TRACE(row.name); + Harness h(GetParam()); + ASSERT_NO_FATAL_FAILURE(activate_visualizer(h)); + for (const bool available : row.before) { + send_state(h, R"({"available":)" + std::string(available ? "true" : "false") + "," + + VISUALIZER + "}"); + EXPECT_TRUE(h.take_all().empty()); + } + ASSERT_FALSE(h.session.visualizer().preconditions().trigger_pending); + send_state(h, R"({"available":)" + std::string(row.available ? "true" : "false") + "}"); + EXPECT_EQ(h.session.visualizer().preconditions().trigger_pending, row.join); + EXPECT_EQ(h.session.available(), row.available); + } +} + +// ============================================================================== +// The source role +// ============================================================================== + +SessionSettings source_settings() { + SessionSettings settings = test_settings(); + settings.hello.source_codecs = {sendspin::server::Codec::PCM, sendspin::server::Codec::FLAC}; + return settings; +} + /// Through an activation with the source role and the client/state source object that follows it void activate_source(Harness& h) { ASSERT_NO_FATAL_FAILURE(h.complete_handshake());