Fix ALSA playback clock after device recovery

src/alsa_sink.cpp

@@ -27,6 +27,7 @@
 #include <cstdio>
 #include <cstdlib>
 #include <cstring>
+#include <limits>
 #include <mutex>
 #include <string>
 #include <thread>
@@ -467,6 +468,9 @@ bool AlsaAudioSink::configure(uint32_t sample_rate, uint8_t channels, uint8_t bi
 
     // Before anything can fail: poll() reopens at this format.
     this->last_format_ = {sample_rate, channels, bits_per_sample};
+    // A new stream starts from zero buffered frames; here, not beside reset(), so a failed open
+    // drops the gap too.
+    this->recovery_.forget_discarded_frames();
 
     if (this->pcm_ != nullptr && this->rate_ == sample_rate && this->channels_ == channels &&
         this->bits_ == bits_per_sample) {
@@ -505,6 +509,7 @@ size_t AlsaAudioSink::write(const uint8_t* data, size_t length, uint32_t timeout
     size_t bytes_per_frame = 0;
     int64_t finish_us = 0;
     bool have_timing = false;
+    uint32_t gap_frames = 0;
 
     {
         const std::lock_guard<std::mutex> lock(this->device_mutex_);
@@ -522,7 +527,11 @@ size_t AlsaAudioSink::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;
         }
 
         bytes_per_frame = this->bytes_per_frame_;
@@ -607,13 +616,23 @@ size_t AlsaAudioSink::write(const uint8_t* data, size_t length, uint32_t timeout
             if (snd_pcm_delay(this->pcm_, &delay) == 0 && delay >= 0) {
                 finish_us = now_us() + ((static_cast<int64_t>(delay) * 1000000) / this->rate_);
                 have_timing = true;
+                // Taken with the timestamp under the same lock, so nothing reports between them.
+                gap_frames = this->recovery_.take_discarded_frames();
             }
         }
+
+        // recover_() closed the device mid-write: these bytes are consumed but never timestamped.
+        if (frames_done > 0 && this->pcm_ == nullptr) {
+            this->recovery_.discard_frames(static_cast<uint32_t>(frames_done));
+        }
     }
 
-    // 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 covers
+    // the gap and this write: the player keeps only the last timestamp it has not read.
     if (have_timing && this->on_frames_played) {
-        this->on_frames_played(static_cast<uint32_t>(frames_done), finish_us);
+        const uint64_t played = static_cast<uint64_t>(gap_frames) + frames_done;
+        const uint64_t ceiling = std::numeric_limits<uint32_t>::max();
+        this->on_frames_played(static_cast<uint32_t>(std::min(played, ceiling)), finish_us);
     }
 
     return frames_done * bytes_per_frame;
@@ -625,6 +644,10 @@ void AlsaAudioSink::clear() {
     // Safe only under device_mutex_; PortAudioSink::clear() must not do the same.
     this->current_multiplier_ = this->target_multiplier_.load(std::memory_order_relaxed);
 
+    // Before the early return: the gap's stream is over. Only the gap goes; reset() owns the
+    // budget.
+    this->recovery_.forget_discarded_frames();
+
     if (this->pcm_ == nullptr) {
         return;
     }
@@ -698,6 +721,7 @@ void AlsaAudioSink::poll(int64_t now_ms) {
         this->close_device_();
         return;
     }
+    // The gap stays pending: reporting it here would race write()'s first timed report.
     this->recovery_.rescan_done(true);
     cli_log(LogLevel::INFO, "alsa: '%s' is back -- recovered without waiting for the next stream",
             this->device_.c_str());

src/sink_recovery.cpp

@@ -14,6 +14,8 @@
 
 #include "sink_recovery.h"
 
+#include <limits>
+
 namespace sendspin_cli {
 
 bool SinkRecovery::reopen_due() {
@@ -73,13 +75,29 @@ bool SinkRecovery::pending() const {
     return this->rescan_owed_.load(std::memory_order_relaxed);
 }
 
+void SinkRecovery::discard_frames(uint32_t frames) {
+    const uint32_t room = std::numeric_limits<uint32_t>::max() - this->discarded_frames_;
+    this->discarded_frames_ += (frames < room) ? frames : room;
+}
+
+uint32_t SinkRecovery::take_discarded_frames() {
+    const uint32_t frames = this->discarded_frames_;
+    this->discarded_frames_ = 0;
+    return frames;
+}
+
+void SinkRecovery::forget_discarded_frames() {
+    this->discarded_frames_ = 0;
+}
+
 void SinkRecovery::reset() {
     this->reopen_spent_ = false;
     this->rescan_spent_ = false;
     this->rescan_in_flight_ = false;
     this->rescan_attempts_ = 0;
     this->rescan_owed_.store(false, std::memory_order_relaxed);
     this->rescan_at_ms_ = NOT_STAMPED;
+    this->discarded_frames_ = 0;
 }
 
 void SinkRecovery::escalate_() {

src/sink_recovery.h

@@ -54,6 +54,16 @@ class SinkRecovery {
     /// True while a rescan is still owed. The one method safe to call without the lock.
     bool pending() const;
 
+    /// Records frames accepted and discarded with no device to play them; saturates.
+    /// Stays pending through a recovered rescan until a timed write takes it.
+    void discard_frames(uint32_t frames);
+
+    /// Returns and clears the frames accumulated by discard_frames().
+    uint32_t take_discarded_frames();
+
+    /// Drops the discarded-frame count, leaving the recovery budget alone.
+    void forget_discarded_frames();
+
     /// Refills the budget; call only when configure() really opened a stream.
     void reset();
 
@@ -71,6 +81,8 @@ class SinkRecovery {
     /// Set from handing out an attempt until rescan_done(); blocks re-arming and double counting.
     bool rescan_in_flight_{false};
     int rescan_attempts_{0};
+    /// Frames accepted with no device to play them, not yet retired; see discard_frames().
+    uint32_t discarded_frames_{0};
     /// Read by the main loop without the sink's lock; see pending().
     std::atomic<bool> rescan_owed_{false};
     int64_t rescan_at_ms_{NOT_STAMPED};

tests/sink_recovery_test.cpp

@@ -19,6 +19,7 @@
 #include <gtest/gtest.h>
 
 #include <cstdint>
+#include <limits>
 
 namespace sendspin_cli {
 namespace {
@@ -75,6 +76,81 @@ TEST(SinkRecovery, AFailedReopenEscalatesToTheRescan) {
     EXPECT_TRUE(recovery.pending());
 }
 
+TEST(SinkRecovery, ReturnsTheDiscardedGapOnceWhenADeviceComesBack) {
+    SinkRecovery recovery;
+    escalate(recovery);
+
+    // Fourteen seconds at 48 kHz, accepted while no DAC played them.
+    recovery.discard_frames(14U * 48'000U);
+
+    EXPECT_EQ(recovery.take_discarded_frames(), 672'000U);
+    EXPECT_EQ(recovery.take_discarded_frames(), 0U);
+}
+
+TEST(SinkRecovery, DiscardedGapSaturatesInsteadOfWrapping) {
+    SinkRecovery recovery;
+    recovery.discard_frames(std::numeric_limits<uint32_t>::max() - 10U);
+    recovery.discard_frames(100U);
+
+    EXPECT_EQ(recovery.take_discarded_frames(), std::numeric_limits<uint32_t>::max());
+}
+
+TEST(SinkRecovery, TheDiscardedGapOutlivesARecoveredRescanUntilTaken) {
+    SinkRecovery recovery;
+    escalate(recovery);
+    recovery.discard_frames(48'000U);
+    ASSERT_GE(rescan_fires_at(recovery, T0, T0 + 10 * SINK_RESCAN_DELAY_MS), T0);
+
+    // The reopen has no device timestamp to retire the gap against; the first timed write does.
+    recovery.rescan_done(true);
+
+    EXPECT_EQ(recovery.take_discarded_frames(), 48'000U);
+    EXPECT_EQ(recovery.take_discarded_frames(), 0U);
+}
+
+TEST(SinkRecovery, ForgettingTheDiscardedGapLeavesTheBudgetAlone) {
+    SinkRecovery recovery;
+    escalate(recovery);
+    recovery.discard_frames(48'000U);
+
+    // What a flush does with no device open: the outage is still owed what it was.
+    recovery.forget_discarded_frames();
+
+    EXPECT_EQ(recovery.take_discarded_frames(), 0U);
+    EXPECT_TRUE(recovery.pending());
+    EXPECT_FALSE(recovery.reopen_due());
+    EXPECT_EQ(rescan_fires_at(recovery, T0, T0 + 10 * SINK_RESCAN_DELAY_MS),
+              T0 + SINK_RESCAN_DELAY_MS);
+}
+
+TEST(SinkRecovery, ForgettingTheDiscardedGapDoesNotRefillASpentBudget) {
+    SinkRecovery recovery;
+    escalate(recovery);
+    int64_t now = T0;
+    for (int attempt = 0; attempt < SINK_RESCAN_ATTEMPTS; ++attempt) {
+        now = rescan_fires_at(recovery, now, now + 100 * SINK_RESCAN_DELAY_MS);
+        ASSERT_GT(now, 0) << "attempt " << attempt << " never fired";
+        recovery.rescan_done(false);
+    }
+    recovery.discard_frames(48'000U);
+
+    recovery.forget_discarded_frames();
+
+    EXPECT_FALSE(recovery.pending());
+    EXPECT_FALSE(recovery.reopen_due());
+    EXPECT_EQ(rescan_fires_at(recovery, now, now + 100 * SINK_RESCAN_DELAY_MS), -1);
+}
+
+TEST(SinkRecovery, ResetClearsTheDiscardedGap) {
+    SinkRecovery recovery;
+    escalate(recovery);
+    recovery.discard_frames(48'000U);
+
+    recovery.reset();
+
+    EXPECT_EQ(recovery.take_discarded_frames(), 0U);
+}
+
 TEST(SinkRecovery, EveryFurtherWriteOfTheOutageIsToldToDiscard) {
     SinkRecovery recovery;
     escalate(recovery);