fix(native): bound cross-platform lifecycle ownership

shared/apple/AtmosProbe/AtmosProbePlugin.swift

@@ -140,8 +140,15 @@ public class AtmosProbePlugin: NSObject, FlutterPlugin {
       }.joined(separator: ", ")
     }
     if let loader = loader {
-      out["fedBytes"] = loader.bytesReceived
-      out["loaderRequests"] = loader.requestLog
+      let snapshot = loader.statusSnapshot()
+      out["fedBytes"] = snapshot.bytesReceived
+      out["loaderRequests"] = snapshot.requestLog
+      out["loaderRetainedBytes"] = snapshot.retainedBytes
+      out["loaderPendingRequests"] = snapshot.pendingRequestCount
+      out["loaderMaximumBytes"] = snapshot.maximumBufferedBytes
+      if let errorCode = snapshot.errorCode {
+        out["loaderError"] = errorCode
+      }
     }
     return out
   }
@@ -157,59 +164,205 @@ public class AtmosProbePlugin: NSObject, FlutterPlugin {
 
 /// Streams an HTTP source into memory and serves it to AVPlayer through an
 /// AVAssetResourceLoader on a custom scheme, mirroring the mpv sink's model.
-private final class RawEc3Loader: NSObject, AVAssetResourceLoaderDelegate, URLSessionDataDelegate {
+final class RawEc3Loader: NSObject, AVAssetResourceLoaderDelegate, URLSessionDataDelegate {
+  static let defaultMaximumBufferedBytes = 64 * 1_024 * 1_024
+  private static let maximumPendingRequests = 256
+
+  struct StatusSnapshot {
+    let bytesReceived: Int
+    let requestLog: String
+    let retainedBytes: Int
+    let pendingRequestCount: Int
+    let maximumBufferedBytes: Int
+    let errorCode: String?
+    let isFinished: Bool
+  }
+
+  private enum State {
+    case active
+    case finished
+    case failed(code: String, error: Error)
+    case cancelled
+
+    var terminalError: Error? {
+      switch self {
+      case .failed(_, let error):
+        return error
+      case .cancelled:
+        return URLError(.cancelled)
+      case .active, .finished:
+        return nil
+      }
+    }
+
+    var errorCode: String? {
+      if case .failed(let code, _) = self { return code }
+      return nil
+    }
+
+    var isFinished: Bool {
+      if case .finished = self { return true }
+      return false
+    }
+  }
+
   let asset: AVURLAsset
+  let maximumBufferedBytes: Int
   private let source: URL
   private let finiteLength: Bool
+  private let sessionConfiguration: URLSessionConfiguration
   private let queue = DispatchQueue(label: "plezy.atmos.probe.loader")
-  private var session: URLSession!
+  private var terminalHandlerForTesting: (() -> Void)?
+  private let queueKey = DispatchSpecificKey<Void>()
+  private var session: URLSession?
   private var buffer = Data()
   private var contentLength: Int64 = -1
-  private var finished = false
+  private var state: State = .active
   private var pending: [AVAssetResourceLoadingRequest] = []
-  private(set) var bytesReceived: Int = 0
-  private(set) var requestLog: String = ""
+  private var bytesReceived = 0
+  private var requestLog = ""
+  private var hasBegun = false
 
-  init(source: URL, finiteLength: Bool) {
+  init(
+    source: URL,
+    finiteLength: Bool,
+    maximumBufferedBytes: Int = RawEc3Loader.defaultMaximumBufferedBytes,
+    sessionConfiguration: URLSessionConfiguration = .default,
+    terminalHandlerForTesting: (() -> Void)? = nil
+  ) {
+    precondition(maximumBufferedBytes > 0)
     self.source = source
     self.finiteLength = finiteLength
+    self.maximumBufferedBytes = maximumBufferedBytes
+    self.sessionConfiguration = sessionConfiguration
+    self.terminalHandlerForTesting = terminalHandlerForTesting
     self.asset = AVURLAsset(url: URL(string: "plezy-ec3-probe://stream/audio.ec3")!)
     super.init()
+    queue.setSpecific(key: queueKey, value: ())
     asset.resourceLoader.setDelegate(self, queue: queue)
   }
 
   func begin() {
-    let config = URLSessionConfiguration.default
-    session = URLSession(configuration: config, delegate: self, delegateQueue: nil)
-    session.dataTask(with: source).resume()
+    queue.async { [weak self] in
+      guard let self, !self.hasBegun else { return }
+      guard case .active = self.state else { return }
+      self.hasBegun = true
+      let session = URLSession(
+        configuration: self.sessionConfiguration,
+        delegate: self,
+        delegateQueue: nil
+      )
+      self.session = session
+      session.dataTask(with: self.source).resume()
+    }
   }
 
   func cancel() {
-    session?.invalidateAndCancel()
-    queue.async {
-      for request in self.pending where !request.isFinished {
-        request.finishLoading(with: NSError(domain: NSURLErrorDomain, code: NSURLErrorCancelled))
-      }
-      self.pending.removeAll()
+    queue.async { [weak self] in
+      self?.cancelOnQueue()
     }
   }
 
+  /// Called only from the method-channel/main path, never from the loader queue.
+  func statusSnapshot() -> StatusSnapshot {
+    syncOnQueue {
+      StatusSnapshot(
+        bytesReceived: bytesReceived,
+        requestLog: requestLog,
+        retainedBytes: buffer.count,
+        pendingRequestCount: pending.count,
+        maximumBufferedBytes: maximumBufferedBytes,
+        errorCode: state.errorCode,
+        isFinished: state.isFinished
+      )
+    }
+  }
+
+  private func syncOnQueue<T>(_ body: () -> T) -> T {
+    if DispatchQueue.getSpecific(key: queueKey) != nil {
+      return body()
+    }
+    return queue.sync(execute: body)
+  }
+
+  private func notifyTerminalForTesting() {
+    let handler = terminalHandlerForTesting
+    terminalHandlerForTesting = nil
+    handler?()
+  }
+
+  private func cancelOnQueue() {
+    guard case .active = state else {
+      if case .finished = state {
+        state = .cancelled
+        finishPending(with: URLError(.cancelled))
+        releaseRetainedBytes()
+      }
+      return
+    }
+    state = .cancelled
+    session?.invalidateAndCancel()
+    session = nil
+    finishPending(with: URLError(.cancelled))
+    releaseRetainedBytes()
+    notifyTerminalForTesting()
+  }
+
+  private func failOnQueue(code: String, error: Error) {
+    guard case .active = state else { return }
+    state = .failed(code: code, error: error)
+    session?.invalidateAndCancel()
+    session = nil
+    finishPending(with: error)
+    releaseRetainedBytes()
+    notifyTerminalForTesting()
+  }
+
+  private func finishPending(with error: Error) {
+    let requests = pending
+    pending.removeAll(keepingCapacity: false)
+    for request in requests where !request.isFinished {
+      request.finishLoading(with: error)
+    }
+  }
+
+  private func releaseRetainedBytes() {
+    buffer.removeAll(keepingCapacity: false)
+  }
+
   // MARK: URLSessionDataDelegate (background queue -> hop to `queue`)
 
   func urlSession(
-    _ session: URLSession, dataTask: URLSessionDataTask,
+    _ session: URLSession,
+    dataTask: URLSessionDataTask,
     didReceive response: URLResponse,
     completionHandler: @escaping (URLSession.ResponseDisposition) -> Void
   ) {
-    queue.async {
+    queue.async { [weak self] in
+      guard let self, case .active = self.state else { return }
       self.contentLength = response.expectedContentLength
-      self.serve()
+      if response.expectedContentLength > Int64(self.maximumBufferedBytes) {
+        self.failOnQueue(
+          code: "response_too_large",
+          error: URLError(.dataLengthExceedsMaximum)
+        )
+      } else {
+        self.serve()
+      }
     }
     completionHandler(.allow)
   }
 
   func urlSession(_ session: URLSession, dataTask: URLSessionDataTask, didReceive data: Data) {
-    queue.async {
+    queue.async { [weak self] in
+      guard let self, case .active = self.state else { return }
+      guard data.count <= self.maximumBufferedBytes - self.buffer.count else {
+        self.failOnQueue(
+          code: "response_too_large",
+          error: URLError(.dataLengthExceedsMaximum)
+        )
+        return
+      }
       self.buffer.append(data)
       self.bytesReceived += data.count
       self.serve()
@@ -217,9 +370,19 @@ private final class RawEc3Loader: NSObject, AVAssetResourceLoaderDelegate, URLSe
   }
 
   func urlSession(_ session: URLSession, task: URLSessionTask, didCompleteWithError error: Error?) {
-    queue.async {
-      self.finished = true
-      self.serve()
+    queue.async { [weak self] in
+      guard let self, case .active = self.state else { return }
+      self.session = nil
+      if let error {
+        self.state = .failed(code: "network_error", error: error)
+        self.finishPending(with: error)
+        self.releaseRetainedBytes()
+      } else {
+        self.state = .finished
+        self.serve()
+      }
+      self.notifyTerminalForTesting()
+      session.finishTasksAndInvalidate()
     }
   }
 
@@ -229,6 +392,14 @@ private final class RawEc3Loader: NSObject, AVAssetResourceLoaderDelegate, URLSe
     _ resourceLoader: AVAssetResourceLoader,
     shouldWaitForLoadingOfRequestedResource loadingRequest: AVAssetResourceLoadingRequest
   ) -> Bool {
+    if let terminalError = state.terminalError {
+      loadingRequest.finishLoading(with: terminalError)
+      return true
+    }
+    guard pending.count < Self.maximumPendingRequests else {
+      loadingRequest.finishLoading(with: URLError(.resourceUnavailable))
+      return true
+    }
     if let dataRequest = loadingRequest.dataRequest {
       requestLog += "[\(dataRequest.requestedOffset)+\(dataRequest.requestedLength)]"
       if requestLog.count > 300 { requestLog = String(requestLog.suffix(300)) }
@@ -246,11 +417,21 @@ private final class RawEc3Loader: NSObject, AVAssetResourceLoaderDelegate, URLSe
   }
 
   private func serve() {
-    // content info: unbounded mirrors the mpv sink; finite passes the real
-    // length through once the HTTP response reveals it
+    let isFinished: Bool
+    switch state {
+    case .active:
+      isFinished = false
+    case .finished:
+      isFinished = true
+    case .failed, .cancelled:
+      return
+    }
+
+    // The logical 1-TiB length remains the raw probe contract. Retained bytes
+    // are independently bounded by maximumBufferedBytes.
     let knownLength: Int64? =
       finiteLength
-      ? (contentLength >= 0 ? contentLength : (finished ? Int64(buffer.count) : nil))
+      ? (contentLength >= 0 ? contentLength : (isFinished ? Int64(buffer.count) : nil))
       : Int64(1) << 40
 
     var index = 0
@@ -259,7 +440,7 @@ private final class RawEc3Loader: NSObject, AVAssetResourceLoaderDelegate, URLSe
       if let info = request.contentInformationRequest {
         guard let length = knownLength else {
           index += 1
-          continue  // wait for the HTTP response before answering
+          continue
         }
         info.contentType = "public.enhanced-ac3-audio"
         info.contentLength = length
@@ -274,13 +455,33 @@ private final class RawEc3Loader: NSObject, AVAssetResourceLoaderDelegate, URLSe
         index += 1
         continue
       }
-      let offset = dataRequest.currentOffset
-      let end = dataRequest.requestedOffset + Int64(dataRequest.requestedLength)
-      if offset < Int64(buffer.count) {
-        let chunkEnd = min(Int64(buffer.count), end)
-        dataRequest.respond(with: buffer.subdata(in: Int(offset)..<Int(chunkEnd)))
+
+      let requestedOffset = dataRequest.requestedOffset
+      let currentOffset = dataRequest.currentOffset
+      let requestedLength = Int64(dataRequest.requestedLength)
+      let (end, overflow) = requestedOffset.addingReportingOverflow(requestedLength)
+      guard requestedOffset >= 0, currentOffset >= requestedOffset, requestedLength >= 0, !overflow else {
+        request.finishLoading(with: URLError(.badServerResponse))
+        pending.remove(at: index)
+        continue
       }
-      if dataRequest.currentOffset >= end || (finished && dataRequest.currentOffset >= Int64(buffer.count)) {
+
+      let bufferedCount = Int64(buffer.count)
+      if currentOffset < bufferedCount {
+        let chunkEnd = min(bufferedCount, end)
+        guard currentOffset <= chunkEnd,
+          let start = Int(exactly: currentOffset),
+          let finish = Int(exactly: chunkEnd)
+        else {
+          request.finishLoading(with: URLError(.badServerResponse))
+          pending.remove(at: index)
+          continue
+        }
+        dataRequest.respond(with: buffer.subdata(in: start..<finish))
+      }
+      if dataRequest.currentOffset >= end
+        || (isFinished && dataRequest.currentOffset >= bufferedCount)
+      {
         request.finishLoading()
         pending.remove(at: index)
         continue

shared/apple/MpvPlayer/MpvPlayerCoreBase.swift

@@ -80,6 +80,27 @@ struct ServerDisplayCriteria {
   }
 }
 
+final class MpvWakeupCallbackContext {
+  private let lock = NSLock()
+  private weak var core: MpvPlayerCoreBase?
+
+  init(core: MpvPlayerCoreBase) {
+    self.core = core
+  }
+
+  func dispatchWakeup() {
+    lock.lock()
+    defer { lock.unlock() }
+    core?.readEventsFromCallback()
+  }
+
+  func detach() {
+    lock.lock()
+    core = nil
+    lock.unlock()
+  }
+}
+
 class MpvPlayerCoreBase: NSObject {
   weak var delegate: MpvPlayerDelegate?
 
@@ -179,6 +200,11 @@ class MpvPlayerCoreBase: NSObject {
 
   private let cacheLock = NSLock()
   private var cachedPaused = true
+  private var confirmedPaused = true
+  private var resolvedPauseGeneration: UInt64 = 0
+  private var pauseIntentGeneration: UInt64 = 0
+  private var pauseObservationRevision: UInt64 = 0
+  private var pendingPauseIntents: [UInt64: Bool] = [:]
   private var cachedDuration = 0.0
   private var cachedTimePos = 0.0
   private var cachedWidth = 0.0
@@ -449,15 +475,18 @@ class MpvPlayerCoreBase: NSObject {
       return false
     }
 
-    // mpv stores this context without retaining it. Retain manually so the
-    // Swift core cannot deallocate while mpv can still fire wakeup callbacks.
-    let wakeupContext = Unmanaged.passRetained(self).toOpaque()
+    // mpv stores this context without retaining it. The core owns the context,
+    // while the context keeps only a weak callback target; disposal atomically
+    // detaches it before removing the C callback. A late callback can therefore
+    // neither retain/dereference a dead core nor enqueue work against a replacement.
+    let callbackOwner = MpvWakeupCallbackContext(core: self)
+    let wakeupContext = Unmanaged.passRetained(callbackOwner).toOpaque()
 
     lifecycleLock.lock()
     guard !lifecycleState.isTerminal, lifecycleState.mpv == nil else {
       lifecycleLock.unlock()
       mpv_terminate_destroy(mpv)
-      Unmanaged<MpvPlayerCoreBase>.fromOpaque(wakeupContext).release()
+      Unmanaged<MpvWakeupCallbackContext>.fromOpaque(wakeupContext).release()
       return false
     }
     lifecycleState.mpv = mpv
@@ -466,8 +495,9 @@ class MpvPlayerCoreBase: NSObject {
       mpv,
       { context in
         guard let context else { return }
-        let core = Unmanaged<MpvPlayerCoreBase>.fromOpaque(context).takeUnretainedValue()
-        core.readEvents()
+        Unmanaged<MpvWakeupCallbackContext>.fromOpaque(context)
+          .takeUnretainedValue()
+          .dispatchWakeup()
       },
       wakeupContext
     )
@@ -490,6 +520,10 @@ class MpvPlayerCoreBase: NSObject {
     value: String,
     completion: @escaping (Result<Void, Error>) -> Void
   ) {
+    guard isLifecycleActive else {
+      completeOnMain { completion(.failure(self.lifecycleUnavailableError())) }
+      return
+    }
     #if targetEnvironment(simulator)
       if name == "hwdec" {
         if value != "no" {
@@ -502,7 +536,7 @@ class MpvPlayerCoreBase: NSObject {
 
     if isManagedRendererProperty(name) {
       print("[MpvPlayerCore] Ignoring managed renderer property: \(name)=\(value)")
-      completion(.success(()))
+      completeOnMain { completion(.success(())) }
       return
     }
 
@@ -510,10 +544,9 @@ class MpvPlayerCoreBase: NSObject {
 
     if name == "pause" {
       let paused = parseBoolProperty(value)
+      let intent = beginCachedPauseIntent(paused)
       setRawStringPropertyAsync(name, value: value) { [weak self] result in
-        if case .success = result {
-          self?.setCachedPaused(paused)
-        }
+        self?.finishCachedPauseIntent(intent, result: result)
         completion(result)
       }
       return
@@ -527,13 +560,13 @@ class MpvPlayerCoreBase: NSObject {
 
     if name == "dv-conversion-mode" {
       setDvConversionMode(value)
-      completion(.success(()))
+      completeOnMain { completion(.success(())) }
       return
     }
 
     if name == "dv-conversion-log" {
       setDvConversionLogEnabled(parseBoolProperty(value))
-      completion(.success(()))
+      completeOnMain { completion(.success(())) }
       return
     }
 
@@ -614,22 +647,12 @@ class MpvPlayerCoreBase: NSObject {
     value: Int64,
     completion: @escaping (Result<Void, Error>) -> Void
   ) {
-    var requestId: UInt64?
     var propertyValue = value
-    guard
-      let status = withActiveMpv({ mpv in
-        let id = registerRequest(.void(completion))
-        requestId = id
-        return name.withCString { namePointer in
-          mpv_set_property_async(mpv, id, namePointer, MPV_FORMAT_INT64, &propertyValue)
-        }
-      }),
-      let requestId
-    else {
-      completion(.failure(lifecycleUnavailableError()))
-      return
+    submitAsyncRequest(.void(completion)) { mpv, requestId in
+      name.withCString { namePointer in
+        mpv_set_property_async(mpv, requestId, namePointer, MPV_FORMAT_INT64, &propertyValue)
+      }
     }
-    completeRequestIfSubmissionFailed(requestId: requestId, status: status)
   }
 
   func setHDREnabled(_ enabled: Bool, completion: ((Result<Void, Error>) -> Void)? = nil) {
@@ -677,31 +700,25 @@ class MpvPlayerCoreBase: NSObject {
   }
 
   func getPropertyAsync(_ name: String, completion: @escaping (Result<String?, Error>) -> Void) {
+    guard isLifecycleActive else {
+      completeOnMain { completion(.failure(self.lifecycleUnavailableError())) }
+      return
+    }
     if name == "dv-conversion-mode" {
-      completion(.success(getDvConversionMode()))
+      completeOnMain { completion(.success(self.getDvConversionMode())) }
       return
     }
 
     if name == "dv-conversion-log" {
-      completion(.success(getDvConversionLogEnabled() ? "yes" : "no"))
+      completeOnMain { completion(.success(self.getDvConversionLogEnabled() ? "yes" : "no")) }
       return
     }
 
-    var requestId: UInt64?
-    guard
-      let status = withActiveMpv({ mpv in
-        let id = registerRequest(.getProperty(completion))
-        requestId = id
-        return name.withCString { namePointer in
-          mpv_get_property_async(mpv, id, namePointer, MPV_FORMAT_STRING)
-        }
-      }),
-      let requestId
-    else {
-      completion(.failure(lifecycleUnavailableError()))
-      return
+    submitAsyncRequest(.getProperty(completion)) { mpv, requestId in
+      name.withCString { namePointer in
+        mpv_get_property_async(mpv, requestId, namePointer, MPV_FORMAT_STRING)
+      }
     }
-    completeRequestIfSubmissionFailed(requestId: requestId, status: status)
   }
 
   func observeProperty(_ name: String, format: String) {
@@ -730,32 +747,23 @@ class MpvPlayerCoreBase: NSObject {
 
   func commandAsync(_ args: [String], completion: @escaping (Result<Void, Error>) -> Void) {
     guard !args.isEmpty else {
-      completion(.success(()))
+      completeOnMain { completion(.success(())) }
       return
     }
 
     var cargs: [UnsafeMutablePointer<CChar>?] = args.map { strdup($0) }
     cargs.append(nil)
 
-    var requestId: UInt64?
-    let status = withActiveMpv { mpv in
-      let id = registerRequest(.void(completion))
-      requestId = id
-      return cargs.withUnsafeBufferPointer { buffer in
+    submitAsyncRequest(.void(completion)) { mpv, requestId in
+      cargs.withUnsafeBufferPointer { buffer in
         var constPointers = buffer.map { UnsafePointer($0) }
-        return mpv_command_async(mpv, id, &constPointers)
+        return mpv_command_async(mpv, requestId, &constPointers)
       }
     }
 
     for pointer in cargs {
       free(pointer)
     }
-
-    guard let status, let requestId else {
-      completion(.failure(lifecycleUnavailableError()))
-      return
-    }
-    completeRequestIfSubmissionFailed(requestId: requestId, status: status)
   }
 
   private func setRawStringPropertyAsync(
@@ -763,24 +771,14 @@ class MpvPlayerCoreBase: NSObject {
     value: String,
     completion: @escaping (Result<Void, Error>) -> Void
   ) {
-    var requestId: UInt64?
-    guard
-      let status = withActiveMpv({ mpv in
-        let id = registerRequest(.void(completion))
-        requestId = id
-        return name.withCString { namePointer in
-          value.withCString { valuePointer in
-            var propertyValue: UnsafePointer<CChar>? = valuePointer
-            return mpv_set_property_async(mpv, id, namePointer, MPV_FORMAT_STRING, &propertyValue)
-          }
+    submitAsyncRequest(.void(completion)) { mpv, requestId in
+      name.withCString { namePointer in
+        value.withCString { valuePointer in
+          var propertyValue: UnsafePointer<CChar>? = valuePointer
+          return mpv_set_property_async(mpv, requestId, namePointer, MPV_FORMAT_STRING, &propertyValue)
         }
-      }),
-      let requestId
-    else {
-      completion(.failure(lifecycleUnavailableError()))
-      return
+      }
     }
-    completeRequestIfSubmissionFailed(requestId: requestId, status: status)
   }
 
   var isPaused: Bool {
@@ -829,6 +827,11 @@ class MpvPlayerCoreBase: NSObject {
     lifecycleState.mpv = nil
     lifecycleState.wakeupCallbackContext = nil
     lifecycleLock.unlock()
+    if let callbackContext {
+      Unmanaged<MpvWakeupCallbackContext>.fromOpaque(callbackContext)
+        .takeUnretainedValue()
+        .detach()
+    }
 
     let destroy = {
       if let mpvHandle {
@@ -836,7 +839,7 @@ class MpvPlayerCoreBase: NSObject {
         mpv_terminate_destroy(mpvHandle)
       }
       if let callbackContext {
-        Unmanaged<MpvPlayerCoreBase>.fromOpaque(callbackContext).release()
+        Unmanaged<MpvWakeupCallbackContext>.fromOpaque(callbackContext).release()
       }
     }
 
@@ -927,6 +930,14 @@ class MpvPlayerCoreBase: NSObject {
     #endif
   }
 
+  private func completeOnMain(_ completion: @escaping () -> Void) {
+    if Thread.isMainThread {
+      completion()
+    } else {
+      DispatchQueue.main.async(execute: completion)
+    }
+  }
+
   private func cancelPendingRequests() {
     pendingRequestsLock.lock()
     let pending = pendingRequests
@@ -982,6 +993,33 @@ class MpvPlayerCoreBase: NSObject {
     )
   }
 
+  private func submitAsyncRequest(
+    _ request: PendingRequest,
+    submission: (OpaquePointer, UInt64) -> CInt
+  ) {
+    var requestId: UInt64?
+    guard
+      let status = withActiveMpv({ mpv in
+        let id = registerRequest(request)
+        requestId = id
+        return submission(mpv, id)
+      }),
+      let requestId
+    else {
+      let error = lifecycleUnavailableError()
+      completeOnMain {
+        switch request {
+        case .void(let completion):
+          completion(.failure(error))
+        case .getProperty(let completion):
+          completion(.failure(error))
+        }
+      }
+      return
+    }
+    completeRequestIfSubmissionFailed(requestId: requestId, status: status)
+  }
+
   private func completeRequestIfSubmissionFailed(requestId: UInt64, status: CInt) {
     guard status < 0, let request = takeRequest(requestId) else { return }
     let error = mpvError(status)
@@ -996,29 +1034,27 @@ class MpvPlayerCoreBase: NSObject {
   }
 
   private func completeVoidRequest(requestId: UInt64, error status: CInt) {
-    guard let request = takeRequest(requestId) else { return }
+    guard case .void(let completion) = takeRequest(requestId) else { return }
+    let result: Result<Void, Error> =
+      status < 0 ? .failure(mpvError(status)) : .success(())
     DispatchQueue.main.async {
-      switch request {
-      case .void(let completion):
-        if status < 0 {
-          completion(.failure(self.mpvError(status)))
-        } else {
-          completion(.success(()))
-        }
-      case .getProperty:
-        break
-      }
+      completion(result)
     }
   }
 
   private func completeGetPropertyRequest(_ event: mpv_event) {
-    guard let request = takeRequest(event.reply_userdata) else { return }
-    guard case .getProperty(let completion) = request else { return }
+    guard case .getProperty(let completion) = takeRequest(event.reply_userdata) else { return }
+
+    if event.error < 0 {
+      let error = mpvError(event.error)
+      DispatchQueue.main.async {
+        completion(.failure(error))
+      }
+      return
+    }
 
     var value: String?
-    if event.error >= 0,
-      let propertyPointer = event.data?.assumingMemoryBound(to: mpv_event_property.self)
-    {
+    if let propertyPointer = event.data?.assumingMemoryBound(to: mpv_event_property.self) {
       let property = propertyPointer.pointee
       if property.format == MPV_FORMAT_STRING, let data = property.data {
         let cstring = data.assumingMemoryBound(to: UnsafePointer<CChar>?.self).pointee
@@ -1031,7 +1067,7 @@ class MpvPlayerCoreBase: NSObject {
     }
   }
 
-  private func readEvents() {
+  fileprivate func readEventsFromCallback() {
     queue.async { [weak self] in
       guard let self else { return }
 
@@ -1048,14 +1084,14 @@ class MpvPlayerCoreBase: NSObject {
     }
   }
 
-  private func dispatchDelegateEvent(name: String, data: [String: Any]?) {
+  func dispatchDelegateEvent(name: String, data: [String: Any]?) {
     DispatchQueue.main.async { [weak self] in
       guard let self, self.isLifecycleActive else { return }
       self.delegate?.onEvent(name: name, data: data)
     }
   }
 
-  private func dispatchDelegateProperty(name: String, value: Any?) {
+  func dispatchDelegateProperty(name: String, value: Any?) {
     DispatchQueue.main.async { [weak self] in
       guard let self, self.isLifecycleActive else { return }
       self.delegate?.onPropertyChange(name: name, value: value)
@@ -1219,7 +1255,11 @@ class MpvPlayerCoreBase: NSObject {
 
     switch name {
     case "pause":
-      if let paused = value as? Bool { cachedPaused = paused }
+      if let paused = value as? Bool {
+        pauseObservationRevision &+= 1
+        confirmedPaused = paused
+        if pendingPauseIntents.isEmpty { cachedPaused = paused }
+      }
     case "duration":
       if let duration = value as? Double { cachedDuration = duration }
     case "time-pos":
@@ -1233,16 +1273,149 @@ class MpvPlayerCoreBase: NSObject {
     }
   }
 
-  private func setCachedPaused(_ paused: Bool) {
+  #if DEBUG
+    func observeCachedPauseForTesting(_ paused: Bool) {
+      updateCachedProperty(name: "pause", value: paused)
+    }
+  #endif
+
+  private func retirePauseIntents(through generation: UInt64) {
+    pendingPauseIntents = pendingPauseIntents.filter { $0.key > generation }
+  }
+
+  func beginCachedPauseIntent(
+    _ paused: Bool
+  ) -> (generation: UInt64, observationRevision: UInt64, paused: Bool) {
     cacheLock.lock()
+    pauseIntentGeneration &+= 1
+    let intent = (
+      generation: pauseIntentGeneration,
+      observationRevision: pauseObservationRevision,
+      paused: paused
+    )
+    pendingPauseIntents[intent.generation] = paused
     cachedPaused = paused
     cacheLock.unlock()
+    return intent
+  }
+
+  func finishCachedPauseIntent(
+    _ intent: (generation: UInt64, observationRevision: UInt64, paused: Bool),
+    result: Result<Void, Error>
+  ) {
+    cacheLock.lock()
+    pendingPauseIntents.removeValue(forKey: intent.generation)
+    if intent.generation >= resolvedPauseGeneration {
+      resolvedPauseGeneration = intent.generation
+      if case .success = result,
+        intent.observationRevision == pauseObservationRevision
+      {
+        confirmedPaused = intent.paused
+      }
+      retirePauseIntents(through: intent.generation)
+    }
+    if let latest = pendingPauseIntents.lazy
+      .filter({ $0.key > self.resolvedPauseGeneration })
+      .max(by: { $0.key < $1.key })
+    {
+      cachedPaused = latest.value
+    } else {
+      cachedPaused = confirmedPaused
+    }
+    cacheLock.unlock()
   }
 
-  private func convertNode(_ node: mpv_node) -> Any? {
+  private static let maximumNodeDepth = 32
+  private static let maximumNodeEntries = 4_096
+  private static let maximumNodeByteCount = 16 * 1_024 * 1_024
+  private static let maximumSideDataDimension: Int64 = 16_384
+  private static let maximumSideDataPixels: Int64 = 64 * 1_024 * 1_024
+
+  private struct NodeConversionBudget {
+    var remainingEntries = MpvPlayerCoreBase.maximumNodeEntries
+    var remainingBytes = MpvPlayerCoreBase.maximumNodeByteCount
+  }
+
+  func convertNode(_ node: mpv_node) -> Any? {
+    var budget = NodeConversionBudget()
+    return convertNode(node, depth: 0, budget: &budget)
+  }
+
+  func validateSideDataDimensions(width: Int64, height: Int64) -> Bool {
+    guard width > 0, height > 0,
+      width <= Self.maximumSideDataDimension,
+      height <= Self.maximumSideDataDimension
+    else { return false }
+    let (pixels, overflow) = width.multipliedReportingOverflow(by: height)
+    return !overflow && pixels <= Self.maximumSideDataPixels
+  }
+
+  private func dimensionValue(_ node: mpv_node) -> Int64? {
+    switch node.format {
+    case MPV_FORMAT_INT64:
+      return node.u.int64
+    case MPV_FORMAT_DOUBLE:
+      let value = node.u.double_
+      guard value.isFinite, value.rounded() == value,
+        value >= Double(Int64.min), value <= Double(Int64.max)
+      else { return nil }
+      return Int64(value)
+    default:
+      return nil
+    }
+  }
+
+  private func convertNodeString(
+    _ pointer: UnsafePointer<CChar>,
+    budget: inout NodeConversionBudget
+  ) -> String? {
+    let length = strnlen(pointer, budget.remainingBytes + 1)
+    guard length <= budget.remainingBytes else { return nil }
+    budget.remainingBytes -= length
+    if let string = String(validatingUTF8: pointer) {
+      return string
+    }
+    let bytes = UnsafeBufferPointer(
+      start: UnsafeRawPointer(pointer).assumingMemoryBound(to: UInt8.self),
+      count: length
+    )
+    return String(bytes.map { Character(Unicode.Scalar($0)) })
+  }
+
+  private func hasValidSideDataDimensions(_ list: mpv_node_list, count: Int) -> Bool {
+    var hasByteArray = false
+    var widthNode: mpv_node?
+    var heightNode: mpv_node?
+    for index in 0..<count {
+      let value = list.values[index]
+      hasByteArray = hasByteArray || value.format == MPV_FORMAT_BYTE_ARRAY
+      guard let keyPointer = list.keys[index] else { continue }
+      if strcmp(keyPointer, "width") == 0 || strcmp(keyPointer, "w") == 0 {
+        widthNode = value
+      } else if strcmp(keyPointer, "height") == 0 || strcmp(keyPointer, "h") == 0 {
+        heightNode = value
+      }
+    }
+    guard hasByteArray, widthNode != nil || heightNode != nil else { return true }
+    guard let widthNode, let heightNode,
+      let width = dimensionValue(widthNode),
+      let height = dimensionValue(heightNode)
+    else { return false }
+    return validateSideDataDimensions(width: width, height: height)
+  }
+
+  private func convertNode(
+    _ node: mpv_node,
+    depth: Int,
+    budget: inout NodeConversionBudget
+  ) -> Any? {
+    guard depth <= Self.maximumNodeDepth, budget.remainingEntries > 0 else { return nil }
+    budget.remainingEntries -= 1
+
     switch node.format {
     case MPV_FORMAT_STRING:
-      return node.u.string.map { safeString($0) }
+      guard let string = node.u.string else { return nil }
+      return convertNodeString(string, budget: &budget)
 
     case MPV_FORMAT_FLAG:
       return node.u.flag != 0
@@ -1253,11 +1426,24 @@ class MpvPlayerCoreBase: NSObject {
     case MPV_FORMAT_DOUBLE:
       return node.u.double_
 
+    case MPV_FORMAT_BYTE_ARRAY:
+      guard let byteArray = node.u.ba?.pointee else { return nil }
+      let byteCount = byteArray.size
+      guard byteCount <= budget.remainingBytes else { return nil }
+      guard byteCount == 0 || byteArray.data != nil else { return nil }
+      budget.remainingBytes -= byteCount
+      guard byteCount > 0, let data = byteArray.data else { return Data() }
+      return Data(bytes: data, count: byteCount)
+
     case MPV_FORMAT_NODE_ARRAY:
       guard let list = node.u.list?.pointee else { return nil }
+      let count = Int(list.num)
+      guard list.num >= 0, count <= budget.remainingEntries else { return nil }
+      guard count == 0 || list.values != nil else { return nil }
       var array = [Any]()
-      for index in 0..<Int(list.num) {
-        if let item = convertNode(list.values[index]) {
+      array.reserveCapacity(count)
+      for index in 0..<count {
+        if let item = convertNode(list.values[index], depth: depth + 1, budget: &budget) {
           array.append(item)
         }
       }
@@ -1265,10 +1451,16 @@ class MpvPlayerCoreBase: NSObject {
 
     case MPV_FORMAT_NODE_MAP:
       guard let list = node.u.list?.pointee else { return nil }
+      let count = Int(list.num)
+      guard list.num >= 0, count <= budget.remainingEntries else { return nil }
+      guard count == 0 || (list.values != nil && list.keys != nil) else { return nil }
+      guard hasValidSideDataDimensions(list, count: count) else { return nil }
       var dictionary = [String: Any]()
-      for index in 0..<Int(list.num) {
-        if let key = list.keys?[index].map({ safeString($0) }),
-          let value = convertNode(list.values[index])
+      dictionary.reserveCapacity(count)
+      for index in 0..<count {
+        if let keyPointer = list.keys[index],
+          let key = convertNodeString(keyPointer, budget: &budget),
+          let value = convertNode(list.values[index], depth: depth + 1, budget: &budget)
         {
           dictionary[key] = value
         }

shared/cpp/sanitize_utf8.h

@@ -12,10 +12,8 @@
 // mpv does not guarantee UTF-8 for log messages, error strings, or
 // system-encoded paths — sending these unsanitized through Flutter's
 // StandardMessageCodec causes FormatException crashes.
-static inline std::string SanitizeUtf8(const char* input) {
-  if (!input) return std::string();
-  size_t len = strlen(input);
-  if (len == 0) return std::string();
+static inline std::string SanitizeUtf8(const char* input, size_t len) {
+  if (!input || len == 0) return std::string();
 
   // Fast path: SIMD-accelerated validation — almost all strings pass this
   if (simdutf::validate_utf8(input, len)) {
@@ -46,4 +44,8 @@ static inline std::string SanitizeUtf8(const char* input) {
   return result;
 }
 
+static inline std::string SanitizeUtf8(const char* input) {
+  return input ? SanitizeUtf8(input, strlen(input)) : std::string();
+}
+
 #endif  // SANITIZE_UTF8_H_

shared/mpv/mpv_player_common.h

@@ -20,9 +20,9 @@ namespace mpv_common {
 using StatusCallback = std::function<void(int error)>;
 using GetPropertyCallback = std::function<void(int error, const std::string& value)>;
 
-inline constexpr char kSetPropertyFailedCode[] = "SET_PROPERTY_FAILED";
-inline constexpr char kSetPropertyNotInitializedCode[] = "NOT_INITIALIZED";
-inline constexpr size_t kSetPropertyErrorDescriptionLimit = 160;
+static constexpr char kSetPropertyFailedCode[] = "SET_PROPERTY_FAILED";
+static constexpr char kSetPropertyNotInitializedCode[] = "NOT_INITIALIZED";
+static constexpr size_t kSetPropertyErrorDescriptionLimit = 160;
 
 inline bool SetPropertyStatusSucceeded(int status) { return status >= 0; }
 
@@ -123,16 +123,19 @@ struct ObservationRequest {
 class PropertyObservationRegistry {
  public:
   ObservationRequest Register(const std::string& name, const std::string& format, int id) {
+    const mpv_format parsed_format = ParsePropertyFormat(format);
+    std::lock_guard<std::mutex> lock(mutex_);
     if (userdata_by_name_.find(name) != userdata_by_name_.end()) {
       return {false, 0, MPV_FORMAT_NONE};
     }
     const uint64_t userdata = next_userdata_++;
     userdata_by_name_[name] = userdata;
     id_by_name_[name] = id;
-    return {true, userdata, ParsePropertyFormat(format)};
+    return {true, userdata, parsed_format};
   }
 
   bool LookupId(const std::string& name, int* id) const {
+    std::lock_guard<std::mutex> lock(mutex_);
     const auto it = id_by_name_.find(name);
     if (it == id_by_name_.end()) return false;
     *id = it->second;
@@ -140,6 +143,7 @@ class PropertyObservationRegistry {
   }
 
   void Clear() {
+    std::lock_guard<std::mutex> lock(mutex_);
     userdata_by_name_.clear();
     id_by_name_.clear();
   }
@@ -148,6 +152,7 @@ class PropertyObservationRegistry {
   uint64_t next_userdata_ = 1;
   std::map<std::string, uint64_t> userdata_by_name_;
   std::map<std::string, int> id_by_name_;
+  mutable std::mutex mutex_;
 };
 
 inline bool ParseEnabledFlag(const std::string& value) { return value == "yes" || value == "true" || value == "1"; }
@@ -160,6 +165,7 @@ struct AudioReloadAction {
   AudioReloadReason reason = AudioReloadReason::kNone;
   int attempt = 0;
   bool exhausted = false;
+  uint64_t request_generation = 0;
 };
 
 enum class AudioOutputTransition { kNone, kFellBackToNull, kRecovered };
@@ -168,23 +174,45 @@ class AudioRecoveryState {
  public:
   using Clock = std::chrono::steady_clock;
 
-  void SetFileLoaded(bool loaded) {
+  void SetFileLoaded(bool loaded, Clock::time_point now = Clock::now()) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    const bool was_loaded = file_loaded_;
     file_loaded_ = loaded;
     if (!loaded) {
+      resume_requested_ = false;
       resume_attempts_left_ = 0;
       null_attempts_left_ = 0;
+      reload_pending_ = false;
+      pending_request_generation_ = 0;
+      return;
     }
-  }
-
-  void RequestResume() { resume_requested_.store(true); }
-
-  AudioOutputTransition SetCurrentAudioOutputNull(bool is_null, Clock::time_point now) {
-    if (is_null == current_ao_is_null_) return AudioOutputTransition::kNone;
-    current_ao_is_null_ = is_null;
-    if (is_null) {
+    if (!was_loaded && current_ao_is_null_) {
       null_attempts_left_ = kNullRetryBudget;
       null_backoff_ = NullFirstDelay();
       null_next_attempt_ = now + NullFirstDelay();
+    }
+  }
+
+  void RequestResume() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (!file_loaded_) {
+      resume_requested_ = false;
+      resume_attempts_left_ = 0;
+      return;
+    }
+    resume_requested_ = true;
+  }
+
+  AudioOutputTransition SetCurrentAudioOutputNull(bool is_null, Clock::time_point now) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (is_null == current_ao_is_null_) return AudioOutputTransition::kNone;
+    current_ao_is_null_ = is_null;
+    if (is_null) {
+      if (file_loaded_) {
+        null_attempts_left_ = kNullRetryBudget;
+        null_backoff_ = NullFirstDelay();
+        null_next_attempt_ = now + NullFirstDelay();
+      }
       return AudioOutputTransition::kFellBackToNull;
     }
     null_attempts_left_ = 0;
@@ -192,7 +220,8 @@ class AudioRecoveryState {
   }
 
   bool OnAudioDeviceListChanged(Clock::time_point now) {
-    if (!current_ao_is_null_) return false;
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (!file_loaded_ || !current_ao_is_null_) return false;
     const auto candidate = now + DeviceListDebounce();
     if (null_attempts_left_ <= 0 || candidate < null_next_attempt_) {
       null_next_attempt_ = candidate;
@@ -203,7 +232,9 @@ class AudioRecoveryState {
   }
 
   AudioReloadAction NextReload(Clock::time_point now) {
-    if (resume_requested_.exchange(false) && file_loaded_) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (resume_requested_ && file_loaded_) {
+      resume_requested_ = false;
       resume_attempts_left_ = kResumeReloadAttempts;
       resume_next_attempt_ = now + ResumeFirstDelay();
     }
@@ -214,7 +245,8 @@ class AudioRecoveryState {
       --resume_attempts_left_;
       resume_next_attempt_ = now + ResumeRetryDelay();
       reload_pending_ = true;
-      return {AudioReloadReason::kResume, attempt, false};
+      pending_request_generation_ = ++next_request_generation_;
+      return {AudioReloadReason::kResume, attempt, false, pending_request_generation_};
     }
 
     if (null_attempts_left_ > 0 && now >= null_next_attempt_) {
@@ -227,20 +259,27 @@ class AudioRecoveryState {
       null_next_attempt_ = now + null_backoff_;
       null_backoff_ = std::min(null_backoff_ * 2, NullBackoffCap());
       reload_pending_ = true;
-      return {AudioReloadReason::kNullFallback, attempt, null_attempts_left_ == 0};
+      pending_request_generation_ = ++next_request_generation_;
+      return {AudioReloadReason::kNullFallback, attempt, null_attempts_left_ == 0, pending_request_generation_};
     }
     return {};
   }
 
-  void CompleteReload() { reload_pending_ = false; }
-
-  bool HasPendingWork() const {
-    return resume_requested_.load() || resume_attempts_left_ > 0 || null_attempts_left_ > 0 || reload_pending_;
+  bool CompleteReload(uint64_t request_generation) {
+    std::lock_guard<std::mutex> lock(mutex_);
+    if (!reload_pending_ || pending_request_generation_ != request_generation) {
+      return false;
+    }
+    reload_pending_ = false;
+    pending_request_generation_ = 0;
+    return true;
   }
 
-  bool current_audio_output_is_null() const { return current_ao_is_null_; }
-
-  static int NullRetryBudget() { return kNullRetryBudget; }
+  bool HasPendingWork() const {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return file_loaded_ &&
+           (resume_requested_ || resume_attempts_left_ > 0 || null_attempts_left_ > 0 || reload_pending_);
+  }
 
  private:
   static constexpr int kResumeReloadAttempts = 2;
@@ -252,15 +291,18 @@ class AudioRecoveryState {
   static std::chrono::milliseconds NullBackoffCap() { return std::chrono::milliseconds(8000); }
   static std::chrono::milliseconds DeviceListDebounce() { return std::chrono::milliseconds(250); }
 
-  std::atomic<bool> resume_requested_{false};
+  bool resume_requested_ = false;
   bool file_loaded_ = false;
   bool current_ao_is_null_ = false;
   bool reload_pending_ = false;
+  uint64_t next_request_generation_ = 0;
+  uint64_t pending_request_generation_ = 0;
   int resume_attempts_left_ = 0;
   Clock::time_point resume_next_attempt_{};
   int null_attempts_left_ = 0;
   Clock::time_point null_next_attempt_{};
   std::chrono::milliseconds null_backoff_{0};
+  mutable std::mutex mutex_;
 };
 
 }  // namespace mpv_common

shared/mpv/mpv_player_common_test.cpp

@@ -4,9 +4,12 @@
 #undef NDEBUG
 #endif
 
+#include <atomic>
 #include <cassert>
 #include <chrono>
 #include <string>
+#include <thread>
+#include <vector>
 
 namespace {
 
@@ -41,6 +44,35 @@ void TestRequestRegistry() {
   assert(cancelled.properties.size() == 1);
 }
 
+void TestConcurrentRequestCompletion() {
+  for (int iteration = 0; iteration < 200; ++iteration) {
+    plezy::mpv_common::AsyncRequestRegistry registry;
+    std::atomic<int> completions{0};
+    const auto id = registry.RegisterStatus([&](int) { completions.fetch_add(1); });
+    std::atomic<bool> start{false};
+
+    std::thread taker([&]() {
+      while (!start.load(std::memory_order_acquire)) {
+      }
+      auto callback = registry.TakeStatus(id);
+      if (callback) callback(0);
+    });
+    std::thread canceller([&]() {
+      while (!start.load(std::memory_order_acquire)) {
+      }
+      auto cancelled = registry.CancelAll();
+      for (auto& callback : cancelled.status) {
+        callback(MPV_ERROR_UNINITIALIZED);
+      }
+    });
+
+    start.store(true, std::memory_order_release);
+    taker.join();
+    canceller.join();
+    assert(completions.load() == 1);
+  }
+}
+
 void TestSetPropertyResultContract() {
   using namespace plezy::mpv_common;
 
@@ -84,6 +116,66 @@ void TestPropertyObservationRegistry() {
   assert(!registry.LookupId("pause", &id));
 }
 
+void TestConcurrentPropertyObservationRegistry() {
+  constexpr int kPropertyCount = 512;
+  constexpr int kClearRounds = 32;
+  plezy::mpv_common::PropertyObservationRegistry registry;
+  std::vector<std::string> names;
+  names.reserve(kPropertyCount);
+  for (int i = 0; i < kPropertyCount; ++i) {
+    names.push_back("property-" + std::to_string(i));
+  }
+
+  std::atomic<bool> start{false};
+  std::atomic<bool> writer_done{false};
+  std::thread writer([&]() {
+    while (!start.load(std::memory_order_acquire)) {
+    }
+    for (int round = 0; round < kClearRounds; ++round) {
+      for (int i = 0; i < kPropertyCount; ++i) {
+        registry.Register(names[i], "int64", 1000 + i);
+      }
+    }
+    writer_done.store(true, std::memory_order_release);
+  });
+  std::thread reader([&]() {
+    while (!start.load(std::memory_order_acquire)) {
+    }
+    while (!writer_done.load(std::memory_order_acquire)) {
+      for (int i = 0; i < kPropertyCount; ++i) {
+        int id = 0;
+        if (registry.LookupId(names[i], &id)) {
+          assert(id == 1000 + i);
+        }
+      }
+    }
+  });
+  std::thread clearer([&]() {
+    while (!start.load(std::memory_order_acquire)) {
+    }
+    for (int round = 0; round < kClearRounds; ++round) {
+      registry.Clear();
+      std::this_thread::yield();
+    }
+  });
+
+  start.store(true, std::memory_order_release);
+  writer.join();
+  reader.join();
+  clearer.join();
+
+  registry.Clear();
+  for (int i = 0; i < kPropertyCount; ++i) {
+    const auto request = registry.Register(names[i], "int64", 1000 + i);
+    assert(request.added);
+  }
+  for (int i = 0; i < kPropertyCount; ++i) {
+    int id = 0;
+    assert(registry.LookupId(names[i], &id));
+    assert(id == 1000 + i);
+  }
+}
+
 void TestResumeRecoverySchedule() {
   AudioRecoveryState state;
   const auto start = AudioRecoveryState::Clock::time_point{};
@@ -98,15 +190,72 @@ void TestResumeRecoverySchedule() {
   assert(first.reason == AudioReloadReason::kResume);
   assert(first.attempt == 1);
   assert(!first.exhausted);
-  state.CompleteReload();
+  assert(state.CompleteReload(first.request_generation));
 
   const auto second = state.NextReload(start + std::chrono::milliseconds(6000));
   assert(second.reason == AudioReloadReason::kResume);
   assert(second.attempt == 2);
-  state.CompleteReload();
+  assert(state.CompleteReload(second.request_generation));
   assert(!state.HasPendingWork());
 }
 
+void TestConcurrentAudioRecoveryState() {
+  AudioRecoveryState state;
+  const auto start = AudioRecoveryState::Clock::time_point{};
+  state.SetFileLoaded(true);
+  std::atomic<bool> begin{false};
+
+  std::thread resume([&]() {
+    while (!begin.load(std::memory_order_acquire)) {
+    }
+    for (int i = 0; i < 1000; ++i) state.RequestResume();
+  });
+  std::thread device([&]() {
+    while (!begin.load(std::memory_order_acquire)) {
+    }
+    for (int i = 0; i < 1000; ++i) {
+      state.SetCurrentAudioOutputNull(true, start);
+      state.OnAudioDeviceListChanged(start);
+    }
+  });
+  std::thread timer([&]() {
+    while (!begin.load(std::memory_order_acquire)) {
+    }
+    for (int i = 0; i < 1000; ++i) {
+      const auto action = state.NextReload(start + std::chrono::hours(1));
+      if (action.reason != AudioReloadReason::kNone) {
+        state.CompleteReload(action.request_generation);
+      }
+    }
+  });
+
+  begin.store(true, std::memory_order_release);
+  resume.join();
+  device.join();
+  timer.join();
+  state.SetFileLoaded(false);
+  assert(!state.HasPendingWork());
+}
+
+void TestFileBoundaryRestartsNullRecoveryOnlyAfterLoad() {
+  AudioRecoveryState state;
+  const auto start = AudioRecoveryState::Clock::time_point{};
+  state.SetFileLoaded(true, start);
+  assert(state.SetCurrentAudioOutputNull(true, start) == AudioOutputTransition::kFellBackToNull);
+  assert(state.HasPendingWork());
+
+  state.SetFileLoaded(false, start + std::chrono::milliseconds(100));
+  assert(!state.HasPendingWork());
+  assert(!state.OnAudioDeviceListChanged(start + std::chrono::milliseconds(200)));
+
+  state.SetFileLoaded(true, start + std::chrono::milliseconds(300));
+  assert(state.HasPendingWork());
+  assert(state.NextReload(start + std::chrono::milliseconds(799)).reason == AudioReloadReason::kNone);
+  const auto retry = state.NextReload(start + std::chrono::milliseconds(800));
+  assert(retry.reason == AudioReloadReason::kNullFallback);
+  assert(retry.attempt == 1);
+}
+
 void TestNullFallbackRecoverySchedule() {
   AudioRecoveryState state;
   const auto start = AudioRecoveryState::Clock::time_point{};
@@ -116,35 +265,35 @@ void TestNullFallbackRecoverySchedule() {
   auto action = state.NextReload(start + std::chrono::milliseconds(500));
   assert(action.reason == AudioReloadReason::kNullFallback);
   assert(action.attempt == 1);
-  state.CompleteReload();
+  assert(state.CompleteReload(action.request_generation));
 
   action = state.NextReload(start + std::chrono::milliseconds(1000));
   assert(action.reason == AudioReloadReason::kNullFallback);
   assert(action.attempt == 2);
-  state.CompleteReload();
+  assert(state.CompleteReload(action.request_generation));
 
   action = state.NextReload(start + std::chrono::milliseconds(2000));
   assert(action.reason == AudioReloadReason::kNullFallback);
   assert(action.attempt == 3);
-  state.CompleteReload();
+  assert(state.CompleteReload(action.request_generation));
 
   action = state.NextReload(start + std::chrono::milliseconds(4000));
   assert(action.reason == AudioReloadReason::kNullFallback);
   assert(action.attempt == 4);
-  state.CompleteReload();
+  assert(state.CompleteReload(action.request_generation));
 
   action = state.NextReload(start + std::chrono::milliseconds(8000));
   assert(action.reason == AudioReloadReason::kNullFallback);
   assert(action.attempt == 5);
   assert(action.exhausted);
-  state.CompleteReload();
+  assert(state.CompleteReload(action.request_generation));
   assert(!state.HasPendingWork());
 
   assert(state.OnAudioDeviceListChanged(start + std::chrono::milliseconds(9000)));
   action = state.NextReload(start + std::chrono::milliseconds(9250));
   assert(action.reason == AudioReloadReason::kNullFallback);
   assert(action.attempt == 1);
-  state.CompleteReload();
+  assert(state.CompleteReload(action.request_generation));
 
   assert(
       state.SetCurrentAudioOutputNull(false, start + std::chrono::milliseconds(9300)) ==
@@ -152,6 +301,37 @@ void TestNullFallbackRecoverySchedule() {
   assert(!state.HasPendingWork());
 }
 
+void TestUnloadedResumeIsConsumed() {
+  AudioRecoveryState state;
+  const auto start = AudioRecoveryState::Clock::time_point{};
+
+  state.RequestResume();
+  assert(!state.HasPendingWork());
+  assert(state.NextReload(start + std::chrono::hours(1)).reason == AudioReloadReason::kNone);
+
+  state.SetFileLoaded(true, start);
+  assert(!state.HasPendingWork());
+}
+
+void TestStaleReloadCompletionCannotClearCurrentRequest() {
+  AudioRecoveryState state;
+  const auto start = AudioRecoveryState::Clock::time_point{};
+  state.SetFileLoaded(true, start);
+  assert(state.SetCurrentAudioOutputNull(true, start) == AudioOutputTransition::kFellBackToNull);
+  const auto old_request = state.NextReload(start + std::chrono::milliseconds(500));
+  assert(old_request.reason == AudioReloadReason::kNullFallback);
+
+  state.SetFileLoaded(false, start + std::chrono::milliseconds(600));
+  state.SetFileLoaded(true, start + std::chrono::milliseconds(700));
+  const auto current_request = state.NextReload(start + std::chrono::milliseconds(1200));
+  assert(current_request.reason == AudioReloadReason::kNullFallback);
+  assert(current_request.request_generation != old_request.request_generation);
+
+  assert(!state.CompleteReload(old_request.request_generation));
+  assert(state.NextReload(start + std::chrono::hours(1)).reason == AudioReloadReason::kNone);
+  assert(state.CompleteReload(current_request.request_generation));
+}
+
 void TestHdrHelpers() {
   assert(plezy::mpv_common::ParseEnabledFlag("yes"));
   assert(plezy::mpv_common::ParseEnabledFlag("true"));
@@ -165,10 +345,16 @@ void TestHdrHelpers() {
 
 int main() {
   TestRequestRegistry();
+  TestConcurrentRequestCompletion();
   TestSetPropertyResultContract();
   TestPropertyObservationRegistry();
+  TestConcurrentPropertyObservationRegistry();
   TestResumeRecoverySchedule();
+  TestConcurrentAudioRecoveryState();
   TestNullFallbackRecoverySchedule();
+  TestFileBoundaryRestartsNullRecoveryOnlyAfterLoad();
+  TestUnloadedResumeIsConsumed();
+  TestStaleReloadCompletionCannotClearCurrentRequest();
   TestHdrHelpers();
   return 0;
 }