Mar 25, 2025 · Mar 25, 2025 · Mar 25, 2025 · Mar 25, 2025
diff --git a/coderd/database/pubsub/pubsub.go b/coderd/database/pubsub/pubsub.go
 p.connected.Set(0)
 // Creates a new listener using pq.
 var (
 errCh  = make(chan error)
 dialer = logDialer{
 logger: p.logger,
 // pq.defaultDialer uses a zero net.Dialer as well.
 dc.Dialer(dialer)
 }

 var (
 errCh     = make(chan error, 1)
 sentErrCh = false
 )
 p.pgListener = pqListenerShim{
 Listener: pq.NewConnectorListener(connector, connectURL, time.Second, time.Minute, func(t pq.ListenerEventType, err error) {
 switch t {
 p.logger.Error(ctx, "pubsub failed to connect to postgres", slog.Error(err))
 }
 // This callback gets events whenever the connection state changes.
 // Don't send if the errChannel has already been closed.
 select {
 case <-errCh:
 // Only send the first error.
 if sentErrCh {
 return
 default:
 errCh <- err
 close(errCh)
 }
 errCh <- err // won't block because we are buffered.
 sentErrCh = true
 }),
 }
 select {
 case err:= <-errCh:
 case err = <-errCh:
 if err != nil {
 _ = p.pgListener.Close()
 return xerrors.Errorf("create pq listener: %w", err)
Original file line number	Diff line number	Diff line change
Expand Up		@@ -492,7 +492,6 @@ func (p *PGPubsub) startListener(ctx context.Context, connectURL string) error {
		p.connected.Set(0)
		// Creates a new listener using pq.
		var (
		errCh = make(chan error)
		dialer = logDialer{
		logger: p.logger,
		// pq.defaultDialer uses a zero net.Dialer as well.
Expand DownExpand Up		@@ -525,6 +524,10 @@ func (p *PGPubsub) startListener(ctx context.Context, connectURL string) error {
		dc.Dialer(dialer)
		}

		var (
		errCh = make(chan error, 1)
		sentErrCh = false
		)
		p.pgListener = pqListenerShim{
		Listener: pq.NewConnectorListener(connector, connectURL, time.Second, time.Minute, func(t pq.ListenerEventType, err error) {
		switch t {
Expand All		@@ -541,18 +544,16 @@ func (p *PGPubsub) startListener(ctx context.Context, connectURL string) error {
		p.logger.Error(ctx, "pubsub failed to connect to postgres", slog.Error(err))
		}
		// This callback gets events whenever the connection state changes.
		// Don't send if the errChannel has already been closed.
		select {
		case <-errCh:
		// Only send the first error.
		if sentErrCh {
		return
		default:
		errCh <- err
		close(errCh)
		}
		errCh <- err // won't block because we are buffered.
		sentErrCh = true
Copy link Member mafredriMar 25, 2025 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others.Learn more. Do we need to protect this or is sequential execution guaranteed? Copy link ContributorAuthor spikecurtisMar 25, 2025 Choose a reason for hiding this comment The reason will be displayed to describe this comment to others.Learn more. It executes sequentially, at least by inspection. There is always a chance they could change the implementation and call-back concurrently, but it didn't seem worth while defending for that.
		}),
		}
		select {
		case err:= <-errCh:
		case err = <-errCh:
		if err != nil {
		_ = p.pgListener.Close()
		return xerrors.Errorf("create pq listener: %w", err)
Expand Down