diff --git a/internal/ircserver/conn.go b/internal/ircserver/conn.go index bba99ca..e11d09b 100644 --- a/internal/ircserver/conn.go +++ b/internal/ircserver/conn.go @@ -222,7 +222,9 @@ func (c *Conn) serve( ) { ctx, cancel := context.WithCancel(ctx) defer cancel() - defer c.cleanup(ctx) + // Stopping the server cancels ctx, and the cleanup must + // still end the session then. + defer c.cleanup(context.WithoutCancel(ctx)) ping := time.AfterFunc(interval, func() { c.send("PING :" + c.serverSfx) diff --git a/internal/ircserver/export_test.go b/internal/ircserver/export_test.go index db02999..11f4b8d 100644 --- a/internal/ircserver/export_test.go +++ b/internal/ircserver/export_test.go @@ -85,6 +85,10 @@ const PingInterval = pingInterval // for a line before it closes the connection. const PongDeadline = pongDeadline +// StopTimeout is the longest Stop waits for the connections' +// cleanup. +const StopTimeout = stopTimeout + // NewTestConn returns a Conn for nick on tcpConn, without a // server, database or service behind it. func NewTestConn( diff --git a/internal/ircserver/server.go b/internal/ircserver/server.go index 8d5ce86..3459860 100644 --- a/internal/ircserver/server.go +++ b/internal/ircserver/server.go @@ -16,6 +16,10 @@ import ( "sneak.berlin/go/neoirc/internal/service" ) +// stopTimeout is the longest stop waits for the connections' +// cleanup, which ends their sessions. +const stopTimeout = 5 * time.Second + // Params defines the dependencies for creating an IRC // Server. type Params struct { @@ -40,6 +44,10 @@ type Server struct { conns map[*Conn]struct{} cancel context.CancelFunc + // serving counts the connections whose serve, cleanup + // included, has not returned yet. + serving sync.WaitGroup + // pingInterval and pongDeadline are the constants of the // same names, except in tests, which shorten them. pingInterval time.Duration @@ -63,6 +71,7 @@ func New( listener: nil, cancel: nil, mu: sync.Mutex{}, + serving: sync.WaitGroup{}, pingInterval: pingInterval, pongDeadline: pongDeadline, @@ -112,7 +121,10 @@ func (s *Server) start(ctx context.Context, addr string) error { return nil } -// stop shuts down the listener and all connections. +// stop shuts down the listener and all connections, then +// waits up to stopTimeout for every connection's cleanup to +// finish, so that their sessions are ended before the +// database closes. func (s *Server) stop() { if s.cancel != nil { s.cancel() @@ -127,6 +139,22 @@ func (s *Server) stop() { c.conn.Close() //nolint:errcheck,gosec } s.mu.Unlock() + + done := make(chan struct{}) + + go func() { + s.serving.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(stopTimeout): + s.log.Error( + "irc connections not cleaned up before stop timeout", + "timeout", stopTimeout, + ) + } } // acceptLoop accepts new connections. @@ -152,10 +180,24 @@ func (s *Server) acceptLoop(ctx context.Context) { ) s.mu.Lock() + + // stop cancels ctx before it takes s.mu to close the + // connections, so a connection accepted while stop + // runs is either closed and waited for by stop or + // closed here. + if ctx.Err() != nil { + s.mu.Unlock() + tcpConn.Close() //nolint:errcheck,gosec + + return + } + s.conns[client] = struct{}{} + s.serving.Add(1) s.mu.Unlock() go func() { + defer s.serving.Done() defer func() { s.mu.Lock() delete(s.conns, client) diff --git a/internal/ircserver/server_test.go b/internal/ircserver/server_test.go index e7f92d4..0e62431 100644 --- a/internal/ircserver/server_test.go +++ b/internal/ircserver/server_test.go @@ -4,6 +4,7 @@ import ( "bufio" "crypto/rand" "database/sql" + "errors" "fmt" "log/slog" "net" @@ -812,6 +813,117 @@ func TestQuit(t *testing.T) { assertContains(t, lines, "goodbye", "QUIT reason") } +// TestStopEndsSessions checks that stopping the server ends +// a registered connection's session before Stop returns, as +// closing the connection does: the session, its client and +// its channel membership are gone from the database, and a +// channel peer has been sent the QUIT. +func TestStopEndsSessions(t *testing.T) { + t.Parallel() + + env := newTestEnv(t) + client := env.dial(t) + + client.register("shutirc") + client.joinAndDrain("#shut") + + ctx := t.Context() + + // A peer on the HTTP API, whose session outlives the stop. + peerID, peerClientID, _, err := env.database.CreateSession( + ctx, "peer", "peer", "", "", + ) + if err != nil { + t.Fatalf("create peer session: %v", err) + } + + channelID, err := env.database.GetChannelByName(ctx, "#shut") + if err != nil { + t.Fatalf("get channel: %v", err) + } + + err = env.database.JoinChannel(ctx, channelID, peerID) + if err != nil { + t.Fatalf("join peer: %v", err) + } + + env.srv.Stop() + + _, err = env.database.GetSessionByNick(ctx, "shutirc") + if !errors.Is(err, sql.ErrNoRows) { + t.Errorf("look up shutirc after stop: %v, want no rows", err) + } + + clients, err := env.database.GetClientCount(ctx) + if err != nil { + t.Fatalf("count clients: %v", err) + } + + if clients != 1 { + t.Errorf("%d clients after stop, want only the peer's", clients) + } + + members, err := env.database.CountChannelMembers(ctx, channelID) + if err != nil { + t.Fatalf("count members: %v", err) + } + + if members != 1 { + t.Errorf("%d members of #shut after stop, want only the peer", members) + } + + msgs, _, err := env.database.PollMessages(ctx, peerClientID, 0, 0) + if err != nil { + t.Fatalf("poll peer: %v", err) + } + + for _, msg := range msgs { + if msg.Command == "QUIT" && msg.From == "shutirc" { + return + } + } + + t.Errorf("peer was not sent the QUIT of shutirc: %v", msgs) +} + +// TestStopReturnsWhileCleanupIsStuck checks that Stop stops +// waiting for a connection's cleanup after StopTimeout, so a +// cleanup that cannot finish does not hold up stopping. +func TestStopReturnsWhileCleanupIsStuck(t *testing.T) { + t.Parallel() + + env := newTestEnv(t) + client := env.dial(t) + + client.register("stuck") + + // The test database has one connection. While the test + // holds it, the cleanup's first query waits for it. + held, err := env.database.GetDB().Conn(t.Context()) + if err != nil { + t.Fatalf("take the database connection: %v", err) + } + + stopped := make(chan struct{}) + + go func() { + env.srv.Stop() + close(stopped) + }() + + select { + case <-stopped: + case <-time.After(ircserver.StopTimeout + testTimeout): + t.Error("Stop did not return while a cleanup was stuck") + } + + // Lets the cleanup finish before the database is closed. + err = held.Close() + if err != nil { + t.Errorf("release the database connection: %v", err) + } +} + func TestTopicSetAndGet(t *testing.T) { t.Parallel() diff --git a/internal/server/server.go b/internal/server/server.go index d19f869..49bcd50 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -45,11 +45,8 @@ type Params struct { // It manages routing, middleware, and lifecycle. type Server struct { startupTime time.Time - exitCode int sentryEnabled bool log *slog.Logger - ctx context.Context //nolint:containedctx // signal handling pattern - cancelFunc context.CancelFunc httpServer *http.Server router *chi.Mux params Params @@ -78,13 +75,16 @@ func New( srv.enableSentry() srv.SetupRoutes() - // The start hook's context ends when the hook - // returns; serving must outlive it. - go srv.serve() //nolint:contextcheck,gosec // G118 + // A closed stdout or stderr must not end the process. + signal.Ignore(syscall.SIGPIPE) + + srv.startHTTP() return nil }, - OnStop: func(_ context.Context) error { + OnStop: func(ctx context.Context) error { + srv.cleanShutdown(ctx) + return nil }, }) @@ -129,44 +129,15 @@ func (srv *Server) enableSentry() { srv.sentryEnabled = true } -func (srv *Server) serve() int { - srv.ctx, srv.cancelFunc = context.WithCancel( - context.Background(), - ) - - go func() { - sigCh := make(chan os.Signal, 1) - - signal.Ignore(syscall.SIGPIPE) - signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM) - - sig := <-sigCh - - srv.log.Info("signal received", "signal", sig) - - if srv.cancelFunc != nil { - srv.cancelFunc() - } - }() - - go srv.serveUntilShutdown() - - <-srv.ctx.Done() - - srv.cleanShutdown() - - return srv.exitCode -} - func (srv *Server) cleanupForExit() { srv.log.Info("cleaning up") } -func (srv *Server) cleanShutdown() { - srv.exitCode = 0 - +// cleanShutdown stops the HTTP server, waiting up to +// shutdownTimeout for the requests in progress to finish. +func (srv *Server) cleanShutdown(ctx context.Context) { ctxShutdown, shutdownCancel := context.WithTimeout( - context.Background(), shutdownTimeout, + ctx, shutdownTimeout, ) err := srv.httpServer.Shutdown(ctxShutdown) @@ -191,7 +162,11 @@ func (srv *Server) configure() { // Server configuration placeholder. } -func (srv *Server) serveUntilShutdown() { +// startHTTP creates the HTTP server and serves it from a new +// goroutine. srv.httpServer is set before that goroutine +// starts and never changes, so it and cleanShutdown read it +// without a lock. +func (srv *Server) startHTTP() { listenAddr := fmt.Sprintf( ":%d", srv.params.Config.Port, ) @@ -204,16 +179,16 @@ func (srv *Server) serveUntilShutdown() { Handler: srv, } + go srv.serveUntilShutdown() +} + +func (srv *Server) serveUntilShutdown() { srv.log.Info( - "http begin listen", "listenaddr", listenAddr, + "http begin listen", "listenaddr", srv.httpServer.Addr, ) err := srv.httpServer.ListenAndServe() if err != nil && !errors.Is(err, http.ErrServerClosed) { srv.log.Error("listen error", "error", err) - - if srv.cancelFunc != nil { - srv.cancelFunc() - } } } diff --git a/internal/server/server_test.go b/internal/server/server_test.go new file mode 100644 index 0000000..c740ba1 --- /dev/null +++ b/internal/server/server_test.go @@ -0,0 +1,247 @@ +package server_test + +import ( + "crypto/rand" + "fmt" + "net" + "net/http" + "strconv" + "testing" + "time" + + "go.uber.org/fx" + "go.uber.org/fx/fxtest" + "sneak.berlin/go/neoirc/internal/broker" + "sneak.berlin/go/neoirc/internal/config" + "sneak.berlin/go/neoirc/internal/db" + "sneak.berlin/go/neoirc/internal/globals" + "sneak.berlin/go/neoirc/internal/handlers" + "sneak.berlin/go/neoirc/internal/healthcheck" + "sneak.berlin/go/neoirc/internal/logger" + "sneak.berlin/go/neoirc/internal/middleware" + "sneak.berlin/go/neoirc/internal/server" + "sneak.berlin/go/neoirc/internal/service" + "sneak.berlin/go/neoirc/internal/stats" +) + +const ( + testTimeout = 5 * time.Second + + // checkStep is how often waitUntil checks again. + checkStep = 10 * time.Millisecond + + // stopHoldTime is how long a test waits to see that the + // app has not stopped. + stopHoldTime = 200 * time.Millisecond +) + +// TestStopWaitsForRequestInProgress checks that stopping the +// app, which runs the stop hook server.New registers, waits +// for a long poll the HTTP server is serving to finish, and +// that the poll gets its response. +func TestStopWaitsForRequestInProgress(t *testing.T) { + t.Parallel() + + port := freePort(t) + addr := net.JoinHostPort("127.0.0.1", strconv.Itoa(port)) + + app, brk, database := startApp(t, port) + + sessionID, _, token, err := database.CreateSession( + t.Context(), "poller", "poller", "", "", + ) + if err != nil { + t.Fatalf("create session: %v", err) + } + + waitUntil(t, "the HTTP server to listen", func() bool { + return canConnect(t, addr) + }) + + statuses := make(chan int, 1) + + go func() { + statuses <- getAuthenticated( + t, "http://"+addr+"/api/v1/messages?timeout=30", token, + ) + }() + + waitUntil(t, "the long poll to wait for a message", func() bool { + return brk.ListenerCount(sessionID) == 1 + }) + + var stopErr error + + stopped := make(chan struct{}) + + go func() { + stopErr = app.Stop(t.Context()) + + close(stopped) + }() + + select { + case <-stopped: + t.Error("the app stopped while a request was in progress") + case <-time.After(stopHoldTime): + } + + brk.Notify(sessionID) + + status := <-statuses + if status != http.StatusOK { + t.Errorf("status %d, want %d", status, http.StatusOK) + } + + select { + case <-stopped: + if stopErr != nil { + t.Errorf("stop: %v", stopErr) + } + case <-time.After(testTimeout): + t.Fatal("the app did not stop after the request finished") + } +} + +// startApp starts the server with the dependencies +// cmd/neoircd gives it, on a fresh in-memory database, with +// the HTTP server listening on port. +func startApp( + t *testing.T, + port int, +) (*fxtest.App, *broker.Broker, *db.Database) { + t.Helper() + + var ( + brk *broker.Broker + database *db.Database + ) + + app := fxtest.New(t, + fx.Supply(&globals.Globals{ + Appname: "neoirc-test", + Version: "test", + StartTime: time.Now(), + }), + fx.Provide( + broker.New, + config.New, + db.New, + handlers.New, + healthcheck.New, + logger.New, + middleware.New, + server.New, + service.New, + stats.New, + ), + fx.Decorate(func(cfg *config.Config) *config.Config { + // A random name, so that no other test can share it. + cfg.DBURL = fmt.Sprintf( + "file:%s?mode=memory&cache=shared", rand.Text(), + ) + cfg.Port = port + + return cfg + }), + fx.Invoke(func(*server.Server) {}), + fx.Populate(&brk, &database), + ) + + app.RequireStart() + + // Stops the app if the test ends before stopping it. + t.Cleanup(app.RequireStop) + + return app, brk, database +} + +// freePort returns a TCP port that nothing was listening on a +// moment ago. +func freePort(t *testing.T) int { + t.Helper() + + var listenConfig net.ListenConfig + + listener, err := listenConfig.Listen( + t.Context(), "tcp", "127.0.0.1:0", + ) + if err != nil { + t.Fatalf("listen: %v", err) + } + + tcpAddr, ok := listener.Addr().(*net.TCPAddr) + if !ok { + t.Fatalf("listener address %v is not TCP", listener.Addr()) + } + + err = listener.Close() + if err != nil { + t.Fatalf("close listener: %v", err) + } + + return tcpAddr.Port +} + +// canConnect reports whether addr accepts a TCP connection. +func canConnect(t *testing.T, addr string) bool { + t.Helper() + + var dialer net.Dialer + + conn, err := dialer.DialContext(t.Context(), "tcp", addr) + if err != nil { + return false + } + + _ = conn.Close() + + return true +} + +// waitUntil calls done every checkStep until it returns true, +// and fails the test if it does not within testTimeout. +func waitUntil(t *testing.T, what string, done func() bool) { + t.Helper() + + deadline := time.Now().Add(testTimeout) + + for !done() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", what) + } + + time.Sleep(checkStep) + } +} + +// getAuthenticated sends a GET request to url as the session +// whose auth token is token, and returns the response status, +// or 0 if there was no response. +func getAuthenticated(t *testing.T, url, token string) int { + t.Helper() + + request, err := http.NewRequestWithContext( + t.Context(), http.MethodGet, url, nil, + ) + if err != nil { + t.Errorf("new request: %v", err) + + return 0 + } + + request.Header.Set("Cookie", "neoirc_auth="+token) + + client := http.Client{Timeout: testTimeout} + + response, err := client.Do(request) + if err != nil { + t.Errorf("get %s: %v", url, err) + + return 0 + } + + _ = response.Body.Close() + + return response.StatusCode +}