fix: stop reading closed channel for /watch devcontainers endpoint (#19373)

Fixes https://github.com/coder/coder/issues/19372

We increase the read limit to 4MiB (we use this limit elsewhere). We
also make sure to stop sending messages when `containersCh` becomes
closed.

agent/agentcontainers/api.go

@@ -763,7 +763,11 @@ func (api *API) broadcastUpdatesLocked() {
 func (api *API) watchContainers(rw http.ResponseWriter, r *http.Request) {
 	ctx := r.Context()
 
-	conn, err := websocket.Accept(rw, r, nil)
+	conn, err := websocket.Accept(rw, r, &websocket.AcceptOptions{
+		// We want `NoContextTakeover` compression to balance improving
+		// bandwidth cost/latency with minimal memory usage overhead.
+		CompressionMode: websocket.CompressionNoContextTakeover,
+	})
 	if err != nil {
 		httpapi.Write(ctx, rw, http.StatusInternalServerError, codersdk.Response{
 			Message: "Failed to upgrade connection to websocket.",

coderd/workspaceagents.go

@@ -896,7 +896,11 @@ func (api *API) watchWorkspaceAgentContainers(rw http.ResponseWriter, r *http.Re
 		case <-ctx.Done():
 			return
 
-		case containers := <-containersCh:
+		case containers, ok := <-containersCh:
+			if !ok {
+				return
+			}
+
 			if err := encoder.Encode(containers); err != nil {
 				api.Logger.Error(ctx, "encode containers", slog.Error(err))
 				return

coderd/workspaceagents_test.go

@@ -1389,169 +1389,147 @@ func TestWorkspaceAgentContainers(t *testing.T) {
 func TestWatchWorkspaceAgentDevcontainers(t *testing.T) {
 	t.Parallel()
 
-	var (
-		ctx               = testutil.Context(t, testutil.WaitLong)
-		logger            = slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}).Leveled(slog.LevelDebug)
-		mClock            = quartz.NewMock(t)
-		updaterTickerTrap = mClock.Trap().TickerFunc("updaterLoop")
-		mCtrl             = gomock.NewController(t)
-		mCCLI             = acmock.NewMockContainerCLI(mCtrl)
+	t.Run("OK", func(t *testing.T) {
+		t.Parallel()
 
-		client, db = coderdtest.NewWithDatabase(t, &coderdtest.Options{Logger: &logger})
-		user       = coderdtest.CreateFirstUser(t, client)
-		r          = dbfake.WorkspaceBuild(t, db, database.WorkspaceTable{
-			OrganizationID: user.OrganizationID,
-			OwnerID:        user.UserID,
-		}).WithAgent(func(agents []*proto.Agent) []*proto.Agent {
-			return agents
-		}).Do()
+		var (
+			ctx               = testutil.Context(t, testutil.WaitLong)
+			logger            = slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}).Leveled(slog.LevelDebug)
+			mClock            = quartz.NewMock(t)
+			updaterTickerTrap = mClock.Trap().TickerFunc("updaterLoop")
+			mCtrl             = gomock.NewController(t)
+			mCCLI             = acmock.NewMockContainerCLI(mCtrl)
 
-		fakeContainer1 = codersdk.WorkspaceAgentContainer{
-			ID:           "container1",
-			CreatedAt:    dbtime.Now(),
-			FriendlyName: "container1",
-			Image:        "busybox:latest",
-			Labels: map[string]string{
-				agentcontainers.DevcontainerLocalFolderLabel: "/home/coder/project1",
-				agentcontainers.DevcontainerConfigFileLabel:  "/home/coder/project1/.devcontainer/devcontainer.json",
-			},
-			Running: true,
-			Status:  "running",
-		}
+			client, db = coderdtest.NewWithDatabase(t, &coderdtest.Options{Logger: &logger})
+			user       = coderdtest.CreateFirstUser(t, client)
+			r          = dbfake.WorkspaceBuild(t, db, database.WorkspaceTable{
+				OrganizationID: user.OrganizationID,
+				OwnerID:        user.UserID,
+			}).WithAgent(func(agents []*proto.Agent) []*proto.Agent {
+				return agents
+			}).Do()
 
-		fakeContainer2 = codersdk.WorkspaceAgentContainer{
-			ID:           "container1",
-			CreatedAt:    dbtime.Now(),
-			FriendlyName: "container2",
-			Image:        "busybox:latest",
-			Labels: map[string]string{
-				agentcontainers.DevcontainerLocalFolderLabel: "/home/coder/project2",
-				agentcontainers.DevcontainerConfigFileLabel:  "/home/coder/project2/.devcontainer/devcontainer.json",
-			},
-			Running: true,
-			Status:  "running",
-		}
-	)
+			fakeContainer1 = codersdk.WorkspaceAgentContainer{
+				ID:           "container1",
+				CreatedAt:    dbtime.Now(),
+				FriendlyName: "container1",
+				Image:        "busybox:latest",
+				Labels: map[string]string{
+					agentcontainers.DevcontainerLocalFolderLabel: "/home/coder/project1",
+					agentcontainers.DevcontainerConfigFileLabel:  "/home/coder/project1/.devcontainer/devcontainer.json",
+				},
+				Running: true,
+				Status:  "running",
+			}
 
-	stages := []struct {
-		containers []codersdk.WorkspaceAgentContainer
-		expected   codersdk.WorkspaceAgentListContainersResponse
-	}{
-		{
-			containers: []codersdk.WorkspaceAgentContainer{fakeContainer1},
-			expected: codersdk.WorkspaceAgentListContainersResponse{
-				Containers: []codersdk.WorkspaceAgentContainer{fakeContainer1},
-				Devcontainers: []codersdk.WorkspaceAgentDevcontainer{
-					{
-						Name:            "project1",
-						WorkspaceFolder: fakeContainer1.Labels[agentcontainers.DevcontainerLocalFolderLabel],
-						ConfigPath:      fakeContainer1.Labels[agentcontainers.DevcontainerConfigFileLabel],
-						Status:          "running",
-						Container:       &fakeContainer1,
+			fakeContainer2 = codersdk.WorkspaceAgentContainer{
+				ID:           "container1",
+				CreatedAt:    dbtime.Now(),
+				FriendlyName: "container2",
+				Image:        "busybox:latest",
+				Labels: map[string]string{
+					agentcontainers.DevcontainerLocalFolderLabel: "/home/coder/project2",
+					agentcontainers.DevcontainerConfigFileLabel:  "/home/coder/project2/.devcontainer/devcontainer.json",
+				},
+				Running: true,
+				Status:  "running",
+			}
+		)
+
+		stages := []struct {
+			containers []codersdk.WorkspaceAgentContainer
+			expected   codersdk.WorkspaceAgentListContainersResponse
+		}{
+			{
+				containers: []codersdk.WorkspaceAgentContainer{fakeContainer1},
+				expected: codersdk.WorkspaceAgentListContainersResponse{
+					Containers: []codersdk.WorkspaceAgentContainer{fakeContainer1},
+					Devcontainers: []codersdk.WorkspaceAgentDevcontainer{
+						{
+							Name:            "project1",
+							WorkspaceFolder: fakeContainer1.Labels[agentcontainers.DevcontainerLocalFolderLabel],
+							ConfigPath:      fakeContainer1.Labels[agentcontainers.DevcontainerConfigFileLabel],
+							Status:          "running",
+							Container:       &fakeContainer1,
+						},
 					},
 				},
 			},
-		},
-		{
-			containers: []codersdk.WorkspaceAgentContainer{fakeContainer1, fakeContainer2},
-			expected: codersdk.WorkspaceAgentListContainersResponse{
-				Containers: []codersdk.WorkspaceAgentContainer{fakeContainer1, fakeContainer2},
-				Devcontainers: []codersdk.WorkspaceAgentDevcontainer{
-					{
-						Name:            "project1",
-						WorkspaceFolder: fakeContainer1.Labels[agentcontainers.DevcontainerLocalFolderLabel],
-						ConfigPath:      fakeContainer1.Labels[agentcontainers.DevcontainerConfigFileLabel],
-						Status:          "running",
-						Container:       &fakeContainer1,
-					},
-					{
-						Name:            "project2",
-						WorkspaceFolder: fakeContainer2.Labels[agentcontainers.DevcontainerLocalFolderLabel],
-						ConfigPath:      fakeContainer2.Labels[agentcontainers.DevcontainerConfigFileLabel],
-						Status:          "running",
-						Container:       &fakeContainer2,
+			{
+				containers: []codersdk.WorkspaceAgentContainer{fakeContainer1, fakeContainer2},
+				expected: codersdk.WorkspaceAgentListContainersResponse{
+					Containers: []codersdk.WorkspaceAgentContainer{fakeContainer1, fakeContainer2},
+					Devcontainers: []codersdk.WorkspaceAgentDevcontainer{
+						{
+							Name:            "project1",
+							WorkspaceFolder: fakeContainer1.Labels[agentcontainers.DevcontainerLocalFolderLabel],
+							ConfigPath:      fakeContainer1.Labels[agentcontainers.DevcontainerConfigFileLabel],
+							Status:          "running",
+							Container:       &fakeContainer1,
+						},
+						{
+							Name:            "project2",
+							WorkspaceFolder: fakeContainer2.Labels[agentcontainers.DevcontainerLocalFolderLabel],
+							ConfigPath:      fakeContainer2.Labels[agentcontainers.DevcontainerConfigFileLabel],
+							Status:          "running",
+							Container:       &fakeContainer2,
+						},
 					},
 				},
 			},
-		},
-		{
-			containers: []codersdk.WorkspaceAgentContainer{fakeContainer2},
-			expected: codersdk.WorkspaceAgentListContainersResponse{
-				Containers: []codersdk.WorkspaceAgentContainer{fakeContainer2},
-				Devcontainers: []codersdk.WorkspaceAgentDevcontainer{
-					{
-						Name:            "",
-						WorkspaceFolder: fakeContainer1.Labels[agentcontainers.DevcontainerLocalFolderLabel],
-						ConfigPath:      fakeContainer1.Labels[agentcontainers.DevcontainerConfigFileLabel],
-						Status:          "stopped",
-						Container:       nil,
-					},
-					{
-						Name:            "project2",
-						WorkspaceFolder: fakeContainer2.Labels[agentcontainers.DevcontainerLocalFolderLabel],
-						ConfigPath:      fakeContainer2.Labels[agentcontainers.DevcontainerConfigFileLabel],
-						Status:          "running",
-						Container:       &fakeContainer2,
+			{
+				containers: []codersdk.WorkspaceAgentContainer{fakeContainer2},
+				expected: codersdk.WorkspaceAgentListContainersResponse{
+					Containers: []codersdk.WorkspaceAgentContainer{fakeContainer2},
+					Devcontainers: []codersdk.WorkspaceAgentDevcontainer{
+						{
+							Name:            "",
+							WorkspaceFolder: fakeContainer1.Labels[agentcontainers.DevcontainerLocalFolderLabel],
+							ConfigPath:      fakeContainer1.Labels[agentcontainers.DevcontainerConfigFileLabel],
+							Status:          "stopped",
+							Container:       nil,
+						},
+						{
+							Name:            "project2",
+							WorkspaceFolder: fakeContainer2.Labels[agentcontainers.DevcontainerLocalFolderLabel],
+							ConfigPath:      fakeContainer2.Labels[agentcontainers.DevcontainerConfigFileLabel],
+							Status:          "running",
+							Container:       &fakeContainer2,
+						},
 					},
 				},
 			},
-		},
-	}
-
-	// Set up initial state for immediate send on connection
-	mCCLI.EXPECT().List(gomock.Any()).Return(codersdk.WorkspaceAgentListContainersResponse{Containers: stages[0].containers}, nil)
-	mCCLI.EXPECT().DetectArchitecture(gomock.Any(), gomock.Any()).Return("<none>", nil).AnyTimes()
-
-	_ = agenttest.New(t, client.URL, r.AgentToken, func(o *agent.Options) {
-		o.Logger = logger.Named("agent")
-		o.Devcontainers = true
-		o.DevcontainerAPIOptions = []agentcontainers.Option{
-			agentcontainers.WithClock(mClock),
-			agentcontainers.WithContainerCLI(mCCLI),
-			agentcontainers.WithWatcher(watcher.NewNoop()),
 		}
-	})
 
-	resources := coderdtest.NewWorkspaceAgentWaiter(t, client, r.Workspace.ID).Wait()
-	require.Len(t, resources, 1, "expected one resource")
-	require.Len(t, resources[0].Agents, 1, "expected one agent")
-	agentID := resources[0].Agents[0].ID
+		// Set up initial state for immediate send on connection
+		mCCLI.EXPECT().List(gomock.Any()).Return(codersdk.WorkspaceAgentListContainersResponse{Containers: stages[0].containers}, nil)
+		mCCLI.EXPECT().DetectArchitecture(gomock.Any(), gomock.Any()).Return("<none>", nil).AnyTimes()
 
-	updaterTickerTrap.MustWait(ctx).MustRelease(ctx)
-	defer updaterTickerTrap.Close()
+		_ = agenttest.New(t, client.URL, r.AgentToken, func(o *agent.Options) {
+			o.Logger = logger.Named("agent")
+			o.Devcontainers = true
+			o.DevcontainerAPIOptions = []agentcontainers.Option{
+				agentcontainers.WithClock(mClock),
+				agentcontainers.WithContainerCLI(mCCLI),
+				agentcontainers.WithWatcher(watcher.NewNoop()),
+			}
+		})
 
-	containers, closer, err := client.WatchWorkspaceAgentContainers(ctx, agentID)
-	require.NoError(t, err)
-	defer func() {
-		closer.Close()
-	}()
+		resources := coderdtest.NewWorkspaceAgentWaiter(t, client, r.Workspace.ID).Wait()
+		require.Len(t, resources, 1, "expected one resource")
+		require.Len(t, resources[0].Agents, 1, "expected one agent")
+		agentID := resources[0].Agents[0].ID
 
-	// Read initial state sent immediately on connection
-	var got codersdk.WorkspaceAgentListContainersResponse
-	select {
-	case <-ctx.Done():
-	case got = <-containers:
-	}
-	require.NoError(t, ctx.Err())
+		updaterTickerTrap.MustWait(ctx).MustRelease(ctx)
+		defer updaterTickerTrap.Close()
 
-	require.Equal(t, stages[0].expected.Containers, got.Containers)
-	require.Len(t, got.Devcontainers, len(stages[0].expected.Devcontainers))
-	for j, expectedDev := range stages[0].expected.Devcontainers {
-		gotDev := got.Devcontainers[j]
-		require.Equal(t, expectedDev.Name, gotDev.Name)
-		require.Equal(t, expectedDev.WorkspaceFolder, gotDev.WorkspaceFolder)
-		require.Equal(t, expectedDev.ConfigPath, gotDev.ConfigPath)
-		require.Equal(t, expectedDev.Status, gotDev.Status)
-		require.Equal(t, expectedDev.Container, gotDev.Container)
-	}
-
-	// Process remaining stages through updater loop
-	for i, stage := range stages[1:] {
-		mCCLI.EXPECT().List(gomock.Any()).Return(codersdk.WorkspaceAgentListContainersResponse{Containers: stage.containers}, nil)
-
-		_, aw := mClock.AdvanceNext()
-		aw.MustWait(ctx)
+		containers, closer, err := client.WatchWorkspaceAgentContainers(ctx, agentID)
+		require.NoError(t, err)
+		defer func() {
+			closer.Close()
+		}()
 
+		// Read initial state sent immediately on connection
 		var got codersdk.WorkspaceAgentListContainersResponse
 		select {
 		case <-ctx.Done():
@@ -1559,9 +1537,9 @@ func TestWatchWorkspaceAgentDevcontainers(t *testing.T) {
 		}
 		require.NoError(t, ctx.Err())
 
-		require.Equal(t, stages[i+1].expected.Containers, got.Containers)
-		require.Len(t, got.Devcontainers, len(stages[i+1].expected.Devcontainers))
-		for j, expectedDev := range stages[i+1].expected.Devcontainers {
+		require.Equal(t, stages[0].expected.Containers, got.Containers)
+		require.Len(t, got.Devcontainers, len(stages[0].expected.Devcontainers))
+		for j, expectedDev := range stages[0].expected.Devcontainers {
 			gotDev := got.Devcontainers[j]
 			require.Equal(t, expectedDev.Name, gotDev.Name)
 			require.Equal(t, expectedDev.WorkspaceFolder, gotDev.WorkspaceFolder)
@@ -1569,7 +1547,103 @@ func TestWatchWorkspaceAgentDevcontainers(t *testing.T) {
 			require.Equal(t, expectedDev.Status, gotDev.Status)
 			require.Equal(t, expectedDev.Container, gotDev.Container)
 		}
-	}
+
+		// Process remaining stages through updater loop
+		for i, stage := range stages[1:] {
+			mCCLI.EXPECT().List(gomock.Any()).Return(codersdk.WorkspaceAgentListContainersResponse{Containers: stage.containers}, nil)
+
+			_, aw := mClock.AdvanceNext()
+			aw.MustWait(ctx)
+
+			var got codersdk.WorkspaceAgentListContainersResponse
+			select {
+			case <-ctx.Done():
+			case got = <-containers:
+			}
+			require.NoError(t, ctx.Err())
+
+			require.Equal(t, stages[i+1].expected.Containers, got.Containers)
+			require.Len(t, got.Devcontainers, len(stages[i+1].expected.Devcontainers))
+			for j, expectedDev := range stages[i+1].expected.Devcontainers {
+				gotDev := got.Devcontainers[j]
+				require.Equal(t, expectedDev.Name, gotDev.Name)
+				require.Equal(t, expectedDev.WorkspaceFolder, gotDev.WorkspaceFolder)
+				require.Equal(t, expectedDev.ConfigPath, gotDev.ConfigPath)
+				require.Equal(t, expectedDev.Status, gotDev.Status)
+				require.Equal(t, expectedDev.Container, gotDev.Container)
+			}
+		}
+	})
+
+	t.Run("PayloadTooLarge", func(t *testing.T) {
+		t.Parallel()
+
+		var (
+			ctx    = testutil.Context(t, testutil.WaitShort)
+			logger = slogtest.Make(t, &slogtest.Options{IgnoreErrors: true}).Leveled(slog.LevelDebug)
+			mCtrl  = gomock.NewController(t)
+			mCCLI  = acmock.NewMockContainerCLI(mCtrl)
+
+			client, db = coderdtest.NewWithDatabase(t, &coderdtest.Options{Logger: &logger})
+			user       = coderdtest.CreateFirstUser(t, client)
+			r          = dbfake.WorkspaceBuild(t, db, database.WorkspaceTable{
+				OrganizationID: user.OrganizationID,
+				OwnerID:        user.UserID,
+			}).WithAgent(func(agents []*proto.Agent) []*proto.Agent {
+				return agents
+			}).Do()
+		)
+
+		// WebSocket limit is 4MiB, so we want to ensure we create _more_ than 4MiB worth of payload.
+		// Creating 20,000 fake containers creates a payload of roughly 7MiB.
+		var fakeContainers []codersdk.WorkspaceAgentContainer
+		for range 20_000 {
+			fakeContainers = append(fakeContainers, codersdk.WorkspaceAgentContainer{
+				CreatedAt:    time.Now(),
+				ID:           uuid.NewString(),
+				FriendlyName: uuid.NewString(),
+				Image:        "busybox:latest",
+				Labels: map[string]string{
+					agentcontainers.DevcontainerLocalFolderLabel: "/home/coder/project",
+					agentcontainers.DevcontainerConfigFileLabel:  "/home/coder/project/.devcontainer/devcontainer.json",
+				},
+				Running: false,
+				Ports:   []codersdk.WorkspaceAgentContainerPort{},
+				Status:  string(codersdk.WorkspaceAgentDevcontainerStatusRunning),
+				Volumes: map[string]string{},
+			})
+		}
+
+		mCCLI.EXPECT().List(gomock.Any()).Return(codersdk.WorkspaceAgentListContainersResponse{Containers: fakeContainers}, nil)
+		mCCLI.EXPECT().DetectArchitecture(gomock.Any(), gomock.Any()).Return("<none>", nil).AnyTimes()
+
+		_ = agenttest.New(t, client.URL, r.AgentToken, func(o *agent.Options) {
+			o.Logger = logger.Named("agent")
+			o.Devcontainers = true
+			o.DevcontainerAPIOptions = []agentcontainers.Option{
+				agentcontainers.WithContainerCLI(mCCLI),
+				agentcontainers.WithWatcher(watcher.NewNoop()),
+			}
+		})
+
+		resources := coderdtest.NewWorkspaceAgentWaiter(t, client, r.Workspace.ID).Wait()
+		require.Len(t, resources, 1, "expected one resource")
+		require.Len(t, resources[0].Agents, 1, "expected one agent")
+		agentID := resources[0].Agents[0].ID
+
+		containers, closer, err := client.WatchWorkspaceAgentContainers(ctx, agentID)
+		require.NoError(t, err)
+		defer func() {
+			closer.Close()
+		}()
+
+		select {
+		case <-ctx.Done():
+			t.Fail()
+		case _, ok := <-containers:
+			require.False(t, ok)
+		}
+	})
 }
 
 func TestWorkspaceAgentRecreateDevcontainer(t *testing.T) {

codersdk/workspaceagents.go

@@ -550,7 +550,9 @@ func (c *Client) WatchWorkspaceAgentContainers(ctx context.Context, agentID uuid
 	}})
 
 	conn, res, err := websocket.Dial(ctx, reqURL.String(), &websocket.DialOptions{
-		CompressionMode: websocket.CompressionDisabled,
+		// We want `NoContextTakeover` compression to balance improving
+		// bandwidth cost/latency with minimal memory usage overhead.
+		CompressionMode: websocket.CompressionNoContextTakeover,
 		HTTPClient: &http.Client{
 			Jar:       jar,
 			Transport: c.HTTPClient.Transport,
@@ -563,6 +565,12 @@ func (c *Client) WatchWorkspaceAgentContainers(ctx context.Context, agentID uuid
 		return nil, nil, ReadBodyAsError(res)
 	}
 
+	// When a workspace has a few devcontainers running, or a single devcontainer
+	// has a large amount of apps, then each payload can easily exceed 32KiB.
+	// We up the limit to 4MiB to give us plenty of headroom for workspaces that
+	// have lots of dev containers with lots of apps.
+	conn.SetReadLimit(1 << 22) // 4MiB
+
 	d := wsjson.NewDecoder[WorkspaceAgentListContainersResponse](conn, websocket.MessageText, c.logger)
 	return d.Chan(), d, nil
 }

codersdk/workspacesdk/agentconn.go

@@ -400,6 +400,10 @@ func (c *AgentConn) WatchContainers(ctx context.Context, logger slog.Logger) (<-
 
 	conn, res, err := websocket.Dial(ctx, url, &websocket.DialOptions{
 		HTTPClient: c.apiClient(),
+
+		// We want `NoContextTakeover` compression to balance improving
+		// bandwidth cost/latency with minimal memory usage overhead.
+		CompressionMode: websocket.CompressionNoContextTakeover,
 	})
 	if err != nil {
 		if res == nil {
@@ -411,6 +415,12 @@ func (c *AgentConn) WatchContainers(ctx context.Context, logger slog.Logger) (<-
 		defer res.Body.Close()
 	}
 
+	// When a workspace has a few devcontainers running, or a single devcontainer
+	// has a large amount of apps, then each payload can easily exceed 32KiB.
+	// We up the limit to 4MiB to give us plenty of headroom for workspaces that
+	// have lots of dev containers with lots of apps.
+	conn.SetReadLimit(1 << 22) // 4MiB
+
 	d := wsjson.NewDecoder[codersdk.WorkspaceAgentListContainersResponse](conn, websocket.MessageText, logger)
 	return d.Chan(), d, nil
 }