Remove unnecessary logs from streams

.nanpa/remove-stream-logs.kdl

@@ -0,0 +1 @@
+patch type="removed" "Removed unnecessary logs from stream handlers"

Sources/LiveKit/DataStream/Incoming/IncomingStreamManager.swift

@@ -28,6 +28,8 @@
 
     /// Mapping between stream ID and descriptor for open streams.
     private var openStreams: [String: Descriptor] = [:]
+    /// Stream topics without a registered handler.
+    private var failedToOpenStreams: Set<String> = []
 
     private var byteStreamHandlers: [String: ByteStreamHandler] = [:]
     private var textStreamHandlers: [String: TextStreamHandler] = [:]
@@ -63,27 +65,28 @@
         let identity = Participant.Identity(from: identityString)
 
         guard let streamInfo = Self.streamInfo(from: header) else {
-            log("Invalid header for stream '\(header.streamID)'", .error)
             return
         }
         openStream(with: streamInfo, from: identity)
     }
 
     private func openStream(with info: StreamInfo, from identity: Participant.Identity) {
         guard openStreams[info.id] == nil else {
-            log("Stream '\(info.id)' already open", .error)
             return
         }
         guard let handler = handler(for: info) else {
-            log("Unable to find handler for stream '\(info.id)'", .warning)
+            let topic = info.topic
+            if !failedToOpenStreams.contains(topic) {
+                logger.warning("Unable to find handler for incoming stream: \(info.id), topic: \(topic), opened by: \(identity)")
+                failedToOpenStreams.insert(topic)
+            }
             return
         }
 
         var continuation: StreamReaderSource.Continuation!
         let source = StreamReaderSource {
-            $0.onTermination = { @Sendable [weak self] termination in
+            $0.onTermination = { @Sendable [weak self] _ in
                 guard let self else { return }
-                self.log("Continuation terminated: \(termination)", .debug)
                 Task { await self.closeStream(with: info.id) }
             }
             continuation = $0
@@ -94,23 +97,18 @@
             openTime: Date.timeIntervalSinceReferenceDate,
             continuation: continuation
         )
-        log("Opened stream '\(info.id)'", .debug)
         openStreams[info.id] = descriptor
 
         Task.detached {
-            do { try await handler(source, identity) }
-            catch { logger.error("Unhandled error in stream handler") }
+            try await handler(source, identity)
         }
     }
 
     /// Close the stream with the given id.
     private func closeStream(with id: String) {
         guard let descriptor = openStreams[id] else {
-            log("No descriptor for stream '\(id)'", .debug)
             return
         }
-        let openDuration = Date.timeIntervalSinceReferenceDate - descriptor.openTime
-        log("Closed stream '\(id)' (open for \(openDuration))", .debug)
         openStreams[id] = nil
     }
 
@@ -133,7 +131,6 @@
     /// Handles a data stream trailer.
     func handle(trailer: Livekit_DataStream.Trailer) {
         guard let descriptor = openStreams[trailer.streamID] else {
-            log("Received trailer for unknown stream '\(trailer.streamID)'", .warning)
             return
         }
         if let totalLength = descriptor.info.totalLength {
@@ -175,7 +172,6 @@
 
     deinit {
         guard !openStreams.isEmpty else { return }
-        log("Terminating \(openStreams.count) open stream(s)", .debug)
         for descriptor in openStreams.values {
             descriptor.continuation.finish(throwing: StreamError.terminated)
         }

Sources/LiveKit/DataStream/Outgoing/OutgoingStreamManager.swift

@@ -167,8 +167,6 @@ actor OutgoingStreamManager: Loggable {
 
         let descriptor = Descriptor(info: info)
         openStreams[info.id] = descriptor
-
-        log("Opened stream '\(info.id)'", .debug)
     }
 
     private func send(_ data: some StreamData, to id: String) async throws {
@@ -210,8 +208,6 @@ actor OutgoingStreamManager: Loggable {
 
         try await packetHandler(packet)
         openStreams[id] = nil
-
-        log("Closed stream '\(id)'", .debug)
     }
 
     // MARK: - Destination