diff --git a/internal/server/bootstrap_test.go b/internal/server/bootstrap_test.go index 738a12b..e621079 100644 --- a/internal/server/bootstrap_test.go +++ b/internal/server/bootstrap_test.go @@ -228,24 +228,18 @@ func TestOnlineUsersSyntheticBadge(t *testing.T) { func TestRunServiceHTTPAndListen(t *testing.T) { gin.SetMode(gin.TestMode) - ln, err := net.Listen("tcp", "127.0.0.1:0") - require.NoError(t, err) - addr := ln.Addr().String() - _ = ln.Close() - // Pre-bind HTTP listener for HTTPListen inject + // Pre-bind HTTP listener for HTTPListen inject (service HTTP independent of FSD). httpLn, err := net.Listen("tcp", "127.0.0.1:0") require.NoError(t, err) httpAddr := httpLn.Addr().String() - fsdLn, err := net.Listen("tcp", "127.0.0.1:0") - require.NoError(t, err) - fsdAddr := fsdLn.Addr().String() - + fsdBound := make(chan string, 2) kv := &mapConfig{m: map[string]string{db.ConfigJwtSecretKey: TestJWTSecret}} srv, err := New(Deps{ Config: &Config{ - FsdListenAddrs: []string{fsdAddr}, + FsdListenAddrs: []string{"127.0.0.1:0"}, + FsdNumEventLoop: 1, ServiceHTTPListenAddr: httpAddr, }, Users: stubUserStore{}, @@ -253,9 +247,7 @@ func TestRunServiceHTTPAndListen(t *testing.T) { Registry: postoffice.New(), Metar: &recordingMetar{}, Clock: realClock{}, - Listen: func(ctx context.Context, network, address string) (net.Listener, error) { - return fsdLn, nil - }, + FSDBound: fsdBound, HTTPListen: func(network, address string) (net.Listener, error) { return httpLn, nil }, @@ -266,13 +258,26 @@ func TestRunServiceHTTPAndListen(t *testing.T) { done := make(chan error, 1) go func() { done <- srv.Run(ctx) }() - // Give listeners a moment - time.Sleep(50 * time.Millisecond) - // Connect TCP briefly - conn, err := net.DialTimeout("tcp", fsdAddr, time.Second) - if err == nil { - _ = conn.Close() + // Readiness: FSDBound select (not sleep-only), same style as StartTestServer. + var fsdAddr string + select { + case fsdAddr = <-fsdBound: + case err := <-done: + cancel() + t.Fatalf("server exited before FSD bind: %v", err) + case <-time.After(8 * time.Second): + cancel() + t.Fatal("timeout waiting for FSD listener (gnet OnBoot)") } + if host, port, splitErr := net.SplitHostPort(fsdAddr); splitErr == nil { + if host == "0.0.0.0" || host == "::" { + fsdAddr = net.JoinHostPort("127.0.0.1", port) + } + } + + conn, err := net.DialTimeout("tcp", fsdAddr, time.Second) + require.NoError(t, err) + _ = conn.Close() cancel() select { @@ -280,7 +285,6 @@ func TestRunServiceHTTPAndListen(t *testing.T) { case <-time.After(5 * time.Second): t.Fatal("Run did not return") } - _ = addr } type mapConfig struct{ m map[string]string } diff --git a/internal/server/config.go b/internal/server/config.go index 24df92e..7558c9b 100644 --- a/internal/server/config.go +++ b/internal/server/config.go @@ -15,8 +15,7 @@ type Config struct { FsdListenAddrs []string `env:"FSD_LISTEN_ADDRS, default=:6809"` // FSD listen addresses // FsdNumEventLoop is the number of gnet event-loop goroutines for the FSD - // TCP plane. 0 (default) means GOMAXPROCS. Only used when the gnet path is - // active (Deps.Listen is nil). + // TCP plane. 0 (default) means GOMAXPROCS. FsdNumEventLoop int `env:"FSD_NUM_EVENT_LOOP, default=0"` // DatabaseDriver is accepted for backward compatibility only. diff --git a/internal/server/conn.go b/internal/server/conn.go index cb30bb0..3de6a26 100644 --- a/internal/server/conn.go +++ b/internal/server/conn.go @@ -1,15 +1,11 @@ package server import ( - "bufio" - "context" "errors" "fmt" "io" - "net" "strconv" "strings" - "time" "github.com/renorris/openfsd/internal/auth" "github.com/renorris/openfsd/internal/db" @@ -27,138 +23,6 @@ func sendError(conn io.Writer, code int, message string) (err error) { return protocol.WriteError(conn, protocol.ErrorCode(code), message) } -// handleConn manages a single client connection (classic net path). -// If any errors occur during the process, it sends an error to the client and closes the connection. -func (s *Server) handleConn(ctx context.Context, conn net.Conn) { - defer func() { - if err := recover(); err != nil { - s.logger.Error("FSD connection goroutine panicked", "err", err) - } - }() - - ip := remoteIPFromConn(conn) - if !s.limits.tryAcquireConn(ip, s.cfg.FsdMaxConnections, s.cfg.FsdMaxConnectionsPerIP) { - _ = sendError(conn, ServerFullError, "Server full") - _ = conn.Close() - s.logger.Debug("connection rejected: connection limit", "ip", ip) - return - } - defer s.limits.releaseConn(ip) - defer conn.Close() - - // Login deadline (classic path). - if s.cfg.FsdLoginTimeout > 0 { - _ = conn.SetReadDeadline(time.Now().Add(s.cfg.FsdLoginTimeout)) - } - - if err := sendServerIdent(conn); err != nil { - s.logger.Debug("error sending server ident", "err", err) - return - } - - scanner := bufio.NewScanner(conn) - buf := make([]byte, 4096) - scanner.Buffer(buf, len(buf)) - - data, token, err := readLoginPackets(conn, scanner, s.clock) - if err != nil { - return - } - - // Clear login deadline; idle timeouts applied in eventLoop. - _ = conn.SetReadDeadline(time.Time{}) - - // Check if the requested callsign is OK - if !isValidClientCallsign([]byte(data.Callsign)) { - sendError(conn, CallsignInvalidError, "Callsign invalid") - return - } - - client := session.New(ctx, conn, scanner, data) - client.Auth = &auth.AuthState{} - client.SetRemoteIP(ip) - client.LastInboundNs.Store(s.clock.Now().UnixNano()) - - // Attempt to authenticate connection (login-phase errors still use sendError on conn) - if err = s.attemptAuthentication(client, token); err != nil { - return - } - - // Per-CID session cap (after auth, before register). - if !s.limits.tryAcquireCID(client.CID, s.cfg.FsdMaxSessionsPerCID) { - sendError(conn, ServerFullError, "Too many sessions for this CID") - return - } - cidHeld := true - defer func() { - if cidHeld { - s.limits.releaseCID(client.CID) - } - }() - - // Attempt to register to registry - if err = s.registry.Register(client); err != nil { - if errors.Is(err, ErrCallsignInUse) { - sendError(conn, CallsignInUseError, "Callsign already in use") - } - return - } - defer s.registry.Release(client) - - // Start sender before any post-login outbound traffic (MOTD, etc.). - // After this point, all writes go through client.Send → SenderWorker. - // Direct conn.Write is forbidden outside SenderWorker. - go client.SenderWorker() - - // Send hello message to client - if err = s.sendMotd(client); err != nil { - client.Disconnect() - return - } - - // Broadcast add packet to entire server - s.broadcastAddPacket(client) - defer s.broadcastDisconnectPacket(client) - - s.eventLoop(client) -} - -// eventLoop reads packets from the session and dispatches handlers. -// SenderWorker must already be running before eventLoop is entered. -func (s *Server) eventLoop(client *session.Session) { - defer client.Disconnect() - - for { - if s.cfg.FsdIdleTimeout > 0 && client.Conn != nil { - _ = client.Conn.SetReadDeadline(time.Now().Add(s.cfg.FsdIdleTimeout)) - } - - if !client.Scanner.Scan() { - return - } - - // Copy packet out of the scanner buffer (reused on next Scan) and - // re-append CRLF so handlers / fan-out own an immutable payload. - raw := client.Scanner.Bytes() - packet := make([]byte, len(raw)+2) - copy(packet, raw) - packet[len(raw)] = '\r' - packet[len(raw)+1] = '\n' - - client.LastInboundNs.Store(s.clock.Now().UnixNano()) - - // Verify packet and obtain type - packetType, ok := verifyPacket(packet, client) - if !ok { - continue - } - - // Run handler - handler := s.getHandler(packetType) - handler(client, packet) - } -} - // sendServerIdent sends the initial server identification packet to the client. // It returns an error if writing to the connection fails. func sendServerIdent(conn io.Writer) (err error) { @@ -176,37 +40,6 @@ var ErrInvalidAddPacket = errors.New("invalid add packet") // ErrInvalidIDPacket is returned when the ID packet from the client is invalid. var ErrInvalidIDPacket = errors.New("invalid ID packet") -// readLoginPackets reads the two expected login packets from the client: -// the client identification packet and the add packet. -// It parses these packets to extract the client's data and returns it in a LoginData struct. -// If any errors occur during reading or parsing, it sends an error to the client and returns an error. -func readLoginPackets(conn net.Conn, scanner *bufio.Scanner, clock Clock) (data session.LoginData, token string, err error) { - // Client ident - if !scanner.Scan() { - err = ErrInvalidIDPacket - sendError(conn, SyntaxError, "Error reading Client ident packet") - return - } - idPacket := append([]byte{}, scanner.Bytes()...) - - // Add packet - if !scanner.Scan() { - err = ErrInvalidAddPacket - sendError(conn, SyntaxError, "Error reading add packet") - return - } - addPacket := append([]byte{}, scanner.Bytes()...) - - var errCode int - var errMsg string - data, token, errCode, errMsg, err = parseLoginPackets(idPacket, addPacket, clock.Now()) - if err != nil { - sendError(conn, errCode, errMsg) - return - } - return -} - func (s *Server) attemptAuthentication(client *session.Session, token string) (err error) { ip := client.RemoteIP() now := s.clock.Now() @@ -422,18 +255,3 @@ func (s *Server) sendServerTextMessage(client *session.Session, msg string) (err return client.Send(packet.String()) } - -func remoteIPFromConn(conn net.Conn) string { - if conn == nil { - return "" - } - addr := conn.RemoteAddr() - if addr == nil { - return "" - } - host, _, err := net.SplitHostPort(addr.String()) - if err != nil { - return addr.String() - } - return host -} diff --git a/internal/server/deps.go b/internal/server/deps.go index de798c5..0fc6acc 100644 --- a/internal/server/deps.go +++ b/internal/server/deps.go @@ -67,15 +67,8 @@ type Deps struct { Metar MetarQueue Clock Clock // nil => real clock Logger *slog.Logger - // Listen optionally injects FSD net listener creation (tests). - // When non-nil, the classic per-connection goroutine path is used instead of gnet. - // nil => production gnet FSD plane (or defaultListen only if ForceClassicFSD). - Listen func(ctx context.Context, network, addr string) (net.Listener, error) - // ForceClassicFSD forces the classic net.Listener accept loop even when Listen - // is nil. Used by tests that need the classic path without a custom Listen. - ForceClassicFSD bool // FSDBound is an optional channel that receives each bound FSD listen address - // after the listener starts (useful for :0). Buffered; non-blocking send. + // after the gnet listener starts (useful for :0). Buffered; non-blocking send. FSDBound chan<- string // HTTPListen optionally injects service HTTP listener creation (tests). // Signature matches net.Listen. nil => net.Listen("tcp", ServiceHTTPListenAddr). @@ -89,8 +82,3 @@ type Deps struct { type realClock struct{} func (realClock) Now() time.Time { return time.Now() } - -func defaultListen(ctx context.Context, network, addr string) (net.Listener, error) { - var lc net.ListenConfig - return lc.Listen(ctx, network, addr) -} diff --git a/internal/server/deps_test.go b/internal/server/deps_test.go index 4344a38..e53138c 100644 --- a/internal/server/deps_test.go +++ b/internal/server/deps_test.go @@ -72,9 +72,6 @@ func TestNewRequiresDeps(t *testing.T) { if srv.logger == nil { t.Fatal("logger should default") } - if srv.listen == nil { - t.Fatal("listen should default") - } if !srv.clock.Now().Equal(time.Unix(1_700_000_000, 0)) { t.Fatalf("clock not wired: %v", srv.clock.Now()) } diff --git a/internal/server/gnet_fsd.go b/internal/server/gnet_fsd.go index 547ca8c..cbd746c 100644 --- a/internal/server/gnet_fsd.go +++ b/internal/server/gnet_fsd.go @@ -9,8 +9,7 @@ package server // per-conn SenderWorker). Reliable packets flush immediately; position // packets latest-wins and coalesce by size/idle. // - Login-phase sync writes use gnet.Conn.Write from OnTraffic (same loop). -// - HTTP admin remains on net/http. Classic net.Listener path is used when -// Deps.Listen is injected (tests). +// - HTTP admin remains on net/http. import ( "bytes" @@ -46,7 +45,7 @@ type fsdConnCtx struct { client *session.Session registered bool // disconnect broadcast deferred until OnClose after successful register - // (mirrors handleConn defer broadcastDisconnectPacket). + // (mirrors gnet disconnect / synthetic cleanup broadcastDisconnectPacket). remoteIP string connHeld bool // limits.tryAcquireConn succeeded diff --git a/internal/server/gnet_fsd_test.go b/internal/server/gnet_fsd_test.go index 44d7d56..22a7e45 100644 --- a/internal/server/gnet_fsd_test.go +++ b/internal/server/gnet_fsd_test.go @@ -8,7 +8,7 @@ import ( ) // TestGnetFSD_LoginAndPosition exercises the production gnet path (StartTestServer -// leaves Deps.Listen nil) for dual login, MOTD, and ranged position fan-out. +// for dual login, MOTD, and ranged position fan-out. func TestGnetFSD_LoginAndPosition(t *testing.T) { ts := server.StartTestServer(t) diff --git a/internal/server/handler_admin.go b/internal/server/handler_admin.go index 3b650e9..339fc1a 100644 --- a/internal/server/handler_admin.go +++ b/internal/server/handler_admin.go @@ -15,7 +15,7 @@ func (s *Server) handleKillRequest(client *session.Session, packet []byte) { return } - // Synthetic (sweatbox) sessions have no handleConn Release defer — Remove + // Synthetic (sweatbox) sessions have no gnet disconnect Release defer — Remove // performs pointer-scoped #DP + registry.Release + engine.Delete. if victim.Synthetic && s.sweatbox != nil { _ = s.sweatbox.Remove(victim.Callsign) diff --git a/internal/server/handler_position.go b/internal/server/handler_position.go index d4f5814..986fcca 100644 --- a/internal/server/handler_position.go +++ b/internal/server/handler_position.go @@ -140,7 +140,7 @@ func (s *Server) handlePilotPosition(client *session.Session, packet []byte) { const pilotVisRange = 50.0 * 1852.0 // 50 nautical miles // Update registry position then fan-out (hot path). - // packet is an owned immutable copy (eventLoop / gnet dispatch). + // packet is an owned immutable copy (gnet dispatch). s.registry.UpdatePosition(client, [2]float64{lat, lon}, pilotVisRange) // Rewrite rating field (index 3) from authenticated session rating. diff --git a/internal/server/http_service.go b/internal/server/http_service.go index 6dc0a3c..73bf4fe 100644 --- a/internal/server/http_service.go +++ b/internal/server/http_service.go @@ -186,7 +186,7 @@ func (s *Server) handleKickUser(c *gin.Context) { return } - // Synthetic (sweatbox) sessions have no handleConn Release defer — Remove + // Synthetic (sweatbox) sessions have no gnet disconnect Release defer — Remove // performs pointer-scoped #DP + registry.Release + engine.Delete. if client.Synthetic && s.sweatbox != nil { _ = s.sweatbox.Remove(client.Callsign) diff --git a/internal/server/io_ab_verify_test.go b/internal/server/io_ab_verify_test.go index 7729d00..40f047d 100644 --- a/internal/server/io_ab_verify_test.go +++ b/internal/server/io_ab_verify_test.go @@ -1,8 +1,8 @@ //go:build verifyperf -// A/B verification: gnet FSD plane vs classic 2-goroutine-per-conn path. +// Gnet FSD plane baseline smoke (historical A/B vs classic removed with dual-path). // -// go test -tags=verifyperf -count=1 -timeout=180s ./internal/server/ -run TestIOAB -v +// go test -tags=verifyperf -count=1 -timeout=180s ./internal/server/ -run TestIO_GnetBaseline -v // // Reports goroutine counts, send throughput, and peer receive counts under a // hub-like position storm. Not compiled into default test runs. @@ -76,60 +76,25 @@ type abResult struct { recvPerSec float64 } -func TestIOAB_GnetVsClassic(t *testing.T) { +func TestIO_GnetBaseline(t *testing.T) { if testing.Short() { - t.Skip("verifyperf A/B under -short") + t.Skip("verifyperf under -short") } m := abPilots() dur := abDuration() hz := abHz() - classic := runIOAB(t, "classic", true, m, dur, hz) - runtime.GC() - time.Sleep(250 * time.Millisecond) - gnetRes := runIOAB(t, "gnet", false, m, dur, hz) + gnetRes := runIOAB(t, "gnet", m, dur, hz) - t.Logf("=== A/B summary (M=%d T=%s hz=%.1f) ===", m, dur, hz) - logAB(t, classic) + t.Logf("=== gnet baseline (M=%d T=%s hz=%.1f) ===", m, dur, hz) logAB(t, gnetRes) - t.Logf("goroutine_end classic=%d gnet=%d gnet_saves=%.2fx", - classic.goroutinesEnd, gnetRes.goroutinesEnd, - float64(classic.goroutinesEnd)/float64(max1(gnetRes.goroutinesEnd))) - t.Logf("goroutine_peak classic=%d gnet=%d gnet_saves=%.2fx", - classic.goroutinesPeak, gnetRes.goroutinesPeak, - float64(classic.goroutinesPeak)/float64(max1(gnetRes.goroutinesPeak))) - t.Logf("send/s classic=%.0f gnet=%.0f ratio=%.2f", - classic.sendPerSec, gnetRes.sendPerSec, - gnetRes.sendPerSec/maxF1(classic.sendPerSec)) - t.Logf("recv/s classic=%.0f gnet=%.0f ratio=%.2f", - classic.recvPerSec, gnetRes.recvPerSec, - gnetRes.recvPerSec/maxF1(classic.recvPerSec)) - - // Objective success criteria for the I/O pass: - // 1) fewer goroutines under load (no 2N reader/writer pairs) - // 2) fan-out delivery not worse than classic by >15% - // 3) send throughput not worse than classic by >15% - if gnetRes.goroutinesEnd >= classic.goroutinesEnd { - t.Errorf("gnet end goroutines %d not better than classic %d", - gnetRes.goroutinesEnd, classic.goroutinesEnd) - } - saved := classic.goroutinesPeak - gnetRes.goroutinesPeak - if saved < m { - t.Errorf("expected gnet to save at least ~M=%d peak goroutines, saved only %d (classic peak=%d gnet peak=%d)", - m, saved, classic.goroutinesPeak, gnetRes.goroutinesPeak) - } - if gnetRes.recvPerSec < classic.recvPerSec*0.85 { - t.Errorf("gnet recv/s %.0f is >15%% worse than classic %.0f", - gnetRes.recvPerSec, classic.recvPerSec) - } - if gnetRes.sendPerSec < classic.sendPerSec*0.85 { - t.Errorf("gnet send/s %.0f is >15%% worse than classic %.0f", - gnetRes.sendPerSec, classic.sendPerSec) - } if gnetRes.sends == 0 || gnetRes.recvPos == 0 { t.Fatal("gnet path produced zero traffic") } + if gnetRes.goroutinesEnd < 1 { + t.Fatal("unexpected zero goroutines") + } } func logAB(t *testing.T, r abResult) { @@ -153,9 +118,9 @@ func maxF1(a float64) float64 { return a } -func runIOAB(t *testing.T, name string, classic bool, m int, dur time.Duration, hz float64) abResult { +func runIOAB(t *testing.T, name string, m int, dur time.Duration, hz float64) abResult { t.Helper() - ts := startABServer(t, classic, m) + ts := startABServer(t, m) clients := make([]*fsdclient.Client, m) callsigns := make([]string, m) @@ -350,7 +315,7 @@ func (s *abServer) makeJWT(cid int) (string, error) { return tok.SignedString([]byte(s.secret)) } -func startABServer(t *testing.T, classic bool, m int) *abServer { +func startABServer(t *testing.T, m int) *abServer { t.Helper() gin.SetMode(gin.TestMode) ctx, cancel := context.WithCancel(context.Background()) @@ -436,7 +401,6 @@ func startABServer(t *testing.T, classic bool, m int) *abServer { Metar: metar.New(1, abNoopHTTP{}), Logger: logger, SweatboxEnabled: false, - ForceClassicFSD: classic, FSDBound: fsdAddrCh, HTTPListen: func(network, addr string) (net.Listener, error) { return httpLn, nil diff --git a/internal/server/server.go b/internal/server/server.go index 1e0fbb1..f7af01e 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -10,7 +10,6 @@ import ( "io" "log/slog" "net" - "sync" "time" "github.com/renorris/openfsd/internal/db" @@ -23,23 +22,20 @@ import ( // // # FSD I/O model // -// Production default is the gnet event-driven plane (fixed event-loop count, -// coalesced AsyncWrite outbound). When Deps.Listen is injected or -// ForceClassicFSD is set, the classic net.Listener accept loop runs instead -// (one reader + one SenderWorker per connection) for test harnesses. +// FSD TCP uses the gnet event-driven plane (fixed event-loop count, +// coalesced AsyncWrite outbound). There is no classic net.Listener accept +// path. Tests learn the bound address via Deps.FSDBound (including :0). +// session.SenderWorker remains for sweatbox synthetic sessions (nil Conn). type Server struct { - cfg *Config - users UserStore - configKV ConfigStore - registry Registry - metar MetarQueue - clock Clock - logger *slog.Logger - listen func(ctx context.Context, network, addr string) (net.Listener, error) - // useClassicFSD selects the classic 2-goroutine-per-conn path. - useClassicFSD bool - fsdBound chan<- string - httpListen func(network, addr string) (net.Listener, error) + cfg *Config + users UserStore + configKV ConfigStore + registry Registry + metar MetarQueue + clock Clock + logger *slog.Logger + fsdBound chan<- string + httpListen func(network, addr string) (net.Listener, error) // httpDone is closed when runServiceHTTP returns (after Serve exits). httpDone chan struct{} // sweatbox is the integrated simulator host (nil when disabled). @@ -77,28 +73,20 @@ func New(d Deps) (*Server, error) { if logger == nil { logger = slog.Default() } - listen := d.Listen - if listen == nil { - listen = defaultListen - } - // Classic path when tests inject Listen or ForceClassicFSD. - useClassic := d.ForceClassicFSD || d.Listen != nil s := &Server{ - cfg: d.Config, - users: d.Users, - configKV: d.ConfigKV, - registry: d.Registry, - metar: d.Metar, - clock: clock, - logger: logger, - listen: listen, - useClassicFSD: useClassic, - fsdBound: d.FSDBound, - httpListen: d.HTTPListen, - httpDone: make(chan struct{}), - limits: newConnLimits(), - authFails: newAuthFailLimiter(d.Config.AuthFailMax, d.Config.AuthFailWindow), + cfg: d.Config, + users: d.Users, + configKV: d.ConfigKV, + registry: d.Registry, + metar: d.Metar, + clock: clock, + logger: logger, + fsdBound: d.FSDBound, + httpListen: d.HTTPListen, + httpDone: make(chan struct{}), + limits: newConnLimits(), + authFails: newAuthFailLimiter(d.Config.AuthFailMax, d.Config.AuthFailWindow), } // Two-phase: Server exists so SweatboxHost can hold a back-ref for // unexported broadcast helpers, registry, clock, and logger. @@ -225,11 +213,7 @@ func (s *Server) Run(ctx context.Context) (err error) { // Start HTTP service go s.runServiceHTTP(ctx) - if s.useClassicFSD { - err = s.runClassicFSD(ctx) - } else { - err = s.runGnetFSD(ctx) - } + err = s.runGnetFSD(ctx) // Join service HTTP so callers (and tests) can close shared resources safely. select { @@ -322,84 +306,6 @@ func (s *Server) runGnetFSD(ctx context.Context) error { } } -// runClassicFSD is the legacy accept-loop path (2 goroutines per connection). -func (s *Server) runClassicFSD(ctx context.Context) error { - errCh := make(chan error, len(s.cfg.FsdListenAddrs)) - var listenerWg sync.WaitGroup - - for _, addr := range s.cfg.FsdListenAddrs { - s.logger.Info(fmt.Sprintf("Listening (classic) on %s\n", addr)) - listenerWg.Add(1) - go func(ctx context.Context, addr string) { - defer listenerWg.Done() - s.listenLoop(ctx, addr, errCh) - }(ctx, addr) - } - - go func() { - listenerWg.Wait() - close(errCh) - }() - - var startupErrors []error - for err := range errCh { - startupErrors = append(startupErrors, err) - } - - if len(startupErrors) > 0 { - select { - case <-s.httpDone: - case <-time.After(2 * time.Second): - } - return fmt.Errorf("some listeners failed: %v", startupErrors) - } - - <-ctx.Done() - return nil -} - -func (s *Server) listenLoop(ctx context.Context, addr string, errCh chan<- error) { - listener, err := s.listen(ctx, "tcp4", addr) - if err != nil { - errCh <- fmt.Errorf("failed to listen on %s: %w", addr, err) - return - } - defer listener.Close() - - if s.fsdBound != nil { - select { - case s.fsdBound <- listener.Addr().String(): - default: - } - } - - // Start a goroutine to close the listener when the context is cancelled - go func() { - <-ctx.Done() - listener.Close() - }() - - // Accept connections in a loop - for { - conn, err := listener.Accept() - if err != nil { - if errors.Is(err, net.ErrClosed) { - // Listener was closed due to context cancellation; exit the loop - return - } - // Log or handle non-fatal accept errors - continue - } - // Optional TCP keepalive for half-open detection. - if tc, ok := conn.(*net.TCPConn); ok { - _ = tc.SetKeepAlive(true) - _ = tc.SetKeepAlivePeriod(60 * time.Second) - } - // Handle the connection in another goroutine - go s.handleConn(ctx, conn) - } -} - // Compile-time interface satisfaction checks against production implementors. var ( _ Registry = (*postoffice.PostOffice)(nil) diff --git a/internal/server/testserver.go b/internal/server/testserver.go index f74044e..a5606d8 100644 --- a/internal/server/testserver.go +++ b/internal/server/testserver.go @@ -250,7 +250,7 @@ func StartTestServerOpts(t testing.TB, opts TestServerOptions) *TestServer { } httpAddr := httpLn.Addr().String() - // Bound FSD address reported by gnet OnBoot (or classic listenLoop). + // Bound FSD address reported by gnet OnBoot. fsdAddrCh := make(chan string, 2) cfg := &Config{ @@ -279,7 +279,7 @@ func StartTestServerOpts(t testing.TB, opts TestServerOptions) *TestServer { Metar: metarSvc, Logger: logger, SweatboxEnabled: true, // e2e convenience; empty until airport/scenario load - // Production gnet path (Listen nil). Bound address via FSDBound. + // gnet FSD plane. Bound address via FSDBound. FSDBound: fsdAddrCh, // Return the already-bound listener; never rebind. HTTPListen: func(network, addr string) (net.Listener, error) { @@ -310,7 +310,7 @@ func StartTestServerOpts(t testing.TB, opts TestServerOptions) *TestServer { case <-time.After(8 * time.Second): cancel() _ = sqlDB.Close() - t.Fatal("timeout waiting for FSD listener (gnet OnBoot / classic bind)") + t.Fatal("timeout waiting for FSD listener (gnet OnBoot)") } // gnet may report 0.0.0.0 — clients should dial loopback. if host, port, err := net.SplitHostPort(fsdAddr); err == nil {