Fix source manager lifetime race

obs-studio-server/source/osn-source.cpp

@@ -250,8 +250,8 @@ void osn::Source::IsConfigurable(void *data, const int64_t id, const std::vector
 
 void osn::Source::GetProperties(void *data, const int64_t id, const std::vector<ipc::value> &args, std::vector<ipc::value> &rval)
 {
-	// Atomically find and acquire a strong reference under the manager lock,
-	// preventing the source from being destroyed between find() and obs_source_get_ref().
+	// Promote the manager's retained weak reference while its registration is
+	// locked, so the source remains alive for this call or is rejected as expired.
 	OBSSourceAutoRelease src = osn::Source::Manager::GetInstance().findAndRef(args[0].value_union.ui64);
 	if (!src) {
 		PRETTY_ERROR_RETURN(ErrorCode::InvalidReference, "Source reference is not valid.");

obs-studio-server/source/osn-source.hpp

@@ -20,6 +20,7 @@
 #include <ipc-server.hpp>
 #include <obs.h>
 #include <obs.hpp>
+#include <utility>
 #include "utility.hpp"
 #undef strtoll
 #include "nlohmann/json.hpp"
@@ -32,7 +33,10 @@ class Source {
 
 	protected:
 		Manager() {}
-		~Manager() {}
+		~Manager() { clear(); }
+
+	private:
+		std::map<utility::unique_id::id_t, OBSWeakSourceAutoRelease> weak_sources;
 
 	public:
 		Manager(Manager const &) = delete;
@@ -41,16 +45,58 @@ class Source {
 	public:
 		static Manager &GetInstance();
 
-		// Atomically finds the source and acquires a strong reference under the
-		// manager lock, preventing destruction between find() and obs_source_get_ref().
-		// Returns null (as OBSSourceAutoRelease) if not found or already destroyed.
+		utility::unique_id::id_t allocate(obs_source_t *source)
+		{
+			std::lock_guard<std::recursive_mutex> lock(internal_mutex);
+			const auto existingUid = utility::unique_object_manager<obs_source_t>::find(source);
+			if (existingUid != std::numeric_limits<utility::unique_id::id_t>::max())
+				return existingUid;
+
+			OBSWeakSourceAutoRelease weakSource(obs_source_get_weak_source(source));
+			const auto uid = utility::unique_object_manager<obs_source_t>::allocate(source);
+			if (uid != std::numeric_limits<utility::unique_id::id_t>::max()) {
+				try {
+					weak_sources.emplace(uid, std::move(weakSource));
+				} catch (...) {
+					utility::unique_object_manager<obs_source_t>::free(uid);
+					throw;
+				}
+			}
+			return uid;
+		}
+
+		utility::unique_id::id_t free(obs_source_t *source)
+		{
+			std::lock_guard<std::recursive_mutex> lock(internal_mutex);
+			const auto uid = utility::unique_object_manager<obs_source_t>::free(source);
+			weak_sources.erase(uid);
+			return uid;
+		}
+
+		obs_source_t *free(utility::unique_id::id_t uid)
+		{
+			std::lock_guard<std::recursive_mutex> lock(internal_mutex);
+			obs_source_t *source = utility::unique_object_manager<obs_source_t>::free(uid);
+			weak_sources.erase(uid);
+			return source;
+		}
+
+		void clear()
+		{
+			std::lock_guard<std::recursive_mutex> lock(internal_mutex);
+			object_map.clear();
+			weak_sources.clear();
+		}
+
+		// Promote the retained weak reference instead of deriving a reference from
+		// the raw source pointer, whose control block may already be destroyed.
 		OBSSourceAutoRelease findAndRef(utility::unique_id::id_t id)
 		{
 			std::lock_guard<std::recursive_mutex> lock(internal_mutex);
-			auto iter = object_map.find(id);
-			if (iter == object_map.end())
+			auto iter = weak_sources.find(id);
+			if (iter == weak_sources.end())
 				return nullptr;
-			return obs_source_get_ref(iter->second);
+			return obs_weak_source_get_source(iter->second.Get());
 		}
 	};
 

obs-studio-server/tests/test-osn-source.cpp

@@ -6,43 +6,118 @@
 #include "osn-source.hpp"
 #include <obs.h>
 #include "shared.hpp"
-#include <chrono>
+#include <condition_variable>
 #include <cstddef>
 #include <cstdint>
+#include <mutex>
 #include <string>
-#include "obs-setup.hpp"
-#include <thread>
-#include <utility>
 #include <vector>
 
-// Since we do not use C++ 20 (std::jthread), defining a scoped thread.
-struct joining_thread {
-	std::thread t;
-	explicit joining_thread(std::thread t_) : t(std::move(t_)) {}
-	joining_thread(joining_thread &&) = default;
-	joining_thread &operator=(joining_thread &&) = default;
-	joining_thread(const joining_thread &) = delete;
-	joining_thread &operator=(const joining_thread &) = delete;
-	~joining_thread()
+namespace {
+
+// Keep the test source independent of OBS plugins so its behavior depends only
+// on the source manager and libobs reference lifecycle.
+constexpr char TEST_SOURCE_ID[] = "source_manager_lifetime_test_source";
+
+const char *testSourceGetName(void *)
+{
+	return "Source Manager Lifetime Test Source";
+}
+
+void *testSourceCreate(obs_data_t *, obs_source_t *source)
+{
+	return source;
+}
+
+void testSourceDestroy(void *) {}
+
+obs_properties_t *testSourceGetProperties(void *)
+{
+	obs_properties_t *properties = obs_properties_create();
+	obs_properties_add_bool(properties, "enabled", "Enabled");
+	return properties;
+}
+
+obs_source_info makeTestSourceInfo()
+{
+	obs_source_info info{};
+	info.id = TEST_SOURCE_ID;
+	info.type = OBS_SOURCE_TYPE_INPUT;
+	info.get_name = testSourceGetName;
+	info.create = testSourceCreate;
+	info.destroy = testSourceDestroy;
+	info.get_properties = testSourceGetProperties;
+	return info;
+}
+
+// These tests only need the OBS core and its deferred-destruction queue.
+class ObsCoreSetup {
+public:
+	ObsCoreSetup() { REQUIRE(obs_startup("en-US", nullptr, nullptr)); }
+	~ObsCoreSetup()
 	{
-		if (t.joinable())
-			t.join();
+		// Do not let a queued source destruction outlive the OBS core.
+		obs_wait_for_destroy_queue();
+		obs_shutdown();
 	}
+
+	ObsCoreSetup(const ObsCoreSetup &) = delete;
+	ObsCoreSetup &operator=(const ObsCoreSetup &) = delete;
 };
 
-static bool wait_for_source_manager_size(std::size_t expectedSize)
-{
-	for (int i = 0; i < 100; i++) {
-		obs_wait_for_destroy_queue();
+// Source destruction runs on OBS_TASK_DESTROY. Blocking that queue creates the
+// precise window where the last strong reference is gone but the manager's
+// destroy callback has not removed the source registration yet.
+class DestroyQueueGate {
+public:
+	DestroyQueueGate()
+	{
+		obs_queue_task(OBS_TASK_DESTROY, waitForRelease, this, false);
+		// Wait until the gate is running so later destruction is guaranteed to
+		// queue behind it.
+		std::unique_lock lock(mutex);
+		condition.wait(lock, [this] { return entered; });
+	}
+
+	~DestroyQueueGate() { releaseAndWait(); }
 
-		if (osn::Source::Manager::GetInstance().size() == expectedSize)
-			return true;
+	void releaseAndWait()
+	{
+		{
+			std::lock_guard lock(mutex);
+			released = true;
+		}
+		condition.notify_all();
 
-		std::this_thread::sleep_for(std::chrono::milliseconds(10));
+		if (!drained) {
+			// Draining also runs the source's destroy signal, which unregisters
+			// it from Source::Manager.
+			obs_wait_for_destroy_queue();
+			drained = true;
+		}
 	}
 
-	return false;
-}
+	DestroyQueueGate(const DestroyQueueGate &) = delete;
+	DestroyQueueGate &operator=(const DestroyQueueGate &) = delete;
+
+private:
+	static void waitForRelease(void *data)
+	{
+		auto *gate = static_cast<DestroyQueueGate *>(data);
+		std::unique_lock lock(gate->mutex);
+		gate->entered = true;
+		gate->condition.notify_all();
+		gate->condition.wait(lock, [gate] { return gate->released; });
+	}
+
+	std::mutex mutex;
+	std::condition_variable condition;
+	bool entered = false;
+	bool released = false;
+	bool drained = false;
+};
+
+} // namespace
 
 TEST_CASE("Scene AddSource rejects malformed argument counts")
 {
@@ -60,59 +135,77 @@ TEST_CASE("Scene AddSource rejects malformed argument counts")
 	}
 }
 
-TEST_CASE("Run osn::source tests")
+TEST_CASE("Source manager safely promotes references during deferred destruction")
 {
-	osn::tests::ObsSetup setupOBS;
+	ObsCoreSetup setupOBS;
+	static const obs_source_info testSourceInfo = makeTestSourceInfo();
+	obs_register_source(&testSourceInfo);
+	auto &manager = osn::Source::Manager::GetInstance();
+	const auto sourceCount = manager.size();
 
-	SECTION("Get properties of browser source while releasing concurrently does not crash")
 	{
-		auto sourceCount = osn::Source::Manager::GetInstance().size();
-		const int iterations = 20;
-		std::vector<joining_thread> workers;
-		std::vector<uint8_t> releaseOk(iterations, 0);
-		std::vector<ErrorCode> getPropertiesCode(iterations, ErrorCode::Error);
-
-		for (int i = 0; i < iterations; i++) {
-			const std::string sourceName = "test-input-" + std::to_string(i);
-			std::vector<ipc::value> args = {ipc::value("browser_source"), ipc::value(sourceName)};
-			std::vector<ipc::value> response;
-
-			osn::Input::Create(nullptr, 0, args, response);
-			REQUIRE(response.size() >= 2);
-			ErrorCode error = (ErrorCode)response[0].value_union.ui64;
-			REQUIRE(error == ErrorCode::Ok);
-
-			uint64_t sourceId = response[1].value_union.ui64;
-
-			workers.push_back(joining_thread(std::thread([sourceId, i, &getPropertiesCode]() {
-				std::vector<ipc::value> propArgs = {ipc::value(sourceId)};
-				std::vector<ipc::value> propResponse;
-				osn::Source::GetProperties(nullptr, 0, propArgs, propResponse);
-				if (propResponse.size() >= 1) {
-					getPropertiesCode[i] = (ErrorCode)propResponse[0].value_union.ui64;
-				}
-			})));
-
-			workers.push_back(joining_thread(std::thread([sourceId, i, &releaseOk]() {
-				std::vector<ipc::value> propArgs = {ipc::value(sourceId)};
-				std::vector<ipc::value> propResponse;
-				osn::Source::Release(nullptr, 0, propArgs, propResponse);
-				// Capture result for checking on the main thread after join.
-				if (propResponse.size() >= 1) {
-					releaseOk[i] = ((ErrorCode)propResponse[0].value_union.ui64 == ErrorCode::Ok);
-				}
-			})));
-		}
-
-		workers.clear();
-		// Check release results on the main thread where Catch2 is safe to use.
-		for (int i = 0; i < iterations; i++) {
-			CHECK(releaseOk[i]);
-			// ErrorCode::InvalidReference is possible if the source was deleted before we could acquire the source
-			bool expectedErrorCode = getPropertiesCode[i] == ErrorCode::Ok || getPropertiesCode[i] == ErrorCode::InvalidReference;
-			CHECK(expectedErrorCode);
-		}
+		INFO("The final source release wins the race");
+		DestroyQueueGate destroyQueue;
+		OBSSourceAutoRelease source = obs_source_create_private(TEST_SOURCE_ID, "released source", nullptr);
+		REQUIRE(source != nullptr);
+
+		// Private sources do not emit the global source-create signal, so mirror
+		// production registration here. Repeated allocation must keep one ID and
+		// one retained weak reference for the source.
+		const uint64_t sourceId = manager.allocate(source);
+		osn::Source::attach_source_signals(source);
+		REQUIRE(sourceId != UINT64_MAX);
+		CHECK(manager.allocate(source) == sourceId);
+		CHECK(manager.size() == sourceCount + 1);
+
+		// The gate keeps deferred destruction from unregistering the source.
+		// Dropping the only strong reference therefore leaves an expired source
+		// ID in the manager, reproducing the original race window.
+		source = nullptr;
+
+		CHECK(manager.size() == sourceCount + 1);
+		std::vector<ipc::value> args = {ipc::value(sourceId)};
+		std::vector<ipc::value> response;
+		// Promotion of the retained weak reference must fail cleanly instead of
+		// dereferencing the stale raw pointer stored by the old implementation.
+		osn::Source::GetProperties(nullptr, 0, args, response);
+		REQUIRE(!response.empty());
+		CHECK((ErrorCode)response[0].value_union.ui64 == ErrorCode::InvalidReference);
+
+		// Let OBS finish destruction and deliver the manager's destroy callback.
+		destroyQueue.releaseAndWait();
+		CHECK(manager.size() == sourceCount);
+	}
 
-		CHECK(wait_for_source_manager_size(sourceCount)); // Check to see if all objects released.
+	{
+		INFO("The source lookup wins the race");
+		DestroyQueueGate destroyQueue;
+		OBSSourceAutoRelease source = obs_source_create_private(TEST_SOURCE_ID, "retained source", nullptr);
+		REQUIRE(source != nullptr);
+
+		const uint64_t sourceId = manager.allocate(source);
+		osn::Source::attach_source_signals(source);
+		REQUIRE(sourceId != UINT64_MAX);
+		// This time lookup wins: promotion happens while the original strong
+		// reference is still alive and must keep the source usable.
+		OBSSourceAutoRelease retainedSource = manager.findAndRef(sourceId);
+		REQUIRE(retainedSource != nullptr);
+
+		source = nullptr;
+		CHECK(std::string(obs_source_get_name(retainedSource)) == "retained source");
+		OBSSourceAutoRelease secondReference = manager.findAndRef(sourceId);
+		CHECK(secondReference != nullptr);
+
+		// Releasing every promoted reference expires the source. Its registration
+		// remains until the blocked destroy callback runs, but another promotion
+		// must already report that the source is gone.
+		secondReference = nullptr;
+		retainedSource = nullptr;
+		CHECK(manager.size() == sourceCount + 1);
+		CHECK(!manager.findAndRef(sourceId));
+
+		// Draining the queue completes destruction and removes the registration.
+		destroyQueue.releaseAndWait();
+		CHECK(manager.size() == sourceCount);
 	}
 }