fix: Leaking yamux session after HTTP handler is closed (#329)

* fix: Leaking yamux session after HTTP handler is closed

Closes #317. The httptest server cancels the context after the connection
is closed, but if a connection takes a long time to close, the request
would never end. This applies a context to the entire listener that cancels
on test cleanup.

After discussion with @bryphe-coder, reducing the parallel limit on
Windows is likely to reduce failures as well.

* Switch to windows-2022 to improve decompression

* Invalidate cache on matrix OS

.github/workflows/coder.yaml

@@ -122,7 +122,7 @@ jobs:
         os:
           - ubuntu-latest
           - macos-latest
-          - windows-latest
+          - windows-2022
     steps:
       - uses: actions/checkout@v2
 
@@ -138,9 +138,9 @@ jobs:
             ~/.cache/go-build
             ~/Library/Caches/go-build
             %LocalAppData%\go-build
-          key: ${{ runner.os }}-go-${{ hashFiles('**/go.sum') }}
+          key: ${{ matrix.os }}-go-${{ hashFiles('**/go.sum') }}
           restore-keys: |
-            ${{ runner.os }}-go-
+            ${{ matrix.os }}-go-
 
       - run: go install gotest.tools/gotestsum@latest
 
@@ -150,9 +150,13 @@ jobs:
           terraform_wrapper: false
 
       - name: Test with Mock Database
+        shell: bash
+        env:
+          GOCOUNT: ${{ runner.os == 'Windows' && 3 || 5 }}
+          GOMAXPROCS: ${{ runner.os == 'Windows' && 1 || 2 }}
         run: gotestsum --junitfile="gotests.xml" --packages="./..." --
           -covermode=atomic -coverprofile="gotests.coverage"
-          -timeout=3m -count=5 -race -short -parallel=2
+          -timeout=3m -count=$GOCOUNT -race -short -failfast
 
       - name: Upload DataDog Trace
         if: (success() || failure()) && github.actor != 'dependabot[bot]'
@@ -166,10 +170,10 @@ jobs:
         if: runner.os == 'Linux'
         run: DB=true gotestsum --junitfile="gotests.xml" --packages="./..." --
           -covermode=atomic -coverprofile="gotests.coverage" -timeout=3m
-          -count=1 -race -parallel=2
+          -count=1 -race -parallel=2 -failfast
 
       - name: Upload DataDog Trace
-        if: (success() || failure()) && github.actor != 'dependabot[bot]'
+        if: (success() || failure()) && github.actor != 'dependabot[bot]' && runner.os == 'Linux'
         env:
           DATADOG_API_KEY: ${{ secrets.DATADOG_API_KEY }}
           DD_DATABASE: postgresql

coderd/coderdtest/coderdtest.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"database/sql"
 	"io"
+	"net"
 	"net/http/httptest"
 	"net/url"
 	"os"
@@ -59,7 +60,13 @@ func New(t *testing.T) *codersdk.Client {
 		Database: db,
 		Pubsub:   pubsub,
 	})
-	srv := httptest.NewServer(handler)
+	srv := httptest.NewUnstartedServer(handler)
+	srv.Config.BaseContext = func(_ net.Listener) context.Context {
+		ctx, cancelFunc := context.WithCancel(context.Background())
+		t.Cleanup(cancelFunc)
+		return ctx
+	}
+	srv.Start()
 	serverURL, err := url.Parse(srv.URL)
 	require.NoError(t, err)
 	t.Cleanup(srv.Close)

peer/conn.go

@@ -183,12 +183,15 @@ func (c *Conn) init() error {
 		}
 	})
 	c.rtc.OnConnectionStateChange(func(peerConnectionState webrtc.PeerConnectionState) {
-		if c.isClosed() {
-			// Make sure we don't log after Close() has been called.
-			return
-		}
-		c.opts.Logger.Debug(context.Background(), "rtc connection updated",
-			slog.F("state", peerConnectionState))
+		go func() {
+			c.closeMutex.Lock()
+			defer c.closeMutex.Unlock()
+			if c.isClosed() {
+				return
+			}
+			c.opts.Logger.Debug(context.Background(), "rtc connection updated",
+				slog.F("state", peerConnectionState))
+		}()
 
 		switch peerConnectionState {
 		case webrtc.PeerConnectionStateDisconnected:

provisionerd/provisionerd.go

@@ -110,10 +110,13 @@ func (p *provisionerDaemon) connect(ctx context.Context) {
 			if errors.Is(err, context.Canceled) {
 				return
 			}
+			p.closeMutex.Lock()
 			if p.isClosed() {
+				p.closeMutex.Unlock()
 				return
 			}
 			p.opts.Logger.Warn(context.Background(), "failed to dial", slog.Error(err))
+			p.closeMutex.Unlock()
 			continue
 		}
 		p.opts.Logger.Debug(context.Background(), "connected")

pty/pty_other.go

@@ -6,6 +6,7 @@ package pty
 import (
 	"io"
 	"os"
+	"sync"
 
 	"github.com/creack/pty"
 )
@@ -23,6 +24,7 @@ func newPty() (PTY, error) {
 }
 
 type otherPty struct {
+	mutex    sync.Mutex
 	pty, tty *os.File
 }
 
@@ -41,6 +43,8 @@ func (p *otherPty) Output() io.ReadWriter {
 }
 
 func (p *otherPty) Resize(cols uint16, rows uint16) error {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
 	return pty.Setsize(p.tty, &pty.Winsize{
 		Rows: rows,
 		Cols: cols,
@@ -48,6 +52,9 @@ func (p *otherPty) Resize(cols uint16, rows uint16) error {
 }
 
 func (p *otherPty) Close() error {
+	p.mutex.Lock()
+	defer p.mutex.Unlock()
+
 	err := p.pty.Close()
 	if err != nil {
 		return err

pty/start_other.go

@@ -8,13 +8,17 @@ import (
 	"syscall"
 
 	"github.com/creack/pty"
+	"golang.org/x/xerrors"
 )
 
 func startPty(cmd *exec.Cmd) (PTY, error) {
 	ptty, tty, err := pty.Open()
 	if err != nil {
-		return nil, err
+		return nil, xerrors.Errorf("open: %w", err)
 	}
+	defer func() {
+		_ = tty.Close()
+	}()
 	cmd.SysProcAttr = &syscall.SysProcAttr{
 		Setsid:  true,
 		Setctty: true,
@@ -25,7 +29,7 @@ func startPty(cmd *exec.Cmd) (PTY, error) {
 	err = cmd.Start()
 	if err != nil {
 		_ = ptty.Close()
-		return nil, err
+		return nil, xerrors.Errorf("start: %w", err)
 	}
 	return &otherPty{
 		pty: ptty,