fix: concurrency issues on ios

Author: Ivan Sekovanikj

examples/SampleApp/ios/Podfile

@@ -5,6 +5,34 @@ require Pod::Executable.execute_command('node', ['-p',
     {paths: [process.argv[1]]},
   )', __dir__]).strip
 
+react_native_path = File.dirname(
+  Pod::Executable.execute_command('node', ['-p',
+    'require.resolve(
+      "react-native/package.json",
+      {paths: [process.argv[1]]},
+    )', __dir__]).strip,
+)
+
+fmt_podspec_path = File.join(react_native_path, 'third-party-podspecs', 'fmt.podspec')
+rct_folly_podspec_path = File.join(react_native_path, 'third-party-podspecs', 'RCT-Folly.podspec')
+
+fmt_podspec = File.read(fmt_podspec_path)
+fmt_podspec = fmt_podspec.gsub('spec.version = "11.0.2"', 'spec.version = "12.1.0"')
+fmt_podspec = fmt_podspec.gsub(':tag => "11.0.2"', ':tag => "12.1.0"')
+fmt_podspec = fmt_podspec.gsub(
+  '"GCC_WARN_INHIBIT_ALL_WARNINGS" => "YES" # Disable warnings because we don\'t control this library',
+  "\"GCC_WARN_INHIBIT_ALL_WARNINGS\" => \"YES\", \"OTHER_CPLUSPLUSFLAGS\" => \"$(inherited) -DFMT_USE_CONSTEVAL=0\" # Disable warnings because we don't control this library",
+)
+File.write(fmt_podspec_path, fmt_podspec)
+
+rct_folly_podspec = File.read(rct_folly_podspec_path)
+rct_folly_podspec = rct_folly_podspec.gsub('spec.dependency "fmt", "11.0.2"', 'spec.dependency "fmt", "12.1.0"')
+rct_folly_podspec = rct_folly_podspec.gsub(
+  '"GCC_WARN_INHIBIT_ALL_WARNINGS" => "YES" # Disable warnings because we don\'t control this library',
+  "\"GCC_WARN_INHIBIT_ALL_WARNINGS\" => \"YES\", \"OTHER_CPLUSPLUSFLAGS\" => \"$(inherited) -DFMT_USE_CONSTEVAL=0\" # Disable warnings because we don't control this library",
+)
+File.write(rct_folly_podspec_path, rct_folly_podspec)
+
 platform :ios, min_ios_version_supported
 prepare_react_native_project!
 
@@ -55,5 +83,18 @@ target 'SampleApp' do
       :mac_catalyst_enabled => false,
       # :ccache_enabled => true
     )
+
+    installer.pods_project.targets.each do |target|
+      next unless ['fmt', 'RCT-Folly'].include?(target.name)
+
+      target.build_configurations.each do |config|
+        flags = Array(config.build_settings['OTHER_CPLUSPLUSFLAGS'] || '$(inherited)')
+        unless flags.include?('-DFMT_USE_CONSTEVAL=0')
+          flags << '-DFMT_USE_CONSTEVAL=0'
+        end
+        config.build_settings['OTHER_CPLUSPLUSFLAGS'] = flags
+      end
+    end
+
   end
 end

examples/SampleApp/ios/Podfile.lock

@@ -73,7 +73,7 @@ PODS:
     - GoogleUtilities/Reachability (~> 8.1)
     - GoogleUtilities/UserDefaults (~> 8.1)
     - nanopb (~> 3.30910.0)
-  - FirebaseRemoteConfigInterop (11.14.0)
+  - FirebaseRemoteConfigInterop (11.15.0)
   - FirebaseSessions (11.13.0):
     - FirebaseCore (~> 11.13.0)
     - FirebaseCoreExtension (~> 11.13.0)
@@ -83,7 +83,7 @@ PODS:
     - GoogleUtilities/UserDefaults (~> 8.1)
     - nanopb (~> 3.30910.0)
     - PromisesSwift (~> 2.1)
-  - fmt (11.0.2)
+  - fmt (12.1.0)
   - glog (0.3.5)
   - GoogleAppMeasurement (11.13.0):
     - GoogleAppMeasurement/AdIdSupport (= 11.13.0)
@@ -251,20 +251,20 @@ PODS:
     - boost
     - DoubleConversion
     - fast_float (= 8.0.0)
-    - fmt (= 11.0.2)
+    - fmt (= 12.1.0)
     - glog
     - RCT-Folly/Default (= 2024.11.18.00)
   - RCT-Folly/Default (2024.11.18.00):
     - boost
     - DoubleConversion
     - fast_float (= 8.0.0)
-    - fmt (= 11.0.2)
+    - fmt (= 12.1.0)
     - glog
   - RCT-Folly/Fabric (2024.11.18.00):
     - boost
     - DoubleConversion
     - fast_float (= 8.0.0)
-    - fmt (= 11.0.2)
+    - fmt (= 12.1.0)
     - glog
   - RCTDeprecation (0.81.6)
   - RCTRequired (0.81.6)
@@ -3731,9 +3731,9 @@ SPEC CHECKSUMS:
   FirebaseCrashlytics: 8281e577b6f85a08ea7aeb8b66f95e1ae430c943
   FirebaseInstallations: 0ee9074f2c1e86561ace168ee1470dc67aabaf02
   FirebaseMessaging: 195bbdb73e6ca1dbc76cd46e73f3552c084ef6e4
-  FirebaseRemoteConfigInterop: 7b74ceaa54e28863ed17fa39da8951692725eced
+  FirebaseRemoteConfigInterop: 1c6135e8a094cc6368949f5faeeca7ee8948b8aa
   FirebaseSessions: eaa8ec037e7793769defe4201c20bd4d976f9677
-  fmt: a40bb5bd0294ea969aaaba240a927bd33d878cdd
+  fmt: 12a698626610c2fef5e7d8de472b100baf225f93
   glog: 5683914934d5b6e4240e497e0f4a3b42d1854183
   GoogleAppMeasurement: 0dfca1a4b534d123de3945e28f77869d10d0d600
   GoogleDataTransport: aae35b7ea0c09004c3797d53c8c41f66f219d6a7
@@ -3746,7 +3746,7 @@ SPEC CHECKSUMS:
   op-sqlite: 2e58f87227360fa6251d1fe103d189f11ae8c95f
   PromisesObjC: f5707f49cb48b9636751c5b2e7d227e43fba9f47
   PromisesSwift: 9d77319bbe72ebf6d872900551f7eeba9bce2851
-  RCT-Folly: 59ec0ac1f2f39672a0c6e6cecdd39383b764646f
+  RCT-Folly: 5a8bea092f38495b327c6eff2dc52ee25c10f637
   RCTDeprecation: d4ef510f229cea15314176aee5e3ba10064a8496
   RCTRequired: 1e41b794629558f6626e2bc39c166ac0ec1c5878
   RCTTypeSafety: 62c8105cf08af634c93d38ea1e8ec8a57b7abc2c
@@ -3839,6 +3839,6 @@ SPEC CHECKSUMS:
   Teleport: ed828b19e62ca8b9ec101d991bf0594b1c1c8812
   Yoga: ff16d80456ce825ffc9400eeccc645a0dfcccdf5
 
-PODFILE CHECKSUM: 4f662370295f8f9cee909f1a4c59a614999a209d
+PODFILE CHECKSUM: 84efea5f3e8c9c79671ee6e525f700f244c17388
 
 COCOAPODS: 1.15.2

package/expo-package/src/native/NativeStreamMultipartUploader.ts

@@ -9,29 +9,29 @@ export type UploadHeader = {
 
 export type UploadPart = {
   fieldName: string;
-  fileName?: string | null;
+  fileName?: string;
   kind: string;
-  mimeType?: string | null;
-  uri?: string | null;
-  value?: string | null;
+  mimeType?: string;
+  uri?: string;
+  value?: string;
 };
 
 export type UploadProgressConfig = {
-  count?: number | null;
-  intervalMs?: number | null;
+  count?: number;
+  intervalMs?: number;
 };
 
 export type UploadProgressEvent = {
   loaded: number;
-  total?: number | null;
+  total?: number;
   uploadId: string;
 };
 
 export type UploadResponse = {
   body: string;
-  headers?: ReadonlyArray<UploadHeader> | null;
+  headers?: ReadonlyArray<UploadHeader>;
   status: number;
-  statusText?: string | null;
+  statusText?: string;
 };
 
 export interface Spec extends TurboModule {
@@ -44,7 +44,7 @@ export interface Spec extends TurboModule {
     method: string,
     headers: ReadonlyArray<UploadHeader>,
     parts: ReadonlyArray<UploadPart>,
-    progress?: UploadProgressConfig | null,
+    progress?: UploadProgressConfig,
     timeoutMs?: number | null,
   ): Promise<UploadResponse>;
 }

package/native-package/src/native/NativeStreamMultipartUploader.ts

@@ -9,29 +9,29 @@ export type UploadHeader = {
 
 export type UploadPart = {
   fieldName: string;
-  fileName?: string | null;
+  fileName?: string;
   kind: string;
-  mimeType?: string | null;
-  uri?: string | null;
-  value?: string | null;
+  mimeType?: string;
+  uri?: string;
+  value?: string;
 };
 
 export type UploadProgressConfig = {
-  count?: number | null;
-  intervalMs?: number | null;
+  count?: number;
+  intervalMs?: number;
 };
 
 export type UploadProgressEvent = {
   loaded: number;
-  total?: number | null;
+  total?: number;
   uploadId: string;
 };
 
 export type UploadResponse = {
   body: string;
-  headers?: ReadonlyArray<UploadHeader> | null;
+  headers?: ReadonlyArray<UploadHeader>;
   status: number;
-  statusText?: string | null;
+  statusText?: string;
 };
 
 export interface Spec extends TurboModule {
@@ -44,7 +44,7 @@ export interface Spec extends TurboModule {
     method: string,
     headers: ReadonlyArray<UploadHeader>,
     parts: ReadonlyArray<UploadPart>,
-    progress?: UploadProgressConfig | null,
+    progress?: UploadProgressConfig,
     timeoutMs?: number | null,
   ): Promise<UploadResponse>;
 }

package/shared-native/android/upload/StreamMultipartUploader.kt

@@ -35,7 +35,10 @@ object StreamMultipartUploader {
 
     val httpRequest = createRequest(context, request, onProgress)
     val call = clientFor(request).newCall(httpRequest)
-    inFlightCalls[request.uploadId] = call
+    val existingCall = inFlightCalls.putIfAbsent(request.uploadId, call)
+    if (existingCall != null) {
+      throw IllegalStateException("Upload already in flight for id: ${request.uploadId}")
+    }
 
     try {
       if (cancelledUploadIds.remove(request.uploadId)) {
@@ -54,7 +57,7 @@ object StreamMultipartUploader {
         )
       }
     } finally {
-      inFlightCalls.remove(request.uploadId)
+      inFlightCalls.remove(request.uploadId, call)
       cancelledUploadIds.remove(request.uploadId)
     }
   }

package/shared-native/ios/StreamMultipartUploadManager.swift

@@ -1,5 +1,62 @@
 import Foundation
 
+private actor StreamMultipartUploadConcurrencyLimiter {
+  private var activeUploads = 0
+  private let maxConcurrentUploads: Int
+  private var waiterOrder = [UUID]()
+  private var waiters = [UUID: CheckedContinuation<Void, Error>]()
+
+  init(maxConcurrentUploads: Int) {
+    self.maxConcurrentUploads = max(1, maxConcurrentUploads)
+  }
+
+  func acquire() async throws {
+    if activeUploads < maxConcurrentUploads {
+      activeUploads += 1
+      return
+    }
+
+    let waiterId = UUID()
+
+    try await withTaskCancellationHandler {
+      try await withCheckedThrowingContinuation { continuation in
+        if activeUploads < maxConcurrentUploads {
+          activeUploads += 1
+          continuation.resume()
+          return
+        }
+
+        waiterOrder.append(waiterId)
+        waiters[waiterId] = continuation
+      }
+    } onCancel: {
+      Task {
+        await self.cancelWaiter(id: waiterId)
+      }
+    }
+  }
+
+  func release() {
+    while !waiterOrder.isEmpty {
+      let waiterId = waiterOrder.removeFirst()
+
+      guard let continuation = waiters.removeValue(forKey: waiterId) else {
+        continue
+      }
+
+      continuation.resume()
+      return
+    }
+
+    activeUploads = max(0, activeUploads - 1)
+  }
+
+  private func cancelWaiter(id: UUID) {
+    waiterOrder.removeAll { $0 == id }
+    waiters.removeValue(forKey: id)?.resume(throwing: StreamMultipartUploadError.cancelled)
+  }
+}
+
 private final class StreamMultipartUploadTaskState {
   let bodyFactory: StreamMultipartUploadBodyStreamFactory
   let progressThrottler: StreamMultipartUploadProgressThrottler
@@ -29,15 +86,20 @@ private final class StreamMultipartUploadTaskState {
 final class StreamMultipartUploadManager: NSObject {
   static let shared = StreamMultipartUploadManager()
   private let maxResponseBodyBytes = 1_048_576
+  private let maxConcurrentUploads = min(max(ProcessInfo.processInfo.activeProcessorCount, 2), 4)
 
   private lazy var session: URLSession = {
     let delegateQueue = OperationQueue()
     delegateQueue.maxConcurrentOperationCount = 1
     delegateQueue.qualityOfService = .userInitiated
     let configuration = URLSessionConfiguration.ephemeral
+    configuration.httpMaximumConnectionsPerHost = maxConcurrentUploads
     configuration.waitsForConnectivity = false
     return URLSession(configuration: configuration, delegate: self, delegateQueue: delegateQueue)
   }()
+  private lazy var uploadLimiter = StreamMultipartUploadConcurrencyLimiter(
+    maxConcurrentUploads: maxConcurrentUploads
+  )
 
   private let lock = NSLock()
   private var cancelledUploadIds = Set<String>()
@@ -109,6 +171,7 @@ final class StreamMultipartUploadManager: NSObject {
 
     let progressThrottler =
       StreamMultipartUploadProgressThrottler(options: request.progress, onProgress: onProgress)
+    try await uploadLimiter.acquire()
 
     return try await withCheckedThrowingContinuation { continuation in
       let task = session.uploadTask(withStreamedRequest: urlRequest)
@@ -118,11 +181,17 @@ final class StreamMultipartUploadManager: NSObject {
         task: task,
         uploadId: uploadId
       ) { result in
+        Task {
+          await self.uploadLimiter.release()
+        }
         continuation.resume(with: result)
       }
 
       guard register(state) else {
         task.cancel()
+        Task {
+          await self.uploadLimiter.release()
+        }
         continuation.resume(throwing: StreamMultipartUploadError.cancelled)
         return
       }
@@ -245,7 +314,9 @@ final class StreamMultipartUploadManager: NSObject {
     lock.lock()
     let state = statesByTaskIdentifier.removeValue(forKey: taskIdentifier)
     if let uploadId = state?.uploadId {
-      taskIdentifiersByUploadId.removeValue(forKey: uploadId)
+      if taskIdentifiersByUploadId[uploadId] == taskIdentifier {
+        taskIdentifiersByUploadId.removeValue(forKey: uploadId)
+      }
       cancelledUploadIds.remove(uploadId)
     }
     lock.unlock()

package/shared-native/ios/StreamMultipartUploader.mm

@@ -52,7 +52,7 @@ - (void)uploadMultipart:(NSString *)uploadId
                headers:(NSArray<NSDictionary<NSString *, NSString *> *> *)headers
                  parts:(NSArray<NSDictionary<NSString *, id> *> *)parts
                progress:(JS::NativeStreamMultipartUploader::UploadProgressConfig &)progress
-             timeoutMs:(NSNumber * _Nullable)timeoutMs
+             timeoutMs:(NSNumber *)timeoutMs
                 resolve:(RCTPromiseResolveBlock)resolve
                  reject:(RCTPromiseRejectBlock)reject
 {

package/shared-native/ios/StreamMultipartUploaderBridge.swift

@@ -45,11 +45,16 @@ public final class StreamMultipartUploaderBridge: NSObject {
     completion: @escaping (NSDictionary?, NSError?) -> Void
   ) {
     let taskBox = StreamMultipartUploadBridgeTaskBox()
+    var replacedTaskBox: StreamMultipartUploadBridgeTaskBox?
 
     taskLock.lock()
-    tasksByUploadId[uploadId]?.cancel()
+    replacedTaskBox = tasksByUploadId[uploadId]
     tasksByUploadId[uploadId] = taskBox
     taskLock.unlock()
+    if replacedTaskBox != nil {
+      replacedTaskBox?.cancel()
+      StreamMultipartUploadManager.shared.cancel(uploadId: uploadId)
+    }
 
     let task = Task(priority: .userInitiated) {
       defer {