From 064b8528848b40b8da347b49bdad5a6bb56b6e5e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Thu, 13 Aug 2026 01:08:32 +0100 Subject: [PATCH] fix(gateway): stop TunnelManager deadlocking on its own reservation (#2597) Co-authored-by: Kandev Agent Co-authored-by: Claude Opus 5 (1M context) --- .../internal/gateway/websocket/port_tunnel.go | 170 ++++++++++--- .../websocket/port_tunnel_concurrency_test.go | 223 ++++++++++++++++++ .../websocket/port_tunnel_reservation_test.go | 142 +++++++++++ .../websocket/port_tunnel_serve_test.go | 6 +- .../websocket/port_tunnel_teardown_test.go | 181 ++++++++++++++ .../gateway/websocket/port_tunnel_test.go | 2 +- 6 files changed, 686 insertions(+), 38 deletions(-) create mode 100644 apps/backend/internal/gateway/websocket/port_tunnel_concurrency_test.go create mode 100644 apps/backend/internal/gateway/websocket/port_tunnel_reservation_test.go create mode 100644 apps/backend/internal/gateway/websocket/port_tunnel_teardown_test.go diff --git a/apps/backend/internal/gateway/websocket/port_tunnel.go b/apps/backend/internal/gateway/websocket/port_tunnel.go index daa20256ed..1d9236b993 100644 --- a/apps/backend/internal/gateway/websocket/port_tunnel.go +++ b/apps/backend/internal/gateway/websocket/port_tunnel.go @@ -2,6 +2,7 @@ package websocket import ( "context" + "errors" "fmt" "net" "net/http" @@ -24,7 +25,8 @@ type TunnelInfo struct { TunnelPort int `json:"tunnel_port"` } -// tunnelEntry tracks a running tunnel HTTP server. +// tunnelEntry tracks a running tunnel HTTP server. Entries reach m.tunnels +// only once they are complete, so every reader can dereference them. type tunnelEntry struct { port int tunnelPort int @@ -33,6 +35,25 @@ type tunnelEntry struct { cancel context.CancelFunc } +// pendingTunnel is a StartTunnel that is still resolving and binding. It is +// kept out of m.tunnels so that a concurrent caller waits for the in-flight +// bind and shares its outcome instead of observing a half-built entry, and so +// that StopTunnel, ListTunnels, InvalidateSession and Shutdown never see one. +// tunnelPort and err are written before done is closed and read only after. +type pendingTunnel struct { + done chan struct{} + tunnelPort int + err error +} + +// errStartCanceled reports a start that was abandoned mid-bind because +// InvalidateSession or Shutdown dropped its reservation. +var errStartCanceled = errors.New("tunnel start canceled by session invalidation or shutdown") + +// errManagerClosed reports a start refused because Shutdown already ran. +// Shutdown is terminal: nothing may bind a listener it will never close. +var errManagerClosed = errors.New("tunnel manager is shut down") + // TunnelManager manages per-session port tunnels that bind dedicated host // ports and reverse-proxy traffic to agentctl's port-proxy endpoint. // Unlike the path-based PortProxyHandler, tunnels serve apps at the root @@ -42,7 +63,9 @@ type TunnelManager struct { logger *logger.Logger mu sync.Mutex - tunnels map[string]*tunnelEntry // key: "sessionId:port" + closed bool // set by Shutdown; refuses every later start + tunnels map[string]*tunnelEntry // key: "sessionId:port"; complete entries only + pending map[string]*pendingTunnel // key: "sessionId:port"; starts still in flight } // NewTunnelManager creates a new TunnelManager. @@ -51,6 +74,7 @@ func NewTunnelManager(lifecycleMgr *lifecycle.Manager, log *logger.Logger) *Tunn lifecycleMgr: lifecycleMgr, logger: log.WithFields(zap.String("component", "port-tunnel")), tunnels: make(map[string]*tunnelEntry), + pending: make(map[string]*pendingTunnel), } } @@ -60,20 +84,16 @@ func NewTunnelManager(lifecycleMgr *lifecycle.Manager, log *logger.Logger) *Tunn func (m *TunnelManager) StartTunnel(sessionID string, port int, tunnelPort int) (int, error) { cacheKey := sessionID + ":" + strconv.Itoa(port) - // Check for existing tunnel under lock to prevent duplicate creation. - m.mu.Lock() - if entry, ok := m.tunnels[cacheKey]; ok { - tp := entry.tunnelPort - m.mu.Unlock() - return tp, nil + started, existingPort, err := m.claimStart(cacheKey) + if started == nil { + // Another caller already owns this session:port, either as a running + // tunnel or as an in-flight start we just waited for. + return existingPort, err } - // Reserve the key with a nil entry to prevent concurrent creation. - m.tunnels[cacheKey] = nil - m.mu.Unlock() - target, authToken, ln, err := m.resolveAndBind(cacheKey, sessionID, tunnelPort) + target, authToken, ln, err := m.resolveAndBind(sessionID, tunnelPort) if err != nil { - return 0, err + return m.finishStart(cacheKey, started, 0, err) } actualPort := ln.Addr().(*net.TCPAddr).Port @@ -89,9 +109,14 @@ func (m *TunnelManager) StartTunnel(sessionID string, port int, tunnelPort int) cancel: cancel, } - m.mu.Lock() - m.tunnels[cacheKey] = entry - m.mu.Unlock() + if !m.publish(cacheKey, started, entry) { + cancel() + _ = ln.Close() + m.logger.Info("tunnel start canceled while binding", + zap.String("session_id", sessionID), + zap.Int("port", port)) + return m.finishStart(cacheKey, started, 0, errStartCanceled) + } m.serveTunnel(ctx, srv, ln, cacheKey, sessionID, port, actualPort) @@ -100,43 +125,94 @@ func (m *TunnelManager) StartTunnel(sessionID string, port int, tunnelPort int) zap.Int("port", port), zap.Int("tunnel_port", actualPort)) - return actualPort, nil + return m.finishStart(cacheKey, started, actualPort, nil) +} + +// claimStart reserves cacheKey for the calling StartTunnel. It returns a +// non-nil pendingTunnel when the caller owns the start; otherwise it returns +// the result of the tunnel that already exists or was being built +// concurrently, having waited for the latter to settle. +func (m *TunnelManager) claimStart(cacheKey string) (*pendingTunnel, int, error) { + m.mu.Lock() + closed := m.closed + entry, running := m.tunnels[cacheKey] + inFlight, starting := m.pending[cacheKey] + var started *pendingTunnel + if !closed && !running && !starting { + started = &pendingTunnel{done: make(chan struct{})} + m.pending[cacheKey] = started + } + m.mu.Unlock() + + switch { + case closed: + return nil, 0, errManagerClosed + case running: + return nil, entry.tunnelPort, nil + case starting: + <-inFlight.done // Never while holding m.mu: the owner needs it to finish. + return nil, inFlight.tunnelPort, inFlight.err + default: + return started, 0, nil + } +} + +// publish installs a finished tunnel under cacheKey, before the reservation is +// released, so a caller arriving in between finds the tunnel rather than +// starting a second one. It reports false when the reservation is no longer +// registered, which means InvalidateSession or Shutdown tore this session down +// while the bind was in flight: the caller must then abandon its listener +// instead of installing a tunnel nobody will ever close. +func (m *TunnelManager) publish(cacheKey string, started *pendingTunnel, entry *tunnelEntry) bool { + m.mu.Lock() + defer m.mu.Unlock() + + if m.pending[cacheKey] != started { + return false + } + m.tunnels[cacheKey] = entry + return true +} + +// finishStart releases the reservation taken by claimStart and hands its +// outcome to every caller waiting on the same session:port. The identity check +// matters: once a teardown has canceled this reservation another start may own +// the key, and that newcomer's reservation must survive. +func (m *TunnelManager) finishStart(cacheKey string, started *pendingTunnel, tunnelPort int, err error) (int, error) { + started.tunnelPort = tunnelPort + started.err = err + + m.mu.Lock() + if m.pending[cacheKey] == started { + delete(m.pending, cacheKey) + } + m.mu.Unlock() + + close(started.done) + return tunnelPort, err } // resolveAndBind resolves the agentctl target URL for the session and binds a -// local TCP listener for the tunnel. On failure it cleans up the placeholder -// entry reserved by StartTunnel. -func (m *TunnelManager) resolveAndBind(cacheKey, sessionID string, tunnelPort int) (*url.URL, string, net.Listener, error) { - cleanup := func() { - m.mu.Lock() - if m.tunnels[cacheKey] == nil { - delete(m.tunnels, cacheKey) - } - m.mu.Unlock() - } - +// local TCP listener for the tunnel. +func (m *TunnelManager) resolveAndBind(sessionID string, tunnelPort int) (*url.URL, string, net.Listener, error) { execution, ok := m.lifecycleMgr.GetExecutionBySessionID(sessionID) if !ok { - cleanup() return nil, "", nil, fmt.Errorf("session not found or no active execution") } agentctlClient := execution.GetAgentCtlClient() if agentctlClient == nil { - cleanup() return nil, "", nil, fmt.Errorf("agentctl client not available") } target, err := url.Parse(agentctlClient.BaseURL()) if err != nil { - cleanup() return nil, "", nil, fmt.Errorf("failed to parse agentctl URL: %w", err) } authToken := agentctlClient.AuthToken() ln, err := net.Listen("tcp", fmt.Sprintf(":%d", tunnelPort)) if err != nil { - cleanup() if netutil.IsAddrInUse(err) { return nil, "", nil, fmt.Errorf("port %d is already in use, choose a different port", tunnelPort) } @@ -219,6 +295,7 @@ func (m *TunnelManager) InvalidateSession(sessionID string) { delete(m.tunnels, key) } } + canceled := m.cancelPendingStarts(prefix) m.mu.Unlock() for _, entry := range toStop { @@ -226,10 +303,11 @@ func (m *TunnelManager) InvalidateSession(sessionID string) { _ = entry.ln.Close() } - if len(toStop) > 0 { + if len(toStop) > 0 || canceled > 0 { m.logger.Info("invalidated session tunnels", zap.String("session_id", sessionID), - zap.Int("count", len(toStop))) + zap.Int("count", len(toStop)), + zap.Int("canceled_starts", canceled)) } } @@ -241,6 +319,8 @@ func (m *TunnelManager) Shutdown() { entries = append(entries, entry) } m.tunnels = make(map[string]*tunnelEntry) + canceled := m.cancelPendingStarts("") + m.closed = true m.mu.Unlock() for _, entry := range entries { @@ -248,11 +328,29 @@ func (m *TunnelManager) Shutdown() { _ = entry.ln.Close() } - if len(entries) > 0 { - m.logger.Info("shutdown all tunnels", zap.Int("count", len(entries))) + if len(entries) > 0 || canceled > 0 { + m.logger.Info("shutdown all tunnels", + zap.Int("count", len(entries)), + zap.Int("canceled_starts", canceled)) } } +// cancelPendingStarts drops every reservation whose key has the given prefix +// ("" for all of them) and reports how many were dropped. A start whose +// reservation is gone abandons its listener in publish rather than installing a +// tunnel into a session, or a manager, that has already been torn down. +// The caller must hold m.mu. +func (m *TunnelManager) cancelPendingStarts(prefix string) int { + canceled := 0 + for key := range m.pending { + if strings.HasPrefix(key, prefix) { + delete(m.pending, key) + canceled++ + } + } + return canceled +} + func (m *TunnelManager) removeTunnel(cacheKey string) { m.mu.Lock() defer m.mu.Unlock() diff --git a/apps/backend/internal/gateway/websocket/port_tunnel_concurrency_test.go b/apps/backend/internal/gateway/websocket/port_tunnel_concurrency_test.go new file mode 100644 index 0000000000..1aae66760f --- /dev/null +++ b/apps/backend/internal/gateway/websocket/port_tunnel_concurrency_test.go @@ -0,0 +1,223 @@ +package websocket + +import ( + "net/http" + "testing" + "time" + + "github.com/kandev/kandev/internal/agent/runtime/lifecycle" + "github.com/kandev/kandev/internal/common/logger" +) + +// startOutcome is one concurrent StartTunnel result. A panic in a racing +// goroutine is recovered and reported here so it fails the test that caused it +// instead of taking the whole test binary down with it. +type startOutcome struct { + port int + err error + recovered any +} + +// startTunnelRacing runs StartTunnel in its own goroutine, releasing gate first +// so every caller enters the reservation window together. +func startTunnelRacing( + t *testing.T, + manager *TunnelManager, + gate <-chan struct{}, + sessionID string, + port int, +) <-chan startOutcome { + t.Helper() + out := make(chan startOutcome, 1) + go func() { + outcome := startOutcome{} + defer func() { + outcome.recovered = recover() + out <- outcome + }() + <-gate + outcome.port, outcome.err = manager.StartTunnel(sessionID, port, 0) + }() + return out +} + +// awaitStart collects one racing StartTunnel result under a bound. A leaked +// m.mu blocks the caller forever, so an unbounded receive here would hang the +// suite instead of reporting the defect. +func awaitStart(t *testing.T, what string, out <-chan startOutcome) startOutcome { + t.Helper() + timer := time.NewTimer(wsTestTimeout) + defer timer.Stop() + select { + case outcome := <-out: + if outcome.recovered != nil { + t.Errorf("%s panicked: %v", what, outcome.recovered) + } + return outcome + case <-timer.C: + t.Fatalf("%s did not return within %s: TunnelManager.mu is held by a goroutine that never released it", what, wsTestTimeout) + return startOutcome{} + } +} + +// mustNotBlock fails if fn has not returned within wsTestTimeout. Every +// TunnelManager method takes m.mu, so a lock leaked by a panicking StartTunnel +// shows up here as a hang rather than as an error. +func mustNotBlock(t *testing.T, what string, fn func()) { + t.Helper() + done := make(chan struct{}) + go func() { + defer close(done) + fn() + }() + timer := time.NewTimer(wsTestTimeout) + defer timer.Stop() + select { + case <-done: + case <-timer.C: + t.Fatalf("%s blocked for %s: TunnelManager.mu was never released", what, wsTestTimeout) + } +} + +// newTunnelRaceBackend returns a lifecycle manager whose named sessions all +// resolve to one fake agentctl, so tunnel starts against them succeed. +func newTunnelRaceBackend(t *testing.T, log *logger.Logger, sessionIDs ...string) *lifecycle.Manager { + t.Helper() + upstream := newFakeAgentctl(t, func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + lifecycleMgr := newProxyTestManager(t, log) + for _, sessionID := range sessionIDs { + addExecutionForURL(t, lifecycleMgr, sessionID, upstream.server.URL, "", log) + } + return lifecycleMgr +} + +// newBoundedTunnelManager returns a manager whose cleanup cannot hang the +// suite: Shutdown takes m.mu, so a leaked lock would block teardown forever. +func newBoundedTunnelManager(t *testing.T, lifecycleMgr *lifecycle.Manager, log *logger.Logger) *TunnelManager { + t.Helper() + manager := NewTunnelManager(lifecycleMgr, log) + t.Cleanup(func() { mustNotBlock(t, "Shutdown", manager.Shutdown) }) + return manager +} + +func newRacingTunnelManager(t *testing.T, sessionIDs ...string) *TunnelManager { + t.Helper() + log, _ := observedTerminalLogger(t) + return newBoundedTunnelManager(t, newTunnelRaceBackend(t, log, sessionIDs...), log) +} + +// Two clicks on "expose port" arrive as two ports.tunnel.start actions, and +// client.go dispatches each in its own goroutine. Both callers must end up +// sharing a single tunnel instead of observing the reservation the first one +// left behind mid-bind. +func TestStartTunnelIsSafeUnderConcurrentDuplicateStarts(t *testing.T) { + manager := newRacingTunnelManager(t, "sess-race") + + gate := make(chan struct{}) + first := startTunnelRacing(t, manager, gate, "sess-race", 3000) + second := startTunnelRacing(t, manager, gate, "sess-race", 3000) + close(gate) + + firstOutcome := awaitStart(t, "first StartTunnel", first) + secondOutcome := awaitStart(t, "second StartTunnel", second) + + for _, outcome := range []struct { + what string + got startOutcome + }{{"first", firstOutcome}, {"second", secondOutcome}} { + if outcome.got.err != nil { + t.Fatalf("%s StartTunnel() error = %v, want nil", outcome.what, outcome.got.err) + } + if outcome.got.port <= 0 { + t.Fatalf("%s StartTunnel() port = %d, want an OS-assigned port", outcome.what, outcome.got.port) + } + } + if firstOutcome.port != secondOutcome.port { + t.Fatalf("concurrent starts returned different ports (%d and %d), want one shared tunnel", + firstOutcome.port, secondOutcome.port) + } + + listed := tunnelList(t, manager, "sess-race") + if len(listed) != 1 { + t.Fatalf("tunnels = %v, want exactly 1 (a duplicate start must not bind a second listener)", listed) + } + if listed[0].TunnelPort != firstOutcome.port { + t.Fatalf("listed tunnel port = %d, want the started %d", listed[0].TunnelPort, firstOutcome.port) + } + if status, _ := getThroughTunnel(t, newTunnelClient(t), firstOutcome.port, "/"); status != http.StatusOK { + t.Fatalf("shared tunnel status = %d, want %d", status, http.StatusOK) + } +} + +// The serious half of the defect: StartTunnel dereferenced the reservation +// while holding m.mu behind a non-deferred Unlock, so a panic there stranded +// the lock and every later start, stop, list and invalidate blocked for the +// process lifetime. +func TestTunnelManagerStaysUsableAfterConcurrentDuplicateStarts(t *testing.T) { + manager := newRacingTunnelManager(t, "sess-usable") + + gate := make(chan struct{}) + first := startTunnelRacing(t, manager, gate, "sess-usable", 3000) + second := startTunnelRacing(t, manager, gate, "sess-usable", 3000) + close(gate) + tunnelPort := awaitStart(t, "first StartTunnel", first).port + awaitStart(t, "second StartTunnel", second) + + mustNotBlock(t, "ListTunnels after a duplicate start", func() { + _ = manager.ListTunnels("sess-usable") + }) + mustNotBlock(t, "StartTunnel for another port after a duplicate start", func() { + if _, err := manager.StartTunnel("sess-usable", 4000, 0); err != nil { + t.Errorf("StartTunnel(4000) error = %v, want nil", err) + } + }) + mustNotBlock(t, "InvalidateSession after a duplicate start", func() { + manager.InvalidateSession("sess-nobody") + }) + mustNotBlock(t, "StopTunnel after a duplicate start", func() { + if err := manager.StopTunnel("sess-usable", 3000); err != nil { + t.Errorf("StopTunnel() error = %v, want nil", err) + } + }) + assertPortClosed(t, tunnelPort) + + mustNotBlock(t, "Shutdown after a duplicate start", manager.Shutdown) +} + +// Every reservation must be released even when many callers pile onto the same +// key and the start fails, or the session:port pair stays permanently unusable. +func TestConcurrentFailingStartsReleaseTheReservation(t *testing.T) { + manager := newRacingTunnelManager(t) + + const callers = 8 + gate := make(chan struct{}) + outs := make([]<-chan startOutcome, 0, callers) + for range callers { + outs = append(outs, startTunnelRacing(t, manager, gate, "sess-missing", 3000)) + } + close(gate) + + for i, out := range outs { + outcome := awaitStart(t, "failing StartTunnel", out) + if outcome.err == nil { + t.Fatalf("caller %d: StartTunnel() error = nil, want a failure", i) + } + if outcome.port != 0 { + t.Fatalf("caller %d: StartTunnel() port = %d, want 0 on failure", i, outcome.port) + } + } + + // The key is usable again: nothing was left reserved by the failed round. + log, _ := observedTerminalLogger(t) + upstream := newFakeAgentctl(t, func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + }) + addExecutionForURL(t, manager.lifecycleMgr, "sess-missing", upstream.server.URL, "", log) + mustNotBlock(t, "StartTunnel after concurrent failures", func() { + if _, err := manager.StartTunnel("sess-missing", 3000, 0); err != nil { + t.Errorf("StartTunnel after the failed round: %v", err) + } + }) +} diff --git a/apps/backend/internal/gateway/websocket/port_tunnel_reservation_test.go b/apps/backend/internal/gateway/websocket/port_tunnel_reservation_test.go new file mode 100644 index 0000000000..6adafd2ec6 --- /dev/null +++ b/apps/backend/internal/gateway/websocket/port_tunnel_reservation_test.go @@ -0,0 +1,142 @@ +package websocket + +import ( + "errors" + "testing" + "time" +) + +// reserveInFlightStart seeds the reservation StartTunnel takes while it +// resolves and binds, and returns the settle function the owning start would +// call. Binding is a real syscall of unbounded duration, so this window is wide +// enough in production for every other method to run inside it. +func reserveInFlightStart(t *testing.T, manager *TunnelManager, cacheKey string) (*pendingTunnel, func(port int, err error)) { + t.Helper() + started := &pendingTunnel{done: make(chan struct{})} + manager.mu.Lock() + manager.pending[cacheKey] = started + manager.mu.Unlock() + + var settled bool + settle := func(port int, err error) { + if settled { + return + } + settled = true + _, _ = manager.finishStart(cacheKey, started, port, err) + } + // Registered immediately: a t.Fatal below must not strand callers that are + // already blocked on this reservation. + t.Cleanup(func() { settle(0, errors.New("test cleanup")) }) + return started, settle +} + +// A start that has not finished binding is invisible to every other method: +// none of them can observe a half-built tunnel, and none of them blocks on the +// in-flight bind. +func TestInFlightStartIsInvisibleToTheOtherReaders(t *testing.T) { + manager := newRacingTunnelManager(t) + _, settle := reserveInFlightStart(t, manager, "sess-inflight:3000") + + mustNotBlock(t, "ListTunnels during an in-flight start", func() { + if listed := tunnelList(t, manager, "sess-inflight"); len(listed) != 0 { + t.Errorf("ListTunnels() = %v during an in-flight start, want none", listed) + } + }) + mustNotBlock(t, "StopTunnel during an in-flight start", func() { + if err := manager.StopTunnel("sess-inflight", 3000); err == nil { + t.Error("StopTunnel() error = nil during an in-flight start, want a not-found error") + } + }) + mustNotBlock(t, "InvalidateSession for another session during an in-flight start", func() { + manager.InvalidateSession("sess-unrelated") + }) + // A start for a different session:port is not blocked by this reservation. + mustNotBlock(t, "StartTunnel for another key during an in-flight start", func() { + if _, err := manager.StartTunnel("sess-other", 3000, 0); err == nil { + t.Error("StartTunnel(sess-other) error = nil, want the missing-session failure") + } + }) + // Last, because Shutdown is terminal. + mustNotBlock(t, "Shutdown during an in-flight start", manager.Shutdown) + + settle(41234, nil) +} + +// A caller that loses the reservation race waits for the in-flight bind and +// shares its result, rather than binding a second listener or reading a +// half-built entry. +func TestDuplicateStartWaitsForTheInFlightResult(t *testing.T) { + tests := []struct { + name string + port int + err error + wantPort int + wantError string + }{ + {name: "in-flight start succeeds", port: 41234, wantPort: 41234}, + {name: "in-flight start fails", err: errors.New("bind exploded"), wantError: "bind exploded"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + manager := newRacingTunnelManager(t) + _, settle := reserveInFlightStart(t, manager, "sess-wait:3000") + + gate := make(chan struct{}) + close(gate) + waiter := startTunnelRacing(t, manager, gate, "sess-wait", 3000) + + // The waiter must be parked without holding m.mu, so an unrelated + // reader still runs while it waits. + mustNotBlock(t, "ListTunnels while a duplicate start waits", func() { + _ = manager.ListTunnels("sess-wait") + }) + // ListTunnels above is the m.mu liveness proof. This select is a + // secondary sanity-check that the waiter has not returned early; + // 50ms is ample for an in-process channel wait to park. + select { + case outcome := <-waiter: + t.Fatalf("duplicate StartTunnel returned %+v before the in-flight start settled", outcome) + case <-time.After(50 * time.Millisecond): + } + + settle(tt.port, tt.err) + + outcome := awaitStart(t, "duplicate StartTunnel", waiter) + if outcome.port != tt.wantPort { + t.Fatalf("duplicate StartTunnel() port = %d, want the in-flight %d", outcome.port, tt.wantPort) + } + if tt.wantError == "" { + if outcome.err != nil { + t.Fatalf("duplicate StartTunnel() error = %v, want nil", outcome.err) + } + return + } + if outcome.err == nil || outcome.err.Error() != tt.wantError { + t.Fatalf("duplicate StartTunnel() error = %v, want %q", outcome.err, tt.wantError) + } + }) + } +} + +// The reservation is released even when the start fails, so the same +// session:port can be started again. +func TestFinishStartReleasesTheReservation(t *testing.T) { + manager := newRacingTunnelManager(t) + _, settle := reserveInFlightStart(t, manager, "sess-release:3000") + settle(0, errors.New("bind exploded")) + + manager.mu.Lock() + _, stillReserved := manager.pending["sess-release:3000"] + manager.mu.Unlock() + if stillReserved { + t.Fatal("finishStart left the reservation in place") + } + + // A later start owns the key again instead of waiting on a dead reservation. + mustNotBlock(t, "StartTunnel after a released reservation", func() { + if _, err := manager.StartTunnel("sess-release", 3000, 0); err == nil { + t.Error("StartTunnel() error = nil, want the missing-session failure") + } + }) +} diff --git a/apps/backend/internal/gateway/websocket/port_tunnel_serve_test.go b/apps/backend/internal/gateway/websocket/port_tunnel_serve_test.go index bec80c759a..cd9ffde3ff 100644 --- a/apps/backend/internal/gateway/websocket/port_tunnel_serve_test.go +++ b/apps/backend/internal/gateway/websocket/port_tunnel_serve_test.go @@ -180,8 +180,12 @@ func TestStartTunnelReleasesReservationWhenResolutionFails(t *testing.T) { t.Fatalf("StartTunnel() port = %d, want 0 on failure", port) } manager.mu.Lock() - _, stillReserved := manager.tunnels[tt.sessionID+":3000"] + _, stillRunning := manager.tunnels[tt.sessionID+":3000"] + _, stillReserved := manager.pending[tt.sessionID+":3000"] manager.mu.Unlock() + if stillRunning { + t.Fatal("a failed start left an entry in the tunnel map") + } if stillReserved { t.Fatal("the reserved cache key was not released after the failure") } diff --git a/apps/backend/internal/gateway/websocket/port_tunnel_teardown_test.go b/apps/backend/internal/gateway/websocket/port_tunnel_teardown_test.go new file mode 100644 index 0000000000..3cc8d98e90 --- /dev/null +++ b/apps/backend/internal/gateway/websocket/port_tunnel_teardown_test.go @@ -0,0 +1,181 @@ +package websocket + +import ( + "errors" + "net/http" + "testing" +) + +// startTunnelRacingOnPort is startTunnelRacing for a caller that needs to know +// the tunnel port up front, so the port can be checked whichever way the race +// went. +func startTunnelRacingOnPort( + t *testing.T, + manager *TunnelManager, + gate <-chan struct{}, + sessionID string, + port, tunnelPort int, +) <-chan startOutcome { + t.Helper() + out := make(chan startOutcome, 1) + go func() { + outcome := startOutcome{} + defer func() { + outcome.recovered = recover() + out <- outcome + }() + <-gate + outcome.port, outcome.err = manager.StartTunnel(sessionID, port, tunnelPort) + }() + return out +} + +// A start still binding when its session or the whole manager is torn down must +// be canceled. Its reservation is the cancellation token: once the teardown has +// dropped it, publish refuses to install the tunnel, so no listener survives a +// teardown that could not see it. +func TestTeardownCancelsInFlightStarts(t *testing.T) { + tests := []struct { + name string + teardown func(*TunnelManager) + }{ + {name: "Shutdown", teardown: func(m *TunnelManager) { m.Shutdown() }}, + {name: "InvalidateSession", teardown: func(m *TunnelManager) { m.InvalidateSession("sess-teardown") }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + manager := newRacingTunnelManager(t) + started, settle := reserveInFlightStart(t, manager, "sess-teardown:3000") + + tt.teardown(manager) + + manager.mu.Lock() + _, stillReserved := manager.pending["sess-teardown:3000"] + manager.mu.Unlock() + if stillReserved { + t.Fatalf("%s left the in-flight reservation in place", tt.name) + } + + // The owning start now finishes its bind and must abandon it. + if manager.publish("sess-teardown:3000", started, &tunnelEntry{port: 3000, tunnelPort: 41000}) { + t.Fatalf("publish installed a tunnel after %s canceled its reservation", tt.name) + } + manager.mu.Lock() + installed := len(manager.tunnels) + manager.mu.Unlock() + if installed != 0 { + t.Fatalf("tunnels after a canceled start = %d, want 0", installed) + } + + settle(0, errStartCanceled) + }) + } +} + +// A canceled reservation must not take a newer one down with it: after a +// teardown, the next start owns the key, and the canceled start's finishStart +// must leave that newcomer's reservation alone. +func TestFinishStartKeepsANewerReservation(t *testing.T) { + manager := newRacingTunnelManager(t) + canceled, settleCanceled := reserveInFlightStart(t, manager, "sess-newer:3000") + manager.InvalidateSession("sess-newer") + + newer := &pendingTunnel{done: make(chan struct{})} + manager.mu.Lock() + manager.pending["sess-newer:3000"] = newer + manager.mu.Unlock() + t.Cleanup(func() { _, _ = manager.finishStart("sess-newer:3000", newer, 0, errors.New("test cleanup")) }) + + settleCanceled(0, errStartCanceled) + + manager.mu.Lock() + owner := manager.pending["sess-newer:3000"] + manager.mu.Unlock() + if owner != newer { + t.Fatalf("the canceled start's finishStart dropped the newer reservation (owner = %p, want %p)", owner, newer) + } + if manager.publish("sess-newer:3000", canceled, &tunnelEntry{port: 3000}) { + t.Fatal("publish installed a tunnel for the canceled start while a newer one owned the key") + } +} + +// End-to-end property, whatever the interleaving: a StartTunnel racing Shutdown +// either publishes a tunnel that Shutdown then closes, or is canceled and +// closes its own listener. Nothing may still be listening afterwards, and the +// manager must be left empty. +func TestStartTunnelRacingShutdownLeavesNoListener(t *testing.T) { + log, _ := observedTerminalLogger(t) + backend := newTunnelRaceBackend(t, log, "sess-shutdown-race") + + const attempts = 20 + published, canceled, refused, unbound := 0, 0, 0, 0 + for range attempts { + manager := newBoundedTunnelManager(t, backend, log) + wanted := reserveLoopbackPort(t) + + gate := make(chan struct{}) + start := startTunnelRacingOnPort(t, manager, gate, "sess-shutdown-race", 3000, wanted) + shutdownDone := make(chan struct{}) + go func() { + defer close(shutdownDone) + <-gate + manager.Shutdown() + }() + close(gate) + + outcome := awaitStart(t, "StartTunnel racing Shutdown", start) + mustNotBlock(t, "Shutdown racing StartTunnel", func() { <-shutdownDone }) + + switch { + case outcome.err == nil: + published++ + assertPortClosed(t, wanted) // Shutdown drained the published entry. + case errors.Is(outcome.err, errStartCanceled): + canceled++ + assertPortClosed(t, wanted) // The canceled start closed its own listener. + case errors.Is(outcome.err, errManagerClosed): + refused++ + assertPortClosed(t, wanted) // Refused before binding anything. + default: + // The reserved port was taken between reserve and bind; another + // process owns it, so its state says nothing about this manager. + unbound++ + } + + manager.mu.Lock() + leftTunnels, leftPending := len(manager.tunnels), len(manager.pending) + manager.mu.Unlock() + if leftTunnels != 0 || leftPending != 0 { + t.Fatalf("after the race: tunnels = %d, pending = %d, want 0 and 0", leftTunnels, leftPending) + } + } + t.Logf("outcomes over %d attempts: published=%d canceled=%d refused=%d port-unavailable=%d", + attempts, published, canceled, refused, unbound) +} + +// A tunnel published before a teardown is still closed by it; this pins the +// ordering the property test above relies on. +func TestShutdownClosesATunnelPublishedBeforeIt(t *testing.T) { + manager := newRacingTunnelManager(t, "sess-ordered") + + tunnelPort, err := manager.StartTunnel("sess-ordered", 3000, 0) + if err != nil { + t.Fatalf("StartTunnel: %v", err) + } + if status, _ := getThroughTunnel(t, newTunnelClient(t), tunnelPort, "/"); status != http.StatusOK { + t.Fatalf("status before shutdown = %d, want %d", status, http.StatusOK) + } + + manager.Shutdown() + assertPortClosed(t, tunnelPort) + + // Shutdown is terminal. A start afterwards must be refused rather than bind + // a listener that nothing is left to close. + again, err := manager.StartTunnel("sess-ordered", 3000, 0) + if !errors.Is(err, errManagerClosed) { + t.Fatalf("StartTunnel after Shutdown error = %v, want %v", err, errManagerClosed) + } + if again != 0 { + t.Fatalf("StartTunnel after Shutdown port = %d, want 0", again) + } +} diff --git a/apps/backend/internal/gateway/websocket/port_tunnel_test.go b/apps/backend/internal/gateway/websocket/port_tunnel_test.go index bb5d20c717..192f56ab45 100644 --- a/apps/backend/internal/gateway/websocket/port_tunnel_test.go +++ b/apps/backend/internal/gateway/websocket/port_tunnel_test.go @@ -46,7 +46,7 @@ func TestResolveAndBindReportsOccupiedPort(t *testing.T) { port := occupied.Addr().(*net.TCPAddr).Port mgr := NewTunnelManager(lifecycleMgr, log) - _, _, _, err = mgr.resolveAndBind("session:1", "session", port) + _, _, _, err = mgr.resolveAndBind("session", port) if err == nil { t.Fatal("resolveAndBind() = nil, want occupied-port error") }