From 1eaa1abbe9c9d5411b6d02fd4821ca34bf9d131a Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 10 Oct 2026 12:31:06 +0200 Subject: [PATCH] Prove a write by compare-and-set loses no update under contention, on a real bus (hq research 042) The issues module will hand out numbers through mesh/state.create and update. Two writers incrementing one counter a thousand times each through the module's bus now end at two thousand with every number handed out once; a stale, zero or future revision is refused with the lost-write code and writes nothing; a deleted key can be created again; and the lost-write code reaches a bundle wrapped or not, and no other refusal carries it. Both runtime tests run in the check's bundleless list, so mesh/repo-check runs them. Part of the issue tracker decision, hq research 042. --- merge-check.sh | 5 +- node-tools/internal/launch/error_code_test.go | 30 +++ .../internal/runtime/state_counter_test.go | 200 ++++++++++++++++++ 3 files changed, 233 insertions(+), 2 deletions(-) create mode 100644 node-tools/internal/launch/error_code_test.go create mode 100644 node-tools/internal/runtime/state_counter_test.go diff --git a/merge-check.sh b/merge-check.sh index fa55750..1bd567e 100644 --- a/merge-check.sh +++ b/merge-check.sh @@ -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 diff --git a/node-tools/internal/launch/error_code_test.go b/node-tools/internal/launch/error_code_test.go new file mode 100644 index 0000000..ace06ec --- /dev/null +++ b/node-tools/internal/launch/error_code_test.go @@ -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) + } +} diff --git a/node-tools/internal/runtime/state_counter_test.go b/node-tools/internal/runtime/state_counter_test.go new file mode 100644 index 0000000..cd18788 --- /dev/null +++ b/node-tools/internal/runtime/state_counter_test.go @@ -0,0 +1,200 @@ +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. 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) + } +}