LineSequence: throw error when max line length exceeded instead of returning partial (and potentially incorrect) data

Author: iCharlesHu

Sources/Subprocess/AsyncBufferSequence.swift

@@ -81,16 +81,18 @@ public struct AsyncBufferSequence: AsyncSequence, Sendable {
         return Iterator(diskIO: self.diskIO)
     }
 
+    // [New API: 0.0.1]
     public func lines<Encoding: _UnicodeEncoding>(
         encoding: Encoding.Type = UTF8.self,
-        bufferingPolicy: LineSequence<Encoding>.BufferingPolicy = .unbounded
+        bufferingPolicy: LineSequence<Encoding>.BufferingPolicy = .maxLineLength(128 * 1024)
     ) -> LineSequence<Encoding> {
         return LineSequence(underlying: self, encoding: encoding, bufferingPolicy: bufferingPolicy)
     }
 }
 
 // MARK: - LineSequence
 extension AsyncBufferSequence {
+    // [New API: 0.0.1]
     public struct LineSequence<Encoding: _UnicodeEncoding>: AsyncSequence, Sendable {
         public typealias Element = String
 
@@ -140,25 +142,20 @@ extension AsyncBufferSequence {
                     }
                     #else
                     // Unfortunately here we _have to_ copy the bytes out because
-                    // DisptachIO (rightfully) reuses buffer, which means `buffer.data`
+                    // DispatchIO (rightfully) reuses buffer, which means `buffer.data`
                     // has the same address on all iterations, therefore we can't directly
                     // create the result array from buffer.data
-                    let temporary = UnsafeMutableBufferPointer<Encoding.CodeUnit>.allocate(
-                        capacity: buffer.data.count
-                    )
-                    defer { temporary.deallocate() }
-                    let actualBytesCopied = buffer.data.copyBytes(
-                        to: temporary,
-                        count: buffer.data.count
-                    )
 
                     // Calculate how many CodePoint elements we have
-                    let elementCount = actualBytesCopied / MemoryLayout<Encoding.CodeUnit>.stride
+                    let elementCount = buffer.data.count / MemoryLayout<Encoding.CodeUnit>.stride
 
                     // Create array by copying from the buffer reinterpreted as CodePoint
-                    let result: Array<Encoding.CodeUnit> = Array(
-                        UnsafeBufferPointer(start: temporary.baseAddress, count: elementCount)
-                    )
+                    let result: Array<Encoding.CodeUnit> = buffer.data.withUnsafeBytes { ptr -> Array<Encoding.CodeUnit> in
+                        return Array(
+                            UnsafeBufferPointer(start: ptr.baseAddress?.assumingMemoryBound(to: Encoding.CodeUnit.self), count: elementCount)
+                        )
+                    }
+
                     #endif
                     return result.isEmpty ? nil : result
                 }
@@ -180,18 +177,38 @@ extension AsyncBufferSequence {
                 /// let formFeed        = Encoding.CodeUnit(0x0C)
                 let carriageReturn      = Encoding.CodeUnit(0x0D)
                 // carriageReturn + lineFeed
-                let newLine: Encoding.CodeUnit
-                let lineSeparator: Encoding.CodeUnit
-                let paragraphSeparator: Encoding.CodeUnit
+                let newLine1: Encoding.CodeUnit
+                let newLine2: Encoding.CodeUnit
+                let lineSeparator1: Encoding.CodeUnit
+                let lineSeparator2: Encoding.CodeUnit
+                let lineSeparator3: Encoding.CodeUnit
+                let paragraphSeparator1: Encoding.CodeUnit
+                let paragraphSeparator2: Encoding.CodeUnit
+                let paragraphSeparator3: Encoding.CodeUnit
                 switch Encoding.CodeUnit.self {
                 case is UInt8.Type:
-                    newLine             = Encoding.CodeUnit(0xC2) // 0xC2 0x85
-                    lineSeparator       = Encoding.CodeUnit(0xE2) // 0xE2 0x80 0xA8
-                    paragraphSeparator  = Encoding.CodeUnit(0xE2) // 0xE2 0x80 0xA9
+                    newLine1            = Encoding.CodeUnit(0xC2)
+                    newLine2            = Encoding.CodeUnit(0x85)
+
+                    lineSeparator1      = Encoding.CodeUnit(0xE2)
+                    lineSeparator2      = Encoding.CodeUnit(0x80)
+                    lineSeparator3      = Encoding.CodeUnit(0xA8)
+
+                    paragraphSeparator1 = Encoding.CodeUnit(0xE2)
+                    paragraphSeparator2 = Encoding.CodeUnit(0x80)
+                    paragraphSeparator3 = Encoding.CodeUnit(0xA9)
                 case is UInt16.Type, is UInt32.Type:
-                    newLine             = Encoding.CodeUnit(0x0085)
-                    lineSeparator       = Encoding.CodeUnit(0x2028)
-                    paragraphSeparator  = Encoding.CodeUnit(0x2029)
+                    // UTF16 and UTF32 use one byte for all
+                    newLine1            = Encoding.CodeUnit(0x0085)
+                    newLine2            = Encoding.CodeUnit(0x0085)
+
+                    lineSeparator1      = Encoding.CodeUnit(0x2028)
+                    lineSeparator2      = Encoding.CodeUnit(0x2028)
+                    lineSeparator3      = Encoding.CodeUnit(0x2028)
+
+                    paragraphSeparator1 = Encoding.CodeUnit(0x2029)
+                    paragraphSeparator2 = Encoding.CodeUnit(0x2029)
+                    paragraphSeparator3 = Encoding.CodeUnit(0x2029)
                 default:
                     fatalError("Unknown encoding type \(Encoding.self)")
                 }
@@ -210,10 +227,13 @@ extension AsyncBufferSequence {
                     var currentIndex: Int = self.startIndex
                     for index in self.startIndex ..< self.buffer.count {
                         currentIndex = index
-                        // Early return if we exceed max line length
+                        // Throw if we exceed max line length
                         if case .maxLineLength(let maxLength) = self.bufferingPolicy,
                            currentIndex >= maxLength {
-                            return yield(at: currentIndex)
+                            throw SubprocessError(
+                                code: .init(.streamOutputExceedsLimit(maxLength)),
+                                underlyingError: nil
+                            )
                         }
                         let byte = self.buffer[currentIndex]
                         switch byte {
@@ -232,12 +252,12 @@ extension AsyncBufferSequence {
                                 continue
                             }
                             return result
-                        case newLine:
+                        case newLine1:
                             var targetIndex = currentIndex
                             if Encoding.CodeUnit.self is UInt8.Type {
                                 // For UTF8, look for the next 0x85 byte
                                 guard (targetIndex + 1) < self.buffer.count,
-                                      self.buffer[targetIndex + 1] == Encoding.CodeUnit(0x85) else {
+                                      self.buffer[targetIndex + 1] == newLine2 else {
                                     // Not a valid new line. Keep looking
                                     continue
                                 }
@@ -248,21 +268,22 @@ extension AsyncBufferSequence {
                                 continue
                             }
                             return result
-                        case lineSeparator, paragraphSeparator:
+                        case lineSeparator1, paragraphSeparator1:
                             var targetIndex = currentIndex
                             if Encoding.CodeUnit.self is UInt8.Type {
-                                // For UTF8, look for the next 0x80 byte
+                                // For UTF8, look for the next byte
                                 guard (targetIndex + 1) < self.buffer.count,
-                                      self.buffer[targetIndex + 1] == Encoding.CodeUnit(0x80) else {
+                                      self.buffer[targetIndex + 1] == lineSeparator2 ||
+                                      self.buffer[targetIndex + 1] == paragraphSeparator2 else {
                                     // Not a valid new line. Keep looking
                                     continue
                                 }
-                                // Swallow 0x80 byte
+                                // Swallow next byte
                                 targetIndex += 1
-                                // Look for the final 0xA8 (lineSeparator) or 0xA9 (paragraphSeparator)
+                                // Look for the final byte
                                 guard (targetIndex + 1) < self.buffer.count,
-                                      (self.buffer[targetIndex + 1] == Encoding.CodeUnit(0xA8) ||
-                                       self.buffer[targetIndex + 1] == Encoding.CodeUnit(0xA9)) else {
+                                      (self.buffer[targetIndex + 1] == lineSeparator3 ||
+                                       self.buffer[targetIndex + 1] == paragraphSeparator3) else {
                                     // Not a valid new line. Keep looking
                                     continue
                                 }
@@ -308,9 +329,8 @@ extension AsyncBufferSequence.LineSequence {
         /// on the number of buffered elements (line length).
         case unbounded
         /// Impose a max buffer size (line length) limit.
-        /// When using this policy, `LineSequence` will return a line if:
-        /// - A newline character is encountered (standard behavior)
-        /// - The current line in the buffer reaches or exceeds the specified maximum length
+        /// Subprocess **will throw an error** if the number of buffered
+        /// elements (line length) exceeds the limit
         case maxLineLength(Int)
     }
 }

Sources/Subprocess/Buffer.swift

@@ -28,17 +28,17 @@ extension AsyncBufferSequence {
             return [.init(data: data)]
         }
         #else
-        // We need to keep the backingData alive while Slice is alive
+        // We need to keep the backingData alive while _ContiguousBufferView is alive
         internal let backingData: DispatchData
-        internal let data: DispatchData.Slice
+        internal let data: DispatchData._ContiguousBufferView
 
-        internal init(data: DispatchData.Slice, backingData: DispatchData) {
+        internal init(data: DispatchData._ContiguousBufferView, backingData: DispatchData) {
             self.data = data
             self.backingData = backingData
         }
 
         internal static func createFrom(_ data: DispatchData) -> [Buffer] {
-            let slices = data.slices
+            let slices = data.contiguousBufferViews
             // In most (all?) cases data should only have one slice
             if _fastPath(slices.count == 1) {
                 return [.init(data: slices[0], backingData: data)]
@@ -114,7 +114,7 @@ extension AsyncBufferSequence.Buffer: Equatable, Hashable {
 #if canImport(Darwin) || canImport(Glibc) || canImport(Android) || canImport(Musl)
 extension DispatchData {
     /// Unfortunately `DispatchData.Region` is not available on Linux, hence our own wrapper
-    internal struct Slice: @unchecked Sendable, RandomAccessCollection {
+    internal struct _ContiguousBufferView: @unchecked Sendable, RandomAccessCollection {
         typealias Element = UInt8
 
         internal let bytes: UnsafeBufferPointer<UInt8>
@@ -130,24 +130,17 @@ extension DispatchData {
             return try body(UnsafeRawBufferPointer(self.bytes))
         }
 
-        @discardableResult
-        internal func copyBytes<DestinationType>(
-            to ptr: UnsafeMutableBufferPointer<DestinationType>, count: Int
-        ) -> Int {
-            self.bytes.copyBytes(to: ptr, count: count)
-        }
-
         subscript(position: Int) -> UInt8 {
             _read {
                 yield self.bytes[position]
             }
         }
     }
 
-    internal var slices: [Slice] {
-        var slices = [Slice]()
+    internal var contiguousBufferViews: [_ContiguousBufferView] {
+        var slices = [_ContiguousBufferView]()
         enumerateBytes { (bytes, index, stop) in
-            slices.append(Slice(bytes: bytes))
+            slices.append(_ContiguousBufferView(bytes: bytes))
         }
         return slices
     }

Sources/Subprocess/Error.swift

@@ -40,6 +40,7 @@ extension SubprocessError {
             case failedToReadFromSubprocess
             case failedToWriteToSubprocess
             case failedToMonitorProcess
+            case streamOutputExceedsLimit(Int)
             // Signal
             case failedToSendSignal(Int32)
             // Windows Only
@@ -64,18 +65,20 @@ extension SubprocessError {
                 return 4
             case .failedToMonitorProcess:
                 return 5
-            case .failedToSendSignal(_):
+            case .streamOutputExceedsLimit(_):
                 return 6
-            case .failedToTerminate:
+            case .failedToSendSignal(_):
                 return 7
-            case .failedToSuspend:
+            case .failedToTerminate:
                 return 8
-            case .failedToResume:
+            case .failedToSuspend:
                 return 9
-            case .failedToCreatePipe:
+            case .failedToResume:
                 return 10
-            case .invalidWindowsPath(_):
+            case .failedToCreatePipe:
                 return 11
+            case .invalidWindowsPath(_):
+                return 12
             }
         }
 
@@ -103,6 +106,8 @@ extension SubprocessError: CustomStringConvertible, CustomDebugStringConvertible
             return "Failed to write bytes to the child process."
         case .failedToMonitorProcess:
             return "Failed to monitor the state of child process with underlying error: \(self.underlyingError!)"
+        case .streamOutputExceedsLimit(let limit):
+            return "Failed to create output from current buffer because the output limit (\(limit)) was reached."
         case .failedToSendSignal(let signal):
             return "Failed to send signal \(signal) to the child process."
         case .failedToTerminate:

Sources/Subprocess/IO/Output.swift

@@ -144,14 +144,10 @@ public struct BytesOutput: OutputProtocol {
     internal func captureOutput(
         from diskIO: consuming TrackedPlatformDiskIO?
     ) async throws -> [UInt8] {
-        guard let diskIO = diskIO else {
-            // Show not happen due to type system constraints
-            fatalError("Trying to capture output without file descriptor")
-        }
         #if os(Windows)
-        return try await diskIO.fileDescriptor.read(upToLength: self.maxSize) ?? []
+        return try await diskIO?.fileDescriptor.read(upToLength: self.maxSize) ?? []
         #else
-        let result = try await diskIO.dispatchIO.read(upToLength: self.maxSize)
+        let result = try await diskIO!.dispatchIO.read(upToLength: self.maxSize)
         return result?.array() ?? []
         #endif
     }
@@ -270,16 +266,11 @@ extension OutputProtocol {
             return () as! OutputType
         }
 
-        guard let diskIO = diskIO else {
-            // Show not happen due to type system constraints
-            fatalError("Trying to capture output without file descriptor")
-        }
-
         #if os(Windows)
-        let result = try await diskIO.fileDescriptor.read(upToLength: self.maxSize)
+        let result = try await diskIO?.fileDescriptor.read(upToLength: self.maxSize)
         return try self.output(from: result ?? [])
         #else
-        let result = try await diskIO.dispatchIO.read(upToLength: self.maxSize)
+        let result = try await diskIO!.dispatchIO.read(upToLength: self.maxSize)
         return try self.output(from: result ?? .empty)
         #endif
     }

Sources/Subprocess/Teardown.swift

@@ -85,7 +85,7 @@ extension Execution {
         using sequence: some Sequence<TeardownStep> & Sendable,
         on processIdentifier: ProcessIdentifier
     ) async {
-        await withUncanceledTask {
+        await withUncancelledTask {
             await Self.runTeardownSequence(sequence, on: processIdentifier)
         }
     }
@@ -222,7 +222,7 @@ extension Execution {
     }
 }
 
-func withUncanceledTask<Result: Sendable>(
+func withUncancelledTask<Result: Sendable>(
     returning: Result.Type = Result.self,
     _ body: @Sendable @escaping () async -> Result
 ) async -> Result {

Tests/SubprocessTests/SubprocessTests+Unix.swift

@@ -980,7 +980,7 @@ extension SubprocessUnixTests {
         // Generate test cases
         func generateString(size: TestCaseSize) -> [UInt8] {
             // Basic Latin has the range U+0020 ... U+007E
-            let range: ClosedRange<UInt8> = 0x20 ... 0x7E // 0x4E ... 0x5A
+            let range: ClosedRange<UInt8> = 0x20 ... 0x7E
 
             let length: Int
             switch size {
@@ -1044,7 +1044,7 @@ extension SubprocessUnixTests {
         }
 
         let testCaseCount = 60
-        let testFilePath = URL.temporaryDirectory.appending(path: "NewLines.txt")
+        let testFilePath = URL.temporaryDirectory.appending(path: "NewLines-\(UUID().uuidString).txt")
         if FileManager.default.fileExists(atPath: testFilePath.path()) {
             try FileManager.default.removeItem(at: testFilePath)
         }
@@ -1070,6 +1070,7 @@ extension SubprocessUnixTests {
                 index += 1
             }
         }
+        try FileManager.default.removeItem(at: testFilePath)
     }
 }