mesh-tools/node-tools/internal/runtime/state_counter_test.go

package runtime

import (
	"encoding/json"
	"errors"
	"fmt"
	"sync"
	"testing"

	"github.com/novox/mesh-tools/node-tools/internal/bus"
	"github.com/novox/mesh-tools/node-tools/internal/launch"
	mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
)

// counterBus is a module, issues, that keeps a state numbers in a bucket of a real bus, reached as a bundle
// reaches it: mesh/state.<verb> with its JSON, through the module's bus.
func counterBus(t *testing.T) launch.Bus {
	t.Helper()
	mesh := mt.New(t)
	mesh.Bucket(t, "issues_numbers")
	m := mt.MembershipOf("issues", "anchor", false, nil)
	m.State = []bus.StateIssued{{Name: "numbers", Bucket: "issues_numbers", Writes: true}}
	mesh.Issue(t, m)
	conn := connect(t, "node-tools", "anchor")
	conn.Follow("issues")
	mt.Until(t, func() error {
		if conn.Membership("issues") == nil {
			return errors.New("not issued yet")
		}
		return nil
	})
	return (&consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}).forModule("issues")
}

func ask(b launch.Bus, verb string, params map[string]any) (json.RawMessage, error) {
	raw, _ := json.Marshal(params)
	return b.State(verb, raw)
}

// increment is what a bundle's counter does over the wire: get, then create (no value) or update at the
// revision read; a lost write — told by its code alone — is read again and tried again.
func increment(b launch.Bus, key string, attempts int) (n uint64, lost int, err error) {
	for i := 0; i < attempts; i++ {
		raw, err := ask(b, "get", map[string]any{"state": "numbers", "key": key})
		if err != nil {
			return 0, lost, err
		}
		var e *bus.StateEntry
		if err := json.Unmarshal(raw, &e); err != nil {
			return 0, lost, err
		}
		if e == nil {
			n = 1
			_, err = ask(b, "create", map[string]any{"state": "numbers", "key": key, "value": n})
		} else {
			if err := json.Unmarshal(e.Value, &n); err != nil {
				return 0, lost, err
			}
			n++
			_, err = ask(b, "update", map[string]any{"state": "numbers", "key": key, "value": n, "revision": e.Revision})
		}
		var coded interface{ ErrorCode() int }
		switch {
		case err == nil:
			return n, lost, nil
		case errors.As(err, &coded) && coded.ErrorCode() == launch.CodeChanged:
			lost++
		default:
			return 0, lost, err
		}
	}
	return 0, lost, fmt.Errorf("%s lost %d times", key, attempts)
}

// Two writers incrementing one counter a thousand times each, against a real bus, end at two thousand:
// every number handed out once and none lost, because the runtime writes only at the revision read
// (novox/hq research 042: an issue's number is handed out once).
func TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate(t *testing.T) {
	b := counterBus(t)
	const each = 1000
	var wg sync.WaitGroup
	numbers := make([][]uint64, 2)
	errs := make([]error, 2)
	lost := make([]int, 2)
	for w := range numbers {
		wg.Add(1)
		go func(w int) {
			defer wg.Done()
			for i := 0; i < each; i++ {
				n, l, err := increment(b, "issue", 1000)
				lost[w] += l
				if err != nil {
					errs[w] = err
					return
				}
				numbers[w] = append(numbers[w], n)
			}
		}(w)
	}
	wg.Wait()
	if err := errors.Join(errs...); err != nil {
		t.Fatal(err)
	}
	seen := map[uint64]bool{}
	for w, ns := range numbers {
		for i, n := range ns {
			if seen[n] {
				t.Errorf("the number %d was handed out twice", n)
			}
			seen[n] = true
			if i > 0 && n <= ns[i-1] {
				t.Errorf("writer %d was handed %d after %d", w, n, ns[i-1])
			}
		}
	}
	raw, err := ask(b, "get", map[string]any{"state": "numbers", "key": "issue"})
	if err != nil {
		t.Fatal(err)
	}
	var e bus.StateEntry
	_ = json.Unmarshal(raw, &e)
	if string(e.Value) != "2000" || len(seen) != 2*each {
		t.Fatalf("the counter ended at %s with %d numbers handed out, not 2000: an increment was lost", e.Value, len(seen))
	}
	t.Logf("2000 increments, %d writes lost to the other writer and tried again", lost[0]+lost[1])
}

// A write at a revision that is no longer the key's is refused with the lost-write code and writes
// nothing; create is refused for a key with a value and taken for one deleted; an update names the
// revision it read, never zero.
func TestAStaleWriteIsRefusedAndWritesNothing(t *testing.T) {
	b := counterBus(t)
	state := func(verb string, extra map[string]any) (uint64, error) {
		p := map[string]any{"state": "numbers", "key": "issue"}
		for k, v := range extra {
			p[k] = v
		}
		raw, err := ask(b, verb, p)
		if err != nil {
			return 0, err
		}
		var r struct {
			Revision uint64 `json:"revision"`
		}
		_ = json.Unmarshal(raw, &r)
		return r.Revision, nil
	}
	isLost := func(err error) bool {
		var coded interface{ ErrorCode() int }
		return errors.As(err, &coded) && coded.ErrorCode() == launch.CodeChanged && errors.Is(err, bus.ErrStateChanged)
	}
	value := func() string {
		raw, err := ask(b, "get", map[string]any{"state": "numbers", "key": "issue"})
		if err != nil {
			t.Fatal(err)
		}
		var e *bus.StateEntry
		_ = json.Unmarshal(raw, &e)
		if e == nil {
			return "none"
		}
		return string(e.Value)
	}

	if _, err := state("update", map[string]any{"value": 1, "revision": 0}); err == nil || isLost(err) {
		t.Fatalf("an update at revision zero: %v", err)
	}
	if value() != "none" {
		t.Fatal("an update at revision zero wrote")
	}
	first, err := state("create", map[string]any{"value": 1})
	if err != nil || first == 0 {
		t.Fatalf("create of a key with no value: %d %v", first, err)
	}
	if _, err := state("create", map[string]any{"value": 9}); !isLost(err) {
		t.Fatalf("a second create: %v", err)
	}
	second, err := state("update", map[string]any{"value": 2, "revision": first})
	if err != nil || second <= first {
		t.Fatalf("an update at the revision read: %d %v", second, err)
	}
	if _, err := state("update", map[string]any{"value": 9, "revision": first}); !isLost(err) {
		t.Fatalf("an update at a stale revision: %v", err)
	}
	if _, err := state("update", map[string]any{"value": 9, "revision": second + 100}); !isLost(err) {
		t.Fatalf("an update at a revision not yet written: %v", err)
	}
	if v := value(); v != "2" {
		t.Fatalf("refused writes changed the value to %s", v)
	}
	if _, err := ask(b, "delete", map[string]any{"state": "numbers", "key": "issue"}); err != nil {
		t.Fatal(err)
	}
	if _, err := state("create", map[string]any{"value": 1}); err != nil {
		t.Fatalf("create of a deleted key: %v", err)
	}
	if v := value(); v != "1" {
		t.Fatalf("after delete and create: %s", v)
	}
}