fix: close database pools and notification listener on shutdown

Pool closing was commented out ("@@ to be fixed, now blocks"), so the
process exited with every pgx pool open and the listener's dedicated
connection up: Postgres saw abrupt connection resets on each stop, and
connection slots lingered until TCP teardown.

Two things made Close() block:

- pgxpool.Pool.Close() waits for acquired connections, and with the old
  1-second shutdown window HTTP.Shutdown regularly gave up while
  handlers were still running and holding connections.
- NotificationListener.Stop() only closed a channel that the listen
  loop checked between notifications, while the loop normally sits
  blocked inside WaitForNotification — the goroutine (and its
  connection) survived Stop.

Now the listener's Stop cancels the loop context, which interrupts
WaitForNotification, closes the connection cleanly with a bounded
fresh context (so Postgres receives a Terminate message), and waits
for the goroutine to exit — also when racing another Stop.
Server.Shutdown then closes the engine after the HTTP drain completes
— at that point no handler holds a connection — bounding the wait by
the shutdown context (plus a 5s backstop when it has no deadline) so
an expired drain window cannot hang the process past the supervisor's
grace period.

Author: coroluca

CHANGELOG.md

@@ -20,6 +20,7 @@
 * Session watcher no longer deadlocks when paused.
 * **SIGTERM** — the server now shuts down gracefully on SIGTERM (the default stop signal of Docker, Kubernetes and systemd), not only on SIGINT/Ctrl-C. A raw SIGTERM previously terminated the process immediately, cutting in-flight requests and resetting database connections.
 * **Exit status** — a graceful shutdown now exits with status 0; it previously exited 1, which supervisors interpret as a crash.
+* **Database connections on shutdown** — connection pools and the notification listener's dedicated connection are now closed after the HTTP server drains, so Postgres sees clean disconnects instead of connection resets on every stop. `NotificationListener.Stop()` now interrupts a pending `WaitForNotification` and waits for the listener goroutine to exit before returning.
 
 ## 0.8.1 - 2026-07-08
 

database/notification_listener.go

@@ -35,11 +35,11 @@ type NotificationListener struct {
 	config     *ListenerConfig
 	logger     *logging.Logger
 	connString string
-	conn       *pgx.Conn
 	handlers   map[string]NotificationHandler
 	mu         sync.RWMutex
 	running    bool
-	stop       chan struct{}
+	cancel     context.CancelFunc
+	stopped    chan struct{}
 }
 
 // NewNotificationListener creates a new notification listener
@@ -52,7 +52,6 @@ func NewNotificationListener(connString string, config *ListenerConfig, logger *
 		logger:     logger,
 		connString: connString,
 		handlers:   make(map[string]NotificationHandler),
-		stop:       make(chan struct{}),
 	}
 }
 
@@ -78,44 +77,52 @@ func (l *NotificationListener) Start(ctx context.Context) error {
 		return fmt.Errorf("listener is already running")
 	}
 	l.running = true
+	ctx, l.cancel = context.WithCancel(ctx)
+	l.stopped = make(chan struct{})
+	stopped := l.stopped
 	l.mu.Unlock()
 
-	go l.listenLoop(ctx)
+	go func() {
+		defer close(stopped)
+		l.listenLoop(ctx)
+	}()
 	return nil
 }
 
-// Stop stops the listener
+// Stop stops the listener, interrupting a pending WaitForNotification, and
+// waits for the loop to exit, so its dedicated connection is closed when
+// Stop returns — also when racing another Stop. It must not be called from
+// a notification handler, which runs on the loop it would wait for.
 func (l *NotificationListener) Stop() {
 	l.mu.Lock()
-	defer l.mu.Unlock()
 	if l.running {
-		close(l.stop)
 		l.running = false
+		l.cancel()
+	}
+	stopped := l.stopped
+	l.mu.Unlock()
+	if stopped != nil {
+		<-stopped
 	}
 }
 
 // listenLoop maintains the connection and processes notifications
 func (l *NotificationListener) listenLoop(ctx context.Context) {
 	for {
-		select {
-		case <-ctx.Done():
-			return
-		case <-l.stop:
+		err := l.connectAndListen(ctx)
+		if ctx.Err() != nil {
+			// Stopped: WaitForNotification reports the canceled
+			// context as an error, not worth logging
 			return
-		default:
-			err := l.connectAndListen(ctx)
-			if err != nil {
-				if l.logger != nil {
-					l.logger.Err(err).Msg("Notification listener error")
-				}
-				select {
-				case <-time.After(l.config.ReconnectDelay):
-					continue
-				case <-ctx.Done():
-					return
-				case <-l.stop:
-					return
-				}
+		}
+		if err != nil {
+			if l.logger != nil {
+				l.logger.Err(err).Msg("Notification listener error")
+			}
+			select {
+			case <-time.After(l.config.ReconnectDelay):
+			case <-ctx.Done():
+				return
 			}
 		}
 	}
@@ -127,9 +134,13 @@ func (l *NotificationListener) connectAndListen(ctx context.Context) error {
 	if err != nil {
 		return fmt.Errorf("failed to connect: %w", err)
 	}
-	defer conn.Close(ctx)
-
-	l.conn = conn
+	defer func() {
+		// Close cleanly even when ctx is already canceled (i.e. during
+		// Stop), so Postgres sees a Terminate message instead of a reset
+		closeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second)
+		defer cancel()
+		conn.Close(closeCtx)
+	}()
 
 	_, err = conn.Exec(ctx, fmt.Sprintf("LISTEN %s", l.config.Channel))
 	if err != nil {
@@ -141,20 +152,12 @@ func (l *NotificationListener) connectAndListen(ctx context.Context) error {
 	}
 
 	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case <-l.stop:
-			return nil
-		default:
-			notification, err := conn.WaitForNotification(ctx)
-			if err != nil {
-				return fmt.Errorf("error waiting for notification: %w", err)
-			}
-
-			if notification != nil {
-				l.handleNotification(ctx, notification.Payload)
-			}
+		notification, err := conn.WaitForNotification(ctx)
+		if err != nil {
+			return fmt.Errorf("error waiting for notification: %w", err)
+		}
+		if notification != nil {
+			l.handleNotification(ctx, notification.Payload)
 		}
 	}
 }

database/notification_listener_test.go

@@ -0,0 +1,37 @@
+package database
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// Stop must interrupt the listener even while it is blocked in
+// WaitForNotification, and return only once the loop has exited and its
+// dedicated connection is closed. It used to leave both behind: the stop
+// channel was only checked between notifications.
+func TestListenerStopInterruptsWait(t *testing.T) {
+	l := NewNotificationListener(dbe.config.URL, nil, nil)
+	if err := l.Start(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	// Let the listener connect and block waiting for notifications
+	time.Sleep(200 * time.Millisecond)
+
+	done := make(chan struct{})
+	go func() {
+		l.Stop()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("Stop did not return: listener stuck in WaitForNotification")
+	}
+
+	// The listener must be restartable after a stop
+	if err := l.Start(context.Background()); err != nil {
+		t.Fatalf("restart after Stop failed: %v", err)
+	}
+	l.Stop()
+}

server/server.go

@@ -117,8 +117,24 @@ func (s *Server) Shutdown(ctx context.Context) {
 		}
 		// Close goroutines (for now just the checker in the service manager)
 		close(s.shutdown)
-		// Close database pools - ok, not so graceful
-		// s.DBE.Close() @@ to be fixed, now blocks
+		// Release database resources: the notification listener and all the
+		// connection pools. Handlers have completed by now, so pool Close()
+		// does not wait on acquired connections — unless the graceful window
+		// expired with requests still running, so bound the wait (by ctx,
+		// plus a backstop when ctx has no deadline) rather than hang past
+		// the supervisor's grace period.
+		closed := make(chan struct{})
+		go func() {
+			s.DBE.Close()
+			close(closed)
+		}()
+		select {
+		case <-closed:
+		case <-ctx.Done():
+			s.logger.Warn().Msg("Shutdown context expired while closing database pools")
+		case <-time.After(5 * time.Second):
+			s.logger.Warn().Msg("Timed out closing database pools")
+		}
 		close(s.shutdownCompleted)
 	})
 }

server/shutdown_test.go

@@ -79,6 +79,11 @@ func TestSIGTERMStartsGracefulShutdown(t *testing.T) {
 		t.Fatal(err)
 	}
 	waitStopped(t, done, 10*time.Second)
+
+	// Shutdown must also have closed the database pools
+	if _, err := s.DBE.AcquireConnection(context.Background()); err == nil {
+		t.Fatal("expected the main pool to be closed after shutdown")
+	}
 }
 
 func getStatus(t *testing.T, url string) int {