From aef908b57b818e983010eb574b7af94309bdd447 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 10 Oct 2026 17:18:35 +0200 Subject: [PATCH 1/2] Keep an event alive while its bundle handles it, and hold one at a time (hq issue 387) The consumer's ack wait is 30 seconds and the runtime gives a handler two minutes, so an event handled longer was offered again while it was handled; events waiting in the client's buffer ran out their ack wait too, so a burst reached max-deliver with nothing failed. Say in-progress every 10 seconds, as a worker's work does, and pull one event at a time. --- node-tools/internal/bus/bus.go | 29 +++++++++++- node-tools/internal/bus/consume_test.go | 62 +++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 2 deletions(-) create mode 100644 node-tools/internal/bus/consume_test.go diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 5e0a0a8..0e25931 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,31 @@ 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 := make(chan struct{}) + go func() { + tick := time.NewTicker(EventHeartbeat) + defer tick.Stop() + for { + select { + case <-done: + return + case <-tick.C: + _ = msg.InProgress() + } + } + }() + err := deliver(env) + close(done) + 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) + } + } +} From 560876c07435bae61f57c0b1b8b260e0df94963c Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 10 Oct 2026 17:21:54 +0200 Subject: [PATCH 2/2] No in-progress after an event is answered (review, hq issue 387) --- node-tools/internal/bus/bus.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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