2026-10-10 10:37:11 +00:00
3 changed files with 328 additions and 0 deletions
@@ -104,6 +104,9 @@ publishes it under the module's name; `Cancel`; `Where` reads its state from the
 `Take` hand the module each warrant once — marked acted on by compare-and-set before the module acts, and only
 for an ask it made, an option it offered, at that option's level, before it expired. A lost compare-and-set is
 `stdio.ErrChanged`, told by the runtime's error code.
+`State(...).Increment(key, attempts)` is a counter on top of the two (for numbers handed out once, such as an
+issue's): it reads the key, writes one more at the revision read, and on a lost write reads and tries again, at
+most `attempts` times; past that it is `stdio.ErrChanged` and nothing was counted.

 ### `primitives` — sealing, semver, resolved-env
 Seal/unseal for secrets, semantic version comparison, and the accessor a module uses to read its
@@ -2,6 +2,8 @@ package stdio

 import (
 	"encoding/json"
+	"errors"
+	"fmt"
 	"time"
 )

@@ -74,6 +76,51 @@ func (s KeptState) Delete(key string) error {
 	return err
 }

+// Increment adds one to a counter kept at key and returns the number it now holds: one for a key with no
+// value. Read, then written only if the key is still at the revision read (Create when it had none, Update
+// when it had one), so of any number of writers each number is handed out once and no increment is lost.
+// A write another writer won is read again and tried again, at most attempts times in all; past that the
+// error is ErrChanged and nothing was counted. Any other error is returned at once, as is a value that is
+// not a whole number (null included): nothing is ever reported counted that was not written. The reverse
+// does not hold: an error that is not ErrChanged — a timeout above all — may come after the write landed, so
+// that number may be spent and never handed out. A caller tolerates gaps, never a number handed out twice.
+// Deleting the key starts the count again at one; a key whose numbers must stay unique is never deleted.
+func (s KeptState) Increment(key string, attempts int) (uint64, error) {
+	if attempts < 1 {
+		return 0, fmt.Errorf("an increment of %s is tried at least once: %d attempts", key, attempts)
+	}
+	for i := 0; i < attempts; i++ {
+		e, err := s.Get(key)
+		if err != nil {
+			return 0, err
+		}
+		var next uint64 = 1
+		if e == nil {
+			_, err = s.Create(key, next)
+		} else {
+			var n uint64
+			if string(e.Value) == "null" {
+				return 0, fmt.Errorf("%s holds null, not a counter", key)
+			}
+			if err := json.Unmarshal(e.Value, &n); err != nil {
+				return 0, fmt.Errorf("%s holds %s, not a counter: %w", key, e.Value, err)
+			}
+			if n == ^uint64(0) {
+				return 0, fmt.Errorf("%s is at the largest number a counter holds", key)
+			}
+			next = n + 1
+			_, err = s.Update(key, next, e.Revision)
+		}
+		if err == nil {
+			return next, nil
+		}
+		if !errors.Is(err, ErrChanged) {
+			return 0, err
+		}
+	}
+	return 0, fmt.Errorf("%s was written by another writer first %d times: %w", key, attempts, ErrChanged)
+}
+
 // Keys is every key with a current value, sorted.
 func (s KeptState) Keys() ([]string, error) {
 	raw, err := s.ask("keys", map[string]any{})
@@ -0,0 +1,278 @@
+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)
+		}
+	})
+}