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 939097202..e463c1359 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -75,6 +75,7 @@ func main() { } h.init() h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention} + h.Merges = recordedMerges{MergeStore: kvMerges{}, records: h.Records, members: h.groupMembers} h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf} l := &listening{now: "not yet: starting"} go run(h, l) @@ -95,6 +96,7 @@ func main() { func run(h *Holder, l *listening) { <-stdio.Served() // Serve first: the state is reached through it, and an event is heard from as soon as it serves loadKept(h, l) + switchToRecords(h, l) go h.Work() for _, pattern := range followed { pattern := pattern @@ -171,6 +173,21 @@ func loadKept(h *Holder, l *listening) { } } +// switchToRecords carries the groups in flight into their delivery records (switch.go), retrying until it can: no event is +// taken before, so no outcome reaches a record the switch has not opened yet. +func switchToRecords(h *Holder, l *listening) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + opened, found, err := h.Switch() + if err == nil { + logf("[mesh-delivery] the switch to the delivery records: %d opened, %d already there", opened, found) + return + } + l.set("not yet: " + err.Error()) + logf("[mesh-delivery] the switch to the delivery records cannot be made yet (%v); asking again in %s", err, wait) + time.Sleep(wait) + } +} + // readOnStart reads once what happened while the gate was away: the forge's open pull requests (heads nobody checks) are // read by the first look, and the asks the router closed meanwhile are read here. func readOnStart(h *Holder) { diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go index 3d1ac5f5c..c1b11d87a 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" ) @@ -403,3 +404,48 @@ func TestTheRecordsOutcomesArriveThroughTheGatesConsumer(t *testing.T) { } }) } + +// W11.3 (design 47 *The switch carries today's state in, once*): at the gate's start every group in flight becomes a +// record at its heads; a second switch (a restart, or milestone G delivered again after a revert) opens none and asks +// nothing; and from then on every merge the gate hands is in its record, whose merge commits the record's outcomes +// are found by. +func TestTheSwitchCarriesTheGroupsInFlightIntoRecordsOnce(t *testing.T) { + w := approvedGroup(t) + l := newLedgerFake(t) + w.h.Records = &Records{Ledger: l, Now: w.h.Now} + w.h.Merges = recordedMerges{MergeStore: w.merges, records: w.h.Records, members: w.h.groupMembers} + opened, found, err := w.h.Switch() + if err != nil || opened != 1 || found != 0 { + t.Fatalf("the first switch: %d opened, %d found, %v", opened, found, err) + } + asked := w.op.count() + opened, found, err = w.h.Switch() + if err != nil || opened != 0 || found != 1 || w.op.count() != asked { + t.Fatalf("the switch again: %d opened, %d found, %v, %d asks (was %d)", opened, found, err, w.op.count(), asked) + } + w.op.answer(w.op.open()[0], OptionApprove) + w.settleAll() + w.settleAll() + records, err := w.h.Records.Current() + if err != nil || len(records) != 1 { + t.Fatalf("records %+v %v: the handed merge opened another", records, err) + } + r := records[0] + if r.Milestone != "feat/g" || r.Group == nil || r.Group.Ask == "" || len(r.Group.Pulls) != 2 || len(r.Group.Members) != 2 { + t.Fatalf("the record's group state %+v", r.Group) + } + for _, p := range r.Group.Pulls { + if p.State != string(mdc.MergeMerged) || p.Commit == "" { + t.Fatalf("a member's merge is not in the record: %+v", p) + } + } + c := &StepConsumer{Records: w.h.Records, Families: stepFamilies, Now: w.h.Now} + body := `{"v":1,"kind":"taken","key":"m.00000000000000000001","what":"revision","merges":[{"repository":"` + gCat + + `","number":2,"merge":"` + r.Group.Pulls[1].Commit + `"}],"since":"2026-11-09T20:00:00Z"}` + if err := c.Hear(stdio.Envelope{Key: "mesh-delivery-indexer.take-in.taken", Body: json.RawMessage(body)}); err != nil { + t.Fatal(err) + } + if back, _ := w.h.Records.Read(r.Instance); back.Outcomes[OutcomeKey("", "", "", "take-in")].Outcome != OutcomeTakenIn { + t.Fatalf("the take-in of the group's merge is not its record's outcome: %+v", back.Outcomes) + } +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go new file mode 100644 index 000000000..02f9893bc --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go @@ -0,0 +1,141 @@ +package main + +// The switch to the delivery record (design 47 *The delivery record*: "The switch carries today's state in, once"): at +// the new gate's start, every group in flight and its merge as `merges` holds it become records, keyed by the +// milestone and the heads, and from then on every merge the gate hands is written into its record. It is safe to run +// twice: a group in flight whose record is there already leaves it as it is (Records.Open), and a group state the +// record holds already is not written again (Records.SetGroup), so delivering milestone G again after a revert makes no +// second record and asks nothing again. + +import ( + "fmt" + "sort" + + mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" +) + +// milestoneOf is the milestone a delivery's record is kept under: the name its group's Approve is asked and its merge is +// handed by (askGroup), the one name the gate gives a group's delivery today. +func milestoneOf(d *Delivery) string { return askGroup(d) } + +// recordGroupOf is a merge, as `merges` holds it, as the record's group state. +func recordGroupOf(m Merge, members []string) RecordGroup { + g := RecordGroup{Group: m.Group, Members: members, Ask: m.Ask, Approved: m.Approved} + for _, p := range m.Pulls { + g.Pulls = append(g.Pulls, RecordPull{Repository: p.Repository, Number: p.Number, Head: p.Head.Head, State: string(p.State), + Commit: p.Commit, Words: p.Words}) + } + return g +} + +// recordedMerges is the gate's merges with every merge it writes also written into its delivery record: the record holds +// the group's state, which `merges` holds today. +type recordedMerges struct { + MergeStore + records *Records + members func(group string) []string +} + +func (m recordedMerges) Put(group string, v Merge) error { + if err := m.MergeStore.Put(group, v); err != nil { + return err + } + if m.records == nil || len(v.Approved) == 0 { + return nil + } + r, _, err := m.records.Open(group, v.Approved, false) + if err != nil { + return fmt.Errorf("the merge of %s is handed but not in its delivery record: %w", group, err) + } + var members []string + if m.members != nil { + members = m.members(group) + } + return m.records.SetGroup(r, recordGroupOf(v, members)) +} + +// switchGroup is one group in flight as the switch finds it: its milestone, its heads and its members. +type switchGroup struct { + milestone string + heads map[string]string + members []string +} + +// inFlight are the groups in flight, under the lock: every milestone with a delivery not final, its live members' heads +// as `@` to the full commit. +func (h *Holder) inFlight() []switchGroup { + by := map[string]*switchGroup{} + for _, d := range h.deliveries { + if d.State.Final() && !(d.State == Merged && !d.InSource) || d.Number == 0 || d.NewerHead != "" || d.ClosedUnmerged { + continue + } + name := milestoneOf(d) + g := by[name] + if g == nil { + g = &switchGroup{milestone: name, heads: map[string]string{}} + by[name] = g + } + g.heads[mdc.ApprovedKey(d.Repository, d.Number)] = d.Commit + g.members = append(g.members, d.ID) + } + out := make([]switchGroup, 0, len(by)) + for _, g := range by { + sort.Strings(g.members) + out = append(out, *g) + } + sort.Slice(out, func(i, j int) bool { return out[i].milestone < out[j].milestone }) + return out +} + +// Switch carries today's state into the records, once and safe to run again: each group in flight becomes a record at +// its heads, or at the heads its merge was approved at when `merges` holds one, with that merge as its group state. +// It answers how many records it opened and how many it found there. +func (h *Holder) Switch() (opened, found int, err error) { + if h.Records == nil { + return 0, 0, nil + } + h.mu.Lock() + groups := h.inFlight() + h.mu.Unlock() + for _, g := range groups { + var merge *Merge + if h.Merges != nil { + if merge, err = h.Merges.Get(g.milestone); err != nil { + return opened, found, fmt.Errorf("the merge of %s cannot be read for the switch: %w", g.milestone, err) + } + } + heads := g.heads + if merge != nil && len(merge.Approved) > 0 { + heads = merge.Approved + } + r, made, err := h.Records.Open(g.milestone, heads, false) + if err != nil { + return opened, found, fmt.Errorf("the switch cannot open the record of %s: %w", g.milestone, err) + } + if made { + opened++ + } else { + found++ + } + if merge != nil { + if err := h.Records.SetGroup(r, recordGroupOf(*merge, g.members)); err != nil { + return opened, found, fmt.Errorf("the switch cannot carry the merge of %s: %w", g.milestone, err) + } + } + } + return opened, found, nil +} + +// groupMembers are the members' ids of the milestone a merge is handed under, under the lock. +func (h *Holder) groupMembers(name string) []string { + h.mu.Lock() + defer h.mu.Unlock() + var out []string + for _, d := range h.deliveries { + if milestoneOf(d) == name && d.NewerHead == "" && !d.ClosedUnmerged { + out = append(out, d.ID) + } + } + sort.Strings(out) + return out +}