jochen
c58ce2c7fe
mesh/delivery-group group feat/a-counter-by-compare-and-set checking: 0 of 2 member(s) ready
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate pass: builds mesh-sdk, sent nowhere; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (typescript), go/merge-check.sh (go)
The coming issues module hands out issue numbers from its own state; two instances must never hand out one number twice. Create and Update already write by compare-and-set (ADR 0259), so State(...).Increment reads, writes one more at the revision read, and on a lost write reads and tries again, bounded by the caller's attempts. A lost write past the bound is ErrChanged, any other refusal and a value that is no counter are returned at once: nothing is reported counted that was not written. Part of the issue tracker decision, hq research 042.
267 lines
8.6 KiB
Go
267 lines
8.6 KiB
Go
package stdio
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"errors"
|
|
"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
|
|
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 {
|
|
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.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.
|
|
r.revision++
|
|
r.keys["issue"] = StateEntry{Key: "issue", Value: json.RawMessage(`42`), Revision: r.revision}
|
|
if lost == 2 {
|
|
r.keys["issue"] = StateEntry{Key: "issue", Value: json.RawMessage(`43`), 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("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)
|
|
}
|
|
})
|
|
}
|