Handle source discontinuities and reject stale device epochs

src/call/call_jitter.cpp

@@ -32,6 +32,8 @@ struct call_jitter {
     call_capture_time timing[CALL_JITTER_MAX_FRAMES];
     CallTimeline timeline;
     uint64_t playback_epoch = 1;
+    uint64_t source_epoch = 0;
+    int64_t capture_end = INT64_MIN;
     int head = 0, count = 0;
     uint32_t dropped = 0, underruns = 0;
     call_jitter_diagnostics diagnostics{};
@@ -168,7 +170,14 @@ int call_jitter_pull(struct call_jitter* j, int16_t* out, int max_samples) {
         }
         for (int i = 0; i < n; i++) j->input[i] = j->pcm[i] / 32768.0f;
         int64_t pts = timing.pts_us == INT64_MIN ? INT64_MIN : timing.pts_us - (int64_t)timing.delay_samples * 1000000 / kSampleRate;
-        j->timeline.feed(pts, (unsigned)n, timing.epoch);
+        bool discontinuity = j->source_epoch != timing.epoch || (timing.pts_us != INT64_MIN && j->capture_end != INT64_MIN
+            && (timing.pts_us < j->capture_end - 2000 || timing.pts_us > j->capture_end + 2000));
+        unsigned prefix = discontinuity ? std::min((unsigned)n, timing.delay_samples) : 0;
+        if (prefix) j->timeline.feed(INT64_MIN, prefix, timing.epoch);
+        if (prefix < (unsigned)n) j->timeline.feed(pts == INT64_MIN ? pts : pts + prefix * 1000000ll / kSampleRate,
+                                                  n - prefix, timing.epoch);
+        j->source_epoch = timing.epoch;
+        j->capture_end = timing.pts_us == INT64_MIN ? INT64_MIN : timing.pts_us + n * 1000000ll / kSampleRate;
         j->st.putSamples(j->input, (uint)n);
     }
     j->output.resize((size_t)max_samples);
@@ -231,6 +240,7 @@ void call_jitter_reset(struct call_jitter* j) {
     j->dropped += discard;
     j->st.clear(); j->st.setTempo(1.0);
     j->timeline.reset(++j->playback_epoch);
+    j->source_epoch = 0; j->capture_end = INT64_MIN;
     j->primed = false; j->catching_up = false;
     j->tempo = 1.0; j->ema_ms = j->arrival_window.reserve;
     j->processing_ms = 0; j->tempo_x100 = 100;

src/call/call_timeline.cpp

@@ -11,10 +11,10 @@ double CallTimeline::at(const Span& s, double position) {
 
 void CallTimeline::reset(uint64_t newEpoch) {
     input.clear(); ready.clear(); output.clear();
-    inputWritten = generated = consumed = written = anchorPosition = minimumPosition = lastSourceEpoch = 0;
+    inputWritten = generated = consumed = written = anchorPosition = minimumPosition = 0;
     epoch = newEpoch; anchorTime = anchorReceived = 0; lastPts = -1;
     endPts = 0; endEpoch = 0; sourceError = 0;
-    outputFailed = haveAnchor = measured = endValid = false;
+    outputFailed = deviceFailed = haveAnchor = measured = endValid = false;
     DEBUG_DEBUG(DEBUG_CATEGORY_CALL, "audio timeline reset epoch=%llu", (unsigned long long)epoch);
 }
 
@@ -78,6 +78,17 @@ void CallTimeline::source(const soundtouch::SourceSpan& s) {
 
 void CallTimeline::appendOutput(const Span& s) {
     if (outputFailed) return;
+    // Источники без PTS и длинные локальные паузы не расходуют по записи на каждый pull.
+    if (output.count) {
+        Span& previous = output.data[(output.head + output.count - 1) % output.data.size()];
+        bool sameHold = previous.pts == previous.end && s.pts == s.end && previous.end == s.pts;
+        if (previous.first + previous.count == s.first && previous.epoch == s.epoch && previous.valid == s.valid
+            && (!s.valid || sameHold) && previous.count <= UINT32_MAX - s.count) {
+            previous.count += s.count; previous.end = s.end;
+            endPts = s.end; endEpoch = s.epoch; endValid = s.valid;
+            return;
+        }
+    }
     if (!output.push(s)) {
         outputFailed = true;
         DEBUG_ERROR(DEBUG_CATEGORY_CALL, "audio timeline output overflow epoch=%llu written=%llu anchor=%llu spans=%zu",
@@ -118,12 +129,12 @@ void CallTimeline::silence(unsigned samples) {
 
 /* at_us — время конца известного устройству участка; будущая отметка учитывает очередь backend. */
 void CallTimeline::progress(uint64_t deviceEpoch, uint64_t position, int64_t atUs, int64_t receivedUs, bool isMeasured) {
-    if (deviceEpoch != epoch) return;
+    if (deviceEpoch != epoch || deviceFailed) return;
     if (position > written || atUs < 0 || receivedUs < 0 || (haveAnchor && (position < anchorPosition || receivedUs < anchorReceived))) {
         DEBUG_WARN(DEBUG_CATEGORY_CALL, "audio clock invalid device progress epoch=%llu position=%llu written=%llu time=%lld received=%lld",
                    (unsigned long long)epoch, (unsigned long long)position, (unsigned long long)written,
                    (long long)atUs, (long long)receivedUs);
-        haveAnchor = false; return;
+        deviceFailed = true; haveAnchor = false; return;
     }
     anchorPosition = position; anchorTime = atUs; anchorReceived = receivedUs; measured = isMeasured; haveAnchor = true;
     uint64_t pending = atUs > receivedUs ? (uint64_t)std::ceil((atUs - receivedUs) / sampleUs) : 0;
@@ -135,7 +146,7 @@ void CallTimeline::progress(uint64_t deviceEpoch, uint64_t position, int64_t atU
 call_clock_snapshot CallTimeline::clock(int64_t nowUs) {
     call_clock_snapshot result{};
     result.epoch = epoch; result.written_samples = written; result.pts_us = lastPts; result.measured = measured;
-    if (outputFailed || !haveAnchor || !output.count || nowUs < anchorReceived || nowUs - anchorReceived > 250000) return result;
+    if (outputFailed || deviceFailed || !haveAnchor || !output.count || nowUs < anchorReceived || nowUs - anchorReceived > 250000) return result;
     uint64_t pending = anchorTime > nowUs ? (uint64_t)std::ceil((anchorTime - nowUs) / sampleUs) : 0;
     uint64_t position = pending > anchorPosition ? 0 : anchorPosition - pending;
     position = std::max(position, minimumPosition);
@@ -147,7 +158,7 @@ call_clock_snapshot CallTimeline::clock(int64_t nowUs) {
     int64_t pts = (int64_t)std::llround(at(s, position));
     if (pts < 0 || (lastPts >= 0 && pts < lastPts - 100000)) return result;
     result.held_back_us = pts < lastPts ? (uint32_t)(lastPts - pts) : 0;
-    lastPts = std::max(pts, lastPts); lastSourceEpoch = s.epoch;
+    lastPts = std::max(pts, lastPts);
     result.pts_us = lastPts; result.valid = 1;
     return result;
 }

src/call/call_timeline.h

@@ -45,12 +45,12 @@ private:
     Ring<1024> ready;
     Ring<2048> output;
     uint64_t inputWritten = 0, generated = 0, consumed = 0, written = 0, epoch = 0;
-    uint64_t anchorPosition = 0, minimumPosition = 0, lastSourceEpoch = 0;
+    uint64_t anchorPosition = 0, minimumPosition = 0;
     int64_t anchorTime = 0, anchorReceived = 0, lastPts = -1;
     double endPts = 0;
     uint64_t endEpoch = 0;
     int sourceError = 0;
-    bool outputFailed = false, haveAnchor = false, measured = false, endValid = false;
+    bool outputFailed = false, deviceFailed = false, haveAnchor = false, measured = false, endValid = false;
     static double at(const Span& s, double position);
     const Span* findInput(uint64_t position) const;
     void appendOutput(const Span& s);