From 68117bd74db6b98da051b241fde3dac3018fb08b Mon Sep 17 00:00:00 2001 From: mesh-tools Date: Tue, 20 Oct 2026 01:20:09 +0000 Subject: [PATCH] Hold the seat only when no assignment stands, every have key says removed and both watches are current; fix the two bed tests that failed the try (review round 1, G1) --- .../cmd/mesh-delivery-gate/bus_test.go | 25 +++- .../cmd/mesh-delivery-gate/main.go | 9 +- .../cmd/mesh-delivery-gate/occupancy_test.go | 80 +++++++++++++ .../cmd/mesh-delivery-gate/seat.go | 108 +++++++++++++++--- 4 files changed, 201 insertions(+), 21 deletions(-) create mode 100644 modules/mesh-delivery-gate/cmd/mesh-delivery-gate/occupancy_test.go diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bus_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bus_test.go index 33edebaf..ae311e49 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bus_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bus_test.go @@ -12,6 +12,7 @@ import ( "os" "os/exec" "path/filepath" + "slices" "strings" "sync" "testing" @@ -93,6 +94,10 @@ func newTestBus(t *testing.T) *testBus { } b.kvs[name] = kv } + // The node-engine has said today's module removed on the one machine: the gate may hold the seat (ADR 0450 §6, issue 1369). + if _, err := b.kvs["node-engine.have"].Put(context.Background(), "test.mesh-delivery", []byte(`{"v":1,"phase":"removed","health":"unknown"}`)); err != nil { + t.Fatal(err) + } // The streams the events are on (mesh.mod..event., mesh.seat..event.). if _, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.gitea.event.>", "mesh.seat." + ControllerSeat + ".event.>", "mesh.seat." + asks.Seat + ".event.>", @@ -528,7 +533,8 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB b.publish(t, "mesh.mod.gitea.event.forge.pull-request.edited", map[string]any{"owner": "novox", "repo": "app", "number": 7, "title": announced.Title, "head_sha": head, "base": "main", "body": "Reviewed-by: lead@shanks at " + head}) rt.until("the ask to be written again with the record", 5*time.Second, func() bool { - return strings.Contains(askAt(t, b, key).Ask.Explanation, "reviewed by lead on shanks") + a, ok := askNow(b, key) // the key may be absent for an instant while the ask is taken away and written again + return ok && strings.Contains(a.Ask.Explanation, "reviewed by lead on shanks") }) if h, err := b.kvs["asks"].History(ctx, key); err != nil || len(h) != 3 || h[1].Operation() != jetstream.KeyValueDelete { t.Fatalf("the ask's history is %d revisions (%v); wanted written, taken away, written", len(h), err) @@ -597,6 +603,16 @@ func onlyAsk(b *testBus) string { return keys[0] } +// askNow is the ask under a key at this moment; false in the instant a rewrite has taken the key away and not yet written it. +func askNow(b *testBus, key string) (sdk.Ask, bool) { + e, err := b.kvs["asks"].Get(context.Background(), key) + if err != nil { + return sdk.Ask{}, false + } + a, err := sdk.DecodeAsk(e.Value()) + return a, err == nil +} + func askAt(t *testing.T, b *testBus, key string) sdk.Ask { t.Helper() e, err := b.kvs["asks"].Get(context.Background(), key) @@ -875,8 +891,11 @@ func TestTheGateCarriesALiveCapturesDeliveriesAndGroupsOnceOverARealBus(t *testi if err != nil { t.Fatalf("the group %s did not arrive: %v", g.ID, err) } - if string(mustJSON(g)) != string(mustJSON(mustDecode[Group](t, e.Value()))) { - t.Fatalf("the group %s changed in the carry", g.ID) + // The carry copies the group byte for byte (TestTheCarryCopiesWhatItDoesNotClearByteForByte); by the time this reads, the gate + // holds the seat and acts, and writes a group's said state as it does, so what is compared is what only the carry could change. + got := mustDecode[Group](t, e.Value()) + if got.ID != g.ID || !slices.Equal(got.Members, g.Members) || !got.Created.Equal(g.Created) { + t.Fatalf("the group %s changed in the carry: %+v", g.ID, got) } } if _, err := b.kvs["deliveries"].Get(ctx, carriedKey); err != nil { diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go index e5f4348a..1a5ae466 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -40,6 +40,7 @@ var followed = []string{"gitea.pull.updated", "gitea.pull.merged", "gitea.pull.c const forgePattern = "mesh-delivery-forge-bridge.forge.**" func main() { + occupancy := &busOccupancy{} h := &Holder{ Store: kvStore{}, Controller: seatController{ask: stdio.Ask}, @@ -52,13 +53,13 @@ func main() { Owed: kvOwed{}, Pulls: kvPulls{}, // The gate acts only while it holds the seat, and carries today's state over when it first does (seat.go). - Seat: busOccupancy{}, + Seat: occupancy, Carry: carryFromBus, } h.OnHold = func() { go readOnHolding(h) } h.init() l := &listening{now: "not yet: starting"} - go run(h, l) + go run(h, l, occupancy) if err := stdio.Serve("", tools(h, l)); err != nil { logf("%v", err) os.Exit(1) @@ -70,10 +71,10 @@ func main() { // holds the seat (SeatChanged), and what it may have missed meanwhile is read once then (readOnHolding): the controller's // walks, the forge's open pull requests, the operator's asks — and it acts on events from then on (novox/hq ADR 0406 §1, // issue 975): no tick, no forge poll, no reconcile loop. Each step is retried, each failure said, never given up on quietly. -func run(h *Holder, l *listening) { +func run(h *Holder, l *listening, occupancy *busOccupancy) { time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it go h.Work() - go watchSeat(h, l) + go watchSeat(h, l, occupancy) for _, pattern := range followed { pattern := pattern for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/occupancy_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/occupancy_test.go new file mode 100644 index 00000000..d8fb63d2 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/occupancy_test.go @@ -0,0 +1,80 @@ +package main + +import ( + "strings" + "testing" +) + +// **Where today's module stands** (novox/hq ADR 0450 §6 as corrected, issues 1347 and 1369; review round 1, G1). Fixture +// sources: an assignment is mesh-controller internal/chain/records.go Assignment (a key is never removed: an unassignment writes +// `ended`, a pending one has a `state`); a have is mesh-sdk go/delivery Have, whose phase `removed` the node-engine writes for a +// module its declaration no longer holds. +const ( + standing = `{"node":"novox","module":"mesh-delivery","assigned":"2026-10-01T10:00:00Z","build":"b1","since":"2026-10-01T10:00:00Z"}` + ended = `{"node":"novox","module":"mesh-delivery","assigned":"2026-10-01T10:00:00Z","build":"b1","since":"2026-10-01T10:00:00Z","ended":"2026-10-20T03:00:00Z"}` + pending = `{"node":"novox","module":"mesh-delivery","state":"waiting"}` + haveRun = `{"v":1,"build":"b1","phase":"running","health":"ok"}` + haveGone = `{"v":1,"build":"b1","phase":"removed","health":"unknown"}` + haveFailed = `{"v":1,"build":"b1","phase":"failed","health":"failed"}` +) + +func bucket(current bool, kv ...string) seatBucket { + b := seatBucket{values: map[string][]byte{}, current: current} + for i := 0; i+1 < len(kv); i += 2 { + b.values[kv[i]] = []byte(kv[i+1]) + } + return b +} + +func TestAnEndedAssignmentKeepsItsKeyAndIsNoOccupant(t *testing.T) { + who, err := occupantsOf(bucket(true, "novox.mesh-delivery", ended), bucket(true, "novox.mesh-delivery", haveGone)) + if err != nil || len(who) != 0 { + t.Fatalf("an ended assignment and a removed have gave %v, %v; wanted a free seat", who, err) + } + if who, _ := occupantsOf(bucket(true, "novox.mesh-delivery", standing), bucket(true, "novox.mesh-delivery", haveGone)); len(who) != 1 || !strings.Contains(who[0], "assigned on novox") { + t.Fatalf("a standing assignment gave %v", who) + } + // A pending assignment, waiting for its build, is no assignment (every reader of the chain says so). + if who, _ := occupantsOf(bucket(true, "novox.mesh-delivery", pending), bucket(true, "novox.mesh-delivery", haveGone)); len(who) != 0 { + t.Fatalf("a pending assignment gave %v", who) + } + // Other modules' keys are no business of the gate's. + if who, _ := occupantsOf(bucket(true, "novox.mesh-delivery-gate", standing), bucket(true, "novox.mesh-delivery", haveGone)); len(who) != 0 { + t.Fatalf("another module's assignment gave %v", who) + } +} + +func TestAHaveKeyCountsUnlessItsPhaseIsRemoved(t *testing.T) { + for _, have := range []string{haveRun, haveFailed, `{"v":1,"phase":"applying","health":"unknown"}`, `not json`} { + who, err := occupantsOf(bucket(true), bucket(true, "novox.mesh-delivery", have)) + if err != nil || len(who) != 1 || !strings.Contains(who[0], "node-engine of novox") { + t.Fatalf("the have %s gave %v, %v; wanted an occupant", have, who, err) + } + } + // One machine removed it, another runs it still. + who, _ := occupantsOf(bucket(true), bucket(true, "novox.mesh-delivery", haveGone, "ace.mesh-delivery", haveRun)) + if len(who) != 1 || !strings.Contains(who[0], "ace") { + t.Fatalf("a module removed on one machine and running on another gave %v", who) + } +} + +// An empty read says nothing of the module: the seat is not free until a node-engine said the module removed. +func TestAnEmptyReadOfTheHaveKeysDoesNotFreeTheSeat(t *testing.T) { + who, err := occupantsOf(bucket(true), bucket(true)) + if err != nil || len(who) != 1 || !strings.Contains(who[0], "no have key") { + t.Fatalf("an empty read gave %v, %v; wanted the seat held back", who, err) + } + if who, _ := occupantsOf(bucket(true), bucket(true, "novox.node", `{"v":1}`, "novox.mesh-delivery-gate", haveRun)); len(who) != 1 { + t.Fatalf("have keys of other modules gave %v", who) + } +} + +// A watch that has not delivered its current values yet says nothing: the seat is not judged. +func TestTheSeatIsNotJudgedBeforeBothWatchesAreCurrent(t *testing.T) { + for _, c := range [][2]bool{{false, true}, {true, false}, {false, false}} { + who, err := occupantsOf(bucket(c[0]), bucket(c[1], "novox.mesh-delivery", haveGone)) + if err == nil || len(who) != 0 { + t.Fatalf("watches current %v gave %v, %v; wanted an error", c, who, err) + } + } +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go index 3160c98e..854ce277 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go @@ -2,9 +2,11 @@ package main import ( "encoding/json" + "errors" "fmt" "sort" "strings" + "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -193,27 +195,99 @@ func CarryOver(old, own func(bucket string) Bucket, at time.Time) (CarryReport, return report, nil } -// busOccupancy reads the controller's assignments and the node-engine's have keys, both `.`. -type busOccupancy struct{} +// What a bucket says of where today's module stands: the values by key, `.`, and whether the watch of the bucket +// has delivered its current values. +type seatBucket struct { + values map[string][]byte + current bool +} -func (busOccupancy) Occupants() ([]string, error) { +// occupantsOf says where today's module stands from what the controller's assignments and the node-engine's have keys hold +// (novox/hq ADR 0450 §6 as corrected). Neither bucket loses a key: an unassignment writes `ended` on the assignment and keeps +// its key (mesh-controller internal/chain/records.go Assignment), and the node-engine writes the phase `removed` on the have +// key of a module its declaration no longer holds (issue 1369). So an assignment counts only while it stands (no `ended`, no +// `state`), a have key counts unless its phase is `removed`, and the module is gone only once both watches have delivered +// their current values and at least one have key of it exists and every one says removed. A value that cannot be read counts. +func occupantsOf(assignments, haves seatBucket) ([]string, error) { + if !assignments.current || !haves.current { + return nil, errors.New("the watches of the assignments and the have keys have not both delivered their current values yet") + } var who []string - for _, src := range []struct{ bucket, says string }{ - {"mesh-controller.assignments", "assigned on %s"}, {"node-engine.have", "the node-engine of %s has it"}} { - keys, err := stdio.State(src.bucket).Keys() - if err != nil { - return nil, fmt.Errorf("%s cannot be read: %w", src.bucket, err) + for key, raw := range assignments.values { + node, module, ok := strings.Cut(key, ".") + if !ok || module != OldModule { + continue } - for _, key := range keys { - if node, module, ok := strings.Cut(key, "."); ok && module == OldModule { - who = append(who, fmt.Sprintf(src.says, node)) - } + var a struct { + Ended time.Time `json:"ended,omitzero"` + State string `json:"state,omitempty"` } + if err := json.Unmarshal(raw, &a); err != nil || (a.Ended.IsZero() && a.State == "") { + who = append(who, "assigned on "+node) + } + } + seen := 0 + for key, raw := range haves.values { + node, module, ok := strings.Cut(key, ".") + if !ok || module != OldModule { + continue + } + seen++ + var h struct { + Phase string `json:"phase"` + } + if err := json.Unmarshal(raw, &h); err != nil || h.Phase != "removed" { + who = append(who, "the node-engine of "+node+" has it") + } + } + if seen == 0 { + who = append(who, "no node-engine has said the module removed yet: no have key of it") } sort.Strings(who) return who, nil } +// busOccupancy reads the controller's assignments and the node-engine's have keys through the runtime. It answers nothing until +// the watches of both have delivered their current values (watchSeat says so). +type busOccupancy struct { + assignmentsCurrent, haveCurrent atomic.Bool +} + +func (b *busOccupancy) Occupants() ([]string, error) { + read := func(bucket string, current bool) (seatBucket, error) { + out := seatBucket{values: map[string][]byte{}, current: current} + if !current { + return out, nil + } + keys, err := stdio.State(bucket).Keys() + if err != nil { + return out, fmt.Errorf("%s cannot be read: %w", bucket, err) + } + for _, key := range keys { + if _, module, ok := strings.Cut(key, "."); !ok || module != OldModule { + continue + } + e, err := stdio.State(bucket).Get(key) + if err != nil { + return out, fmt.Errorf("%s %s cannot be read: %w", bucket, key, err) + } + if e != nil { + out.values[key] = e.Value + } + } + return out, nil + } + a, err := read("mesh-controller.assignments", b.assignmentsCurrent.Load()) + if err != nil { + return nil, err + } + h, err := read("node-engine.have", b.haveCurrent.Load()) + if err != nil { + return nil, err + } + return occupantsOf(a, h) +} + // keptBucket is a bucket of the bus through the runtime: today's module's by `.`, the gate's by its name. type keptBucket struct{ name string } @@ -237,7 +311,7 @@ func carryFromBus() (CarryReport, error) { // watchSeat reads who has the seat at once, and again at every change of the assignments or the have keys, for ever. A read // that fails is said and tried again in ten seconds; a change read meanwhile is not lost. -func watchSeat(h *Holder, l *listening) { +func watchSeat(h *Holder, l *listening, occ *busOccupancy) { changed := make(chan struct{}, 1) note := func(stdio.StateChange) error { select { @@ -246,11 +320,17 @@ func watchSeat(h *Holder, l *listening) { } return nil } - for _, bucket := range []string{"mesh-controller.assignments", "node-engine.have"} { + for _, w := range []struct { + bucket string + current *atomic.Bool + }{{"mesh-controller.assignments", &occ.assignmentsCurrent}, {"node-engine.have", &occ.haveCurrent}} { + bucket := w.bucket go func() { for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := stdio.State(bucket).Watch("", note) if err == nil { + w.current.Store(true) // the current values are in: the seat may be judged + note(stdio.StateChange{}) return } logf("[mesh-delivery] not hearing changes of %s yet (%v); asking again in %s", bucket, err, wait)