diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/deliveries.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/deliveries.go new file mode 100644 index 000000000..24780ce18 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/deliveries.go @@ -0,0 +1,458 @@ +package main + +// The gate keeps a group's members from their announcement (design 47 G §716; A-279, A-236): its states `deliveries` and +// `groups`, written from the seat git's pull-updated, pull-merged and pull-closed, deleted when a record carries a member (the +// take-in, takenin.go) or its pull request closes unmerged, and put right at the gate's start from the seat git's open-pulls and +// pull. A delivery's key is its id `@` escaped as a key name; a group's key is its head branch +// escaped as a key name. Only the gate writes either state. + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + + gitseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// The two states, by their names in the gate's manifest. +const ( + deliveriesState = "deliveries" + groupsState = "groups" +) + +// The states of a delivery the gate writes; any other state an old table kept counts open (§716). +const ( + deliveryOpen = "open" + deliveryMerged = "merged" +) + +// Delivery is one pull request as the state `deliveries` keeps it. +type Delivery struct { + ID string `json:"id"` + Repository string `json:"repository"` + Number int `json:"number"` + Commit string `json:"commit"` + // Base is the pull request's base branch as the seat git says it (A-279); Branch its head branch, which names its group. + Base string `json:"base,omitempty"` + Branch string `json:"branch,omitempty"` + Group string `json:"group,omitempty"` + State string `json:"state"` +} + +// openOrMerged is a delivery's state as the gate reads it: merged, or else open. +func (d Delivery) openOrMerged() string { + if d.State == deliveryMerged { + return deliveryMerged + } + return deliveryOpen +} + +// keptGroup is one key of the state `groups`: the group, its head branch, and its members' delivery ids. +type keptGroup struct { + Group string `json:"group"` + Members []string `json:"members"` +} + +// DeliveryStore is the gate's two states over the runtime. +type DeliveryStore interface { + Keys(state string) ([]string, error) + Get(state, key string) ([]byte, error) + Put(state, key string, value any) error + Delete(state, key string) error +} + +// ForgeReads is what the gate reads of the seat git: open-pulls and pull. +type ForgeReads interface { + OpenPulls() (gitseat.OpenPullsAnswer, error) + Pull(repository string, number int64) (gitseat.PullAnswer, error) +} + +// kvKept is the store over the runtime. +type kvKept struct{} + +func (kvKept) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } + +func (kvKept) Get(state, key string) ([]byte, error) { + e, err := stdio.State(state).Get(key) + if err != nil || e == nil { + return nil, err + } + return e.Value, nil +} + +func (kvKept) Put(state, key string, value any) error { + _, err := stdio.State(state).Put(key, value) + return err +} + +func (kvKept) Delete(state, key string) error { return stdio.State(state).Delete(key) } + +// seatGit reads the seat git through the tool runner. +type seatGit struct{ ask asking } + +func (s seatGit) call(verb string, in any, out any) error { + raw, err := s.ask("seat:"+gitseat.Seat+"."+verb, in) + if err != nil { + return err + } + answer, said, ok := answerOf(raw) + if !ok { + return fmt.Errorf("%s refused: %s", verb, lastLine(said)) + } + if err := json.Unmarshal(answer, out); err != nil { + return fmt.Errorf("the answer of %s is not readable: %w", verb, err) + } + return nil +} + +func (s seatGit) OpenPulls() (gitseat.OpenPullsAnswer, error) { + var a gitseat.OpenPullsAnswer + return a, s.call(gitseat.VerbOpenPulls, gitseat.OpenPullsInput{}, &a) +} + +func (s seatGit) Pull(repository string, number int64) (gitseat.PullAnswer, error) { + var a gitseat.PullAnswer + return a, s.call(gitseat.VerbPull, gitseat.PullInput{Repository: repository, Number: number}, &a) +} + +func deliveryKey(id string) string { return keys.EscapeName(id) } + +func samePull(d Delivery, repository string, number int) bool { + return strings.EqualFold(d.Repository, repository) && d.Number == number +} + +// deliveries are every delivery the state keeps by its key; a key that is no delivery (a mark an earlier owner kept) is left. +func (h *Holder) deliveries() (map[string]Delivery, error) { + if h.Kept == nil { + return nil, errors.New("this gate keeps no state deliveries") + } + names, err := h.Kept.Keys(deliveriesState) + if err != nil { + return nil, fmt.Errorf("the state deliveries cannot be read: %w", err) + } + out := map[string]Delivery{} + for _, k := range names { + raw, err := h.Kept.Get(deliveriesState, k) + if err != nil { + return nil, fmt.Errorf("the delivery %s cannot be read: %w", k, err) + } + var d Delivery + if raw == nil || json.Unmarshal(raw, &d) != nil || d.ID == "" || d.Number == 0 { + continue + } + out[k] = d + } + return out, nil +} + +// keptGroups are every group the state keeps by its key. +func (h *Holder) keptGroups() (map[string]keptGroup, error) { + if h.Kept == nil { + return nil, errors.New("this gate keeps no state groups") + } + names, err := h.Kept.Keys(groupsState) + if err != nil { + return nil, fmt.Errorf("the state groups cannot be read: %w", err) + } + out := map[string]keptGroup{} + for _, k := range names { + raw, err := h.Kept.Get(groupsState, k) + if err != nil { + return nil, fmt.Errorf("the group %s cannot be read: %w", k, err) + } + var g keptGroup + if raw == nil || json.Unmarshal(raw, &g) != nil || g.Group == "" { + continue + } + out[k] = g + } + return out, nil +} + +// PullUpdated takes one pull-updated: the delivery is written under its id with its base and head branch, and a newer head of the +// same pull request takes the older's place, in its group too. +func (h *Holder) PullUpdated(body []byte) error { + var e gitseat.PullUpdatedEvent + if err := json.Unmarshal(body, &e); err != nil || e.Repository == "" || e.Number <= 0 || e.Head == "" { + h.Refuse(fmt.Errorf("a pull-updated of the seat git is not readable: %v", err)) + return nil + } + h.dmu.Lock() + defer h.dmu.Unlock() + return h.writeDelivery(Delivery{ID: IDOf(e.Repository, e.Head), Repository: e.Repository, Number: int(e.Number), Commit: e.Head, + Base: e.Base, Branch: e.Branch, State: deliveryOpen}) +} + +// writeDelivery writes a delivery in place of any other head of its pull request and puts the groups right; under dmu. +func (h *Holder) writeDelivery(d Delivery) error { + all, err := h.deliveries() + if err != nil { + return err + } + branches := map[string]bool{d.Branch: true} + key := deliveryKey(d.ID) + for k, o := range all { + if k == key || !samePull(o, d.Repository, d.Number) { + continue + } + if err := h.Kept.Delete(deliveriesState, k); err != nil { + return fmt.Errorf("the older delivery %s cannot be deleted: %w", o.ID, err) + } + delete(all, k) + branches[o.Branch] = true + } + if o, ok := all[key]; ok { + d.State = o.openOrMerged() + d.Group = o.Group + } + if err := h.Kept.Put(deliveriesState, key, d); err != nil { + return fmt.Errorf("the delivery %s cannot be written: %w", d.ID, err) + } + return h.refreshGroups(branches) +} + +// DeliveryMerged takes one pull-merged: the delivery of that pull request is merged. A pull request the gate keeps no delivery of is +// left: nothing would delete it. +func (h *Holder) DeliveryMerged(body []byte) error { + var e gitseat.PullMergedEvent + if err := json.Unmarshal(body, &e); err != nil || e.Repository == "" || e.Number <= 0 { + h.Refuse(fmt.Errorf("a pull-merged of the seat git is not readable: %v", err)) + return nil + } + h.dmu.Lock() + defer h.dmu.Unlock() + return h.setMerged(e.Repository, int(e.Number), e.Base) +} + +// setMerged sets the state of a pull request's delivery to merged; under dmu. +func (h *Holder) setMerged(repository string, number int, base string) error { + all, err := h.deliveries() + if err != nil { + return err + } + for k, d := range all { + if !samePull(d, repository, number) || d.State == deliveryMerged { + continue + } + d.State = deliveryMerged + if base != "" { + d.Base = base + } + if err := h.Kept.Put(deliveriesState, k, d); err != nil { + return fmt.Errorf("the delivery %s cannot be set merged: %w", d.ID, err) + } + } + return nil +} + +// DeliveryClosed takes one pull-closed: the delivery of a pull request not merged is deleted, and its group put right. +func (h *Holder) DeliveryClosed(body []byte) error { + var e gitseat.PullClosedEvent + if err := json.Unmarshal(body, &e); err != nil || e.Repository == "" || e.Number <= 0 { + h.Refuse(fmt.Errorf("a pull-closed of the seat git is not readable: %v", err)) + return nil + } + h.dmu.Lock() + defer h.dmu.Unlock() + return h.dropUnmerged(e.Repository, int(e.Number)) +} + +// dropUnmerged deletes the deliveries of a pull request that are not merged and puts their groups right; under dmu. +func (h *Holder) dropUnmerged(repository string, number int) error { + all, err := h.deliveries() + if err != nil { + return err + } + branches := map[string]bool{} + for k, d := range all { + if !samePull(d, repository, number) || d.State == deliveryMerged { + continue + } + if err := h.Kept.Delete(deliveriesState, k); err != nil { + return fmt.Errorf("the delivery %s cannot be deleted: %w", d.ID, err) + } + branches[d.Branch] = true + } + return h.refreshGroups(branches) +} + +// recordGroupsHeld are the group ids the delivery records carry, ended or not. +func (h *Holder) recordGroupsHeld() (map[string]bool, error) { + out := map[string]bool{} + if h.Records == nil { + return out, nil + } + kept, err := h.Records.Kept() + if err != nil { + return nil, fmt.Errorf("the delivery records cannot be read for the groups: %w", err) + } + for _, r := range kept { + if r.Group != nil && r.Group.Group != "" && len(membersOf(r)) > 0 { + out[r.Group.Group] = true + } + } + return out, nil +} + +// refreshGroups writes, for each head branch, the group key of its deliveries and their `group`: pull requests with one head branch +// form one group; a group of one with no member in a record is no group, its key deleted and its member's `group` cleared; a group +// key once no member remains outside a record is deleted (§716). Under dmu. +func (h *Holder) refreshGroups(branches map[string]bool) error { + inRecords, err := h.recordGroupsHeld() + if err != nil { + return err + } + all, err := h.deliveries() + if err != nil { + return err + } + for branch := range branches { + if branch == "" { + continue + } + var members []string + keysOf := map[string]string{} + for k, d := range all { + if d.Branch == branch { + members = append(members, d.ID) + keysOf[d.ID] = k + } + } + sort.Strings(members) + gkey := keys.EscapeName(branch) + isGroup := len(members) > 1 || len(members) == 1 && inRecords[branch] + if !isGroup { + if err := h.Kept.Delete(groupsState, gkey); err != nil { + return fmt.Errorf("the group %s cannot be deleted: %w", branch, err) + } + } else if err := h.Kept.Put(groupsState, gkey, keptGroup{Group: branch, Members: members}); err != nil { + return fmt.Errorf("the group %s cannot be written: %w", branch, err) + } + for _, id := range members { + k := keysOf[id] + d := all[k] + want := "" + if isGroup { + want = branch + } + if d.Group != want { + d.Group = want + all[k] = d + if err := h.Kept.Put(deliveriesState, k, d); err != nil { + return fmt.Errorf("the delivery %s cannot be put in its group: %w", id, err) + } + } + } + } + return nil +} + +// ReadForge puts right what a missed pull event left, once at the gate's start (§716, after §613's copy): each open pull request +// the seat git's open-pulls says that the gate lacks, or keeps at another head, is written; each delivery kept open that the answer +// does not name is read by pull: merged sets merged, closed deletes it. A read that fails is said in the journal and made again at +// the next start. +func (h *Holder) ReadForge() error { + if h.Forge == nil || h.Kept == nil { + return nil + } + answer, err := h.Forge.OpenPulls() + if err != nil { + return fmt.Errorf("open-pulls: %w", err) + } + if answer.Error != "" { + return fmt.Errorf("open-pulls: %s", answer.Error) + } + h.dmu.Lock() + defer h.dmu.Unlock() + listed := map[string]bool{} + var failed []error + for _, o := range answer.Pulls { + p := o.Pull + if p.Repository == "" || p.Number <= 0 || p.Head == "" { + continue + } + listed[strings.ToLower(p.Repository)+"#"+fmt.Sprint(p.Number)] = true + all, err := h.deliveries() + if err != nil { + return err + } + if d, ok := all[deliveryKey(IDOf(p.Repository, p.Head))]; ok && d.Base == p.Base && d.Branch == p.Branch { + continue + } + if err := h.writeDelivery(Delivery{ID: IDOf(p.Repository, p.Head), Repository: p.Repository, Number: int(p.Number), Commit: p.Head, + Base: p.Base, Branch: p.Branch, State: deliveryOpen}); err != nil { + failed = append(failed, err) + } + } + all, err := h.deliveries() + if err != nil { + return err + } + for _, d := range all { + if d.openOrMerged() != deliveryOpen || listed[strings.ToLower(d.Repository)+"#"+fmt.Sprint(d.Number)] { + continue + } + got, err := h.Forge.Pull(d.Repository, int64(d.Number)) + if err == nil && got.Error != "" { + err = errors.New(got.Error) + } + if err == nil && got.Pull == nil { + err = errors.New("the answer names no pull request") + } + if err != nil { + h.Logf("[mesh-delivery] the pull request %s#%d cannot be read at the start: %v", d.Repository, d.Number, err) + failed = append(failed, err) + continue + } + switch got.Pull.State { + case gitseat.PullStateMerged: + err = h.setMerged(d.Repository, d.Number, got.Pull.Base) + case gitseat.PullStateClosed: + err = h.dropUnmerged(d.Repository, d.Number) + } + if err != nil { + failed = append(failed, err) + } + } + return errors.Join(failed...) +} + +// baseOf is the base of a pull request's delivery, and whether the gate keeps one. +func baseOf(all map[string]Delivery, repository string, number int) (string, bool) { + for _, d := range all { + if samePull(d, repository, number) { + return d.Base, true + } + } + return "", false +} + +// carried puts the deliveries of the pull requests a record carries away, once the record holds them with their base: each delivery +// is deleted and the groups put right. A deletion is made in the old section too (§613): the state's keys are the one table. +func (h *Holder) carried(pulls []RecordPull) error { + if h.Kept == nil { + return nil + } + h.dmu.Lock() + defer h.dmu.Unlock() + all, err := h.deliveries() + if err != nil { + return err + } + branches := map[string]bool{} + for k, d := range all { + for _, p := range pulls { + if samePull(d, p.Repository, p.Number) { + if err := h.Kept.Delete(deliveriesState, k); err != nil { + return fmt.Errorf("the delivery %s cannot be deleted: %w", d.ID, err) + } + branches[d.Branch] = true + } + } + } + return h.refreshGroups(branches) +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/deliveries_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/deliveries_test.go new file mode 100644 index 000000000..7710e6fe1 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/deliveries_test.go @@ -0,0 +1,253 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "testing" + "time" + + gitseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git" +) + +// memKept is the gate's states deliveries and groups in memory. +type memKept struct{ m map[string]map[string][]byte } + +func (k *memKept) Keys(state string) ([]string, error) { + var out []string + for key := range k.m[state] { + out = append(out, key) + } + sort.Strings(out) + return out, nil +} +func (k *memKept) Get(state, key string) ([]byte, error) { return k.m[state][key], nil } +func (k *memKept) Put(state, key string, v any) error { + if k.m == nil { + k.m = map[string]map[string][]byte{} + } + if k.m[state] == nil { + k.m[state] = map[string][]byte{} + } + raw, _ := json.Marshal(v) + k.m[state][key] = raw + return nil +} +func (k *memKept) Delete(state, key string) error { delete(k.m[state], key); return nil } + +// fakeForge is the seat git's reads: the open pull requests and each pull request by its number. +type fakeForge struct { + open []gitseat.OpenPull + pulls map[string]gitseat.Pull + fail bool +} + +func (f *fakeForge) OpenPulls() (gitseat.OpenPullsAnswer, error) { + if f.fail { + return gitseat.OpenPullsAnswer{}, errors.New("the forge is away") + } + return gitseat.OpenPullsAnswer{Pulls: f.open}, nil +} +func (f *fakeForge) Pull(repository string, number int64) (gitseat.PullAnswer, error) { + p, ok := f.pulls[fmt.Sprintf("%s#%d", repository, number)] + if !ok { + return gitseat.PullAnswer{Error: "no such pull request"}, nil + } + return gitseat.PullAnswer{Pull: &p}, nil +} + +func gateWithDeliveries(t *testing.T) (*Holder, *memKept) { + t.Helper() + k := &memKept{} + h := &Holder{Now: func() time.Time { return time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC) }, Kept: k, Logf: func(string, ...any) {}} + h.Records = &Records{Ledger: newLedgerFake(t), Now: h.Now} + return h, k +} + +func updated(repository string, number int, head, base, branch string) []byte { + raw, _ := json.Marshal(gitseat.PullUpdatedEvent{Repository: repository, Number: int64(number), Head: head, Base: base, Branch: branch}) + return raw +} + +func groupNamed(t *testing.T, h *Holder, id string) []string { + t.Helper() + a := h.Groups(false, id) + if a.Error != "" || len(a.Groups) != 1 { + t.Fatalf("groups %q: %+v", id, a) + } + var out []string + for _, m := range a.Groups[0].Members { + out = append(out, fmt.Sprintf("%s#%d@%s/%s/%s/%v", m.Repository, m.Number, m.Base, m.State, m.Head[:1], m.Open)) + } + return out +} + +// §716, A-279: a group read before any take-in names its members with their base, from the seat git's pull-updated. +func TestAGroupReadBeforeAnyTakeInNamesItsMembersWithTheirBase(t *testing.T) { + h, k := gateWithDeliveries(t) + a, b := strings.Repeat("a", 40), strings.Repeat("b", 40) + for _, body := range [][]byte{updated("novox/app", 3, a, "main", "feat"), updated("novox/lib", 4, b, "beta", "feat")} { + if err := h.PullUpdated(body); err != nil { + t.Fatal(err) + } + } + got := groupNamed(t, h, "feat") + want := []string{"novox/app#3@main/open/a/true", "novox/lib#4@beta/open/b/true"} + if strings.Join(got, " ") != strings.Join(want, " ") { + t.Fatalf("members %v, wanted %v", got, want) + } + if len(k.m[groupsState]) != 1 || len(k.m[deliveriesState]) != 2 { + t.Fatalf("states %v", k.m) + } + // The delivery's key is its id escaped as a key name, with its group set (§42). + var d Delivery + if err := json.Unmarshal(k.m[deliveriesState]["novox/app=40"+strings.Repeat("a", 12)], &d); err != nil || d.Group != "feat" || d.Base != "main" || d.State != "open" || d.Number != 3 { + t.Fatalf("the delivery %+v %v (keys %v)", d, err, k.m[deliveriesState]) + } +} + +// §716: a newer head of the same pull request takes the older's place in its group, and the older key is deleted. +func TestANewerHeadReplacesTheOlder(t *testing.T) { + h, k := gateWithDeliveries(t) + a, b, c := strings.Repeat("a", 40), strings.Repeat("b", 40), strings.Repeat("c", 40) + for _, body := range [][]byte{updated("novox/app", 3, a, "main", "feat"), updated("novox/lib", 4, b, "main", "feat"), updated("novox/app", 3, c, "main", "feat")} { + if err := h.PullUpdated(body); err != nil { + t.Fatal(err) + } + } + got := groupNamed(t, h, "feat") + if strings.Join(got, " ") != "novox/app#3@main/open/c/true novox/lib#4@main/open/b/true" || len(k.m[deliveriesState]) != 2 { + t.Fatalf("members %v, deliveries %d", got, len(k.m[deliveriesState])) + } +} + +// §716: a closed unmerged pull request leaves; a group of one with no member in a record is no group, its key deleted and the +// member's group cleared. +func TestAClosedUnmergedPullRequestLeavesItsGroup(t *testing.T) { + h, k := gateWithDeliveries(t) + a, b := strings.Repeat("a", 40), strings.Repeat("b", 40) + _ = h.PullUpdated(updated("novox/app", 3, a, "main", "feat")) + _ = h.PullUpdated(updated("novox/lib", 4, b, "main", "feat")) + closed, _ := json.Marshal(gitseat.PullClosedEvent{Repository: "novox/lib", Number: 4, Head: b, Base: "main", Branch: "feat"}) + if err := h.DeliveryClosed(closed); err != nil { + t.Fatal(err) + } + if len(k.m[groupsState]) != 0 || len(k.m[deliveriesState]) != 1 { + t.Fatalf("states %v", k.m) + } + for _, raw := range k.m[deliveriesState] { + var d Delivery + _ = json.Unmarshal(raw, &d) + if d.Group != "" || d.Number != 3 { + t.Fatalf("the member of a group of one keeps its group: %+v", d) + } + } +} + +// §716, A-279: the take-in copies the delivery's base onto the record's member and deletes the delivery; the member not taken in stays +// in the group, which names both. +func TestTheTakeInCopiesTheBaseAndKeepsTheRestOfTheGroup(t *testing.T) { + h, k := gateWithDeliveries(t) + merge := strings.Repeat("m", 40) + a, b := strings.Repeat("a", 40), strings.Repeat("b", 40) + _ = h.PullUpdated(updated("novox/app", 3, a, "main", "feat")) + _ = h.PullUpdated(updated("novox/lib", 4, b, "beta", "feat")) + merged, _ := json.Marshal(gitseat.PullMergedEvent{Repository: "novox/app", Number: 3, Head: a, Base: "main", Branch: "feat", MergeCommit: merge}) + if err := h.DeliveryMerged(merged); err != nil { + t.Fatal(err) + } + if got := groupNamed(t, h, "feat"); got[0] != "novox/app#3@main/merged/a/false" { + t.Fatalf("a merged member before its take-in: %v", got) + } + body := json.RawMessage(`{"v":1,"kind":"taken","key":"g.feat","what":"group","group":"feat","branch":"main","source":7,` + + `"merges":[{"repository":"novox/app","number":3,"merge":"` + merge + `"}],` + + `"entries":[{"module":"app","commit":"` + merge + `","fingerprint":"f","repository":"novox/app","pull":3}],"since":"2026-11-12T09:00:00Z"}`) + for i := 0; i < 2; i++ { // heard again: the base the record holds stays + if err := h.TakenIn(body); err != nil { + t.Fatal(err) + } + } + kept, _ := h.Records.Kept() + if len(kept) != 1 || kept[0].Group == nil || len(kept[0].Group.Pulls) != 1 || kept[0].Group.Pulls[0].Base != "main" { + t.Fatalf("the record's member lacks its base: %+v", kept) + } + if len(k.m[deliveriesState]) != 1 || len(k.m[groupsState]) != 1 { + t.Fatalf("states after the take-in: %v", k.m) + } + got := groupNamed(t, h, "feat") + if strings.Join(got, " ") != "novox/app#3@main/merged/m/false novox/lib#4@beta/open/b/true" && strings.Join(got, " ") != "novox/app#3@main/merged/a/false novox/lib#4@beta/open/b/true" { + t.Fatalf("members %v", got) + } +} + +// §716: a missed event is put right at the start: an open pull request the gate lacks is written, one kept at another head is +// replaced, and a kept delivery the answer does not name is read: merged is set, closed is deleted; a read that fails is +// said and left. +func TestAMissedEventIsPutRightAtStart(t *testing.T) { + h, k := gateWithDeliveries(t) + a, a2, b, c, d := strings.Repeat("a", 40), strings.Repeat("1", 40), strings.Repeat("b", 40), strings.Repeat("c", 40), strings.Repeat("d", 40) + _ = h.PullUpdated(updated("novox/app", 3, a, "main", "feat")) + _ = h.PullUpdated(updated("novox/gone", 5, c, "main", "feat")) + _ = h.PullUpdated(updated("novox/done", 6, d, "main", "feat")) + _ = h.PullUpdated(updated("novox/away", 7, d, "main", "other")) + h.Forge = &fakeForge{ + open: []gitseat.OpenPull{ + {Pull: gitseat.Pull{Repository: "novox/app", Number: 3, Head: a2, Base: "main", Branch: "feat"}}, + {Pull: gitseat.Pull{Repository: "novox/lib", Number: 4, Head: b, Base: "beta", Branch: "feat"}}}, + pulls: map[string]gitseat.Pull{ + "novox/gone#5": {Repository: "novox/gone", Number: 5, State: gitseat.PullStateClosed, Base: "main"}, + "novox/done#6": {Repository: "novox/done", Number: 6, State: gitseat.PullStateMerged, Base: "main"}}, + } + err := h.ReadForge() + if err == nil || !strings.Contains(err.Error(), "no such pull request") { + t.Fatalf("a pull request the forge does not know is said: %v", err) + } + all, _ := h.deliveries() + got := map[string]string{} + for _, d := range all { + got[fmt.Sprintf("%s#%d", d.Repository, d.Number)] = d.Commit[:1] + "/" + d.State + "/" + d.Base + } + want := map[string]string{"novox/app#3": "1/open/main", "novox/lib#4": "b/open/beta", "novox/done#6": "d/merged/main", "novox/away#7": "d/open/main"} + if fmt.Sprint(got) != fmt.Sprint(want) { + t.Fatalf("deliveries %v, wanted %v", got, want) + } + if len(k.m[groupsState]) != 1 { + t.Fatalf("groups %v", k.m[groupsState]) + } + h.Forge = &fakeForge{fail: true} + if err := h.ReadForge(); err == nil { + t.Fatal("a forge that cannot be read is said") + } +} + +// A-281: the switch runs once; a start, then a pull-updated, then a second start opens no record. +func TestTheSwitchRunsOnceSoDeliveriesNeverBecomeRecordsAtAStart(t *testing.T) { + h, k := gateWithDeliveries(t) + h.Old = kvKeptOld{k} + if _, _, err := h.Switch(); err != nil { + t.Fatal(err) + } + if err := h.PullUpdated(updated("novox/app", 3, strings.Repeat("a", 40), "main", "feat")); err != nil { + t.Fatal(err) + } + opened, found, err := h.Switch() + if err != nil || opened != 0 || found != 0 { + t.Fatalf("the second start switched: %d %d %v", opened, found, err) + } + if kept, _ := h.Records.Kept(); len(kept) != 0 { + t.Fatalf("a delivery became a record: %+v", kept) + } + if done, err := h.Records.SwitchDone(); err != nil || !done { + t.Fatalf("switch.done %v %v", done, err) + } +} + +type kvKeptOld struct{ k *memKept } + +func (o kvKeptOld) Keys(state string) ([]string, error) { return o.k.Keys(state) } +func (o kvKeptOld) Get(state, key string) ([]byte, uint64, error) { + v, err := o.k.Get(state, key) + return v, 1, err +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/findings.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/findings.go index a38acdc97..15844c158 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/findings.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/findings.go @@ -28,8 +28,9 @@ const ( // pullMergedEvent and pullClosedEvent are the seat git's events that end a pull request, as the gate's manifest consumes them. var ( - pullMergedEvent = declared(gitseat.EventPullMerged) - pullClosedEvent = declared(gitseat.EventPullClosed) + pullUpdatedEvent = declared(gitseat.EventPullUpdated) + pullMergedEvent = declared(gitseat.EventPullMerged) + pullClosedEvent = declared(gitseat.EventPullClosed) ) // FindingWriter is what the gate holds of the one held writer (views.NewStdioHeldWriter over its state `findings`). diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/gate.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/gate.go index 2942e6dca..94de1c9b7 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/gate.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/gate.go @@ -22,6 +22,10 @@ type Holder struct { Logf func(format string, a ...any) // Merges and Old are the old sections the switch reads (switch.go); Source is the gate's source (merges.go). Merges MergeStore + // Kept is the gate's states deliveries and groups, and Forge the seat git's reads (deliveries.go, §716). + Kept DeliveryStore + Forge ForgeReads + dmu sync.Mutex Old OldState Source SourceStore // Owed says what the gate owes the forge as the seat's event owed (owed.go). 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 651af2836..465482dea 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -32,7 +32,7 @@ func (l *listening) get() string { l.mu.Lock(); defer l.mu.Unlock(); return l.n // The events this owner follows: the forge bridge's word on an act owed, and the step events the delivery records are built // from (consumer.go). -var followed = append(append([]string{owedEndedEvent}, stepPatterns()...), pullMergedEvent, pullClosedEvent) +var followed = append(append([]string{owedEndedEvent}, stepPatterns()...), pullUpdatedEvent, pullMergedEvent, pullClosedEvent) func main() { settingsFile := os.Getenv("MESH_DELIVERY_GATE_SETTINGS") @@ -66,6 +66,8 @@ func main() { h.Records = &Records{Ledger: sdkLedger{}, Now: h.Now, Retention: h.recordRetention, Say: stdio.SeatPublish, Logf: logf} h.Records.Changed = h.recordChanged h.Old = kvOld{} + h.Kept = kvKept{} + h.Forge = seatGit{ask: stdio.Ask} h.Consumer = &StepConsumer{Records: h.Records, Families: stepFamilies, Now: h.Now, Logf: logf, Taken: h.TakenIn} // The gate's one held writer of its own findings (design 47 G A-39, A-37, A-40): made at once, started once the runtime serves. fw, err := NewFindingWriter() @@ -114,6 +116,10 @@ func run(h *Holder, l *listening) { } l.set("listening") logf("[mesh-delivery] listening: %s", strings.Join(followed, ", ")) + // A missed pull event is put right from the forge's own record, once at each start (§716). + if err := h.ReadForge(); err != nil { + logf("[mesh-delivery] the open pull requests were not read from the forge, they are read again at the next start: %v", err) + } } // switchToRecords carries the groups in flight into their delivery records (switch.go), retrying until it can: no event is @@ -160,11 +166,14 @@ func take(h *Holder, e stdio.Envelope) error { if e.Key == owedEndedEvent { return h.OwedEnded(e.Body) } + if e.Key == pullUpdatedEvent { + return h.PullUpdated(e.Body) + } if e.Key == pullMergedEvent { - return h.pullEnded(e.Body, "merged") + return errors.Join(h.DeliveryMerged(e.Body), h.pullEnded(e.Body, "merged")) } if e.Key == pullClosedEvent { - return h.pullEnded(e.Body, "closed") + return errors.Join(h.DeliveryClosed(e.Body), h.pullEnded(e.Body, "closed")) } return nil } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/manifest_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/manifest_test.go index 551122b85..a679e1fa7 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/manifest_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/manifest_test.go @@ -141,7 +141,7 @@ func TestItClaimsTheDeliverySeatAndFollowsWhatItHandles(t *testing.T) { t.Fatalf("consumes %v, follows %v", m.Consumes, followed) } for _, c := range m.Consumes { - for _, gone := range []string{".forge.", "teller", "view", "forge-pull", "forge-status", "pull-updated"} { + for _, gone := range []string{".forge.", "teller", "view", "forge-pull", "forge-status"} { if strings.Contains(c, gone) { t.Errorf("the manifest still consumes %s, which G's gate does not hear (§431, §592, §593)", c) } @@ -149,7 +149,7 @@ func TestItClaimsTheDeliverySeatAndFollowsWhatItHandles(t *testing.T) { } // A-30: the events its held writer says are declared in the manifest, and the two events that clear its finding are consumed. if !strings.Contains(raw, `"finding-written.*"`) || !strings.Contains(raw, `"finding-cleared.*"`) || !strings.Contains(raw, `"findings-stated.*"`) || - pullMergedEvent == "" || pullClosedEvent == "" { + pullMergedEvent == "" || pullClosedEvent == "" || pullUpdatedEvent == "" { t.Fatalf("the finding events are not declared, or the pull events not consumed: %q %q", pullMergedEvent, pullClosedEvent) } if !reflect.DeepEqual(m.Emits, []string{"transition"}) { // `group` and `merged` go with the merge states (§431) @@ -181,7 +181,7 @@ func TestItClaimsTheDeliverySeatAndFollowsWhatItHandles(t *testing.T) { } // It calls no controller verb and no forge verb: the forge's reads and writes are the bridge's (ADR 0437 §8; design 47 §592). // It uses operator-channel only to say its conditions (ADR 0437 §9, issue 1275): the teller asks no more (§431). - want := []string{"seat:mesh-hand-acts.hand-act", WantsEntryVerb, EngineCompositionVerb} + want := []string{"seat:git.open-pulls", "seat:git.pull", "seat:mesh-hand-acts.hand-act", WantsEntryVerb, EngineCompositionVerb} if !reflect.DeepEqual(m.Uses, []string{"operator-channel"}) { t.Fatalf("uses %v: operator-channel, for its conditions alone", m.Uses) } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go index 99edd8393..831f3f682 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/record.go @@ -39,6 +39,8 @@ const RecordForm = 1 const ( TypeRecord = "record" TypeHeld = "held" + // TypeSwitch holds the one key `switch.done` (A-281): the gate's fingerprint and the time of the switch to the records. + TypeSwitch = "switch" ) // The operations of a record, each one effect (ADR 0597 §2), and of the month's list. The manifest's `requires` declares @@ -57,6 +59,7 @@ const ( OpAct = "act" // set-entry into acts, by id: an act owed the forge with its outcome OpHold = "hold" // a month's list: set-entry into records, by instance OpDrop = "drop" // a month's list: remove from records, by instance + OpDone = "done" // the switch: set-fields, the gate's fingerprint and the time ) // RecordOperations are a record's operations, in the order the manifest's type allows them. @@ -65,6 +68,27 @@ var RecordOperations = []string{OpOpen, OpGroup, OpTrace, OpEntry, OpOutcome, Op // HeldOperations are the month's list's operations. var HeldOperations = []string{OpHold, OpDrop} +// SwitchOperations are the switch key's operations. +var SwitchOperations = []string{OpDone} + +// switchInstance is the instance of the key `switch.done`, written once the switch has carried today's state in (A-281). +const switchInstance = TypeSwitch + "/done" + +// SwitchDone says whether the key `switch.done` is there: the switch has run, and no later start runs it again (A-281). +func (rs *Records) SwitchDone() (bool, error) { + raw, _, err := rs.Ledger.Get(switchInstance) + if err != nil { + return false, fmt.Errorf("the key switch.done cannot be read: %w", err) + } + return len(raw) > 0 && string(raw) != "null", nil +} + +// MarkSwitched writes the key `switch.done`: the gate's fingerprint and the time (A-281). +func (rs *Records) MarkSwitched(fingerprint string) error { + _, err := rs.Ledger.Apply(switchInstance, OpDone, map[string]any{"fingerprint": fingerprint, "at": rs.now().UTC().Format(time.RFC3339)}) + return err +} + // DefaultRecordRetention is how long a record is kept when the setting record-retention gives none (design 47: 90 days // covers every trace the operator has asked of a past delivery so far). const DefaultRecordRetention = 90 * 24 * time.Hour 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 86e0a44a9..0cb23b623 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 @@ -386,7 +386,7 @@ func TestTheDeliveryRecordIsALedgerInstanceKeyedByMilestoneAndHeads(t *testing.T // W11.1: the manifest's ledger types allow exactly the operations the gate applies, and the ledger is capped at 768 MiB. func TestTheManifestsLedgerTypesAreTheRecordsOperations(t *testing.T) { l := newLedgerFake(t) - for name, want := range map[string][]string{TypeRecord: RecordOperations, TypeHeld: HeldOperations} { + for name, want := range map[string][]string{TypeRecord: RecordOperations, TypeHeld: HeldOperations, TypeSwitch: SwitchOperations} { var got []string for _, op := range l.types[name].Operations { got = append(got, op.Name) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go index a152ccedf..1cfaacb4f 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/switch.go @@ -10,6 +10,7 @@ import ( "encoding/json" "errors" "fmt" + "os" "sort" "strings" @@ -104,11 +105,22 @@ func (h *Holder) inFlight() ([]switchGroup, error) { // 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. +// It answers how many records it opened and how many it found there. It runs once (A-281): when it has carried today's state in +// it writes the key `switch.done`, and a start that finds the key runs no switch, so the deliveries the gate writes from the +// seat git's events (§716) never become records at a start. A revert's restore brings back the state without the key, so +// delivering G again switches again. func (h *Holder) Switch() (opened, found int, err error) { if h.Records == nil { return 0, 0, nil } + if done, err := h.Records.SwitchDone(); err != nil || done { + return 0, 0, err + } + defer func() { + if err == nil { + err = h.Records.MarkSwitched(os.Getenv("MESH_FINGERPRINT")) + } + }() groups, err := h.inFlight() if err != nil { return 0, 0, err @@ -142,7 +154,8 @@ func (h *Holder) Switch() (opened, found int, err error) { if err := h.carryOwed(); err != nil { return opened, found, err } - return opened, found, h.carryApprovals() + err = h.carryApprovals() + return opened, found, err } // oldMerge is a group's merge as the old state merges holds it, with its key and revision; nil when it holds none. diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go index de51c0257..d69ae8dc1 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/takenin.go @@ -58,13 +58,41 @@ func (h *Holder) TakenIn(body json.RawMessage) error { g.Members = append(g.Members, mdc.ApprovedKey(e.Repository, int(e.Pull))) } sort.Strings(g.Members) + // §716: each member's base is copied from its delivery onto the record's member; a record that holds the member already + // (a take-in heard again, its delivery deleted) keeps the base it holds. + var kept map[string]Delivery + if h.Kept != nil { + h.dmu.Lock() + kept, err = h.deliveries() + h.dmu.Unlock() + if err != nil { + return err + } + } for _, m := range t.Merges { - g.Pulls = append(g.Pulls, RecordPull{Repository: m.Repository, Number: int(m.Number), Head: heads[mdc.ApprovedKey(m.Repository, int(m.Number))], - State: string(mdc.MergeMerged), Commit: m.Merge}) + p := RecordPull{Repository: m.Repository, Number: int(m.Number), Head: heads[mdc.ApprovedKey(m.Repository, int(m.Number))], + State: string(mdc.MergeMerged), Commit: m.Merge} + if base, ok := baseOf(kept, m.Repository, int(m.Number)); ok { + p.Base = base + } else if r.Group != nil { + for _, o := range r.Group.Pulls { + if samePull(Delivery{Repository: o.Repository, Number: o.Number}, m.Repository, int(m.Number)) { + p.Base = o.Base + } + } + } + if p.Base == "" && h.Kept != nil && h.Logf != nil { + h.Logf("[mesh-delivery] the take-in of %s names no delivery of %s#%d: its base is not known, the take-in's branch is %s", t.Group, m.Repository, m.Number, t.Branch) + } + g.Pulls = append(g.Pulls, p) } if err := h.Records.SetGroup(r, g); err != nil { return err } + // The record that carries the members now holds their bases: their deliveries, and a group no member of which remains outside a record, go. + if err := h.carried(g.Pulls); err != nil { + return err + } } return h.moveSource(t) } 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 b24561694..52e21c370 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/verbs.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/verbs.go @@ -185,10 +185,74 @@ func (h *Holder) Groups(all bool, only string) seat.GroupsAnswer { } out.Groups = append(out.Groups, g) } + if err := h.addKeptMembers(&out, kept, all, only); err != nil { + return seat.GroupsAnswer{Error: err.Error()} + } sort.SliceStable(out.Groups, func(i, j int) bool { return out.Groups[i].ID < out.Groups[j].ID }) return out } +// memberOf is a pull request as one key: the repository in lower case and the number. +func memberOf(repository string, number int64) string { + return fmt.Sprintf("%s#%d", strings.ToLower(repository), number) +} + +// addKeptMembers answers a group from its records and its key in the state `groups` together, each member once, a record's first +// (§716, A-279): the members of a group the gate keeps and no record carries, with their delivery's base, head, state and open +// while not merged, so before any take-in the verb names every member and its base. A group only its key holds is open. +func (h *Holder) addKeptMembers(out *seat.GroupsAnswer, kept []*Record, all bool, only string) error { + if h.Kept == nil { + return nil + } + groups, err := h.keptGroups() + if err != nil { + return err + } + deliveries, err := h.deliveries() + if err != nil { + return err + } + carried := map[string]map[string]bool{} + for _, r := range kept { + id := groupOf(r) + if carried[id] == nil { + carried[id] = map[string]bool{} + } + for _, m := range membersOf(r) { + carried[id][memberOf(m.Repository, m.Number)] = true + } + } + for _, kg := range groups { + if only != "" && kg.Group != only { + continue + } + var add []seat.Member + for _, d := range deliveries { + if d.Branch != kg.Group || carried[kg.Group][memberOf(d.Repository, int64(d.Number))] { + continue + } + add = append(add, seat.Member{ID: d.ID, Repository: d.Repository, State: d.openOrMerged(), Number: int64(d.Number), Head: d.Commit, + Open: d.openOrMerged() != deliveryMerged, Base: d.Base}) + } + if len(add) == 0 { + continue + } + sort.Slice(add, func(i, j int) bool { return add[i].ID < add[j].ID }) + placed := false + for i := range out.Groups { + if out.Groups[i].ID == kg.Group { + out.Groups[i].Members = append(out.Groups[i].Members, add...) + placed = true + } + } + if !placed { + out.Groups = append(out.Groups, seat.Group{ID: kg.Group, State: stateOpen, Members: add, Acts: []seat.Act{}, + Why: "its pull requests are not taken in yet"}) + } + } + return nil +} + // recordedActs are the acts a record holds, each with its outcome when the forge bridge said one, ordered by id. func recordedActs(r *Record) []seat.Act { out := []seat.Act{} diff --git a/modules/mesh-delivery-gate/module.json b/modules/mesh-delivery-gate/module.json index 194772ff1..aad02e5cf 100644 --- a/modules/mesh-delivery-gate/module.json +++ b/modules/mesh-delivery-gate/module.json @@ -19,6 +19,7 @@ "consumes": [ "mesh-delivery-forge-bridge.owed.ended", "mesh-delivery-indexer.take-in.taken", + "git.pull-updated", "git.pull-merged", "git.pull-closed", "mesh-conditions.findings-asked" @@ -279,6 +280,27 @@ "key": "instance" } ] + }, + { + "name": "switch", + "schema": { + "fingerprint": { + "type": "string" + }, + "at": { + "type": "string" + } + }, + "operations": [ + { + "name": "done", + "effect": "set-fields", + "fields": [ + "fingerprint", + "at" + ] + } + ] } ] }, @@ -329,6 +351,8 @@ "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg": "0.1" }, "invokes": [ + "seat:git.open-pulls", + "seat:git.pull", "seat:mesh-hand-acts.hand-act", "seat:mesh-wants.entry", "seat:node-engine.composition"