diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 1b765f8e..c3a64a47 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -2,8 +2,20 @@ name: Tests # Called by top-level ci.yml on: - workflow_call: {} - workflow_dispatch: {} + workflow_call: + inputs: + platform_audio_stress: + description: Repeat the macOS x64 PlatformAudio lifecycle seam + required: false + type: boolean + default: false + workflow_dispatch: + inputs: + platform_audio_stress: + description: Repeat the macOS x64 PlatformAudio lifecycle seam + required: false + type: boolean + default: false permissions: contents: read @@ -277,6 +289,44 @@ jobs: build-release/bin/livekit_integration_tests \ --gtest_output=xml:build-release/integration-test-results.xml + - name: Stress platform audio lifecycle seam + if: matrix.name == 'macos-x64' && inputs.platform_audio_stress + timeout-minutes: 10 + shell: bash + env: + RUST_BACKTRACE: full + RUST_LOG: "livekit::rtc_engine::lk_runtime=debug,livekit::platform_audio=debug,livekit_ffi=info" + run: | + set -euo pipefail + source .token_helpers/set_data_track_test_tokens.bash + build-release/bin/livekit_integration_tests \ + --gtest_filter='PlatformAudioIntegrationTest.ReleaseAndReacquirePreserveFrameFlow' \ + --gtest_repeat=20 \ + --gtest_recreate_environments_when_repeating=1 \ + --gtest_output=xml:build-release/platform-audio-reacquire-results.xml + build-release/bin/livekit_integration_tests \ + --gtest_filter='MediaMultiStreamIntegrationTest.PublishTwoVideoAndTwoAudioTracks_SinglePeerConnection:PlatformAudioIntegrationTest.PublishPlatformAudioTrackEndToEnd' \ + --gtest_repeat=100 \ + --gtest_recreate_environments_when_repeating=1 \ + --gtest_shuffle=false \ + --gtest_output=xml:build-release/platform-audio-seam-results.xml + + - name: Collect platform audio crash diagnostics + if: failure() && matrix.name == 'macos-x64' && inputs.platform_audio_stress + shell: bash + run: | + mkdir -p build-release/platform-audio-diagnostics + cp "$HOME"/Library/Logs/DiagnosticReports/livekit_integration_tests*.ips \ + build-release/platform-audio-diagnostics/ 2>/dev/null || true + cp build-release/bin/livekit_integration_tests \ + build-release/bin/liblivekit.dylib \ + build-release/bin/liblivekit_ffi.dylib \ + build-release/platform-audio-diagnostics/ 2>/dev/null || true + xcrun dwarfdump --uuid build-release/bin/livekit_integration_tests \ + build-release/bin/liblivekit.dylib \ + build-release/bin/liblivekit_ffi.dylib \ + > build-release/platform-audio-diagnostics/image-uuids.txt 2>&1 || true + - name: Dump livekit-server log on failure if: failure() && matrix.e2e-testing shell: bash @@ -296,6 +346,8 @@ jobs: path: | build-release/unit-test-results.xml build-release/integration-test-results.xml + build-release/platform-audio-*-results.xml + build-release/platform-audio-diagnostics/ ${{ steps.livekit_server.outputs.log-path }} if-no-files-found: ignore retention-days: 7 diff --git a/client-sdk-rust b/client-sdk-rust index dad794d4..85083ada 160000 --- a/client-sdk-rust +++ b/client-sdk-rust @@ -1 +1 @@ -Subproject commit dad794d414fda9e8c1de83af1c0f190506a15f8f +Subproject commit 85083ada32313dde19b39f8cfd539ffc71e819b5 diff --git a/docs/platform-audio-instability-investigation.md b/docs/platform-audio-instability-investigation.md new file mode 100644 index 00000000..b43b1e43 --- /dev/null +++ b/docs/platform-audio-instability-investigation.md @@ -0,0 +1,163 @@ +# macOS PlatformAudio Instability — Investigation and Remediation + +*Updated 2026-07-19 after reproducing the failure matrix on macOS ARM64.* + +## 1. Outcome + +Two separate defects had been conflated: + +1. **Confirmed teardown crash.** WebRTC's macOS `CaptureWorkerThread` could + deliver recorded audio through a stale `AudioTransportImpl` while peer + transports were being destroyed. +2. **Release/reacquire frame regression.** The first attempted fix stopped the + capture worker and detached the transport callback when the final + `PlatformAudio` reference was released, but retained the platform ADM. A + later acquire reused that ADM without restoring its callback, so publication + and subscription succeeded while decoded frames never arrived. + +The corrected lifecycle separates reusable quiescing from terminal shutdown: + +- Releasing the final platform-ADM reference stops recording and playout and + waits for their worker threads, but retains the transport callback. +- Acquiring an inactive platform ADM explicitly ensures the current callback is + bound before capture can restart. +- Terminal peer-connection-factory teardown stops all audio I/O first, then + detaches platform and synthetic callbacks before destroying the factory. +- `LkRuntime` generations cannot overlap teardown and construction. Generation + logs make creation and completed teardown observable. + +## 2. Reproduction matrix + +The exact seam was run in one process and in declaration order: + +```text +MediaMultiStreamIntegrationTest.PublishTwoVideoAndTwoAudioTracks_SinglePeerConnection +PlatformAudioIntegrationTest.PublishPlatformAudioTrackEndToEnd +``` + +The decoded-frame canary was also repeated independently. + +| Rust revision | Seam result on ARM64 | Frame-flow result | +| --- | --- | --- | +| `da3ee007` (C++ main pin, no fix) | SIGSEGV on iteration 35 | Repeated frame flow passes | +| `fd48e5c2` (original fix tip) | 30 iterations pass | First run passes; later runs deterministically receive no frames | +| Corrected rebased candidate | Pending final stress totals below | Full suite and repeated frame flow pass | + +The `da3ee007` crash report records: + +- `EXC_BAD_ACCESS` / `SIGSEGV` +- faulting thread `CaptureWorkerThread` +- crash while `PublishPlatformAudioTrackEndToEnd` was starting immediately + after the MediaMultiStream test + +The original fix's full PlatformAudio suite creates one `LkRuntime` and reuses +it across all tests. Its first three subscription/lifecycle assertions pass; +`PlatformAudioFramesReachRemote` then waits 20 seconds and fails. Running that +test alone repeatedly gives the sharper signature: iteration 1 passes and +iterations 2 onward receive no frames. This proves the regression is callback +lifecycle state, not the previously suspected publisher negotiation change. + +## 3. Confirmed crash mechanism + +The pre-fix crash family includes stacks of this shape: + +```text +webrtc::AudioDeviceMac::CaptureWorkerThread() + -> AudioDeviceBuffer::DeliverRecordedData() + -> AudioTransportImpl::RecordedDataIsAvailable() +``` + +concurrent with peer transport teardown: + +```text +~JsepTransportController() + -> ~BundleManager() + -> transport / audio state destruction +``` + +Stopping capture is the synchronization boundary: WebRTC's +`StopRecording()`/`StopPlayout()` joins the platform workers. Callback +detachment must happen only after those calls because `AudioDeviceBuffer` +refuses callback changes while media is active. + +## 4. Corrected implementation + +The Rust SDK stability work is rebased onto Rust main `62359c35`. + +### Reusable PlatformAudio release + +`AdmProxy::ReleasePlatformAdm()` calls `StopPlatformAudioIO()` at refcount zero. +That helper stops and joins platform recording/playout but does **not** detach +the callback or destroy the ADM. This preserves iOS ADM reuse and allows a +later acquire on the same runtime. + +`AdmProxy::AcquirePlatformAdm()` binds the saved `audio_transport_` on every +inactive-to-active transition. Besides making reacquire explicit, this covers +Android's lazily created platform ADM, which did not exist when the factory +originally registered its callback. + +### Terminal factory shutdown + +`PeerConnectionFactory::~PeerConnectionFactory()` invokes terminal audio +shutdown on WebRTC's worker thread before releasing `peer_factory_`. +`AdmProxy::StopAudioIO()` stops both ADMs, detaches both callbacks, and clears +the stored transport only after workers have stopped. + +### Runtime generation gate + +`LkRuntime` has a teardown guard that is declared after the peer connection +factory. Its static counter reaches zero only after factory and ADM destruction +complete. `LkRuntime::instance()` waits on that gate before constructing a new +generation after the prior weak reference stops upgrading. + +The gate closes create-during-drop overlap; it does not force a still-referenced +runtime to drop. Debug logs identify runtime generation creation and completed +teardown so these cases can be distinguished without sleeps. + +## 5. Regression coverage + +`PlatformAudioIntegrationTest.ReleaseAndReacquirePreserveFrameFlow` performs two +complete publish/subscribe/decoded-frame cycles separated by destruction of the +last `PlatformAudio`, source, track, and rooms. It fails deterministically with +the callback-detaching release implementation and passes with the corrected +split lifecycle. + +The Tests workflow has an opt-in `platform_audio_stress` input. On macOS x64 it +runs: + +- 20 release/reacquire decoded-frame iterations +- 100 exact MediaMultiStream-to-PlatformAudio seam iterations + +On failure it uploads the XML results, macOS `.ips` reports, image UUIDs, and +the matching test and dylib binaries for symbolication. + +## 6. Runtime retention correction + +Closed engines could remain alive briefly in detached reconnect or close tasks. +Keeping a strong `Arc` directly in `EngineInner` therefore retained +the peer connection factory after the active-session guard had stopped audio +I/O. Forcing the singleton weak reference to reset was rejected: instrumentation +showed that it created overlapping runtime generations after bounded teardown +waits expired. + +`EngineInner` now stores only a weak runtime reference. The +`ActiveRtcSessionGuard` remains the authoritative strong owner while an RTC +session can use the factory, and `AudioCapturePauseGuard` owns a temporary +strong reference while sender teardown is in progress. Detached engine tasks +can finish without retaining a closed factory. + +In the macOS ARM64 Release stress run, every instrumented runtime generation +completed teardown before the next generation was created. All 20 +release/reacquire iterations and all 100 exact seam iterations passed without a +teardown timeout or crash. + +## 7. Validation criteria + +The remediation is considered ready when all of the following pass: + +- Full PlatformAudio integration suite. +- Repeated release/reacquire decoded-frame test. +- At least 100 exact seam iterations in Release on macOS ARM64. +- Opt-in 100-iteration macOS x64 Actions stress run. +- C++ formatting, Rust formatting, workflow validation, and the normal test + build. diff --git a/include/livekit/local_participant.h b/include/livekit/local_participant.h index 9369a914..2c1d5cf5 100644 --- a/include/livekit/local_participant.h +++ b/include/livekit/local_participant.h @@ -246,6 +246,14 @@ class LIVEKIT_API LocalParticipant : public Participant { /// prune expired @c weak_ptr entries. mutable TrackMap published_tracks_by_sid_; + /// Guards @ref published_tracks_by_sid_. The map is written from the + /// application thread (@ref publishTrack / @ref unpublishTrack) and both read + /// and pruned from the FFI callback thread (@ref trackPublications / + /// @ref findTrackPublication, reached via Room::onEvent). Without this lock + /// those concurrent accesses race and free map nodes out from under each + /// other (heap-use-after-free). Leaf lock: no other lock is taken while held. + mutable std::mutex published_tracks_mutex_; + std::unordered_map rpc_handlers_; // Shared state for RPC invocation tracking. Using shared_ptr so the state diff --git a/include/livekit/platform_audio.h b/include/livekit/platform_audio.h index 4e12343e..66755d96 100644 --- a/include/livekit/platform_audio.h +++ b/include/livekit/platform_audio.h @@ -114,8 +114,8 @@ class LIVEKIT_API PlatformAudioSource { PlatformAudioSource(FfiHandle handle, std::shared_ptr platform_audio) noexcept; - FfiHandle handle_; std::shared_ptr platform_audio_; + FfiHandle handle_; }; /// Platform audio device manager backed by WebRTC's Audio Device Module. diff --git a/src/data_track_stream.cpp b/src/data_track_stream.cpp index 155f4788..6a777377 100644 --- a/src/data_track_stream.cpp +++ b/src/data_track_stream.cpp @@ -52,22 +52,28 @@ void DataTrackStream::init(FfiHandle subscription_handle) { bool DataTrackStream::read(DataTrackFrame& out) { proto::DataTrackStreamReadResponse read_response; bool missing_read_response = false; + std::uint64_t subscription_handle = 0; { const std::scoped_lock lock(mutex_); if (closed_ || eof_) { return false; } + subscription_handle = static_cast(subscription_handle_.get()); + } - const auto subscription_handle = static_cast(subscription_handle_.get()); + // Do not hold mutex_ across sendRequest: readFrameWithTimeout may call close() + // from another thread on timeout, and close() also needs mutex_. + proto::FfiRequest req; + auto* msg = req.mutable_data_track_stream_read(); + msg->set_stream_handle(subscription_handle); + const proto::FfiResponse resp = FfiClient::instance().sendRequest(req); - // Signal the Rust side that we're ready to receive the next frame. - // The Rust SubscriptionTask uses a demand-driven protocol: it won't pull - // from the underlying stream until notified via this request. - proto::FfiRequest req; - auto* msg = req.mutable_data_track_stream_read(); - msg->set_stream_handle(subscription_handle); - const proto::FfiResponse resp = FfiClient::instance().sendRequest(req); + { + const std::scoped_lock lock(mutex_); + if (closed_ || eof_) { + return false; + } if (!resp.has_data_track_stream_read()) { missing_read_response = true; } else { diff --git a/src/local_participant.cpp b/src/local_participant.cpp index 7fda68ac..1e8d9a0f 100644 --- a/src/local_participant.cpp +++ b/src/local_participant.cpp @@ -197,7 +197,10 @@ void LocalParticipant::publishTrack(const std::shared_ptr& track, const T auto publication = std::make_shared(owned_pub); const std::string sid = publication->sid(); - published_tracks_by_sid_[sid] = std::weak_ptr(track); + { + const std::scoped_lock lock(published_tracks_mutex_); + published_tracks_by_sid_[sid] = std::weak_ptr(track); + } track->setPublication(publication); } @@ -237,6 +240,7 @@ void LocalParticipant::unpublishTrack(const std::string& track_sid) { fut.get(); + const std::scoped_lock lock(published_tracks_mutex_); if (auto it = published_tracks_by_sid_.find(track_sid); it != published_tracks_by_sid_.end()) { if (auto t = it->second.lock()) { t->setPublication(nullptr); @@ -247,6 +251,7 @@ void LocalParticipant::unpublishTrack(const std::string& track_sid) { LocalParticipant::PublicationMap LocalParticipant::trackPublications() const { PublicationMap out; + const std::scoped_lock lock(published_tracks_mutex_); for (auto it = published_tracks_by_sid_.begin(); it != published_tracks_by_sid_.end();) { auto t = it->second.lock(); if (!t) { @@ -443,6 +448,7 @@ void LocalParticipant::handleRpcMethodInvocation(uint64_t invocation_id, const s } std::shared_ptr LocalParticipant::findTrackPublication(const std::string& sid) const { + const std::scoped_lock lock(published_tracks_mutex_); auto it = published_tracks_by_sid_.find(sid); if (it == published_tracks_by_sid_.end()) { return nullptr; diff --git a/src/room.cpp b/src/room.cpp index 32dac5eb..61bf373e 100644 --- a/src/room.cpp +++ b/src/room.cpp @@ -434,7 +434,7 @@ void Room::onEvent(const FfiEvent& event) { if (event.message_case() == FfiEvent::kRpcMethodInvocation) { const auto& rpc = event.rpc_method_invocation(); - LocalParticipant* lp = nullptr; + std::shared_ptr lp; { const std::scoped_lock guard(lock_); if (!local_participant_) { @@ -446,7 +446,7 @@ void Room::onEvent(const FfiEvent& event) { // RPC is not targeted at this room's local participant; ignore. return; } - lp = local_participant_.get(); + lp = local_participant_; } // Call outside the lock to avoid deadlocks / re-entrancy issues. diff --git a/src/tests/integration/test_platform_audio.cpp b/src/tests/integration/test_platform_audio.cpp index c6e86596..706a695f 100644 --- a/src/tests/integration/test_platform_audio.cpp +++ b/src/tests/integration/test_platform_audio.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include "../common/test_common.h" @@ -74,6 +75,73 @@ class PlatformTrackCollectorDelegate : public RoomDelegate { PlatformTrackState& state_; }; +::testing::AssertionResult verifyPlatformAudioFrameFlow(const TestConfig& config, + std::unique_ptr platform_audio, + const std::string& track_name) { + RoomOptions options; + options.auto_subscribe = true; + + PlatformTrackState receiver_state; + PlatformTrackCollectorDelegate receiver_delegate(receiver_state); + auto receiver_room = std::make_unique(); + receiver_room->setDelegate(&receiver_delegate); + if (!receiver_room->connect(config.url, config.token_b, options)) { + return ::testing::AssertionFailure() << "Receiver failed to connect"; + } + + auto sender_room = std::make_unique(); + if (!sender_room->connect(config.url, config.token_a, options)) { + return ::testing::AssertionFailure() << "Sender failed to connect"; + } + + const std::string sender_identity = lockLocalParticipant(*sender_room)->identity(); + const auto source = platform_audio->createAudioSource(); + if (!source) { + return ::testing::AssertionFailure() << "PlatformAudio did not create an audio source"; + } + const auto track = LocalAudioTrack::createLocalAudioTrack(track_name, source); + if (!track) { + return ::testing::AssertionFailure() << "PlatformAudio did not create a local track"; + } + + constexpr int kRequiredFrames = 10; + constexpr auto kFrameTimeout = 20s; + std::mutex frame_mutex; + std::condition_variable frame_cv; + int received_frames = 0; + receiver_room->setOnAudioFrameCallback(sender_identity, track_name, [&](const AudioFrame& frame) { + if (frame.totalSamples() > 0) { + std::lock_guard lock(frame_mutex); + ++received_frames; + frame_cv.notify_all(); + } + }); + + TrackPublishOptions publish_options; + publish_options.source = TrackSource::SOURCE_MICROPHONE; + lockLocalParticipant(*sender_room)->publishTrack(track, publish_options); + + std::unique_lock subscription_lock(receiver_state.mutex); + const bool subscribed = receiver_state.cv.wait_for(subscription_lock, kSubscriptionTimeout, [&]() { + return receiver_state.subscribed_audio_names.count(track_name) > 0; + }); + subscription_lock.unlock(); + + std::unique_lock frame_lock(frame_mutex); + const bool frames_received = + subscribed && frame_cv.wait_for(frame_lock, kFrameTimeout, [&]() { return received_frames >= kRequiredFrames; }); + frame_lock.unlock(); + receiver_room->clearOnAudioFrameCallback(sender_identity, track_name); + + if (!subscribed) { + return ::testing::AssertionFailure() << "Receiver never subscribed to " << track_name; + } + if (!frames_received) { + return ::testing::AssertionFailure() << "Receiver did not get frames from " << track_name; + } + return ::testing::AssertionSuccess(); +} + } // namespace class PlatformAudioIntegrationTest : public LiveKitTestBase {}; @@ -319,4 +387,35 @@ TEST_F(PlatformAudioIntegrationTest, PlatformAudioFramesReachRemote) { receiver_room->clearOnAudioFrameCallback(sender_identity, track_name); } +// Releasing the final PlatformAudio reference quiesces platform capture, while +// reacquiring it on the same retained runtime must restore decoded frame flow. +// This guards the distinction between reusable release and terminal factory +// shutdown, where the AudioTransport callback is detached permanently. +TEST_F(PlatformAudioIntegrationTest, ReleaseAndReacquirePreserveFrameFlow) { + EXPECT_TRUE(config_.available) << "Missing integration configuration"; + + std::unique_ptr first_platform_audio; + try { + first_platform_audio = std::make_unique(); + } catch (const PlatformAudioError& error) { + GTEST_SKIP() << "PlatformAudio unavailable: " << error.what(); + } + + if (first_platform_audio->recordingDeviceCount() == 0) { + GTEST_SKIP() << "No recording device available; cannot capture platform audio frames"; + } + + ASSERT_TRUE(verifyPlatformAudioFrameFlow(config_, std::move(first_platform_audio), "platform-mic-before-release")); + + std::unique_ptr reacquired_platform_audio; + try { + reacquired_platform_audio = std::make_unique(); + } catch (const PlatformAudioError& error) { + FAIL() << "PlatformAudio reacquire failed: " << error.what(); + } + + ASSERT_TRUE( + verifyPlatformAudioFrameFlow(config_, std::move(reacquired_platform_audio), "platform-mic-after-reacquire")); +} + } // namespace livekit::test