diff --git a/docs/integration-guide.md b/docs/integration-guide.md index d1a4070..7595702 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -289,14 +289,28 @@ struct MyArtworkListener : ArtworkRoleListener { display.show_image(slot, decoded_images[slot]); } - // Called from the main loop thread when artwork should be cleared. + // Called from the main loop thread when artwork should be cleared, either for this slot + // alone or for every slot at the end of a stream. void on_image_clear(uint8_t slot) override { display.clear_slot(slot); } }; ``` -**Cross-fades with back-pressure (opt-in).** By default the role decodes and displays every frame as it arrives. A slot can instead opt into a back-pressure gate by setting `ImageSlotPreference::require_frame_done`. With the gate on, the role keeps at most one un-acked *delivery* (a frame or a clear) in flight for that slot; any newer payload that arrives is buffered latest-wins and delivered only after the consumer calls `ArtworkRole::frame_done(slot)` from the main loop -- e.g. once a cross-fade animation finishes. A clear is itself a delivery and supersedes any un-acked frame, so exactly one `frame_done()` is owed after it. There is no timeout: the acknowledgment is the contract. +**Knowing when there is no artwork.** Artwork stays valid until the server replaces or clears it, and the artwork role is independent of the metadata role, so a track change alone sends nothing: the next track of the same album keeps showing the image already delivered. When an item genuinely has no artwork, the server clears that channel and `on_image_clear()` fires for that slot alone, scheduled to its server timestamp like a display (`display_offset_ms` included) so it lands on the item boundary. `on_image_clear()` also fires for every configured slot on stream end, stream clear, and disconnect. + +| What happened | What the listener sees | +| --- | --- | +| Artwork unchanged (e.g. next track of the same album) | nothing; the current image stays valid | +| Item has no artwork | `on_image_clear(slot)` for that slot | +| Stream ended, cleared, or connection lost | `on_image_clear(slot)` for every configured slot | + +**Cross-fades with back-pressure (opt-in).** By default the role decodes and displays every frame as it arrives. A slot can instead opt into a back-pressure gate by setting `ImageSlotPreference::require_frame_done`. With the gate on, the role keeps at most one un-acked *delivery* (a frame or a clear) in flight for that slot. Call `ArtworkRole::frame_done(slot)` from the main loop exactly once for every `on_image_display()` and `on_image_clear()` that slot receives -- e.g. once a cross-fade animation finishes. An extra call is a harmless no-op, but a missed one wedges the slot: there is no timeout, the acknowledgment is the contract. + +Payloads and stream-level clears reach the gate differently: + +- A **frame or per-channel clear** arriving while a delivery is un-acked is buffered latest-wins and delivered only after `frame_done(slot)`, and then owes its own `frame_done()`. It waits behind the outstanding delivery rather than replacing it, so a consumer is never interrupted mid-fade. +- A **stream end or stream clear** is a lifecycle event, not a payload, so it is never buffered: it fires `on_image_clear()` immediately for every configured slot, discards anything buffered, and replaces whatever delivery was outstanding. Exactly one `frame_done()` is owed afterward whatever was in flight. Pair the gate with `ImageSlotPreference::display_offset_ms` to start a fade before the track boundary (positive fires the display early, mirroring `PlayerRoleConfig::fixed_delay_us`), and use `lateness_ms` to shorten the fade so it still ends on schedule: diff --git a/include/sendspin/artwork_role.h b/include/sendspin/artwork_role.h index 833d17b..4003791 100644 --- a/include/sendspin/artwork_role.h +++ b/include/sendspin/artwork_role.h @@ -35,13 +35,25 @@ class SendspinClient; /// /// ACK GATE (opt-in per slot via ImageSlotPreference::require_frame_done): a "delivery" is /// either a frame (on_image_decode() followed later by on_image_display()) or a clear -/// (on_image_clear()). For an ack-enabled slot, at most one un-acked delivery is ever in flight; -/// the newest payload that arrives while a delivery is un-acked is buffered latest-wins and -/// delivered only after the consumer calls ArtworkRole::frame_done(slot). A clear supersedes any -/// un-acked frame for that slot -- exactly one ack is owed, and it is for the clear. A stream -/// restart automatically releases a frame that was decoded but never displayed (its display can -/// no longer fire), but a delivery that already reached on_image_display()/on_image_clear() stays -/// gated until frame_done() is called; there is no timeout. +/// (on_image_clear()). Call ArtworkRole::frame_done(slot) exactly once for every +/// on_image_display() and on_image_clear() that slot receives. An extra call is a harmless no-op, +/// but a missed one wedges the slot forever: there is no timeout. +/// +/// For an ack-enabled slot, at most one un-acked delivery is ever in flight. The two ways a clear +/// reaches the gate differ, so they are worth keeping apart: +/// - A payload -- a frame, or the server's per-channel clear for that slot -- arriving while a +/// delivery is un-acked is buffered latest-wins and delivered only after frame_done(slot), and +/// then owes its own frame_done(). It waits behind the outstanding delivery rather than +/// replacing it, so a consumer is never interrupted mid-presentation. +/// - A stream end or stream clear is a lifecycle event, not a payload, so it is never buffered: +/// it fires on_image_clear() immediately for every configured slot, discards anything buffered, +/// and replaces whatever delivery was outstanding. Exactly one frame_done() is owed afterward +/// whatever was in flight -- including when it lands on an un-acked per-channel clear, which +/// fires on_image_clear() again and still owes exactly one ack. +/// +/// A stream restart automatically releases a frame that was decoded but never displayed (its +/// display can no longer fire), but a delivery that already reached on_image_display()/ +/// on_image_clear() stays gated until frame_done() is called. class ArtworkRoleListener { public: virtual ~ArtworkRoleListener() = default; @@ -77,7 +89,15 @@ class ArtworkRoleListener { /// @brief Called on the main loop thread when artwork should be cleared for a slot /// - /// Fires on stream end or stream clear for each configured slot. + /// Fires on stream end or stream clear for each configured slot, and for a single slot when + /// the server clears that channel (the artwork for the current item is gone, e.g. a track + /// with no album art). A per-channel clear is scheduled to its server timestamp exactly like + /// on_image_display(), ImageSlotPreference::display_offset_ms included, so it lands on the + /// item boundary rather than as soon as it arrives. + /// + /// Artwork stays valid until it is replaced or cleared, so the server does not resend an + /// unchanged image on every track: no callback at a track boundary means the image already + /// delivered still applies. /// @param slot The artwork slot index to clear. virtual void on_image_clear(uint8_t /*slot*/) {} }; diff --git a/src/artwork_role.cpp b/src/artwork_role.cpp index 0c9d07a..ea4f576 100644 --- a/src/artwork_role.cpp +++ b/src/artwork_role.cpp @@ -53,27 +53,6 @@ static int64_t be64_to_host(const uint8_t* bytes) { namespace sendspin { -namespace { - -/// @brief Merges a single-slot display delta into the accumulated cross-thread update -/// -/// Called under the Inbox mutex via InboxSlot::merge() (see Impl::drain_thread_func), so it must -/// stay a pure data operation with no callbacks into application code. `delta` carries exactly -/// one slot's bit (set by the decode thread after a single image finishes decoding); OR-ing -/// valid_mask and overwriting only the masked timestamps entries preserves latest-wins per slot -/// while leaving any other slot's already-accumulated (not yet drained) timestamp untouched. -void merge_artwork_display_update(ArtworkDisplayUpdate& current, ArtworkDisplayUpdate&& delta) { - current.valid_mask |= delta.valid_mask; - for (uint8_t slot = 0; slot < ARTWORK_MAX_SLOTS; ++slot) { - if (delta.valid_mask & (1U << slot)) { - current.timestamps[slot] = delta.timestamps[slot]; - current.epochs[slot] = delta.epochs[slot]; - } - } -} - -} // namespace - // ============================================================================ // ArtworkRole::Impl lifecycle // ============================================================================ @@ -148,6 +127,26 @@ void ArtworkRole::Impl::build_hello_fields(ClientHelloMessage& msg) const { // Display-deadline and ack-gate helpers (used from network, decode, and main threads) // ============================================================================ +void ArtworkRole::Impl::merge_artwork_display_update(ArtworkDisplayUpdate& current, + ArtworkDisplayUpdate&& delta) { + current.valid_mask |= delta.valid_mask; + for (uint8_t slot = 0; slot < ARTWORK_MAX_SLOTS; ++slot) { + const uint8_t bit = static_cast(1U << slot); + if (delta.valid_mask & bit) { + current.timestamps[slot] = delta.timestamps[slot]; + current.epochs[slot] = delta.epochs[slot]; + // clear_mask is assigned, not OR-ed: it says what kind of delivery this slot's + // (latest-wins) pending entry is, so a frame arriving after an undrained clear must + // reset the bit just as a clear after an undrained frame sets it. + if (delta.clear_mask & bit) { + current.clear_mask |= bit; + } else { + current.clear_mask &= static_cast(~bit); + } + } + } +} + int64_t ArtworkRole::Impl::display_overdue_us(int64_t client_ts, int32_t display_offset_ms, int64_t now) { // get_client_time returns 0 when there is no current connection. Without a connection we @@ -218,9 +217,16 @@ void ArtworkRole::Impl::handle_binary(uint8_t slot, const uint8_t* data, size_t image_format = this->config.preferred_formats[slot].format; } + // An empty payload -- a binary message carrying only the type byte and timestamp -- is the + // protocol's per-channel clear: the artwork on this channel is no longer valid, as distinct + // from the server simply not resending an image that still is. There are no bytes to stage, so + // the buffer machinery below is skipped entirely and the notification travels with + // data_length == 0 (see ArtworkNotification). It still goes through the decode thread rather + // than straight to the main loop, so it stays ordered behind any image already queued for this + // slot and takes the same ack-gate path a frame does. uint32_t generation = 0; uint8_t write_idx = 0; - { + if (image_len > 0) { // Hold the slot mutex across the read-modify-write of write_idx/drain_active/ // write_generation and the memcpy itself, so the decode thread can never observe a // buffer mid-write (torn image) and can never have a buffer stolen out from under it @@ -260,7 +266,8 @@ void ArtworkRole::Impl::handle_binary(uint8_t slot, const uint8_t* data, size_t ArtworkNotification notif{slot, write_idx, image_len, timestamp, image_format, generation, epoch}; if (!this->drain_task->notify_queue.send(notif, 0)) { - SS_LOGW(TAG, "Artwork notify queue full; dropping image for slot %u", slot); + SS_LOGW(TAG, "Artwork notify queue full; dropping %s for slot %u", + image_len > 0 ? "image" : "clear", slot); } } @@ -363,6 +370,7 @@ void ArtworkRole::Impl::handle_stream_ring_event(ArtworkEventType event) { case ArtworkEventType::STREAM_END: case ArtworkEventType::STREAM_CLEAR: this->held_display_mask = 0; + this->held_display_clear = 0; this->event_state->display_slot.reset(); { // A clear is itself a delivery that must be acked: it may drive a fade-out, and @@ -404,10 +412,21 @@ void ArtworkRole::Impl::drain_events() { ArtworkDisplayUpdate update{}; if (this->event_state->display_slot.take(update)) { for (uint8_t slot = 0; slot < ARTWORK_MAX_SLOTS; ++slot) { - if (update.valid_mask & (1U << slot)) { + const uint8_t bit = static_cast(1U << slot); + if (update.valid_mask & bit) { this->held_display_ts[slot] = update.timestamps[slot]; this->held_display_epoch[slot] = update.epochs[slot]; - this->held_display_mask |= static_cast(1U << slot); + this->held_display_mask |= bit; + // Assigned rather than OR-ed, for the same latest-wins reason as the cross-thread + // merge: this slot's held entry has just been replaced wholesale, so the kind of + // delivery it is must be replaced too. This mirrors + // merge_artwork_display_update(), which is unit-tested directly + // (ArtworkDisplayMerge); the two must stay in agreement. + if (update.clear_mask & bit) { + this->held_display_clear |= bit; + } else { + this->held_display_clear &= static_cast(~bit); + } } } } @@ -425,13 +444,15 @@ void ArtworkRole::Impl::drain_events() { const int64_t now = platform_time_us(); const uint32_t current_epoch = this->stream_epoch.load(std::memory_order_relaxed); for (uint8_t slot = 0; slot < ARTWORK_MAX_SLOTS; ++slot) { - if (!(this->held_display_mask & (1U << slot))) { + const uint8_t bit = static_cast(1U << slot); + if (!(this->held_display_mask & bit)) { continue; } // Drop a display decoded under a since-replaced stream (restart with no intervening // end/clear bumps the epoch but cannot reach these main-thread holds to cancel it). if (this->held_display_epoch[slot] != current_epoch) { - this->held_display_mask &= static_cast(~(1U << slot)); + this->held_display_mask &= static_cast(~bit); + this->held_display_clear &= static_cast(~bit); if (this->ack_enabled(slot)) { bool should_wake = false; { @@ -460,16 +481,25 @@ void ArtworkRole::Impl::drain_events() { if (overdue_us < 0) { continue; } - this->held_display_mask &= static_cast(~(1U << slot)); + this->held_display_mask &= static_cast(~bit); + // A per-channel clear is scheduled exactly like a frame, offset shift included, so a + // consumer can fade out on the same lead it would have faded in on. + const bool is_clear = (this->held_display_clear & bit) != 0; + this->held_display_clear &= static_cast(~bit); if (this->ack_enabled(slot)) { // Arm the "awaiting frame_done()" state before the callback fires and release the // mutex before invoking it: frame_done() may be called synchronously from inside - // on_image_display(), which would deadlock if this mutex were still held. + // on_image_display()/on_image_clear(), which would deadlock if this mutex were still + // held. std::lock_guard lock(this->drain_task->slot_mutex); this->drain_task->slot_buffers[slot].ack_state = SlotAckState::PRESENTED; } if (this->listener) { - this->listener->on_image_display(slot, display_lateness_ms(client_ts, overdue_us)); + if (is_clear) { + this->listener->on_image_clear(slot); + } else { + this->listener->on_image_display(slot, display_lateness_ms(client_ts, overdue_us)); + } } } } @@ -487,6 +517,7 @@ void ArtworkRole::Impl::cleanup() { // call, which runs before any role's cleanup() -- so there is no per-event ring reset to do // here. this->held_display_mask = 0; + this->held_display_clear = 0; this->event_state->display_slot.reset(); // Enqueue a clean STREAM_END - handle_stream_ring_event() will fire the on_image_clear() @@ -529,6 +560,13 @@ void ArtworkRole::Impl::process_notification(const ArtworkNotification& notif) { uint8_t slot = notif.slot; uint8_t buf_idx = notif.buffer_idx; + // A per-channel clear (see handle_binary) names no buffer, so it skips the buffer validation + // and the decode callback below. Everything else is deliberately shared with a frame: the same + // stream-epoch staleness check, the same ack gate (a clear is a delivery owing exactly one + // frame_done()), and the same timestamp-scheduled hand-off to the main loop, which fires + // on_image_clear() rather than on_image_display() when the deadline is reached. + const bool is_clear = notif.data_length == 0; + uint8_t* decode_data = nullptr; size_t decode_length = 0; { @@ -544,13 +582,13 @@ void ArtworkRole::Impl::process_notification(const ArtworkNotification& notif) { if (notif.stream_epoch != this->stream_epoch.load(std::memory_order_relaxed)) { return; } - if (notif.generation != sb.write_generation[buf_idx]) { - return; - } - - auto& buf = sb.buffers[buf_idx]; - if (notif.data_length == 0 || buf.data() == nullptr) { - return; + if (!is_clear) { + if (notif.generation != sb.write_generation[buf_idx]) { + return; + } + if (sb.buffers[buf_idx].data() == nullptr) { + return; + } } // Ack gate: a slot with require_frame_done set allows only one un-acked delivery in @@ -570,18 +608,20 @@ void ArtworkRole::Impl::process_notification(const ArtworkNotification& notif) { sb.ack_state = SlotAckState::DECODE_DELIVERED; } - // Mark this buffer as in-use so the network thread avoids it while we decode. - sb.drain_buf_idx = buf_idx; - sb.drain_active = true; - decode_data = buf.data(); - decode_length = notif.data_length; + if (!is_clear) { + // Mark this buffer as in-use so the network thread avoids it while we decode. + sb.drain_buf_idx = buf_idx; + sb.drain_active = true; + decode_data = sb.buffers[buf_idx].data(); + decode_length = notif.data_length; + } } - if (this->listener) { - this->listener->on_image_decode(slot, decode_data, decode_length, notif.format); - } + if (!is_clear) { + if (this->listener) { + this->listener->on_image_decode(slot, decode_data, decode_length, notif.format); + } - { std::lock_guard lock(this->drain_task->slot_mutex); this->drain_task->slot_buffers[slot].drain_active = false; } @@ -597,6 +637,9 @@ void ArtworkRole::Impl::process_notification(const ArtworkNotification& notif) { // the display if the stream is replaced after this hand-off (see held_display_epoch). delta.epochs[slot] = notif.stream_epoch; delta.valid_mask = static_cast(1U << slot); + if (is_clear) { + delta.clear_mask = static_cast(1U << slot); + } this->event_state->display_slot.merge(merge_artwork_display_update, delta); } } diff --git a/src/artwork_role_impl.h b/src/artwork_role_impl.h index f1851c9..5d96931 100644 --- a/src/artwork_role_impl.h +++ b/src/artwork_role_impl.h @@ -51,8 +51,9 @@ static constexpr uint8_t ARTWORK_RECHECK_SLOT = 0xFF; /// @brief Ack-gate state for a slot with require_frame_done enabled enum class SlotAckState : uint8_t { - IDLE, // no un-acked delivery; next frame may decode - DECODE_DELIVERED, // on_image_decode fired, on_image_display not yet fired + IDLE, // no un-acked delivery; next frame or clear may be processed + DECODE_DELIVERED, // the decode thread claimed a delivery (on_image_decode fired for a frame; + // nothing fires for a per-channel clear), its display/clear not yet fired PRESENTED, // on_image_display or on_image_clear fired, awaiting frame_done() }; @@ -67,6 +68,11 @@ enum class SlotAckState : uint8_t { /// buffer it names has since been overwritten (generation mismatch) or the stream has moved on /// (stream_epoch mismatch), the notification is skipped rather than decoding torn or /// superseded data. See ArtworkRole::Impl::drain_thread_func. +/// +/// `data_length == 0` marks the protocol's per-channel clear (an artwork binary message carrying +/// only the type byte and timestamp). It names no buffer, so `buffer_idx`/`generation` are unused +/// and left at 0; everything else about it -- queue ordering, the ack gate, and the +/// timestamp-scheduled hand-off to the main loop -- matches a frame. See handle_binary(). struct ArtworkNotification { uint8_t slot; uint8_t buffer_idx; @@ -108,11 +114,15 @@ struct SlotBuffer { /// main-loop drain; a bit set in valid_mask means timestamps[i] holds a pending display. /// epochs[i] carries the stream_epoch the decode ran under, so the main-loop deadline check can /// drop a display whose stream has since been replaced (a stream restart bumps the epoch but -/// cannot reach a display already folded into the main-thread holds). +/// cannot reach a display already folded into the main-thread holds). A bit set in clear_mask +/// means slot i's pending delivery is a per-channel clear rather than a decoded frame, so the +/// deadline fires on_image_clear() instead of on_image_display(); it is meaningful only where +/// valid_mask is set. struct ArtworkDisplayUpdate { int64_t timestamps[ARTWORK_MAX_SLOTS]{}; uint32_t epochs[ARTWORK_MAX_SLOTS]{}; uint8_t valid_mask{0}; + uint8_t clear_mask{0}; }; /// @brief Private implementation of the artwork role @@ -176,6 +186,17 @@ struct ArtworkRole::Impl { void stop() const; void enqueue_stream_event(ArtworkEventType event) const; + // Merges a single-slot display delta into the accumulated cross-thread update. Called under + // the Inbox mutex via InboxSlot::merge() (see process_notification), so it must stay a pure + // data operation with no callbacks into application code. `delta` carries exactly one slot's + // bit (set by the decode thread after a single image finishes decoding or a clear is + // validated); OR-ing valid_mask and overwriting only the masked entries preserves latest-wins + // per slot while leaving any other slot's already-accumulated (not yet drained) entry + // untouched. clear_mask is assigned per bit rather than OR-ed; drain_events() folds the taken + // update into the main-thread holds with the same per-bit assignment, so the two must agree. + // Pure and static for direct unit testing. + static void merge_artwork_display_update(ArtworkDisplayUpdate& current, + ArtworkDisplayUpdate&& delta); // How far past its display deadline a held slot is, in microseconds: >= 0 means due (the // value is the lateness reported to on_image_display), < 0 means not yet due. client_ts is // the server-clock deadline already converted to the client clock (0 = no connection: due @@ -236,6 +257,10 @@ struct ArtworkRole::Impl { std::atomic stream_active{false}; // Main-thread only; see held_display_ts. uint8_t held_display_mask{0}; + // Which held deliveries are per-channel clears rather than decoded frames: bit i selects + // on_image_clear() over on_image_display() when slot i's deadline fires. Only meaningful where + // held_display_mask is set. Main-thread only; see held_display_ts. + uint8_t held_display_clear{0}; /// @brief Bumped on stream start/end/clear/cleanup so in-flight notifications from a /// previous stream are recognized as stale and skipped by the decode thread, instead of diff --git a/tests/test_artwork_role.cpp b/tests/test_artwork_role.cpp index 2d93665..c557f8c 100644 --- a/tests/test_artwork_role.cpp +++ b/tests/test_artwork_role.cpp @@ -25,6 +25,7 @@ #include #include #include +#include #include using namespace sendspin; @@ -120,6 +121,13 @@ class RecordingListener : public ArtworkRoleListener { return this->clears.size(); } + // Slot recorded for the clear at `index`, used to tell a per-channel clear (one slot) from a + // stream-level one (every configured slot). + uint8_t clear_at(size_t index) { + std::lock_guard lock(this->mutex); + return this->clears.at(index); + } + // First byte of the payload decoded at `index`, used to identify which frame decoded. uint8_t decode_marker_at(size_t index) { std::lock_guard lock(this->mutex); @@ -204,6 +212,15 @@ void send_frame(ArtworkRole::Impl& impl, uint8_t slot, uint8_t marker, int64_t t impl.handle_binary(slot, data.data(), data.size()); } +// Sends a per-channel clear to `slot`: an artwork binary message carrying only the timestamp and +// no image bytes, which is how the server says the artwork on that channel is no longer valid +// (as opposed to simply not resending an image that still is). +void send_clear(ArtworkRole::Impl& impl, uint8_t slot, int64_t timestamp = 1) { + std::vector data; + put_be64(data, timestamp); + impl.handle_binary(slot, data.data(), data.size()); +} + // Polls drain_events() until `pred` is true or the timeout elapses. drain_events() must run on // the "main loop" thread (here, the test thread), so it cannot be driven from inside the // listener's condition variable wait -- it has to be called from an ordinary polling loop. @@ -220,6 +237,19 @@ bool poll_drain_until(ArtworkRole::Impl& impl, Pred pred, std::chrono::milliseco return pred(); } +// Asserts pred() stays false for the whole window while the main loop keeps draining: the +// drain-driving counterpart of RecordingListener::never_within(). A negative check on clears or +// displays must use this one rather than never_within(), because on_image_clear()/ +// on_image_display() fire only from drain_events() and handle_stream_ring_event(), both on this +// (main loop) thread -- a window that parks the test thread instead of driving the loop freezes +// the very counter it is watching, so the assertion could never fail. never_within() stays correct +// for decodes, which the decode thread produces on its own. Returns true if pred() never became +// true (the expected outcome). +template +bool poll_drain_never(ArtworkRole::Impl& impl, Pred pred, std::chrono::milliseconds window) { + return !poll_drain_until(impl, pred, window); +} + // Polls until `pred` (evaluated under impl.drain_task->slot_mutex) is true or the timeout // elapses. SlotBuffer::has_parked/ack_state are decode-thread-owned state with no listener // callback to hang a condition variable off of, so tests that need to synchronize with "the @@ -403,6 +433,207 @@ TEST(ArtworkFrameDoneGate, ClearGateHoldsNextStreamFirstFrame) { EXPECT_EQ(listener.decode_marker_at(1), 'B'); } +// ============================================================================ +// Per-channel clear: an artwork binary message with no image bytes clears just that channel, +// scheduled to its timestamp like any other delivery +// ============================================================================ + +TEST(ArtworkChannelClear, EmptyPayloadFiresClearWithoutDecoding) { + auto impl = make_impl(make_single_slot_config(false)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + send_clear(*impl, 0); + + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.clear_count() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.clear_at(0), 0); + EXPECT_TRUE( + poll_drain_never(*impl, [&] { return listener.clear_count() >= 2; }, NEGATIVE_WINDOW)); + // There are no image bytes, so nothing may reach the decode callback -- and nothing may be + // presented as a frame either. + EXPECT_EQ(listener.decode_count(), 0U); + EXPECT_EQ(listener.display_count(), 0U); +} + +TEST(ArtworkChannelClear, ClearAfterDisplayedFrameFiresAgain) { + auto impl = make_impl(make_single_slot_config(false)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + // The album's first track: artwork arrives and is displayed. + send_frame(*impl, 0, 'A'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 1; }, POSITIVE_TIMEOUT)); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.display_count() >= 1; }, POSITIVE_TIMEOUT)); + + // A later track with no artwork of its own: the clear must reach the listener while the + // stream is still running, so a consumer can tell "no artwork for this item" apart from + // "artwork unchanged, nothing sent". + send_clear(*impl, 0); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.clear_count() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_TRUE( + poll_drain_never(*impl, [&] { return listener.clear_count() >= 2; }, NEGATIVE_WINDOW)); + EXPECT_EQ(listener.display_count(), 1U); + EXPECT_EQ(listener.decode_count(), 1U); +} + +TEST(ArtworkChannelClear, ClearOnlyAffectsItsOwnSlot) { + auto impl = make_impl(make_two_slot_config()); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + // Slot 1 (ungated) is cleared; slot 0 (gated) must be left alone entirely -- a stream-level + // clear fires for every configured slot, a per-channel clear for exactly one. + send_clear(*impl, 1); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.clear_count() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.clear_at(0), 1); + EXPECT_TRUE( + poll_drain_never(*impl, [&] { return listener.clear_count() >= 2; }, NEGATIVE_WINDOW)); + + // Slot 0's gate was never armed by slot 1's clear, so its frame decodes without any ack. + send_frame(*impl, 0, 'A'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(0), 'A'); +} + +TEST(ArtworkChannelClear, GatedClearParksBehindUnackedFrame) { + auto impl = make_impl(make_single_slot_config(true)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + send_frame(*impl, 0, 'A'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 1; }, POSITIVE_TIMEOUT)); + + // A's delivery is un-acked, so the clear parks rather than overtaking it: the consumer is + // mid-presentation of A and its buffers must not be disturbed. + send_clear(*impl, 0); + ASSERT_TRUE(wait_slot_state( + *impl, [&] { return impl->drain_task->slot_buffers[0].has_parked; }, POSITIVE_TIMEOUT)); + EXPECT_TRUE( + poll_drain_never(*impl, [&] { return listener.clear_count() >= 1; }, NEGATIVE_WINDOW)); + + // A's own display still fires; only then does acking it release the parked clear. + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.display_count() >= 1; }, POSITIVE_TIMEOUT)); + impl->frame_done(0); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.clear_count() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.clear_at(0), 0); + EXPECT_TRUE( + poll_drain_never(*impl, [&] { return listener.clear_count() >= 2; }, NEGATIVE_WINDOW)); +} + +TEST(ArtworkChannelClear, GatedClearOwesExactlyOneAck) { + auto impl = make_impl(make_single_slot_config(true)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + send_clear(*impl, 0); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.clear_count() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_TRUE( + poll_drain_never(*impl, [&] { return listener.clear_count() >= 2; }, NEGATIVE_WINDOW)); + + // The clear is a delivery like any frame, so it holds the gate until it is acked. + send_frame(*impl, 0, 'A'); + EXPECT_TRUE( + listener.never_within([&] { return !listener.decodes.empty(); }, NEGATIVE_WINDOW)); + + impl->frame_done(0); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(0), 'A'); +} + +TEST(ArtworkChannelClear, GatedClearSupersedesParkedClear) { + auto impl = make_impl(make_single_slot_config(true)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + send_frame(*impl, 0, 'A'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 1; }, POSITIVE_TIMEOUT)); + + // Two clears arrive back to back while A is un-acked. Both park, and the second must overwrite + // the first (latest-wins) rather than queue behind it, so the consumer is asked to clear once + // rather than twice. Distinct timestamps make the handoff observable: waiting for the parked + // notification to carry the second clear's timestamp is what keeps this deterministic, since + // has_parked is already true from the first. + send_clear(*impl, 0, /*timestamp=*/1); + ASSERT_TRUE(wait_slot_state( + *impl, [&] { return impl->drain_task->slot_buffers[0].has_parked; }, POSITIVE_TIMEOUT)); + send_clear(*impl, 0, /*timestamp=*/2); + ASSERT_TRUE(wait_slot_state( + *impl, [&] { return impl->drain_task->slot_buffers[0].parked.timestamp == 2; }, + POSITIVE_TIMEOUT)); + + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.display_count() >= 1; }, POSITIVE_TIMEOUT)); + impl->frame_done(0); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.clear_count() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_TRUE( + poll_drain_never(*impl, [&] { return listener.clear_count() >= 2; }, NEGATIVE_WINDOW)); +} + +TEST(ArtworkChannelClear, StreamEndOnTopOfUnackedChannelClearFiresAgain) { + auto impl = make_impl(make_single_slot_config(true)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + // A per-channel clear is delivered and left un-acked, e.g. the consumer is running a fade-out. + send_clear(*impl, 0); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.clear_count() >= 1; }, POSITIVE_TIMEOUT)); + + // The queue then ends. stream/end is a distinct lifecycle event, so it fires on_image_clear() + // again rather than being swallowed because a clear is already outstanding -- it supersedes + // that clear the same way it supersedes an un-acked frame. + impl->handle_stream_ring_event(ArtworkEventType::STREAM_END); + ASSERT_TRUE(listener.wait_for([&] { return listener.clears.size() >= 2; }, POSITIVE_TIMEOUT)); + + // Superseded, not stacked: exactly one ack is owed for the two clears, so a single frame_done() + // releases the gate for the next stream's first frame. + impl->handle_stream_start(ServerArtworkStreamObject{}); + send_frame(*impl, 0, 'A'); + EXPECT_TRUE(listener.never_within([&] { return !listener.decodes.empty(); }, NEGATIVE_WINDOW)); + + impl->frame_done(0); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 1; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(0), 'A'); +} + +TEST(ArtworkChannelClear, ClearIgnoredWithoutActiveStream) { + auto impl = make_impl(make_single_slot_config(false)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + + // No stream/start yet, so handle_binary()'s stream_active guard rejects the message before any + // clear-specific handling runs. That guard is not new, so unlike the tests above this one does + // not fail without the per-channel clear path -- it pins that the clear path stays behind the + // guard rather than short-circuiting ahead of it. + send_clear(*impl, 0); + EXPECT_TRUE( + poll_drain_never(*impl, [&] { return listener.clear_count() >= 1; }, NEGATIVE_WINDOW)); + EXPECT_EQ(listener.clear_count(), 0U); +} + // ============================================================================ // frame_done() edge cases // ============================================================================ @@ -548,6 +779,98 @@ TEST(ArtworkFrameDoneGate, UngatedSlotUnaffectedBesideGatedSlot) { EXPECT_EQ(listener.decode_count_for_slot(0), 1U); } +// ============================================================================ +// merge_artwork_display_update: the cross-thread latest-wins accumulation the decode thread runs +// under the Inbox mutex. Pure function, so tested directly: reaching it end to end needs two +// same-slot deliveries to accumulate before the main loop takes the slot, and the integration +// tests above all run without a connection (client_ts == 0), so every folded-in entry fires in +// the same drain_events() call that folds it in and nothing is ever left pending to replace. +// ============================================================================ + +namespace { + +// A single-slot delta shaped like the one process_notification() publishes. +ArtworkDisplayUpdate make_delta(uint8_t slot, int64_t timestamp, uint32_t epoch, bool is_clear) { + ArtworkDisplayUpdate delta{}; + const auto bit = static_cast(1U << slot); + delta.timestamps[slot] = timestamp; + delta.epochs[slot] = epoch; + delta.valid_mask = bit; + if (is_clear) { + delta.clear_mask = bit; + } + return delta; +} + +void merge_into(ArtworkDisplayUpdate& current, ArtworkDisplayUpdate delta) { + ArtworkRole::Impl::merge_artwork_display_update(current, std::move(delta)); +} + +} // namespace + +TEST(ArtworkDisplayMerge, FrameAfterUndrainedClearResetsKind) { + // The case the assigned-not-OR-ed clear_mask exists for: an item with no artwork is cleared + // and the next item's frame lands before the main loop drains. The pending entry is now a + // frame, so the bit must be reset -- OR-ing it would fire on_image_clear() for a decoded + // image, blanking the display and dropping the frame. + ArtworkDisplayUpdate current{}; + merge_into(current, make_delta(0, 100, 7, /*is_clear=*/true)); + ASSERT_EQ(current.clear_mask, 0x01); + + merge_into(current, make_delta(0, 200, 8, /*is_clear=*/false)); + EXPECT_EQ(current.valid_mask, 0x01); + EXPECT_EQ(current.clear_mask, 0x00); + EXPECT_EQ(current.timestamps[0], 200); + EXPECT_EQ(current.epochs[0], 8U); +} + +TEST(ArtworkDisplayMerge, ClearAfterUndrainedFrameSetsKind) { + ArtworkDisplayUpdate current{}; + merge_into(current, make_delta(0, 100, 7, /*is_clear=*/false)); + ASSERT_EQ(current.clear_mask, 0x00); + + merge_into(current, make_delta(0, 200, 7, /*is_clear=*/true)); + EXPECT_EQ(current.valid_mask, 0x01); + EXPECT_EQ(current.clear_mask, 0x01); + EXPECT_EQ(current.timestamps[0], 200); +} + +TEST(ArtworkDisplayMerge, SameKindReplacementsKeepTheirKind) { + ArtworkDisplayUpdate clears{}; + merge_into(clears, make_delta(0, 100, 7, /*is_clear=*/true)); + merge_into(clears, make_delta(0, 200, 7, /*is_clear=*/true)); + EXPECT_EQ(clears.clear_mask, 0x01); + EXPECT_EQ(clears.timestamps[0], 200); + + ArtworkDisplayUpdate frames{}; + merge_into(frames, make_delta(0, 100, 7, /*is_clear=*/false)); + merge_into(frames, make_delta(0, 200, 7, /*is_clear=*/false)); + EXPECT_EQ(frames.clear_mask, 0x00); + EXPECT_EQ(frames.timestamps[0], 200); +} + +TEST(ArtworkDisplayMerge, OtherSlotsAreUntouched) { + // Latest-wins is per slot: a delta carries exactly one slot's bit and must leave every other + // slot's accumulated entry -- timestamp, epoch, and kind alike -- alone. + ArtworkDisplayUpdate current{}; + merge_into(current, make_delta(1, 100, 7, /*is_clear=*/true)); + merge_into(current, make_delta(0, 200, 8, /*is_clear=*/false)); + + EXPECT_EQ(current.valid_mask, 0x03); + EXPECT_EQ(current.clear_mask, 0x02); + EXPECT_EQ(current.timestamps[1], 100); + EXPECT_EQ(current.epochs[1], 7U); + EXPECT_EQ(current.timestamps[0], 200); + EXPECT_EQ(current.epochs[0], 8U); + + // And the reverse: slot 0's clear must not disturb slot 1's pending frame. + ArtworkDisplayUpdate reverse{}; + merge_into(reverse, make_delta(1, 100, 7, /*is_clear=*/false)); + merge_into(reverse, make_delta(0, 200, 7, /*is_clear=*/true)); + EXPECT_EQ(reverse.valid_mask, 0x03); + EXPECT_EQ(reverse.clear_mask, 0x01); +} + // ============================================================================ // display_overdue_us: the drain_events() display-deadline arithmetic, including the per-slot // display_offset_ms shift and the lateness (>= 0 overdue) value reported to on_image_display.