diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 5e0a0a8..ddf22ce 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -848,6 +848,12 @@ func ConsumerOf(node, module string) string { // (novox/hq issue 330). var NakDelay = 5 * time.Second +// EventHeartbeat is how often an event a bundle is still handling is said to be in progress, well inside the +// consumer's ack wait. The client holds one event at a time (PullMaxMessages(1)): events are handed over in +// order, one at a time, so an event waiting in the client's buffer would have its ack wait run out before it is +// handed over, and a burst would end in max-deliver with nothing failed (novox/hq issue 387). +var EventHeartbeat = 10 * time.Second + // ConsumeAs reads a module's durable consumer and hands each event to deliver (novox/hq ADR 0198): // acknowledged when deliver returns nil, negatively acknowledged after NakDelay when it returns an // error, terminated when it is not an event at all. The consumer is the controller's to create; this @@ -873,12 +879,34 @@ func (c *Conn) ConsumeAs(module string, deliver func(Envelope) error) (func(), e _ = msg.Term() return } - if err := deliver(env); err != nil { + // Kept alive while the bundle works on it, as a worker's work is (seat.go): the consumer's ack wait + // (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, stopped := make(chan struct{}), make(chan struct{}) + go func() { + defer close(stopped) + tick := time.NewTicker(EventHeartbeat) + defer tick.Stop() + for { + select { + case <-done: + return + case <-tick.C: + _ = msg.InProgress() + } + } + }() + 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 } _ = msg.Ack() - }) + }, jetstream.PullMaxMessages(1)) if err != nil { return nil, fmt.Errorf("reading %s's consumer %s: %w", module, durable, err) } diff --git a/node-tools/internal/bus/consume_test.go b/node-tools/internal/bus/consume_test.go new file mode 100644 index 0000000..4c29d44 --- /dev/null +++ b/node-tools/internal/bus/consume_test.go @@ -0,0 +1,62 @@ +package bus_test + +import ( + "encoding/json" + "fmt" + "sync" + "testing" + "time" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// novox/hq issue 387: an event a bundle handles for longer than the consumer's ack wait is kept alive while +// it is handled, and the events behind it wait on the bus rather than in the client, so a slow handler and a +// burst are each handed over once — never offered again with nothing failed, never toward max-deliver. +func TestASlowHandlerAndABurstAreEachHandedOverOnce(t *testing.T) { + mesh := mt.New(t) + restore := bus.EventHeartbeat + bus.EventHeartbeat = 300 * time.Millisecond + t.Cleanup(func() { bus.EventHeartbeat = restore }) + mesh.Consumer(t, "anchor", "slow", []string{"mesh.mod.alpha.event.>"}, time.Second) + c, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + c.Logf = func(string, ...any) {} + t.Cleanup(c.Close) + + var mu sync.Mutex + handled := map[int]int{} + stop, err := c.ConsumeAs("slow", func(env bus.Envelope) error { + var body struct{ N int } + _ = json.Unmarshal(env.Body, &body) + time.Sleep(1500 * time.Millisecond) // longer than the ack wait + mu.Lock() + handled[body.N]++ + mu.Unlock() + return nil + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(stop) + for n := 1; n <= 3; n++ { + mesh.Emit(t, "mesh.mod.alpha.event.happened", map[string]any{"n": n}) + } + mt.Until(t, func() error { + ack, notYet := mesh.Pending(t, "anchor", "slow") + if ack != 0 || notYet != 0 { + return fmt.Errorf("still pending: %d unacknowledged, %d undelivered", ack, notYet) + } + return nil + }) + mu.Lock() + defer mu.Unlock() + for n := 1; n <= 3; n++ { + if handled[n] != 1 { + t.Errorf("event %d was handed over %d times: %v", n, handled[n], handled) + } + } +}