pgdriver: discarding bad listener connection

[ivanakimov]

When running something similar to this, sometimes I notice:

pgdriver: 2022/06/15 07:16:00 listener.go:321: pgdriver: &{%!s(*context.emptyCtx=0xc00003a038) %!s(*pgdriver.Listener=&{0xc00007c720 0xc001473620 [queue_channel bun:ping] {0 0} 0xc0002af2c0 false 0xc008666a80}) %!s(int=100) %!s(time.Duration=5000000000) %!s(time.Duration=60000000000) %!s(chan pgdriver.Notification=0xc00007c900) %!s(chan struct {}=0xc003a7e000)} channel is full for 1m0s (notification is dropped)

pgdriver: 2022/06/15 07:16:00 listener.go:115: bun: discarding bad listener connection: bun: ping timeout

Two questions:

1. When do full channels and bad listener connections happen?
2. Is there anything to do about these, or are they simply notifications?

[vmihailenco]

That error happens when you can't consume all messages from the Listener channel fast enough. Probably the consuming loop crashed or is blocked for too long.

[comerc]

I got a stable reproduction of the error.

Change this lines: https://github.com/uptrace/bun/blob/master/example/pg-listen/main.go#L41-L43

To above:

	// if err := ln.Listen(ctx, "mychan1", "mychan2"); err != nil {
	// 	panic(err)
	// }

	for {
		select {
		case notif := <-ln.Channel():
			fmt.Println(notif.Channel, notif.Payload)
		}
	}

[john8329]

More like "make it like the example in the link". select was the cause of my problem for this, you need one goroutine with a range loop for each channel you listen to.

[kissejau]

I also caught this error.

In my case, code looks like:

type WrappedListener struct {
	ln *pgdriver.Listener
}

func NewWrappedListener() *WrappedListener {
	var ln *pgdriver.Listener

	// any init code here

	return &WrappedListener{ln: ln}
}

func (wl *WrappedListener) Start() {
	for {
		select {
		case <-wl.ln.Channel():
			// any code here
		}
	}
}

But *pgdriver.Listener.Channel is not a Getter. Its a method that initializes wrapped channel and returns channel.

And correct code looks like:

type WrappedListener struct {
	ln             *pgdriver.Listener
	notificationCh <-chan pgdriver.Notification
}

func NewWrappedListener() *WrappedListener {
	var ln *pgdriver.Listener

	// any init code here

	notificationCh := ln.Channel()
	
	return &WrappedListener{ln: ln, notificationCh: notificationCh}
}

func (wl *WrappedListener) Start() {
	for {
		select {
		case <-wl.notificationCh:
			// any code here
		}
	}
}

I think its not correct. *pgdriver.Listener.Channel func must have a name like CreateChannel/NewChannel OR it must return wrapped channel that already initialized in parent's struct - *pgdriver.Listener.

I think 2nd way cant be done, bc it is necessary to support creating multiple listening channels from a single listener. I suggest renaming the method for creating the channel to CreateChannel (bc its not constructor, its method of *pgdriver.Listener struct).