Fix Publish and Flush calls to not fail on async error (slow consumer)

[kozlovic]

• Revisited the use of nc.err to limit it to report async errors.
• Publish and Flush calls will no longer fail if an asyn error occurs
• Flush will report an error if connection is closed/lost during flush
• Flush calls are failed during a reconnect
• Tests updated

[TursMA]

Guys, can you explain me how we handle error if flush returns error.

if err := Bw.Flush(); err != nil {
                if nc.err == nil {
                    nc.err = err // <--------    what happens later with nc.err?
                }
            }

Who processes error in nc.err after assignee? Why we don't make a call of processOpErr?

My background: I have some microservices which live on different hosts. Nats is mq. I've faced with dropping messages in nats flow and now I'm trying to find all possible cases. I have very low message rate and it's not slow customer. Messages disappear. I make a EncodedConn.Request and don't receive message in subscriber. Subscribtion occurs before publishing. I have just one connection to nats per microservice. So I have to receive message, but...
Another suspucted place I found is gnatsd deliverMsg function.

writeErr:
    if deadlineSet {
        client.nc.SetWriteDeadline(time.Time{})
    }
    client.mu.Unlock()

    if ne, ok := err.(net.Error); ok && ne.Timeout() {
        atomic.AddInt64(&client.srv.slowConsumers, 1)
        client.Noticef("Slow Consumer Detected")
        client.closeConnection()
    } else {
        c.Debugf("Error writing msg: %v", err) //<--------- just printing error without disconnecting or reporting to stats.
    }

[derekcollison]

Run the server with -DV, that should help you trace where the messages are being lost. What is the timeout on the request?

I have really only seen dropped messages via slow consumers, but if the request timeout is too short that could be a cause.

[TursMA]

Thank you for very quick response!
timeout is default - 2 seconds.
Tomorrow I'll try to reproduce with -DV flags.
JFIY possible errors on write operations (first google page) - http://man7.org/linux/man-pages/man2/write.2.html

What about error handling for Flush? There are a lot of logic for transmitting error to nc.err and then just ignore it.

[kozlovic]

It's not ignored. The code you refer to is the flusher, which is a go-routine. There is no way to return directly the error condition there. So this is done through nc.err, which you can get by calling LastError() from anywhere.

[TursMA]

Why we do not do reconnect in the same way like on Read error?
LastError() is useless because of around goroutines. We don't know a moment when I have to do a call.

[kozlovic]

Well, that's the thing (and discussed in the PR) we let the readLoop do the reconnect (it should get an error on read too if the socket is gone)