redigo error log: write: connection reset by peer

Question

Almost the same amount of time (point in time as redigo error log: write: connection reset by peer?), redis error log:
Client id=45183 addr=127.0.0.1:40420 fd=39 name= age=39706 idle=46 flags=N db=0 sub=8 psub=0 multi=-1 qbuf=0 qbuf-free=0 obl=16114 oll=528 omem=8545237 events=rw cmd=ping scheduled to be closed ASAP for overcoming of output buffer limits.

go error log

write tcp 127.0.0.1:40806->127.0.0.1:6379: write: connection reset by peer

Before that, the Go program didn't receive the subscription message for about 7 minutes. I presume it was a cache overflow caused by messages not being consumed.

The Redis client-output-buffer-limit is the default configuration. The linux fd and connection count are normal, and I can't find of a reason for the unconsumable.

Here is my code:

server.go

func WaitFroMsg(ctx context.Context, pool *redis.Pool, onMessage func(channel string, data []byte) error, channel ...string) (err error) {
    conn := pool.Get()
    psc := redis.PubSubConn{Conn: conn}
    if err := psc.Subscribe(redis.Args{}.AddFlat(channel)...); err != nil {
        return err
    }
    done := make(chan error, 1)
    go func() {
        for {
            switch n := psc.Receive().(type) {
            case error:
                done <- fmt.Errorf("redis pubsub receive err: %v", n)
                return
            case redis.Message:
                if err = onMessage(n.Channel, n.Data); err != nil {
                    done <- err
                    return
                }
            case redis.Subscription:
                if n.Count == 0 {
                    fmt.Println("all channels are unsubscribed", channel)
                    done <- nil
                    return
                }
            }
        }
    }()
    const healthCheck = time.Minute
    ticker := time.NewTicker(healthCheck) 
    defer ticker.Stop()
    for {
        select {
        case <-ticker.C:
            if err = psc.Ping(""); err != nil {
                fmt.Println("healthCheck ", err, channel)
                return err
            }
        case err := <-done:
            return err
        case <-ctx.Done(): 
            if err := psc.Unsubscribe(); err != nil {
                return fmt.Errorf("redis　unsubscribe failed: %v", err)
            }
            return nil
        }
    }
}

pool.go

func NewPool(addr string, db int) *redis.Pool {
    return &redis.Pool{
        MaxIdle:     3,
        IdleTimeout: 240 * time.Second,
        Dial: func() (redis.Conn, error) {
            c, err := redis.Dial("tcp", addr)
            if err != nil {
                return nil, err
            }
            if _, err = c.Do("SELECT", db); err != nil {
                c.Close()
                return nil, err
            }
            return c, nil
        },
        TestOnBorrow: func(c redis.Conn, t time.Time) error {
            if time.Since(t) < time.Minute {
                return nil
            }
            _, err := c.Do("PING")
            fmt.Println("PING error", err)
            return err
        },
    }
}