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 9b69039c..295641f1 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 @@ -11,6 +11,7 @@ import ( "io" "os" "os/exec" + "path/filepath" "strings" "sync" "testing" @@ -82,6 +83,15 @@ func newTestBus(t *testing.T) *testBus { } b.kvs[name] = kv } + // The buckets of other modules the manifest reads (`.` is the bucket `_`): the gate reads + // who has the seat there, and today's state to carry (seat.go). + for _, name := range m.Reads { + kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: strings.Replace(name, ".", "_", 1), History: 1}) + if err != nil { + t.Fatal(err) + } + b.kvs[name] = kv + } // 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.>"}}); err != nil { @@ -191,6 +201,15 @@ func (r *runtime) read(out io.Reader) { } } +// untilHolding waits for the gate to hold the seat: it holds it when it reads the controller's walks once, the first thing it +// does after (readOnHolding), and it takes no event before. +func (r *runtime) untilHolding() { + r.t.Helper() + r.until("the gate to hold the seat", 20*time.Second, func() bool { + return len(r.asksOf("seat:"+ControllerSeat+".delivery-walks")) > 0 + }) +} + func (r *runtime) ping() { select { case r.notify <- struct{}{}: @@ -284,6 +303,8 @@ func (r *runtime) answer(m rpcMessage) { return map[string]any{}, nil case "mesh/seat.record": return nil, nil + case "mesh/state.watch": + return map[string]any{}, nil // the runtime's watch: the module also reads the buckets by key, so no change is sent here case "mesh/ask": var a struct { Key string `json:"key"` @@ -429,6 +450,7 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB defer rt.mu.Unlock() return len(rt.subs) >= len(followed)+1 }) + rt.untilHolding() for _, k := range followed { if strings.Contains(k, "labeled") || strings.Contains(k, "commented") { t.Fatalf("the module follows %s: a label or a comment starts nothing here (ADR 0421 §4)", k) @@ -531,6 +553,7 @@ func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeInOnePassOv defer rt.mu.Unlock() return len(rt.subs) >= len(followed)+1 }) + rt.untilHolding() const ctlHead, catHead = "c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1", "d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2" for _, p := range []PullEvent{pr("novox/mesh-controller", 1, ctlHead, "feat/x", "internal/a.go"), pr("novox/mesh-catalog", 2, catHead, "feat/x", "modules/app/module.json")} { @@ -628,6 +651,7 @@ func TestAPlanMovedCarryingAFailedBackupHoldsTheDeliveryWithTheBackupsWordsOverA defer rt.mu.Unlock() return len(rt.subs) >= len(followed)+1 }) + rt.untilHolding() const head, merge = "3333333333333333333333333333333333333333", "4444444444444444444444444444444444444444" announced := pr("novox/app", 9, head, "fix/y", "src/a.go") announced.BaseSHA = "t1t1t1t1t1t1" @@ -675,3 +699,96 @@ func TestAPlanMovedCarryingAFailedBackupHoldsTheDeliveryWithTheBackupsWordsOverA t.Fatalf("the controller was asked %d times to let the held walk go or take its backups again", n) } } + +// **AC3 of issue 1347 (ADR 0437 §7's check): the gate carries a live mesh's open deliveries and groups, once.** The buckets of +// today's module are filled from testdata/live-capture-2026-10-20.json (its source is said in the file), the gate starts with +// today's module on no machine, and every delivery and group arrives in its own buckets unchanged but for the ask an Approve +// waited on; a second start copies nothing. +func TestTheGateCarriesALiveCapturesDeliveriesAndGroupsOnceOverARealBus(t *testing.T) { + raw, err := os.ReadFile(filepath.Join("testdata", "live-capture-2026-10-20.json")) + if err != nil { + t.Fatal(err) + } + var capture struct { + Deliveries []json.RawMessage `json:"deliveries"` + Groups []json.RawMessage `json:"groups"` + } + if err := json.Unmarshal(raw, &capture); err != nil { + t.Fatal(err) + } + b := newTestBus(t) + ctx := context.Background() + want := map[string]Delivery{} + for _, one := range capture.Deliveries { + var d Delivery + if err := json.Unmarshal(one, &d); err != nil { + t.Fatal(err) + } + if _, err := b.kvs["mesh-delivery.deliveries"].Put(ctx, kvKey(d.ID), one); err != nil { + t.Fatal(err) + } + if d.Ready != nil { // what the carry clears: the ask nobody answered + d.Ready.AskID, d.Ready.Ask, d.Ready.Heads = "", nil, nil + } + want[kvKey(d.ID)] = d + } + for _, one := range capture.Groups { + var g Group + if err := json.Unmarshal(one, &g); err != nil { + t.Fatal(err) + } + if _, err := b.kvs["mesh-delivery.groups"].Put(ctx, kvKey(g.ID), one); err != nil { + t.Fatal(err) + } + } + rt := b.start(t) + rt.untilHolding() + for key, d := range want { + e, err := b.kvs["deliveries"].Get(ctx, key) + if err != nil { + t.Fatalf("the delivery %s did not arrive: %v", key, err) + } + var got Delivery + if err := json.Unmarshal(e.Value(), &got); err != nil { + t.Fatal(err) + } + if g, w := string(mustJSON(got)), string(mustJSON(d)); g != w { + t.Fatalf("the delivery %s changed in the carry:\n got %s\nwant %s", key, g, w) + } + } + for _, one := range capture.Groups { + var g Group + _ = json.Unmarshal(one, &g) + e, err := b.kvs["groups"].Get(ctx, kvKey(g.ID)) + 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) + } + } + if _, err := b.kvs["deliveries"].Get(ctx, carriedKey); err != nil { + t.Fatalf("the mark carried was not written: %v", err) + } + // A second start copies nothing: a delivery today's module gains meanwhile stays where it is. + _ = rt.in.Close() + _ = rt.cmd.Wait() + late := mustJSON(&Delivery{ID: "novox/app@999999999999", Repository: "novox/app", Commit: "999999999999", Number: 99, State: Proposed}) + if _, err := b.kvs["mesh-delivery.deliveries"].Put(ctx, kvKey("novox/app@999999999999"), late); err != nil { + t.Fatal(err) + } + again := b.start(t) + again.untilHolding() + if _, err := b.kvs["deliveries"].Get(ctx, kvKey("novox/app@999999999999")); !errors.Is(err, jetstream.ErrKeyNotFound) { + t.Fatalf("a second start copied a key (%v)", err) + } +} + +func mustDecode[T any](t *testing.T, raw []byte) T { + t.Helper() + var v T + if err := json.Unmarshal(raw, &v); err != nil { + t.Fatal(err) + } + return v +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events.go index ac92d25d..7713e463 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events.go @@ -42,6 +42,9 @@ func (h *Holder) takeForgeWanted() bool { // AskDecided takes the router's `decided` event for an ask this owner made: its warrant is applied to the deliveries // the ask covers, and what is said on their pull requests is said. func (h *Holder) AskDecided(id, state string, w *asks.Warrant) { + if !h.Holding() { + return + } h.mu.Lock() said := h.decided(id, state, w) h.mu.Unlock() diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/fakes_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/fakes_test.go index e04f0196..4b85bc65 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/fakes_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/fakes_test.go @@ -62,7 +62,10 @@ func (s *memStore) Deliveries() ([]*Delivery, error) { s.mu.Lock() defer s.mu.Unlock() var out []*Delivery - for _, raw := range s.deliveries { + for key, raw := range s.deliveries { + if key == carriedKey { // the gate's mark that today's state was carried (seat.go), as the real store skips it + continue + } var d Delivery if err := json.Unmarshal(raw, &d); err != nil { return nil, err diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go index 362436c6..ff530ac8 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go @@ -9,6 +9,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" ) @@ -31,9 +32,18 @@ type Holder struct { Operator Operator // Pulls keeps the pull requests' keys on the bus (pullkey.go); nil keeps none. Pulls PullKeys + // Seat says where today's module still stands; the gate holds the seat only when it stands nowhere (seat.go). Carry + // copies today's state over at the first holding, OnHold takes the reads of a holder just come. + Seat Occupancy + Carry func() (CarryReport, error) + OnHold func() // cancels are asks to take back, gathered under the lock and run with the other asks of the look. cancels []func() + seatHeld atomic.Bool + seatMu sync.Mutex + seatWhy string + mu sync.Mutex deliveries map[string]*Delivery groups map[string]*Group 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 c8326410..765d2e8d 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -46,7 +46,11 @@ func main() { Logf: logf, Operator: operator, 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{}, + Carry: carryFromBus, } + h.OnHold = func() { go readOnHolding(h) } h.init() l := &listening{now: "not yet: starting"} go run(h, l, operator) @@ -56,21 +60,15 @@ func main() { } } -// run reads back the state, then takes the events, then reads once what it may have missed while it was away — the -// controller's walks, the forge's open pull requests, the operator's asks — and 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. +// run takes the events and the operator's answers, which a gate that does not hold the seat acknowledges and leaves +// alone, and watches who has the seat (watchSeat). The state is read back, after today's is carried, when the gate first +// 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, operator *asker.Client) { time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - err := h.Load() - if err == nil { - break - } - logf("[mesh-delivery] cannot read back its deliveries yet (%v); asking again in %s", err, wait) - time.Sleep(wait) - } go h.Work() + go watchSeat(h, l) for _, pattern := range followed { pattern := pattern for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { @@ -103,14 +101,16 @@ func run(h *Holder, l *listening, operator *asker.Client) { logf("[mesh-delivery] not hearing the operator's answers yet (%v); asking again in %s", err, wait) time.Sleep(wait) } - l.set("listening") logf("[mesh-delivery] listening: %s", strings.Join(followed, ", ")) - // Read once, now that events arrive: what happened while this owner was away. The forge's open pull requests (ready - // marks, heads nobody checks) are read by the first look; the asks the router closed meanwhile are read here. +} + +// readOnHolding reads once, now that the gate holds the seat and events arrive, what happened while it was away: the +// forge's open pull requests (heads nobody checks) are read by the first look, the asks the router closed meanwhile are read +// here, and the controller's walks, until they have been read once; plan-moved events carry every move after. +func readOnHolding(h *Holder) { h.wantForge() h.hearAsks() h.Poke() - // retries: the controller's walks, until they have been read once; plan-moved events carry every move after for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { err := h.Reconcile() if err == nil { @@ -125,6 +125,9 @@ func run(h *Holder, l *listening, operator *asker.Client) { // handle takes one event, then asks for a look at every step the event makes possible (Work, React). An event that cannot be read is // said and acknowledged: read again it would fail again, and the controller's walks are read back by comparison anyway. func handle(h *Holder, e stdio.Envelope) error { + if !h.Holding() { + return nil // not the holder of the seat: the event is acknowledged and taken by no one (seat.go) + } taken := take(h, e) h.Poke() return taken @@ -412,7 +415,11 @@ func tools(h *Holder, l *listening) []stdio.Tool { // A verb that changes a delivery asks for a look at what it makes possible (react.go). for i := range list { run := list[i].Run + ofSeat := strings.HasPrefix(list[i].Name, seat) list[i].Run = func(a map[string]any) (any, error) { + if ofSeat && !h.Holding() { + return nil, h.notHolder() // a verb of the seat is answered by its holder only + } out, err := run(a) h.Poke() return out, err diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ports.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ports.go index c893fe0f..7ca17df5 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ports.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ports.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" "strconv" "strings" @@ -458,7 +459,8 @@ func (kvStore) PutGroup(g *Group) error { } func (kvStore) DeleteGroup(id string) error { return stdio.State("groups").Delete(kvKey(id)) } -func readAll[T any](state string) ([]*T, error) { +// readAll reads every key of a state but the ones it is told to leave (the gate's mark `carried` is no delivery). +func readAll[T any](state string, leave ...string) ([]*T, error) { keys, err := stdio.State(state).Keys() if err != nil { return nil, err @@ -466,6 +468,9 @@ func readAll[T any](state string) ([]*T, error) { sort.Strings(keys) var out []*T for _, k := range keys { + if slices.Contains(leave, k) { + continue + } e, err := stdio.State(state).Get(k) if err != nil { return nil, err @@ -483,5 +488,5 @@ func readAll[T any](state string) ([]*T, error) { return out, nil } -func (kvStore) Deliveries() ([]*Delivery, error) { return readAll[Delivery]("deliveries") } +func (kvStore) Deliveries() ([]*Delivery, error) { return readAll[Delivery]("deliveries", carriedKey) } func (kvStore) Groups() ([]*Group, error) { return readAll[Group]("groups") } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/react.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/react.go index a1edeff8..165c71b5 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/react.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/react.go @@ -61,6 +61,9 @@ func (h *Holder) Work() { func (h *Holder) React() { h.reactMu.Lock() defer h.reactMu.Unlock() + if !h.Holding() { + return // not the holder of the seat: no look, no write, no ask (seat.go) + } h.wanted.mu.Lock() h.wanted.at, h.wanted.step = time.Time{}, "" h.wanted.mu.Unlock() diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go new file mode 100644 index 00000000..3160c98e --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat.go @@ -0,0 +1,279 @@ +package main + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// The gate acts only while it holds the seat, and carries today's state over when it first holds it (novox/hq ADR 0450 +// §6 and its insight of 2026-10-20; ADR 0437 §7). It holds the seat when no machine is assigned today's module +// (mesh-controller.assignments, keys `.`) and no machine's node-engine says it has the module (node-engine.have, +// the same keys): the seat passes by ADR 0449 §4, so the gate never has to take it. While it does not hold it, it takes +// no event, writes no key, asks nothing and answers no verb, and says in its log why it waits. + +// OldModule is today's module, whose place the gate takes: the module `mesh-delivery`. +const OldModule = "mesh-delivery" + +// Occupancy is where today's module still stands. +type Occupancy interface { + // Occupants says, one line each, where the module is assigned ("assigned on ") or has a have key ("on + // the node-engine has it"); none when no machine has it. + Occupants() ([]string, error) +} + +// Bucket is a bucket of the bus: today's module's, read, or the gate's own, written. +type Bucket interface { + Keys() ([]string, error) + // Get is a key's value, or nil when it has none. + Get(key string) ([]byte, error) + Put(key string, raw []byte) error +} + +// carriedKey is the key, in the gate's bucket `deliveries`, that says today's state was carried: no later start or holding +// copies again. +const carriedKey = "carried" + +// CarryReport says what a carry did. +type CarryReport struct { + // Already is true when the key `carried` was there: nothing was copied. + Already bool + // Copied is, by bucket, how many keys were copied. + Copied map[string]int + // Cleared is how many deliveries had the id of an ask of today's module taken off. + Cleared int +} + +// SeatChanged reads who has today's module and moves the gate to or from holding the seat. The first time it finds the seat +// free it carries today's state over (Carry), reads it back (Load) and only then holds; a failure of either is returned and +// leaves the gate silent, to be tried again. +func (h *Holder) SeatChanged() error { + if h.Seat == nil { + return nil + } + who, err := h.Seat.Occupants() + if err != nil { + h.Logf("[mesh-delivery] cannot tell who has the seat yet: %v", err) + return err + } + if len(who) > 0 { + why := "today's module is not gone yet: " + strings.Join(who, "; ") + if h.seatHeld.Swap(false) { + h.Logf("[mesh-delivery] the gate no longer holds the seat and is silent: %s", why) + } else if h.setSeatWhy(why) { + h.Logf("[mesh-delivery] the gate waits for the seat and is silent: %s", why) + } + h.setSeatWhy(why) + return nil + } + if h.seatHeld.Load() { + return nil + } + if h.Carry != nil { + report, err := h.Carry() + if err != nil { + h.Logf("[mesh-delivery] today's state cannot be carried yet: %v", err) + return err + } + if report.Already { + h.Logf("[mesh-delivery] today's state was carried before: nothing is copied") + } else { + h.Logf("[mesh-delivery] today's state is carried: %v copied, %d ask id(s) cleared", report.Copied, report.Cleared) + } + } + if err := h.Load(); err != nil { + h.Logf("[mesh-delivery] the carried state cannot be read back yet: %v", err) + return err + } + h.setSeatWhy("") + h.seatHeld.Store(true) + h.Logf("[mesh-delivery] the gate holds the seat: today's module is on no machine") + if h.OnHold != nil { + h.OnHold() + } + h.Poke() + return nil +} + +// Holding is whether the gate acts. A holder with no Seat to ask always holds (the tests'; main sets one). +func (h *Holder) Holding() bool { return h.Seat == nil || h.seatHeld.Load() } + +// setSeatWhy keeps why the gate waits and says whether it changed. +func (h *Holder) setSeatWhy(why string) bool { + h.seatMu.Lock() + defer h.seatMu.Unlock() + changed := h.seatWhy != why + h.seatWhy = why + return changed +} + +// SeatWhy is why the gate is silent, or empty when it holds. +func (h *Holder) SeatWhy() string { + h.seatMu.Lock() + defer h.seatMu.Unlock() + return h.seatWhy +} + +// notHolder is the answer of a verb asked of a gate that does not hold the seat. +func (h *Holder) notHolder() error { + return fmt.Errorf("not the holder: the gate does not hold the seat mesh-delivery yet, and answers nothing until it does (%s)", h.SeatWhy()) +} + +// CarryOver copies today's state into the gate's own, once. It does nothing when `carried` is in the gate's deliveries. +// +// For each of today's module's buckets `deliveries`, `groups` and `pulls` it copies every key the gate's bucket lacks, and a +// key the gate has is never touched. A delivery whose Approve nobody answered has the ask of today's module taken off, since +// that ask lives in a router the gate does not ask under: the gate asks again through the teller, as an ask that ended without +// a choice is asked again (ADR 0421 §6). A delivery whose Approve was given keeps its answer, which the merge still needs. +// The mark `carried` is written last, so a carry that stopped halfway is done again and a finished one never is. +func CarryOver(old, own func(bucket string) Bucket, at time.Time) (CarryReport, error) { + report := CarryReport{Copied: map[string]int{}} + marks := own("deliveries") + if mark, err := marks.Get(carriedKey); err != nil { + return report, fmt.Errorf("the mark that today's state was carried cannot be read: %w", err) + } else if mark != nil { + report.Already = true + return report, nil + } + for _, bucket := range []string{"deliveries", "groups", "pulls"} { + keys, err := old(bucket).Keys() + if err != nil { + return report, fmt.Errorf("the keys of %s.%s cannot be read: %w", OldModule, bucket, err) + } + sort.Strings(keys) + for _, key := range keys { + if bucket == "deliveries" && key == carriedKey { + continue + } + have, err := own(bucket).Get(key) + if err != nil { + return report, fmt.Errorf("%s %s of the gate cannot be read: %w", bucket, key, err) + } + if have != nil { + continue + } + raw, err := old(bucket).Get(key) + if err != nil { + return report, fmt.Errorf("%s.%s %s cannot be read: %w", OldModule, bucket, key, err) + } + if raw == nil { + continue + } + if bucket == "deliveries" { + var d Delivery + if err := json.Unmarshal(raw, &d); err != nil { + return report, fmt.Errorf("the delivery %s of %s.deliveries is not readable: %w", key, OldModule, err) + } + if r := d.Ready; r != nil && r.AskID != "" && r.Approved == nil && r.Declined == nil { + r.AskID, r.Ask, r.Heads = "", nil, nil + report.Cleared++ + if raw, err = json.Marshal(&d); err != nil { + return report, err + } + } + } + if err := own(bucket).Put(key, raw); err != nil { + return report, fmt.Errorf("%s %s cannot be written: %w", bucket, key, err) + } + report.Copied[bucket]++ + } + } + mark, err := json.Marshal(map[string]any{"v": 1, "at": at.UTC().Format(time.RFC3339), "from": OldModule, + "deliveries": report.Copied["deliveries"], "groups": report.Copied["groups"], "pulls": report.Copied["pulls"], "cleared": report.Cleared}) + if err != nil { + return report, err + } + if err := marks.Put(carriedKey, mark); err != nil { + return report, fmt.Errorf("the mark that today's state was carried cannot be written: %w", err) + } + return report, nil +} + +// busOccupancy reads the controller's assignments and the node-engine's have keys, both `.`. +type busOccupancy struct{} + +func (busOccupancy) Occupants() ([]string, error) { + 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 := range keys { + if node, module, ok := strings.Cut(key, "."); ok && module == OldModule { + who = append(who, fmt.Sprintf(src.says, node)) + } + } + } + sort.Strings(who) + return who, nil +} + +// 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 } + +func (b keptBucket) Keys() ([]string, error) { return stdio.State(b.name).Keys() } +func (b keptBucket) Get(key string) ([]byte, error) { + e, err := stdio.State(b.name).Get(key) + if err != nil || e == nil { + return nil, err + } + return e.Value, nil +} +func (b keptBucket) Put(key string, raw []byte) error { + _, err := stdio.State(b.name).Put(key, json.RawMessage(raw)) + return err +} + +// carryFromBus is the carry over the runtime: today's module's buckets into the gate's own. +func carryFromBus() (CarryReport, error) { + return CarryOver(func(b string) Bucket { return keptBucket{OldModule + "." + b} }, func(b string) Bucket { return keptBucket{b} }, time.Now()) +} + +// 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) { + changed := make(chan struct{}, 1) + note := func(stdio.StateChange) error { + select { + case changed <- struct{}{}: + default: + } + return nil + } + for _, bucket := range []string{"mesh-controller.assignments", "node-engine.have"} { + go func() { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.State(bucket).Watch("", note) + if err == nil { + return + } + logf("[mesh-delivery] not hearing changes of %s yet (%v); asking again in %s", bucket, err, wait) + time.Sleep(wait) + } + }() + } + for { + err := h.SeatChanged() + switch { + case err != nil: + l.set("not yet: " + err.Error()) + select { + case <-changed: + case <-time.After(10 * time.Second): + } + default: + if h.Holding() { + l.set("listening") + } else { + l.set("waiting: not the holder of the seat (" + h.SeatWhy() + ")") + } + <-changed + } + } +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat_test.go new file mode 100644 index 00000000..cb91b961 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/seat_test.go @@ -0,0 +1,323 @@ +package main + +import ( + "bytes" + "errors" + "slices" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// **The gate acts only while it holds the seat, and carries today's state over once** (novox/hq ADR 0450 §6 with its +// insight of 2026-10-20, ADR 0437 §7, issue 1347). Fixture sources: the keys of mesh-controller.assignments and +// node-engine.have are `.` (mesh-controller internal/inventory/chain_assignments.go assignmentKey, +// mesh-sdk go/delivery HaveKey); a delivery is the module's own Delivery, written as ports.go kvStore writes it. + +// fakeOccupancy is where today's module stands, as the test says. +type fakeOccupancy struct { + mu sync.Mutex + who []string + err error +} + +func (o *fakeOccupancy) set(who ...string) { o.mu.Lock(); o.who = who; o.mu.Unlock() } +func (o *fakeOccupancy) Occupants() ([]string, error) { + o.mu.Lock() + defer o.mu.Unlock() + return slices.Clone(o.who), o.err +} + +// mapBucket is a bucket over a map, shared with the store a test reads. +type mapBucket struct { + mu *sync.Mutex + m map[string][]byte +} + +func (b mapBucket) Keys() ([]string, error) { + b.mu.Lock() + defer b.mu.Unlock() + var keys []string + for k := range b.m { + keys = append(keys, k) + } + return keys, nil +} +func (b mapBucket) Get(key string) ([]byte, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.m[key], nil +} +func (b mapBucket) Put(key string, raw []byte) error { + b.mu.Lock() + defer b.mu.Unlock() + b.m[key] = raw + return nil +} + +// seatTest is a world whose holder asks where today's module stands. +func seatTest(t *testing.T, who ...string) (*world, *fakeOccupancy) { + t.Helper() + w := newWorld(t) + occ := &fakeOccupancy{who: who} + w.h.Seat = occ + return w, occ +} + +// anEvent is the runtime handing the holder a pull request's head, as handle takes it. +func anEvent(t *testing.T, w *world) { + t.Helper() + hear(t, w, "gitea.pull.updated", pr("novox/app", 7, readyHead, "feat/x", "src/a.go")) +} + +// nothingWritten says the forge, the controller and the store were left alone. +func nothingWritten(t *testing.T, w *world) { + t.Helper() + w.forge.mu.Lock() + forge := len(w.forge.notes) + len(w.forge.views) + len(w.forge.statuses) + w.forge.statusWrites + len(w.forge.calls) + len(w.forge.comments) + w.forge.mu.Unlock() + w.ctl.mu.Lock() + ctl := len(w.ctl.checks) + w.ctl.asked + len(w.ctl.delivers) + len(w.ctl.stops) + w.ctl.mu.Unlock() + w.store.mu.Lock() + keys := len(w.store.deliveries) + len(w.store.groups) + w.store.mu.Unlock() + if forge != 0 || ctl != 0 || keys != 0 { + t.Fatalf("a gate that does not hold the seat wrote: %d on the forge, %d to the controller, %d keys", forge, ctl, keys) + } + if n := len(w.h.deliveries); n != 0 { + t.Fatalf("a gate that does not hold the seat took the event: %d deliveries", n) + } +} + +func toolNamed(t *testing.T, h *Holder, name string) stdio.Tool { + t.Helper() + for _, tool := range tools(h, &listening{}) { + if tool.Name == name { + return tool + } + } + t.Fatalf("no tool %s", name) + return stdio.Tool{} +} + +// AC1: with the module assigned on a machine, or its have key present, an event changes no key and a verb answers that +// the gate is not the holder. +func TestTheGateIsSilentWhileTodaysModuleIsAssignedOrHasAHaveKey(t *testing.T) { + for _, who := range []string{"assigned on novox", "the node-engine of ace has it"} { + t.Run(who, func(t *testing.T) { + w, _ := seatTest(t, who) + if err := w.h.SeatChanged(); err != nil { + t.Fatal(err) + } + if w.h.Holding() { + t.Fatalf("the gate holds the seat while %s", who) + } + anEvent(t, w) + nothingWritten(t, w) + // Every verb of the seat answers that it is not the holder, and none changes anything. + for _, name := range []string{"mesh-delivery.deliveries", "mesh-delivery.release", "mesh-delivery.stop", "mesh-delivery.recheck"} { + _, err := toolNamed(t, w.h, name).Run(map[string]any{"id": "novox/app@111111111111", "why": "a test"}) + if err == nil || !strings.Contains(err.Error(), "not the holder") || !strings.Contains(err.Error(), who) { + t.Fatalf("%s answered %v, wanted not the holder, naming %q", name, err, who) + } + } + nothingWritten(t, w) + // Its own status still answers, and says why it is silent. + out, err := toolNamed(t, w.h, "delivery_status").Run(nil) + if err != nil { + t.Fatal(err) + } + status := out.(map[string]any) + if status["holds-the-seat"] != false || !strings.Contains(status["seat-waits"].(string), who) { + t.Fatalf("status %v", status) + } + }) + } +} + +// The seat free, the gate holds it and acts; asked again with the module back, it is silent again. +func TestTheGateActsOnceTodaysModuleIsGoneAndStopsIfItComesBack(t *testing.T) { + w, occ := seatTest(t) + if err := w.h.SeatChanged(); err != nil { + t.Fatal(err) + } + if !w.h.Holding() { + t.Fatal("the seat is free and the gate does not hold it") + } + anEvent(t, w) + if len(w.h.deliveries) != 1 { + t.Fatalf("the gate that holds the seat took %d deliveries, wanted 1", len(w.h.deliveries)) + } + if _, err := toolNamed(t, w.h, "mesh-delivery.deliveries").Run(map[string]any{}); err != nil { + t.Fatalf("a verb of the holder: %v", err) + } + occ.set("assigned on novox") + if err := w.h.SeatChanged(); err != nil { + t.Fatal(err) + } + if w.h.Holding() { + t.Fatal("the module is assigned again and the gate still holds") + } + hear(t, w, "gitea.pull.updated", pr("novox/app", 8, "2222222222222222222222222222222222222222", "feat/y", "src/b.go")) + if len(w.h.deliveries) != 1 { + t.Fatalf("an event was taken while the gate was silent: %d deliveries", len(w.h.deliveries)) + } +} + +// Nothing owed is flushed, and no look is taken, while the gate is silent. +func TestASilentGateFlushesNothingItOwes(t *testing.T) { + w, occ := seatTest(t) + if err := w.h.SeatChanged(); err != nil { + t.Fatal(err) + } + w.h.PullUpdated(pr("novox/app", 7, readyHead, "feat/x", "src/a.go")) // owes a note, a view and a status + occ.set("assigned on novox") + if err := w.h.SeatChanged(); err != nil { + t.Fatal(err) + } + counts := func() [3]int { + w.forge.mu.Lock() + defer w.forge.mu.Unlock() + w.store.mu.Lock() + defer w.store.mu.Unlock() + return [3]int{len(w.forge.notes) + len(w.forge.views) + len(w.forge.statuses) + w.forge.statusWrites + len(w.forge.calls), w.store.puts, len(w.said)} + } + before := counts() + w.h.React() + if after := counts(); after != before { + t.Fatalf("a silent gate flushed what it owes: forge, store and events %v before, %v after", before, after) + } +} + +// A seat that cannot be read leaves a gate that did not hold silent, and says so. +func TestTheGateStaysSilentWhenTheSeatCannotBeRead(t *testing.T) { + w, occ := seatTest(t) + occ.err = errors.New("the bus is away") + if err := w.h.SeatChanged(); err == nil { + t.Fatal("a seat that cannot be read was taken as free") + } + if w.h.Holding() { + t.Fatal("holding without knowing") + } +} + +// --- AC2: the carry --------------------------------------------------------------------------------- + +func aDelivery(id string, number int, ready *ReadyRun) []byte { + return mustJSON(&Delivery{ID: id, Repository: "novox/app", Commit: strings.Repeat("1", 40), Number: number, Base: "main", + State: Proposed, Ready: ready}) +} + +type carryFixture struct { + oldD, oldG, oldP map[string][]byte + mu sync.Mutex + own map[string]mapBucket + w *world + occ *fakeOccupancy + approvedAt time.Time +} + +// aCarry is a world whose holder carries from these buckets into its store's: the old module's deliveries, one with an ask +// waiting, one approved and one the gate already has under another value; one group; one pull key. +func aCarry(t *testing.T) *carryFixture { + f := &carryFixture{oldD: map[string][]byte{}, oldG: map[string][]byte{}, oldP: map[string][]byte{}, + approvedAt: time.Date(2026, 10, 20, 9, 0, 0, 0, time.UTC)} + f.w, f.occ = seatTest(t) + ask := &asks.Ask{ID: "merge-7-aaaaaaaaaaaa-1"} + f.oldD["novox/app@aaaaaaaaaaaa"] = aDelivery("novox/app@aaaaaaaaaaaa", 7, &ReadyRun{Head: strings.Repeat("a", 40), AskID: "merge-7-aaaaaaaaaaaa-1", + Ask: ask, Heads: map[string]string{"novox/app@aaaaaaaaaaaa": strings.Repeat("a", 40)}, Asks: 1}) + f.oldD["novox/app@bbbbbbbbbbbb"] = aDelivery("novox/app@bbbbbbbbbbbb", 8, &ReadyRun{Head: strings.Repeat("b", 40), AskID: "merge-8-bbbbbbbbbbbb-1", + Ask: &asks.Ask{ID: "merge-8-bbbbbbbbbbbb-1"}, Heads: map[string]string{"novox/app@bbbbbbbbbbbb": strings.Repeat("b", 40)}, + Approved: &f.approvedAt, Warrant: "the operator on the phone"}) + f.oldD["novox/app@cccccccccccc"] = aDelivery("novox/app@cccccccccccc", 9, &ReadyRun{Head: strings.Repeat("c", 40), AskID: "merge-9-cccccccccccc-1", Ask: ask}) + f.oldG["feat_x"] = []byte(`{"id":"feat/x","members":["novox/app@aaaaaaaaaaaa"]}`) + f.oldP["novox_app_7"] = []byte(`{"v":1,"number":7,"head":"` + strings.Repeat("a", 40) + `"}`) + // The gate already has one of the old module's deliveries, otherwise. + f.w.store.deliveries["novox/app@cccccccccccc"] = aDelivery("novox/app@cccccccccccc", 9, &ReadyRun{Head: strings.Repeat("c", 40)}) + ownG, ownP := map[string][]byte{}, map[string][]byte{} + f.own = map[string]mapBucket{ + "deliveries": {mu: &f.w.store.mu, m: f.w.store.deliveries}, + "groups": {mu: &f.mu, m: ownG}, + "pulls": {mu: &f.mu, m: ownP}, + } + old := map[string]mapBucket{"deliveries": {mu: &f.mu, m: f.oldD}, "groups": {mu: &f.mu, m: f.oldG}, "pulls": {mu: &f.mu, m: f.oldP}} + f.w.h.Carry = func() (CarryReport, error) { + return CarryOver(func(b string) Bucket { return old[b] }, func(b string) Bucket { return f.own[b] }, f.approvedAt) + } + return f +} + +// AC2: when both go, the gate carries once (old ask ids cleared, keys it already has untouched) and then acts; a restart +// copies nothing. +func TestTheGateCarriesTodaysStateOnceWhenItFirstHoldsTheSeat(t *testing.T) { + f := aCarry(t) + before := slices.Clone(f.w.store.deliveries["novox/app@cccccccccccc"]) + f.occ.set("assigned on novox") + if err := f.w.h.SeatChanged(); err != nil { + t.Fatal(err) + } + if len(f.w.store.deliveries) != 1 || f.w.h.Holding() { + t.Fatalf("the gate carried while the module was still assigned: %d keys", len(f.w.store.deliveries)) + } + f.occ.set() + if err := f.w.h.SeatChanged(); err != nil { + t.Fatal(err) + } + if !f.w.h.Holding() { + t.Fatal("the gate does not hold the seat after the module went") + } + d := f.w.h.deliveries + for _, id := range []string{"novox/app@aaaaaaaaaaaa", "novox/app@bbbbbbbbbbbb", "novox/app@cccccccccccc"} { + if d[id] == nil { + t.Fatalf("%s was not carried and read back: %v", id, f.w.store.deliveries) + } + } + // The ask nobody answered is cleared, so the gate asks again; the one answered keeps its answer. + if r := d["novox/app@aaaaaaaaaaaa"].Ready; r.AskID != "" || r.Ask != nil || r.Heads != nil || r.Asks != 1 { + t.Fatalf("the unanswered ask was not cleared: %+v", r) + } + if r := d["novox/app@bbbbbbbbbbbb"].Ready; r.Approved == nil || r.AskID == "" || r.Warrant == "" { + t.Fatalf("an answered Approve lost its answer: %+v", r) + } + // A key the gate had is untouched. + if !bytes.Equal(f.w.store.deliveries["novox/app@cccccccccccc"], before) { + t.Fatal("a key the gate already had was overwritten") + } + for bucket, key := range map[string]string{"groups": "feat_x", "pulls": "novox_app_7"} { + if got, _ := f.own[bucket].Get(key); got == nil { + t.Fatalf("%s %s was not carried", bucket, key) + } + } + if got, _ := f.own["deliveries"].Get(carriedKey); got == nil { + t.Fatal("the key carried was not written") + } + // A restart (a holder that holds again, or starts again) copies nothing. + f.oldD["novox/app@dddddddddddd"] = aDelivery("novox/app@dddddddddddd", 10, nil) + report, err := f.w.h.Carry() + if err != nil || !report.Already { + t.Fatalf("a second carry: %+v %v", report, err) + } + if got, _ := f.own["deliveries"].Get("novox/app@dddddddddddd"); got != nil { + t.Fatal("a second carry copied a key") + } +} + +// A carry that fails leaves the gate silent and writes no mark, so it is tried again. +func TestAFailedCarryLeavesTheGateSilentAndUnmarked(t *testing.T) { + f := aCarry(t) + f.w.h.Carry = func() (CarryReport, error) { return CarryReport{}, errors.New("the old bucket cannot be read") } + if err := f.w.h.SeatChanged(); err == nil { + t.Fatal("a failed carry was not returned") + } + if f.w.h.Holding() { + t.Fatal("the gate holds the seat on state it did not carry") + } + if got, _ := f.own["deliveries"].Get(carriedKey); got != nil { + t.Fatal("a failed carry wrote its mark") + } +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/testdata/live-capture-2026-10-20.json b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/testdata/live-capture-2026-10-20.json new file mode 100644 index 00000000..98d6661f --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/testdata/live-capture-2026-10-20.json @@ -0,0 +1,16 @@ +{ + "source": "mesh-delivery.deliveries and mesh-delivery.groups, called on the live mesh at 2026-10-20 about 02:45 (local) by fix-r3e-gate@shanks; there is no tool that reads a key's value (issue 1333), so each value is built from what the verbs show: id, pull request number, state, since, group, and for the group's members base, head and state. What the verbs do not show is filled with the least that reads: the commit of a delivery is its id's twelve characters (the verbs show no full head), the base is main, created is since. The two ask ids under ready are synthetic (the verbs do not show an ask id): they stand for an Approve asked and not answered, which the carry must clear.", + "deliveries": [ + {"id": "novox/mesh-catalog@b9104454f9a5", "repository": "novox/mesh-catalog", "commit": "b9104454f9a5", "number": 378, "base": "main", "state": "delivering", "since": "2026-10-19T22:58:32Z", "created": "2026-10-19T22:58:32Z"}, + {"id": "novox/mesh-sdk@1c20b9fb9b25", "repository": "novox/mesh-sdk", "commit": "1c20b9fb9b25", "number": 43, "base": "main", "state": "proposed", "since": "2026-10-19T20:39:11Z", "created": "2026-10-19T20:39:11Z"}, + {"id": "novox/de-spiegel@fdcbaefb3123", "repository": "novox/de-spiegel", "commit": "fdcbaefb3123", "number": 4, "base": "main", "state": "ready", "since": "2026-10-16T20:10:46Z", "created": "2026-10-16T20:10:46Z", "ready": {"head": "fdcbaefb3123", "seen": "2026-10-20T00:00:00Z", "last_try": "0001-01-01T00:00:00Z", "ask_id": "merge-4-fdcbaefb3123-1", "asked_at": "2026-10-20T00:00:00Z", "asks": 1, "heads": {"fdcbaefb3123": "fdcbaefb3123"}}}, + {"id": "novox/hq@1552b0e0a3ec", "repository": "novox/hq", "commit": "1552b0e0a3ec", "number": 384, "base": "main", "state": "proposed", "since": "2026-10-16T15:58:05Z", "created": "2026-10-16T15:58:05Z"}, + {"id": "novox/mesh-controller@0350b4e22276", "repository": "novox/mesh-controller", "commit": "0350b4e22276", "number": 325, "base": "main", "state": "proposed", "since": "2026-10-14T13:22:26Z", "created": "2026-10-14T13:22:26Z"}, + {"id": "novox/mesh-catalog@1fe6ea4ee473", "repository": "novox/mesh-catalog", "commit": "1fe6ea4ee473", "number": 152, "base": "main", "state": "ready", "since": "2026-10-09T07:54:44Z", "created": "2026-10-09T07:54:44Z", "ready": {"head": "1fe6ea4ee473", "seen": "2026-10-20T00:00:00Z", "last_try": "0001-01-01T00:00:00Z", "ask_id": "merge-152-1fe6ea4ee473-1", "asked_at": "2026-10-20T00:00:00Z", "asks": 1, "heads": {"1fe6ea4ee473": "1fe6ea4ee473"}}}, + {"id": "novox/mesh-catalog@795806efa7ad", "repository": "novox/mesh-catalog", "commit": "795806efa7ad95fa5cb3210b527117ad0316f4b9", "number": 350, "base": "main", "state": "stopped", "since": "2026-10-09T07:00:00Z", "created": "2026-10-09T07:00:00Z", "group": "feat/whole-modules"}, + {"id": "novox/mesh-controller@874a91a3c64f", "repository": "novox/mesh-controller", "commit": "874a91a3c64ff4a4dfa6b55b85101622e0e2e550", "number": 370, "base": "main", "state": "failed", "since": "2026-10-09T07:00:00Z", "created": "2026-10-09T07:00:00Z", "group": "feat/whole-modules"} + ], + "groups": [ + {"id": "feat/whole-modules", "members": ["novox/mesh-catalog@795806efa7ad", "novox/mesh-controller@874a91a3c64f"], "created": "2026-10-09T07:00:00Z", "updated": "2026-10-09T07:00:00Z"} + ] +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/verbs.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/verbs.go index 3a7dbce7..9a3e5aaf 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/verbs.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/verbs.go @@ -705,7 +705,7 @@ func (h *Holder) Status(listening string) map[string]any { } out := map[string]any{"deliveries": byState, "groups": len(h.groups), "owed": owed, "unkept": len(h.dirty), "walks-unmatched": len(h.unmatched), "events": listening, "started": h.started.Format(time.RFC3339), - "seat-on-record": h.held} + "seat-on-record": h.held, "holds-the-seat": h.Holding(), "seat-waits": h.SeatWhy()} // The owed effects tried since this owner started (novox/hq issue 407): done, failed and to be tried again, // or given up, and when the last was tried. A count since the start, said with that start: two readings in // one run give a window's tries.