diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 0e25931..ddf22ce 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -883,8 +883,9 @@ func (c *Conn) ConsumeAs(module string, deliver func(Envelope) error) (func(), e // (30 seconds) is for a runtime that died, not a handler that is slow — the runtime gives a handler // launch.EventTimeout. Without it an event handled longer than the ack wait was offered again while // it was handled, and counted toward max-deliver (novox/hq issue 387). - done := make(chan struct{}) + done, stopped := make(chan struct{}), make(chan struct{}) go func() { + defer close(stopped) tick := time.NewTicker(EventHeartbeat) defer tick.Stop() for { @@ -898,6 +899,8 @@ func (c *Conn) ConsumeAs(module string, deliver func(Envelope) error) (func(), e }() err := deliver(env) close(done) + // No in-progress after the answer: one after a Nak would stretch its delay to the whole ack wait. + <-stopped if err != nil { _ = msg.NakWithDelay(NakDelay) return