sqlutil: fix deadlock in NOTIFY listener (refactor) (#4299)

* refactor: simplify event listening setup and improve notification handling

* fix: pass context to pgx.ConnectConfig in listener setup

* fix: enhance error logging for context.Canceled and add resume channel in listener

* limit connections during parallel tests

Author: mastercactapus

app/listenevents.go

@@ -3,58 +3,17 @@ package app
 import (
 	"context"
 
-	"github.com/jackc/pgx/v5/pgconn"
-	"github.com/pkg/errors"
 	"github.com/target/goalert/permission"
 	"github.com/target/goalert/util/log"
 	"github.com/target/goalert/util/sqlutil"
 )
 
-func (app *App) listenEvents(ctx context.Context) (<-chan struct{}, error) {
-	l, err := sqlutil.NewListener(ctx, app.cfg.LegacyLogger, app.db, "/goalert/config-refresh")
-	if err != nil {
-		return nil, err
-	}
-	app.events = l
-	go func() {
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case err := <-l.Errors():
-				log.Log(ctx, errors.Wrap(err, "listen events"))
-			}
-		}
-	}()
-
-	doneCh := make(chan struct{})
-	go func() {
-		defer close(doneCh)
-		for {
-			var n *pgconn.Notification
-			select {
-			case n = <-l.Notifications():
-			case <-ctx.Done():
-				return
-			}
-			if n == nil {
-				return
-			}
-
-			log.Debugf(log.WithFields(ctx, log.Fields{
-				"Channel": n.Channel,
-				"PID":     n.PID,
-				"Payload": n.Payload,
-			}), "NOTIFY")
-
-			switch n.Channel {
-			case "/goalert/config-refresh":
-				permission.SudoContext(ctx, func(ctx context.Context) {
-					log.Log(ctx, app.ConfigStore.Reload(ctx))
-				})
-			}
-		}
-	}()
-
-	return doneCh, nil
+func (app *App) setupListenEvents() {
+	app.events = sqlutil.NewListener(app.pgx)
+	app.events.Handle("/goalert/config-refresh", func(ctx context.Context, payload string) error {
+		permission.SudoContext(ctx, func(ctx context.Context) {
+			log.Log(ctx, app.ConfigStore.Reload(ctx))
+		})
+		return nil
+	})
 }

app/runapp.go

@@ -31,12 +31,7 @@ func (app *App) _Run(ctx context.Context) error {
 		}
 	}()
 
-	eventCtx, cancel := context.WithCancel(ctx)
-	defer cancel()
-	eventDoneCh, err := app.listenEvents(eventCtx)
-	if err != nil {
-		return err
-	}
+	go app.events.Run(ctx)
 
 	if app.sysAPISrv != nil {
 		app.Logger.InfoContext(ctx, "System API server started.",
@@ -63,17 +58,14 @@ func (app *App) _Run(ctx context.Context) error {
 		slog.String("address", app.l.Addr().String()),
 		slog.String("url", app.ConfigStore.Config().PublicURL()),
 	)
-	err = app.srv.Serve(app.l)
+	err := app.srv.Serve(app.l)
 	if err != nil && !errors.Is(err, http.ErrServerClosed) {
 		return errors.Wrap(err, "serve HTTP")
 	}
 	if app.hSrv != nil {
 		app.hSrv.Resume()
 	}
 
-	select {
-	case <-eventDoneCh:
-	case <-ctx.Done():
-	}
+	<-ctx.Done()
 	return nil
 }

app/startup.go

@@ -117,5 +117,7 @@ func (app *App) startup(ctx context.Context) error {
 		app.Logger.InfoContext(ctx, "SWO Enabled.")
 	}
 
+	app.setupListenEvents()
+
 	return nil
 }

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/hashicorp/yamux v0.1.2
 	github.com/jackc/pgtype v1.14.4
 	github.com/jackc/pgx/v5 v5.7.2
+	github.com/jackc/pgxlisten v0.0.0-20241106001234-1d6f6656415c
 	github.com/jmespath/go-jmespath v0.4.0
 	github.com/joho/godotenv v1.5.1
 	github.com/kffl/speedbump v1.1.0

go.sum

@@ -206,6 +206,8 @@ github.com/jackc/pgx/v4 v4.18.2 h1:xVpYkNR5pk5bMCZGfClbO962UIqVABcAGt7ha1s/FeU=
 github.com/jackc/pgx/v4 v4.18.2/go.mod h1:Ey4Oru5tH5sB6tV7hDmfWFahwF15Eb7DNXlRKx2CkVw=
 github.com/jackc/pgx/v5 v5.7.2 h1:mLoDLV6sonKlvjIEsV56SkWNCnuNv531l94GaIzO+XI=
 github.com/jackc/pgx/v5 v5.7.2/go.mod h1:ncY89UGWxg82EykZUwSpUKEfccBGGYq1xjrOpsbsfGQ=
+github.com/jackc/pgxlisten v0.0.0-20241106001234-1d6f6656415c h1:bTgmg761ac9Ki27HoLx8IBvc+T+Qj6eptBpKahKIRT4=
+github.com/jackc/pgxlisten v0.0.0-20241106001234-1d6f6656415c/go.mod h1:N4E1APLOYrbM11HH5kdqAjDa8RJWVwD3JqWpvH22h64=
 github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=

test/smoke/harness/harness.go

@@ -306,7 +306,7 @@ func (h *Harness) Start() {
 	appCfg.JSON = true
 	appCfg.DBURL = h.dbURL
 	appCfg.TwilioBaseURL = h.twS.URL
-	appCfg.DBMaxOpen = 5
+	appCfg.DBMaxOpen = 3
 	appCfg.SlackBaseURL = h.slackS.URL
 	appCfg.SMTPListenAddr = "localhost:0"
 	appCfg.EmailIntegrationDomain = "smoketest.example.com"
@@ -325,7 +325,7 @@ func (h *Harness) Start() {
 	if err != nil {
 		h.t.Fatalf("failed to parse db url: %v", err)
 	}
-	poolCfg.MaxConns = 5
+	poolCfg.MaxConns = 3
 
 	h.appPool, err = pgxpool.NewWithConfig(ctx, poolCfg)
 	require.NoError(h.t, err, "create pgx pool")

util/log/log.go

@@ -144,7 +144,13 @@ func (l *Logger) Error(ctx context.Context, err error) {
 	}
 
 	ctx = l.addSource(ctx, err)
-	l.entry(ctx).WithError(err).Errorln()
+	lg := l.entry(ctx).WithError(err)
+	if errors.Is(err, context.Canceled) {
+		lg.Debugln()
+		return
+	}
+
+	lg.Errorln()
 }
 
 // Logf will log application information.