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)

This commit is contained in:
2026-10-20 01:20:09 +00:00
parent 3b03c220ad
commit 68117bd74d
4 changed files with 201 additions and 21 deletions
@@ -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.<module>.event.<name>, mesh.seat.<seat>.event.<name>).
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 {
@@ -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) {
@@ -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)
}
}
}
@@ -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 `<node>.<module>`.
type busOccupancy struct{}
// What a bucket says of where today's module stands: the values by key, `<node>.<module>`, 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 `<module>.<name>`, 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)