fix(storage, apple): release finished tasks instead of keeping them for the life of the process

packages/firebase_storage/firebase_storage/ios/firebase_storage/Sources/firebase_storage/FLTFirebaseStoragePlugin.swift

@@ -21,7 +21,7 @@ public final class FLTFirebaseStoragePlugin: NSObject, FlutterPlugin, FirebaseSt
   private var channel: FlutterMethodChannel?
   private var messenger: FlutterBinaryMessenger?
   private var eventChannels: [String: FlutterEventChannel] = [:]
-  private var streamHandlers: [String: FlutterStreamHandler] = [:]
+  private var streamHandlers: [String: TaskStateChannelStreamHandler] = [:]
   private var handleToTask: [Int64: AnyObject] = [:]
   private var handleToPath: [Int64: String] = [:]
   private var handleToIdentifier: [Int64: String] = [:]
@@ -48,9 +48,20 @@ public final class FLTFirebaseStoragePlugin: NSObject, FlutterPlugin, FirebaseSt
     instance.channel = channel
     instance.messenger = resolvedMessenger
     registrar.addMethodCallDelegate(instance, channel: channel)
+    #if os(iOS)
+      // Published so the engine calls detachFromEngine(for:), which releases
+      // every task still registered on this engine.
+      registrar.publish(instance)
+    #endif
     FirebaseStorageHostApiSetup.setUp(binaryMessenger: resolvedMessenger, api: instance)
   }
 
+  #if os(iOS)
+    public func detachFromEngine(for registrar: FlutterPluginRegistrar) {
+      releaseAllTasks()
+    }
+  #endif
+
   public func handle(_ call: FlutterMethodCall, result: @escaping FlutterResult) {
     result(FlutterMethodNotImplemented)
   }
@@ -429,24 +440,89 @@ public final class FLTFirebaseStoragePlugin: NSObject, FlutterPlugin, FirebaseSt
     task: StorageObservableTask, appName: String, handle: Int64,
     path: String
   ) -> String {
+    // A task whose event stream nobody ever listened to cannot be released
+    // through onCancel; drop the long-finished ones before adding a new task so
+    // the registries stay bounded.
+    sweepAbandonedTasks()
     let uuid = UUID().uuidString
     let channelName = "plugins.flutter.io/firebase_storage/taskEvent/\(uuid)"
     let channel = FlutterEventChannel(name: channelName, binaryMessenger: messenger!)
     let storageInstance = Storage.storage(app: FLTFirebasePlugin.firebaseAppNamed(appName)!)
-    channel.setStreamHandler(
-      TaskStateChannelStreamHandler(
-        task: task,
-        storage: storageInstance,
-        identifier: channelName
-      )
-    )
+    let handler = TaskStateChannelStreamHandler(
+      task: task,
+      storage: storageInstance,
+      identifier: channelName
+    ) { [weak self] identifier in
+      self?.releaseTask(identifier: identifier)
+    }
+    handler.armFinishWatch()
+    channel.setStreamHandler(handler)
+    streamHandlers[channelName] = handler
     eventChannels[channelName] = channel
     handleToTask[handle] = task as AnyObject
     handleToPath[handle] = path
     handleToIdentifier[handle] = channelName
     return uuid
   }
 
+  /// Forgets one task: its stream handler, its event channel and every handle
+  /// map entry. Runs when the Dart side cancels the task's event stream (which
+  /// it does right after the terminal event), from the sweep of finished tasks
+  /// nobody listened to, and on engine detach. Without it the plugin kept every
+  /// task — and, through `StorageUploadTask.uploadData`, every uploaded payload
+  /// — for the life of the process. Mirrors the ObjC plugin's
+  /// `cleanUpTask:handle:` (≤ 13.0.3) and Android's
+  /// `FlutterFirebaseStorageTask.destroy()`. Idempotent; main thread only.
+  private func releaseTask(identifier: String) {
+    if !Thread.isMainThread {
+      DispatchQueue.main.async { self.releaseTask(identifier: identifier) }
+      return
+    }
+    streamHandlers.removeValue(forKey: identifier)
+    FLTFirebaseStoragePlugin.canceledIdentifiers.remove(identifier)
+    let handles = handleToIdentifier.filter { $0.value == identifier }.map(\.key)
+    for handle in handles {
+      handleToTask.removeValue(forKey: handle)
+      handleToPath.removeValue(forKey: handle)
+      handleToIdentifier.removeValue(forKey: handle)
+    }
+    if let channel = eventChannels.removeValue(forKey: identifier) {
+      // One run-loop turn later: when this runs from the handler's onCancel the
+      // Dart `cancel` call that triggered it is still being dispatched on this
+      // channel, and a channel without a handler would answer it with a
+      // MissingPluginException report on the Dart side.
+      DispatchQueue.main.async {
+        channel.setStreamHandler(nil)
+      }
+    }
+  }
+
+  /// Releases tasks that finished without any Dart listener at least
+  /// `TaskStateChannelStreamHandler.unlistenedGraceInterval` ago; a listener
+  /// that attaches within the grace period still gets the terminal event
+  /// replayed by the SDK.
+  private func sweepAbandonedTasks() {
+    let abandoned = streamHandlers.filter { $0.value.isAbandoned }.map(\.key)
+    for identifier in abandoned {
+      releaseTask(identifier: identifier)
+    }
+  }
+
+  private func releaseAllTasks() {
+    for handler in streamHandlers.values {
+      handler.cleanupObservers()
+    }
+    for channel in eventChannels.values {
+      channel.setStreamHandler(nil)
+    }
+    streamHandlers.removeAll()
+    eventChannels.removeAll()
+    handleToTask.removeAll()
+    handleToPath.removeAll()
+    handleToIdentifier.removeAll()
+    FLTFirebaseStoragePlugin.canceledIdentifiers.removeAll()
+  }
+
   private func currentSnapshot(handle: Int64) -> [String: Any] {
     [
       "path": handleToPath[handle] ?? "",

packages/firebase_storage/firebase_storage/ios/firebase_storage/Sources/firebase_storage/TaskStateChannelStreamHandler.swift

@@ -12,25 +12,80 @@ import Foundation
 #endif
 
 final class TaskStateChannelStreamHandler: NSObject, FlutterStreamHandler {
+  /// How long a task that finished without any Dart listener is kept before the
+  /// plugin sweeps it. Dart subscribes within milliseconds of the `put*` call
+  /// returning, so this only has to outlast that round trip generously.
+  static let unlistenedGraceInterval: TimeInterval = 30
+
   private let task: StorageObservableTask
   private let storage: Storage
   private let identifier: String
+  /// Invoked with `identifier` once the Dart side has cancelled its event
+  /// subscription, so the plugin can drop every reference it holds for the
+  /// task (the SDK's `StorageUploadTask` keeps the uploaded `Data` alive for as
+  /// long as the task object lives).
+  private let onRelease: (String) -> Void
+
+  /// Set on the first `onListen`. A task that finishes while this is still
+  /// false has no Dart subscription that could ever cancel it, so the plugin
+  /// sweeps it after `unlistenedGraceInterval` (see `isAbandoned`).
+  private(set) var hasListened = false
+  /// Stamped (main thread) by the finish watch armed in `armFinishWatch`.
+  private var finishedAt: Date?
+  private var finishWatchHandles: [String] = []
 
   private var successHandle: String?
   private var failureHandle: String?
   private var pausedHandle: String?
   private var progressHandle: String?
 
-  init(task: StorageObservableTask, storage: Storage, identifier: String) {
+  init(
+    task: StorageObservableTask, storage: Storage, identifier: String,
+    onRelease: @escaping (String) -> Void
+  ) {
     self.task = task
     self.storage = storage
     self.identifier = identifier
+    self.onRelease = onRelease
+  }
+
+  /// Observes the task's terminal state independently of any Dart listener so
+  /// a task nobody ever subscribed to can still be recognised as finished. The
+  /// SDK reports a cancelled task as `.failure`, so both observers cover every
+  /// terminal state.
+  func armFinishWatch() {
+    let stamp: (StorageTaskSnapshot) -> Void = { [weak self] _ in
+      if Thread.isMainThread {
+        self?.markFinished()
+      } else {
+        DispatchQueue.main.async { self?.markFinished() }
+      }
+    }
+    finishWatchHandles = [
+      task.observe(.success, handler: stamp),
+      task.observe(.failure, handler: stamp),
+    ]
+  }
+
+  /// True for a task that finished without any Dart listener longer than
+  /// `unlistenedGraceInterval` ago: nothing will ever cancel its stream, so the
+  /// plugin releases it itself. Main thread only.
+  var isAbandoned: Bool {
+    guard !hasListened, let finishedAt else { return false }
+    return Date().timeIntervalSince(finishedAt) >= Self.unlistenedGraceInterval
+  }
+
+  private func markFinished() {
+    if finishedAt == nil {
+      finishedAt = Date()
+    }
   }
 
   func onListen(
     withArguments arguments: Any?,
     eventSink events: @escaping FlutterEventSink
   ) -> FlutterError? {
+    hasListened = true
     successHandle = task.observe(.success) { snapshot in
       events([
         "taskState": 2,  // success
@@ -68,18 +123,26 @@ final class TaskStateChannelStreamHandler: NSObject, FlutterStreamHandler {
 
   func onCancel(withArguments arguments: Any?) -> FlutterError? {
     cleanupObservers()
+    // The Dart task stream cancels right after the terminal event (or earlier
+    // when the app stops listening); either way nothing will be delivered on
+    // this channel again, so let the plugin forget the task.
+    onRelease(identifier)
     return nil
   }
 
-  private func cleanupObservers() {
+  /// Removes every observer this handler registered, including the finish
+  /// watch, so the task holds no closure that references this handler.
+  func cleanupObservers() {
     if let h = successHandle { task.removeObserver(withHandle: h) }
     if let h = failureHandle { task.removeObserver(withHandle: h) }
     if let h = pausedHandle { task.removeObserver(withHandle: h) }
     if let h = progressHandle { task.removeObserver(withHandle: h) }
+    for h in finishWatchHandles { task.removeObserver(withHandle: h) }
     successHandle = nil
     failureHandle = nil
     pausedHandle = nil
     progressHandle = nil
+    finishWatchHandles = []
   }
 
   private func parseTaskSnapshot(_ snapshot: StorageTaskSnapshot) -> [String: Any] {