From 548df24c5aff73da9f4c86c02c418eebea1de4b7 Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Mon, 31 Aug 2026 20:50:58 -0500 Subject: [PATCH 1/3] Cover the source role with loopback wire-order tests --- tests/test_source_role.cpp | 395 ++++++++++++++++++++++++++++++++++++- 1 file changed, 394 insertions(+), 1 deletion(-) diff --git a/tests/test_source_role.cpp b/tests/test_source_role.cpp index 295c5e8..732ae84 100644 --- a/tests/test_source_role.cpp +++ b/tests/test_source_role.cpp @@ -12,11 +12,15 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Source role tests. Two layers, mirroring the acceptance criteria: +// Source role tests. Three layers, mirroring the acceptance criteria: // - Pure chunk/timestamp bookkeeping helpers from source_task.h, tested directly. // - SourceRole::Impl-level tests (make_impl pattern from test_artwork_role.cpp): config // validation, hello/state field building, the command latch's fail-closed path, and // lifecycle callback pairing. +// - End-to-end wire tests driving a real SendspinClient against an in-test loopback +// IXWebSocket "server" (the harness test_connection_lifecycle.cpp establishes), proving the +// wire ordering client-stream/start -> typed+stamped binary chunks -> client-stream/end and +// every streaming gate (no start command, stale connection, no time sync, after stop). #include "connection_manager.h" #include "inbox.h" @@ -29,6 +33,7 @@ #include "source_task.h" #include #include +#include #include #include @@ -47,6 +52,18 @@ using namespace sendspin; // NOLINT(google-build-using-namespace): test-local c namespace { +// Distinct ports per wire test (and distinct from test_connection_lifecycle.cpp's 189xx range) +// so a lingering socket from one scenario cannot bleed into the next. +constexpr uint16_t WIRE_TEST_PORT = 19010; +constexpr uint16_t TIME_GATE_TEST_PORT = 19011; +constexpr uint16_t RECONNECT_TEST_PORT = 19012; + +// Default-config wire framing, derived exactly as source_task.cpp derives it: 25 ms at +// 48 kHz stereo 16-bit -> 1200 frames x 4 bytes, behind a 1-byte type + 8-byte timestamp header. +constexpr size_t WIRE_BYTES_PER_FRAME = 4; +constexpr size_t WIRE_CHUNK_PAYLOAD = 1200 * WIRE_BYTES_PER_FRAME; +constexpr size_t WIRE_HEADER = 9; + // ============================================================================ // Pure bookkeeping helpers (source_task.h) // ============================================================================ @@ -360,4 +377,380 @@ TEST(SourceWriteAudio, RejectedWhenNotStreaming) { EXPECT_FALSE(impl->write_audio(frame, sizeof(frame), 0)); } +// ============================================================================ +// Wire tests: a real client against an in-test loopback Sendspin "server" +// ============================================================================ + +std::string server_url(uint16_t port) { + return "ws://127.0.0.1:" + std::to_string(port) + "/sendspin"; +} + +SendspinClientConfig make_config(uint16_t port) { + SendspinClientConfig config; + config.client_id = "source-test-client"; + config.name = "Source Test Client"; + config.server_port = port; + // Fast time-sync cadence so a burst that goes unanswered (the time-gate test's first + // phase) retries within the test budget instead of the production 10 s interval. + config.time_burst_interval_ms = 200; + config.time_burst_response_timeout_ms = 200; + return config; +} + +class TestNetworkProvider : public SendspinNetworkProvider { +public: + bool is_network_ready() override { + return true; + } +}; + +/// Pumps client.loop() until the predicate returns true or the timeout elapses. +bool pump_until(SendspinClient& client, const std::function& pred, int timeout_ms) { + const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms); + while (std::chrono::steady_clock::now() < deadline) { + client.loop(); + if (pred()) { + return true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(2)); + } + return false; +} + +void pump_for(SendspinClient& client, int duration_ms) { + pump_until( + client, [] { return false; }, duration_ms); +} + +/// A minimal Sendspin "server" for source streaming: an IXWebSocket client that connects to +/// the SendspinClient's WS server, answers client/hello with server/hello, optionally answers +/// client/time with a zero-offset server/time, records every received text and binary message +/// in arrival order, and can send server/command source start/stop. +class FakeSourceServer { +public: + struct WireEvent { + bool binary; + std::string data; + }; + + FakeSourceServer(const std::string& url, std::string server_id, bool answer_time = true) + : server_id_(std::move(server_id)), answer_time_(answer_time) { + this->ws_.setUrl(url); + this->ws_.disableAutomaticReconnection(); + this->ws_.setOnMessageCallback([this](const ix::WebSocketMessagePtr& msg) { + if (msg->type != ix::WebSocketMessageType::Message) { + return; + } + { + std::lock_guard lock(this->mutex_); + this->events_.push_back({msg->binary, msg->str}); + } + if (msg->binary) { + return; + } + if (msg->str.find("client/hello") != std::string::npos) { + this->ws_.send( + std::string(R"({"type":"server/hello","payload":{"server_id":")") + + this->server_id_ + + R"(","name":"Fake Source Server","version":1,"active_roles":["source"],)" + + R"("connection_reason":"discovery"}})"); + } else if (msg->str.find("client/time") != std::string::npos && + this->answer_time_.load()) { + // Zero-offset echo: server clock == client clock, which the Kalman filter + // accepts as its first measurement. + JsonDocument doc; + if (deserializeJson(doc, msg->str) == DeserializationError::Ok) { + const int64_t t = doc["payload"]["client_transmitted"] | int64_t{0}; + this->ws_.send(std::string(R"({"type":"server/time","payload":)") + + R"({"client_transmitted":)" + std::to_string(t) + + R"(,"server_received":)" + std::to_string(t) + + R"(,"server_transmitted":)" + std::to_string(t) + "}}"); + } + } + }); + this->ws_.start(); + } + + ~FakeSourceServer() { + this->ws_.stop(); + } + + void enable_time_answers() { + this->answer_time_.store(true); + } + + void send_source_command(const char* command) { + this->ws_.send(std::string(R"({"type":"server/command","payload":{"source":)") + + R"({"command":")" + command + R"("}}})"); + } + + void disconnect() { + this->ws_.stop(); + } + + std::vector snapshot() const { + std::lock_guard lock(this->mutex_); + return this->events_; + } + + size_t count_text_containing(const std::string& needle) const { + std::lock_guard lock(this->mutex_); + size_t count = 0; + for (const auto& event : this->events_) { + if (!event.binary && event.data.find(needle) != std::string::npos) { + ++count; + } + } + return count; + } + + size_t binary_count() const { + std::lock_guard lock(this->mutex_); + size_t count = 0; + for (const auto& event : this->events_) { + if (event.binary) { + ++count; + } + } + return count; + } + +private: + ix::WebSocket ws_; + std::string server_id_; + std::atomic answer_time_; + mutable std::mutex mutex_; + std::vector events_; +}; + +/// Everything a wire test needs running: a client with the source role added and started, plus +/// helpers to bring a fake server to the handshake-complete, time-synced state. +struct WireHarness { + explicit WireHarness(uint16_t port, const SourceRoleConfig& role_config = SourceRoleConfig{}) + : client(make_config(port)) { + auto& role = this->client.add_source(role_config); + role.set_listener(&this->listener); + this->client.set_network_provider(&this->network); + } + + bool start() { + if (!this->client.start_server()) { + return false; + } + // The WS server starts synchronously on the first loop() once the network is ready. + pump_for(this->client, 50); + return true; + } + + bool establish(FakeSourceServer& fake) { + if (!pump_until( + this->client, [&] { return this->client.is_connected(); }, 4000)) { + return false; + } + (void)fake; + return pump_until( + this->client, [&] { return this->client.is_time_synced(); }, 4000); + } + + TestNetworkProvider network; + RecordingSourceListener listener; + SendspinClient client; +}; + +// Deliberately independent of protocol_messages.h's be64_to_host(): the wire tests audit the +// bytes the library produced, so the reader must not share the code under audit. +int64_t read_be64(const uint8_t* bytes) { + uint64_t val = 0; + for (int i = 0; i < 8; ++i) { + val = (val << 8) | bytes[i]; + } + return static_cast(val); +} + +// The end-to-end pipeline and its exact wire ordering: client-stream/start, then binary chunks +// carrying [type 12][BE64 first-sample capture timestamp][PCM payload], then a final short +// chunk and client-stream/end on stop -- with write_audio() gated shut before the start +// command, validating frames while open, and gated shut again after the stop. +TEST(SourceWire, StreamsPcmEndToEndWithExactOrdering) { + WireHarness harness(WIRE_TEST_PORT); + ASSERT_TRUE(harness.start()); + FakeSourceServer fake(server_url(WIRE_TEST_PORT), "source-server-a"); + ASSERT_TRUE(harness.establish(fake)); + + SourceRole* source = harness.client.source(); + ASSERT_NE(source, nullptr); + + // Gate: no streaming (and no accepted capture) before the server commands start. + uint8_t frame[WIRE_BYTES_PER_FRAME] = {9, 9, 9, 9}; + EXPECT_FALSE(source->write_audio(frame, sizeof(frame), 0)); + EXPECT_FALSE(source->is_streaming()); + + fake.send_source_command("start"); + ASSERT_TRUE(pump_until( + harness.client, + [&] { + return harness.listener.started == 1 && + fake.count_text_containing("client-stream/start") == 1; + }, + 4000)); + EXPECT_TRUE(source->is_streaming()); + + // Validation while open: a non-whole-frame write is rejected as a whole. + std::vector odd(WIRE_BYTES_PER_FRAME * 10 + 1, 0xAA); + EXPECT_FALSE(source->write_audio(odd.data(), odd.size(), 0)); + + // Overflow: a write larger than the whole capture ring cannot be buffered -- rejected and + // counted as a drop, never blocked on. + std::vector oversized(WIRE_BYTES_PER_FRAME * 48000, 0); // 1 s >> 150 ms ring + EXPECT_FALSE(source->write_audio(oversized.data(), oversized.size(), 0)); + + // 50 ms of ramp PCM in 10 ms writes, capture-stamped 10 ms apart: exactly two full 25 ms + // chunks. + const int64_t base = platform_time_us(); + std::vector chunk(WIRE_BYTES_PER_FRAME * 480); + size_t ramp = 0; + for (int i = 0; i < 5; ++i) { + for (auto& byte : chunk) { + byte = static_cast(ramp++ & 0xFF); + } + ASSERT_TRUE(source->write_audio(chunk.data(), chunk.size(), base + i * 10000)); + } + ASSERT_TRUE(pump_until( + harness.client, [&] { return fake.binary_count() == 2; }, 4000)); + + // A duplicate start while streaming MUST NOT restart the stream: no second + // client-stream/start, no second listener callback. + fake.send_source_command("start"); + pump_for(harness.client, 300); + EXPECT_EQ(fake.count_text_containing("client-stream/start"), 1U); + EXPECT_EQ(harness.listener.started, 1); + + // A 10 ms remainder: full chunks only mid-stream, so it must NOT be sent yet... + for (auto& byte : chunk) { + byte = static_cast(ramp++ & 0xFF); + } + ASSERT_TRUE(source->write_audio(chunk.data(), chunk.size(), base + 50000)); + pump_for(harness.client, 200); + EXPECT_EQ(fake.binary_count(), 2U); + + // ...until stream end, where the spec allows a final short chunk before client-stream/end. + fake.send_source_command("stop"); + ASSERT_TRUE(pump_until( + harness.client, + [&] { + return harness.listener.stopped == 1 && + fake.count_text_containing("client-stream/end") == 1; + }, + 4000)); + EXPECT_FALSE(source->is_streaming()); + EXPECT_FALSE(source->write_audio(frame, sizeof(frame), 0)); + + // Now audit the recorded wire order and framing. + const auto events = fake.snapshot(); + ptrdiff_t start_idx = -1; + ptrdiff_t end_idx = -1; + std::vector binary_idx; + for (ptrdiff_t i = 0; i < static_cast(events.size()); ++i) { + if (events[i].binary) { + binary_idx.push_back(i); + } else if (events[i].data.find("client-stream/start") != std::string::npos) { + start_idx = i; + } else if (events[i].data.find("client-stream/end") != std::string::npos) { + end_idx = i; + } + } + ASSERT_EQ(binary_idx.size(), 3U); // two full chunks + the final short chunk + ASSERT_GE(start_idx, 0); + ASSERT_GE(end_idx, 0); + EXPECT_LT(start_idx, binary_idx.front()); + EXPECT_LT(binary_idx.back(), end_idx); + + // Framing: type byte 12, big-endian first-sample capture timestamp (zero-offset time sync, + // so the server-domain value stays near the stamps we supplied), untouched PCM payload. + const auto& first = events[static_cast(binary_idx[0])].data; + ASSERT_EQ(first.size(), WIRE_HEADER + WIRE_CHUNK_PAYLOAD); + EXPECT_EQ(static_cast(first[0]), SENDSPIN_BINARY_SOURCE_AUDIO); + const int64_t ts1 = read_be64(reinterpret_cast(first.data()) + 1); + EXPECT_LT(std::llabs(ts1 - base), 1000000); + for (size_t i = 0; i < 16; ++i) { + EXPECT_EQ(static_cast(first[WIRE_HEADER + i]), static_cast(i & 0xFF)); + } + + // The second chunk's anchor is exactly one chunk after the first (contiguous capture + // stamps); allow slack for the time filter refining its offset between the two sends. + const auto& second = events[static_cast(binary_idx[1])].data; + ASSERT_EQ(second.size(), WIRE_HEADER + WIRE_CHUNK_PAYLOAD); + const int64_t ts2 = read_be64(reinterpret_cast(second.data()) + 1); + EXPECT_NEAR(static_cast(ts2 - ts1), 25000.0, 2000.0); + + // The final short chunk carries the 10 ms remainder. + const auto& last = events[static_cast(binary_idx[2])].data; + EXPECT_EQ(last.size(), WIRE_HEADER + WIRE_BYTES_PER_FRAME * 480); +} + +// The time-sync convergence gate: a start command on a connection whose time filter has no +// measurement yet must not open the stream; it opens (without a fresh command) once sync +// converges. +TEST(SourceWire, NoStreamBeforeTimeSyncConvergence) { + WireHarness harness(TIME_GATE_TEST_PORT); + ASSERT_TRUE(harness.start()); + FakeSourceServer fake(server_url(TIME_GATE_TEST_PORT), "source-server-t", + /*answer_time=*/false); + ASSERT_TRUE(pump_until( + harness.client, [&] { return harness.client.is_connected(); }, 4000)); + + fake.send_source_command("start"); + pump_for(harness.client, 400); + EXPECT_EQ(fake.count_text_containing("client-stream/start"), 0U); + EXPECT_EQ(harness.listener.started, 0); + + fake.enable_time_answers(); + ASSERT_TRUE(pump_until( + harness.client, + [&] { + return harness.client.is_time_synced() && + fake.count_text_containing("client-stream/start") == 1 && + harness.listener.started == 1; + }, + 4000)); +} + +// Streaming permission is per-connection: a disconnect closes the stream, and a new connection +// (fresh handshake, fresh time sync) must not stream until IT commands start. +TEST(SourceWire, PermissionDoesNotSurviveReconnect) { + WireHarness harness(RECONNECT_TEST_PORT); + ASSERT_TRUE(harness.start()); + + { + FakeSourceServer fake(server_url(RECONNECT_TEST_PORT), "source-server-r1"); + ASSERT_TRUE(harness.establish(fake)); + fake.send_source_command("start"); + ASSERT_TRUE(pump_until( + harness.client, [&] { return harness.listener.started == 1; }, 4000)); + + fake.disconnect(); + ASSERT_TRUE(pump_until( + harness.client, [&] { return harness.listener.stopped == 1; }, 4000)); + EXPECT_FALSE(harness.client.source()->is_streaming()); + } + + FakeSourceServer fake2(server_url(RECONNECT_TEST_PORT), "source-server-r2"); + ASSERT_TRUE(harness.establish(fake2)); + + // The old connection's permission must not leak into this one. + pump_for(harness.client, 400); + EXPECT_EQ(fake2.count_text_containing("client-stream/start"), 0U); + EXPECT_EQ(harness.listener.started, 1); + + // Only this connection's own start command opens a stream. + fake2.send_source_command("start"); + ASSERT_TRUE(pump_until( + harness.client, + [&] { + return harness.listener.started == 2 && + fake2.count_text_containing("client-stream/start") == 1; + }, + 4000)); +} + } // namespace From fcffae70071b5296f7c58ba496aee0ebd3799679 Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Fri, 4 Sep 2026 18:04:33 -0500 Subject: [PATCH 2/3] Tighten the source wire tests from review Audit every payload byte of every chunk against the write-side ramp, pin the end-to-end test's 25 ms chunk arithmetic explicitly instead of leaning on the config default, split the mid-stream validation probes into their own focused test with an accepting control, and outlast the failed-open retry window before the reconnect test's fresh command. --- tests/test_source_role.cpp | 69 ++++++++++++++++++++++++++++++-------- 1 file changed, 55 insertions(+), 14 deletions(-) diff --git a/tests/test_source_role.cpp b/tests/test_source_role.cpp index 732ae84..881745a 100644 --- a/tests/test_source_role.cpp +++ b/tests/test_source_role.cpp @@ -57,6 +57,7 @@ namespace { constexpr uint16_t WIRE_TEST_PORT = 19010; constexpr uint16_t TIME_GATE_TEST_PORT = 19011; constexpr uint16_t RECONNECT_TEST_PORT = 19012; +constexpr uint16_t WRITE_GATE_TEST_PORT = 19013; // Default-config wire framing, derived exactly as source_task.cpp derives it: 25 ms at // 48 kHz stereo 16-bit -> 1200 frames x 4 bytes, behind a 1-byte type + 8-byte timestamp header. @@ -572,7 +573,11 @@ int64_t read_be64(const uint8_t* bytes) { // chunk and client-stream/end on stop -- with write_audio() gated shut before the start // command, validating frames while open, and gated shut again after the stop. TEST(SourceWire, StreamsPcmEndToEndWithExactOrdering) { - WireHarness harness(WIRE_TEST_PORT); + // The chunk arithmetic below encodes 25 ms chunks; pinned here so the test cannot drift + // with the config default. + SourceRoleConfig wire_config; + wire_config.chunk_duration_ms = 25; + WireHarness harness(WIRE_TEST_PORT, wire_config); ASSERT_TRUE(harness.start()); FakeSourceServer fake(server_url(WIRE_TEST_PORT), "source-server-a"); ASSERT_TRUE(harness.establish(fake)); @@ -595,15 +600,6 @@ TEST(SourceWire, StreamsPcmEndToEndWithExactOrdering) { 4000)); EXPECT_TRUE(source->is_streaming()); - // Validation while open: a non-whole-frame write is rejected as a whole. - std::vector odd(WIRE_BYTES_PER_FRAME * 10 + 1, 0xAA); - EXPECT_FALSE(source->write_audio(odd.data(), odd.size(), 0)); - - // Overflow: a write larger than the whole capture ring cannot be buffered -- rejected and - // counted as a drop, never blocked on. - std::vector oversized(WIRE_BYTES_PER_FRAME * 48000, 0); // 1 s >> 150 ms ring - EXPECT_FALSE(source->write_audio(oversized.data(), oversized.size(), 0)); - // 50 ms of ramp PCM in 10 ms writes, capture-stamped 10 ms apart: exactly two full 25 ms // chunks. const int64_t base = platform_time_us(); @@ -672,9 +668,6 @@ TEST(SourceWire, StreamsPcmEndToEndWithExactOrdering) { EXPECT_EQ(static_cast(first[0]), SENDSPIN_BINARY_SOURCE_AUDIO); const int64_t ts1 = read_be64(reinterpret_cast(first.data()) + 1); EXPECT_LT(std::llabs(ts1 - base), 1000000); - for (size_t i = 0; i < 16; ++i) { - EXPECT_EQ(static_cast(first[WIRE_HEADER + i]), static_cast(i & 0xFF)); - } // The second chunk's anchor is exactly one chunk after the first (contiguous capture // stamps); allow slack for the time filter refining its offset between the two sends. @@ -686,6 +679,52 @@ TEST(SourceWire, StreamsPcmEndToEndWithExactOrdering) { // The final short chunk carries the 10 ms remainder. const auto& last = events[static_cast(binary_idx[2])].data; EXPECT_EQ(last.size(), WIRE_HEADER + WIRE_BYTES_PER_FRAME * 480); + + // Every payload byte of every chunk continues the write-side ramp: chunk boundaries and + // assembly cannot silently reorder, duplicate, or corrupt capture bytes. + size_t expected_ramp = 0; + bool payload_intact = true; + for (const ptrdiff_t idx : binary_idx) { + const auto& chunk_data = events[static_cast(idx)].data; + for (size_t i = WIRE_HEADER; i < chunk_data.size(); ++i) { + if (static_cast(chunk_data[i]) != + static_cast(expected_ramp++ & 0xFF)) { + payload_intact = false; + break; + } + } + if (!payload_intact) { + break; + } + } + EXPECT_TRUE(payload_intact); + EXPECT_EQ(expected_ramp, WIRE_BYTES_PER_FRAME * 480 * 6); +} + +// Mid-stream capture validation, focused so a regression here names itself: a non-whole-frame +// write is rejected as a whole, and a write larger than the whole capture ring is rejected as +// a counted drop rather than blocking the producer. +TEST(SourceWire, RejectsMalformedAndOversizedWritesWhileOpen) { + WireHarness harness(WRITE_GATE_TEST_PORT); + ASSERT_TRUE(harness.start()); + FakeSourceServer fake(server_url(WRITE_GATE_TEST_PORT), "source-server-w"); + ASSERT_TRUE(harness.establish(fake)); + SourceRole* source = harness.client.source(); + ASSERT_NE(source, nullptr); + + fake.send_source_command("start"); + ASSERT_TRUE(pump_until( + harness.client, [&] { return harness.listener.started == 1; }, 4000)); + + std::vector odd(WIRE_BYTES_PER_FRAME * 10 + 1, 0xAA); + EXPECT_FALSE(source->write_audio(odd.data(), odd.size(), 0)); + + std::vector oversized(WIRE_BYTES_PER_FRAME * 48000, 0); // 1 s >> 150 ms ring + EXPECT_FALSE(source->write_audio(oversized.data(), oversized.size(), 0)); + + // Control: a well-formed write on the same open stream is accepted. + std::vector good(WIRE_BYTES_PER_FRAME * 480, 0x42); + EXPECT_TRUE(source->write_audio(good.data(), good.size(), 0)); } // The time-sync convergence gate: a start command on a connection whose time filter has no @@ -738,7 +777,9 @@ TEST(SourceWire, PermissionDoesNotSurviveReconnect) { ASSERT_TRUE(harness.establish(fake2)); // The old connection's permission must not leak into this one. - pump_for(harness.client, 400); + // Longer than the task's 500 ms failed-open retry (SOURCE_OPEN_RETRY_MS): a permission + // leak that merely deferred into the retry window must surface before the fresh command. + pump_for(harness.client, 700); EXPECT_EQ(fake2.count_text_containing("client-stream/start"), 0U); EXPECT_EQ(harness.listener.started, 1); From 590ac1c1c72696b27bd58a43920a566e031b014a Mon Sep 17 00:00:00 2001 From: Chris Uthe Date: Wed, 9 Sep 2026 19:37:06 -0500 Subject: [PATCH 3/3] Make the wire timestamp and quiet-window checks catch regressions The fake server now runs its clock a known 500 ms ahead, so the ordering test asserts each chunk timestamp lands in the server domain (base + offset) rather than near the raw client stamp -- dropping the compute_server_time() conversion would now fail the test. Add an explicit quiet window past the failed-open retry interval before the start command so an erroneous unsolicited open surfaces, and correct the framing comment to describe the test's pinned 25 ms config rather than the 20 ms role default. --- tests/test_source_role.cpp | 46 ++++++++++++++++++++++++++++---------- 1 file changed, 34 insertions(+), 12 deletions(-) diff --git a/tests/test_source_role.cpp b/tests/test_source_role.cpp index 881745a..6c8edd5 100644 --- a/tests/test_source_role.cpp +++ b/tests/test_source_role.cpp @@ -59,8 +59,10 @@ constexpr uint16_t TIME_GATE_TEST_PORT = 19011; constexpr uint16_t RECONNECT_TEST_PORT = 19012; constexpr uint16_t WRITE_GATE_TEST_PORT = 19013; -// Default-config wire framing, derived exactly as source_task.cpp derives it: 25 ms at -// 48 kHz stereo 16-bit -> 1200 frames x 4 bytes, behind a 1-byte type + 8-byte timestamp header. +// Wire framing for the ordering test's pinned 25 ms chunk config (the role default is 20 ms; +// the test overrides to 25 ms so this arithmetic is stable), derived exactly as +// source_task.cpp does: 25 ms at 48 kHz stereo 16-bit -> 1200 frames x 4 bytes, behind a +// 1-byte type + 8-byte timestamp header. constexpr size_t WIRE_BYTES_PER_FRAME = 4; constexpr size_t WIRE_CHUNK_PAYLOAD = 1200 * WIRE_BYTES_PER_FRAME; constexpr size_t WIRE_HEADER = 9; @@ -434,8 +436,11 @@ class FakeSourceServer { std::string data; }; - FakeSourceServer(const std::string& url, std::string server_id, bool answer_time = true) - : server_id_(std::move(server_id)), answer_time_(answer_time) { + FakeSourceServer(const std::string& url, std::string server_id, bool answer_time = true, + int64_t clock_offset_us = 0) + : server_id_(std::move(server_id)), + answer_time_(answer_time), + clock_offset_us_(clock_offset_us) { this->ws_.setUrl(url); this->ws_.disableAutomaticReconnection(); this->ws_.setOnMessageCallback([this](const ix::WebSocketMessagePtr& msg) { @@ -457,15 +462,16 @@ class FakeSourceServer { R"("connection_reason":"discovery"}})"); } else if (msg->str.find("client/time") != std::string::npos && this->answer_time_.load()) { - // Zero-offset echo: server clock == client clock, which the Kalman filter - // accepts as its first measurement. + // The server clock runs clock_offset_us_ ahead of the client's: the filter + // converges to that offset, so compute_server_time(local) == local + offset. JsonDocument doc; if (deserializeJson(doc, msg->str) == DeserializationError::Ok) { const int64_t t = doc["payload"]["client_transmitted"] | int64_t{0}; + const int64_t srv = t + this->clock_offset_us_; this->ws_.send(std::string(R"({"type":"server/time","payload":)") + R"({"client_transmitted":)" + std::to_string(t) + - R"(,"server_received":)" + std::to_string(t) + - R"(,"server_transmitted":)" + std::to_string(t) + "}}"); + R"(,"server_received":)" + std::to_string(srv) + + R"(,"server_transmitted":)" + std::to_string(srv) + "}}"); } } }); @@ -520,6 +526,7 @@ class FakeSourceServer { ix::WebSocket ws_; std::string server_id_; std::atomic answer_time_; + int64_t clock_offset_us_{0}; mutable std::mutex mutex_; std::vector events_; }; @@ -579,7 +586,11 @@ TEST(SourceWire, StreamsPcmEndToEndWithExactOrdering) { wire_config.chunk_duration_ms = 25; WireHarness harness(WIRE_TEST_PORT, wire_config); ASSERT_TRUE(harness.start()); - FakeSourceServer fake(server_url(WIRE_TEST_PORT), "source-server-a"); + // Server clock 500 ms ahead of the client: chunk timestamps must be converted into that + // domain, so a regression that shipped the raw client stamp instead would be caught below. + constexpr int64_t SERVER_OFFSET_US = 500000; + FakeSourceServer fake(server_url(WIRE_TEST_PORT), "source-server-a", /*answer_time=*/true, + SERVER_OFFSET_US); ASSERT_TRUE(harness.establish(fake)); SourceRole* source = harness.client.source(); @@ -590,6 +601,14 @@ TEST(SourceWire, StreamsPcmEndToEndWithExactOrdering) { EXPECT_FALSE(source->write_audio(frame, sizeof(frame), 0)); EXPECT_FALSE(source->is_streaming()); + // No unsolicited open: observe a quiet window past the 500 ms failed-open retry interval and + // confirm nothing streams before the command (an erroneous auto-open would surface here). + pump_for(harness.client, 700); + EXPECT_FALSE(source->is_streaming()); + EXPECT_EQ(harness.listener.started, 0); + EXPECT_EQ(fake.count_text_containing("client-stream/start"), 0U); + EXPECT_FALSE(source->write_audio(frame, sizeof(frame), 0)); + fake.send_source_command("start"); ASSERT_TRUE(pump_until( harness.client, @@ -661,13 +680,16 @@ TEST(SourceWire, StreamsPcmEndToEndWithExactOrdering) { EXPECT_LT(start_idx, binary_idx.front()); EXPECT_LT(binary_idx.back(), end_idx); - // Framing: type byte 12, big-endian first-sample capture timestamp (zero-offset time sync, - // so the server-domain value stays near the stamps we supplied), untouched PCM payload. + // Framing: type byte 12, big-endian first-sample capture timestamp converted into the + // server clock domain (client capture stamp + the 500 ms server offset the filter learned), + // untouched PCM payload. const auto& first = events[static_cast(binary_idx[0])].data; ASSERT_EQ(first.size(), WIRE_HEADER + WIRE_CHUNK_PAYLOAD); EXPECT_EQ(static_cast(first[0]), SENDSPIN_BINARY_SOURCE_AUDIO); const int64_t ts1 = read_be64(reinterpret_cast(first.data()) + 1); - EXPECT_LT(std::llabs(ts1 - base), 1000000); + // Near base + offset, not base: shipping the raw client stamp would land ~500 ms low and + // fail this bound. + EXPECT_LT(std::llabs(ts1 - (base + SERVER_OFFSET_US)), 100000); // The second chunk's anchor is exactly one chunk after the first (contiguous capture // stamps); allow slack for the time filter refining its offset between the two sends.