mesh-tools/node-tools/internal/bus/consume_test.go

package bus_test

import (
	"encoding/json"
	"fmt"
	"sync"
	"sync/atomic"
	"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)
		}
	}
}

// The runtime tells a bundle how its module's consumer stands between events (stdio.OnHeartbeat): a word at each
// idle heartbeat while the bus answers, once Missed when the bus ended the loop, and a word again when it is bound again.
func TestAConsumerSaysItIsBoundWhileIdleAndOnceWhenTheBusEndedIt(t *testing.T) {
	mesh := mt.New(t)
	restore := bus.IdleHeartbeat
	bus.IdleHeartbeat = 200 * time.Millisecond
	t.Cleanup(func() { bus.IdleHeartbeat = restore })
	ticks := make(chan struct{}, 64)
	t.Cleanup(bus.SetTickHook(func() {
		select {
		case ticks <- struct{}{}:
		default:
		}
	}))
	mesh.Consumer(t, "anchor", "quiet", []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
	var beats []bus.Beat
	stop, err := c.ConsumeAsBeating("quiet", func(bus.Envelope) error { return nil }, func(b bus.Beat) {
		mu.Lock()
		beats = append(beats, b)
		mu.Unlock()
	})
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(stop)
	// A consume nobody asked a heartbeat of says nothing and asks the bus nothing, however many intervals pass.
	var unaskedBeats atomic.Int32
	unasked, err := c.ConsumeAsBeatingWhen("quiet", func(bus.Envelope) error { return nil },
		func(bus.Beat) { unaskedBeats.Add(1) }, func() bool { return false })
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(unasked)
	last := func() (bus.Beat, int) {
		mu.Lock()
		defer mu.Unlock()
		if len(beats) == 0 {
			return bus.Beat{}, 0
		}
		return beats[len(beats)-1], len(beats)
	}
	mt.Until(t, func() error {
		if b, n := last(); n == 0 || b.Missed || b.At.IsZero() {
			return fmt.Errorf("no word that the consumer is bound yet: %d beats", n)
		}
		return nil
	})
	mesh.RemoveConsumer(t, bus.EventsStream, "anchor", "quiet")
	mt.Until(t, func() error {
		if b, _ := last(); !b.Missed || b.Why == "" {
			return fmt.Errorf("the consumer is gone from the bus and no Missed word came: %+v", b)
		}
		return nil
	})
	mu.Lock()
	missed := 0
	for _, b := range beats {
		if b.Missed {
			missed++
		}
	}
	mu.Unlock()
	for n := 0; n < 3; n++ { // three more intervals pass, counted by the loop's own tick
		select {
		case <-ticks:
		case <-time.After(10 * time.Second):
			t.Fatal("the heartbeat loop handled no more intervals")
		}
	}
	mu.Lock()
	defer mu.Unlock()
	after := 0
	for _, b := range beats {
		if b.Missed {
			after++
		}
	}
	if unaskedBeats.Load() != 0 {
		t.Errorf("a consume with no heartbeat asked said %d words", unaskedBeats.Load())
	}
	if after != missed || missed != 1 {
		t.Errorf("Missed is said once, not at every beat: %d then %d", missed, after)
	}
}