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.
This commit is contained in:
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user