mesh-sdk/go/state_counter_test.go

package stdio

import (
	"bufio"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"strings"
	"sync"
	"testing"
	"time"
)

// casRuntime is the runtime's side of mesh/state.get, create and update as the bus does them (novox/hq ADR
// 0201, ADR 0259): one revision per write across the bucket, create refused when the key has a value,
// update refused unless the key is still at the revision named, each refusal answered with CodeChanged.
// refuse, when set, is asked before each write and may refuse it with an error of its own.
type casRuntime struct {
	mu       sync.Mutex
	revision uint64
	keys     map[string]StateEntry
	writes   map[string]int
	lost     int
	refuse   func(verb string) (code int, message string, refused bool)
}

func (r *casRuntime) answer(m message) map[string]any {
	var p struct {
		Key      string          `json:"key"`
		Value    json.RawMessage `json:"value"`
		Revision uint64          `json:"revision"`
	}
	_ = json.Unmarshal(m.Params, &p)
	r.mu.Lock()
	defer r.mu.Unlock()
	verb := strings.TrimPrefix(m.Method, "mesh/state.")
	lost := func() map[string]any {
		r.lost++
		return map[string]any{"error": map[string]any{"code": CodeChanged, "message": p.Key + ": the key was written by another writer first"}}
	}
	if verb != "get" && r.refuse != nil {
		if code, msg, refused := r.refuse(verb); refused {
			return map[string]any{"error": map[string]any{"code": code, "message": msg}}
		}
	}
	e, has := r.keys[p.Key]
	switch verb {
	case "get":
		if !has {
			return map[string]any{"result": nil}
		}
		return map[string]any{"result": e}
	case "create":
		if has {
			return lost()
		}
	case "update":
		if p.Revision == 0 {
			return map[string]any{"error": map[string]any{"code": -32000, "message": "an update names the revision it read"}}
		}
		if !has || e.Revision != p.Revision {
			return lost()
		}
	case "put":
	default:
		return map[string]any{"error": map[string]any{"code": -32601, "message": "no " + m.Method}}
	}
	r.revision++
	r.keys[p.Key] = StateEntry{Key: p.Key, Value: p.Value, Revision: r.revision}
	r.writes[verb]++
	return map[string]any{"result": map[string]any{"revision": r.revision}}
}

// counting serves a bundle whose one tool, "count", runs run, against r; it answers the bundle's asks each
// beside the loop, as the runtime does, so two writers' reads and writes interleave. It returns what the
// tool answered.
func counting(t *testing.T, r *casRuntime, run func() (any, error)) map[string]any {
	t.Helper()
	t.Setenv(ServedAs, "issues")
	inR, inW := io.Pipe()
	outR, outW := io.Pipe()
	go func() {
		_ = ServeOn("", []Tool{{Name: "count", Run: func(map[string]any) (any, error) { return run() }}}, inR, outW)
	}()
	var wmu sync.Mutex
	send := func(m map[string]any) {
		m["jsonrpc"] = "2.0"
		b, _ := json.Marshal(m)
		wmu.Lock()
		defer wmu.Unlock()
		_, _ = inW.Write(append(b, '\n'))
	}
	done := make(chan map[string]any, 1)
	go func() {
		lines := bufio.NewScanner(outR)
		lines.Buffer(make([]byte, 1<<20), 1<<24)
		for lines.Scan() {
			var m message
			if json.Unmarshal(lines.Bytes(), &m) != nil {
				continue
			}
			if strings.HasPrefix(m.Method, "mesh/state.") {
				go func(m message) {
					a := r.answer(m)
					a["id"] = m.ID
					send(a)
				}(m)
				continue
			}
			var answer map[string]any
			_ = json.Unmarshal(lines.Bytes(), &answer)
			if answer["id"] == float64(1) {
				done <- answer
				return
			}
		}
	}()
	send(map[string]any{"id": 1, "method": "tools/call", "params": map[string]any{"name": "count"}})
	t.Cleanup(func() { _ = inW.Close() })
	select {
	case a := <-done:
		return a
	case <-time.After(60 * time.Second):
		t.Fatal("the bundle did not answer the count")
		return nil
	}
}

func text(t *testing.T, a map[string]any) (string, bool) {
	t.Helper()
	res, _ := a["result"].(map[string]any)
	content, _ := res["content"].([]any)
	if len(content) != 1 {
		t.Fatalf("answered %v", a)
	}
	isError, _ := res["isError"].(bool)
	return content[0].(map[string]any)["text"].(string), isError
}

// Two writers incrementing one counter a thousand times each end at two thousand: every number handed out
// once, none lost, each writer's numbers rising — read, then written only at the revision read.
func TestTwoWritersIncrementingOneCounterLoseNoUpdate(t *testing.T) {
	r := &casRuntime{keys: map[string]StateEntry{}, writes: map[string]int{}}
	const each = 1000
	got := counting(t, r, func() (any, error) {
		var wg sync.WaitGroup
		numbers := make([][]uint64, 2)
		errs := make([]error, 2)
		for w := range numbers {
			wg.Add(1)
			go func(w int) {
				defer wg.Done()
				for i := 0; i < each; i++ {
					n, err := State("numbers").Increment("issue", 1000)
					if err != nil {
						errs[w] = err
						return
					}
					numbers[w] = append(numbers[w], n)
				}
			}(w)
		}
		wg.Wait()
		if err := errors.Join(errs...); err != nil {
			return nil, 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])
				}
			}
		}
		return len(seen), nil
	})
	if out, isError := text(t, got); isError || out != "2000" {
		t.Fatalf("two writers counted %s (error: %v)", out, isError)
	}
	var final uint64
	_ = json.Unmarshal(r.keys["issue"].Value, &final)
	if final != 2*each {
		t.Fatalf("the counter ended at %d, not %d: an increment was lost", final, 2*each)
	}
	if r.lost == 0 {
		t.Fatal("the two writers never collided: the test proved nothing about contention")
	}
	t.Logf("2000 increments, %d writes lost to the other writer and tried again", r.lost)
	if r.writes["create"] != 1 || r.writes["update"] != 2*each-1 || r.writes["put"] != 0 {
		t.Fatalf("written by %v: one create, then updates at the revision read, never a put", r.writes)
	}
}

// A lost write is read again and tried again, never reported counted; past its attempts it is ErrChanged
// with nothing counted; any other refusal, and a value that is no counter, is returned at once.
func TestAnIncrementThatLosesIsTriedAgainAndNeverReportedCounted(t *testing.T) {
	t.Run("lost twice then won", func(t *testing.T) {
		lost := 0
		r := &casRuntime{keys: map[string]StateEntry{"issue": {Key: "issue", Value: json.RawMessage(`41`), Revision: 7}},
			writes: map[string]int{}, revision: 7}
		r.refuse = func(string) (int, string, bool) {
			if lost < 2 {
				lost++
				// Another writer was first: the counter moved on by one, to 42 and then 43.
				r.revision++
				r.keys["issue"] = StateEntry{Key: "issue", Value: json.RawMessage(fmt.Sprint(41 + lost)), Revision: r.revision}
				return CodeChanged, "issue: the key was written by another writer first", true
			}
			return 0, "", false
		}
		got := counting(t, r, func() (any, error) { return State("numbers").Increment("issue", 3) })
		if out, isError := text(t, got); isError || out != "44" {
			t.Fatalf("answered %s (error: %v), not 44", out, isError)
		}
	})
	t.Run("lost every time", func(t *testing.T) {
		tries := 0
		r := &casRuntime{keys: map[string]StateEntry{}, writes: map[string]int{}}
		r.refuse = func(string) (int, string, bool) {
			tries++
			return CodeChanged, "issue: the key was written by another writer first", true
		}
		var err error
		got := counting(t, r, func() (any, error) {
			n, e := State("numbers").Increment("issue", 5)
			err = e
			return n, e
		})
		if _, isError := text(t, got); !isError || !errors.Is(err, ErrChanged) || tries != 5 {
			t.Fatalf("after %d tries: %v", tries, err)
		}
	})
	t.Run("refused otherwise", func(t *testing.T) {
		tries := 0
		r := &casRuntime{keys: map[string]StateEntry{}, writes: map[string]int{}}
		r.refuse = func(string) (int, string, bool) {
			tries++
			return -32000, "issues keeps no state numbers", true
		}
		var err error
		got := counting(t, r, func() (any, error) {
			n, e := State("numbers").Increment("issue", 5)
			err = e
			return n, e
		})
		if _, isError := text(t, got); !isError || errors.Is(err, ErrChanged) || tries != 1 {
			t.Fatalf("after %d tries: %v", tries, err)
		}
	})
	t.Run("no counter", func(t *testing.T) {
		r := &casRuntime{keys: map[string]StateEntry{"issue": {Key: "issue", Value: json.RawMessage(`"x"`), Revision: 1}},
			writes: map[string]int{}, revision: 1}
		got := counting(t, r, func() (any, error) { return State("numbers").Increment("issue", 5) })
		if _, isError := text(t, got); !isError || len(r.writes) != 0 {
			t.Fatalf("a value that is no counter was counted: %v, writes %v", got, r.writes)
		}
	})
	t.Run("null", func(t *testing.T) {
		r := &casRuntime{keys: map[string]StateEntry{"issue": {Key: "issue", Value: json.RawMessage(`null`), Revision: 1}},
			writes: map[string]int{}, revision: 1}
		got := counting(t, r, func() (any, error) { return State("numbers").Increment("issue", 5) })
		if _, isError := text(t, got); !isError || len(r.writes) != 0 {
			t.Fatalf("a null was counted from zero: %v, writes %v", got, r.writes)
		}
	})
	t.Run("no attempts", func(t *testing.T) {
		r := &casRuntime{keys: map[string]StateEntry{}, writes: map[string]int{}}
		got := counting(t, r, func() (any, error) { return State("numbers").Increment("issue", 0) })
		if _, isError := text(t, got); !isError || len(r.writes) != 0 {
			t.Fatalf("zero attempts counted: %v", got)
		}
	})
}