From 8f5636855d4e974ddf324a496e9e11bf07c74439 Mon Sep 17 00:00:00 2001 From: Paul Wells Date: Tue, 15 Sep 2026 00:31:02 -0700 Subject: [PATCH] agent endpoints: accept webtransport connections rather than racing Serve with Close ListenWebTransport ran wt.Serve on a goroutine per socket and returned a stop func calling wt.Close. Serve takes a reference on the server's refCount WaitGroup and Close waits on it, so the pair breaks the WaitGroup's own rule that an Add starting from zero must happen before a Wait. The race detector models that rule as a read of wg.sema in Add against a write in Wait, and reports it whenever Wait observes a non-zero counter: a listener stopped before its serve goroutine has run at all, which is every test that builds a stack and tears it down without a worker connecting. sync is compiled without instrumentation, so the report names the two ListenWebTransport call sites with no frame in between. webtransport-go v0.13.0 carries the same code. The accept loop moves here. quic.ListenEarly builds the listener Serve would have built, with the datagram and partial-delivery options the session layer requires, and each connection goes to Server.ServeQUICConn, which touches no part of that WaitGroup. The stop func keeps the order Close established: Close first, so every CONNECTION_CLOSE frame is transmitted while the sockets are still open, then the accept loops are cancelled and drained, then the listeners and sockets close. Co-authored-by: Claude Opus 5 (1M context) --- pkg/service/webtransport.go | 59 ++++++++++++++++++++++---- pkg/service/webtransportlisten_test.go | 37 ++++++++++++++++ 2 files changed, 88 insertions(+), 8 deletions(-) create mode 100644 pkg/service/webtransportlisten_test.go diff --git a/pkg/service/webtransport.go b/pkg/service/webtransport.go index 0a7431939..2d0ad276d 100644 --- a/pkg/service/webtransport.go +++ b/pkg/service/webtransport.go @@ -27,8 +27,10 @@ import ( "net" "net/http" "strconv" + "sync" "time" + "github.com/quic-go/quic-go" "github.com/quic-go/quic-go/http3" "github.com/quic-go/webtransport-go" "github.com/urfave/negroni/v3" @@ -139,9 +141,26 @@ func ListenWebTransport(wt *webtransport.Server, addrs []string, port uint32) ([ addrs = []string{""} } - conns := make([]*net.UDPConn, 0, len(addrs)) - bound := make([]net.Addr, 0, len(addrs)) + // webtransport.Server.Serve takes a reference on a WaitGroup that + // Server.Close waits on, so running the two concurrently is a race by the + // WaitGroup's own rules. Accepting here instead leaves that counter at zero: + // Server.ServeQUICConn never touches it. + quicConf := &quic.Config{} + if wt.H3.QUICConfig != nil { + quicConf = wt.H3.QUICConfig.Clone() + } + quicConf.EnableDatagrams = true + quicConf.EnableStreamResetPartialDelivery = true + + var ( + conns []*net.UDPConn + lns []*quic.EarlyListener + bound []net.Addr + ) closeAll := func() { + for _, ln := range lns { + _ = ln.Close() + } for _, c := range conns { _ = c.Close() } @@ -159,19 +178,43 @@ func ListenWebTransport(wt *webtransport.Server, addrs []string, port uint32) ([ } conns = append(conns, udp) bound = append(bound, udp.LocalAddr()) + + ln, err := quic.ListenEarly(udp, wt.H3.TLSConfig, quicConf) + if err != nil { + closeAll() + return nil, nil, err + } + lns = append(lns, ln) } - for _, udp := range conns { - go func() { - if err := wt.Serve(udp); err != nil { - logger.Infow("webtransport listener stopped", "error", err) - } - }() + ctx, cancel := context.WithCancel(context.Background()) + var serving sync.WaitGroup + for _, ln := range lns { + serving.Go(func() { acceptWebTransport(ctx, wt, ln, &serving) }) } logger.Infow("webtransport listener started", "addresses", bound) return bound, func() { + // the sockets stay open until everything has drained, so every + // CONNECTION_CLOSE frame still reaches its peer _ = wt.Close() + cancel() + serving.Wait() closeAll() }, nil } + +func acceptWebTransport(ctx context.Context, wt *webtransport.Server, ln *quic.EarlyListener, serving *sync.WaitGroup) { + for { + conn, err := ln.Accept(ctx) + if err != nil { + logger.Infow("webtransport listener stopped", "error", err) + return + } + serving.Go(func() { + if err := wt.ServeQUICConn(conn); err != nil && !errors.Is(err, http.ErrServerClosed) { + logger.Infow("webtransport connection stopped", "error", err) + } + }) + } +} diff --git a/pkg/service/webtransportlisten_test.go b/pkg/service/webtransportlisten_test.go new file mode 100644 index 000000000..fb0654707 --- /dev/null +++ b/pkg/service/webtransportlisten_test.go @@ -0,0 +1,37 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package service_test + +import ( + "net/http" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/livekit/livekit-server/pkg/service" +) + +// a listener stopped before it has served anything must not race its own accept +// loop: under -race this is what catches webtransport.Server.Serve running +// concurrently with Server.Close. +func TestWebTransportStopBeforeFirstConnection(t *testing.T) { + for range 20 { + wt := service.NewWebTransportServer(selfSignedTLS(t)) + wt.H3.Handler = service.NewWebTransportHandler(nil, wt, http.NewServeMux()) + _, stop, err := service.ListenWebTransport(wt, []string{"127.0.0.1"}, 0) + require.NoError(t, err) + stop() + } +}