Fix memory manager data races - a follow up fix

js/module.ts

@@ -1302,9 +1302,16 @@ export interface ITransition extends ISource {
 
 export interface IConfigurable {
     /**
-     * Update the settings of the source instance
-     * correlating to the values held within the
-     * object passed.
+     * Merge the supplied values into this instance's settings.
+     * Supported settings depend on the source or encoder type.
+     * For video sources, OBS applies changes during video-frame processing,
+     * so their effects may still be pending when this method returns.
+     *
+     * @param settings JSON-serializable settings to merge; omitted keys are preserved.
+     * @returns Nothing. Success does not guarantee that the changes have taken effect.
+     * @throws {TypeError} If settings cannot be converted to an object or JSON serialized.
+     * @throws {Error} If the source or encoder no longer exists, or the request to OBS fails.
+     * If communication fails after the update is sent, the settings may already have changed.
      */
     update(settings: ISettings): void;
 

obs-studio-server/source/memory-manager.cpp

@@ -59,6 +59,10 @@ struct MediaCacheManager::SourceEntry {
 	std::atomic<uint64_t> revision{1};
 	std::atomic<bool> removed{false};
 
+	// Settings guards and the graphics callback access these under the queue mutex.
+	unsigned settingsUpdatesInProgress = 0;
+	uint64_t queryAllowedFromTick = 0;
+
 	// Only the worker changes these fields, under the queue mutex.
 	uint64_t scheduledRevision = 0;
 	uint64_t reservedBytes = 0; // Includes an enable operation awaiting completion.
@@ -83,6 +87,24 @@ MediaCacheManager::~MediaCacheManager()
 	shutdown();
 }
 
+MediaCacheManager::SourceSettingsUpdate::SourceSettingsUpdate(MediaCacheManager *manager, std::shared_ptr<SourceEntry> source) noexcept
+	: m_manager(manager), m_sourceEntry(std::move(source))
+{
+}
+
+MediaCacheManager::SourceSettingsUpdate::SourceSettingsUpdate(SourceSettingsUpdate &&other) noexcept
+	: m_manager(std::exchange(other.m_manager, nullptr)), m_sourceEntry(std::move(other.m_sourceEntry))
+{
+}
+
+MediaCacheManager::SourceSettingsUpdate::~SourceSettingsUpdate()
+{
+	if (m_manager)
+		m_manager->finishSourceSettingsUpdate(m_sourceEntry);
+	// Dropping m_sourceEntry may release the last OBS source reference, so it must
+	// happen after the queue mutex is unlocked.
+}
+
 void MediaCacheManager::initialize()
 {
 	// Initialization and shutdown are serialized by the OBS API lifecycle.
@@ -146,6 +168,40 @@ void MediaCacheManager::requestCacheUpdate(obs_source_t *source)
 	m_changed.notify_all();
 }
 
+MediaCacheManager::SourceSettingsUpdate MediaCacheManager::trackSourceSettingsUpdate(obs_source_t *source)
+{
+	std::shared_ptr<SourceEntry> entry;
+	{
+		std::lock_guard lock(m_mutex);
+		auto it = m_sources.find(source);
+		if (!m_accepting || it == m_sources.end())
+			return {nullptr, {}};
+		entry = it->second;
+		++entry->revision;
+		++entry->settingsUpdatesInProgress;
+		m_notified = true;
+	}
+	m_changed.notify_all();
+	return {this, std::move(entry)};
+}
+
+void MediaCacheManager::finishSourceSettingsUpdate(const std::shared_ptr<SourceEntry> &source)
+{
+	{
+		std::lock_guard lock(m_mutex);
+		if (!m_accepting || source->removed)
+			return;
+		--source->settingsUpdatesInProgress;
+		++source->revision;
+		// Tick callbacks run before deferred source updates. The first callback
+		// after this write must skip querying; the following callback is after
+		// OBS has had a source-update phase, even if the write finished mid-frame.
+		source->queryAllowedFromTick = m_graphicsTick + 2;
+		m_notified = true;
+	}
+	m_changed.notify_all();
+}
+
 void MediaCacheManager::requestAllCacheUpdates()
 {
 	{
@@ -223,7 +279,7 @@ void MediaCacheManager::setCaching(obs_source_t *source, bool caching)
 	// Apply only our setting; never write back an old copy of the source's
 	// unrelated settings after the user has edited them.
 	OBSDataAutoRelease patch = obs_data_create();
-	// "caching" is a custom Streamlabs setting, OBS does not use it
+	// Streamlabs setting consumed by ffmpeg_source to enable media caching.
 	obs_data_set_bool(patch, "caching", caching);
 	obs_source_update(source, patch);
 }
@@ -237,10 +293,20 @@ void MediaCacheManager::graphicsTick(void *param, float)
 		if (!manager.m_accepting)
 			return;
 
-		// Take one batch per tick.
-		while (!manager.m_graphicsJobs.empty() && results.size() < JOBS_PER_TICK) {
-			results.push_back({std::move(manager.m_graphicsJobs.front())});
+		++manager.m_graphicsTick;
+		// Scan one bounded batch. Keep delayed queries queued without consuming
+		// readiness retries or blocking work for another source behind them.
+		const auto count = std::min(manager.m_graphicsJobs.size(), JOBS_PER_TICK);
+		for (size_t i = 0; i < count; ++i) {
+			auto job = std::move(manager.m_graphicsJobs.front());
 			manager.m_graphicsJobs.pop_front();
+			const auto &entry = *job.source;
+			if (job.type == JobType::QuerySource && !entry.removed && entry.revision == job.revision &&
+			    (entry.settingsUpdatesInProgress || manager.m_graphicsTick < entry.queryAllowedFromTick)) {
+				manager.m_graphicsJobs.push_back(std::move(job));
+				continue;
+			}
+			results.push_back({std::move(job)});
 		}
 	}
 	if (results.empty())

obs-studio-server/source/memory-manager.h

@@ -43,8 +43,34 @@
 // not keep that particular player alive, and our mutex does not protect it.
 // Run those queries in the graphics tick to serialize them with replacement,
 // then return copied metadata to the worker.
+// Settings become visible before the plugin applies them. External writers must
+// use trackSourceSettingsUpdate() so queries wait for a source-update phase and
+// cannot associate a new filename with the previous player's metadata.
 class MediaCacheManager {
+	struct SourceEntry;
+
 public:
+	// Keeps cache queries pending while the caller changes a source's settings.
+	class SourceSettingsUpdate {
+	public:
+		// Transfers responsibility for finishing the update; the moved-from guard
+		// is inactive. The manager must outlive the guard.
+		SourceSettingsUpdate(SourceSettingsUpdate &&other) noexcept;
+		// Finishes tracking and schedules evaluation after OBS can apply the update.
+		// Safe after source unregistration or manager shutdown, but must run before
+		// obs_shutdown(): the guard still retains its source reference.
+		~SourceSettingsUpdate();
+		SourceSettingsUpdate(const SourceSettingsUpdate &) = delete;
+		SourceSettingsUpdate &operator=(const SourceSettingsUpdate &) = delete;
+		SourceSettingsUpdate &operator=(SourceSettingsUpdate &&) = delete;
+
+	private:
+		friend class MediaCacheManager;
+		SourceSettingsUpdate(MediaCacheManager *manager, std::shared_ptr<SourceEntry> source) noexcept;
+		MediaCacheManager *m_manager;
+		std::shared_ptr<SourceEntry> m_sourceEntry;
+	};
+
 	// Returns a non-owning reference to the process-wide singleton. Access is
 	// thread-safe, but does not initialize OBS or start the cache worker.
 	static MediaCacheManager &GetInstance();
@@ -75,7 +101,22 @@ class MediaCacheManager {
 	// obs_source_remove() or explicitly disable the source's cache setting.
 	void unregisterSource(obs_source_t *source);
 
-	// Requests reevaluation after a registered source's settings or activity change.
+	// Invalidates older work before external code changes a source's live settings.
+	// Keep the returned guard alive through obs_source_update(), including any
+	// property callbacks that mutate those settings. Retains the original
+	// registration so finishing this guard cannot affect a later registration
+	// of the same source.
+	// Nested updates are supported; queries wait until all guards for this source
+	// finish and OBS has had a source-update phase. Does not wait for executing
+	// queries; their results are invalidated. Does not serialize external settings
+	// writers.
+	// No queue lock is held across the caller's OBS operations. Null/unregistered
+	// sources and calls while stopped or stopping return an inactive guard.
+	// The manager and OBS runtime must outlive the guard.
+	[[nodiscard]] SourceSettingsUpdate trackSourceSettingsUpdate(obs_source_t *source);
+
+	// Requests reevaluation after a registered source's activity change.
+	// Use trackSourceSettingsUpdate() around settings writes instead.
 	// Safe from OBS callbacks; repeated requests are coalesced, and this call does
 	// not wait for media queries or cache-setting changes to complete.
 	// Null/unregistered sources and calls while stopped or stopping are ignored.
@@ -87,7 +128,9 @@ class MediaCacheManager {
 	void requestAllCacheUpdates();
 
 	// Stops accepting requests, removes the tick callback, cancels queued work,
-	// joins the worker and drops all retained source references.
+	// joins the worker and drops its queued and tracked source references.
+	// Outstanding SourceSettingsUpdate guards retain their own references and must
+	// finish before obs_shutdown(), while this manager is still alive.
 	// Call before obs_shutdown(), from the serialized OBS lifecycle, on neither
 	// the graphics thread nor this manager's worker. Do not overlap another
 	// shutdown() or initialize(). Waits for any executing tick callback and worker,
@@ -100,7 +143,6 @@ class MediaCacheManager {
 private:
 	friend class MediaCacheManagerTestAccess;
 	using Clock = std::chrono::steady_clock;
-	struct SourceEntry;
 	struct Snapshot {
 		std::string file;
 		bool eligible = false;
@@ -134,6 +176,7 @@ class MediaCacheManager {
 	void complete(Completion &result);
 	void queueCacheSettingUpdate(const std::shared_ptr<SourceEntry> &source, uint64_t revision, bool targetCachingEnabled);
 	void releaseBudget(SourceEntry &source);
+	void finishSourceSettingsUpdate(const std::shared_ptr<SourceEntry> &source);
 
 	// Protects only queues and bookkeeping. Never held during an OBS call,
 	// source release, wait for graphics, or thread join.
@@ -148,6 +191,8 @@ class MediaCacheManager {
 	bool m_stopping = false;
 	bool m_notified = false;
 	bool m_rebalance = false;
+	// Counts every graphics callback, including empty ticks; survives video reset.
+	uint64_t m_graphicsTick = 0;
 	uint64_t m_reservedCacheBytes = 0;
 	uint64_t m_cacheBudgetBytes;
 	// Injectable clock for deterministic retry tests.

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

@@ -262,6 +262,8 @@ void osn::Source::GetProperties(void *data, const int64_t id, const std::vector<
 
 	rval.push_back(ipc::value((uint64_t)ErrorCode::Ok));
 
+	// Property callbacks may change live settings before obs_source_update().
+	auto settingsUpdate = MediaCacheManager::GetInstance().trackSourceSettingsUpdate(src);
 	obs_properties_t *prp = obs_source_properties(src);
 	obs_data *settings = obs_source_get_settings(src);
 
@@ -354,8 +356,10 @@ void osn::Source::Update(void *data, const int64_t id, const std::vector<ipc::va
 		}
 	}
 
-	obs_source_update(src, sets);
-	MediaCacheManager::GetInstance().requestCacheUpdate(src);
+	{
+		auto settingsUpdate = MediaCacheManager::GetInstance().trackSourceSettingsUpdate(src);
+		obs_source_update(src, sets);
+	}
 	obs_data_release(sets);
 
 	obs_data_t *updatedSettings = obs_source_get_settings(src);