fix(tunnel): join tunnel goroutine on handler exit

pkg/tunnel/pipebridge.go

@@ -74,10 +74,10 @@ func awaitPair(
 
 	select {
 	case handlerErr = <-handlerChan:
-		select {
-		case tunnelErr = <-tunnelChan:
-		default:
-		}
+		cancel()
+		_ = stdoutWriter.Close()
+		_ = stdinWriter.Close()
+		tunnelErr = <-tunnelChan
 	case tunnelErr = <-tunnelChan:
 		cancel()
 		_ = stdoutWriter.Close()

pkg/tunnel/pipebridge_test.go

@@ -214,3 +214,35 @@ func TestRunPairPipesClosedAfterReturn(t *testing.T) {
 		t.Fatal("timed out waiting for write checks")
 	}
 }
+
+func TestRunPairJoinsTunnelWhenHandlerFinishesFirst(t *testing.T) {
+	t.Parallel()
+
+	pb, err := NewPipeBridge()
+	if err != nil {
+		t.Fatalf("NewPipeBridge() error = %v", err)
+	}
+	defer pb.Close()
+
+	tunnelExited := make(chan struct{})
+	tunnel := func(_ context.Context, stdin *os.File, _ *os.File) error {
+		buf := make([]byte, 1)
+		_, err := stdin.Read(buf) // blocks until stdin write end is closed
+		close(tunnelExited)
+		return err
+	}
+
+	handler := func(_ context.Context, _ *os.File, _ *os.File) error {
+		return nil
+	}
+
+	if err := pb.RunPair(context.Background(), tunnel, handler); err != nil {
+		t.Fatalf("RunPair() error = %v", err)
+	}
+
+	select {
+	case <-tunnelExited:
+	case <-time.After(2 * time.Second):
+		t.Fatal("tunnel goroutine was orphaned when handler finished first")
+	}
+}