Finish input writers to prevent dangling file descriptors

Author: cmcgee1024

Sources/Subprocess/PipeConfiguration.swift

@@ -426,7 +426,6 @@ extension PipeConfiguration {
     private func runPipeline() async throws -> CollectedResult<Output, Error> {
         // Create a pipe for standard error
         let sharedErrorPipe = try FileDescriptor.pipe()
-        let sharedErrorPipeOutput = FileDescriptorOutput(fileDescriptor: sharedErrorPipe.writeEnd, closeAfterSpawningProcess: false)
 
         return try await withThrowingTaskGroup(of: CollectedPipeResult.self, returning: CollectedResult<Output, Error>.self) { group in
             // Collect error output from all stages
@@ -465,7 +464,7 @@ extension PipeConfiguration {
                                             configuration,
                                             input: self.input,
                                             output: .fileDescriptor(writeEnd, closeAfterSpawningProcess: true),
-                                            error: sharedErrorPipeOutput
+                                            error: FileDescriptorOutput(fileDescriptor: sharedErrorPipe.writeEnd, closeAfterSpawningProcess: false)
                                         )
 
                                         taskResult = PipelineTaskResult.success(
@@ -565,6 +564,10 @@ extension PipeConfiguration {
                                             }
                                         }
 
+                                        // Close outputs in case the function did not
+                                        try await outWriter.finish()
+                                        try await errWriter.finish()
+
                                         return 0
                                     }
 
@@ -600,7 +603,7 @@ extension PipeConfiguration {
                                             configuration,
                                             input: .fileDescriptor(readEnd, closeAfterSpawningProcess: true),
                                             output: .fileDescriptor(writeEnd, closeAfterSpawningProcess: true),
-                                            error: sharedErrorPipeOutput
+                                            error: FileDescriptorOutput(fileDescriptor: sharedErrorPipe.writeEnd, closeAfterSpawningProcess: false)
                                         )
 
                                         taskResult = PipelineTaskResult.success(
@@ -678,6 +681,10 @@ extension PipeConfiguration {
                                             }
                                         }
 
+                                        // Close outputs in case the function did not
+                                        try await outWriter.finish()
+                                        try await errWriter.finish()
+
                                         return 0
                                     }
 
@@ -712,7 +719,7 @@ extension PipeConfiguration {
                                             configuration,
                                             input: .fileDescriptor(readEnd, closeAfterSpawningProcess: true),
                                             output: self.output,
-                                            error: sharedErrorPipeOutput
+                                            error: FileDescriptorOutput(fileDescriptor: sharedErrorPipe.writeEnd, closeAfterSpawningProcess: false)
                                         )
                                         return PipelineTaskResult.success(lastIndex, SendableCollectedResult(finalResult))
                                     case .replaceStdout:
@@ -844,6 +851,11 @@ extension PipeConfiguration {
                                             return (retVal, .none)
                                         }
 
+                                        // FIXME: determine how best to handle these writers so that the function doesn't finish them, and it doesn't cause deadlock
+                                        // Close outputs in case the function did not
+                                        //try await outWriter.finish()
+                                        //try await errWriter.finish()
+
                                         var exitCode: UInt32 = 0
                                         var output: Output.OutputType? = nil
                                         for try await r in group {

Tests/SubprocessTests/PipeConfigurationTests.swift

@@ -230,7 +230,7 @@ struct PipeConfigurationTests {
     }
 
     // FIXME - these function tests are hanging on Linux
-    #if os(macOS)
+    #if !os(Windows)
     @Test func testBasicSwiftFunctionBeginning() async throws {
         let config =
             pipe { input, output, error in
@@ -261,7 +261,9 @@ struct PipeConfigurationTests {
         #expect(result.standardOutput?.trimmingCharacters(in: CharacterSet.whitespacesAndNewlines) == "Hello World")
         #expect(result.terminationStatus.isSuccess)
     }
+    #endif
 
+    #if !os(Windows)
     @Test func testBasicSwiftFunctionMiddle() async throws {
         let config =
             pipe(
@@ -284,7 +286,7 @@ struct PipeConfigurationTests {
                 }
                 return 0
             } | Cat().configuration
-            |> (
+            |>  (
                 output: .string(limit: .max),
                 error: .string(limit: .max)
             )
@@ -293,7 +295,9 @@ struct PipeConfigurationTests {
         #expect(result.standardOutput?.trimmingCharacters(in: CharacterSet.whitespacesAndNewlines) == "Hello World")
         #expect(result.terminationStatus.isSuccess)
     }
+    #endif
 
+    #if !os(Windows)
     @Test func testBasicSwiftFunctionEnd() async throws {
         let config =
             pipe(
@@ -666,7 +670,7 @@ struct PipeConfigurationTests {
     }
 
     // FIXME - this test is hanging on Linux
-    #if os(macOS)
+    #if !os(Windows)
     @Test func testMultiStageSwiftFunctionPipelineWithStringInput() async throws {
         let numbers = "10\n25\n7\n42\n13\n8\n99"