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,120 @@ 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.GetChannelMemberIDs(ctx, channelID)
+	if err != nil {
+		t.Fatalf("get members: %v", err)
+	}
+
+	if len(members) != 1 || members[0] != peerID {
+		t.Errorf(
+			"members of #shut after stop: %v, want only the peer %d",
+			members, peerID,
+		)
+	}
+
+	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
+}