This is an automated email from the ASF dual-hosted git repository.
Alanxtl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new e1bd4e817 fix(triple): guard httpSrv/http3Srv access with atomic
pointers (#3639)
e1bd4e817 is described below
commit e1bd4e817942e38824e03e4ebf8c79dffa81b98b
Author: Li Zining <[email protected]>
AuthorDate: Tue Aug 11 20:32:32 2026 +0800
fix(triple): guard httpSrv/http3Srv access with atomic pointers (#3639)
* fix(triple): guard httpSrv/http3Srv access with atomic pointers
Server starts the HTTP/2 and HTTP/3 listeners in background goroutines
(startHttp2/startHttp3/startHttp2AndHttp3 write s.httpSrv and s.http3Srv
on the start path) while Stop and GracefulStop read and close them on a
different goroutine, with no synchronization in between. Under -race this
reports a data race on both fields every time TestCfgAPI_Export runs.
Store the two servers in uatomic.Pointer fields: every write goes through
Store and every read through Load, and Stop/GracefulStop load once into a
local snapshot before checking nil and closing.
Signed-off-by: lizining <[email protected]>
* test(triple): verify triple server start and stop behavior
Add 13 tests that start the real HTTP/2, HTTP/3 and dual-protocol
servers on random ports, stop or gracefully stop them, and assert the
httpSrv/http3Srv uatomic.Pointer fields across start and stop. Also
cover stopping before start, repeated start/stop for every protocol,
and the missing-TLS / unsupported-protocol error paths. These tests
fail on the previous unsynchronized fields under -race.
Signed-off-by: lizining <[email protected]>
* test(triple): adapt nil assertions to uatomic.Pointer fields
The fields became uatomic.Pointer in the race fix, which is never nil
itself; load the pointer before the nil assertion.
Signed-off-by: lizining <[email protected]>
---------
Signed-off-by: lizining <[email protected]>
---
protocol/triple/triple_protocol/server.go | 53 ++--
.../triple_protocol/server_lifecycle_test.go | 344 +++++++++++++++++++++
protocol/triple/triple_protocol/server_test.go | 6 +-
3 files changed, 375 insertions(+), 28 deletions(-)
diff --git a/protocol/triple/triple_protocol/server.go
b/protocol/triple/triple_protocol/server.go
index ea88764a4..409b4e485 100644
--- a/protocol/triple/triple_protocol/server.go
+++ b/protocol/triple/triple_protocol/server.go
@@ -31,6 +31,8 @@ import (
"github.com/quic-go/quic-go/http3"
+ uatomic "go.uber.org/atomic"
+
"golang.org/x/net/http2"
"golang.org/x/net/http2/h2c"
@@ -48,8 +50,8 @@ type Server struct {
addr string
mux *methodRouteMux
handlers map[string]*Handler
- httpSrv *http.Server
- http3Srv *http3.Server
+ httpSrv uatomic.Pointer[http.Server]
+ http3Srv uatomic.Pointer[http3.Server]
tripleConfig *global.TripleConfig // Configuration for the triple
protocol
openapiIntegration *openapi.OpenAPIIntegration
}
@@ -196,20 +198,21 @@ func (s *Server) Run(callProtocol string, tlsConf
*tls.Config) error {
}
func (s *Server) startHttp2(tlsConf *tls.Config) error {
- s.httpSrv = &http.Server{
+ s.httpSrv.Store(&http.Server{
Addr: s.addr,
Handler: h2c.NewHandler(s.mux, &http2.Server{}),
TLSConfig: tlsConf,
- }
+ })
logger.Debugf("[Triple][Server] triple HTTP/2 Server starting on %v",
s.addr)
- var err error
+ srv := s.httpSrv.Load()
+ var err error
if tlsConf != nil {
- err = s.httpSrv.ListenAndServeTLS("", "")
+ err = srv.ListenAndServeTLS("", "")
} else {
- err = s.httpSrv.ListenAndServe()
+ err = srv.ListenAndServe()
}
return err
@@ -230,7 +233,7 @@ func (s *Server) startHttp3(tlsConf *tls.Config) error {
return err
}
- s.http3Srv = &http3.Server{
+ s.http3Srv.Store(&http3.Server{
Addr: s.addr,
Handler: s.mux,
// Adapt and enhance a generic tls.Config object into a
configuration
@@ -238,11 +241,11 @@ func (s *Server) startHttp3(tlsConf *tls.Config) error {
// ref:
https://quic-go.net/docs/http3/server/#setting-up-a-http3server
TLSConfig: http3.ConfigureTLSConfig(tlsConf),
QUICConfig: quicConfig,
- }
+ })
logger.Debugf("[Triple][Server] triple HTTP/3 Server starting on %v",
s.addr)
- return s.http3Srv.ListenAndServe()
+ return s.http3Srv.Load().ListenAndServe()
}
func (s *Server) startHttp2AndHttp3(tlsConf *tls.Config) error {
@@ -262,26 +265,26 @@ func (s *Server) startHttp2AndHttp3(tlsConf *tls.Config)
error {
}
// Start HTTP/3 server first to get its configuration
- s.http3Srv = &http3.Server{
+ s.http3Srv.Store(&http3.Server{
Addr: s.addr,
Handler: s.mux,
TLSConfig: http3.ConfigureTLSConfig(tlsConf),
QUICConfig: quicConfig,
- }
+ })
// Create Alt-Svc handler wrapper for HTTP/2 server
var negotiation bool
if s.tripleConfig != nil && s.tripleConfig.Http3 != nil {
negotiation = s.tripleConfig.Http3.Negotiation
}
- altSvcHandler := NewAltSvcHandler(s.mux, s.http3Srv, negotiation)
+ altSvcHandler := NewAltSvcHandler(s.mux, s.http3Srv.Load(), negotiation)
// Start HTTP/2 server with Alt-Svc handler wrapper
- s.httpSrv = &http.Server{
+ s.httpSrv.Store(&http.Server{
Addr: s.addr,
Handler: h2c.NewHandler(altSvcHandler, &http2.Server{}),
TLSConfig: tlsConf,
- }
+ })
logger.Debugf("[Triple][Server] triple HTTP/2 and HTTP/3 Server
starting on %v", s.addr)
@@ -290,7 +293,7 @@ func (s *Server) startHttp2AndHttp3(tlsConf *tls.Config)
error {
// Start HTTP/2 server in a goroutine
eg.Go(func() error {
- if err := s.httpSrv.ListenAndServeTLS("", ""); err != nil &&
err != http.ErrServerClosed {
+ if err := s.httpSrv.Load().ListenAndServeTLS("", ""); err !=
nil && err != http.ErrServerClosed {
return fmt.Errorf("HTTP/2 server error: %w", err)
}
return nil
@@ -298,7 +301,7 @@ func (s *Server) startHttp2AndHttp3(tlsConf *tls.Config)
error {
// Start HTTP/3 server in a goroutine
eg.Go(func() error {
- if err := s.http3Srv.ListenAndServe(); err != nil && err !=
http.ErrServerClosed {
+ if err := s.http3Srv.Load().ListenAndServe(); err != nil && err
!= http.ErrServerClosed {
return fmt.Errorf("HTTP/3 server error: %w", err)
}
return nil
@@ -313,9 +316,9 @@ func (s *Server) Stop() error {
eg, _ := errgroup.WithContext(context.Background())
// stop HTTP server
- if s.httpSrv != nil {
+ if srv := s.httpSrv.Load(); srv != nil {
eg.Go(func() error {
- if err := s.httpSrv.Close(); err != nil {
+ if err := srv.Close(); err != nil {
return fmt.Errorf("http server close failed:
%w", err)
}
return nil
@@ -323,9 +326,9 @@ func (s *Server) Stop() error {
}
// stop HTTP/3 server
- if s.http3Srv != nil {
+ if srv3 := s.http3Srv.Load(); srv3 != nil {
eg.Go(func() error {
- if err := s.http3Srv.Close(); err != nil {
+ if err := srv3.Close(); err != nil {
return fmt.Errorf("http3 server close failed:
%w", err)
}
return nil
@@ -341,9 +344,9 @@ func (s *Server) GracefulStop(ctx context.Context) error {
eg, ctx := errgroup.WithContext(ctx)
// shutdown HTTP server
- if s.httpSrv != nil {
+ if srv := s.httpSrv.Load(); srv != nil {
eg.Go(func() error {
- if err := s.httpSrv.Shutdown(ctx); err != nil {
+ if err := srv.Shutdown(ctx); err != nil {
return fmt.Errorf("http server shutdown failed:
%w", err)
}
return nil
@@ -351,9 +354,9 @@ func (s *Server) GracefulStop(ctx context.Context) error {
}
// shutdown HTTP/3 server
- if s.http3Srv != nil {
+ if srv3 := s.http3Srv.Load(); srv3 != nil {
eg.Go(func() error {
- if err := s.http3Srv.Shutdown(ctx); err != nil {
+ if err := srv3.Shutdown(ctx); err != nil {
return fmt.Errorf("http3 server shutdown
failed: %w", err)
}
return nil
diff --git a/protocol/triple/triple_protocol/server_lifecycle_test.go
b/protocol/triple/triple_protocol/server_lifecycle_test.go
new file mode 100644
index 000000000..2e6c39d4d
--- /dev/null
+++ b/protocol/triple/triple_protocol/server_lifecycle_test.go
@@ -0,0 +1,344 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 triple_protocol
+
+import (
+ "context"
+ "crypto/ecdsa"
+ "crypto/elliptic"
+ "crypto/rand"
+ "crypto/tls"
+ "crypto/x509"
+ "crypto/x509/pkix"
+ "errors"
+ "math/big"
+ "net"
+ "net/http"
+ "syscall"
+ "testing"
+ "time"
+)
+
+import (
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+import (
+ "dubbo.apache.org/dubbo-go/v3/common/constant"
+ "dubbo.apache.org/dubbo-go/v3/global"
+)
+
+// newTestTLSConfig generates an in-memory self-signed certificate for the
+// server-side TLS configuration. HTTP/3 requires TLS, so it is used by all
+// HTTP/3 related tests.
+func newTestTLSConfig(t *testing.T) *tls.Config {
+ t.Helper()
+
+ priv, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
+ require.NoError(t, err)
+
+ template := &x509.Certificate{
+ SerialNumber: big.NewInt(1),
+ Subject: pkix.Name{
+ CommonName: "localhost",
+ },
+ NotBefore: time.Now().Add(-24 * time.Hour),
+ NotAfter: time.Now().Add(24 * time.Hour),
+ KeyUsage: x509.KeyUsageDigitalSignature |
x509.KeyUsageKeyEncipherment,
+ ExtKeyUsage:
[]x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth},
+ DNSNames: []string{"localhost"},
+ IPAddresses: []net.IP{net.ParseIP("127.0.0.1")},
+ BasicConstraintsValid: true,
+ }
+
+ der, err := x509.CreateCertificate(rand.Reader, template, template,
&priv.PublicKey, priv)
+ require.NoError(t, err)
+
+ return &tls.Config{
+ Certificates: []tls.Certificate{
+ {Certificate: [][]byte{der}, PrivateKey: priv},
+ },
+ }
+}
+
+// getFreeAddr returns a free TCP address on loopback for the test server.
+func getFreeAddr(t *testing.T) string {
+ t.Helper()
+
+ l, err := net.Listen("tcp", "127.0.0.1:0")
+ require.NoError(t, err)
+ addr := l.Addr().String()
+ require.NoError(t, l.Close())
+
+ return addr
+}
+
+// runServer starts the server in a goroutine and returns the channel that
+// receives the error returned by Run. ListenAndServe is blocking, so the
+// server must always be started this way in tests.
+func runServer(srv *Server, protocol string, tlsConf *tls.Config) chan error {
+ errCh := make(chan error, 1)
+ go func() {
+ errCh <- srv.Run(protocol, tlsConf)
+ }()
+ return errCh
+}
+
+// isAddrInUse reports whether the error is a port conflict, i.e. the error
+// chain unwraps to syscall.EADDRINUSE.
+func isAddrInUse(err error) bool {
+ return errors.Is(err, syscall.EADDRINUSE)
+}
+
+// startTestServer creates a server on a free port and starts it in a
+// goroutine. It retries with a fresh port when binding fails.
+func startTestServer(t *testing.T, protocol string, cfg *global.TripleConfig,
tlsConf *tls.Config) (*Server, chan error) {
+ t.Helper()
+
+ for range 3 {
+ srv := NewServer(getFreeAddr(t), cfg)
+ errCh := runServer(srv, protocol, tlsConf)
+ select {
+ case err := <-errCh:
+ if isAddrInUse(err) {
+ continue
+ }
+ require.FailNow(t, "server failed to start", err)
+ case <-time.After(100 * time.Millisecond):
+ }
+ return srv, errCh
+ }
+ require.FailNow(t, "failed to find a free port after 3 attempts")
+ return nil, nil
+}
+
+// waitForTCPReady polls the address until a TCP connection can be established.
+func waitForTCPReady(t *testing.T, addr string, timeout time.Duration) {
+ t.Helper()
+
+ require.Eventually(t, func() bool {
+ conn, err := net.DialTimeout("tcp", addr, 200*time.Millisecond)
+ if err != nil {
+ return false
+ }
+ _ = conn.Close()
+ return true
+ }, timeout, 20*time.Millisecond)
+}
+
+// waitForHTTP3Stored waits until the HTTP/3 server has been stored, which
+// means the startup path has passed the Store call. QUIC listens on UDP,
+// so the TCP readiness probe does not apply here.
+func waitForHTTP3Stored(t *testing.T, srv *Server) {
+ t.Helper()
+
+ require.Eventually(t, func() bool {
+ return srv.http3Srv.Load() != nil
+ }, 3*time.Second, 20*time.Millisecond)
+}
+
+// waitForServerExit waits for Run to return. The test fails if the server
+// does not exit within the timeout.
+func waitForServerExit(t *testing.T, errCh chan error, timeout time.Duration)
error {
+ t.Helper()
+
+ select {
+ case err := <-errCh:
+ return err
+ case <-time.After(timeout):
+ require.FailNow(t, "server did not exit within", timeout)
+ return nil
+ }
+}
+
+func TestServer_HTTP2_StartAndStop(t *testing.T) {
+ srv, errCh := startTestServer(t, constant.CallHTTP2, nil, nil)
+ waitForTCPReady(t, srv.addr, 3*time.Second)
+
+ require.NotNil(t, srv.httpSrv.Load())
+ require.Nil(t, srv.http3Srv.Load())
+
+ require.NoError(t, srv.Stop())
+ require.ErrorIs(t, waitForServerExit(t, errCh, 5*time.Second),
http.ErrServerClosed)
+}
+
+func TestServer_HTTP2_StartAndStopWithTLS(t *testing.T) {
+ srv, errCh := startTestServer(t, constant.CallHTTP2, nil,
newTestTLSConfig(t))
+ waitForTCPReady(t, srv.addr, 3*time.Second)
+
+ require.NotNil(t, srv.httpSrv.Load())
+ require.Nil(t, srv.http3Srv.Load())
+
+ require.NoError(t, srv.Stop())
+ require.ErrorIs(t, waitForServerExit(t, errCh, 5*time.Second),
http.ErrServerClosed)
+}
+
+func TestServer_HTTP3_StartAndStop(t *testing.T) {
+ cfg := &global.TripleConfig{
+ Http3: &global.Http3Config{Enable: true},
+ }
+ srv, errCh := startTestServer(t, constant.CallHTTP3, cfg,
newTestTLSConfig(t))
+ waitForHTTP3Stored(t, srv)
+
+ require.NotNil(t, srv.http3Srv.Load())
+ require.Nil(t, srv.httpSrv.Load())
+
+ require.NoError(t, srv.Stop())
+ require.ErrorIs(t, waitForServerExit(t, errCh, 5*time.Second),
http.ErrServerClosed)
+}
+
+func TestServer_HTTP2AndHTTP3_StartAndStop(t *testing.T) {
+ cfg := &global.TripleConfig{
+ Http3: &global.Http3Config{Enable: true},
+ }
+ srv, errCh := startTestServer(t, constant.CallHTTP2AndHTTP3, cfg,
newTestTLSConfig(t))
+ waitForTCPReady(t, srv.addr, 3*time.Second)
+ waitForHTTP3Stored(t, srv)
+
+ require.NotNil(t, srv.httpSrv.Load())
+ require.NotNil(t, srv.http3Srv.Load())
+
+ require.NoError(t, srv.Stop())
+ // startHttp2AndHttp3 swallows http.ErrServerClosed inside the errgroup,
+ // so Run returns nil after the servers are closed.
+ require.NoError(t, waitForServerExit(t, errCh, 5*time.Second))
+}
+
+func TestServer_HTTP2_StartAndGracefulStop(t *testing.T) {
+ srv, errCh := startTestServer(t, constant.CallHTTP2, nil, nil)
+ waitForTCPReady(t, srv.addr, 3*time.Second)
+
+ require.NotNil(t, srv.httpSrv.Load())
+ require.Nil(t, srv.http3Srv.Load())
+
+ graceCtx, cancel := context.WithTimeout(context.Background(),
constant.DefaultGracefulShutdownTimeout)
+ defer cancel()
+ require.NoError(t, srv.GracefulStop(graceCtx))
+ require.ErrorIs(t, waitForServerExit(t, errCh, 5*time.Second),
http.ErrServerClosed)
+}
+
+func TestServer_HTTP3_StartAndGracefulStop(t *testing.T) {
+ cfg := &global.TripleConfig{
+ Http3: &global.Http3Config{Enable: true},
+ }
+ srv, errCh := startTestServer(t, constant.CallHTTP3, cfg,
newTestTLSConfig(t))
+ waitForHTTP3Stored(t, srv)
+
+ require.NotNil(t, srv.http3Srv.Load())
+ require.Nil(t, srv.httpSrv.Load())
+
+ graceCtx, cancel := context.WithTimeout(context.Background(),
constant.DefaultGracefulShutdownTimeout)
+ defer cancel()
+ require.NoError(t, srv.GracefulStop(graceCtx))
+ require.ErrorIs(t, waitForServerExit(t, errCh, 5*time.Second),
http.ErrServerClosed)
+}
+
+func TestServer_HTTP2AndHTTP3_StartAndGracefulStop(t *testing.T) {
+ cfg := &global.TripleConfig{
+ Http3: &global.Http3Config{Enable: true},
+ }
+ srv, errCh := startTestServer(t, constant.CallHTTP2AndHTTP3, cfg,
newTestTLSConfig(t))
+ waitForTCPReady(t, srv.addr, 3*time.Second)
+ waitForHTTP3Stored(t, srv)
+
+ require.NotNil(t, srv.httpSrv.Load())
+ require.NotNil(t, srv.http3Srv.Load())
+
+ graceCtx, cancel := context.WithTimeout(context.Background(),
constant.DefaultGracefulShutdownTimeout)
+ defer cancel()
+ require.NoError(t, srv.GracefulStop(graceCtx))
+ // startHttp2AndHttp3 swallows http.ErrServerClosed inside the errgroup,
+ // so Run returns nil after the servers are closed.
+ require.NoError(t, waitForServerExit(t, errCh, 5*time.Second))
+}
+
+func TestServer_StopBeforeStart(t *testing.T) {
+ srv := NewServer(getFreeAddr(t), nil)
+ require.NoError(t, srv.Stop())
+}
+
+func TestServer_GracefulStopBeforeStart(t *testing.T) {
+ srv := NewServer(getFreeAddr(t), nil)
+ graceCtx, cancel := context.WithTimeout(context.Background(),
constant.DefaultGracefulShutdownTimeout)
+ defer cancel()
+ require.NoError(t, srv.GracefulStop(graceCtx))
+}
+
+func TestServer_Run_HTTP3WithoutTLS(t *testing.T) {
+ srv := NewServer(getFreeAddr(t), nil)
+ err := srv.Run(constant.CallHTTP3, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "must have TLS config")
+}
+
+func TestServer_Run_HTTP2AndHTTP3WithoutTLS(t *testing.T) {
+ srv := NewServer(getFreeAddr(t), nil)
+ err := srv.Run(constant.CallHTTP2AndHTTP3, nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "must have TLS config")
+}
+
+func TestServer_RunUnsupportedProtocol(t *testing.T) {
+ srv := NewServer(getFreeAddr(t), nil)
+ err := srv.Run("tcp", nil)
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "unsupported protocol")
+}
+
+// TestServer_RepeatedStartStop runs several Start/Stop cycles across all
+// protocols. Each iteration creates a fresh Server, because an http.Server
+// cannot be restarted after Close.
+func TestServer_RepeatedStartStop(t *testing.T) {
+ tlsConf := newTestTLSConfig(t)
+
+ protocols := []struct {
+ protocol string
+ tlsConf *tls.Config
+ }{
+ {protocol: constant.CallHTTP2},
+ {protocol: constant.CallHTTP3, tlsConf: tlsConf},
+ {protocol: constant.CallHTTP2AndHTTP3, tlsConf: tlsConf},
+ }
+
+ for range 3 {
+ for _, tc := range protocols {
+ cfg := &global.TripleConfig{}
+ if tc.protocol == constant.CallHTTP3 || tc.protocol ==
constant.CallHTTP2AndHTTP3 {
+ cfg.Http3 = &global.Http3Config{Enable: true}
+ }
+ srv, errCh := startTestServer(t, tc.protocol, cfg,
tc.tlsConf)
+ switch tc.protocol {
+ case constant.CallHTTP2AndHTTP3:
+ waitForTCPReady(t, srv.addr, 3*time.Second)
+ waitForHTTP3Stored(t, srv)
+ case constant.CallHTTP3:
+ waitForHTTP3Stored(t, srv)
+ default:
+ waitForTCPReady(t, srv.addr, 3*time.Second)
+ }
+
+ require.NoError(t, srv.Stop())
+ if tc.protocol == constant.CallHTTP2AndHTTP3 {
+ require.NoError(t, waitForServerExit(t, errCh,
5*time.Second))
+ } else {
+ require.ErrorIs(t, waitForServerExit(t, errCh,
5*time.Second), http.ErrServerClosed)
+ }
+ }
+ }
+}
diff --git a/protocol/triple/triple_protocol/server_test.go
b/protocol/triple/triple_protocol/server_test.go
index 84d87ae45..f002682b0 100644
--- a/protocol/triple/triple_protocol/server_test.go
+++ b/protocol/triple/triple_protocol/server_test.go
@@ -159,7 +159,7 @@ func TestServer_HTTP3PathsUseQUICConfigHelper(t *testing.T)
{
err := srv.startHttp3(&tls.Config{})
require.Error(t, err)
require.ErrorContains(t, err, "keep-alive-period")
- assert.Nil(t, srv.http3Srv)
+ assert.Nil(t, srv.http3Srv.Load())
})
t.Run("start_http2_and_http3_returns_parse_error", func(t *testing.T) {
@@ -172,8 +172,8 @@ func TestServer_HTTP3PathsUseQUICConfigHelper(t *testing.T)
{
err := srv.startHttp2AndHttp3(&tls.Config{})
require.Error(t, err)
require.ErrorContains(t, err, "max-idle-timeout")
- assert.Nil(t, srv.http3Srv)
- assert.Nil(t, srv.httpSrv)
+ assert.Nil(t, srv.http3Srv.Load())
+ assert.Nil(t, srv.httpSrv.Load())
})
}
