Adopt single announcement start timestamp; drop stream/clear; add override_mute

Author: marcelveldt

include/sendspin/announcement_role.h

@@ -45,6 +45,12 @@ struct ServerAnnouncementStreamObject {
 
     ServerPlayerStreamObject format{};
 
+    /// Server clock time in microseconds when announcement output should begin. The client
+    /// translates it via clock synchronization and starts at that time, or as soon as possible
+    /// if it has already passed. The same value on several clients gives a coordinated
+    /// multi-speaker start. Required.
+    int64_t start_timestamp{0};
+
     /// Reduction in decibel (0-50) to apply to this client's own media output while the
     /// announcement stream is active. 0 means no ducking.
     uint8_t media_duck_db{0};
@@ -57,6 +63,9 @@ struct ServerAnnouncementStreamObject {
     /// the announcement follows the current master volume.
     std::optional<uint8_t> volume{};
 
+    /// When true, a muted client still renders the announcement while its media stays muted.
+    bool override_mute{false};
+
     bool is_complete() const {
         return this->format.is_complete();
     }
@@ -65,9 +74,9 @@ struct ServerAnnouncementStreamObject {
 /// @brief Listener for announcement role events
 ///
 /// THREAD SAFETY: on_announcement_write() fires on the announcement task's background thread.
-/// Implementations must be thread-safe for this method. on_announcement_start(),
-/// on_announcement_end(), and on_announcement_clear() fire on the main loop thread via
-/// drain_events(). The listener must outlive the role.
+/// Implementations must be thread-safe for this method. on_announcement_start() and
+/// on_announcement_end() fire on the main loop thread via drain_events(). The listener must
+/// outlive the role.
 class AnnouncementRoleListener {
 public:
     virtual ~AnnouncementRoleListener() = default;
@@ -85,17 +94,16 @@ class AnnouncementRoleListener {
     /// @brief Called when an announcement stream starts. Fires on the main loop thread
     ///
     /// The embedder applies the ducking policy here (e.g. duck the media pipeline by
-    /// `params.media_duck_db` over `params.duck_ramp_ms`).
+    /// `params.media_duck_db` over `params.duck_ramp_ms`) and honors `params.override_mute` and
+    /// `params.volume`. Also fires when the server re-sends `stream/start` to update the active
+    /// announcement's config (duck level, volume, or override_mute): the embedder re-applies the
+    /// updated policy, ramping a new `media_duck_db` from the current gain, without any
+    /// intervening on_announcement_end().
     virtual void on_announcement_start(const ServerAnnouncementStreamObject& /*params*/) {}
 
     /// @brief Called when the announcement stream ends (normally, aborted, or on transport
     /// loss). Fires on the main loop thread. The embedder releases the ducking here
     virtual void on_announcement_end() {}
-
-    /// @brief Called when the server clears the announcement stream (replace). Fires on the
-    /// main loop thread. The embedder drops any announcement audio it has buffered downstream;
-    /// the stream and the ducking state stay active
-    virtual void on_announcement_clear() {}
 };
 
 /**
@@ -105,9 +113,9 @@ class AnnouncementRoleListener {
  * arrive from the WebSocket network thread, are written into a dedicated ring buffer, decoded,
  * and delivered to the platform through AnnouncementRoleListener::on_announcement_write().
  * Unlike the player role there is no sample-accurate sync machinery: announcements are
- * per-client, start at (or as soon as possible after) their first chunk's timestamp - which is
- * also what aligns a coordinated multi-speaker announcement scheduled by the server - and are
- * paced by the sink.
+ * per-client, start at (or as soon as possible after) the stream's single `start_timestamp` -
+ * which is also what aligns a coordinated multi-speaker announcement scheduled by the server -
+ * and are then paced by the sink.
  *
  * Usage:
  * 1. Implement AnnouncementRoleListener with at minimum on_announcement_write()

src/announcement_role.cpp

@@ -22,22 +22,11 @@
 
 static const char* const TAG = "sendspin.announcement";
 
-/// @brief Size of the big-endian 64-bit timestamp at the start of announcement binary messages.
-static constexpr size_t BINARY_TIMESTAMP_SIZE = 8;
 static constexpr uint32_t HEADER_SEND_TIMEOUT_MS = 100U;
 // Denominator for the advertised buffer capacity fraction: advertises (N-1)/N of capacity,
 // matching the player role's ring-buffer metadata headroom.
 static constexpr size_t ANNOUNCEMENT_BUFFER_ADVERTISE_DENOMINATOR = 5;
 
-/// @brief Swaps bytes of a big-endian 64-bit value to host byte order.
-static int64_t be64_to_host(const uint8_t* bytes) {
-    uint64_t val = 0;
-    for (int i = 0; i < 8; ++i) {
-        val = (val << 8) | bytes[i];
-    }
-    return static_cast<int64_t>(val);
-}
-
 namespace sendspin {
 
 // ============================================================================
@@ -154,15 +143,11 @@ SS_HOT void AnnouncementRole::Impl::handle_binary(const uint8_t* data, size_t le
     if (this->config.audio_formats.empty()) {
         return;
     }
-    if (len < BINARY_TIMESTAMP_SIZE) {
-        SS_LOGW(TAG, "Binary message too short for timestamp");
-        return;
-    }
-    int64_t timestamp = be64_to_host(data);
-    // Announcement chunks are not sync-critical and must not be dropped for being late, so a
-    // full ring buffer is the only failure here (bounded by the advertised buffer_capacity).
-    if (!this->task->write_audio_chunk(data + BINARY_TIMESTAMP_SIZE, len - BINARY_TIMESTAMP_SIZE,
-                                       timestamp, CHUNK_TYPE_ENCODED_AUDIO, 0)) {
+    // Announcement chunks are untimed: the whole payload (after the stripped type byte) is one
+    // encoded audio frame. The stream's single start time travels with the codec header, not
+    // the chunks. Chunks are not sync-critical and must not be dropped for being late, so a full
+    // ring buffer is the only failure here (bounded by the advertised buffer_capacity).
+    if (!this->task->write_audio_chunk(data, len, 0, CHUNK_TYPE_ENCODED_AUDIO, 0)) {
         SS_LOGW(TAG, "Failed to buffer announcement chunk");
     }
 }
@@ -175,9 +160,24 @@ void AnnouncementRole::Impl::handle_stream_start(
         return;
     }
 
+    // A stream/start arriving while the announcement is already playing is a configuration update
+    // (duck level, volume, or override_mute), not a new clip: update the params in place and let
+    // the main thread re-apply the policy via on_announcement_start, without a codec header, a
+    // buffer clear, or a task restart. Replacing a clip is done by the server with stream/end
+    // then a fresh stream/start.
+    if (this->task->is_running()) {
+        this->event_state->stream_params_slot.write(announcement_obj);
+        this->enqueue_stream_event(AnnouncementStreamCallbackType::CONFIG_UPDATE);
+        return;
+    }
+
     bool header_sent = false;
     const ServerPlayerStreamObject& format = announcement_obj.format;
 
+    // The codec header chunk carries the stream's single start_timestamp: it is the first chunk
+    // the task consumes, so the task reads the scheduled start time from it before any audio.
+    const int64_t start_timestamp = announcement_obj.start_timestamp;
+
     if (!format.bit_depth.has_value() || !format.channels.has_value() ||
         !format.sample_rate.has_value() || !format.codec.has_value()) {
         SS_LOGE(TAG, "Announcement stream start missing required audio parameters");
@@ -195,8 +195,8 @@ void AnnouncementRole::Impl::handle_stream_start(
                                        : CHUNK_TYPE_OPUS_DUMMY_HEADER;
 
             header_sent = this->task->write_audio_chunk(reinterpret_cast<const uint8_t*>(&header),
-                                                        sizeof(DummyHeader), 0, chunk_type,
-                                                        HEADER_SEND_TIMEOUT_MS);
+                                                        sizeof(DummyHeader), start_timestamp,
+                                                        chunk_type, HEADER_SEND_TIMEOUT_MS);
             if (!header_sent) {
                 SS_LOGE(TAG, "Failed to send announcement codec header");
             }
@@ -205,9 +205,9 @@ void AnnouncementRole::Impl::handle_stream_start(
                 SS_LOGE(TAG, "FLAC codec header missing");
             } else {
                 std::vector<uint8_t> flac_header = base64_decode(format.codec_header.value());
-                header_sent =
-                    this->task->write_audio_chunk(flac_header.data(), flac_header.size(), 0,
-                                                  CHUNK_TYPE_FLAC_HEADER, HEADER_SEND_TIMEOUT_MS);
+                header_sent = this->task->write_audio_chunk(flac_header.data(), flac_header.size(),
+                                                            start_timestamp, CHUNK_TYPE_FLAC_HEADER,
+                                                            HEADER_SEND_TIMEOUT_MS);
                 if (!header_sent) {
                     SS_LOGE(TAG, "Failed to send announcement codec header");
                 }
@@ -237,13 +237,6 @@ void AnnouncementRole::Impl::handle_stream_end() const {
     this->enqueue_stream_event(AnnouncementStreamCallbackType::STREAM_END);
 }
 
-void AnnouncementRole::Impl::handle_stream_clear() const {
-    // The server replaces the announcement audio: discard what is buffered, keep the stream
-    // and the ducking state active. The chunks that follow are the replacement clip.
-    this->task->signal_stream_clear();
-    this->enqueue_stream_event(AnnouncementStreamCallbackType::STREAM_CLEARED);
-}
-
 void AnnouncementRole::Impl::on_stream_ring_event(AnnouncementStreamCallbackType event) {
     this->pending_events.push_back(event);
 }
@@ -295,9 +288,20 @@ void AnnouncementRole::Impl::drain_events() {
                 }
                 break;
             }
-            case AnnouncementStreamCallbackType::STREAM_CLEARED: {
+            case AnnouncementStreamCallbackType::CONFIG_UPDATE: {
+                // Re-sent stream/start on an active stream: adopt the new duck/volume/override_mute
+                // params and re-apply the policy through on_announcement_start. The task keeps
+                // playing; no state change and no task signal.
+                ServerAnnouncementStreamObject stream_params;
+                if (this->event_state->stream_params_slot.take(stream_params)) {
+                    this->current_stream_params = std::move(stream_params);
+                }
                 if (this->listener && this->stream_active) {
-                    this->listener->on_announcement_clear();
+                    const uint32_t generation = this->cleanup_generation;
+                    this->listener->on_announcement_start(this->current_stream_params);
+                    if (this->cleanup_generation != generation) {
+                        teardown_reentered = true;
+                    }
                 }
                 break;
             }
@@ -373,7 +377,7 @@ void AnnouncementRole::Impl::cleanup() {
 void AnnouncementRole::Impl::enqueue_stream_event(AnnouncementStreamCallbackType event) const {
     // A dropped STREAM_START would leave the task waiting for its start signal; a dropped
     // STREAM_END would leave media ducked. Both wedge the announcement, so log drops at ERROR.
-    static const char* const EVENT_NAMES[] = {"STREAM_START", "STREAM_END", "STREAM_CLEARED",
+    static const char* const EVENT_NAMES[] = {"STREAM_START", "STREAM_END", "CONFIG_UPDATE",
                                               "OUTPUT_STARTED", "OUTPUT_FINISHED"};
     push_event_or_log(this->inbox, InboxEventType::ANNOUNCEMENT_STREAM, static_cast<uint8_t>(event),
                       TAG, EVENT_NAMES[static_cast<uint8_t>(event)],

src/announcement_role_impl.h

@@ -34,7 +34,7 @@ struct ClientStateMessage;
 enum class AnnouncementStreamCallbackType : uint8_t {
     STREAM_START,     // New announcement stream is starting (from the network thread)
     STREAM_END,       // Announcement stream ended (from the network thread or cleanup)
-    STREAM_CLEARED,   // Server cleared buffered announcement audio (from the network thread)
+    CONFIG_UPDATE,    // Re-sent stream/start updates the active stream's config (network thread)
     OUTPUT_STARTED,   // Task began writing announcement audio to the sink (from the task thread)
     OUTPUT_FINISHED,  // Task finished draining the announcement (from the task thread)
 };
@@ -63,7 +63,6 @@ struct AnnouncementRole::Impl {
     void handle_binary(const uint8_t* data, size_t len) const;
     void handle_stream_start(const ServerAnnouncementStreamObject& announcement_obj) const;
     void handle_stream_end() const;
-    void handle_stream_clear() const;
     void on_stream_ring_event(AnnouncementStreamCallbackType event);
     // True if this tick has drainable announcement work. All announcement events (lifecycle from
     // the network thread, output transitions from the announcement task) travel over the shared

src/announcement_task.cpp

@@ -95,7 +95,6 @@ bool AnnouncementTask::start(bool task_stack_in_psram, unsigned priority) {
                              AnnouncementTaskBits::ANNOUNCEMENT_TASK_IDLE |
                              AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP |
                              AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END |
-                             AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR |
                              AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_START);
 
     platform_configure_thread("SendspinAnnc", ANNOUNCEMENT_TASK_STACK_SIZE,
@@ -125,13 +124,6 @@ void AnnouncementTask::signal_stream_end() {
     this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END);
 }
 
-void AnnouncementTask::signal_stream_clear() {
-    if (!this->is_initialized()) {
-        return;
-    }
-    this->event_flags_.set(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR);
-}
-
 void AnnouncementTask::signal_stream_start() {
     if (!this->is_initialized()) {
         return;
@@ -217,11 +209,9 @@ bool AnnouncementTask::wait_for_codec_header(AudioStreamInfo* stream_info) {
         if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP) != 0U) {
             return false;
         }
-        if ((flags & (AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END |
-                      AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR)) != 0U) {
-            // Stale commands from a stream that ended while idle
-            this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END |
-                                     AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR);
+        if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END) != 0U) {
+            // Stale command from a stream that ended while idle
+            this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END);
         }
 
         AudioRingBufferEntry* entry =
@@ -231,17 +221,19 @@ bool AnnouncementTask::wait_for_codec_header(AudioStreamInfo* stream_info) {
         }
 
         const ChunkType chunk_type = entry->chunk_type;
-        if (chunk_type == CHUNK_TYPE_ENCODED_AUDIO ||
-            chunk_type == CHUNK_TYPE_STREAM_CLEAR_MARKER) {
+        if (chunk_type == CHUNK_TYPE_ENCODED_AUDIO) {
             // Stale audio from a previous announcement; discard until a codec header arrives
             this->encoded_ring_buffer_->return_chunk(entry);
             continue;
         }
 
+        // The header chunk carries the stream's single scheduled start time.
+        const int64_t start_timestamp = entry->timestamp;
         const bool header_ok = this->decoder_->process_header(entry->data(), entry->data_size,
                                                               chunk_type, stream_info);
         this->encoded_ring_buffer_->return_chunk(entry);
         if (header_ok) {
+            this->start_timestamp_ = start_timestamp;
             this->decode_buffer_.resize(this->decoder_->get_decode_buffer_size());
             return true;
         }
@@ -264,14 +256,6 @@ void AnnouncementTask::play_stream() {
             this->drain_ring_buffer();
             break;
         }
-        if ((flags & AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR) != 0U) {
-            // Replace: drop what is buffered, keep playing what follows. The clear boundary is
-            // approximate (chunks racing the drain may be dropped with the old clip), which the
-            // silence-padded replacement stream tolerates.
-            this->event_flags_.clear(AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR);
-            this->drain_ring_buffer();
-            continue;
-        }
 
         AudioRingBufferEntry* entry =
             this->encoded_ring_buffer_->receive_chunk(ANNOUNCEMENT_CHUNK_RECEIVE_TIMEOUT_MS);
@@ -285,10 +269,6 @@ void AnnouncementTask::play_stream() {
         last_data_us = platform_time_us();
 
         if (entry->chunk_type != CHUNK_TYPE_ENCODED_AUDIO) {
-            if (entry->chunk_type == CHUNK_TYPE_STREAM_CLEAR_MARKER) {
-                this->encoded_ring_buffer_->return_chunk(entry);
-                continue;
-            }
             // Mid-stream codec header (configuration update): re-initialize the decoder
             AudioStreamInfo new_info{};
             if (this->decoder_->process_header(entry->data(), entry->data_size, entry->chunk_type,
@@ -312,7 +292,6 @@ void AnnouncementTask::play_stream() {
                 entry->data(), entry->data_size, this->decode_buffer_.data(),
                 this->decode_buffer_.size(), &decoded_size);
         }
-        const int64_t chunk_timestamp = entry->timestamp;
         this->encoded_ring_buffer_->return_chunk(entry);
 
         if (!decode_ok) {
@@ -324,14 +303,13 @@ void AnnouncementTask::play_stream() {
         }
 
         if (!output_started) {
-            // Loose start scheduling: begin at the first chunk's timestamp when time sync is
-            // available, or as soon as possible otherwise. Announcement chunks are never dropped
-            // for lateness.
-            this->wait_for_start_time(chunk_timestamp);
+            // Loose start scheduling: begin at the stream's single start_timestamp when time sync
+            // is available, or as soon as possible otherwise. Announcement chunks are never
+            // dropped for lateness.
+            this->wait_for_start_time(this->start_timestamp_);
             if ((this->event_flags_.get() &
                  (AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP |
-                  AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END |
-                  AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR)) != 0U) {
+                  AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END)) != 0U) {
                 continue;  // Re-enter the loop head to handle the command
             }
             output_started = true;
@@ -345,8 +323,7 @@ void AnnouncementTask::play_stream() {
         while (offset < decoded_size && listener != nullptr) {
             if ((this->event_flags_.get() &
                  (AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STOP |
-                  AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END |
-                  AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_CLEAR)) != 0U) {
+                  AnnouncementTaskBits::ANNOUNCEMENT_COMMAND_STREAM_END)) != 0U) {
                 break;
             }
             const size_t written = listener->on_announcement_write(