Merge pull request 'Keep an event alive while its bundle handles it, and hold one at a time (hq issue 387)' (#26) from fix/387-an-event-is-kept-alive-while-it-is-handled into main

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)
 	}

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)
+		}
+	}
+}