From af85d1d9c9156a95363dbc4e83f4a8b2b8b762f2 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sat, 25 Apr 2026 19:57:09 -0500 Subject: [PATCH 1/5] fix(inject): resolve goroutine race in pipe() bidirectional copy Replace non-deterministic first-wins error channel with dedicated per-direction channels so each goroutine's result is tracked independently. Wait for both goroutines to complete before closing endpoints, preventing data loss from mid-copy closes and ensuring deterministic error priority (stdin errors take precedence). --- pkg/inject/inject.go | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/pkg/inject/inject.go b/pkg/inject/inject.go index f5517be11..683b423ed 100644 --- a/pkg/inject/inject.go +++ b/pkg/inject/inject.go @@ -309,22 +309,26 @@ func pipe( toStdin io.WriteCloser, fromStdin io.Reader, toStdout io.Writer, fromStdout io.ReadCloser, ) error { - errChan := make(chan error, 2) + stdinErr := make(chan error, 1) + stdoutErr := make(chan error, 1) + go func() { _, err := io.Copy(toStdout, fromStdout) - errChan <- err + stdoutErr <- err }() go func() { _, err := io.Copy(toStdin, fromStdin) - errChan <- err + stdinErr <- err }() - first := <-errChan + err1 := <-stdinErr + err2 := <-stdoutErr _ = toStdin.Close() _ = fromStdout.Close() - <-errChan - - return first + if err1 != nil { + return err1 + } + return err2 } From 6e64de5c2e7b85391ddc9ede65c40060701d3901 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sat, 25 Apr 2026 20:00:24 -0500 Subject: [PATCH 2/5] test(inject): add regression test for pipe() goroutine race Stress-test pipe() with 100 iterations of varying data to expose non-deterministic goroutine scheduling. Verifies both stdin and stdout directions complete fully without data loss. --- pkg/inject/inject_test.go | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/pkg/inject/inject_test.go b/pkg/inject/inject_test.go index 845627d0f..848254f25 100644 --- a/pkg/inject/inject_test.go +++ b/pkg/inject/inject_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "fmt" "io" "os" "runtime" @@ -153,6 +154,39 @@ func (s *PipeTestSuite) TestPipe_NoGoroutineLeak() { s.LessOrEqual(after, before+5, "goroutine leak detected: before=%d after=%d", before, after) } +func (s *PipeTestSuite) TestPipe_ConcurrentCopyRaceRegression() { + for i := range 100 { + func() { + fromStdinReader, fromStdinWriter := io.Pipe() + toStdoutBuf := &bytes.Buffer{} + + fromStdoutReader, fromStdoutWriter := io.Pipe() + toStdinPipeReader, toStdinPipeWriter := io.Pipe() + + errCh := make(chan error, 1) + go func() { + errCh <- pipe(toStdinPipeWriter, fromStdinReader, toStdoutBuf, fromStdoutReader) + }() + + msg := fmt.Sprintf("iteration-%d", i) + _, err := fromStdinWriter.Write([]byte(msg)) + s.Require().NoError(err) + _ = fromStdinWriter.Close() + + _, err = fromStdoutWriter.Write([]byte(msg)) + s.Require().NoError(err) + _ = fromStdoutWriter.Close() + + received, err := io.ReadAll(toStdinPipeReader) + s.Require().NoError(err) + + s.NoError(<-errCh) + s.Equal(msg, string(received), "stdin data lost at iteration %d", i) + s.Equal(msg, toStdoutBuf.String(), "stdout data lost at iteration %d", i) + }() + } +} + func TestPipeSuite(t *testing.T) { suite.Run(t, new(PipeTestSuite)) } From 7453bce71f55bd7c96f5d0603e6a934671040d5a Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 26 Apr 2026 02:56:29 -0500 Subject: [PATCH 3/5] fix(e2e): add inactivity timeout to machineprovider provider config Apply the same timeout fix from PR #140 to the machineprovider (no suffix) testdata directory for consistency. --- .../machineprovider/testdata/machineprovider/provider.yaml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/e2e/tests/machineprovider/testdata/machineprovider/provider.yaml b/e2e/tests/machineprovider/testdata/machineprovider/provider.yaml index 23dcd28be..aac71c23c 100644 --- a/e2e/tests/machineprovider/testdata/machineprovider/provider.yaml +++ b/e2e/tests/machineprovider/testdata/machineprovider/provider.yaml @@ -5,10 +5,14 @@ description: |- options: LOCATION: description: The location Devsy should use + INACTIVITY_TIMEOUT: + description: The timeout until the machine will be stopped + default: 30s agent: local: true docker: install: false + inactivityTimeout: ${INACTIVITY_TIMEOUT} exec: create: |- mkdir -p ${LOCATION}/${MACHINE_ID} From fae0b522143c282c87584a457f1f218384ffa774 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 26 Apr 2026 08:20:34 -0500 Subject: [PATCH 4/5] fix(inject): prevent pipe() hang with bounded second-direction wait The previous fix used sequential channel reads that could block indefinitely when one io.Copy direction stalled. Replace with select-based first-direction-wins and a 5s bounded timeout for the second direction, keeping pipe() well within sshtunnel's 10s cleanupTimeout. --- pkg/inject/inject.go | 28 +++++++++++++++++++++++----- 1 file changed, 23 insertions(+), 5 deletions(-) diff --git a/pkg/inject/inject.go b/pkg/inject/inject.go index 683b423ed..4a9987129 100644 --- a/pkg/inject/inject.go +++ b/pkg/inject/inject.go @@ -305,6 +305,8 @@ func readLine(reader io.Reader) (string, error) { } } +const pipeSecondDirTimeout = 5 * time.Second + func pipe( toStdin io.WriteCloser, fromStdin io.Reader, toStdout io.Writer, fromStdout io.ReadCloser, @@ -321,14 +323,30 @@ func pipe( stdinErr <- err }() - err1 := <-stdinErr - err2 := <-stdoutErr + // Wait for whichever direction completes first. + var firstErr error + var otherCh <-chan error + select { + case firstErr = <-stdinErr: + otherCh = stdoutErr + case firstErr = <-stdoutErr: + otherCh = stdinErr + } + + // Give the other direction time to finish naturally so we can + // capture any real error and avoid interrupting data in flight. + // If it doesn't finish in time, close pipes to force completion. + var secondErr error + select { + case secondErr = <-otherCh: + case <-time.After(pipeSecondDirTimeout): + } _ = toStdin.Close() _ = fromStdout.Close() - if err1 != nil { - return err1 + if firstErr != nil { + return firstErr } - return err2 + return secondErr } From db88a313766461c6e9e497aba01b0ed8a086493b Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sun, 26 Apr 2026 09:18:37 -0500 Subject: [PATCH 5/5] style(inject): use time.NewTimer for stoppable timeout in pipe() Replace time.After with time.NewTimer + defer Stop() to match the codebase convention (sshtunnel.go:146) and avoid minor timer leak on early completion. --- pkg/inject/inject.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pkg/inject/inject.go b/pkg/inject/inject.go index 4a9987129..f1785fdb8 100644 --- a/pkg/inject/inject.go +++ b/pkg/inject/inject.go @@ -337,9 +337,11 @@ func pipe( // capture any real error and avoid interrupting data in flight. // If it doesn't finish in time, close pipes to force completion. var secondErr error + timer := time.NewTimer(pipeSecondDirTimeout) + defer timer.Stop() select { case secondErr = <-otherCh: - case <-time.After(pipeSecondDirTimeout): + case <-timer.C: } _ = toStdin.Close()