Retire the outage gap in the Pulse, PortAudio and PipeWire sinks

src/pipewire_sink.cpp

@@ -359,6 +359,11 @@ bool PipeWireSink::configure(uint32_t sample_rate, uint8_t channels, uint8_t bit
 
     // Before anything can fail: recovery reopens at this format.
     this->last_format_ = {sample_rate, channels, bits_per_sample};
+    // A new stream starts the producer from zero buffered frames, so an outage gap left over from
+    // the last one is owed to nobody, on either side of the handoff. Here rather than beside
+    // reset(), which a failed open skips.
+    this->recovery_.forget_discarded_frames();
+    this->gap_handoff_.forget();
 
     this->close_stream_();
     if (!this->open_stream_(sample_rate, channels, bits_per_sample, PIPEWIRE_STREAM_TIMEOUT_MS)) {
@@ -391,7 +396,11 @@ size_t PipeWireSink::write(const uint8_t* data, size_t length, uint32_t timeout_
                                  ? this->bytes_per_frame_
                                  : static_cast<size_t>(this->last_format_.channels) *
                                        (static_cast<size_t>(this->last_format_.bit_depth) / 8U);
-        return (frame == 0) ? length : length - (length % frame);
+        const size_t consumed = (frame == 0) ? length : length - (length % frame);
+        if (frame != 0) {
+            this->recovery_.discard_frames(static_cast<uint32_t>(consumed / frame));
+        }
+        return consumed;
     }
 
     const size_t bytes_per_frame = this->bytes_per_frame_;
@@ -407,6 +416,10 @@ size_t PipeWireSink::write(const uint8_t* data, size_t length, uint32_t timeout_
     const uint64_t target = this->target_multiplier_.load(std::memory_order_relaxed);
     const uint8_t* src = this->stage_(data, frames_total, start, target);
 
+    // The stream is alive, so an outage's gap can go to the process callback: it has the
+    // timestamp to retire the gap against, and this thread does not.
+    this->gap_handoff_.add(this->recovery_.take_discarded_frames());
+
     size_t done = 0;
     while (done < usable) {
         if (this->stopping_.load()) {
@@ -452,6 +465,13 @@ void PipeWireSink::clear() {
     ++this->stream_generation_;
     this->space_available_.notify_all();
 
+    // The player zeroes its buffered-frame count with a flush, so the gap is owed to nobody. Not
+    // reset(): the recovery budget is configure()'s to refill. A callback that has already taken
+    // the gap may still report it after this -- the same window request_clear() leaves for frames
+    // already read, and not worth a lock on the realtime path.
+    this->recovery_.forget_discarded_frames();
+    this->gap_handoff_.forget();
+
     // Do not snap current_multiplier_: process() keeps running through a flush.
 
     if (this->stream_alive_()) {
@@ -522,6 +542,7 @@ void PipeWireSink::poll(int64_t now_ms) {
     }
 
     const StreamFormat format = this->last_format_;
+    this->discard_ring_tail_();
     this->close_stream_();
     // Rebuild the loop too: a restarted daemon took the old connection with it.
     this->stop_loop_();
@@ -607,6 +628,7 @@ void PipeWireSink::stream_process_cb(void* userdata) {
     pw_time time{};
     int64_t finish_us = 0;
     bool have_timing = false;
+    uint32_t gap_frames = 0;
     if (real_bytes > 0 && self->on_frames_played &&
         pw_stream_get_time_n(self->stream_, &time, sizeof(time)) == 0 && time.rate.denom != 0) {
         const double rate_s =
@@ -618,6 +640,8 @@ void PipeWireSink::stream_process_cb(void* userdata) {
         }
         finish_us = entered_us + static_cast<int64_t>(ahead_s * 1e6);
         have_timing = true;
+        // Only here, so a cycle the graph gave no timing for leaves the gap for the next one.
+        gap_frames = self->gap_handoff_.take();
     }
 
     data.chunk->offset = 0;
@@ -632,8 +656,10 @@ void PipeWireSink::stream_process_cb(void* userdata) {
     // Notified without the mutex; a missed wakeup costs at most one quantum.
     self->space_available_.notify_one();
 
+    // The gap rides on this report rather than its own: the player sums the frames of reports it
+    // has not read yet but keeps only the last timestamp.
     if (have_timing) {
-        self->on_frames_played(static_cast<uint32_t>(real_bytes / stride), finish_us);
+        self->on_frames_played(frames_with_gap(gap_frames, real_bytes / stride), finish_us);
     }
 }
 
@@ -823,6 +849,7 @@ bool PipeWireSink::reopen_in_place_() {
     }
 
     const StreamFormat format = this->last_format_;
+    this->discard_ring_tail_();
     this->close_stream_();
     // Keep the loop: this cheap attempt recovers a node lost while the daemon stayed up.
     if (!this->open_stream_(format.sample_rate, format.channels, format.bit_depth,
@@ -842,6 +869,14 @@ bool PipeWireSink::reopen_in_place_() {
     return true;
 }
 
+void PipeWireSink::discard_ring_tail_() {
+    if (this->bytes_per_frame_ == 0) {
+        return;  // no stream, so close_stream_() has already emptied the ring
+    }
+    this->recovery_.discard_frames(
+        static_cast<uint32_t>(this->ring_.available() / this->bytes_per_frame_));
+}
+
 bool PipeWireSink::stream_alive_() const {
     return this->stream_ != nullptr && this->bytes_per_frame_ != 0 && !this->stream_failed_.load();
 }

src/pipewire_sink.h

@@ -125,6 +125,10 @@ class PipeWireSink final : public AudioSink {
     /// The one in-place reconnect a dead stream gets, from write(). Caller holds mutex_.
     /// @return true if a stream is running again.
     bool reopen_in_place_();
+    /// Adds the frames still in the ring to the outage gap: the player has counted them, and the
+    /// lost stream recovery is about to close will never report them. Only for those closes --
+    /// stop(), configure() and clear() end the stream the gap belonged to. Caller holds mutex_.
+    void discard_ring_tail_();
     /// True while the stream is connected and the graph is still driving it. Caller holds mutex_.
     bool stream_alive_() const;
     /// Ring size in bytes. Caller holds mutex_ and the format fields are set.
@@ -181,6 +185,9 @@ class PipeWireSink final : public AudioSink {
     StreamFormat last_format_{};
     /// Guarded by mutex_, except SinkRecovery::pending().
     SinkRecovery recovery_;
+    /// The outage gap once write() has handed it on, for the process callback to retire lock-free
+    /// with its next timed report. recovery_ cannot be read there; see OutageGapHandoff.
+    OutageGapHandoff gap_handoff_;
 
     std::atomic<uint8_t> volume_{DEFAULT_SINK_VOLUME};
     std::atomic<bool> muted_{false};

src/portaudio_sink.cpp

@@ -412,6 +412,11 @@ bool PortAudioSink::configure(uint32_t sample_rate, uint8_t channels, uint8_t bi
 
     // Before anything can fail: recovery reopens at this format.
     this->last_format_ = {sample_rate, channels, bits_per_sample};
+    // A new stream starts the producer from zero buffered frames, so an outage gap left over from
+    // the last one is owed to nobody, on either side of the handoff. Before the device is
+    // resolved, so the restart_stream_() reuse path and a failed open both drop it too.
+    this->recovery_.forget_discarded_frames();
+    this->gap_handoff_.forget();
 
     // Resolved per stream, so a bare -o portaudio follows the host's default.
     PaDeviceIndex device = paNoDevice;
@@ -465,7 +470,11 @@ size_t PortAudioSink::write(const uint8_t* data, size_t length, uint32_t timeout
                                  ? this->bytes_per_frame_
                                  : static_cast<size_t>(this->last_format_.channels) *
                                        (static_cast<size_t>(this->last_format_.bit_depth) / 8U);
-        return (frame == 0) ? length : length - (length % frame);
+        const size_t consumed = (frame == 0) ? length : length - (length % frame);
+        if (frame != 0) {
+            this->recovery_.discard_frames(static_cast<uint32_t>(consumed / frame));
+        }
+        return consumed;
     }
 
     const size_t bytes_per_frame = this->bytes_per_frame_;
@@ -476,6 +485,10 @@ size_t PortAudioSink::write(const uint8_t* data, size_t length, uint32_t timeout
         return 0;
     }
 
+    // The stream is alive, so an outage's gap can go to the callback: it has the timestamp to
+    // retire the gap against, and this thread does not.
+    this->gap_handoff_.add(this->recovery_.take_discarded_frames());
+
     size_t done = 0;
     while (done < usable) {
         if (this->stopping_.load()) {
@@ -518,6 +531,13 @@ void PortAudioSink::clear() {
     // A flush ends a parked write()'s stream.
     ++this->stream_generation_;
 
+    // The player zeroes its buffered-frame count with a flush, so the gap is owed to nobody. Not
+    // reset(): the recovery budget is configure()'s to refill. A callback that has already taken
+    // the gap may still report it after this -- the same window request_clear() leaves for frames
+    // already read, and not worth a lock on the callback.
+    this->recovery_.forget_discarded_frames();
+    this->gap_handoff_.forget();
+
     // Do not snap current_multiplier_: the callback keeps running through a flush.
 
     if (this->stream_alive_()) {
@@ -561,7 +581,11 @@ void PortAudioSink::poll(int64_t now_ms) {
     this->recovery_.rescan_done(true);
 
     const StreamFormat format = this->last_format_;
-    // Close first: Pa_Terminate() with a stream open is undefined and invalidates every index.
+    // The stream goes first whatever happens next: Pa_Terminate() with one open is undefined,
+    // and every PaDeviceIndex -- device_index_ among them, which this clears -- dies with it.
+    // Nothing playing is torn down by this, because only reopen_in_place_() can ask for a rescan
+    // and it only runs on a stream that has already died.
+    this->discard_ring_tail_();
     this->close_stream_();
 
     if (!this->pa_.reinitialize()) {
@@ -759,6 +783,7 @@ bool PortAudioSink::reopen_in_place_() {
     }
 
     const StreamFormat format = this->last_format_;
+    this->discard_ring_tail_();
     this->close_stream_();
     if (!this->open_stream_(device, format.sample_rate, format.channels, format.bit_depth)) {
         this->recovery_.reopen_done(false);  // open_stream_() has already said why, once
@@ -777,6 +802,14 @@ bool PortAudioSink::reopen_in_place_() {
     return true;
 }
 
+void PortAudioSink::discard_ring_tail_() {
+    if (this->bytes_per_frame_ == 0) {
+        return;  // no stream, so close_stream_() has already emptied the ring
+    }
+    this->recovery_.discard_frames(
+        static_cast<uint32_t>(this->ring_.available() / this->bytes_per_frame_));
+}
+
 bool PortAudioSink::stream_alive_() const {
     if (this->stream_ == nullptr) {
         return false;
@@ -840,7 +873,10 @@ int PortAudioSink::pa_callback(const void* /*input*/, void* output, unsigned lon
                                     (static_cast<double>(frames_played) / self->stream_rate_);
         const int64_t finish_us =
             entered_us + static_cast<int64_t>(std::llround(dac_offset_s * 1e6));
-        self->on_frames_played(frames_played, finish_us);
+        // The gap rides on this report rather than its own: the player sums the frames of reports
+        // it has not read yet but keeps only the last timestamp.
+        self->on_frames_played(frames_with_gap(self->gap_handoff_.take(), frames_played),
+                               finish_us);
     }
 
     return paContinue;

src/portaudio_sink.h

@@ -103,6 +103,10 @@ class PortAudioSink final : public AudioSink {
     /// The one in-place reopen a dead stream gets, from write(). Caller holds mutex_.
     /// @return true if a stream is running again.
     bool reopen_in_place_();
+    /// Adds the frames still in the ring to the outage gap: the player has counted them, and the
+    /// lost stream recovery is about to close will never report them. Only for those closes --
+    /// stop(), configure() and clear() end the stream the gap belonged to. Caller holds mutex_.
+    void discard_ring_tail_();
     /// True while the open stream is still being driven by PortAudio. Caller holds mutex_.
     bool stream_alive_() const;
     /// Ring size in bytes. Caller holds mutex_ and the format fields are set.
@@ -146,6 +150,9 @@ class PortAudioSink final : public AudioSink {
     StreamFormat last_format_{};
     /// Guarded by mutex_, except SinkRecovery::pending().
     SinkRecovery recovery_;
+    /// The outage gap once write() has handed it on, for the callback to retire lock-free with
+    /// its next report. recovery_ cannot be read there; see OutageGapHandoff.
+    OutageGapHandoff gap_handoff_;
 
     std::atomic<uint8_t> volume_{DEFAULT_SINK_VOLUME};
     std::atomic<bool> muted_{false};

src/pulse_sink.cpp

@@ -383,6 +383,9 @@ bool PulseAudioSink::configure(uint32_t sample_rate, uint8_t channels, uint8_t b
 
     // Before anything can fail: recovery reopens at this format.
     this->last_format_ = {sample_rate, channels, bits_per_sample};
+    // A new stream starts the producer from zero buffered frames, so an outage gap left over from
+    // the last one is owed to nobody. Here rather than beside reset(), which a failed open skips.
+    this->recovery_.forget_discarded_frames();
 
     this->close_stream_();
     if (!this->conn_.ready()) {
@@ -422,7 +425,11 @@ size_t PulseAudioSink::write(const uint8_t* data, size_t length, uint32_t timeou
                                  ? this->bytes_per_frame_
                                  : static_cast<size_t>(this->last_format_.channels) *
                                        (static_cast<size_t>(this->last_format_.bit_depth) / 8U);
-        return (frame == 0) ? length : length - (length % frame);
+        const size_t consumed = (frame == 0) ? length : length - (length % frame);
+        if (frame != 0) {
+            this->recovery_.discard_frames(static_cast<uint32_t>(consumed / frame));
+        }
+        return consumed;
     }
 
     const size_t bytes_per_frame = this->bytes_per_frame_;
@@ -495,6 +502,7 @@ size_t PulseAudioSink::write(const uint8_t* data, size_t length, uint32_t timeou
 
     int64_t finish_us = 0;
     bool have_timing = false;
+    uint32_t gap_frames = 0;
     if (frames_done > 0 && this->stream_ != nullptr) {
         // pa_stream_get_latency() is the server's own measure of time until queued audio plays.
         const MainloopLock ml(this->conn_.mainloop());
@@ -503,13 +511,29 @@ size_t PulseAudioSink::write(const uint8_t* data, size_t length, uint32_t timeou
         if (pa_stream_get_latency(this->stream_, &latency_us, &negative) == 0 && negative == 0) {
             finish_us = now_us() + static_cast<int64_t>(latency_us);
             have_timing = true;
+            // Taken with the timestamp, under the same lock, so no report can land between the
+            // outage being retired and the first real playback being reported.
+            gap_frames = this->recovery_.take_discarded_frames();
         }
     }
 
+    // Frames that reached a stream which then died will never be reported, so they join the gap.
+    // A clear() during the wait is a flush rather than a loss -- the player has dropped those
+    // frames already -- and is told apart by the generation. A poll() that closed and reopened the
+    // stream during the wait moves the generation too and so is taken for a flush: that rare case
+    // under-counts.
+    if (frames_done > 0 && !have_timing &&
+        (this->stream_ == nullptr ||
+         (this->stream_failed_.load() && this->stream_generation_ == generation))) {
+        this->recovery_.discard_frames(static_cast<uint32_t>(frames_done));
+    }
+
     lock.unlock();
-    // Outside the lock, so a callback that touches the sink cannot deadlock.
+    // Outside the lock, so a callback that touches the sink cannot deadlock. One report for the
+    // gap and this write together: the player keeps only the last timestamp, so a separate gap
+    // report could lose it.
     if (have_timing && this->on_frames_played) {
-        this->on_frames_played(static_cast<uint32_t>(frames_done), finish_us);
+        this->on_frames_played(frames_with_gap(gap_frames, frames_done), finish_us);
     }
     return frames_done * bytes_per_frame;
 }
@@ -524,6 +548,10 @@ void PulseAudioSink::clear() {
     ++this->stream_generation_;
     this->space_available_.notify_all();
 
+    // Also before the early return: the player zeroes its buffered-frame count with a flush, so
+    // the gap is owed to nobody. Only the gap -- the recovery budget is configure()'s to refill.
+    this->recovery_.forget_discarded_frames();
+
     if (this->stream_ == nullptr) {
         return;
     }

src/sink_recovery.cpp

@@ -107,6 +107,30 @@ void SinkRecovery::escalate_() {
     }
 }
 
+void OutageGapHandoff::add(uint32_t frames) {
+    if (frames == 0) {
+        return;
+    }
+    uint32_t current = this->frames_.load();
+    // A compare-exchange rather than load-then-store: the callback can take() between the two, and
+    // a plain store would hand back the gap it just retired.
+    while (true) {
+        const uint32_t room = std::numeric_limits<uint32_t>::max() - current;
+        const uint32_t next = current + ((frames < room) ? frames : room);
+        if (this->frames_.compare_exchange_weak(current, next)) {
+            return;
+        }
+    }
+}
+
+uint32_t OutageGapHandoff::take() {
+    return this->frames_.exchange(0);
+}
+
+void OutageGapHandoff::forget() {
+    this->frames_.store(0);
+}
+
 int64_t SinkRecovery::delay_for_(int attempts_made) {
     int64_t delay = SINK_RESCAN_DELAY_MS;
     // Doubled in a loop rather than shifted, so raising SINK_RESCAN_ATTEMPTS cannot overflow.

src/sink_recovery.h

@@ -18,6 +18,7 @@
 
 #include <atomic>
 #include <cstdint>
+#include <limits>
 
 namespace sendspin_cli {
 
@@ -88,4 +89,35 @@ class SinkRecovery {
     int64_t rescan_at_ms_{NOT_STAMPED};
 };
 
+/// @brief Carries an outage gap from a sink's locked write() to a callback that takes no lock.
+///
+/// PortAudioSink and PipeWireSink report playback from a realtime callback, which SinkRecovery's
+/// locking rule puts out of its reach. So write() moves the gap in here, under the sink's lock,
+/// once the stream is alive again, and the callback takes it with the first report that has a
+/// timestamp to retire it against. Saturates rather than wraps, like discard_frames().
+class OutageGapHandoff {
+public:
+    /// Producer side, under the sink's lock. Safe against a concurrent take().
+    void add(uint32_t frames);
+
+    /// Consumer side, from the audio callback: the whole gap, leaving none behind.
+    uint32_t take();
+
+    /// For a stream that ended before its gap was retired; see forget_discarded_frames().
+    void forget();
+
+private:
+    std::atomic<uint32_t> frames_{0};
+};
+
+/// @brief The count for one on_frames_played() report that retires `gap_frames` alongside
+/// `played_frames`.
+///
+/// Saturated rather than wrapped: the report is 32-bit and a gap can already sit at the ceiling.
+constexpr uint32_t frames_with_gap(uint32_t gap_frames, uint64_t played_frames) {
+    const uint64_t total = static_cast<uint64_t>(gap_frames) + played_frames;
+    const uint64_t ceiling = std::numeric_limits<uint32_t>::max();
+    return static_cast<uint32_t>((total < ceiling) ? total : ceiling);
+}
+
 }  // namespace sendspin_cli