From a44f62c6599b9dafcf08e0140f1eef704b9cf3ab Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 7 Sep 2026 21:30:52 +0000 Subject: [PATCH 1/4] fix(transport): arbitrate managed lifecycle results Bounded join and semantic result arbitration in RunManaged prevent teardown-only errors from overriding successful commands. --- pkg/transport/lifecycle.go | 178 ++++++++++++++++-- pkg/transport/lifecycle_test.go | 307 ++++++++++++++++++++++++++++++++ 2 files changed, 474 insertions(+), 11 deletions(-) diff --git a/pkg/transport/lifecycle.go b/pkg/transport/lifecycle.go index 3a6331336..0358ac67e 100644 --- a/pkg/transport/lifecycle.go +++ b/pkg/transport/lifecycle.go @@ -4,7 +4,10 @@ import ( "context" "errors" "io" + "net" + "strings" "sync" + "time" "github.com/devsy-org/devsy/pkg/log" ) @@ -38,6 +41,7 @@ const ( TransportSideSSH = SideSSH TransportSideParent = SideParent ) +const DefaultJoinTimeout = 5 * time.Second const ( TransportCloseUnknown = CloseUnknown @@ -153,6 +157,153 @@ type RunManagedOptions struct { Handler func(context.Context) error Metadata LogMetadata TransportSide Side + JoinTimeout time.Duration +} + +type managedOutcome struct { + firstSide Side + parentErr error + handlerErr error + transportErr error + handlerCompleted bool + transportCompleted bool +} + +func isTeardownOrCancellationError(err error) bool { + if err == nil { + return false + } + return isCancellationErr(err) || isClosedOrEOFErr(err) +} + +func isCancellationErr(err error) bool { + if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return true + } + return strings.Contains(err.Error(), "context canceled") +} + +func isClosedOrEOFErr(err error) bool { + return isEOFErr(err) || isClosedNetErr(err) +} + +func isEOFErr(err error) bool { + if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) || + errors.Is(err, io.ErrClosedPipe) { + return true + } + msg := err.Error() + return strings.Contains(msg, ": EOF") || strings.Contains(msg, "closed pipe") +} + +func isClosedNetErr(err error) bool { + if errors.Is(err, net.ErrClosed) { + return true + } + msg := err.Error() + return strings.Contains(msg, "closed network connection") || + strings.Contains(msg, "use of closed network connection") || + strings.Contains(msg, "connection is closed") +} + +func resolveManagedErrors(outcome managedOutcome) error { + if outcome.firstSide == SideParent { + return outcome.parentErr + } + if outcome.handlerCompleted && outcome.handlerErr == nil { + return nil + } + if err := resolveParentCancellation(outcome); err != nil { + return err + } + if outcome.firstSide == SideSSH { + return outcome.handlerErr + } + if outcome.firstSide == SideProvider { + return resolveProviderFirst(outcome) + } + return resolveFallback(outcome) +} + +func resolveParentCancellation(outcome managedOutcome) error { + if outcome.parentErr != nil && + (errors.Is(outcome.parentErr, context.Canceled) || errors.Is(outcome.parentErr, context.DeadlineExceeded)) { + if !outcome.handlerCompleted || isTeardownOrCancellationError(outcome.handlerErr) { + return outcome.parentErr + } + } + return nil +} + +func resolveProviderFirst(outcome managedOutcome) error { + if !outcome.handlerCompleted || isTeardownOrCancellationError(outcome.handlerErr) { + return outcome.transportErr + } + return outcome.handlerErr +} + +func resolveFallback(outcome managedOutcome) error { + if outcome.handlerCompleted && outcome.handlerErr != nil { + return outcome.handlerErr + } + if outcome.transportCompleted && outcome.transportErr != nil { + return outcome.transportErr + } + return outcome.parentErr +} + +func waitForFirst( + parent context.Context, + transportSide Side, + handlerDone <-chan error, + connDone <-chan error, +) (managedOutcome, error) { + var outcome managedOutcome + select { + case err := <-connDone: + outcome.firstSide = transportSide + outcome.transportErr = err + outcome.transportCompleted = true + return outcome, err + case err := <-handlerDone: + outcome.firstSide = SideSSH + outcome.handlerErr = err + outcome.handlerCompleted = true + return outcome, err + case <-parent.Done(): + outcome.firstSide = SideParent + return outcome, parent.Err() + } +} + +func initiateTeardown(conn ManagedConn, firstSide Side, handlerErr error) { + if firstSide == SideSSH && handlerErr == nil { + if cw, ok := conn.(CloseWriter); ok { + _ = cw.CloseWrite() + return + } + } + _ = conn.Close() +} + +func joinRemaining( + joinCtx context.Context, + handlerDone <-chan error, + connDone <-chan error, + outcome *managedOutcome, +) { + for !outcome.handlerCompleted || !outcome.transportCompleted { + select { + case err := <-handlerDone: + outcome.handlerErr = err + outcome.handlerCompleted = true + case err := <-connDone: + outcome.transportErr = err + outcome.transportCompleted = true + case <-joinCtx.Done(): + return + } + } } func RunManaged(opts RunManagedOptions) error { @@ -165,23 +316,28 @@ func RunManaged(opts RunManagedOptions) error { if opts.Handler == nil { return errors.New("handler is required") } + + joinTimeout := opts.JoinTimeout + if joinTimeout <= 0 { + joinTimeout = DefaultJoinTimeout + } + lifecycle, ctx := NewPersistentLifecycle(opts.Parent) handlerDone := make(chan error, 1) go func() { handlerDone <- opts.Handler(ctx) }() connDone := make(chan error, 1) go func() { connDone <- opts.Conn.Wait() }() - var info CloseInfo - select { - case err := <-connDone: - info = Classify(opts.TransportSide, err, opts.Parent) - case err := <-handlerDone: - info = Classify(SideSSH, err, opts.Parent) - case <-opts.Parent.Done(): - info = Classify(SideParent, opts.Parent.Err(), opts.Parent) - } - lifecycle.Close(info) + outcome, firstErr := waitForFirst(opts.Parent, opts.TransportSide, handlerDone, connDone) + lifecycle.Close(Classify(outcome.firstSide, firstErr, opts.Parent)) + + initiateTeardown(opts.Conn, outcome.firstSide, outcome.handlerErr) + joinCtx, cancelJoin := context.WithTimeout(context.WithoutCancel(opts.Parent), joinTimeout) + defer cancelJoin() + joinRemaining(joinCtx, handlerDone, connDone, &outcome) + _ = opts.Conn.Close() LogClose(lifecycle.CloseInfo(), opts.Metadata) - return lifecycle.CloseInfo().Err + outcome.parentErr = opts.Parent.Err() + return resolveManagedErrors(outcome) } diff --git a/pkg/transport/lifecycle_test.go b/pkg/transport/lifecycle_test.go index 7de355db2..47fa248c7 100644 --- a/pkg/transport/lifecycle_test.go +++ b/pkg/transport/lifecycle_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "net" + "sync" "testing" "time" ) @@ -111,6 +112,312 @@ func TestRunManagedReturnsParentCancellation(t *testing.T) { } } +type controlledManagedConn struct { + net.Conn + mu sync.Mutex + triggerOnce sync.Once + waitErr error + triggerWait chan struct{} + waitResultPublished chan struct{} + closed chan struct{} +} + +func newControlledManagedConn() *controlledManagedConn { + return &controlledManagedConn{ + Conn: &stubConn{}, + triggerWait: make(chan struct{}), + waitResultPublished: make(chan struct{}), + closed: make(chan struct{}), + } +} + +func (c *controlledManagedConn) Wait() error { + <-c.triggerWait + close(c.waitResultPublished) + c.mu.Lock() + defer c.mu.Unlock() + return c.waitErr +} + +func (c *controlledManagedConn) Close() error { + select { + case <-c.closed: + default: + close(c.closed) + c.triggerOnce.Do(func() { close(c.triggerWait) }) + } + return nil +} + +func (c *controlledManagedConn) TriggerCleanupError(err error) { + c.mu.Lock() + c.waitErr = err + c.mu.Unlock() + c.triggerOnce.Do(func() { close(c.triggerWait) }) +} + +func TestRunManagedHandlerSuccessBeatsCleanupError(t *testing.T) { + conn := newControlledManagedConn() + errTeardown := errors.New("wait: remote command exited without exit status or exit signal") + + err := RunManaged(RunManagedOptions{ + Parent: context.Background(), + Conn: conn, + TransportSide: SideProvider, + Handler: func(ctx context.Context) error { + conn.TriggerCleanupError(errTeardown) + <-conn.waitResultPublished + time.Sleep(5 * time.Millisecond) + return nil + }, + }) + if err != nil { + t.Fatalf("RunManaged() = %v, want nil", err) + } +} + +func TestRunManagedHandlerErrorWinsOverCleanupError(t *testing.T) { + conn := newControlledManagedConn() + errTeardown := errors.New("wait: remote command exited without exit status or exit signal") + wantErr := errors.New("command exited with status 127") + + err := RunManaged(RunManagedOptions{ + Parent: context.Background(), + Conn: conn, + TransportSide: SideProvider, + Handler: func(ctx context.Context) error { + conn.TriggerCleanupError(errTeardown) + <-conn.waitResultPublished + time.Sleep(5 * time.Millisecond) + return wantErr + }, + }) + if !errors.Is(err, wantErr) { + t.Fatalf("RunManaged() = %v, want %v", err, wantErr) + } +} + +func TestRunManagedGenuineTransportFailure(t *testing.T) { + conn := newControlledManagedConn() + providerErr := errors.New("connection reset by peer") + conn.TriggerCleanupError(providerErr) + + err := RunManaged(RunManagedOptions{ + Parent: context.Background(), + Conn: conn, + TransportSide: SideProvider, + Handler: func(ctx context.Context) error { + <-ctx.Done() + return ctx.Err() + }, + }) + if !errors.Is(err, providerErr) { + t.Fatalf("RunManaged() = %v, want %v", err, providerErr) + } +} + +func TestRunManagedStressHandlerSuccessBeatsCleanup(t *testing.T) { + errTeardown := errors.New("wait: remote command exited without exit status or exit signal") + for i := range 100 { + conn := newControlledManagedConn() + err := RunManaged(RunManagedOptions{ + Parent: context.Background(), + Conn: conn, + TransportSide: SideProvider, + Handler: func(ctx context.Context) error { + conn.TriggerCleanupError(errTeardown) + <-conn.waitResultPublished + return nil + }, + }) + if err != nil { + t.Fatalf("iteration %d: RunManaged() = %v, want nil", i, err) + } + } +} + +func TestRunManagedBoundedSecondSideShutdown(t *testing.T) { + conn := newTestManagedConn(nil) + handlerStuck := make(chan struct{}) + defer close(handlerStuck) + + done := make(chan error, 1) + start := time.Now() + go func() { + done <- RunManaged(RunManagedOptions{ + Parent: context.Background(), + Conn: conn, + TransportSide: SideProvider, + JoinTimeout: 50 * time.Millisecond, + Handler: func(ctx context.Context) error { + <-handlerStuck + return nil + }, + }) + }() + + _ = conn.Close() + + select { + case <-done: + if time.Since(start) > 2*time.Second { + t.Fatal("RunManaged took too long to return after bounded join timeout") + } + case <-time.After(2 * time.Second): + t.Fatal("RunManaged hung waiting for second side") + } +} + +func TestResolveManagedErrors_SuccessAndCancellation(t *testing.T) { + errTeardown := errors.New("wait: remote command exited without exit status or exit signal") + errCanceled := context.Canceled + + tests := []struct { + name string + outcome managedOutcome + wantErr error + }{ + { + name: "parent cancelled first wins", + outcome: managedOutcome{ + firstSide: SideParent, + parentErr: errCanceled, + handlerErr: errCanceled, + transportErr: errTeardown, + handlerCompleted: true, + transportCompleted: true, + }, + wantErr: errCanceled, + }, + { + name: "handler success beats transport teardown when transport was first", + outcome: managedOutcome{ + firstSide: SideProvider, + handlerErr: nil, + transportErr: errTeardown, + handlerCompleted: true, + transportCompleted: true, + }, + wantErr: nil, + }, + { + name: "handler success beats transport teardown when handler was first", + outcome: managedOutcome{ + firstSide: SideSSH, + handlerErr: nil, + transportErr: errTeardown, + handlerCompleted: true, + transportCompleted: true, + }, + wantErr: nil, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := resolveManagedErrors(tt.outcome) + if !errors.Is(got, tt.wantErr) { + t.Fatalf("resolveManagedErrors() = %v, want %v", got, tt.wantErr) + } + }) + } +} + +func TestResolveManagedErrors_HandlerError(t *testing.T) { + errTeardown := errors.New("wait: remote command exited without exit status or exit signal") + errUser := errors.New("command exited with status 127") + + tests := []struct { + name string + outcome managedOutcome + wantErr error + }{ + { + name: "handler error beats transport teardown when handler was first", + outcome: managedOutcome{ + firstSide: SideSSH, + handlerErr: errUser, + transportErr: errTeardown, + handlerCompleted: true, + transportCompleted: true, + }, + wantErr: errUser, + }, + { + name: "handler error beats transport teardown when transport was first", + outcome: managedOutcome{ + firstSide: SideProvider, + handlerErr: errUser, + transportErr: errTeardown, + handlerCompleted: true, + transportCompleted: true, + }, + wantErr: errUser, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := resolveManagedErrors(tt.outcome) + if !errors.Is(got, tt.wantErr) { + t.Fatalf("resolveManagedErrors() = %v, want %v", got, tt.wantErr) + } + }) + } +} + +func TestResolveManagedErrors_ProviderFailure(t *testing.T) { + errProvider := errors.New("provider reset") + errCanceled := context.Canceled + + tests := []struct { + name string + outcome managedOutcome + wantErr error + }{ + { + name: "genuine provider failure wins over handler cancellation consequence", + outcome: managedOutcome{ + firstSide: SideProvider, + handlerErr: errCanceled, + transportErr: errProvider, + handlerCompleted: true, + transportCompleted: true, + }, + wantErr: errProvider, + }, + { + name: "genuine provider failure wins when handler timed out", + outcome: managedOutcome{ + firstSide: SideProvider, + transportErr: errProvider, + handlerCompleted: false, + transportCompleted: true, + }, + wantErr: errProvider, + }, + { + name: "clean provider exit with handler EOF returns nil", + outcome: managedOutcome{ + firstSide: SideProvider, + handlerErr: errors.New("read: EOF"), + handlerCompleted: true, + transportCompleted: true, + }, + wantErr: nil, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := resolveManagedErrors(tt.outcome) + if !errors.Is(got, tt.wantErr) { + t.Fatalf("resolveManagedErrors() = %v, want %v", got, tt.wantErr) + } + }) + } +} + type stubConn struct{} func (*stubConn) Read([]byte) (int, error) { return 0, net.ErrClosed } From b37884cdf2e08c101de2521f7eb9214d92076694 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Mon, 7 Sep 2026 23:41:42 +0000 Subject: [PATCH 2/4] fix(transport): preserve genuine provider failures and handle half-close fallback Narrow handler-success precedence so genuine provider failures are not suppressed by later handler nil results. Fall back to Close() immediately when CloseWrite() fails or is unsupported. Replace sleep-based ordering in tests with deterministic context synchronization and add regression tests. --- pkg/transport/lifecycle.go | 50 +++++++++++++++++++--- pkg/transport/lifecycle_test.go | 73 +++++++++++++++++++++++++++++++-- 2 files changed, 113 insertions(+), 10 deletions(-) diff --git a/pkg/transport/lifecycle.go b/pkg/transport/lifecycle.go index 0358ac67e..12128b324 100644 --- a/pkg/transport/lifecycle.go +++ b/pkg/transport/lifecycle.go @@ -206,23 +206,60 @@ func isClosedNetErr(err error) bool { strings.Contains(msg, "connection is closed") } +func isExitMissingErr(err error) bool { + if err == nil { + return false + } + return strings.Contains(err.Error(), "remote command exited without exit status or exit signal") +} + +func isTransportTeardownError(err error) bool { + if err == nil { + return true + } + return isTeardownOrCancellationError(err) || isExitMissingErr(err) +} + +func isMeaningfulHandlerErr(outcome managedOutcome) bool { + return outcome.handlerCompleted && outcome.handlerErr != nil && + !isTeardownOrCancellationError(outcome.handlerErr) +} + +func isGenuineProviderFailure(outcome managedOutcome) bool { + if !outcome.transportCompleted || outcome.transportErr == nil { + return false + } + return outcome.firstSide == SideProvider && !isTransportTeardownError(outcome.transportErr) +} + func resolveManagedErrors(outcome managedOutcome) error { if outcome.firstSide == SideParent { return outcome.parentErr } + if isMeaningfulHandlerErr(outcome) { + return outcome.handlerErr + } + if isGenuineProviderFailure(outcome) { + return outcome.transportErr + } if outcome.handlerCompleted && outcome.handlerErr == nil { return nil } if err := resolveParentCancellation(outcome); err != nil { return err } - if outcome.firstSide == SideSSH { + return resolveByFirstSide(outcome) +} + +func resolveByFirstSide(outcome managedOutcome) error { + switch outcome.firstSide { + case SideSSH: return outcome.handlerErr - } - if outcome.firstSide == SideProvider { + case SideProvider: return resolveProviderFirst(outcome) + default: + return resolveFallback(outcome) } - return resolveFallback(outcome) } func resolveParentCancellation(outcome managedOutcome) error { @@ -279,8 +316,9 @@ func waitForFirst( func initiateTeardown(conn ManagedConn, firstSide Side, handlerErr error) { if firstSide == SideSSH && handlerErr == nil { if cw, ok := conn.(CloseWriter); ok { - _ = cw.CloseWrite() - return + if err := cw.CloseWrite(); err == nil { + return + } } } _ = conn.Close() diff --git a/pkg/transport/lifecycle_test.go b/pkg/transport/lifecycle_test.go index 47fa248c7..d1aed3a31 100644 --- a/pkg/transport/lifecycle_test.go +++ b/pkg/transport/lifecycle_test.go @@ -166,8 +166,7 @@ func TestRunManagedHandlerSuccessBeatsCleanupError(t *testing.T) { TransportSide: SideProvider, Handler: func(ctx context.Context) error { conn.TriggerCleanupError(errTeardown) - <-conn.waitResultPublished - time.Sleep(5 * time.Millisecond) + <-ctx.Done() return nil }, }) @@ -187,8 +186,7 @@ func TestRunManagedHandlerErrorWinsOverCleanupError(t *testing.T) { TransportSide: SideProvider, Handler: func(ctx context.Context) error { conn.TriggerCleanupError(errTeardown) - <-conn.waitResultPublished - time.Sleep(5 * time.Millisecond) + <-ctx.Done() return wantErr }, }) @@ -197,6 +195,25 @@ func TestRunManagedHandlerErrorWinsOverCleanupError(t *testing.T) { } } +func TestRunManagedProviderFailureWinsOverLaterHandlerSuccess(t *testing.T) { + conn := newControlledManagedConn() + providerErr := errors.New("connection reset by peer") + + err := RunManaged(RunManagedOptions{ + Parent: context.Background(), + Conn: conn, + TransportSide: SideProvider, + Handler: func(ctx context.Context) error { + conn.TriggerCleanupError(providerErr) + <-ctx.Done() + return nil + }, + }) + if !errors.Is(err, providerErr) { + t.Fatalf("RunManaged() = %v, want %v", err, providerErr) + } +} + func TestRunManagedGenuineTransportFailure(t *testing.T) { conn := newControlledManagedConn() providerErr := errors.New("connection reset by peer") @@ -268,6 +285,43 @@ func TestRunManagedBoundedSecondSideShutdown(t *testing.T) { } } +type closeWriteFailingConn struct { + *testManagedConn + closeWriteCalled bool +} + +func (c *closeWriteFailingConn) CloseWrite() error { + c.closeWriteCalled = true + return errors.ErrUnsupported +} + +func TestRunManagedCloseWriteFailureFallsBackToClose(t *testing.T) { + conn := &closeWriteFailingConn{ + testManagedConn: newTestManagedConn(nil), + } + + err := RunManaged(RunManagedOptions{ + Parent: context.Background(), + Conn: conn, + TransportSide: SideProvider, + JoinTimeout: 2 * time.Second, + Handler: func(ctx context.Context) error { + return nil + }, + }) + if err != nil { + t.Fatalf("RunManaged() = %v, want nil", err) + } + if !conn.closeWriteCalled { + t.Fatal("CloseWrite was not called") + } + select { + case <-conn.closed: + default: + t.Fatal("Close was not called after CloseWrite failed") + } +} + func TestResolveManagedErrors_SuccessAndCancellation(t *testing.T) { errTeardown := errors.New("wait: remote command exited without exit status or exit signal") errCanceled := context.Canceled @@ -396,6 +450,17 @@ func TestResolveManagedErrors_ProviderFailure(t *testing.T) { }, wantErr: errProvider, }, + { + name: "genuine provider failure wins over later handler success", + outcome: managedOutcome{ + firstSide: SideProvider, + handlerErr: nil, + transportErr: errProvider, + handlerCompleted: true, + transportCompleted: true, + }, + wantErr: errProvider, + }, { name: "clean provider exit with handler EOF returns nil", outcome: managedOutcome{ From f415b05f3b5930c2365dfd3b958734ba543220cc Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 8 Sep 2026 03:06:29 +0000 Subject: [PATCH 3/4] refactor(transport): use typed ExitMissingError classification Classify ExitMissingError using errors.As() while retaining fallback text matching. --- pkg/transport/lifecycle.go | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/pkg/transport/lifecycle.go b/pkg/transport/lifecycle.go index 12128b324..478b0ce20 100644 --- a/pkg/transport/lifecycle.go +++ b/pkg/transport/lifecycle.go @@ -10,6 +10,7 @@ import ( "time" "github.com/devsy-org/devsy/pkg/log" + "golang.org/x/crypto/ssh" ) type CloseReason string @@ -210,6 +211,10 @@ func isExitMissingErr(err error) bool { if err == nil { return false } + var exitMissing *ssh.ExitMissingError + if errors.As(err, &exitMissing) { + return true + } return strings.Contains(err.Error(), "remote command exited without exit status or exit signal") } From a93f33a45cd43e98195773768c293c5ab994ea22 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Tue, 8 Sep 2026 03:29:23 +0000 Subject: [PATCH 4/4] fix(transport): handle teardown errors in SideSSH resolution Symmetrically classify teardown-noise handler errors in resolveSSHFirst. Narrow isClosedNetErr string fallback to 'use of closed network connection'. Guard c.closed with sync.Once in test double controlledManagedConn.Close. Wrap io.EOF in test fixture and add uncompleted SideSSH test case. --- pkg/transport/lifecycle.go | 17 ++++++++++++----- pkg/transport/lifecycle_test.go | 21 ++++++++++++++++----- 2 files changed, 28 insertions(+), 10 deletions(-) diff --git a/pkg/transport/lifecycle.go b/pkg/transport/lifecycle.go index 478b0ce20..58c0128cb 100644 --- a/pkg/transport/lifecycle.go +++ b/pkg/transport/lifecycle.go @@ -201,10 +201,7 @@ func isClosedNetErr(err error) bool { if errors.Is(err, net.ErrClosed) { return true } - msg := err.Error() - return strings.Contains(msg, "closed network connection") || - strings.Contains(msg, "use of closed network connection") || - strings.Contains(msg, "connection is closed") + return strings.Contains(err.Error(), "use of closed network connection") } func isExitMissingErr(err error) bool { @@ -259,7 +256,7 @@ func resolveManagedErrors(outcome managedOutcome) error { func resolveByFirstSide(outcome managedOutcome) error { switch outcome.firstSide { case SideSSH: - return outcome.handlerErr + return resolveSSHFirst(outcome) case SideProvider: return resolveProviderFirst(outcome) default: @@ -267,6 +264,16 @@ func resolveByFirstSide(outcome managedOutcome) error { } } +func resolveSSHFirst(outcome managedOutcome) error { + if !outcome.handlerCompleted || isTeardownOrCancellationError(outcome.handlerErr) { + if outcome.transportErr != nil { + return outcome.transportErr + } + return nil + } + return outcome.handlerErr +} + func resolveParentCancellation(outcome managedOutcome) error { if outcome.parentErr != nil && (errors.Is(outcome.parentErr, context.Canceled) || errors.Is(outcome.parentErr, context.DeadlineExceeded)) { diff --git a/pkg/transport/lifecycle_test.go b/pkg/transport/lifecycle_test.go index d1aed3a31..3540819f9 100644 --- a/pkg/transport/lifecycle_test.go +++ b/pkg/transport/lifecycle_test.go @@ -3,6 +3,8 @@ package transport import ( "context" "errors" + "fmt" + "io" "net" "sync" "testing" @@ -116,6 +118,7 @@ type controlledManagedConn struct { net.Conn mu sync.Mutex triggerOnce sync.Once + closeOnce sync.Once waitErr error triggerWait chan struct{} waitResultPublished chan struct{} @@ -140,12 +143,10 @@ func (c *controlledManagedConn) Wait() error { } func (c *controlledManagedConn) Close() error { - select { - case <-c.closed: - default: + c.closeOnce.Do(func() { close(c.closed) c.triggerOnce.Do(func() { close(c.triggerWait) }) - } + }) return nil } @@ -408,6 +409,16 @@ func TestResolveManagedErrors_HandlerError(t *testing.T) { }, wantErr: errUser, }, + { + name: "transport error returned when handler did not complete in SideSSH", + outcome: managedOutcome{ + firstSide: SideSSH, + transportErr: errTeardown, + handlerCompleted: false, + transportCompleted: true, + }, + wantErr: errTeardown, + }, } for _, tt := range tests { @@ -465,7 +476,7 @@ func TestResolveManagedErrors_ProviderFailure(t *testing.T) { name: "clean provider exit with handler EOF returns nil", outcome: managedOutcome{ firstSide: SideProvider, - handlerErr: errors.New("read: EOF"), + handlerErr: fmt.Errorf("read: %w", io.EOF), handlerCompleted: true, transportCompleted: true, },