flightsimulatorplatfrom/openfsd
1
0
Fork 0
mirror of https://github.com/renorris/openfsd synced 2026-08-12 04:15:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

server: remove classic FSD accept/event path; gnet only

Browse Source 
Delete handleConn/eventLoop/readLoginPackets, runClassicFSD/listenLoop,
and Deps.Listen/ForceClassicFSD/Server.listen/useClassicFSD. Always run
gnet. Rewrite TestRunServiceHTTPAndListen to use FSDBound select
readiness. Convert verifyperf to gnet-only baseline. Keep shared auth,
broadcast, MOTD helpers and SenderWorker for sweatbox synthetics.
This commit is contained in:
Reese Norris
2026-07-27 13:53:18 -04:00
parent b1f4d09291
commit b7fbae4577
13 changed files with 73 additions and 398 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
internal/server/bootstrap_test.go
View File

@@ -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 }

3
internal/server/config.go
View File

@@ -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.

182
internal/server/conn.go
View File

@@ -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
}

14
internal/server/deps.go
View File

@@ -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)
}

3
internal/server/deps_test.go
View File

@@ -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())
}

5
internal/server/gnet_fsd.go
View File

@@ -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

2
internal/server/gnet_fsd_test.go
View File

@@ -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)

2
internal/server/handler_admin.go
View File

@@ -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)

2
internal/server/handler_position.go
View File

@@ -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.

2
internal/server/http_service.go
View File

@@ -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)

60
internal/server/io_ab_verify_test.go
View File

@@ -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

146
internal/server/server.go
View File

@@ -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)

6
internal/server/testserver.go
View File

@@ -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 {