mesh-issues: hand out decision and research numbers, so sessions stop taking them by reading hq (hq issue 424)
mesh/merge-gate pass: builds mesh-issues → novox; 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 (go), merge-check-typescript.sh (typescript)
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate pass: builds mesh-issues → novox; 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 (go), merge-check-typescript.sh (typescript)
mesh/delivery superseded: a newer head of the same pull request
A new verb, issue-tracker.reserve {kind, for} -> {kind, number}: one counter
per sequence beside the issue counter, advanced the same way, started above
what hq's record holds, a floor setting for open branches and the
reservations kept, and raised when hq holds a higher number. A reserved
number is spent forever, kept with who, when and for what, and said as the
event reserved (hq ADR 0275 rule 8, ADR 0290 rule 5).
This commit is contained in:
@@ -203,6 +203,20 @@ func (i *instance) answer(m rpc) {
|
||||
i.evMu.Unlock()
|
||||
return map[string]any{}, nil
|
||||
}
|
||||
if m.Method == "mesh/ask" {
|
||||
// The records module, as a reservation asks it what numbers hq holds (issue 424).
|
||||
var a struct {
|
||||
Key string `json:"key"`
|
||||
Body struct {
|
||||
Folder string `json:"folder"`
|
||||
} `json:"body"`
|
||||
}
|
||||
if err := json.Unmarshal(m.Params, &a); err != nil || a.Key != "records.records_list" {
|
||||
return nil, fmt.Errorf("the runtime answers no ask %s in this test", a.Key)
|
||||
}
|
||||
listing := hqToday()
|
||||
return map[string]any{"folders": listing.folders[a.Body.Folder], "documents": listing.docs[a.Body.Folder]}, nil
|
||||
}
|
||||
kv := i.b.kvs[p.State]
|
||||
if kv == nil {
|
||||
return nil, fmt.Errorf("the state %q is not declared", p.State)
|
||||
@@ -380,6 +394,87 @@ func TestFiftyOpensAtOnceFromTwoInstancesGetFiftyDistinctConsecutiveNumbers(t *t
|
||||
}
|
||||
}
|
||||
|
||||
// Two instances on one bus, forty decision numbers reserved at the same moment: forty numbers, each once,
|
||||
// consecutive from one above the highest decision hq holds, a reservation and an event for each (issue 424).
|
||||
func TestFortyReservationsAtOnceFromTwoInstancesGetFortyDistinctConsecutiveNumbers(t *testing.T) {
|
||||
b := newBus(t)
|
||||
settings := writeSettings(t, 365)
|
||||
var events []event
|
||||
var evMu sync.Mutex
|
||||
instances := []*instance{b.start(t, settings, &events, &evMu), b.start(t, settings, &events, &evMu)}
|
||||
const reserves = 40
|
||||
numbers := make([]uint64, reserves)
|
||||
var wg sync.WaitGroup
|
||||
start := make(chan struct{})
|
||||
for k := 0; k < reserves; k++ {
|
||||
wg.Add(1)
|
||||
go func(k int) {
|
||||
defer wg.Done()
|
||||
<-start
|
||||
raw, err := instances[k%2].call("reserve", map[string]any{"kind": "decision", "for": fmt.Sprintf("record %d", k)})
|
||||
if err != nil {
|
||||
t.Errorf("reserve %d: %v", k, err)
|
||||
return
|
||||
}
|
||||
var r struct {
|
||||
Kind string `json:"kind"`
|
||||
Number uint64 `json:"number"`
|
||||
}
|
||||
_ = json.Unmarshal(raw, &r)
|
||||
if r.Kind != "decision" {
|
||||
t.Errorf("reserve %d answered %s", k, raw)
|
||||
}
|
||||
numbers[k] = r.Number
|
||||
}(k)
|
||||
}
|
||||
close(start)
|
||||
wg.Wait()
|
||||
if t.Failed() {
|
||||
return
|
||||
}
|
||||
sorted := append([]uint64(nil), numbers...)
|
||||
sort.Slice(sorted, func(i, j int) bool { return sorted[i] < sorted[j] })
|
||||
for k, n := range sorted {
|
||||
if n != uint64(292+k) {
|
||||
t.Fatalf("the numbers handed out are %v: not forty distinct numbers from 292 in a row", sorted)
|
||||
}
|
||||
}
|
||||
t.Logf("%d writes lost to another caller and tried again, %d of them the counters' updates", b.lost.Load(), b.counterLost.Load())
|
||||
if b.counterLost.Load() == 0 {
|
||||
t.Fatal("no counter update was lost to another caller: the test did not make the reservations contend")
|
||||
}
|
||||
ctx := context.Background()
|
||||
if e, err := b.kvs[CountersState].Get(ctx, KindDecision); err != nil || string(e.Value()) != "331" {
|
||||
t.Fatalf("the decision counter holds %v (%v), want 331", e, err)
|
||||
}
|
||||
if _, err := b.kvs[CountersState].Get(ctx, IssueCounter); !errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
t.Fatalf("a reservation touched the issue counter: %v", err)
|
||||
}
|
||||
for k, n := range numbers {
|
||||
e, err := b.kvs[ReservationsState].Get(ctx, ReservationKey(KindDecision, n))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var r Reservation
|
||||
_ = json.Unmarshal(e.Value(), &r)
|
||||
if r.Number != n || r.For != fmt.Sprintf("record %d", k) || r.By != Unproven {
|
||||
t.Fatalf("reservation %d holds %+v, not reserve %d's", n, r, k)
|
||||
}
|
||||
}
|
||||
evMu.Lock()
|
||||
defer evMu.Unlock()
|
||||
said := map[float64]bool{}
|
||||
for _, ev := range events {
|
||||
if ev.Key != EventReserved || ev.Headers["x-source"] != "mesh-issues" {
|
||||
t.Fatalf("event %+v", ev)
|
||||
}
|
||||
said[ev.Body["number"].(float64)] = true
|
||||
}
|
||||
if len(said) != reserves {
|
||||
t.Fatalf("%d reserved events for distinct numbers, want %d", len(said), reserves)
|
||||
}
|
||||
}
|
||||
|
||||
// One instance, every verb over the bus: what an agent does with an issue from opening to resolving.
|
||||
func TestAnIssueIsOpenedNotedLinkedMovedAndListedOverTheBus(t *testing.T) {
|
||||
b := newBus(t)
|
||||
|
||||
@@ -22,7 +22,7 @@ import (
|
||||
const Seat = "issue-tracker"
|
||||
|
||||
// Verbs are the seat's verbs, as the manifest declares and claims them.
|
||||
var Verbs = []string{"open", "get", "list", "move", "note", "link", "import", "groom", "set", "tracking"}
|
||||
var Verbs = []string{"open", "get", "list", "move", "note", "link", "import", "groom", "set", "tracking", "reserve"}
|
||||
|
||||
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
|
||||
|
||||
@@ -47,6 +47,12 @@ type Settings struct {
|
||||
// WorkerStaleAfter is how long an issue's worker stays current with nothing done on it (ADR 0283 §2), as a
|
||||
// duration such as 24h; DefaultWorkerStaleAfter when empty.
|
||||
WorkerStaleAfter string `json:"worker-stale-after,omitempty"`
|
||||
// DecisionFloor and ResearchFloor are the highest decision and research numbers taken where the tracker
|
||||
// cannot read, such as an open branch of hq on the forge (issue 424). Read once, when the sequence's
|
||||
// counter starts: its first number is one above the highest of this floor, the numbers hq's record holds
|
||||
// and the reservations kept. Raised by the operator before the first reserve when a branch holds more.
|
||||
DecisionFloor uint64 `json:"decision-floor,omitempty"`
|
||||
ResearchFloor uint64 `json:"research-floor,omitempty"`
|
||||
}
|
||||
|
||||
func readSettings(path string) (Settings, error) {
|
||||
@@ -623,5 +629,32 @@ func Tools(t *Tracker) []stdio.Tool {
|
||||
}
|
||||
return t.Tracking(keys)
|
||||
}},
|
||||
{Name: Seat + ".reserve",
|
||||
Description: "Reserve the next decision or research number, before the record is written, and get it: never " +
|
||||
"one another caller got, and always above every number hq already holds. The answer is the kind and the " +
|
||||
"number, and the number as a record's name starts with it (0292 for a decision, 046 for research). " +
|
||||
"Optionally for: what the record is about, in one line. A reserved number is spent forever, whether or " +
|
||||
"not its record lands, and is never handed out again; who reserved it, when and for what are kept. " +
|
||||
"Refused, taking no number: a kind that is not decision or research (an issue's number comes from " +
|
||||
"open), a for that is not one line of plain text or carries a secret, an unknown argument, and the " +
|
||||
"first number of a kind while what hq holds cannot be read.",
|
||||
Input: schema([]string{"kind"}, map[string]any{
|
||||
"kind": map[string]any{"type": "string", "enum": ReserveKinds, "description": "which number: decision or research"},
|
||||
"for": text("what the record is about, in one line, such as: the knowledge base holds decisions"),
|
||||
}),
|
||||
Run: func(m map[string]any) (any, error) {
|
||||
a, err := readArgs("reserve", m, "kind", "for")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var ask ReserveAsk
|
||||
if ask.Kind, err = a.str("kind"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if ask.For, err = a.str("for"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return t.Reserve(ask)
|
||||
}},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -84,20 +84,23 @@ func TestItDeclaresAndHoldsTheIssueTrackerSeatOnceForTheMesh(t *testing.T) {
|
||||
t.Fatalf("module %q, seats %+v, claims %+v", m.Module, m.Seats, m.Claims)
|
||||
}
|
||||
if !reflect.DeepEqual(m.Emits, []string{EventOpened, EventMoved, EventNoted, EventLinked, EventImported, EventSet,
|
||||
EventTriaged, EventReady, EventReadyToVerify}) {
|
||||
EventTriaged, EventReady, EventReadyToVerify, EventReserved}) {
|
||||
t.Fatalf("emits %v", m.Emits)
|
||||
}
|
||||
// Three states, none expiring: issues keep a short history per key, the counter one value, the backlog's
|
||||
// pass of the duplicate check one value (ADR 0278 §6).
|
||||
// Four states, none expiring: issues keep a short history per key, the counters one value each, the
|
||||
// backlog's pass of the duplicate check one value (ADR 0278 §6), each reserved number one value (issue 424).
|
||||
d := m.stateDeclarations(t)
|
||||
if len(d) != 3 || d[0].Name != IssuesState || d[0].History < 2 || d[0].TTLSeconds != 0 ||
|
||||
if len(d) != 4 || d[0].Name != IssuesState || d[0].History < 2 || d[0].TTLSeconds != 0 ||
|
||||
d[1].Name != CountersState || d[1].History > 1 || d[1].TTLSeconds != 0 ||
|
||||
d[2].Name != GroomingState || d[2].History > 1 || d[2].TTLSeconds != 0 {
|
||||
d[2].Name != GroomingState || d[2].History > 1 || d[2].TTLSeconds != 0 ||
|
||||
d[3].Name != ReservationsState || d[3].History > 1 || d[3].TTLSeconds != 0 {
|
||||
t.Fatalf("state %+v", d)
|
||||
}
|
||||
// The duplicate check hears the tracker's own opens and asks the agent module on its node (ADR 0278 §1, §5).
|
||||
// The duplicate check hears the tracker's own opens and asks the agent module on its node (ADR 0278 §1, §5);
|
||||
// a reservation reads what numbers the record holds through the records module (issue 424).
|
||||
if !reflect.DeepEqual(m.Consumes, []string{"mesh-issues." + EventOpened, "mesh-issues." + EventMoved, "mesh-delivery.transition"}) ||
|
||||
!reflect.DeepEqual(m.Invokes, []string{"claude-code.claude_code_judge", "seat:mesh-delivery.checks", "seat:operator-channel.notify"}) {
|
||||
!reflect.DeepEqual(m.Invokes, []string{"claude-code.claude_code_judge", "seat:mesh-delivery.checks", "seat:operator-channel.notify",
|
||||
"records.records_list"}) {
|
||||
t.Fatalf("consumes %v, invokes %v", m.Consumes, m.Invokes)
|
||||
}
|
||||
if len(m.Build.Artifacts) != 1 || m.Build.Artifacts[0].From != "cmd/mesh-issues" ||
|
||||
|
||||
@@ -25,7 +25,7 @@ type memory struct {
|
||||
}
|
||||
|
||||
func newMemory() *memory {
|
||||
return &memory{data: map[string]map[string]Entry{IssuesState: {}, CountersState: {}, GroomingState: {}}, fail: map[string]error{}}
|
||||
return &memory{data: map[string]map[string]Entry{IssuesState: {}, CountersState: {}, GroomingState: {}, ReservationsState: {}}, fail: map[string]error{}}
|
||||
}
|
||||
|
||||
func (m *memory) bucket(state string) (map[string]Entry, error) {
|
||||
|
||||
@@ -0,0 +1,300 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// The tracker hands out decision and research numbers too (novox/hq ADR 0275 rule 8, ADR 0290 rule 5, issue
|
||||
// 424): one counter per sequence in the state `counters`, beside `issue`, advanced the same way, so two
|
||||
// sessions writing a record at the same moment never take one number. A reserved number is spent forever,
|
||||
// whether or not its record ever lands; each reservation is kept in the state `reservations` and said as the
|
||||
// event `reserved`.
|
||||
|
||||
const (
|
||||
// ReservationsState keeps one key per reserved number, named by ReservationKey.
|
||||
ReservationsState = "reservations"
|
||||
// EventReserved is said for every number reserved.
|
||||
EventReserved = "reserved"
|
||||
|
||||
KindDecision = "decision"
|
||||
KindResearch = "research"
|
||||
|
||||
// DecisionsFolder and ResearchFolder are where hq keeps the records whose numbers a counter starts above.
|
||||
DecisionsFolder = "02-DECISIONS"
|
||||
ResearchFolder = "01-RESEARCH"
|
||||
|
||||
// raiseAttempts is how many times a counter behind the record is read and raised again.
|
||||
raiseAttempts = 16
|
||||
)
|
||||
|
||||
// ReserveKinds are the sequences reserve hands numbers from; each is also its counter's key in `counters`.
|
||||
var ReserveKinds = []string{KindDecision, KindResearch}
|
||||
|
||||
// sequence is one kind of record numbered by the tracker: where the record keeps it, how many digits its
|
||||
// number is written with, and the operator's floor for it.
|
||||
type sequence struct {
|
||||
kind string
|
||||
folder string
|
||||
digits int
|
||||
floor func(Settings) uint64
|
||||
}
|
||||
|
||||
var sequences = map[string]sequence{
|
||||
KindDecision: {KindDecision, DecisionsFolder, 4, func(s Settings) uint64 { return s.DecisionFloor }},
|
||||
KindResearch: {KindResearch, ResearchFolder, 3, func(s Settings) uint64 { return s.ResearchFloor }},
|
||||
}
|
||||
|
||||
// recordNumber is a record's name that starts with its number: 0291-some-words.md, 045-some-words.
|
||||
var recordNumber = regexp.MustCompile(`^([0-9]{1,9})-`)
|
||||
|
||||
// ReservationKey is a reservation's key: the kind and the number as the record writes it, such as
|
||||
// decision-0292 or research-046.
|
||||
func ReservationKey(kind string, n uint64) string {
|
||||
return kind + "-" + prefixOf(kind, n)
|
||||
}
|
||||
|
||||
func prefixOf(kind string, n uint64) string {
|
||||
return fmt.Sprintf("%0*d", sequences[kind].digits, n)
|
||||
}
|
||||
|
||||
// ReserveAsk is what a reserve carries.
|
||||
type ReserveAsk struct {
|
||||
Kind string
|
||||
For string
|
||||
}
|
||||
|
||||
// Reservation is one reserved number as it is kept.
|
||||
type Reservation struct {
|
||||
Kind string `json:"kind"`
|
||||
Number uint64 `json:"number"`
|
||||
For string `json:"for,omitempty"`
|
||||
By string `json:"by"`
|
||||
At time.Time `json:"at"`
|
||||
// ID is the call's own id, so a write whose answer was lost is known as its own.
|
||||
ID string `json:"id"`
|
||||
}
|
||||
|
||||
// Reserved is the answer to a reserve: kind and number are the contract; the rest says what was kept.
|
||||
type Reserved struct {
|
||||
Kind string `json:"kind"`
|
||||
Number uint64 `json:"number"`
|
||||
// Prefix is the number as the record's name starts with it, such as 0292.
|
||||
Prefix string `json:"prefix"`
|
||||
For string `json:"for,omitempty"`
|
||||
By string `json:"by"`
|
||||
At time.Time `json:"at"`
|
||||
Said string `json:"said"`
|
||||
Announced bool `json:"announced"`
|
||||
Warning string `json:"warning,omitempty"`
|
||||
}
|
||||
|
||||
// Reserve hands out the next number of a sequence and keeps who reserved it, when and for what. A refused
|
||||
// reserve spends no number; a number handed out is spent forever.
|
||||
func (t *Tracker) Reserve(a ReserveAsk) (*Reserved, error) {
|
||||
kind := a.Kind
|
||||
seq, ok := sequences[kind]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("kind %q is not a kind of number the tracker hands out: one of %s (an issue's number "+
|
||||
"comes from open). No number was taken", a.Kind, strings.Join(ReserveKinds, ", "))
|
||||
}
|
||||
purpose := ""
|
||||
if strings.TrimSpace(a.For) != "" {
|
||||
var err error
|
||||
if purpose, err = singleLine("for", a.For, MostTitle); err != nil {
|
||||
return nil, fmt.Errorf("%v. No number was taken", err)
|
||||
}
|
||||
}
|
||||
records := t.Records
|
||||
if records == nil {
|
||||
records = busRecords{}
|
||||
}
|
||||
highest, rerr := highestIn(records, seq)
|
||||
n, err := t.nextOf(seq, highest, rerr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r := Reservation{Kind: kind, Number: n, For: purpose, By: t.by(), At: t.now(), ID: newID()}
|
||||
if err := t.keep(&r); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := &Reserved{Kind: kind, Number: n, Prefix: prefixOf(kind, n), For: purpose, By: r.By, At: r.At,
|
||||
Said: fmt.Sprintf("Reserved %s number %s; it is spent, whether or not its record lands.", kind, prefixOf(kind, n))}
|
||||
body := map[string]any{"kind": kind, "number": n, "prefix": out.Prefix, "by": r.By, "at": r.At}
|
||||
if purpose != "" {
|
||||
body["for"] = purpose
|
||||
}
|
||||
if err := t.Emit(EventReserved, body); err != nil {
|
||||
t.Logf("[mesh-issues] %s %s was reserved and its reserved event could not be said: %v", kind, out.Prefix, err)
|
||||
out.Warning = fmt.Sprintf("the reservation is kept, and its reserved event could not be said on the bus (%v): "+
|
||||
"what watches the events did not hear of it", err)
|
||||
} else {
|
||||
out.Announced = true
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// highestIn is the highest number the record holds in a sequence's folder, sub-folders and documents alike.
|
||||
func highestIn(records Records, seq sequence) (uint64, error) {
|
||||
folders, docs, err := records.List(seq.folder)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var highest uint64
|
||||
for _, name := range append(folders, docs...) {
|
||||
if m := recordNumber.FindStringSubmatch(name); m != nil {
|
||||
if n, err := strconv.ParseUint(m[1], 10, 64); err == nil && n > highest {
|
||||
highest = n
|
||||
}
|
||||
}
|
||||
}
|
||||
return highest, nil
|
||||
}
|
||||
|
||||
// nextOf takes a sequence's next number. A counter not yet started is started once, above the highest the
|
||||
// record holds, the operator's floor and every reservation kept; it is never started without the record
|
||||
// read. A started counter behind the record (a number taken by hand) is raised to it first; a record that
|
||||
// cannot be read then leaves the counter as it is.
|
||||
func (t *Tracker) nextOf(seq sequence, highest uint64, recordErr error) (uint64, error) {
|
||||
e, err := t.Store.Get(CountersState, seq.kind)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("the %s counter cannot be read (%v); no number was taken", seq.kind, err)
|
||||
}
|
||||
if e == nil {
|
||||
if recordErr != nil {
|
||||
return 0, fmt.Errorf("the %s counter has not started, and what numbers hq's %s holds cannot be read through "+
|
||||
"the records module (%v): it starts above them, so no number was taken. Ask again when the records "+
|
||||
"module answers", seq.kind, seq.folder, recordErr)
|
||||
}
|
||||
if err := t.startSequence(seq, highest); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
} else if recordErr != nil {
|
||||
t.Logf("[mesh-issues] the record's %s could not be read (%v); the %s counter is used as it stands", seq.folder, recordErr, seq.kind)
|
||||
} else if err := t.raise(seq.kind, highest); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
for round := 0; round < counterRounds; round++ {
|
||||
n, err := t.Store.Increment(CountersState, seq.kind, counterAttempts)
|
||||
if err == nil {
|
||||
return n, nil
|
||||
}
|
||||
if !errors.Is(err, stdio.ErrChanged) {
|
||||
t.Logf("[mesh-issues] the %s counter could not be advanced: %v", seq.kind, err)
|
||||
return 0, fmt.Errorf("the %s counter could not be advanced (%v); nothing was reserved. If the bus took the "+
|
||||
"write before it failed, one number stays unused", seq.kind, err)
|
||||
}
|
||||
t.Sleep(jitter(time.Duration(round+1) * 10 * time.Millisecond))
|
||||
}
|
||||
return 0, fmt.Errorf("so many %s numbers are being reserved at this moment that the counter was taken by another "+
|
||||
"caller %d times; no number was taken. Reserve again", seq.kind, counterRounds*counterAttempts)
|
||||
}
|
||||
|
||||
// startSequence makes a sequence's counter, only when there is none, at the highest of the record, the
|
||||
// floor and the reservations kept: the first number handed out is one above all three.
|
||||
func (t *Tracker) startSequence(seq sequence, highest uint64) error {
|
||||
s, err := t.Settings()
|
||||
if err != nil {
|
||||
return fmt.Errorf("the %s counter has not started and the settings cannot be read (%v); no number was taken", seq.kind, err)
|
||||
}
|
||||
start := max(highest, seq.floor(s))
|
||||
keys, err := t.Store.Keys(ReservationsState)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the %s counter has not started and the reservations kept cannot be listed (%v); no number "+
|
||||
"was taken", seq.kind, err)
|
||||
}
|
||||
for _, k := range keys {
|
||||
if rest, ok := strings.CutPrefix(k, seq.kind+"-"); ok {
|
||||
if n, err := strconv.ParseUint(rest, 10, 64); err == nil && n > start {
|
||||
start = n
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := t.Store.Create(CountersState, seq.kind, start); err != nil && !errors.Is(err, stdio.ErrChanged) {
|
||||
return fmt.Errorf("the %s counter could not be started (%v); no number was taken", seq.kind, err)
|
||||
}
|
||||
// ErrChanged: another caller started it first; the raise on the next reserve covers any difference.
|
||||
return nil
|
||||
}
|
||||
|
||||
// raise sets a counter to the highest number the record holds when it stands below it, written only at the
|
||||
// revision read: it only ever goes up.
|
||||
func (t *Tracker) raise(key string, highest uint64) error {
|
||||
for i := 0; i < raiseAttempts; i++ {
|
||||
e, err := t.Store.Get(CountersState, key)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the %s counter cannot be read (%v); no number was taken", key, err)
|
||||
}
|
||||
var n uint64
|
||||
if e != nil {
|
||||
if err := json.Unmarshal(e.Value, &n); err != nil {
|
||||
return fmt.Errorf("the %s counter holds %s, not a number; no number was taken", key, e.Value)
|
||||
}
|
||||
}
|
||||
if e == nil || n >= highest {
|
||||
return nil
|
||||
}
|
||||
_, err = t.Store.Update(CountersState, key, highest, e.Revision)
|
||||
if err == nil {
|
||||
t.Logf("[mesh-issues] the %s counter stood at %d below %d in the record; raised to it", key, n, highest)
|
||||
return nil
|
||||
}
|
||||
if !errors.Is(err, stdio.ErrChanged) {
|
||||
return fmt.Errorf("the %s counter stands below the record's %d and could not be raised (%v); no number was "+
|
||||
"taken", key, highest, err)
|
||||
}
|
||||
t.Sleep(jitter(time.Duration(i+1) * 5 * time.Millisecond))
|
||||
}
|
||||
return fmt.Errorf("the %s counter was changed by other callers %d times while it was raised to the record's %d; "+
|
||||
"no number was taken. Reserve again", key, raiseAttempts, highest)
|
||||
}
|
||||
|
||||
// keep writes a reservation under its key, never over one. A write that cannot be confirmed is refused, and
|
||||
// the number stays spent: it is never handed out again.
|
||||
func (t *Tracker) keep(r *Reservation) error {
|
||||
key := ReservationKey(r.Kind, r.Number)
|
||||
var err error
|
||||
for i := 0; i < createAttempts; i++ {
|
||||
if i > 0 {
|
||||
t.Sleep(jitter(time.Duration(i) * 50 * time.Millisecond))
|
||||
}
|
||||
_, err = t.Store.Create(ReservationsState, key, r)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
if errors.Is(err, stdio.ErrChanged) {
|
||||
if mine, rerr := t.isKept(key, r.ID); rerr == nil && mine {
|
||||
return nil
|
||||
}
|
||||
t.Logf("[mesh-issues] the %s counter handed out %d and a reservation %s is already kept: not overwritten", r.Kind, r.Number, key)
|
||||
return fmt.Errorf("a fault, not your mistake: the %s counter handed out %d, and %s is already reserved. "+
|
||||
"Nothing was overwritten; %d stays spent. Reserve again for the next number, and tell the operator "+
|
||||
"the counter is behind the reservations kept", r.Kind, r.Number, key, r.Number)
|
||||
}
|
||||
}
|
||||
if mine, rerr := t.isKept(key, r.ID); rerr == nil && mine {
|
||||
return nil
|
||||
}
|
||||
t.Logf("[mesh-issues] %s was taken and its reservation's write was not confirmed: %v", key, err)
|
||||
return fmt.Errorf("%s number %d was taken and its reservation could not be kept (%v): the number is spent and "+
|
||||
"is not yours to use. Reserve again for the next one", r.Kind, r.Number, err)
|
||||
}
|
||||
|
||||
func (t *Tracker) isKept(key, id string) (bool, error) {
|
||||
e, err := t.Store.Get(ReservationsState, key)
|
||||
if err != nil || e == nil {
|
||||
return false, err
|
||||
}
|
||||
var there Reservation
|
||||
if err := json.Unmarshal(e.Value, &there); err != nil {
|
||||
return false, err
|
||||
}
|
||||
return there.ID == id, nil
|
||||
}
|
||||
@@ -0,0 +1,342 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The tracker hands out decision and research numbers too (novox/hq ADR 0275 rule 8, ADR 0290 rule 5, issue
|
||||
// 424): one counter per sequence, advanced like the issue counter, started above every number the record
|
||||
// holds and the operator's floor, and a reserved number is spent forever.
|
||||
|
||||
// listedRecords is the record as a folder listing: what records_list answers per folder, or an error.
|
||||
type listedRecords struct {
|
||||
mu sync.Mutex
|
||||
folders map[string][]string
|
||||
docs map[string][]string
|
||||
err error
|
||||
lists int
|
||||
}
|
||||
|
||||
func (l *listedRecords) Standing() (string, error) { return "1ea678c", nil }
|
||||
func (l *listedRecords) Read(string) (string, string, error) { return "", "", errors.New("not read") }
|
||||
func (l *listedRecords) List(folder string) ([]string, []string, error) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
l.lists++
|
||||
if l.err != nil {
|
||||
return nil, nil, l.err
|
||||
}
|
||||
return l.folders[folder], l.docs[folder], nil
|
||||
}
|
||||
|
||||
// hqToday is hq as the records module holds it: decisions as documents, research as folders.
|
||||
func hqToday() *listedRecords {
|
||||
return &listedRecords{
|
||||
docs: map[string][]string{DecisionsFolder: {"README.md", "0001-the-first.md", "0290-no-copy-in-hq.md",
|
||||
"0291-an-issue-lives-whole-in-the-issue-tracker.md"}},
|
||||
folders: map[string][]string{ResearchFolder: {"001-the-first", "044-what-leaves-the-controller-next",
|
||||
"045-a-knowledge-base-the-mesh-writes", "_template"}},
|
||||
}
|
||||
}
|
||||
|
||||
func reserveHarness(t *testing.T) (*harness, *listedRecords) {
|
||||
t.Helper()
|
||||
h := newHarness(t)
|
||||
r := hqToday()
|
||||
h.Records = r
|
||||
return h, r
|
||||
}
|
||||
|
||||
func (h *harness) reserve(t *testing.T, kind string) uint64 {
|
||||
t.Helper()
|
||||
out, err := h.Reserve(ReserveAsk{Kind: kind})
|
||||
if err != nil {
|
||||
t.Fatalf("reserve %s: %v", kind, err)
|
||||
}
|
||||
return out.Number
|
||||
}
|
||||
|
||||
func (m *memory) counterOf(t *testing.T, key string) uint64 {
|
||||
t.Helper()
|
||||
e, _ := m.Get(CountersState, key)
|
||||
if e == nil {
|
||||
return 0
|
||||
}
|
||||
var n uint64
|
||||
if err := json.Unmarshal(e.Value, &n); err != nil {
|
||||
t.Fatalf("the counter %s holds %s", key, e.Value)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
func TestTheFirstReservedNumberIsAboveTheHighestTheRecordHolds(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
if got := h.reserve(t, KindDecision); got != 292 {
|
||||
t.Fatalf("the first decision number is %d, want 292: one above 0291", got)
|
||||
}
|
||||
if got := h.reserve(t, KindResearch); got != 46 {
|
||||
t.Fatalf("the first research number is %d, want 46: one above 045", got)
|
||||
}
|
||||
if got := h.reserve(t, KindDecision); got != 293 {
|
||||
t.Fatalf("the second decision number is %d, want 293", got)
|
||||
}
|
||||
if h.mem.counter(t) != 0 {
|
||||
t.Fatal("a reservation moved the issue counter")
|
||||
}
|
||||
}
|
||||
|
||||
// An open branch the tracker cannot read holds a higher number: the operator's floor says it.
|
||||
func TestTheCounterStartsAboveTheFloorWhenItIsHigher(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
h.settings.DecisionFloor = 293
|
||||
h.settings.ResearchFloor = 40 // below what the record holds: the record wins
|
||||
if got := h.reserve(t, KindDecision); got != 294 {
|
||||
t.Fatalf("got %d, want 294: one above the floor 293", got)
|
||||
}
|
||||
if got := h.reserve(t, KindResearch); got != 46 {
|
||||
t.Fatalf("got %d, want 46: the floor 40 is below 045", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheCounterNeverStartsBelowAReservationAlreadyKept(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
h.mem.put(ReservationsState, ReservationKey(KindDecision, 310), Reservation{Kind: KindDecision, Number: 310})
|
||||
if got := h.reserve(t, KindDecision); got != 311 {
|
||||
t.Fatalf("got %d, want 311: the counter started below a reservation kept", got)
|
||||
}
|
||||
}
|
||||
|
||||
// A number the record holds above the counter — taken by hand while sessions still did — raises it: no
|
||||
// number at or below one that exists is handed out.
|
||||
func TestARecordAboveTheCounterRaisesIt(t *testing.T) {
|
||||
h, r := reserveHarness(t)
|
||||
h.reserve(t, KindDecision) // 292, the counter started
|
||||
r.mu.Lock()
|
||||
r.docs[DecisionsFolder] = append(r.docs[DecisionsFolder], "0300-taken-by-hand.md")
|
||||
r.mu.Unlock()
|
||||
if got := h.reserve(t, KindDecision); got != 301 {
|
||||
t.Fatalf("got %d, want 301: one above 0300", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheCounterStartsOnlyOnceAndTheFloorIsThenNotRead(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
h.reserve(t, KindDecision) // 292
|
||||
h.settings.DecisionFloor = 1
|
||||
if got := h.reserve(t, KindDecision); got != 293 {
|
||||
t.Fatalf("got %d, want 293", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManyConcurrentReservationsGetDistinctConsecutiveNumbers(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
const each = 40
|
||||
var mu sync.Mutex
|
||||
got := map[string][]uint64{}
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 2*each; i++ {
|
||||
kind := KindDecision
|
||||
if i%2 == 1 {
|
||||
kind = KindResearch
|
||||
}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
out, err := h.Reserve(ReserveAsk{Kind: kind, For: "concurrent"})
|
||||
if err != nil {
|
||||
t.Errorf("reserve %s: %v", kind, err)
|
||||
return
|
||||
}
|
||||
mu.Lock()
|
||||
got[kind] = append(got[kind], out.Number)
|
||||
mu.Unlock()
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
for kind, first := range map[string]uint64{KindDecision: 292, KindResearch: 46} {
|
||||
ns := got[kind]
|
||||
sort.Slice(ns, func(i, j int) bool { return ns[i] < ns[j] })
|
||||
if len(ns) != each {
|
||||
t.Fatalf("%s: %d numbers, want %d", kind, len(ns), each)
|
||||
}
|
||||
for i, n := range ns {
|
||||
if n != first+uint64(i) {
|
||||
t.Fatalf("%s: numbers %v are not %d.. distinct and consecutive", kind, ns, first)
|
||||
}
|
||||
if _, err := h.mem.Get(ReservationsState, ReservationKey(kind, n)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if c := h.mem.counterOf(t, kind); c != first+each-1 {
|
||||
t.Fatalf("%s counter at %d, want %d", kind, c, first+each-1)
|
||||
}
|
||||
}
|
||||
if keys, _ := h.mem.Keys(ReservationsState); len(keys) != 2*each {
|
||||
t.Fatalf("%d reservations kept, want %d", len(keys), 2*each)
|
||||
}
|
||||
}
|
||||
|
||||
func TestARefusedReservationSpendsNothing(t *testing.T) {
|
||||
h, r := reserveHarness(t)
|
||||
refused := []ReserveAsk{
|
||||
{Kind: ""},
|
||||
{Kind: "issue"},
|
||||
{Kind: "Decision"},
|
||||
{Kind: "design"},
|
||||
{Kind: KindDecision, For: "two\nlines"},
|
||||
{Kind: KindDecision, For: strings.Repeat("x", MostTitle+1)},
|
||||
{Kind: KindResearch, For: "the key is sk-ant-api03-abcdefghijklmnopqrstuvwxyz"},
|
||||
{Kind: KindResearch, For: "t\x1b[31m"},
|
||||
}
|
||||
for i, a := range refused {
|
||||
if out, err := h.Reserve(a); err == nil {
|
||||
t.Errorf("case %d (%+v) was reserved: %+v", i, a, out)
|
||||
}
|
||||
}
|
||||
keys, _ := h.mem.Keys(CountersState)
|
||||
kept, _ := h.mem.Keys(ReservationsState)
|
||||
if len(keys) != 0 || len(kept) != 0 || len(h.said()) != 0 || r.lists != 0 {
|
||||
t.Fatalf("a refused reservation left counters %v, reservations %v, events %v, record reads %d", keys, kept, h.said(), r.lists)
|
||||
}
|
||||
if got := h.reserve(t, KindDecision); got != 292 {
|
||||
t.Fatalf("the first valid reservation got %d, want 292", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Before its counter starts, a sequence must know what the record holds: unreadable, nothing is handed out.
|
||||
// Once started, the counter alone is enough.
|
||||
func TestAnUnreadableRecordRefusesTheFirstNumberAndNotTheNext(t *testing.T) {
|
||||
h, r := reserveHarness(t)
|
||||
r.err = errors.New("records did not answer")
|
||||
if _, err := h.Reserve(ReserveAsk{Kind: KindDecision}); err == nil || !strings.Contains(err.Error(), "records did not answer") {
|
||||
t.Fatalf("a first reservation with the record unreadable: %v", err)
|
||||
}
|
||||
if h.mem.counterOf(t, KindDecision) != 0 || len(h.said()) != 0 {
|
||||
t.Fatal("a refused first reservation started the counter or said something")
|
||||
}
|
||||
r.err = nil
|
||||
h.reserve(t, KindDecision) // 292
|
||||
r.err = errors.New("records did not answer")
|
||||
if got := h.reserve(t, KindDecision); got != 293 {
|
||||
t.Fatalf("got %d, want 293 from the started counter", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAReservationIsKeptWithWhoWhenAndWhatFor(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
out, err := h.Reserve(ReserveAsk{Kind: KindDecision, For: " the knowledge base holds decisions "})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if out.Kind != KindDecision || out.Number != 292 || out.Prefix != "0292" || out.By != Unproven ||
|
||||
out.For != "the knowledge base holds decisions" || !out.Announced {
|
||||
t.Fatalf("answered %+v", out)
|
||||
}
|
||||
e, err := h.mem.Get(ReservationsState, "decision-0292")
|
||||
if err != nil || e == nil {
|
||||
t.Fatalf("the reservation is not kept: %v", err)
|
||||
}
|
||||
var kept Reservation
|
||||
if err := json.Unmarshal(e.Value, &kept); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if kept.Kind != KindDecision || kept.Number != 292 || kept.By != Unproven || kept.At.IsZero() ||
|
||||
kept.For != "the knowledge base holds decisions" {
|
||||
t.Fatalf("kept %+v", kept)
|
||||
}
|
||||
ev := h.said()
|
||||
if len(ev) != 1 || ev[0].Event != EventReserved || ev[0].Body["kind"] != KindDecision ||
|
||||
ev[0].Body["number"] != float64(292) || ev[0].Body["by"] != Unproven || ev[0].Body["for"] != kept.For {
|
||||
t.Fatalf("said %+v", ev)
|
||||
}
|
||||
r, err := h.Reserve(ReserveAsk{Kind: KindResearch})
|
||||
if err != nil || r.Prefix != "046" {
|
||||
t.Fatalf("a research number's prefix: %+v, %v", r, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAReservationWhoseEventCannotBeSaidIsKeptAndSaysSo(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
h.emitErr = errors.New("the bus is away")
|
||||
out, err := h.Reserve(ReserveAsk{Kind: KindResearch})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if out.Announced || out.Warning == "" || out.Number != 46 {
|
||||
t.Fatalf("answered %+v", out)
|
||||
}
|
||||
}
|
||||
|
||||
// A reservation whose write cannot be confirmed is refused, and the number stays spent.
|
||||
func TestANumberWhoseReservationIsNotKeptIsSaidAndNeverHandedOutAgain(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
h.mem.fail[ReservationsState] = errors.New("the bus did not answer")
|
||||
if got := h.reserve(t, KindDecision); got != 292 {
|
||||
t.Fatalf("one failure is tried again: got %d", got)
|
||||
}
|
||||
failing := &failingReservations{memory: h.mem, fail: 1000}
|
||||
h.Store = failing
|
||||
_, err := h.Reserve(ReserveAsk{Kind: KindDecision})
|
||||
if err == nil || !strings.Contains(err.Error(), "293") || !strings.Contains(err.Error(), "spent") {
|
||||
t.Fatalf("a reservation never kept: %v", err)
|
||||
}
|
||||
h.Store = h.mem
|
||||
if got := h.reserve(t, KindDecision); got != 294 {
|
||||
t.Fatalf("got %d, want 294: a number is never handed out twice", got)
|
||||
}
|
||||
}
|
||||
|
||||
type failingReservations struct {
|
||||
*memory
|
||||
fail int
|
||||
}
|
||||
|
||||
func (f *failingReservations) Create(state, key string, value any) (uint64, error) {
|
||||
if state == ReservationsState && f.fail > 0 {
|
||||
f.fail--
|
||||
return 0, errors.New("the bus did not answer")
|
||||
}
|
||||
return f.memory.Create(state, key, value)
|
||||
}
|
||||
|
||||
func TestTheReserveVerbReadsItsArgumentsStrictlyAndAnswersTheContract(t *testing.T) {
|
||||
h, _ := reserveHarness(t)
|
||||
for _, args := range []map[string]any{
|
||||
{},
|
||||
{"kind": 5},
|
||||
{"kind": KindDecision, "for": 7},
|
||||
{"kind": KindDecision, "by": "operator"},
|
||||
{"kind": KindDecision, "number": 400},
|
||||
} {
|
||||
if out, err := h.call(t, "reserve", args); err == nil {
|
||||
t.Errorf("reserve %v was taken: %+v", args, out)
|
||||
}
|
||||
}
|
||||
out, err := h.call(t, "reserve", map[string]any{"kind": "research", "for": "a spike"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
raw, _ := json.Marshal(out)
|
||||
var answer map[string]any
|
||||
_ = json.Unmarshal(raw, &answer)
|
||||
if answer["kind"] != "research" || answer["number"] != float64(46) || strings.Contains(string(raw), "operator") {
|
||||
t.Fatalf("answered %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAReservationKeyNamesTheKindAndTheRecordsPrefix(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
kind string
|
||||
n uint64
|
||||
want string
|
||||
}{{KindDecision, 292, "decision-0292"}, {KindResearch, 46, "research-046"}, {KindDecision, 10000, "decision-10000"}} {
|
||||
if got := ReservationKey(c.kind, c.n); got != c.want {
|
||||
t.Errorf("%s %d: %s, want %s", c.kind, c.n, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user