Wait for IRC listener cleanup and HTTP shutdown on stop (closes #159)

Stopping the server returned before the IRC listener's connections had
run their cleanup, and the database closed under them, so their
sessions, clients and channel memberships stayed and their nicks stayed
taken after a restart.

The IRC listener's stop now waits up to five seconds for every
connection's cleanup, which now runs on a context that stopping does
not cancel. The HTTP server's stop hook now runs Shutdown itself and
waits for it; httpServer is set before serving starts, and the server's
own signal watcher, which raced fx's stop hooks, is gone.

Model: opus-5-5

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)

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(

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)

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

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

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