Keep a segment's boundary time from overflowing and resync a source after its ASRC refuses

CLAUDE.md

@@ -95,7 +95,7 @@ script/clang-tidy.sh      - clang-tidy over src/ using build/compile_commands.js
 - Library target `sendspin_server_cpp` (alias `sendspin_server_cpp::sendspin_server_cpp`); CMake options are prefixed `SENDSPIN_SERVER_`. `SENDSPIN_SERVER_ENABLE_WERROR` (default off) makes warnings in this project's targets errors; CI builds with it on, so local builds should too. `SENDSPIN_SERVER_EMBEDDED_PROFILE` (default off) builds with `src/profile.h`'s embedded limits, as the ESP-IDF component (`cmake/esp_idf.cmake`) always does.
 - Plain: `cmake -B build && cmake --build build`, which builds `examples/` too (`SENDSPIN_SERVER_BUILD_EXAMPLES`, default on when top-level). Uses ccache when installed (`SENDSPIN_SERVER_USE_CCACHE`, default on when top-level). `SENDSPIN_SERVER_ENABLE_DISCOVERY` (default on) builds the mDNS backend: dns_sd on macOS, avahi-client through pkg-config on Linux (`libavahi-client-dev`). `SENDSPIN_SERVER_ENABLE_OPUS` (default on) builds the Opus encoder and the source role's Opus decoder; off compiles them out (`src/frame_encoder.cpp` and `src/source_decoder.cpp` are the places it is guarded) and is refused with the tests on.
 - Tests never use the network's mDNS: `tests/test_recorder.h` `local_config()` turns advertising and browsing off and listens on a port the system picks, and `test_server_discovery.cpp` supplies its own `DiscoveryProvider` through `ServerConfig::discovery`.
-- Tests: `cmake -B build-asan -DSENDSPIN_SERVER_ENABLE_WERROR=ON -DSENDSPIN_SERVER_BUILD_TESTS=ON -DSENDSPIN_SERVER_ENABLE_SANITIZERS=ON`, then `cmake --build build-asan` and `ctest --test-dir build-asan --output-on-failure`. The suites whose assertions depend on players keeping up with real time carry the CTest label `realtime` (`tests/CMakeLists.txt`); CI runs them on their own and the rest in parallel (`-LE realtime -j`). The tests also build `server_cli`'s WAV and FLAC readers, feeders, `--one-group`, timing figures and log and `--web` dashboard; `tests/temp_dir.h` gives a test a temporary directory, and `tests/job_view.h` a copy of what a group's pipelines read of a host view. `tests/test_ports.h` hands out loopback ports for test peers, and holds one that nothing listens on for a dial that must be refused; `tests/raw_client.h` is the raw test client (noise-c and Beast, either suite, scriptable misbehavior), `tests/silent_listener.h` a listener that holds the server's dials in their upgrade or handshake, `tests/test_recorder.h` a recording listener, an in-memory provider and `LocalServer`, a server started with them, `tests/wire_messages.h` the spec's binary message types and fields as the tests decode them, `tests/flac_reader.h` decodes a FLAC stream as a player gets it, `tests/opus_reader.h` an Opus stream, with the test tones and an SNR to compare lossy audio by, `tests/player_sink.h` holds the test signal and a real-time audio output for sendspin-cpp players, `tests/peer.h` what each in-process sendspin-cpp client shares (its network provider and blob store, its config and the URL the server dials, its main loop run on the test thread, the client_id at a URL), `tests/trust_fixtures.h` distinct PSKs and the client record a server saved, `tests/hex.h` hex text for known answers, `tests/gate.h` a gate a job waits at, the fixtures and harnesses a class's two test files share (`tests/host_view_fixture.h`, `tests/group_pipelines.h`, `tests/session_harness.h`, `tests/trust_harness.h`), `tests/raw_roles.h` the messages a raw client sends to drive the roles and the readers of what it receives, `tests/players_rig.h` the rig, writes and checks of the in-process player tests, `tests/group_job.h` a group's pipeline job wrapped between runs, for the files compiled with `-fno-access-control` alone, and `tests/freed_blocks.h` counts the heap blocks freed still holding a secret, through the sanitizer's free hook (a test that needs it skips itself without ASan or TSan).
+- Tests: `cmake -B build-asan -DSENDSPIN_SERVER_ENABLE_WERROR=ON -DSENDSPIN_SERVER_BUILD_TESTS=ON -DSENDSPIN_SERVER_ENABLE_SANITIZERS=ON`, then `cmake --build build-asan` and `ctest --test-dir build-asan --output-on-failure`. The suites whose assertions depend on players keeping up with real time carry the CTest label `realtime` (`tests/CMakeLists.txt`); CI runs them on their own and the rest in parallel (`-LE realtime -j`). The tests also build `server_cli`'s WAV and FLAC readers, feeders, `--one-group`, timing figures and log and `--web` dashboard; `tests/temp_dir.h` gives a test a temporary directory, and `tests/job_view.h` a copy of what a group's pipelines read of a host view. `tests/test_ports.h` hands out loopback ports for test peers, and holds one that nothing listens on for a dial that must be refused; `tests/raw_client.h` is the raw test client (noise-c and Beast, either suite, scriptable misbehavior), `tests/silent_listener.h` a listener that holds the server's dials in their upgrade or handshake, `tests/test_recorder.h` a recording listener, an in-memory provider and `LocalServer`, a server started with them, `tests/wire_messages.h` the spec's binary message types and fields as the tests decode them, `tests/flac_reader.h` decodes a FLAC stream as a player gets it, `tests/opus_reader.h` an Opus stream, with the test tones and an SNR to compare lossy audio by, `tests/player_sink.h` holds the test signal and a real-time audio output for sendspin-cpp players, `tests/peer.h` what each in-process sendspin-cpp client shares (its network provider and blob store, its config and the URL the server dials, its main loop run on the test thread, the client_id at a URL), `tests/trust_fixtures.h` distinct PSKs and the client record a server saved, `tests/hex.h` hex text for known answers, `tests/gate.h` a gate a job waits at, the fixtures and harnesses a class's two test files share (`tests/host_view_fixture.h`, `tests/group_pipelines.h`, `tests/source_sink_rig.h`, `tests/session_harness.h`, `tests/trust_harness.h`), `tests/raw_roles.h` the messages a raw client sends to drive the roles and the readers of what it receives, `tests/players_rig.h` the rig, writes and checks of the in-process player tests, `tests/group_job.h` a group's pipeline job wrapped between runs, for the files compiled with `-fno-access-control` alone, and `tests/freed_blocks.h` counts the heap blocks freed still holding a secret, through the sanitizer's free hook (a test that needs it skips itself without ASan or TSan).
 - Soak: `tests/soak.cpp` (its trace checker in `tests/soak_wire_checker.h`, tested in the default suite), target `sendspin_server_cpp_soak` with its own entry point, registered as the CTest test `Soak` (label `soak`), which skips itself unless `SENDSPIN_SERVER_SOAK_SECONDS` is set (`SENDSPIN_SERVER_SOAK_SEED` picks the script); README.md has the command.
 - CI also builds the library with Opus and discovery off and memory statistics on; as a subproject (`tests/consumer/`); and as the ESP-IDF component (`tests/esp_idf/`: `idf.py -C tests/esp_idf set-target esp32s3`, then `idf.py -C tests/esp_idf build`).
 - Memory statistics: `SENDSPIN_SERVER_MEMORY_STATS` (default off; `CONFIG_SENDSPIN_SERVER_MEMORY_STATS` in the component's `Kconfig`) logs the heaps and the tasks' stacks at the server's start (on the thread calling `start()`) and, on the core thread, at a connection's start and end and a group's playback (`profile::MEMORY_STATS`); the tests count log lines and expect it off.

src/sched/timeline.cpp

@@ -29,7 +29,7 @@ const Segment& segment_for(const Anchor& anchor, uint64_t position) {
     return it == anchor.segments.begin() ? anchor.segments.front() : *std::prev(it);
 }
 
-/// Picoseconds in a second, the unit of a segment's elapsed time
+/// Picoseconds in a second, the unit of a segment's elapsed time past its whole seconds
 constexpr int64_t PS_PER_SECOND = 1000000000000;
 
 /// `floor((p - p_j) * 10^6 / r_j)` for the given segment
@@ -57,11 +57,15 @@ void reset_anchor(Anchor& anchor, int64_t t0_us, uint64_t p0, uint32_t rate, uin
 void add_segment(Anchor& anchor, uint64_t position, uint32_t rate, uint8_t channels) {
     const Segment& last = anchor.segments.back();
     const int64_t unshifted = last.unshifted_us + offset_us(last, position);
-    // Whole seconds and the rest apart, so neither product overflows
+    // The segment's whole seconds and the rest apart, so the rest's product stays small
     const auto samples = static_cast<int64_t>(position - last.position);
-    const int64_t elapsed_ps = last.elapsed_ps + ((samples / last.rate) * PS_PER_SECOND) +
-                               ((samples % last.rate) * PS_PER_SECOND / last.rate);
-    anchor.segments.push_back({position, unshifted, rate, channels, elapsed_ps});
+    int64_t elapsed_s = last.elapsed_s + (samples / last.rate);
+    int64_t elapsed_ps = last.elapsed_ps + ((samples % last.rate) * PS_PER_SECOND / last.rate);
+    if (elapsed_ps >= PS_PER_SECOND) {
+        ++elapsed_s;
+        elapsed_ps -= PS_PER_SECOND;
+    }
+    anchor.segments.push_back({position, unshifted, rate, channels, elapsed_s, elapsed_ps});
 }
 
 void add_shift(Anchor& anchor, int64_t index, int64_t delta_us) {
@@ -140,11 +144,10 @@ int64_t first_index_at_or_after_position(const Anchor& anchor, uint64_t position
 }
 
 int64_t segment_output_index(const Anchor& anchor, size_t segment, uint32_t rate) {
-    const int64_t elapsed = anchor.segments[segment].elapsed_ps;
-    // Whole seconds and the rest apart, so neither product overflows; the rest rounds half up
-    const int64_t rest = elapsed % PS_PER_SECOND;
-    return ((elapsed / PS_PER_SECOND) * int64_t{rate}) +
-           (((2 * rest * int64_t{rate}) + PS_PER_SECOND) / (2 * PS_PER_SECOND));
+    const Segment& boundary = anchor.segments[segment];
+    // The picoseconds past the whole seconds round half up
+    return (boundary.elapsed_s * int64_t{rate}) +
+           (((2 * boundary.elapsed_ps * int64_t{rate}) + PS_PER_SECOND) / (2 * PS_PER_SECOND));
 }
 
 int64_t instance_start(const Anchor& anchor, int64_t requested, uint32_t cells) {

src/sched/timeline.h

@@ -49,9 +49,10 @@ struct Segment {
     int64_t unshifted_us{0};
     uint32_t rate{0};
     uint8_t channels{0};
-    /// The time from `T0` to `position` in picoseconds, floored at each segment: the boundary
-    /// time output indices derive from, which `u_j`'s whole microseconds would drift by up to
-    /// one per segment
+    /// The time from `T0` to `position`, whole seconds and the picoseconds past them (below
+    /// 10^12), floored to the picosecond at each segment: the boundary time output indices derive
+    /// from, which `u_j`'s whole microseconds would drift by up to one per segment
+    int64_t elapsed_s{0};
     int64_t elapsed_ps{0};
 };
 

src/source_clock.h

@@ -154,7 +154,7 @@ class SourceAsrc {
     /// what the filter holds at the current ratio
     double next_position() const;
 
-    /// @brief The output frames written since create()
+    /// @brief The output frames written since create() or the last reset()
     uint64_t produced() const {
         return this->produced_;
     }

src/source_sink.cpp

@@ -159,7 +159,15 @@ void LiveSourceSink::correct(const uint8_t* pcm, size_t frames, int64_t capture_
     const uint64_t first = this->asrc_->produced();
     const std::optional<SourceAsrc::Processed> processed = this->asrc_->process(
         floats, frames, engaged ? ppm : 0.0, corrected, SourceAsrc::max_output(frames));
-    if (!engaged || !processed) {
+    if (!processed) {
+        // Written unaltered; the ASRC's position, which did not move with it, and its state past
+        // the blocks it took start over, else its next output would be cut as written already. What
+        // its filter held of the chunk before is dropped.
+        (void)this->view_->write(floats, frames);
+        this->restart_correction();
+        return;
+    }
+    if (!engaged) {
         (void)this->view_->write(floats, frames);
         this->written_ += frames;
         return;

src/source_sink.h

@@ -113,7 +113,7 @@ class LiveSourceSink : public SourcePcmSink {
 private:
     /// The run ends unless a stop by the server holds it or the input closed
     void stop_run();
-    /// The corrected positions start over, for a new run
+    /// The corrected positions start over: for a new run, or after the ASRC refuses a chunk
     void restart_correction();
     /// Corrects one chunk's frames and writes them
     void correct(const uint8_t* pcm, size_t frames, int64_t capture_us);
@@ -128,8 +128,8 @@ class LiveSourceSink : public SourcePcmSink {
     /// The corrected frames written since the corrected positions started
     uint64_t written_{0};
     SampleSlip slip_;
-    /// With profile::SOURCE_ASRC: runs from the stream's start, its output written once the
-    /// correction starts, from the position the unaltered frames reached
+    /// With profile::SOURCE_ASRC: runs from the stream's start or its last restart, its output
+    /// written once the correction starts, from the position the unaltered frames reached
     std::unique_ptr<SourceAsrc> asrc_;
     /// A chunk as floats, then corrected (profile::SOURCE_ASRC), or slipped; in PSRAM where there
     /// is one

tests/CMakeLists.txt

@@ -98,6 +98,7 @@ add_executable(sendspin_server_cpp_tests
     test_source_decoder.cpp
     test_source_feed.cpp
     test_source_sink.cpp
+    test_source_sink_internals.cpp
     test_source_stream.cpp
     test_soak_wire_checker.cpp
     test_task_queue.cpp
@@ -236,6 +237,12 @@ set_source_files_properties(test_session_driver_internals.cpp PROPERTIES COMPILE
 # them, which nothing the session sends or reports shows.
 set_source_files_properties(test_session_internals.cpp PROPERTIES COMPILE_OPTIONS
                             -fno-access-control)
+# test_source_sink_internals.cpp reads LiveSourceSink::clock_ to learn when its correction has
+# started and sets SourceAsrc::max_block_ of LiveSourceSink::asrc_ to zero for one chunk, so
+# micro-resampler takes nothing and process() refuses: no input the server receives makes it
+# refuse.
+set_source_files_properties(test_source_sink_internals.cpp PROPERTIES COMPILE_OPTIONS
+                            -fno-access-control)
 
 # sendspin_server_cpp_beast last: a system Boost's include directory (e.g. /opt/homebrew/include) must
 # not shadow the fetched dependencies' headers.

tests/source_sink_rig.h

@@ -0,0 +1,128 @@
+// 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.
+
+/// @file source_sink_rig.h
+/// @brief SinkRig, a live input a LiveSourceSink writes into and the rig reads back, with the
+/// source's test tone
+
+#pragma once
+
+#include "dispatcher.h"
+#include "host_view.h"
+#include "job_view.h"
+#include "pcm_convert.h"
+#include "source_sink.h"
+#include <gtest/gtest.h>
+
+#include <cmath>
+#include <cstdint>
+#include <memory>
+#include <numbers>
+#include <vector>
+
+namespace sendspin::server::test {
+
+/// The source's write jitter, the staleness bound its chunks are judged by
+inline constexpr int64_t JITTER_US = 60000;
+/// The path from capture end to receipt when nothing holds the audio back
+inline constexpr int64_t PATH_US = 3000;
+
+/// The rig's rate, and a source's 20 ms chunk at it
+inline constexpr uint32_t RATE = 48000;
+inline constexpr size_t CHUNK = RATE / 50;
+/// The test tone's largest step between two frames at RATE: 100 Hz at half scale
+inline const double TONE_STEP = 0.5 * 2.0 * std::numbers::pi * 100.0 / RATE;
+
+/// A source's tone at frame position, as a 16-bit sample
+inline int16_t tone(uint64_t position) {
+    return static_cast<int16_t>(std::lround(
+        16383.0 * std::sin(2.0 * std::numbers::pi * 100.0 * static_cast<double>(position) / RATE)));
+}
+
+/// A mono live input a LiveSourceSink writes into, whose staged audio the rig reads back as
+/// floats and releases, as the pipelines would. Its group lead is raised past everything written
+/// once the run is anchored, so the input never cuts a write to lower its latency.
+class SinkRig {
+public:
+    explicit SinkRig(uint32_t rate = RATE, ResamplerQuality quality = ResamplerQuality::MEDIUM)
+        : rate_(rate) {
+        InputOptions options{{rate, 1, SampleFormat::S16}};
+        options.kind = InputKind::LIVE;
+        this->view = std::make_shared<HostView>(GroupId{1}, InputId{2}, options,
+                                                GroupParams{300000, 0, WRITE_AHEAD_US},
+                                                this->dispatcher, nullptr);
+        this->sink = std::make_unique<LiveSourceSink>(this->view, JITTER_US, quality);
+    }
+
+    ~SinkRig() {
+        this->dispatcher->finish();
+        this->dispatcher->join();
+    }
+
+    SinkRig(const SinkRig&) = delete;
+    SinkRig& operator=(const SinkRig&) = delete;
+
+    /// One chunk of the tone from position, captured at capture_us and received on its path
+    void write(uint64_t position, size_t frames, int64_t capture_us, int64_t late_us = 0) {
+        std::vector<int16_t> pcm(frames);
+        for (size_t i = 0; i < frames; ++i) {
+            pcm[i] = tone(position + i);
+        }
+        const int64_t end = capture_us + static_cast<int64_t>(frames) * 1000000 / this->rate_;
+        this->sink->write(reinterpret_cast<const uint8_t*>(pcm.data()), frames, capture_us,
+                          end + PATH_US + late_us);
+        if (!this->anchored_) {
+            this->view->set_group_params({LEAD_PAST_ALL_US, 0, WRITE_AHEAD_US});
+            this->anchored_ = true;
+        }
+        this->collect();
+    }
+
+    std::shared_ptr<Dispatcher> dispatcher = std::make_shared<Dispatcher>();
+    std::shared_ptr<HostView> view;
+    std::unique_ptr<LiveSourceSink> sink;
+    /// What the input staged, as floats, unless only counted
+    std::vector<float> staged;
+    bool count_only{false};
+    uint64_t staged_count{0};
+
+private:
+    static constexpr int64_t WRITE_AHEAD_US = int64_t{1000} * 1000000;
+    static constexpr int64_t LEAD_PAST_ALL_US = int64_t{1000} * 1000000;
+
+    void collect() {
+        const JobView job = job_view_of(*this->view);
+        if (job.head <= this->read_to_) {
+            return;
+        }
+        const auto frames = static_cast<size_t>(job.head - this->read_to_);
+        this->staged_count += frames;
+        if (!this->count_only) {
+            std::vector<uint8_t> bytes;
+            ASSERT_TRUE(this->view->read(job.run.epoch, this->read_to_, frames, bytes));
+            const SampleFormat format = this->view->format().format.sample_format;
+            std::vector<float> floats(frames);
+            to_float(bytes, format, floats);
+            this->staged.insert(this->staged.end(), floats.begin(), floats.end());
+        }
+        this->read_to_ = job.head;
+        this->view->release_before(job.run.epoch, job.head);
+    }
+
+    uint32_t rate_;
+    bool anchored_{false};
+    uint64_t read_to_{0};
+};
+
+}  // namespace sendspin::server::test