diff --git a/docs/integration-guide.md b/docs/integration-guide.md index 2bc6c6a..d1a4070 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -271,6 +271,8 @@ struct MyControllerListener : ControllerRoleListener { The artwork role uses a dedicated decode thread for the CPU-bound decode step and the main loop for scheduled display. `on_image_decode()` fires on the decode thread immediately when encoded image data arrives; once decode returns, the server display timestamp is handed off to the main loop, which fires `on_image_display()` once the timestamp is reached. If a newer frame for the same slot finishes decoding before its predecessor's display fires, only the newer one is delivered. Lifecycle callbacks also fire on the main loop thread. +`on_image_display()` reports `lateness_ms`: how far past the (offset-shifted) deadline it fired. Displays are best-effort, so an image that arrives or decodes after its deadline fires as soon as it is ready. Treat a small value as on time; a huge value is the cue to snap instantly. `lateness_ms` is `0` only when there is no connection (no deadline exists), so a connected on-time display always reports a small nonzero value. + ```cpp struct MyArtworkListener : ArtworkRoleListener { // THREAD SAFETY: Called from the dedicated decode thread. @@ -282,8 +284,8 @@ struct MyArtworkListener : ArtworkRoleListener { } // Called from the main loop thread once the server display timestamp is reached. - // Swap the decoded image onto the display. - void on_image_display(uint8_t slot) override { + // lateness_ms reports how late the display fired (0 = no connection). + void on_image_display(uint8_t slot, uint32_t lateness_ms) override { display.show_image(slot, decoded_images[slot]); } @@ -294,6 +296,27 @@ struct MyArtworkListener : ArtworkRoleListener { }; ``` +**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. + +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: + +```cpp +// Slot 0 has require_frame_done set, so on_image_display() starts a cross-fade and the gate +// stays held until on_fade_complete() acks it. +void on_image_display(uint8_t slot, uint32_t lateness_ms) override { + display.start_fade(slot, decoded_images[slot], FADE_MS - std::min(lateness_ms, FADE_MS)); +} +void on_image_clear(uint8_t slot) override { + display.clear_slot(slot); + artwork_role->frame_done(slot); // a clear is a delivery; ack it +} +void on_fade_complete(uint8_t slot) { + artwork_role->frame_done(slot); // release the gate so the next frame can decode +} +``` + +Call `frame_done()` from the main loop thread. It is a safe no-op when the slot has nothing un-acked (including slots where `require_frame_done` is false), so calling it from inside `on_image_display()`/`on_image_clear()` for an instant, non-animated swap is fine. + ### VisualizerRoleListener ```cpp @@ -608,6 +631,8 @@ Most listener callbacks fire on the main loop thread (the thread calling `client `PlayerRole::notify_audio_played()` is thread-safe and is designed to be called from an audio output callback thread. +`ArtworkRole::frame_done()` must be called from the main loop thread (typically from inside `on_image_display()`/`on_image_clear()` or when a cross-fade animation completes). + ## Minimal Example A minimal integration that receives and discards audio: @@ -777,6 +802,8 @@ Each entry in `preferred_formats` is an `ImageSlotPreference`. The slot/channel | `format` | `SendspinImageFormat` | Image format (`JPEG`, `PNG`, or `BMP`) | | `width` | `uint16_t` | Desired image width in pixels | | `height` | `uint16_t` | Desired image height in pixels | +| `require_frame_done` | `bool` | Opt-in back-pressure gate (default `false`). When set, the role delivers at most one un-acked frame or clear at a time for this slot; the consumer must call `ArtworkRole::frame_done(slot)` to release the gate. See [ArtworkRoleListener](#artworkrolelistener). | +| `display_offset_ms` | `int32_t` | Shifts the display deadline (default `0`). Positive fires `on_image_display()` earlier (mirroring `PlayerRoleConfig::fixed_delay_us`), negative delays it; lets a cross-fade straddle the track boundary. | --- diff --git a/docs/internals.md b/docs/internals.md index b47c795..1e089a4 100644 --- a/docs/internals.md +++ b/docs/internals.md @@ -287,7 +287,8 @@ The `awaiting_sync_idle_events` list (on `PlayerRole::Impl`) is the key ordering - **ControllerRole**: Takes the latest `ServerStateControllerObject` from its `InboxSlot`, fires `on_controller_state()`. The disconnect clear is not handled here; it arrives as a `CONTROLLER_CLEARED` event on the shared ring, whose `handle_cleared_event()` fires `on_controller_state_clear()` (deferred from `cleanup()` to avoid invoking the listener while `ConnectionManager` holds `conn_ptr_mutex_`). - **MetadataRole**: `InboxSlot` has no `take_if`, so the deadline gate that used to run under the shadow slot's mutex is split in two: `take()` unconditionally moves any pending delta into a main-thread-only `held_delta` (folding it into an already-held delta), then the server-clock deadline is evaluated with no lock held, applying deltas and firing `on_metadata()` once the `timestamp` is reached (or immediately if there is no active connection). A future-dated `held_delta` persists across ticks with no topic bit set, which is why `needs_drain()` ORs in `held_delta.has_value()` alongside the `INBOX_TOPIC_METADATA` bit test: the deadline sets no inbox bit, so without that term a bit-gated tick would strand the delta until an unrelated new delta happened to re-set the bit, silently starving deadline-based delivery. The clear arrives separately as a `METADATA_CLEARED` ring event (`handle_cleared_event()` fires `on_metadata_clear()`), deferred from `cleanup()` for the same `conn_ptr_mutex_` reason. - **ColorRole**: Same structure as MetadataRole: `take()` into `held_delta`, a lock-free server-clock deadline gate firing `on_color()`, and a `COLOR_CLEARED` ring event driving `on_color_clear()`. -- **ArtworkRole**: Stream end/clear lifecycle is handled earlier in the tick by `handle_stream_ring_event()` (dispatched from the ring drain, before this call), which clears `held_display_mask`/`display_slot` and fires `on_image_clear()` for each configured slot - preserving the "lifecycle before display" ordering the old single-function drain guaranteed. `drain_events()` itself folds any taken `display_slot` update into the main-thread-only `held_display_*` state (latest-wins per slot), then sweeps the held slots and fires `on_image_display(slot)` for any whose timestamp is due on the synced client clock (or immediately if there is no active connection). Per-slot epochs drop a held display whose stream was replaced after the decode hand-off. `needs_drain()` ORs a nonzero `held_display_mask` into the `INBOX_TOPIC_ARTWORK_DISPLAY` bit test (the same carry-over pattern the metadata role uses for `held_delta`) so held displays keep getting a drain every tick until their deadline fires, even though the deadline sets no inbox bit; `on_image_decode` still happens on the dedicated artwork decode thread. +- **ArtworkRole**: Stream end/clear lifecycle is handled earlier in the tick by `handle_stream_ring_event()` (dispatched from the ring drain, before this call), which clears `held_display_mask`/`display_slot` and fires `on_image_clear()` for each configured slot - preserving the "lifecycle before display" ordering the old single-function drain guaranteed. `drain_events()` itself folds any taken `display_slot` update into the main-thread-only `held_display_*` state (latest-wins per slot), then sweeps the held slots and fires `on_image_display(slot, lateness_ms)` for any whose timestamp is due on the synced client clock (or immediately if there is no active connection). The deadline is computed by the pure `display_overdue_us()` helper, which applies the slot's `display_offset_ms` shift (positive fires early) and returns the overdue microseconds; `display_lateness_ms()` maps that to the `lateness_ms` argument, reserving `0` for the no-connection case (a connected on-time display is floored to 1 ms so it never collides with that sentinel). Per-slot epochs drop a held display whose stream was replaced after the decode hand-off. `needs_drain()` ORs a nonzero `held_display_mask` into the `INBOX_TOPIC_ARTWORK_DISPLAY` bit test (the same carry-over pattern the metadata role uses for `held_delta`) so held displays keep getting a drain every tick until their deadline fires, even though the deadline sets no inbox bit; `on_image_decode` still happens on the dedicated artwork decode thread. + - **Ack gate (`require_frame_done`)**: A slot can opt into per-slot back-pressure. Each `SlotBuffer` carries a `SlotAckState` (`IDLE` -> `DECODE_DELIVERED` once `on_image_decode()` fires -> `PRESENTED` once `on_image_display()`/`on_image_clear()` fires), all guarded by `slot_mutex`. While a gated slot is not `IDLE`, the decode thread (`process_notification()`) does not decode a newer notification; it *parks* it latest-wins in `SlotBuffer::parked` (`has_parked`) instead of decoding concurrently with the un-acked delivery. `ArtworkRole::frame_done(slot)` (main loop) returns the gate to `IDLE` and, if a notification is parked, calls `wake_drain_thread()` -- a sentinel `ARTWORK_RECHECK_SLOT` notification that unblocks the decode thread's `notify_queue.receive()` so it re-runs the top-of-loop parked-slot sweep (a dropped wake is covered by the `DRAIN_RECEIVE_TIMEOUT_MS` fallback). The parked notification is re-validated on replay, so a since-stale generation/epoch is simply skipped. A clear counts as a delivery: `handle_stream_ring_event()` drops any parked notification and forces gated slots to `PRESENTED`, so exactly one `frame_done()` is owed after it. A stream restart releases only `DECODE_DELIVERED` slots (their display can no longer fire); `PRESENTED` stays armed because the consumer may still be mid-fade on the prior stream's last delivery. There is no timeout. - **VisualizerRole**: Has no `drain_events()`. STREAM_START/END/CLEAR are dispatched entirely from `handle_stream_ring_event()` (from the ring drain): STREAM_START `take()`s the config from `config_slot` and fires `on_visualizer_stream_start()`; STREAM_END/CLEAR fire `on_visualizer_stream_end()`/`on_visualizer_stream_clear()`. ## Sync Task State Machine diff --git a/include/sendspin/artwork_role.h b/include/sendspin/artwork_role.h index 4744f23..833d17b 100644 --- a/include/sendspin/artwork_role.h +++ b/include/sendspin/artwork_role.h @@ -32,6 +32,16 @@ class SendspinClient; /// THREAD SAFETY: on_image_decode() fires on a dedicated decode thread and must be /// thread-safe with respect to the other callbacks. on_image_display() and on_image_clear() /// fire on the main loop thread. +/// +/// 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. class ArtworkRoleListener { public: virtual ~ArtworkRoleListener() = default; @@ -50,11 +60,20 @@ class ArtworkRoleListener { /// @brief Called on the main loop thread at the correct timestamp when the decoded image /// should be displayed /// - /// Fires after on_image_decode() once the server timestamp is reached. If a newer frame for - /// the same slot finishes decoding before the pending display fires, the older pending - /// display is superseded and only the newer one is delivered. + /// Fires after on_image_decode() once the server timestamp is reached. The deadline can be + /// shifted per slot via ImageSlotPreference::display_offset_ms (positive fires early, e.g. + /// to start a cross-fade before the track boundary). If a newer frame for the same slot + /// finishes decoding before the pending display fires, the older pending display is + /// superseded and only the newer one is delivered. /// @param slot The artwork slot index. - virtual void on_image_display(uint8_t /*slot*/) {} + /// @param lateness_ms How far past the (offset-shifted) deadline this display fired. Displays + /// are best-effort: an image that arrives or decodes after its deadline fires as soon as it + /// is ready, and lateness_ms reports the slip so a consumer can compensate (e.g. shorten a + /// cross-fade by the lateness so it still ends on schedule, or snap instantly on a huge + /// value). On-time displays report a few milliseconds of main-loop polling granularity, never + /// exactly 0, so treat small values as on time. Reports 0 when there is no connection, since + /// no deadline exists. + virtual void on_image_display(uint8_t /*slot*/, uint32_t /*lateness_ms*/) {} /// @brief Called on the main loop thread when artwork should be cleared for a slot /// @@ -72,6 +91,10 @@ class ArtworkRoleListener { * loop thread, with on_image_display() scheduled to the server timestamp. Supports multiple * image slots with configurable format and resolution preferences. * + * A slot may opt into a back-pressure gate via ImageSlotPreference::require_frame_done: see + * the ArtworkRoleListener class comment for the ack contract. Call frame_done() once the + * consumer has finished presenting a delivery for such a slot. + * * Usage: * 1. Implement ArtworkRoleListener with on_image_decode() and on_image_display() * 2. Build an ArtworkRoleConfig with the desired slot/format/resolution preferences @@ -84,19 +107,29 @@ class ArtworkRoleListener { * SendspinImageFormat format) override { * decoded_images[slot] = decode(data, length, format); * } - * void on_image_display(uint8_t slot) override { - * display.show_image(slot, decoded_images[slot]); + * void on_image_display(uint8_t slot, uint32_t lateness_ms) override { + * // Slot 0 has require_frame_done set, so this starts a cross-fade; frame_done() is + * // called once the fade finishes instead of immediately. Shortening the fade by the + * // lateness keeps it ending on schedule even when the image arrived late. + * display.start_fade(slot, decoded_images[slot], FADE_MS - std::min(lateness_ms, FADE_MS)); * } * void on_image_clear(uint8_t slot) override { * display.clear_slot(slot); + * artwork_role->frame_done(slot); + * } + * void on_fade_complete(uint8_t slot) { + * artwork_role->frame_done(slot); * } + * + * ArtworkRole* artwork_role{nullptr}; * }; * * MyArtworkListener listener; * ArtworkRoleConfig config; * config.preferred_formats = {{SendspinImageSource::ALBUM, - * SendspinImageFormat::JPEG, 240, 240}}; + * SendspinImageFormat::JPEG, 240, 240, true}}; * auto& artwork = client.add_artwork(config); + * listener.artwork_role = &artwork; * artwork.set_listener(&listener); * @endcode */ @@ -114,6 +147,16 @@ class ArtworkRole { /// @param listener Pointer to the listener implementation; must outlive this role void set_listener(ArtworkRoleListener* listener); + /// @brief Acknowledges the most recent delivery for an ack-gated slot, releasing the gate + /// + /// Call from the main loop thread after finishing presentation of the most recent delivery + /// (frame or clear) for a slot with ImageSlotPreference::require_frame_done set, e.g. once a + /// cross-fade animation completes. Safe no-op if the slot has nothing un-acked (including + /// slots where require_frame_done is false). Also safe to call from inside + /// on_image_display() or on_image_clear() for instant (non-animated) presentation. + /// @param slot The artwork slot index to acknowledge. + void frame_done(uint8_t slot); + private: std::unique_ptr impl_; }; diff --git a/include/sendspin/config.h b/include/sendspin/config.h index fc18d2f..89977c6 100644 --- a/include/sendspin/config.h +++ b/include/sendspin/config.h @@ -181,6 +181,23 @@ struct ImageSlotPreference { SendspinImageFormat format{}; uint16_t width{}; uint16_t height{}; + + /// @brief Opt-in per-slot back-pressure gate. When true, the role delivers at most one + /// un-acked "delivery" at a time for this slot: a delivery is either a frame + /// (on_image_decode() followed by on_image_display()) or a clear (on_image_clear()). While a + /// delivery is un-acked, any newer payload that arrives is buffered latest-wins and only + /// delivered once the consumer calls ArtworkRole::frame_done(slot) from the main loop (e.g. + /// after a cross-fade animation completes). Defaults to false, which preserves today's + /// behavior of decoding and displaying every frame as it arrives. + bool require_frame_done{false}; + + /// @brief Fires on_image_display() this many milliseconds before the server's display + /// timestamp (negative delays it). Lets a cross-fade straddle the track boundary: with a + /// 2 s fade, an offset of 1000 starts the fade 1 s before the boundary so the incoming image + /// is fully shown 1 s after it. Positive-equals-earlier mirrors + /// PlayerRoleConfig::fixed_delay_us. Best-effort: an image that arrives or decodes after the + /// offset deadline fires as soon as it is ready, same as any past-timestamp display. + int32_t display_offset_ms{0}; }; /// @brief Configuration for the artwork role diff --git a/src/artwork_role.cpp b/src/artwork_role.cpp index 863af84..0c9d07a 100644 --- a/src/artwork_role.cpp +++ b/src/artwork_role.cpp @@ -144,6 +144,53 @@ void ArtworkRole::Impl::build_hello_fields(ClientHelloMessage& msg) const { msg.artwork_v1_support = artwork_support; } +// ============================================================================ +// Display-deadline and ack-gate helpers (used from network, decode, and main threads) +// ============================================================================ + +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 + // cannot honor the server-clock deadline, so fire immediately rather than starving the + // listener; the lateness is 0 by definition since no deadline exists. The check must precede + // the offset shift so the sentinel is never mistaken for a real deadline. + if (client_ts == 0) { + return 0; + } + // Positive display_offset_ms fires the display early (mirroring + // PlayerRoleConfig::fixed_delay_us), negative delays it; see ImageSlotPreference. + return now - (client_ts - static_cast(display_offset_ms) * US_PER_MS); +} + +uint32_t ArtworkRole::Impl::display_lateness_ms(int64_t client_ts, int64_t overdue_us) { + // No connection: no deadline exists, so report the documented 0 sentinel (see + // display_overdue_us and on_image_display's contract). + if (client_ts == 0) { + return 0; + } + // Connected: floor at 1 ms. A display firing under a millisecond late truncates to 0 ms, + // which would collide with the no-connection sentinel above; on-time displays must report a + // small nonzero value, never exactly 0. Clamp the top so a huge lateness (~49 days) saturates + // instead of wrapping. + int64_t ms = std::min(overdue_us / US_PER_MS, UINT32_MAX); + return static_cast(std::max(ms, 1)); +} + +bool ArtworkRole::Impl::ack_enabled(uint8_t slot) const { + return slot < this->config.preferred_formats.size() && + this->config.preferred_formats[slot].require_frame_done; +} + +void ArtworkRole::Impl::wake_drain_thread() const { + // Best-effort wakeup: a dropped send just means the decode thread's own + // DRAIN_RECEIVE_TIMEOUT_MS receive timeout, plus the parked-slot sweep it runs at the top + // of every loop iteration, picks up the parked notification a little later instead of + // immediately. + ArtworkNotification wake{}; + wake.slot = ARTWORK_RECHECK_SLOT; + this->drain_task->notify_queue.send(wake, 0); +} + // ============================================================================ // Binary handling (network thread) // ============================================================================ @@ -265,6 +312,22 @@ void ArtworkRole::Impl::handle_stream_start(const ServerArtworkStreamObject& str // holds is not reachable from here, but it carries the epoch it was decoded under // (held_display_epoch), so the epoch bump above makes the main-loop deadline check drop it. this->event_state->display_slot.reset(); + + { + // Release any DECODE_DELIVERED ack gate: display_slot was just reset and the epoch was + // just bumped, so that decode's eventual display can no longer fire, and leaving the + // gate armed would wedge the slot forever. PRESENTED must stay armed here: the consumer + // may still be mid-fade on the previous stream's last delivery, and its buffers must not + // be disturbed until frame_done() is called. Protocol messages are serialized on the + // network thread, so this runs before any of the new stream's handle_binary() calls. + std::lock_guard lock(this->drain_task->slot_mutex); + for (auto& sb : this->drain_task->slot_buffers) { + sb.has_parked = false; + if (sb.ack_state == SlotAckState::DECODE_DELIVERED) { + sb.ack_state = SlotAckState::IDLE; + } + } + } } void ArtworkRole::Impl::handle_stream_end() { @@ -301,6 +364,26 @@ void ArtworkRole::Impl::handle_stream_ring_event(ArtworkEventType event) { case ArtworkEventType::STREAM_CLEAR: this->held_display_mask = 0; this->event_state->display_slot.reset(); + { + // A clear is itself a delivery that must be acked: it may drive a fade-out, and + // it supersedes any un-acked frame for the slot, so exactly one frame_done() is + // owed afterward regardless of what ack_state held before. Drop any notification + // parked behind an un-acked frame -- it is superseded by the clear. Released + // before firing the callbacks below so a listener calling frame_done() from + // inside on_image_clear() does not deadlock on this same mutex. + std::lock_guard lock(this->drain_task->slot_mutex); + // Sweep the whole fixed-size slot_buffers array (ARTWORK_MAX_SLOTS), matching + // handle_stream_start(): ack_enabled() already gates the PRESENTED arm to + // configured ack slots, and clearing has_parked on any others is a harmless reset + // (they never park). + for (size_t i = 0; i < ARTWORK_MAX_SLOTS; ++i) { + auto& sb = this->drain_task->slot_buffers[i]; + sb.has_parked = false; + if (this->ack_enabled(static_cast(i))) { + sb.ack_state = SlotAckState::PRESENTED; + } + } + } if (this->listener) { // Array index is the authoritative slot number; see the Impl constructor. for (size_t i = 0; i < this->config.preferred_formats.size(); ++i) { @@ -349,18 +432,44 @@ void ArtworkRole::Impl::drain_events() { // 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)); + if (this->ack_enabled(slot)) { + bool should_wake = false; + { + std::lock_guard lock(this->drain_task->slot_mutex); + auto& sb = this->drain_task->slot_buffers[slot]; + // The consumer got a decode whose display will never fire now; release the + // gate so the slot does not wedge on this stream restart. PRESENTED is left + // untouched: a delivery that already reached on_image_display()/ + // on_image_clear() still owes its frame_done() regardless of epoch. + if (sb.ack_state == SlotAckState::DECODE_DELIVERED) { + sb.ack_state = SlotAckState::IDLE; + } + should_wake = sb.has_parked; + } + if (should_wake) { + this->wake_drain_thread(); + } + } continue; } - // get_client_time returns 0 when there is no current connection. Without a connection we - // cannot honor the server-clock deadline, so fire immediately rather than starving the - // listener. int64_t client_ts = this->client->get_client_time(this->held_display_ts[slot]); - if (client_ts != 0 && client_ts > now) { + int32_t display_offset_ms = slot < this->config.preferred_formats.size() + ? this->config.preferred_formats[slot].display_offset_ms + : 0; + int64_t overdue_us = display_overdue_us(client_ts, display_offset_ms, now); + if (overdue_us < 0) { continue; } this->held_display_mask &= static_cast(~(1U << slot)); + 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. + 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); + this->listener->on_image_display(slot, display_lateness_ms(client_ts, overdue_us)); } } } @@ -386,10 +495,112 @@ void ArtworkRole::Impl::cleanup() { this->enqueue_stream_event(ArtworkEventType::STREAM_END); } +// ============================================================================ +// Consumer-facing methods (main thread) +// ============================================================================ + +void ArtworkRole::Impl::frame_done(uint8_t slot) const { + if (slot >= ARTWORK_MAX_SLOTS) { + return; + } + + bool should_wake = false; + { + std::lock_guard lock(this->drain_task->slot_mutex); + auto& sb = this->drain_task->slot_buffers[slot]; + if (sb.ack_state == SlotAckState::IDLE) { + // Safe no-op: nothing un-acked for this slot, whether because require_frame_done is + // disabled, the delivery was already acked, or a clear already acked it for us. + return; + } + sb.ack_state = SlotAckState::IDLE; + should_wake = sb.has_parked; + } + if (should_wake) { + this->wake_drain_thread(); + } +} + // ============================================================================ // Decode thread // ============================================================================ +void ArtworkRole::Impl::process_notification(const ArtworkNotification& notif) { + uint8_t slot = notif.slot; + uint8_t buf_idx = notif.buffer_idx; + + uint8_t* decode_data = nullptr; + size_t decode_length = 0; + { + // Validate the notification is still current before touching the buffer: a newer + // stream (stream_epoch changed) or a newer write to the same buffer (write_generation + // changed) means this notification is stale and the bytes it names may have already + // been overwritten by the network thread, or are about to be. Skip it instead of + // risking a torn read; a fresher notification for the same slot is already queued or + // on its way. + std::lock_guard lock(this->drain_task->slot_mutex); + auto& sb = this->drain_task->slot_buffers[slot]; + + 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; + } + + // Ack gate: a slot with require_frame_done set allows only one un-acked delivery in + // flight. If one is already outstanding, park this (newer) notification instead of + // decoding it now -- overwriting any previously parked notification is latest-wins by + // design. Otherwise arm the gate (DECODE_DELIVERED) before decoding, so any later + // notification for this slot parks instead of decoding concurrently with this un-acked + // delivery. Arming gates on ack_enabled() alone, matching drain_events() and + // handle_stream_ring_event(); the listener is set before start() (see set_listener) so it + // is non-null here, and the callback invocation below is the crash-guard for that pointer. + if (this->ack_enabled(slot) && sb.ack_state != SlotAckState::IDLE) { + sb.parked = notif; + sb.has_parked = true; + return; + } + if (this->ack_enabled(slot)) { + 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 (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; + } + + // Hand off the timestamp to the main loop. Skip if the stream ended while we were + // decoding so the main loop doesn't fire a display after on_image_clear. The delta + // carries just this slot's bit; merge_artwork_display_update ORs it into whatever the + // main loop hasn't drained out of display_slot yet. + if (this->stream_active.load(std::memory_order_acquire)) { + ArtworkDisplayUpdate delta{}; + delta.timestamps[slot] = notif.timestamp; + // The epoch this decode was validated under: lets the main-loop deadline check drop + // 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); + this->event_state->display_slot.merge(merge_artwork_display_update, delta); + } +} + void ArtworkRole::Impl::drain_thread_func(ArtworkRole::Impl* self) { SS_LOGD(TAG, "Decode thread started"); @@ -403,71 +614,50 @@ void ArtworkRole::Impl::drain_thread_func(ArtworkRole::Impl* self) { break; } - // Blocking receive with 100ms timeout (allows periodic command checks) + // Replay any parked notification whose slot's gate has reopened (ack_state back to + // IDLE via frame_done() or an epoch-mismatch release in drain_events()). + // process_notification() revalidates the notification itself, so a since-stale + // generation/epoch is simply skipped -- correct, since a fresher notification is either + // already queued or has itself been freshly parked. Loop until no parked slot is ready + // so one wakeup can drain several slots without waiting on separate receive timeouts. + while (true) { + ArtworkNotification parked_notif{}; + bool found = false; + { + std::lock_guard lock(self->drain_task->slot_mutex); + for (auto& sb : self->drain_task->slot_buffers) { + if (sb.has_parked && sb.ack_state == SlotAckState::IDLE) { + parked_notif = sb.parked; + sb.has_parked = false; + found = true; + break; + } + } + } + if (!found) { + break; + } + self->process_notification(parked_notif); + } + + // Blocking receive with 100ms timeout (allows periodic command checks and, when nothing + // ever wakes the queue, an upper bound on how long a parked notification waits before the + // sweep above rechecks it). ArtworkNotification notif{}; if (!queue.receive(notif, DRAIN_RECEIVE_TIMEOUT_MS)) { continue; } - uint8_t slot = notif.slot; - uint8_t buf_idx = notif.buffer_idx; - if (slot >= ARTWORK_MAX_SLOTS) { + if (notif.slot == ARTWORK_RECHECK_SLOT) { + // Sentinel used only to unblock receive() so the parked-slot sweep above re-runs + // promptly; carries no work of its own. continue; } - - uint8_t* decode_data = nullptr; - size_t decode_length = 0; - { - // Validate the notification is still current before touching the buffer: a newer - // stream (stream_epoch changed) or a newer write to the same buffer (write_generation - // changed) means this notification is stale and the bytes it names may have already - // been overwritten by the network thread, or are about to be. Skip it instead of - // risking a torn read; a fresher notification for the same slot is already queued or - // on its way. - std::lock_guard lock(self->drain_task->slot_mutex); - auto& sb = self->drain_task->slot_buffers[slot]; - - if (notif.stream_epoch != self->stream_epoch.load(std::memory_order_relaxed)) { - continue; - } - if (notif.generation != sb.write_generation[buf_idx]) { - continue; - } - - auto& buf = sb.buffers[buf_idx]; - if (notif.data_length == 0 || buf.data() == nullptr) { - continue; - } - - // 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 (self->listener) { - self->listener->on_image_decode(slot, decode_data, decode_length, notif.format); - } - - { - std::lock_guard lock(self->drain_task->slot_mutex); - self->drain_task->slot_buffers[slot].drain_active = false; + if (notif.slot >= ARTWORK_MAX_SLOTS) { + continue; } - // Hand off the timestamp to the main loop. Skip if the stream ended while we were - // decoding so the main loop doesn't fire a display after on_image_clear. The delta - // carries just this slot's bit; merge_artwork_display_update ORs it into whatever the - // main loop hasn't drained out of display_slot yet. - if (self->stream_active.load(std::memory_order_acquire)) { - ArtworkDisplayUpdate delta{}; - delta.timestamps[slot] = notif.timestamp; - // The epoch this decode was validated under: lets the main-loop deadline check drop - // 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); - self->event_state->display_slot.merge(merge_artwork_display_update, delta); - } + self->process_notification(notif); } SS_LOGD(TAG, "Decode thread stopped"); @@ -486,4 +676,8 @@ void ArtworkRole::set_listener(ArtworkRoleListener* listener) { this->impl_->listener = listener; } +void ArtworkRole::frame_done(uint8_t slot) { + this->impl_->frame_done(slot); +} + } // namespace sendspin diff --git a/src/artwork_role_impl.h b/src/artwork_role_impl.h index 415b7d5..f1851c9 100644 --- a/src/artwork_role_impl.h +++ b/src/artwork_role_impl.h @@ -46,22 +46,14 @@ enum class ArtworkEventType : uint8_t { /// @brief Maximum number of artwork slots (2-bit slot field in protocol binary type byte) static constexpr size_t ARTWORK_MAX_SLOTS = 4; -/// @brief Double-buffered image storage for a single artwork slot -/// -/// All fields here are guarded by DrainTask::slot_mutex (shared across all slots; artwork is -/// not a hot path so contention is negligible). The network thread and decode thread both read -/// and write these fields, so they must never be touched outside that lock: -/// - write_idx: which buffer the network thread writes to next. -/// - drain_active / drain_buf_idx: which buffer the decode thread is currently decoding. -/// - write_generation[i]: bumped every time buffers[i] is overwritten by the network thread. -/// The decode thread compares this against the generation stamped on the notification it -/// dequeued to detect whether the buffer was overwritten again before it could be claimed. -struct SlotBuffer { - PlatformBuffer buffers[2]; - uint8_t write_idx{0}; - bool drain_active{false}; - uint8_t drain_buf_idx{0}; - uint32_t write_generation[2]{0, 0}; +/// @brief Sentinel notification slot used to wake the decode thread for a parked-frame recheck +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 + PRESENTED, // on_image_display or on_image_clear fired, awaiting frame_done() }; /// @brief Notification sent from the network thread to the decode thread when new image data @@ -85,6 +77,31 @@ struct ArtworkNotification { uint32_t stream_epoch; }; +/// @brief Double-buffered image storage for a single artwork slot +/// +/// All fields here are guarded by DrainTask::slot_mutex (shared across all slots; artwork is +/// not a hot path so contention is negligible). The network thread and decode thread both read +/// and write these fields, so they must never be touched outside that lock: +/// - write_idx: which buffer the network thread writes to next. +/// - drain_active / drain_buf_idx: which buffer the decode thread is currently decoding. +/// - write_generation[i]: bumped every time buffers[i] is overwritten by the network thread. +/// The decode thread compares this against the generation stamped on the notification it +/// dequeued to detect whether the buffer was overwritten again before it could be claimed. +/// - ack_state: only meaningful when the slot has require_frame_done set (see ack_enabled()); +/// tracks whether a delivery is currently un-acked for the slot (see SlotAckState). +/// - has_parked / parked: while ack_state is not IDLE, at most one newer notification is parked +/// here (latest-wins) instead of being decoded; it is replayed once the gate reopens. +struct SlotBuffer { + PlatformBuffer buffers[2]; + uint8_t write_idx{0}; + bool drain_active{false}; + uint8_t drain_buf_idx{0}; + uint32_t write_generation[2]{0, 0}; + SlotAckState ack_state{SlotAckState::IDLE}; + bool has_parked{false}; + ArtworkNotification parked{}; +}; + /// @brief Latest-wins display timestamps accumulated across artwork slots /// /// Merged cross-thread by the decode thread (one slot per merge) and taken whole by the @@ -147,12 +164,43 @@ struct ArtworkRole::Impl { void drain_events(); void cleanup(); + // ======================================== + // Consumer-facing method implementations + // ======================================== + + void frame_done(uint8_t slot) const; + // ======================================== // Helpers // ======================================== void stop() const; void enqueue_stream_event(ArtworkEventType event) const; + // 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 + // immediately with lateness 0, since no deadline exists); display_offset_ms shifts the + // deadline, positive firing early (mirroring PlayerRoleConfig::fixed_delay_us) and negative + // delaying. Pure and static for direct unit testing. + static int64_t display_overdue_us(int64_t client_ts, int32_t display_offset_ms, int64_t now); + // Maps a due display's overdue microseconds (from display_overdue_us) to the lateness_ms + // reported to on_image_display(). client_ts == 0 means no connection: report 0, the + // documented "no deadline exists" sentinel. When connected, floor the result at 1 ms so a + // sub-millisecond-late display never truncates to 0 and collides with that sentinel; the top + // is clamped at UINT32_MAX ms (~49 days), past which lateness is not meaningful. Pure and + // static for direct unit testing. + static uint32_t display_lateness_ms(int64_t client_ts, int64_t overdue_us); + // True if `slot` is within range and configured with require_frame_done. + bool ack_enabled(uint8_t slot) const; + // Sends a sentinel ARTWORK_RECHECK_SLOT notification to unblock the decode thread's queue + // receive so it re-runs the parked-slot sweep at the top of its loop. Best-effort: the send + // uses a 0 timeout and any failure is ignored, since the decode thread's 100ms receive + // timeout plus the loop-top sweep is the fallback that guarantees the parked notification is + // eventually rechecked even if this wakeup is dropped. + void wake_drain_thread() const; + // Validates and, if appropriate, decodes a single notification; called both from the normal + // queue-receive path and from the parked-slot sweep in drain_thread_func(). + void process_notification(const ArtworkNotification& notif); static void drain_thread_func(ArtworkRole::Impl* self); // ======================================== diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index c050522..ac5113c 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -28,6 +28,7 @@ add_executable(sendspin_tests test_spsc_ring_buffer.cpp test_inbox.cpp test_visualizer_role.cpp + test_artwork_role.cpp ) # Reach the library's private headers (protocol_messages.h, time_filter.h, ...). diff --git a/tests/test_artwork_role.cpp b/tests/test_artwork_role.cpp new file mode 100644 index 0000000..2d93665 --- /dev/null +++ b/tests/test_artwork_role.cpp @@ -0,0 +1,634 @@ +// Copyright 2026 Sendspin Contributors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "artwork_role_impl.h" +#include "constants.h" +#include "protocol_messages.h" +#include "sendspin/client.h" +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace sendspin; + +namespace { + +// Appends val as 8 big-endian bytes (the server timestamp prefix of every artwork binary +// message), mirroring put_be64 in test_visualizer_role.cpp. +void put_be64(std::vector& out, int64_t val) { + auto u = static_cast(val); + for (int i = 7; i >= 0; --i) { + out.push_back(static_cast((u >> (8 * i)) & 0xFF)); + } +} + +// Negative-wait window: long enough to be safely past the decode thread's 100ms parked-slot +// sweep fallback (see DRAIN_RECEIVE_TIMEOUT_MS in artwork_role.cpp), short enough to keep the +// suite fast. +constexpr auto NEGATIVE_WINDOW = std::chrono::milliseconds(300); +constexpr auto POSITIVE_TIMEOUT = std::chrono::milliseconds(1500); + +// Records every callback fired by an ArtworkRole::Impl under test, guarded by its own mutex so +// the test thread can safely poll state produced on the decode thread and the main thread. If +// frame_done_on_display is set, on_image_display() immediately (and reentrantly) calls +// frame_done() on the Impl this listener was bound to, exercising the reentrant-ack path. +class RecordingListener : public ArtworkRoleListener { +public: + struct DecodeEvent { + uint8_t slot; + std::vector payload; + }; + + void on_image_decode(uint8_t slot, const uint8_t* data, size_t length, + SendspinImageFormat /*format*/) override { + { + std::lock_guard lock(this->mutex); + this->decodes.push_back({slot, std::vector(data, data + length)}); + } + this->cv.notify_all(); + } + + void on_image_display(uint8_t slot, uint32_t /*lateness_ms*/) override { + { + std::lock_guard lock(this->mutex); + this->displays.push_back(slot); + } + this->cv.notify_all(); + // Deliberately outside the lock above: frame_done() takes the Impl's own slot_mutex, and + // this call must not be made while holding this listener's mutex (which nothing else + // needs, but keeping the pattern lock-then-release-then-reenter is the safe shape the + // production code itself uses -- see drain_events()/handle_stream_ring_event()). + if (this->frame_done_on_display && this->impl != nullptr) { + this->impl->frame_done(slot); + } + } + + void on_image_clear(uint8_t slot) override { + { + std::lock_guard lock(this->mutex); + this->clears.push_back(slot); + } + this->cv.notify_all(); + } + + // Waits (up to timeout) for pred() to become true, evaluated under this->mutex so it can + // safely read decodes/displays/clears. + template + bool wait_for(Pred pred, std::chrono::milliseconds timeout) { + std::unique_lock lock(this->mutex); + return this->cv.wait_for(lock, timeout, pred); + } + + // Asserts pred() stays false for the whole window; used for "must NOT fire" checks. Returns + // true if pred() never became true (the expected outcome). + template + bool never_within(Pred pred, std::chrono::milliseconds window) { + std::unique_lock lock(this->mutex); + return !this->cv.wait_for(lock, window, pred); + } + + size_t decode_count() { + std::lock_guard lock(this->mutex); + return this->decodes.size(); + } + + size_t display_count() { + std::lock_guard lock(this->mutex); + return this->displays.size(); + } + + size_t clear_count() { + std::lock_guard lock(this->mutex); + return this->clears.size(); + } + + // 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); + return this->decodes.at(index).payload.at(0); + } + + // True if any recorded decode for `slot` carries `marker` as its first payload byte. + bool has_decoded_marker(uint8_t slot, uint8_t marker) { + std::lock_guard lock(this->mutex); + for (const auto& d : this->decodes) { + if (d.slot == slot && !d.payload.empty() && d.payload[0] == marker) { + return true; + } + } + return false; + } + + size_t decode_count_for_slot(uint8_t slot) { + std::lock_guard lock(this->mutex); + size_t count = 0; + for (const auto& d : this->decodes) { + if (d.slot == slot) { + ++count; + } + } + return count; + } + + std::mutex mutex; + std::condition_variable cv; + std::vector decodes; + std::vector displays; + std::vector clears; + bool frame_done_on_display{false}; + ArtworkRole::Impl* impl{nullptr}; +}; + +// Builds a one-slot ArtworkRoleConfig; slot 0 opts into the ack gate iff `gated`. +ArtworkRoleConfig make_single_slot_config(bool gated) { + ArtworkRoleConfig config; + config.preferred_formats.push_back( + {SendspinImageSource::ALBUM, SendspinImageFormat::JPEG, 100, 100, gated}); + return config; +} + +// Builds a two-slot ArtworkRoleConfig: slot 0 gated, slot 1 not. +ArtworkRoleConfig make_two_slot_config() { + ArtworkRoleConfig config; + config.preferred_formats.push_back( + {SendspinImageSource::ALBUM, SendspinImageFormat::JPEG, 100, 100, true}); + config.preferred_formats.push_back( + {SendspinImageSource::ARTIST, SendspinImageFormat::JPEG, 100, 100, false}); + return config; +} + +// A real, never-started SendspinClient plus a bound ArtworkRole::Impl running a live decode +// thread. Both are heap-allocated with program lifetime (static deques, mirroring make_impl() in +// test_visualizer_role.cpp): Impl holds atomics so it is neither copyable nor movable, and it +// keeps a raw SendspinClient* that drain_events() dereferences (get_client_time()), so the client +// must outlive the Impl. A default-constructed, never-started SendspinClient never opens a +// connection, so get_client_time() always returns 0 -- drain_events() then treats every pending +// display as immediately due instead of honoring a server-clock deadline (see the comment at its +// call site in artwork_role.cpp), which is exactly what these tests want. +std::unique_ptr make_impl(ArtworkRoleConfig config) { + static std::deque clients; + static std::deque inboxes; + + clients.emplace_back(SendspinClientConfig{}); + auto impl = std::make_unique(std::move(config), &clients.back()); + inboxes.emplace_back(); + impl->attach_inbox(inboxes.back()); + return impl; +} + +// Sends one fake frame to `slot` whose image payload is [marker, 0xAA] (a distinct first byte +// per frame so tests can tell which frame decoded). +void send_frame(ArtworkRole::Impl& impl, uint8_t slot, uint8_t marker, int64_t timestamp = 1) { + std::vector data; + put_be64(data, timestamp); + data.push_back(marker); + data.push_back(0xAA); + 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. +template +bool poll_drain_until(ArtworkRole::Impl& impl, Pred pred, std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + do { + impl.drain_events(); + if (pred()) { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } while (std::chrono::steady_clock::now() < deadline); + return pred(); +} + +// 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 +// decode thread has parked this notification" (rather than "the decode thread has decoded +// something") poll the (public, per artwork_role_impl.h) SlotBuffer fields directly under the +// same mutex the production code uses. +template +bool wait_slot_state(ArtworkRole::Impl& impl, Pred pred, std::chrono::milliseconds timeout) { + const auto deadline = std::chrono::steady_clock::now() + timeout; + do { + { + std::lock_guard lock(impl.drain_task->slot_mutex); + if (pred()) { + return true; + } + } + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + } while (std::chrono::steady_clock::now() < deadline); + std::lock_guard lock(impl.drain_task->slot_mutex); + return pred(); +} + +} // namespace + +// ============================================================================ +// Ungated behavior: require_frame_done = false must reproduce today's behavior exactly +// ============================================================================ + +TEST(ArtworkFrameDoneGate, DefaultUngatedUnchanged) { + auto impl = make_impl(make_single_slot_config(false)); + 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)); + EXPECT_EQ(listener.decode_marker_at(0), 'A'); + + send_frame(*impl, 0, 'B'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(1), 'B'); +} + +// ============================================================================ +// Basic gate: at most one un-acked delivery per gated slot +// ============================================================================ + +TEST(ArtworkFrameDoneGate, GateHoldsSecondFrame) { + 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)); + EXPECT_EQ(listener.decode_marker_at(0), 'A'); + + send_frame(*impl, 0, 'B'); + EXPECT_TRUE( + listener.never_within([&] { return listener.decodes.size() >= 2; }, NEGATIVE_WINDOW)); + + impl->frame_done(0); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(1), 'B'); +} + +TEST(ArtworkFrameDoneGate, GateHoldsThroughDisplay) { + 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)); + + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.display_count() >= 1; }, POSITIVE_TIMEOUT)); + + // The gate must still be held after the display fires -- only frame_done() releases it. + send_frame(*impl, 0, 'B'); + EXPECT_TRUE( + listener.never_within([&] { return listener.decodes.size() >= 2; }, NEGATIVE_WINDOW)); + + impl->frame_done(0); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(1), 'B'); +} + +TEST(ArtworkFrameDoneGate, SupersedeKeepsNewestParked) { + 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)); + + send_frame(*impl, 0, 'B'); + // Wait for B to actually be parked before sending C, so C deterministically observes an + // already-parked notification to supersede (see the wait_slot_state comment on its first use + // in ClearIsADeliveryAndDropsParked for why this matters instead of a fixed sleep). + ASSERT_TRUE(wait_slot_state( + *impl, [&] { return impl->drain_task->slot_buffers[0].has_parked; }, POSITIVE_TIMEOUT)); + send_frame(*impl, 0, 'C'); + + impl->frame_done(0); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + + // Only one more decode fires, and it is the newest (C); B was superseded while parked. + EXPECT_TRUE( + listener.never_within([&] { return listener.decodes.size() >= 3; }, NEGATIVE_WINDOW)); + EXPECT_EQ(listener.decode_marker_at(1), 'C'); + EXPECT_FALSE(listener.has_decoded_marker(0, 'B')); +} + +// ============================================================================ +// Clear as a delivery: stream/end and stream/clear each owe exactly one ack +// ============================================================================ + +TEST(ArtworkFrameDoneGate, ClearIsADeliveryAndDropsParked) { + 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)); + + send_frame(*impl, 0, 'B'); // parks: A's delivery is still un-acked + // Wait for the decode thread to actually park B (has_parked observed under slot_mutex) + // before delivering the clear -- otherwise the clear could race ahead of the still-in-flight + // notification and land before B is parked, in which case B would park *behind* the clear's + // own owed ack instead of being dropped by it, which is a different (also-tested, see + // ClearGateHoldsNextStreamFirstFrame) scenario. + ASSERT_TRUE(wait_slot_state( + *impl, [&] { return impl->drain_task->slot_buffers[0].has_parked; }, POSITIVE_TIMEOUT)); + + impl->handle_stream_ring_event(ArtworkEventType::STREAM_CLEAR); + ASSERT_TRUE(listener.wait_for([&] { return listener.clears.size() >= 1; }, POSITIVE_TIMEOUT)); + + // The clear itself owes an ack; acking it must NOT resurrect the dropped, parked B. + impl->frame_done(0); + EXPECT_TRUE( + listener.never_within([&] { return listener.decodes.size() >= 2; }, NEGATIVE_WINDOW)); + + // A fresh stream's frame decodes normally: the gate is IDLE again. + impl->handle_stream_start(ServerArtworkStreamObject{}); + send_frame(*impl, 0, 'C'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(1), 'C'); +} + +TEST(ArtworkFrameDoneGate, ClearGateHoldsNextStreamFirstFrame) { + 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)); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.display_count() >= 1; }, POSITIVE_TIMEOUT)); + + // stream/end fires the clear callback but the clear's own ack is still outstanding. + impl->handle_stream_ring_event(ArtworkEventType::STREAM_END); + ASSERT_TRUE(listener.wait_for([&] { return listener.clears.size() >= 1; }, POSITIVE_TIMEOUT)); + + impl->handle_stream_start(ServerArtworkStreamObject{}); + send_frame(*impl, 0, 'B'); + EXPECT_TRUE( + listener.never_within([&] { return listener.decodes.size() >= 2; }, NEGATIVE_WINDOW)); + + impl->frame_done(0); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(1), 'B'); +} + +// ============================================================================ +// frame_done() edge cases +// ============================================================================ + +TEST(ArtworkFrameDoneGate, FrameDoneNoOpWhenIdle) { + auto impl = make_impl(make_single_slot_config(true)); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + // Nothing outstanding: both calls must be safe no-ops (including the out-of-range slot). + impl->frame_done(0); + impl->frame_done(99); + + 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'); +} + +// ============================================================================ +// Stream restart interaction with the gate +// ============================================================================ + +TEST(ArtworkFrameDoneGate, RestartReleasesUndisplayedDecode) { + 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)); + // Deliberately never call drain_events() here: A's display must never fire. + + impl->handle_stream_start(ServerArtworkStreamObject{}); // restart + + // Give the decode thread's async display hand-off a chance to land, then confirm the restart + // (epoch bump + display_slot reset) keeps it from ever reaching the listener. + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + impl->drain_events(); + impl->drain_events(); + EXPECT_EQ(listener.display_count(), 0U); + + // The DECODE_DELIVERED gate was auto-released by the restart: B decodes without any ack. + send_frame(*impl, 0, 'B'); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(1), 'B'); +} + +TEST(ArtworkFrameDoneGate, RestartKeepsPresentedGate) { + 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)); + ASSERT_TRUE( + poll_drain_until(*impl, [&] { return listener.display_count() >= 1; }, POSITIVE_TIMEOUT)); + + impl->handle_stream_start(ServerArtworkStreamObject{}); // restart; PRESENTED stays armed + + send_frame(*impl, 0, 'B'); + EXPECT_TRUE( + listener.never_within([&] { return listener.decodes.size() >= 2; }, NEGATIVE_WINDOW)); + + impl->frame_done(0); + ASSERT_TRUE(listener.wait_for([&] { return listener.decodes.size() >= 2; }, POSITIVE_TIMEOUT)); + EXPECT_EQ(listener.decode_marker_at(1), 'B'); +} + +// ============================================================================ +// Reentrant frame_done() from inside on_image_display() +// ============================================================================ + +TEST(ArtworkFrameDoneGate, FrameDoneReentrantFromDisplay) { + auto impl = make_impl(make_single_slot_config(true)); + RecordingListener listener; + listener.frame_done_on_display = true; + listener.impl = impl.get(); + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + send_frame(*impl, 0, 'A'); + // B may arrive while A is still un-acked (it will park, then replay once the reentrant ack + // from A's on_image_display() fires) or after; either way both must eventually decode and + // display without any external frame_done() call and without deadlock. + send_frame(*impl, 0, 'B'); + + ASSERT_TRUE(poll_drain_until( + *impl, [&] { return listener.decode_count() >= 2 && listener.display_count() >= 2; }, + POSITIVE_TIMEOUT)); + + EXPECT_TRUE(listener.has_decoded_marker(0, 'A')); + EXPECT_TRUE(listener.has_decoded_marker(0, 'B')); + EXPECT_EQ(listener.display_count(), 2U); +} + +// ============================================================================ +// One gated slot must not affect an ungated slot +// ============================================================================ + +TEST(ArtworkFrameDoneGate, UngatedSlotUnaffectedBesideGatedSlot) { + auto impl = make_impl(make_two_slot_config()); + RecordingListener listener; + impl->listener = &listener; + ASSERT_TRUE(impl->start()); + impl->handle_stream_start(ServerArtworkStreamObject{}); + + // wait_for()'s predicate runs under RecordingListener::mutex (via condition_variable's + // predicate overload), so it must touch listener.decodes directly rather than going through + // a helper like decode_count_for_slot() that re-locks the same non-recursive mutex. + auto count_for_slot = [&](uint8_t slot) { + size_t n = 0; + for (const auto& d : listener.decodes) { + if (d.slot == slot) { + ++n; + } + } + return n; + }; + + // Gate slot 0 with an un-acked delivery. + send_frame(*impl, 0, 'A'); + ASSERT_TRUE(listener.wait_for([&] { return count_for_slot(0) >= 1; }, POSITIVE_TIMEOUT)); + + // Slot 1 keeps decoding every frame freely, ungated by slot 0's outstanding delivery. Each + // send waits for its own decode before the next is sent: slot 1 is double-buffered like any + // other slot (see SlotBuffer::write_generation), so three back-to-back writes with nothing + // draining them could legitimately overwrite an unclaimed buffer and drop a frame -- a + // real (and separately-covered) property of the double-buffering scheme, not of the ack + // gate this test is about, so it must not be exercised here. + send_frame(*impl, 1, 'X'); + ASSERT_TRUE(listener.wait_for([&] { return count_for_slot(1) >= 1; }, POSITIVE_TIMEOUT)); + send_frame(*impl, 1, 'Y'); + ASSERT_TRUE(listener.wait_for([&] { return count_for_slot(1) >= 2; }, POSITIVE_TIMEOUT)); + send_frame(*impl, 1, 'Z'); + ASSERT_TRUE(listener.wait_for([&] { return count_for_slot(1) >= 3; }, POSITIVE_TIMEOUT)); + + EXPECT_EQ(listener.decode_count_for_slot(0), 1U); +} + +// ============================================================================ +// 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. +// Pure function, so tested directly: the integration tests above all run without a connection +// (client_ts == 0), which bypasses the offset and lateness paths. +// ============================================================================ + +TEST(ArtworkDisplayDeadline, NoConnectionSentinelFiresImmediately) { + // client_ts == 0 means no connection: due immediately with lateness 0, regardless of offset + // in either direction (no deadline exists to be late against). + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(0, 0, 5'000'000), 0); + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(0, 1000, 5'000'000), 0); + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(0, -1000, 5'000'000), 0); +} + +TEST(ArtworkDisplayDeadline, ZeroOffsetMatchesServerDeadline) { + const int64_t now = 10'000'000; // 10 s in us + EXPECT_LT(ArtworkRole::Impl::display_overdue_us(now + 1, 0, now), 0); + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now, 0, now), 0); + // 1 us past the deadline: due, with 1 us of lateness. + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now - 1, 0, now), 1); +} + +TEST(ArtworkDisplayDeadline, PositiveOffsetFiresEarly) { + const int64_t now = 10'000'000; + // Deadline 900 ms in the future, offset 1000 ms: already due, 100 ms past the shifted + // deadline. + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now + 900 * US_PER_MS, 1000, now), + 100 * US_PER_MS); + // Deadline 1100 ms in the future, offset 1000 ms: still 100 ms out. + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now + 1100 * US_PER_MS, 1000, now), + -100 * US_PER_MS); + // Exact boundary: deadline minus offset equals now. + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now + 1000 * US_PER_MS, 1000, now), 0); +} + +TEST(ArtworkDisplayDeadline, NegativeOffsetDelays) { + const int64_t now = 10'000'000; + // Deadline 500 ms in the past, but a -1000 ms offset holds it another 500 ms. + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now - 500 * US_PER_MS, -1000, now), + -500 * US_PER_MS); + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now - 1000 * US_PER_MS, -1000, now), 0); +} + +TEST(ArtworkDisplayDeadline, LatenessReportsPastDeadlineSlip) { + const int64_t now = 10'000'000; + // A frame that arrived 600 ms after its shifted deadline reports exactly that slip, letting + // a consumer shorten its cross-fade (e.g. 2000 ms - 600 ms) so the fade still ends on time. + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now + 400 * US_PER_MS, 1000, now), + 600 * US_PER_MS); + // A deadline far in the past reports a correspondingly huge lateness, the cue for a consumer + // to snap instead of fading. + EXPECT_EQ(ArtworkRole::Impl::display_overdue_us(now - 120'000 * US_PER_MS, 0, now), + 120'000 * US_PER_MS); +} + +TEST(ArtworkDisplayDeadline, LargeOffsetDoesNotOverflow) { + // INT32_MIN/MAX offsets must be widened to 64-bit before the ms-to-us multiply. + const int64_t now = 10'000'000; + EXPECT_GE(ArtworkRole::Impl::display_overdue_us(now + US_PER_MS, INT32_MAX, now), 0); + EXPECT_LT(ArtworkRole::Impl::display_overdue_us(now - US_PER_MS, INT32_MIN, now), 0); +} + +// ============================================================================ +// display_lateness_ms: maps a due display's overdue microseconds to the lateness_ms passed to +// on_image_display(). Pure function; the integration tests above all run without a connection so +// only its client_ts == 0 branch is otherwise exercised. +// ============================================================================ + +TEST(ArtworkDisplayLateness, ZeroIsReservedForNoConnection) { + // The one non-obvious invariant on_image_display() consumers rely on: lateness_ms == 0 means + // "no connection" and nothing else. With a connection, a display firing under a millisecond + // late must not truncate to 0 and collide with that sentinel -- it is floored to 1 ms -- while + // a normal multi-millisecond slip passes through unchanged. + EXPECT_EQ(ArtworkRole::Impl::display_lateness_ms(0, 0), 0u); // no connection + EXPECT_EQ(ArtworkRole::Impl::display_lateness_ms(1, US_PER_MS - 1), 1u); // connected, <1ms + EXPECT_EQ(ArtworkRole::Impl::display_lateness_ms(1, 600 * US_PER_MS), 600u); // connected slip +} + +TEST(ArtworkDisplayLateness, HugeLatenessSaturatesAtUint32Max) { + // A pathological far-past deadline can exceed UINT32_MAX ms (~49 days); the ms value must + // saturate there rather than wrap when narrowed to uint32_t. + EXPECT_EQ(ArtworkRole::Impl::display_lateness_ms(1, INT64_MAX), UINT32_MAX); +}