Merge pull request 'Prove a write by compare-and-set loses no update under contention, on a real bus (hq research 042)' (#25) from feat/a-counter-by-compare-and-set into main

2026-10-10 12:37:42 +00:00
parent 83c1994183 d0128adf5b
commit c761c22c2a
3 changed files with 236 additions and 2 deletions
@@ -34,8 +34,9 @@ if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; t
 	# and how a search ranks (novox/hq ADR 0245).
 	bundleless='^(TestASeatAndAModuleOfOneNameAreEachReached|TestSearch.*|TestASearchByCommandSaysWhatItReplaces)$'
 	# And what of the runtime launches no bundle: how a bundle's seat traffic reaches the bus, what its
-	# environment holds, and what a runtime on a module's own account serves (novox/hq ADR 0259).
-	runtimeless='^(TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode)$'
+	# environment holds, what a runtime on a module's own account serves (novox/hq ADR 0259), and that a write
+	# by compare-and-set loses no update under contention (novox/hq research 042).
+	runtimeless='^(TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing)$'
 fi
 if command -v gcc >/dev/null 2>&1; then
 	CGO_ENABLED=1 go test -race -count=1 $packages
@@ -0,0 +1,30 @@
+package launch
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+)
+
+type lostWrite struct{}
+
+func (lostWrite) Error() string {
+	return "issues.numbers.issue: the key was written by another writer first"
+}
+func (lostWrite) ErrorCode() int { return CodeChanged }
+
+// A lost compare-and-set is answered to a bundle with its own code, wrapped or not, and the SDK tells it by
+// that code alone (mesh-sdk's stdio.CodeChanged); every other refusal is the general code, never that one.
+func TestALostWriteIsAnsweredWithItsOwnCodeAndNothingElseIs(t *testing.T) {
+	if CodeChanged != -32010 {
+		t.Fatalf("the lost-write code is %d; mesh-sdk tells it by -32010", CodeChanged)
+	}
+	for _, err := range []error{lostWrite{}, fmt.Errorf("asked by issues: %w", lostWrite{})} {
+		if got := errorCode(err); got != CodeChanged {
+			t.Errorf("%v is answered %d", err, got)
+		}
+	}
+	if got := errorCode(errors.New("issues keeps no state numbers")); got == CodeChanged || got != -32000 {
+		t.Errorf("a plain refusal is answered %d", got)
+	}
+}
@@ -0,0 +1,203 @@
+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))
+	}
+	if lost[0]+lost[1] == 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", 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)
+	}
+}