livekit · pblazej · Jan 7, 2026 · Dec 15, 2025 · Dec 15, 2025 · Dec 15, 2025
diff --git a/.changes/dc-leak b/.changes/dc-leak
@@ -0,0 +1 @@
+patch type="fixed" "Memory leaks in data channel cancellation code"
diff --git a/.swiftlint.yml b/.swiftlint.yml
@@ -15,3 +15,10 @@ type_name:
   excluded:
     - ID
     - OS
+
+custom_rules:
+  no_manual_task_management:
+    name: "No Manual Task Management"
+    regex: "(let|var)\\s+\\w+\\s*:\\s*Task<Void,"
+    message: "Prefer AnyTaskCancellable over manually managing Task<Void, ...>. This allows for safer cleanup, similar to Combine's AnyCancellable."
+    severity: warning
diff --git a/Sources/LiveKit/Agent/Agent.swift b/Sources/LiveKit/Agent/Agent.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Chat/Message.swift b/Sources/LiveKit/Agent/Chat/Message.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Chat/Receive/MessageReceiver.swift b/Sources/LiveKit/Agent/Chat/Receive/MessageReceiver.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Chat/Receive/TranscriptionDelegateReceiver.swift b/Sources/LiveKit/Agent/Chat/Receive/TranscriptionDelegateReceiver.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Chat/Receive/TranscriptionStreamReceiver.swift b/Sources/LiveKit/Agent/Chat/Receive/TranscriptionStreamReceiver.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Chat/Send/MessageSender.swift b/Sources/LiveKit/Agent/Chat/Send/MessageSender.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Chat/Send/TextMessageSender.swift b/Sources/LiveKit/Agent/Chat/Send/TextMessageSender.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Participant+Agent.swift b/Sources/LiveKit/Agent/Participant+Agent.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Room+Agent.swift b/Sources/LiveKit/Agent/Room+Agent.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Agent/Session.swift b/Sources/LiveKit/Agent/Session.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -110,7 +110,8 @@ open class Session: ObservableObject {
 
     // MARK: - Internal state
 
-    private var waitForAgentTask: Task<Void, Swift.Error>?
+    private var tasks = Set<AnyTaskCancellable>()
+    private var waitForAgentTask: AnyTaskCancellable?
 
     // MARK: - Init
 
@@ -193,17 +194,10 @@ open class Session: ObservableObject {
                 receivers: receivers)
     }
 
-    deinit {
-        waitForAgentTask?.cancel()
-    }
-
     private func observe(room: Room) {
-        Task { [weak self] in
-            for try await _ in room.changes {
-                guard let self else { return }
-                updateAgent(in: room)
-            }
-        }
+        room.changes.subscribeOnMainActor(self) { observer, _ in
+            observer.updateAgent(in: room)
+        }.store(in: &tasks)
     }
 
     private func updateAgent(in room: Room) {
@@ -221,18 +215,25 @@ open class Session: ObservableObject {
     }
 
     private func observe(receivers: [any MessageReceiver]) {
+        let (stream, continuation) = AsyncStream.makeStream(of: ReceivedMessage.self)
+
+        // Multiple producers → single stream
         for receiver in receivers {
             Task { [weak self] in
                 do {
                     for await message in try await receiver.messages() {
-                        guard let self else { return }
-                        messagesDict.updateValue(message, forKey: message.id)
+                        continuation.yield(message)
                     }
                 } catch {
                     self?.error = .receiver(error)
                 }
-            }
+            }.cancellable().store(in: &tasks)
         }
+
+        // Single consumer
+        stream.subscribeOnMainActor(self) { observer, message in
+            observer.messagesDict.updateValue(message, forKey: message.id)
+        }.store(in: &tasks)
     }
 
     // MARK: - Lifecycle
@@ -242,7 +243,7 @@ open class Session: ObservableObject {
         guard connectionState == .disconnected else { return }
 
         error = nil
-        waitForAgentTask?.cancel()
+        waitForAgentTask = nil
 
         let timeout = options.agentConnectTimeout
 
@@ -278,7 +279,7 @@ open class Session: ObservableObject {
                     if isConnected, !agent.isConnected {
                         agent.failed(error: .timeout)
                     }
-                }
+                }.cancellable()
             }
         } catch {
             self.error = .connection(error)

diff --git a/Sources/LiveKit/Agent/SessionOptions.swift b/Sources/LiveKit/Agent/SessionOptions.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Audio/AudioDeviceModuleDelegateAdapter.swift b/Sources/LiveKit/Audio/AudioDeviceModuleDelegateAdapter.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Audio/AudioEngineObserver.swift b/Sources/LiveKit/Audio/AudioEngineObserver.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Audio/AudioSessionEngineObserver.swift b/Sources/LiveKit/Audio/AudioSessionEngineObserver.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Audio/Manager/AudioManager+ModuleType.swift b/Sources/LiveKit/Audio/Manager/AudioManager+ModuleType.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Audio/Manager/AudioManager+MuteMode.swift b/Sources/LiveKit/Audio/Manager/AudioManager+MuteMode.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Audio/Manager/AudioManager+Testing.swift b/Sources/LiveKit/Audio/Manager/AudioManager+Testing.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Audio/Manager/AudioManager.swift b/Sources/LiveKit/Audio/Manager/AudioManager.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -103,10 +103,10 @@
        // Keep this var within State so it's protected by UnfairLock
        public var localTracksCount: Int = 0
        public var remoteTracksCount: Int = 0
        public var customConfigureFunc: ConfigureAudioSessionFunc?
        public var sessionConfiguration: AudioSessionConfiguration?

        public var trackState: TrackState {
            switch (localTracksCount > 0, remoteTracksCount > 0) {
            case (true, false): .localOnly
            case (false, true): .remoteOnly

diff --git a/Sources/LiveKit/Audio/MixerEngineObserver.swift b/Sources/LiveKit/Audio/MixerEngineObserver.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/BroadcastBundleInfo.swift b/Sources/LiveKit/Broadcast/BroadcastBundleInfo.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/BroadcastManager.swift b/Sources/LiveKit/Broadcast/BroadcastManager.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/BroadcastScreenCapturer.swift b/Sources/LiveKit/Broadcast/BroadcastScreenCapturer.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/IPC/BroadcastAudioCodec.swift b/Sources/LiveKit/Broadcast/IPC/BroadcastAudioCodec.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -93,7 +93,7 @@
    }
 }

 extension AudioStreamBasicDescription: Codable {
    public func encode(to encoder: any Encoder) throws {
        var container = encoder.unkeyedContainer()
        try container.encode(mSampleRate)

diff --git a/Sources/LiveKit/Broadcast/IPC/BroadcastIPCHeader.swift b/Sources/LiveKit/Broadcast/IPC/BroadcastIPCHeader.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/IPC/BroadcastImageCodec.swift b/Sources/LiveKit/Broadcast/IPC/BroadcastImageCodec.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/IPC/BroadcastReceiver.swift b/Sources/LiveKit/Broadcast/IPC/BroadcastReceiver.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/IPC/BroadcastUploader.swift b/Sources/LiveKit/Broadcast/IPC/BroadcastUploader.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -29,6 +29,7 @@ final class BroadcastUploader: Sendable, Loggable {
     private struct State {
         var isUploadingImage = false
         var shouldUploadAudio = false
+        var messageLoopTask: AnyTaskCancellable?
     }
 
     private let state = StateSync(State())
@@ -40,8 +41,19 @@ final class BroadcastUploader: Sendable, Loggable {
 
     /// Creates an uploader with an open connection to another process.
     init(socketPath: SocketPath) async throws {
-        channel = try await IPCChannel(connectingTo: socketPath)
-        Task { try await handleIncomingMessages() }
+        let channel = try await IPCChannel(connectingTo: socketPath)
+        self.channel = channel
+
+        let messageLoopTask = channel.incomingMessages(BroadcastIPCHeader.self).subscribe(self) { observer, message in
+            observer.processMessageHeader(message.0)
+        } onFailure: { observer, error in
+            observer.log("IPCChannel returned error: \(error)")
+        }
+        state.mutate { $0.messageLoopTask = messageLoopTask }
+    }
+
+    deinit {
+        close()
     }
 
     /// Whether or not the connection to the receiver has been closed.
@@ -92,15 +104,12 @@ final class BroadcastUploader: Sendable, Loggable {
         }
     }
 
-    private func handleIncomingMessages() async throws {
-        for try await (header, _) in channel.incomingMessages(BroadcastIPCHeader.self) {
-            switch header {
-            case let .wantsAudio(wantsAudio):
-                state.mutate { $0.shouldUploadAudio = wantsAudio }
-            default:
-                log("Unhandled incoming message: \(header)", .debug)
-                continue
-            }
+    private func processMessageHeader(_ header: BroadcastIPCHeader) {
+        switch header {
+        case let .wantsAudio(wantsAudio):
+            state.mutate { $0.shouldUploadAudio = wantsAudio }
+        default:
+            log("Unhandled incoming message: \(header)", .debug)
         }
     }
 }

diff --git a/Sources/LiveKit/Broadcast/IPC/IPCChannel.swift b/Sources/LiveKit/Broadcast/IPC/IPCChannel.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/IPC/IPCProtocol.swift b/Sources/LiveKit/Broadcast/IPC/IPCProtocol.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/IPC/SocketPath.swift b/Sources/LiveKit/Broadcast/IPC/SocketPath.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/LKSampleHandler.swift b/Sources/LiveKit/Broadcast/LKSampleHandler.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Broadcast/Support/DarwinNotificationCenter.swift b/Sources/LiveKit/Broadcast/Support/DarwinNotificationCenter.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.

diff --git a/Sources/LiveKit/Convenience/AudioProcessing.swift b/Sources/LiveKit/Convenience/AudioProcessing.swift
@@ -1,5 +1,5 @@
 /*
- * Copyright 2025 LiveKit
+ * Copyright 2026 LiveKit
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		patch type="fixed" "Memory leaks in data channel cancellation code"