Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions pkg/repository/multiplexer.go
Original file line number Diff line number Diff line change
Expand Up @@ -57,12 +57,19 @@ func (m *multiplexedListener) startListening() {
// listen for multiplexed messages
listener := &pgxlisten.Listener{
Connect: func(ctx context.Context) (*pgx.Conn, error) {
// Acquire a new connection each time
// Acquire a connection from the pool, then Hijack it so pgxlisten owns
// its lifecycle. pgxlisten Close()s this conn on every reconnect (e.g.
// server-side idle-session termination). Without Hijack, the
// *pgxpool.Conn wrapper returned by Acquire falls out of scope without
// Release(), so pgxpool permanently counts the slot as acquired and
// leaks one slot per reconnect cycle until Acquire blocks at MaxConns
// (#3694). Hijack removes the conn from the pool's accounting entirely,
// so pgxlisten closing it leaks nothing.
poolConn, err := m.pool.Acquire(ctx)
if err != nil {
return nil, err
}
return poolConn.Conn(), nil
return poolConn.Hijack(), nil

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

cc @mnafees -- wdyt?

},
LogError: func(innerCtx context.Context, err error) {
m.l.Warn().Err(err).Msg("error in listener")
Expand Down