diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go index bfae30d9..3fbb0070 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval.go @@ -1,6 +1,7 @@ package main import ( + "encoding/json" "fmt" "regexp" "sort" @@ -8,15 +9,18 @@ import ( "strings" "time" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asks" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" ) // A merge into main waits for the operator's Approve (novox/hq ADR 0398 §3, issue 948, ADR 0421). // -// This owner asks the operator on the router (the operator channel, as the knowledge base asks for a record) as soon as -// a pull request's checks pass: Approve or Decline, each option binding the act "merge these heads". It merges only after -// an Approve whose warrant holds for that ask — the same asker, the same ask's digest, an option it offered, in time, and -// the act that option bound. +// This owner writes the ask into its bucket asks (novox/hq ADR 0437 §6) as soon as a pull request's checks pass: +// Approve or Decline, each option binding the act "merge these heads". The teller publishes it to the operator on the +// router and, on a warrant that holds the ask (its asker, the ask's digest, an option it offered, in time, and the act +// that option bound), writes the answer into its bucket answers. This owner reads the answer and acts on it only while +// the heads of the key it was written under are still its group's heads; it asks the router nothing. // // - Asked once per head. A group is asked once, for all its members' heads together, by the first member that reaches // this point; the Approve covers every member at those heads. @@ -43,14 +47,27 @@ const ( // AskLasts is how long the operator has to answer: a day less a margin, under the router's bound for an approval. const AskLasts = asks.ApproveLasts - 10*time.Minute -// Operator is the operator channel as this owner reaches it: asker.Client in the running module. +// Operator is how this owner asks the operator: through the keys of its bucket asks and the teller's answers. kvOperator in +// the running module. type Operator interface { - // Ask checks, keeps and publishes an ask under this module's name. - Ask(a asks.Ask) error - // Cancel takes an ask back. - Cancel(id string) error - // Where is an ask's state as the router keeps it, and its warrant once there is one. - Where(id string) (string, *asks.Warrant, error) + // Ask writes the ask under its key (sdk.MergeAskKey); it writes nothing when that key holds the same ask already. + Ask(key string, a sdk.Ask) error + // Cancel takes the ask at a key away, which the teller reads as the ask withdrawn. + Cancel(key string) error + // Where is the teller's answer to the ask at a key, or nil when there is none. + Where(key string) (*sdk.Answer, error) +} + +// askGroup is the name a delivery's ask and merge are kept under: its group's, or, for a lone pull request, a group of one +// under its branch name (novox/hq ADR 0437 §6). +func askGroup(d *Delivery) string { + switch { + case d.Group != "": + return d.Group + case d.Branch != "": + return d.Branch + } + return d.ID } // headsString is the heads an act names, as an act's value: "=" for each, sorted, joined by commas. @@ -287,6 +304,10 @@ func (h *Holder) endStaleAsks() { h.cancels = append(h.cancels, h.dropAsk(r.AskID, "its head "+shortOf(d.Commit, 12)+" is no longer the pull request's head, or the pull request closed")) case !h.headsHold(d, r): h.cancels = append(h.cancels, h.dropAsk(r.AskID, "a head it covers moved on: the heads it covers are no longer the group's heads")) + case r.Approved == nil && r.AskID != "" && !r.AskedAt.IsZero() && !h.Now().Before(r.AskedAt.Add(AskLasts)): + // The ask lapsed at the router with nobody's answer, and nothing in the gate's inputs says so: it is taken away and + // written again, which the teller asks afresh (ADR 0441 §1). + h.cancels = append(h.cancels, h.withdrawAsk(r.AskID, "nobody answered in "+AskLasts.String())) } } } @@ -301,6 +322,9 @@ func (h *Holder) approval(d *Delivery, now time.Time) (string, func()) { r.Approved, r.Warrant, r.AskID, r.Ask, r.Heads = c.Approved, c.Warrant, c.AskID, c.Ask, c.Heads return "", nil } + if !c.AskedAt.IsZero() { + h.lookAt(c.AskedAt.Add(AskLasts), "the operator's Approve, unanswered until the ask lapses") + } return "the operator's Approve is asked (" + c.AskID + ")", nil } if r.Refused != "" { @@ -315,12 +339,18 @@ func (h *Holder) approval(d *Delivery, now time.Time) (string, func()) { group = d.Group } heads := map[string]string{} + var covered []sdk.Head for _, m := range members { heads[m.ID] = m.Commit + covered = append(covered, sdk.Head{Repository: m.Repository, Number: m.Number, Head: m.Commit}) + } + name := askGroup(d) + id, err := sdk.MergeAskKey(name, covered) + if err != nil { + return "the operator cannot be asked: " + err.Error(), nil } r.Asks++ - id := fmt.Sprintf("merge-%d-%s-%d", d.Number, shortOf(d.Commit, 12), r.Asks) - ask := composeMergeAsk(id, d, group, members, heads, now) + ask := composeMergeAsk(sdk.AskID(id), d, group, members, heads, now) taken := []*Delivery{} var reads []memberRead for _, m := range members { @@ -386,7 +416,7 @@ func (h *Holder) approval(d *Delivery, now time.Time) (string, func()) { h.keepGroup(g) h.mu.Unlock() } - if err := h.Operator.Ask(ask); err != nil { + if err := h.Operator.Ask(id, sdk.Ask{Group: name, Heads: covered, Ask: ask}); err != nil { h.mu.Lock() defer h.mu.Unlock() undo() @@ -469,25 +499,25 @@ func (h *Holder) pendingAsks() []string { return out } -// hearAsks reads the router's record of every pending ask and applies the warrant of each that closed. +// hearAsks reads the teller's answer to every pending ask and applies each that is there. func (h *Holder) hearAsks() { if h.Operator == nil { return } h.mu.Lock() - ids := h.pendingAsks() + keys := h.pendingAsks() h.mu.Unlock() - for _, id := range ids { - state, warrant, err := h.Operator.Where(id) + for _, key := range keys { + answer, err := h.Operator.Where(key) if err != nil { - h.Logf("[mesh-delivery] the router's record of the ask %s cannot be read: %v", id, err) + h.Logf("[mesh-delivery] the teller's answer to the ask %s cannot be read: %v", key, err) continue } - if state == "" || state == "open" || state == "held" { + if answer == nil { continue } h.mu.Lock() - said := h.decided(id, state, warrant) + said := h.answered(key, *answer) h.mu.Unlock() for _, s := range said { s() @@ -495,57 +525,54 @@ func (h *Holder) hearAsks() { } } -// decided applies one closed ask's warrant to the deliveries it covers; under the lock. It returns what to say on the -// pull requests. -func (h *Holder) decided(id, state string, w *asks.Warrant) []func() { +// answered applies the teller's answer to the ask at a key to the deliveries it covers; under the lock. The teller has +// held the warrant to the ask (novox/hq ADR 0437 §6); this owner acts on the answer only if the key it was written under is +// still the key of the heads asked: an answer to an ask taken away, or to other heads, merges nothing (ADR 0441 §1). It +// returns what to say on the pull requests. +func (h *Holder) answered(key string, ans sdk.Answer) []func() { var covered []*Delivery for _, d := range h.deliveries { - if d.Ready != nil && d.Ready.AskID == id { + if d.Ready != nil && d.Ready.AskID == key { covered = append(covered, d) } } if len(covered) == 0 { + h.Logf("[mesh-delivery] the answer to the ask %s names no ask that is asked now: nothing is merged on it", key) return nil } sort.Slice(covered, func(i, j int) bool { return covered[i].ID < covered[j].ID }) first, now := covered[0].Ready, h.Now() - end := func(why string) { + refuse := func(why string) { + h.Logf("[mesh-delivery] REFUSED the answer to the ask %s: %s", key, why) for _, d := range covered { d.Ready.AskID, d.Ready.Ask, d.Ready.Heads = "", nil, nil + d.Ready.Refused = "was answered, and the answer does not hold here: " + clip(oneLine(why), 300) h.keep(d) } - h.Logf("[mesh-delivery] the operator's ask %s ended without a choice (%s); asked again at once", id, why) } - if w == nil || w.Outcome != asks.OutcomeChosen { - why := "the router says " + state - if w != nil { - why = "ended " + string(w.Outcome) - if w.Words != "" { - why += " (" + w.Words + ")" - } + var heads []sdk.Head + for id, sha := range first.Heads { + m := h.deliveries[id] + if m == nil { + refuse("the head " + id + " it covers is no delivery of this owner") + return nil } - end(why) + heads = append(heads, sdk.Head{Repository: m.Repository, Number: m.Number, Head: sha}) + } + if want, err := sdk.MergeAskKey(askGroup(covered[0]), heads); err != nil || want != key { + refuse(fmt.Sprintf("its key %s is not the key %s of the heads asked", key, want)) return nil } - option, err := w.For(Module, *first.Ask) - if err == nil { - err = option.Performs(mergeAct(id, option.ID, first.Heads)) - } - if err != nil { - h.Logf("[mesh-delivery] REFUSED the answer to the ask %s: %v", id, err) - for _, d := range covered { - d.Ready.AskID, d.Ready.Ask, d.Ready.Heads = "", nil, nil - d.Ready.Refused = "was answered, and the answer does not hold here: " + clip(oneLine(err.Error()), 300) - h.keep(d) - } + if ans.Option != OptionApprove && ans.Option != OptionDecline { + refuse("it chose " + ans.Option + ", which the ask did not offer") return nil } var said []func() for _, d := range covered { r := d.Ready - r.Warrant = w.Says() + r.Warrant = ans.Words h.mergeAgainNow(d) // the operator chose: the merge is asked at once - switch option.ID { + switch ans.Option { case OptionApprove: r.Approved = &now h.Logf("[mesh-delivery] %s: the operator approved its merge into main (%s)", d.ID, r.Warrant) @@ -559,3 +586,35 @@ func (h *Holder) decided(id, state string, w *asks.Warrant) []func() { } return said } + +// kvOperator is the operator over the runtime: the ask is a key of the gate's bucket asks, the answer a key of the teller's +// bucket answers. +type kvOperator struct{} + +func (kvOperator) Ask(key string, a sdk.Ask) error { + raw, err := sdk.EncodeAsk(a) + if err != nil { + return err + } + if e, err := stdio.State("asks").Get(key); err == nil && e != nil { + if kept, err := sdk.DecodeAsk(e.Value); err == nil && sdk.SameAsk(kept, a) { + return nil // the same ask is there: the teller has it + } + } + _, err = stdio.State("asks").Put(key, json.RawMessage(raw)) + return err +} + +func (kvOperator) Cancel(key string) error { return stdio.State("asks").Delete(key) } + +func (kvOperator) Where(key string) (*sdk.Answer, error) { + e, err := stdio.State(sdk.TellerModule + ".answers").Get(key) + if err != nil || e == nil { + return nil, err + } + a, err := sdk.DecodeAnswer(e.Value) + if err != nil { + return nil, fmt.Errorf("the answer %s is not readable: %w", key, err) + } + return &a, nil +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval_test.go index d440d1af..ea7dd7bd 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/approval_test.go @@ -6,6 +6,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/asks" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" ) // **A merge into main waits for the operator's Approve** (novox/hq ADR 0398 §3, issue 948): a marked, green delivery @@ -84,19 +85,39 @@ func TestAMarkedGreenDeliveryIntoMainIsAskedAndNotMergedUntilApproved(t *testing } } -func TestAnAnswerThatIsNotForThisAskMergesNothing(t *testing.T) { +// AC3 of issue 1348: an answer naming old heads merges nothing. The teller's answer to an ask that was withdrawn when the head +// moved is still in its bucket; the gate acts on the key of the heads asked now alone. +func TestAnAnswerNamingOldHeadsMergesNothing(t *testing.T) { w := markedGreen(t) - id := w.op.open()[0] - w.op.answer(id, OptionApprove) - w.op.mu.Lock() - w.op.warrant[id].AskDigest = "sha256:not-the-ask-that-was-shown" - w.op.mu.Unlock() + old := w.op.open()[0] + w.h.PullUpdated(pr("novox/app", 7, pushed, "feat/x", "src/a.go")) + w.settleAll() + w.op.answer(old, OptionApprove) // the teller wrote it after the ask was taken away + w.settleAll() + if len(w.forge.merges) != 0 || len(w.merges.puts) != 0 { + t.Fatalf("an answer to old heads merged: %v, handed %d", w.forge.merges, len(w.merges.puts)) + } + if d := w.delivery(IDOf("novox/app", readyHead)); d.Ready != nil && d.Ready.Approved != nil { + t.Fatalf("an answer to old heads approved the old delivery: %+v", d.Ready) + } +} + +// An answer under a key that is not the key of the heads the ask covers is refused and said: the gate acts on its own +// ask's key alone (the heads hash is the key's last part). +func TestAnAnswerUnderAKeyThatIsNotTheHeadsKeyIsRefused(t *testing.T) { + w := markedGreen(t) + id := IDOf("novox/app", readyHead) + key := w.op.open()[0] + w.h.mu.Lock() + w.h.deliveries[id].Ready.Heads[id] = strings.Repeat("9", 40) // the heads no longer give the key + w.h.mu.Unlock() + w.op.answer(key, OptionApprove) w.settleAll() if len(w.forge.merges) != 0 { - t.Fatalf("a warrant for another ask merged: %v", w.forge.merges) + t.Fatalf("an answer under another key merged: %v", w.forge.merges) } - d := w.delivery(IDOf("novox/app", readyHead)) - if !strings.Contains(d.Ready.Refused, "digest") || d.Ready.Approved != nil { + d := w.delivery(id) + if !strings.Contains(d.Ready.Refused, "does not hold") || d.Ready.Approved != nil { t.Fatalf("the refused answer is not kept: %+v", d.Ready) } if lines := w.h.Stalled(); len(lines) != 1 || !strings.Contains(lines[0].Says, "answer") { @@ -142,13 +163,15 @@ func TestAPushAfterTheAskCancelsItAndTheNewHeadIsAskedOnItsOwnWhenItsChecksPass( } } -func TestAnExpiredAskIsAskedAgainOnce(t *testing.T) { +// The ask lapses at the router after AskLasts with nobody's answer, and the gate has no word of it: it takes its key away and +// writes it again, once, which the teller asks afresh (it asks again under the same id when the gate writes its key again). +func TestAnUnansweredAskIsWrittenAgainOnceWhenItLapses(t *testing.T) { w := markedGreen(t) first := w.op.open()[0] - w.op.end(first) + w.later(AskLasts + time.Minute) w.settleAll() - if w.op.count() != 2 || len(w.op.open()) != 1 || w.op.open()[0] == first { - t.Fatalf("an expired ask was not asked again once: %d asks, open %v", w.op.count(), w.op.open()) + if w.op.count() != 2 || len(w.op.open()) != 1 || w.op.open()[0] != first || len(w.op.cancelled) != 1 || w.op.cancelled[0] != first { + t.Fatalf("a lapsed ask was not written again once: %d asks, open %v, taken away %v", w.op.count(), w.op.open(), w.op.cancelled) } for range 3 { w.later(2 * time.Minute) @@ -176,12 +199,13 @@ func TestAGroupIsAskedOnceForAllItsHeadsAndMergedInOnePassOnOneApprove(t *testin if open := w.op.open(); len(open) != 1 || len(w.forge.merges) != 0 { t.Fatalf("a group is asked once before any merge: open %v, merges %v", w.op.open(), w.forge.merges) } - a := w.op.ask(w.op.open()[0]) + key := w.op.open()[0] + a := w.op.ask(key) if !strings.Contains(a.Headline, "group feat") || !strings.Contains(a.Details, "mesh-controller#1") || !strings.Contains(a.Details, "mesh-catalog#2") { t.Fatalf("the ask does not name the group and its members: %+v", a) } - w.op.answer(a.ID, OptionApprove) + w.op.answer(key, OptionApprove) w.settleAll() if len(w.forge.merges) != 2 || w.op.count() != 1 { t.Fatalf("both members were not merged in one pass on the one Approve: merges %v, asks %d", w.forge.merges, w.op.count()) @@ -263,7 +287,7 @@ func TestIssue1025AMergeOnTheApproveIsAnnouncedWithItsWarrant(t *testing.T) { } m := got[0] if m["owner"] != "novox" || m["repo"] != "app" || m["number"] != float64(7) || m["head_sha"] != readyHead || - m["ask_id"] != open[0] || m["option"] != OptionApprove { + m["ask_id"] != sdk.AskID(open[0]) || m["asker"] != sdk.TellerModule || m["option"] != OptionApprove { t.Fatalf("the event does not name the pull request and its warrant: %v", m) } hs, _ := m["heads"].(map[string]any) @@ -324,7 +348,7 @@ func TestIssue1025AMergeInAGroupCarriesTheGroupAndItsHeads(t *testing.T) { } for _, m := range got { hs, _ := m["heads"].(map[string]any) - if m["group"] != "feat/x" || m["ask_id"] != open[0] || m["option"] != OptionApprove || + if m["group"] != "feat/x" || m["ask_id"] != sdk.AskID(open[0]) || m["asker"] != sdk.TellerModule || m["option"] != OptionApprove || hs[ctl] != ctlHead || hs[cat] != catHead || len(hs) != 2 { t.Fatalf("the group's merge event: %v", m) } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bridge_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bridge_test.go new file mode 100644 index 00000000..fb16da09 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/bridge_test.go @@ -0,0 +1,136 @@ +package main + +import ( + "sort" + "strconv" + "strings" + "sync" + + sdk "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// fakeMerges is the gate's bucket merges with the forge bridge behind it. Every pull request the gate writes in the state +// merging is merged on the fake forge, as the bridge's merge does (mesh-delivery modules/mesh-delivery-forge-bridge +// internal/bridge acts.go MergeActs): the forge's merge tool is asked at the head with the approved heads of the whole +// group, `@` sorted, and what the fake forge answers is written on FORGE as the bridge writes it, a +// merged message or a refused one with the forge's words. With manual set the bridge says nothing. +type fakeMerges struct { + mu sync.Mutex + forge *fakeForge + h *Holder + values map[string]sdk.Merge + puts []sdk.Merge + manual bool + failed error + // said are the messages the bridge wrote on FORGE, in order. + said []sdk.Forge +} + +func newFakeMerges(f *fakeForge) *fakeMerges { + return &fakeMerges{forge: f, values: map[string]sdk.Merge{}} +} + +func (m *fakeMerges) Get(group string) (*sdk.Merge, error) { + m.mu.Lock() + defer m.mu.Unlock() + if v, ok := m.values[group]; ok { + return &v, nil + } + return nil, nil +} + +func (m *fakeMerges) Put(group string, v sdk.Merge) error { + if _, err := sdk.EncodeMerge(v); err != nil { + return err + } + m.mu.Lock() + if m.failed != nil { + defer m.mu.Unlock() + return m.failed + } + m.values[group] = v + m.puts = append(m.puts, v) + manual := m.manual + m.mu.Unlock() + if manual { + return nil + } + for _, p := range v.Pulls { + if p.State != sdk.MergeMerging { + continue + } + owner, repo, _ := strings.Cut(p.Repository, "/") + var approved []string + for key, sha := range v.Approved { + repository, _, _ := strings.Cut(key, "@") + approved = append(approved, repository+"@"+sha) + } + sortStrings(approved) + f := m.forge + f.mu.Lock() + f.merges = append(f.merges, owner+"/"+repo+"#"+itoa(p.Number)+"@"+p.Head.Head) + f.approvedHeads = append(f.approvedHeads, approved) + err := f.refuseRepo[owner+"/"+repo] + if err == nil { + err = f.refuse + } + f.mu.Unlock() + msg := sdk.Forge{Repository: p.Repository, Number: p.Number, Head: p.Head.Head} + if err != nil { + msg.Kind, msg.Act, msg.Words = sdk.ForgeRefused, "merge", err.Error() + } else { + msg.Kind, msg.Merge, msg.Root = sdk.ForgeMerged, strings.Repeat("c", 40), "root" + msg.Trees = map[string]string{"app": "tree"} + } + m.say(msg) + } + return nil +} + +// say writes a message on FORGE and has the gate hear it, as the runtime does. +func (m *fakeMerges) say(msg sdk.Forge) { + body, err := sdk.EncodeForge(msg) + if err != nil { + panic(err) + } + event, err := sdk.ForgeEvent(msg.Repository, msg.Kind) + if err != nil { + panic(err) + } + m.mu.Lock() + m.said = append(m.said, msg) + m.mu.Unlock() + m.h.Forged("mesh-delivery-forge-bridge."+event, body) +} + +// last is the newest value written for a group. +func (m *fakeMerges) last(group string) sdk.Merge { + m.mu.Lock() + defer m.mu.Unlock() + return m.values[group] +} + +// fakeSource is the gate's bucket source with its one key. +type fakeSource struct { + mu sync.Mutex + main *sdk.Source + moves []sdk.Source +} + +func (s *fakeSource) Move(update func(prev *sdk.Source) (sdk.Source, bool)) error { + s.mu.Lock() + defer s.mu.Unlock() + next, write := update(s.main) + if !write { + return nil + } + if _, err := sdk.EncodeSource(next); err != nil { + return err + } + s.main = &next + s.moves = append(s.moves, next) + return nil +} + +func sortStrings(s []string) { sort.Strings(s) } +func itoa(n int) string { return strconv.Itoa(n) } 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 295641f1..e53ac606 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 @@ -19,6 +19,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asks" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -77,7 +78,7 @@ func newTestBus(t *testing.T) *testBus { // The controller's job: a bucket per state the manifest declares. m, _ := readManifest(t) for _, name := range m.buckets() { - kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: m.Module + "_" + name, History: 1}) + kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: m.Module + "_" + name, History: 10}) if err != nil { t.Fatal(err) } @@ -86,7 +87,7 @@ func newTestBus(t *testing.T) *testBus { // 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}) + kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: strings.Replace(name, ".", "_", 1), History: 10}) if err != nil { t.Fatal(err) } @@ -94,7 +95,8 @@ func newTestBus(t *testing.T) *testBus { } // 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 { + Subjects: []string{"mesh.mod.gitea.event.>", "mesh.seat." + ControllerSeat + ".event.>", "mesh.seat." + asks.Seat + ".event.>", + "mesh.mod." + sdk.ForgeBridgeModule + ".event.>"}}); err != nil { t.Fatal(err) } return b @@ -304,7 +306,16 @@ func (r *runtime) answer(m rpcMessage) { 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 + // The runtime's watch (ADR 0201): each current value, then each change, handed to the bundle as mesh/state under the + // id the watch was asked with. + var id string + _ = json.Unmarshal(m.ID, &id) + kv := r.b.kvs[p.State] + if kv == nil { + return nil, fmt.Errorf("the state %q is not declared", p.State) + } + go r.watch(kv, id, p.State) + return map[string]any{}, nil case "mesh/ask": var a struct { Key string `json:"key"` @@ -390,6 +401,42 @@ func (r *runtime) answer(m rpcMessage) { r.say(map[string]any{"id": m.ID, "result": result}) } +// watch hands the bundle each key of a bucket and each change of it, as the runtime does for a state watch. +func (r *runtime) watch(kv jetstream.KeyValue, id, state string) { + ctx, cancel := context.WithCancel(context.Background()) + r.t.Cleanup(cancel) + w, err := kv.Watch(ctx, ">") + if err != nil { + return + } + defer w.Stop() + current := true + for { + select { + case <-ctx.Done(): + return + case e, ok := <-w.Updates(): + if !ok { + return + } + if e == nil { + current = false + continue + } + op := "put" + if e.Operation() != jetstream.KeyValuePut { + op = "delete" + } + r.mu.Lock() + n := r.next + r.next++ + r.mu.Unlock() + r.say(map[string]any{"id": fmt.Sprintf("watch-%d", n), "method": "mesh/state", "params": map[string]any{"watch": id, + "change": stdio.StateChange{State: state, Key: e.Key(), Op: op, Value: e.Value(), Revision: e.Revision(), Current: current}}}) + } + } +} + // deliverEvents is the runtime reading the events streams and handing each event to the bundle, once, as // `mesh/event`, the way the tool runner does for a module's consumer. func (r *runtime) deliverEvents(ctx context.Context) { @@ -440,15 +487,16 @@ func (b *testBus) publish(t *testing.T, subject string, body any) time.Time { func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealBus(t *testing.T) { b := newTestBus(t) + ctx := context.Background() rt := b.start(t) - ctx, cancel := context.WithCancel(context.Background()) + evctx, cancel := context.WithCancel(ctx) defer cancel() - rt.deliverEvents(ctx) - // The module has started when it hears what it follows, and the operator's answers. + rt.deliverEvents(evctx) + // The module has started when it hears what it follows. rt.until("the module to subscribe to the events it follows", 20*time.Second, func() bool { rt.mu.Lock() defer rt.mu.Unlock() - return len(rt.subs) >= len(followed)+1 + return len(rt.subs) >= len(followed) }) rt.untilHolding() for _, k := range followed { @@ -466,67 +514,59 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB b.publish(t, "mesh.mod.gitea.event.pull.updated", announced) passed := b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", judgedAt("novox/app", 7, head, "pass", "t1t1t1t1t1t1")) - // The operator is asked, within seconds of the verdict's event, with no label, comment or mark written by anyone. - rt.until("the operator to be asked to Approve the merge", 5*time.Second, func() bool { - rt.mu.Lock() - defer rt.mu.Unlock() - return len(rt.operatorAsks) == 1 - }) - rt.mu.Lock() - ask := rt.operatorAsks[0] - asked := time.Now() - rt.mu.Unlock() - if !strings.Contains(ask.Explanation, "not reviewed at this head") { - t.Fatalf("the first ask does not say the record is missing: %s", ask.Explanation) + // The ask is written into the gate's bucket asks within seconds of the verdict's event, with no label, comment or mark + // written by anyone, and nothing is asked of a router. + var key string + rt.until("the ask to be written", 5*time.Second, func() bool { key = onlyAsk(b); return key != "" }) + ask := askAt(t, b, key) + t.Logf("the ask was written %s after the verdict's event", time.Since(passed)) + if !strings.Contains(ask.Ask.Explanation, "not reviewed at this head") || ask.Ask.ID != sdk.AskID(key) { + t.Fatalf("the first ask does not say the record is missing: %s", ask.Ask.Explanation) } - t.Logf("the operator was asked %s after the verdict's event", asked.Sub(passed)) // The description is edited to add the review record (issue 1079): the forge module's edit event carries it. The unanswered - // ask is withdrawn from the router and asked again, once, with the record read; nothing else is asked of the mesh. - first := ask + // ask is taken away and written again, once, with the record read (the heads are the same, so is the key). b.publish(t, "mesh.mod.gitea.event.forge.pull-request.edited", map[string]any{"owner": "novox", "repo": "app", "number": 7, "title": announced.Title, "head_sha": head, "base": "main", "body": "Reviewed-by: lead@shanks at " + head}) - rt.until("the first ask to be withdrawn and the operator asked again", 5*time.Second, func() bool { - rt.mu.Lock() - defer rt.mu.Unlock() - return len(rt.operatorCancels) >= 1 && len(rt.operatorAsks) >= 2 + rt.until("the ask to be written again with the record", 5*time.Second, func() bool { + return strings.Contains(askAt(t, b, key).Ask.Explanation, "reviewed by lead on shanks") }) - rt.mu.Lock() - if len(rt.operatorCancels) != 1 || rt.operatorCancels[0] != first.ID || len(rt.operatorAsks) != 2 { - t.Fatalf("withdrawn %v, asked %d; wanted the first ask %s withdrawn once and exactly one new ask", rt.operatorCancels, len(rt.operatorAsks), first.ID) - } - ask = rt.operatorAsks[1] - rt.mu.Unlock() - if ask.ID == first.ID || !strings.Contains(ask.Explanation, "reviewed by lead on shanks") || strings.Contains(ask.Explanation, "not reviewed") { - t.Fatalf("the new ask is not a fresh one naming the reviewer: %s %s", ask.ID, ask.Explanation) + if h, err := b.kvs["asks"].History(ctx, key); err != nil || len(h) != 3 || h[1].Operation() != jetstream.KeyValueDelete { + t.Fatalf("the ask's history is %d revisions (%v); wanted written, taken away, written", len(h), err) } if len(rt.asksOf("gitea.gitea_merge_pull_request")) != 0 { t.Fatalf("merged before the Approve") } // The pull request's key holds its head, the trunk commit it was asked against and the verdict. rt.until("the pull request's key to hold head, trunk and verdict", 5*time.Second, func() bool { - e, err := b.kvs["pulls"].Get(context.Background(), "novox.app.7") + e, err := b.kvs["pulls"].Get(ctx, "novox.app.7") var k PullKey return err == nil && json.Unmarshal(e.Value(), &k) == nil && k == PullKey{Head: head, Trunk: "t1t1t1t1t1t1", Verdict: "pass"} }) - // The operator approves (a person's Approve, proven by the router): the merge is asked within seconds. - opt, _ := ask.Option(OptionApprove) - approved := b.publish(t, "mesh.seat."+asks.Seat+".event.decided."+Module, asks.Warrant{Ask: ask.ID, Asker: Module, - Outcome: asks.OutcomeChosen, Option: OptionApprove, Label: opt.Label, Level: opt.Level, - By: &asks.Person{Who: asks.Operator, Kind: "telegram", Verified: "user id verified"}, At: time.Now(), AskDigest: ask.Digest()}) - rt.until("the merge to be asked of the forge", 5*time.Second, func() bool { return len(rt.asksOf("gitea.gitea_merge_pull_request")) >= 1 }) - merges := rt.asksOf("gitea.gitea_merge_pull_request") - if len(merges) != 1 || merges[0].Body["head"] != head || merges[0].Body["number"] != float64(7) { - t.Fatalf("the merge was asked %+v; wanted once, for pull request 7 at the approved head", merges) + // The operator approves; the teller holds the warrant and writes the answer: the merge is handed to the bridge within + // seconds, with the Approve's heads, and the gate asks the forge nothing. + approved := b.answer(t, key, OptionApprove) + var handed sdk.Merge + rt.until("the merge to be handed to the bridge", 5*time.Second, func() bool { handed = mergeAt(b, "fix/x"); return len(handed.Pulls) > 0 }) + if len(handed.Pulls) != 1 || handed.Pulls[0].Head.Head != head || handed.Pulls[0].Number != 7 || handed.Pulls[0].State != sdk.MergeMerging || + handed.Approved[sdk.ApprovedKey("novox/app", 7)] != head || handed.Ask != key { + t.Fatalf("the merge handed is %+v; wanted pull request 7 at the approved head, merging, under the ask %s", handed, key) } - if got := fmt.Sprint(merges[0].Body["approved_heads"]); got != "[novox/app@"+head+"]" { - t.Fatalf("the merge tool was handed the approved heads %s", got) - } - if took := merges[0].At.Sub(approved); took > 5*time.Second { - t.Fatalf("the merge was asked %s after the Approve's event; wanted within 5s", took) - } else { - t.Logf("the merge was asked %s after the Approve's event", took) + t.Logf("the merge was handed %s after the answer", time.Since(approved)) + if n := len(rt.asksOf("gitea.gitea_merge_pull_request")); n != 0 { + t.Fatalf("the gate asked the forge to merge %d times: the bridge merges", n) } + // The bridge says it merged; the mesh's source then holds the merge, in one write. + b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: "novox/app", Number: 7, Head: head, Merge: strings.Repeat("b", 40), + Root: "r", Trees: map[string]string{"app": "t"}}) + rt.until("source.main to hold the merge", 5*time.Second, func() bool { + e, err := b.kvs["source"].Get(ctx, sdk.SourceMainKey) + if err != nil { + return false + } + src, err := sdk.DecodeSource(e.Value()) + return err == nil && src.Group == "fix/x" && src.Heads["novox/app"] == strings.Repeat("b", 40) + }) // Quiet follows: nothing asks the forge or the controller again with no event to say so. quiet := len(rt.asked) time.Sleep(1500 * time.Millisecond) @@ -538,20 +578,81 @@ func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealB } } -// **The acceptance of one group, one Approve** (novox/hq ADR 0418 §4, ADR 0421, design 47 milestone 3, issues 978 and -// 1022): a group of two pull requests on one branch name, in two repositories, with green checks, gets one ask of the -// operator — no label, comment or mark written by anyone — and on the Approve both merge in one pass, each handed the -// approved heads of both. -func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeInOnePassOverARealBus(t *testing.T) { +// onlyAsk is the key of the one ask in the gate's bucket asks, or empty. +func onlyAsk(b *testBus) string { + keys, err := b.kvs["asks"].Keys(context.Background()) + if err != nil || len(keys) != 1 { + return "" + } + return keys[0] +} + +func askAt(t *testing.T, b *testBus, key string) sdk.Ask { + t.Helper() + e, err := b.kvs["asks"].Get(context.Background(), key) + if err != nil { + t.Fatalf("the ask %s: %v", key, err) + } + a, err := sdk.DecodeAsk(e.Value()) + if err != nil { + t.Fatal(err) + } + return a +} + +func mergeAt(b *testBus, group string) sdk.Merge { + e, err := b.kvs["merges"].Get(context.Background(), group) + if err != nil { + return sdk.Merge{} + } + m, _ := sdk.DecodeMerge(e.Value()) + return m +} + +// answer is the teller's answer to an ask, written under the ask's own key in its bucket answers. +func (b *testBus) answer(t *testing.T, key, option string) time.Time { + t.Helper() + raw, err := sdk.EncodeAnswer(sdk.Answer{Option: option, Warrant: "w-" + key, Words: "the operator, via telegram, chose " + option}) + if err != nil { + t.Fatal(err) + } + if _, err := b.kvs["mesh-delivery-teller.answers"].Put(context.Background(), key, raw); err != nil { + t.Fatal(err) + } + return time.Now() +} + +// forgeSays is the forge bridge's message on FORGE, as its stream carries it. +func (b *testBus) forgeSays(t *testing.T, f sdk.Forge) { + t.Helper() + raw, err := sdk.EncodeForge(f) + if err != nil { + t.Fatal(err) + } + event, err := sdk.ForgeEvent(f.Repository, f.Kind) + if err != nil { + t.Fatal(err) + } + if _, err := b.js.Publish(context.Background(), "mesh.mod."+sdk.ForgeBridgeModule+".event."+event, raw); err != nil { + t.Fatal(err) + } +} + +// **The acceptance of one group, one Approve** (novox/hq ADR 0418 §4, ADR 0421, ADR 0441 §5, design 47 milestone 3, issues 978 +// and 1022): a group of two pull requests on one branch name, in two repositories, with green checks, gets one ask of the +// operator — no label, comment or mark written by anyone — and on the Approve the members are handed to the bridge one at a time, +// the second after the first merged, each with the approved heads of both. +func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeOneAfterTheOtherOverARealBus(t *testing.T) { b := newTestBus(t) + ctx := context.Background() rt := b.start(t) - ctx, cancel := context.WithCancel(context.Background()) + evctx, cancel := context.WithCancel(ctx) defer cancel() - rt.deliverEvents(ctx) + rt.deliverEvents(evctx) rt.until("the module to subscribe to the events it follows", 20*time.Second, func() bool { rt.mu.Lock() defer rt.mu.Unlock() - return len(rt.subs) >= len(followed)+1 + return len(rt.subs) >= len(followed) }) rt.untilHolding() const ctlHead, catHead = "c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1", "d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2" @@ -577,61 +678,62 @@ func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeInOnePassOv if n := len(rt.asksOf("seat:" + ControllerSeat + ".delivery-order")); n != 0 { t.Fatalf("the controller's delivery-order was asked %d times: no order is asked for", n) } - rt.mu.Lock() - nothing := len(rt.operatorAsks) - rt.mu.Unlock() - if nothing != 0 { - t.Fatalf("the operator was asked before the group's composed check passed") + if keys, _ := b.kvs["asks"].Keys(ctx); len(keys) != 0 { + t.Fatalf("the operator was asked before the group's composed check passed: %v", keys) } // The composed check passes at both heads: the operator is asked once, for the group. b.publish(t, "mesh.seat."+ControllerSeat+".event.checked", map[string]any{"group": "feat/x", "id": "check-1", "verdict": "pass", "summary": "both compose", "members": []map[string]any{ {"owner": "novox", "repo": "mesh-controller", "commit": ctlHead, "trunk": "t1t1t1t1t1t1"}, {"owner": "novox", "repo": "mesh-catalog", "commit": catHead, "trunk": "t1t1t1t1t1t1"}}}) - rt.until("the operator to be asked once for the group", 10*time.Second, func() bool { - rt.mu.Lock() - defer rt.mu.Unlock() - return len(rt.operatorAsks) == 1 - }) + var key string + rt.until("the operator to be asked once for the group", 10*time.Second, func() bool { key = onlyAsk(b); return key != "" }) time.Sleep(500 * time.Millisecond) - rt.mu.Lock() - asked, ask := len(rt.operatorAsks), rt.operatorAsks[0] - rt.mu.Unlock() - if asked != 1 { - t.Fatalf("the operator was asked %d times for one group", asked) + if keys, _ := b.kvs["asks"].Keys(ctx); len(keys) != 1 { + t.Fatalf("the operator was asked %d times for one group", len(keys)) } - if len(rt.asksOf("gitea.gitea_merge_pull_request")) != 0 { + if g := askAt(t, b, key).Group; g != "feat/x" { + t.Fatalf("the ask is for the group %q", g) + } + if len(rt.asksOf("gitea.gitea_merge_pull_request")) != 0 || mergeAt(b, "feat/x").Group != "" { t.Fatalf("merged before the Approve") } - opt, _ := ask.Option(OptionApprove) - approved := b.publish(t, "mesh.seat."+asks.Seat+".event.decided."+Module, asks.Warrant{Ask: ask.ID, Asker: Module, - Outcome: asks.OutcomeChosen, Option: OptionApprove, Label: opt.Label, Level: opt.Level, - By: &asks.Person{Who: asks.Operator, Kind: "telegram", Verified: "user id verified"}, At: time.Now(), AskDigest: ask.Digest()}) + approved := b.answer(t, key, OptionApprove) - // Both merge on that one Approve, in one pass: asked within seconds of the Approve's event, each with both heads. - rt.until("both merges to be asked of the forge", 5*time.Second, func() bool { return len(rt.asksOf("gitea.gitea_merge_pull_request")) >= 2 }) - merges := rt.asksOf("gitea.gitea_merge_pull_request") - if len(merges) != 2 { - t.Fatalf("the merge was asked %d times; wanted once for each member: %+v", len(merges), merges) + // One member is handed, with both heads approved, within seconds of the answer; the second only after the bridge says the + // first merged. + var first sdk.Merge + rt.until("the first member to be handed", 5*time.Second, func() bool { first = mergeAt(b, "feat/x"); return len(first.Pulls) > 0 }) + if len(first.Pulls) != 1 || first.Pulls[0].State != sdk.MergeMerging || len(first.Approved) != 2 || + first.Approved[sdk.ApprovedKey("novox/mesh-controller", 1)] != ctlHead || first.Approved[sdk.ApprovedKey("novox/mesh-catalog", 2)] != catHead { + t.Fatalf("the first merge handed is %+v", first) } - heads := map[string]bool{} - for _, m := range merges { - heads[fmt.Sprint(m.Body["repo"], "@", m.Body["head"])] = true - if got := fmt.Sprint(m.Body["approved_heads"]); got != "[novox/mesh-catalog@"+catHead+" novox/mesh-controller@"+ctlHead+"]" { - t.Fatalf("a merge was handed the approved heads %s", got) + t.Logf("the first member was handed %s after the answer", time.Since(approved)) + time.Sleep(500 * time.Millisecond) + if again := mergeAt(b, "feat/x"); len(again.Pulls) != 1 { + t.Fatalf("the second member was handed before the first merged: %+v", again.Pulls) + } + one := first.Pulls[0] + b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: one.Repository, Number: one.Number, Head: one.Head.Head, + Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}}) + var both sdk.Merge + rt.until("the second member to be handed", 5*time.Second, func() bool { both = mergeAt(b, "feat/x"); return len(both.Pulls) == 2 }) + if both.Pulls[0].State != sdk.MergeMerged || both.Pulls[1].State != sdk.MergeMerging || both.Pulls[1].Repository == one.Repository { + t.Fatalf("the second merge handed is %+v", both.Pulls) + } + two := both.Pulls[1] + b.forgeSays(t, sdk.Forge{Kind: sdk.ForgeMerged, Repository: two.Repository, Number: two.Number, Head: two.Head.Head, + Merge: strings.Repeat("b", 40), Root: "r", Trees: map[string]string{"x": "t"}}) + rt.until("source.main to hold both merges", 5*time.Second, func() bool { + e, err := b.kvs["source"].Get(ctx, sdk.SourceMainKey) + if err != nil { + return false } - if took := m.At.Sub(approved); took > 5*time.Second { - t.Fatalf("a merge was asked %s after the Approve's event; wanted within 5s", took) - } - } - if !heads["mesh-controller@"+ctlHead] || !heads["mesh-catalog@"+catHead] { - t.Fatalf("the merges were asked for %v", heads) - } - rt.mu.Lock() - again := len(rt.operatorAsks) - rt.mu.Unlock() - if again != 1 { - t.Fatalf("a second ask was made: %d", again) + src, err := sdk.DecodeSource(e.Value()) + return err == nil && src.Group == "feat/x" && len(src.Heads) == 2 + }) + if keys, _ := b.kvs["asks"].Keys(ctx); len(keys) != 1 { + t.Fatalf("a second ask was made: %v", keys) } } @@ -649,7 +751,7 @@ func TestAPlanMovedCarryingAFailedBackupHoldsTheDeliveryWithTheBackupsWordsOverA rt.until("the module to subscribe to the events it follows", 20*time.Second, func() bool { rt.mu.Lock() defer rt.mu.Unlock() - return len(rt.subs) >= len(followed)+1 + return len(rt.subs) >= len(followed) }) rt.untilHolding() const head, merge = "3333333333333333333333333333333333333333", "4444444444444444444444444444444444444444" 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 7713e463..8eb4bab5 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events.go @@ -4,8 +4,6 @@ import ( "fmt" "slices" "strings" - - "git.novox.be/novox/mesh-sdk/go/asks" ) // Events that start a step (novox/hq ADR 0406 §1, issue 975, ADR 0421 issue 1022). @@ -39,17 +37,10 @@ func (h *Holder) takeForgeWanted() bool { return wanted } -// 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() - for _, s := range said { - s() +// AnswersChanged takes a change of the teller's answers: the answers to the asks pending are read at the look it asks for. +func (h *Holder) AnswersChanged() { + if h.Holding() { + h.Poke() } } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events_test.go index 84324b11..fcea942e 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/events_test.go @@ -66,9 +66,7 @@ func TestAFailedVerdictAsksNothing(t *testing.T) { func approve(w *world) { open := w.op.open() w.op.answer(open[0], OptionApprove) - state, warrant, _ := w.op.Where(open[0]) - w.h.AskDecided(open[0], state, warrant) - w.h.React() + w.settleAll() } func TestAnApproveHeardAsAnEventMergesAtOnceAtTheApprovedHead(t *testing.T) { 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 4b85bc65..3360b384 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 @@ -591,12 +591,16 @@ func (f *fakeForge) notesOn(commit string) []string { // world is a holder over fakes, with a clock a test moves. type world struct { - t *testing.T - h *Holder - store *memStore - ctl *fakeController - forge *fakeForge - op *fakeOperator + t *testing.T + h *Holder + store *memStore + ctl *fakeController + forge *fakeForge + op *fakeOperator + merges *fakeMerges + source *fakeSource + // asked is the key of the ask a test answered. + asked string now time.Time said []map[string]any mu sync.Mutex @@ -624,6 +628,9 @@ func newWorldOver(t *testing.T, store *memStore, ctl *fakeController, forge *fak w.op = newFakeOperator(w.h.Now) w.op.auto = true w.h.Operator = w.op + w.merges, w.source = newFakeMerges(forge), &fakeSource{} + w.h.Merges, w.h.Source = w.merges, w.source + w.merges.h = w.h w.h.init() if err := w.h.Load(); err != nil { t.Fatal(err) @@ -637,28 +644,14 @@ func (w *world) later(d time.Duration) { w.mu.Unlock() } -// settleAll lets the router's answers reach the holder as its `decided` events, and takes a look, a few times. +// settleAll takes a look a few times: the answers the teller wrote are read at each, and each member of a group handed to the +// forge bridge is merged by the fake bridge, whose word is heard by the next look. func (w *world) settleAll() { for range 4 { - w.decide() w.h.React() } } -// decide hands the holder the router's `decided` event for every ask of its that has closed. -func (w *world) decide() { - w.h.mu.Lock() - ids := w.h.pendingAsks() - w.h.mu.Unlock() - for _, id := range ids { - state, warrant, _ := w.op.Where(id) - if state == "" || state == "open" || state == "held" { - continue - } - w.h.AskDecided(id, state, warrant) - } -} - func (w *world) delivery(id string) *Delivery { w.t.Helper() w.h.mu.Lock() @@ -715,18 +708,7 @@ func aWalk(id, repo, commit string, waits bool) Walk { return w } -// Merge and Comment record what was asked (novox/hq issue 722); Merge answers refuse when set. -func (f *fakeForge) Merge(owner, repo string, number int, head string, approved []string) error { - f.mu.Lock() - defer f.mu.Unlock() - f.merges = append(f.merges, fmt.Sprintf("%s/%s#%d@%s", owner, repo, number, head)) - f.approvedHeads = append(f.approvedHeads, approved) - if err := f.refuseRepo[owner+"/"+repo]; err != nil { - return err - } - return f.refuse -} - +// Comment records what was asked (novox/hq issue 722). func (f *fakeForge) Comment(owner, repo string, number int, body string) error { f.mu.Lock() defer f.mu.Unlock() diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/group_merge.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/group_merge.go index 049d8179..c676babe 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/group_merge.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/group_merge.go @@ -65,66 +65,33 @@ func (h *Holder) unmergedLive(g *Group) []*Delivery { return out } -// groupPass is the one pass that merges a group's members, made by the first of them in the group's order that may be -// merged; nil when it is not d's turn, or a member is not approved at its head. Under the lock; the pass runs outside it. -func (h *Holder) groupPass(g *Group, d *Delivery, now time.Time) func() { - order := h.unmergedLive(g) - if len(order) == 0 || order[0] != d { - return nil - } - var pass []passMember - for _, m := range order { - c := h.covering(m) - if m.Ready == nil || m.Ready.Declined != nil || c == nil || c.Approved == nil || c.Heads[m.ID] != m.Commit || h.readyWaits(m) != "" { - return nil // the others are not approved at their heads yet: nothing is merged before they are - } - pass = append(pass, passMember{read: readOf(m, c.Heads[m.ID]), heads: c.Heads}) - } - for _, m := range order { - m.Ready.Tries++ - m.Ready.LastTry, m.Ready.merging = now, true - h.keep(m) - } - gid := g.ID - return func() { h.mergeGroupPass(gid, pass) } -} - -// passMember is one member of a pass as it was taken under the lock. -type passMember struct { - read memberRead - heads map[string]string -} - -// mergeGroupPass reads every member's required statuses, and only when all are a success merges the members in order, -// stopping at the first the merge tool refuses. -func (h *Holder) mergeGroupPass(gid string, pass []passMember) { - release := func(from int, retry bool) { - for _, p := range pass[from:] { - if d := h.deliveries[p.read.id]; d != nil && d.Ready != nil { - d.Ready.merging = false - if retry { - d.Ready.LastTry = time.Time{} - } - } +// groupMergeable reads every member's required statuses at the head the Approve covers, and says whether all are a success. +// When one is not, or cannot be read, nothing is handed to the bridge: the group is held, said urgent (ADR 0441 §2, §4), and +// the target's turn is given back so the next event that decides whether it may merge asks it again. Outside the lock. +func (h *Holder) groupMergeable(gid, target string, reads []memberRead) bool { + release := func() { + if d := h.deliveries[target]; d != nil && d.Ready != nil { + d.Ready.merging = false + d.Ready.LastTry = time.Time{} } } - for _, p := range pass { - says, ownOnly, err := h.notASuccess(p.read) + for _, m := range reads { + says, ownOnly, err := h.notASuccess(m) if err == nil && says == "" { continue } h.mu.Lock() if err != nil { - h.Logf("[mesh-delivery] group %s: the required statuses of %s cannot be read, so nothing is merged yet: %v", gid, p.read.id, err) - release(0, true) + h.Logf("[mesh-delivery] group %s: the required statuses of %s cannot be read, so nothing is merged yet: %v", gid, m.id, err) + release() h.again("a group's merge whose required statuses could not be read") h.mu.Unlock() - return + return false } var own func() if g := h.groups[gid]; g != nil { - if g.Hold == nil || g.Hold.Kind != "blocked" || g.Hold.Says != says || g.Hold.Member != p.read.id { - h.Logf("[mesh-delivery] group %s is approved and NOT merged: %s at %s: %s", gid, p.read.id, shortOf(p.read.head, 12), says) + if g.Hold == nil || g.Hold.Kind != "blocked" || g.Hold.Says != says || g.Hold.Member != m.id { + h.Logf("[mesh-delivery] group %s is approved and NOT merged: %s at %s: %s", gid, m.id, shortOf(m.head, 12), says) } since := time.Time{} if g.Hold != nil && g.Hold.Kind == "blocked" { @@ -133,19 +100,19 @@ func (h *Holder) mergeGroupPass(gid string, pass []passMember) { if since.IsZero() { since = h.Now() } - g.Hold = &GroupHold{Kind: "blocked", Member: p.read.id, Says: says, Since: since} + g.Hold = &GroupHold{Kind: "blocked", Member: m.id, Says: says, Since: since} h.keepGroup(g) } - if d := h.deliveries[p.read.id]; ownOnly && d != nil && d.Ready != nil { + if d := h.deliveries[m.id]; ownOnly && d != nil && d.Ready != nil { own = h.askOwnAfterRefusal(d) // its own repository check is what is missing: asked, once for the head } - release(0, true) + release() h.again("a group's merge held for a required status") h.mu.Unlock() if own != nil { own() } - return + return false } h.mu.Lock() if g := h.groups[gid]; g != nil && g.Hold != nil { @@ -153,19 +120,7 @@ func (h *Holder) mergeGroupPass(gid string, pass []passMember) { h.keepGroup(g) } h.mu.Unlock() - for i, p := range pass { - if h.mergeOne(p.read.id, p.heads) { - continue - } - h.mu.Lock() - release(i+1, false) - if g := h.groups[gid]; g != nil { - h.Logf("[mesh-delivery] group %s: the merge of %s was refused, so none of the members after it is merged", gid, p.read.id) - h.keepGroup(g) - } - h.mu.Unlock() - return - } + return true } // groupLines are the groups that are held or half merged, as urgent stalled lines the controller raises as conditions: 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 ff530ac8..d96772ce 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go @@ -28,8 +28,11 @@ type Holder struct { Emit func(event string, body any) error Now func() time.Time Logf func(format string, a ...any) - // Operator is the operator channel, asked for the Approve of a merge into main (approval.go). + // Operator keeps the ask of the Approve of a merge into main in the gate's asks and reads the teller's answers + // (approval.go); Merges and Source are the gate's merges and source (merges.go). Operator Operator + Merges MergeStore + Source SourceStore // 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 diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1079_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1079_test.go index c5cb6cec..25b8fa78 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1079_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1079_test.go @@ -39,8 +39,10 @@ func TestAReviewedByLineWrittenAfterTheAskWithdrawsTheUnansweredAskAndAsksAgain( w.h.PullEdited(edited("Reviewed-by: reviewer@shanks at " + readyHead)) w.settleAll() open := w.op.open() - if len(open) != 1 || open[0] == first || w.op.count() != 2 { - t.Fatalf("after the edit the open asks are %v (first %s), %d asked; wanted the first withdrawn and one new", open, first, w.op.count()) + // The heads are the same, so the key is: the ask is taken away and written again with the record in its words, which the + // teller reads as the ask to withdraw and ask afresh. + if len(open) != 1 || open[0] != first || w.op.count() != 2 { + t.Fatalf("after the edit the open asks are %v (first %s), %d asked; wanted the first withdrawn and written again once", open, first, w.op.count()) } if len(w.op.cancelled) != 1 || w.op.cancelled[0] != first { t.Fatalf("the first ask was not withdrawn: %v", w.op.cancelled) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1348_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1348_test.go new file mode 100644 index 00000000..07bb1c26 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/issue1348_test.go @@ -0,0 +1,175 @@ +package main + +import ( + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/asks" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **The gate asks through its bucket asks and merges through the bridge** (novox/hq ADR 0437 §6, §8, ADR 0441 §1 to §5, +// ADR 0450 §3, issue 1348). Fixture sources: the key and the id of an ask and the layout of a merge are mesh-sdk go/delivery's +// (MergeAskKey, AskID, Merge); the act an ask binds is spelled as mesh-controller cmd/mesh-controller/moved.go:98-106 verifies +// it and as the teller checks it (mesh-delivery modules/mesh-delivery-teller teller.go mergeAct, headsOf); the bridge's +// messages are the ones mesh-delivery modules/mesh-delivery-forge-bridge internal/bridge acts.go writes (RefusedMessage) and +// bridge.go merged writes, and the fake bridge here (bridge_test.go) writes them in that layout. + +// AC1: a ready group writes one ask key, with today's words, and the gate calls no router. +func TestAReadyGroupWritesOneAskKeyWithTodaysWordsAndCallsNoRouter(t *testing.T) { + w := approvedGroup(t) + allGreen(w) + if w.op.count() != 1 { + t.Fatalf("%d asks written, wanted one", w.op.count()) + } + key := w.op.open()[0] + kind, group, hash, err := sdk.ParseAskKey(key) + if err != nil || kind != "merge" || group != "feat/g" { + t.Fatalf("the key %q is not merge..: %q %q %v", key, kind, group, err) + } + ask := w.op.byKey[key] + if ask.Group != "feat/g" || len(ask.Heads) != 2 { + t.Fatalf("the ask value %+v does not name the group and both heads", ask) + } + if want, err := sdk.MergeAskKey("feat/g", ask.Heads); err != nil || want != key || !strings.HasSuffix(key, hash) { + t.Fatalf("the heads %v give the key %q, the ask is under %q (%v)", ask.Heads, want, key, err) + } + if ask.Ask.ID != sdk.AskID(key) { + t.Fatalf("the router's id of the ask is %q, wanted %q", ask.Ask.ID, sdk.AskID(key)) + } + // Today's words (approval.go composeMergeAsk): the headline names the group, the details its members. + if !strings.Contains(ask.Ask.Headline, "group feat") || !strings.Contains(ask.Ask.Details, "mesh-controller#1") || + !strings.Contains(ask.Ask.Details, "mesh-catalog#2") || len(ask.Ask.Options) != 2 || ask.Ask.Who != asks.Operator { + t.Fatalf("the ask is not today's merge ask: %+v", ask.Ask) + } + // The act the Approve binds is the one the teller checks and the controller verifies: this ask's router id, the option, and + // the heads as `owner/repo@` to the whole commit. + approve, _ := ask.Ask.Option(OptionApprove) + heads := map[string]string{IDOf(gCtl, gHc): gHc, IDOf(gCat, gHk): gHk} + want, _ := asks.ActDigest(asks.Act{"verb": "mesh-delivery.merge", "ask": sdk.AskID(key), "option": "approve", + "heads": "novox/mesh-catalog@d2d2d2d2d2d2=" + gHk + ",novox/mesh-controller@c1c1c1c1c1c1=" + gHc}) + if approve.Binds != want { + t.Fatalf("the Approve binds %s, wanted the act %v of %v", approve.Binds, mergeAct(sdk.AskID(key), "approve", heads), heads) + } + if len(w.forge.calls) != 0 || len(w.merges.puts) != 0 || len(w.forge.merges) != 0 { + t.Fatalf("something was written beside the ask: forge %v, merges %d", w.forge.calls, len(w.merges.puts)) + } +} + +// answeredFirst is a group approved by the teller's answer, with the bridge saying nothing yet. +func answeredFirst(t *testing.T) *world { + w := approvedGroup(t) + allGreen(w) + w.merges.manual = true + w.asked = w.op.open()[0] + w.op.answer(w.asked, OptionApprove) + w.settleAll() + return w +} + +// AC2: an answer naming the heads asked now hands the first member only; the bridge's merged hands the next; its refused +// hands none. +func TestAnAnswerHandsTheFirstMemberOnlyAndTheBridgesMergedHandsTheNext(t *testing.T) { + w := answeredFirst(t) + if len(w.merges.puts) != 1 { + t.Fatalf("the gate wrote %d merges, wanted the first member handed once", len(w.merges.puts)) + } + first := w.merges.last("feat/g") + if len(first.Pulls) != 1 || first.Pulls[0].Repository != gCat || first.Pulls[0].State != sdk.MergeMerging || first.Pulls[0].Head.Head != gHk { + t.Fatalf("the first member is not handed alone, merging, at its head: %+v", first.Pulls) + } + if first.Approved[sdk.ApprovedKey(gCat, 2)] != gHk || first.Approved[sdk.ApprovedKey(gCtl, 1)] != gHc || len(first.Approved) != 2 { + t.Fatalf("the merge does not carry the Approve's heads of both members: %v", first.Approved) + } + if first.Ask != w.asked { + t.Fatalf("the merge names no ask: %+v", first) + } + w.settleAll() + if len(w.merges.puts) != 1 { + t.Fatalf("the next member was handed before the bridge said the first merged: %d merges written", len(w.merges.puts)) + } + w.merges.say(sdk.Forge{Kind: sdk.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}}) + w.settleAll() + if len(w.merges.puts) != 2 { + t.Fatalf("the second member was not handed after the first merged: %d merges written", len(w.merges.puts)) + } + second := w.merges.last("feat/g") + if len(second.Pulls) != 2 || second.Pulls[0].State != sdk.MergeMerged || second.Pulls[0].Commit != strings.Repeat("a", 40) || + second.Pulls[1].Repository != gCtl || second.Pulls[1].State != sdk.MergeMerging { + t.Fatalf("the next merge is not the first merged and the second merging: %+v", second.Pulls) + } +} + +// AC2: the bridge's refused hands no member, and the group is said urgent, naming the member and the forge's words. +func TestTheBridgesRefusalHandsNoMemberAndIsSaidUrgent(t *testing.T) { + w := answeredFirst(t) + w.merges.say(sdk.Forge{Kind: sdk.ForgeRefused, Repository: gCat, Number: 2, Head: gHk, Act: "merge", + Words: "the forge refused gitea_merge_pull_request: mesh/repo-check (missing)"}) + w.later(time30s) + w.settleAll() + if len(w.merges.puts) != 2 { + t.Fatalf("%d merges written; wanted the first handed and then written refused, and none handed after", len(w.merges.puts)) + } + refused := w.merges.last("feat/g") + if len(refused.Pulls) != 1 || refused.Pulls[0].State != sdk.MergeRefused || !strings.Contains(refused.Pulls[0].Words, "mesh/repo-check (missing)") { + t.Fatalf("the refusal is not written into the merge: %+v", refused.Pulls) + } + lines := urgentLines(w) + if len(lines) != 1 || lines[0].ID != IDOf(gCat, gHk) || !strings.Contains(lines[0].Says, "feat/g") || + !strings.Contains(lines[0].Says, "mesh/repo-check (missing)") { + t.Fatalf("the refusal is not said as one urgent line: %+v", lines) + } +} + +// ADR 0441 §2: before a member is handed, every member's required statuses are read again; one that is not a success holds +// the group, even after the first member merged. +func TestTheNextMemberIsNotHandedWhileARequiredStatusOfTheGroupIsNotASuccess(t *testing.T) { + w := answeredFirst(t) + w.merges.say(sdk.Forge{Kind: sdk.ForgeMerged, Repository: gCat, Number: 2, Head: gHk, Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}}) + setChecks(w, gCtl, gHc, "success", "") // the second member's repository check went missing meanwhile + w.settleAll() + if len(w.merges.puts) != 1 { + t.Fatalf("the second member was handed with a required status missing: %d merges written", len(w.merges.puts)) + } + if lines := urgentLines(w); len(lines) != 1 || !strings.Contains(lines[0].Says, "mesh/repo-check (missing)") { + t.Fatalf("the held group is not said urgent: %+v", lines) + } +} + +// AC4: once every member is merged the gate writes source.main once, and each merge is announced with the ask, its asker the +// teller, and the heads. +func TestAGroupMergedWholeMovesSourceMainOnceAndAnnouncesWithTheTeller(t *testing.T) { + w := approvedGroup(t) + allGreen(w) + w.op.answer(w.op.open()[0], OptionApprove) + w.settleAll() + w.settleAll() + if len(w.forge.merges) != 2 { + t.Fatalf("merges %v", w.forge.merges) + } + if len(w.source.moves) != 1 { + t.Fatalf("source.main moved %d times, wanted once", len(w.source.moves)) + } + main := w.source.moves[0] + commit := strings.Repeat("c", 40) + if main.Group != "feat/g" || main.Heads[gCat] != commit || main.Heads[gCtl] != commit || len(main.Heads) != 2 { + t.Fatalf("source.main is %+v; wanted both repositories at their merge commits, moved by the group", main) + } + final := w.merges.last("feat/g") + if len(final.Pulls) != 2 || final.Pulls[0].State != sdk.MergeMerged || final.Pulls[1].State != sdk.MergeMerged { + t.Fatalf("the merge is not written as all merged: %+v", final.Pulls) + } + say := mergedSaid(w) + if len(say) != 2 { + t.Fatalf("announced %d merges, wanted each member's once: %v", len(say), say) + } + for _, m := range say { + if m["asker"] != sdk.TellerModule || !strings.HasPrefix(m["ask_id"].(string), "merge-feat_g-") || m["option"] != OptionApprove { + t.Fatalf("the announcement does not name the ask and its asker the teller: %v", m) + } + } + w.settleAll() + if len(w.source.moves) != 1 { + t.Fatalf("source.main moved again: %d", len(w.source.moves)) + } +} 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 765d2e8d..4588a811 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/main.go @@ -16,8 +16,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/asker" - "git.novox.be/novox/mesh-sdk/go/asks" + sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery" ) func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } @@ -34,17 +33,22 @@ func (l *listening) get() string { l.mu.Lock(); defer l.mu.Unlock(); return l.n // The events this owner follows: the forge's pull requests, and the controller's verdicts and walks. // The forge's status events say that a check answered (events.go). var followed = []string{"gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed", - "gitea.forge.status", "gitea.forge.pull-request.edited", "gitea.forge.pull-request.closed", ControllerSeat + ".checked", ControllerSeat + ".plan-moved"} + "gitea.forge.status", "gitea.forge.pull-request.edited", "gitea.forge.pull-request.closed", ControllerSeat + ".checked", ControllerSeat + ".plan-moved", + forgePattern} + +// forgePattern is the forge bridge's messages on FORGE (novox/hq ADR 0437 §8): the gate follows a merge and the refusal of one. +const forgePattern = "mesh-delivery-forge-bridge.forge.**" func main() { - operator := &asker.Client{Module: Module, State: stdio.State("asked")} h := &Holder{ Store: kvStore{}, Controller: seatController{ask: stdio.Ask}, Forge: toolForge{ask: stdio.Ask}, Emit: func(event string, body any) error { return stdio.Emit(event, body) }, Logf: logf, - Operator: operator, + Operator: kvOperator{}, + Merges: kvMerges{}, + Source: kvSource{}, 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{}, @@ -53,7 +57,7 @@ func main() { h.OnHold = func() { go readOnHolding(h) } h.init() l := &listening{now: "not yet: starting"} - go run(h, l, operator) + go run(h, l) if err := stdio.Serve("", tools(h, l)); err != nil { logf("%v", err) os.Exit(1) @@ -65,7 +69,7 @@ func main() { // holds the seat (SeatChanged), and what it may have missed meanwhile is read once then (readOnHolding): the controller's // walks, the forge's open pull requests, the operator's asks — and it acts on events from then on (novox/hq ADR 0406 §1, // issue 975): no tick, no forge poll, no reconcile loop. Each step is retried, each failure said, never given up on quietly. -func run(h *Holder, l *listening, operator *asker.Client) { +func run(h *Holder, l *listening) { time.Sleep(500 * time.Millisecond) // Serve first: the state is reached through it go h.Work() go watchSeat(h, l) @@ -81,26 +85,17 @@ func run(h *Holder, l *listening, operator *asker.Client) { time.Sleep(wait) } } - // The operator's answers: the router's `decided` event, in place of reading every pending ask each tick. - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - err := operator.Listen( - func(_ asks.Option, w asks.Warrant) error { h.AskDecided(w.Ask, "answered", &w); h.Poke(); return nil }, - func(handled asker.Handled, w asks.Warrant, err error) { - if err != nil { - logf("[mesh-delivery] the answer to the ask %s: %s: %v", w.Ask, handled, err) - } - if handled == asker.Refused || handled == asker.Recorded { - h.AskDecided(w.Ask, string(handled), &w) - h.Poke() - } - }) - if err == nil { - break + // The teller's answers: a change of its bucket asks for a look, which reads the answers to the asks pending. + go func() { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + err := stdio.State(sdkdelivery.TellerModule+".answers").Watch("", func(stdio.StateChange) error { h.AnswersChanged(); return nil }) + if err == nil { + return + } + logf("[mesh-delivery] not hearing the teller's answers yet (%v); asking again in %s", err, wait) + time.Sleep(wait) } - l.set("not yet: " + err.Error()) - logf("[mesh-delivery] not hearing the operator's answers yet (%v); asking again in %s", err, wait) - time.Sleep(wait) - } + }() logf("[mesh-delivery] listening: %s", strings.Join(followed, ", ")) } @@ -134,6 +129,10 @@ func handle(h *Holder, e stdio.Envelope) error { } func take(h *Holder, e stdio.Envelope) error { + if strings.HasPrefix(e.Key, "mesh-delivery-forge-bridge.forge.") { + h.Forged(e.Key, e.Body) + return nil + } switch e.Key { case "gitea.pull.updated", "gitea.pull.merged", "gitea.pull.closed": var p PullEvent 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 cafb2c4a..b1ca5f0f 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 @@ -5,6 +5,7 @@ import ( "os" "path/filepath" "reflect" + "slices" "sort" "strings" "testing" @@ -71,8 +72,8 @@ func TestItClaimsTheDeliverySeatAndFollowsWhatItHandles(t *testing.T) { if len(m.Claims[0].Serves) != 13 { t.Fatalf("serves %d verbs, wanted 13: %v", len(m.Claims[0].Serves), m.Claims[0].Serves) } - if !reflect.DeepEqual(m.Consumes, append(append([]string{}, followed...), "mesh-delivery-forge-bridge.forge.**")) { - t.Fatalf("consumes %v, follows %v and the bridge's forge messages", m.Consumes, followed) + if !reflect.DeepEqual(m.Consumes, followed) || !slices.Contains(m.Consumes, "mesh-delivery-forge-bridge.forge.**") { + t.Fatalf("consumes %v, follows %v (the bridge's forge messages among them)", m.Consumes, followed) } if !reflect.DeepEqual(m.Emits, []string{"transition", "group", "merged"}) { t.Fatalf("emits %v", m.Emits) diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go new file mode 100644 index 00000000..0ec30658 --- /dev/null +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/merges.go @@ -0,0 +1,406 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "reflect" + "sort" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The gate merges through the forge bridge, never itself (novox/hq ADR 0437 §8, ADR 0441 §5, ADR 0450 §3). It writes into +// `merges.` the Approve's heads and ONE member at a time, the next only after the bridge's message on FORGE says the +// one before it merged, and none after a refusal. It re-reads every member's required statuses before it hands a member +// (ADR 0441 §2). Once every member is merged it moves `source.main` in one compare-and-set write and the mesh's source has +// the group as one revision. What it announces for each merge (`merged`) names the ask, its asker the teller, and the heads. + +// MergeStore is the gate's bucket merges. +type MergeStore interface { + Get(group string) (*sdk.Merge, error) + Put(group string, m sdk.Merge) error +} + +// SourceStore is the gate's bucket source, one key, moved by compare-and-set. +type SourceStore interface { + // Move reads source.main and writes what update makes of it, unless update says nothing is to be written; a write + // another writer won is read again and tried again. + Move(update func(prev *sdk.Source) (sdk.Source, bool)) error +} + +// kvMerges is merges over the runtime. +type kvMerges struct{} + +func (kvMerges) Get(group string) (*sdk.Merge, error) { + key, err := sdk.MergesKey(group) + if err != nil { + return nil, err + } + e, err := stdio.State("merges").Get(key) + if err != nil || e == nil { + return nil, err + } + m, err := sdk.DecodeMerge(e.Value) + if err != nil { + return nil, fmt.Errorf("the merge of %s is not readable: %w", group, err) + } + return &m, nil +} + +func (kvMerges) Put(group string, m sdk.Merge) error { + key, err := sdk.MergesKey(group) + if err != nil { + return err + } + raw, err := sdk.EncodeMerge(m) + if err != nil { + return err + } + _, err = stdio.State("merges").Put(key, json.RawMessage(raw)) + return err +} + +// kvSource is source over the runtime. +type kvSource struct{} + +func (kvSource) Move(update func(prev *sdk.Source) (sdk.Source, bool)) error { + state := stdio.State("source") + for range changeAttempts { + e, err := state.Get(sdk.SourceMainKey) + if err != nil { + return err + } + var prev *sdk.Source + if e != nil { + p, err := sdk.DecodeSource(e.Value) + if err != nil { + return fmt.Errorf("source.main is not readable: %w", err) + } + prev = &p + } + next, write := update(prev) + if !write { + return nil + } + raw, err := sdk.EncodeSource(next) + if err != nil { + return err + } + if e == nil { + _, err = state.Create(sdk.SourceMainKey, json.RawMessage(raw)) + } else { + _, err = state.Update(sdk.SourceMainKey, json.RawMessage(raw), e.Revision) + } + if err == nil { + return nil + } + if !errors.Is(err, stdio.ErrChanged) { + return err + } + } + return fmt.Errorf("source.main was written by another writer first %d times: %w", changeAttempts, stdio.ErrChanged) +} + +// approvedOf are the heads an Approve covers as a merge carries them: `@` to the full head. +func (h *Holder) approvedOf(heads map[string]string) map[string]string { + out := map[string]string{} + for id, sha := range heads { + if m := h.deliveries[id]; m != nil { + out[sdk.ApprovedKey(m.Repository, m.Number)] = sha + } + } + return out +} + +// handTurn is d's turn to be handed to the bridge, when it is: the first member of its group not merged (a lone pull request +// is a group of one), every member approved at its head and ready to merge. Nil when it is not. Under the lock; the hand +// runs outside it. +func (h *Holder) handTurn(g *Group, d *Delivery, now time.Time) func() { + var reads []memberRead + c := h.covering(d) + if g != nil { + order := h.unmergedLive(g) + if len(order) == 0 || order[0] != d { + return nil + } + for _, m := range order { + mc := h.covering(m) + if m.Ready == nil || m.Ready.Declined != nil || mc == nil || mc.Approved == nil || mc.Heads[m.ID] != m.Commit || h.readyWaits(m) != "" { + return nil // the others are not approved at their heads yet: nothing is merged before they are + } + reads = append(reads, readOf(m, mc.Heads[m.ID])) + } + } + if c == nil || c.Approved == nil || c.Heads[d.ID] == "" { + return nil + } + d.Ready.Tries++ + d.Ready.LastTry, d.Ready.merging = now, true + h.keep(d) + gid, name, key, target, approved := "", askGroup(d), c.AskID, d.ID, h.approvedOf(c.Heads) + if g != nil { + gid = g.ID + } + return func() { h.hand(gid, name, key, target, reads, approved) } +} + +// hand reads a group's required statuses and, when all are a success, writes the target as the member being merged. +func (h *Holder) hand(gid, name, key, target string, reads []memberRead, approved map[string]string) { + if gid != "" && !h.groupMergeable(gid, target, reads) { + return + } + h.mu.Lock() + value, ok := h.mergeValue(name, key, approved, target, sdk.MergeMerging) + h.mu.Unlock() + if !ok { + h.giveBack(target, nil) + return + } + cur, err := h.Merges.Get(name) + if err == nil && cur != nil && sdk.SameMerge(*cur, value) { + return // handed already, the bridge has the revision + } + if err == nil { + err = h.Merges.Put(name, value) + } + if err != nil { + h.giveBack(target, err) + return + } + h.Logf("[mesh-delivery] %s is handed to the forge bridge to merge, at the heads the operator approved", target) +} + +// giveBack gives a target its turn back after a hand that could not be made, to be asked again after a while. +func (h *Holder) giveBack(target string, err error) { + h.mu.Lock() + defer h.mu.Unlock() + if d := h.deliveries[target]; d != nil && d.Ready != nil { + d.Ready.merging = false + d.Ready.LastTry = time.Time{} + } + if err != nil { + h.Logf("[mesh-delivery] %s could not be handed to the forge bridge: %v; asked again in %s", target, err, askAgainAfter) + } + h.again("a merge that could not be handed to the forge bridge") +} + +// mergeValue is the value of a group's merges key with these members merged, and the target in the state asked (merging, or +// refused with the forge's words): the group's members in order, those the Approve covers. Under the lock. False when it +// cannot be made (the target is not among the heads the Approve covers). +func (h *Holder) mergeValue(name, key string, approved map[string]string, target string, state sdk.MergeState) (sdk.Merge, bool) { + t := h.deliveries[target] + if t == nil { + return sdk.Merge{}, false + } + members := []*Delivery{t} + if g := h.groups[t.Group]; t.Group != "" && g != nil { + members = h.membersOf(g) + } + m := sdk.Merge{Group: name, Ask: key, Approved: approved} + for _, o := range members { + head := approved[sdk.ApprovedKey(o.Repository, o.Number)] + if head == "" { + continue + } + pull := sdk.MergePull{Head: sdk.Head{Repository: o.Repository, Number: o.Number, Head: head}} + switch { + case o.MergedAs != "" || (o.Ready != nil && o.Ready.Merged != nil): + commit := o.MergedAs + if o.Ready != nil && o.Ready.MergeCommit != "" { + commit = o.Ready.MergeCommit + } + if commit == "" { + continue + } + pull.State, pull.Commit = sdk.MergeMerged, commit + case o == t: + pull.State = state + if state == sdk.MergeRefused { + pull.Words = o.Ready.Answer + } + m.Pulls = append(m.Pulls, pull) + return m, true + default: + return m, false + } + m.Pulls = append(m.Pulls, pull) + } + return m, len(m.Pulls) > 0 && state == sdk.MergeMerged +} + +// Forged takes a message of the forge bridge on FORGE: the gate follows a merge and the refusal of a merge, and nothing else. +func (h *Holder) Forged(event string, body []byte) { + f, err := sdk.DecodeForge(body) + if err != nil { + h.Refuse(fmt.Errorf("a message of the forge bridge is not readable: %v", err)) + return + } + switch { + case f.Kind == sdk.ForgeMerged: + h.ForgeMerged(f) + case f.Kind == sdk.ForgeRefused && f.Act == "merge": + h.ForgeRefused(f) + } +} + +// memberOfForge is the delivery a forge message is about: at the head it names. +func (h *Holder) memberOfForge(f sdk.Forge) *Delivery { + if d := h.deliveries[IDOf(f.Repository, f.Head)]; d != nil && d.Number == f.Number { + return d + } + return nil +} + +// ForgeMerged takes the bridge's word that a pull request the gate handed to it merged: the merge is announced once, +// with the ask, its asker the teller and the heads (issue 1025), and the next member's turn comes at the next look. +func (h *Holder) ForgeMerged(f sdk.Forge) { + h.mu.Lock() + defer h.mu.Unlock() + d := h.memberOfForge(f) + if d == nil || d.Ready == nil || d.Ready.Approved == nil || d.Ready.Merged != nil { + return + } + r, now := d.Ready, h.Now() + d.Owed = append(d.Owed, Effect{Kind: EffectEmit, Event: "merged", Line: string(mustJSON(mergedEvent(d, f.Head))), Since: now}) + r.Merged, r.Answer, r.RefusedSince, r.merging, r.MergeCommit = &now, "merged", nil, false, f.Merge + h.Logf("[mesh-delivery] %s merged on the operator's Approve at %s, its check passed against its trunk's head", d.ID, shortOf(f.Head, 12)) + h.keep(d) +} + +// ForgeRefused takes the bridge's word that the forge refused a merge: nothing after it is merged, and the refusal is said +// as a stalled line (ADR 0441 §3, §4). The merge is handed again only at the next event that decides whether it may +// merge (mergeAgainNow). +func (h *Holder) ForgeRefused(f sdk.Forge) { + h.mu.Lock() + defer h.mu.Unlock() + d := h.memberOfForge(f) + if d == nil || d.Ready == nil || d.Ready.Merged != nil { + return + } + r, now := d.Ready, h.Now() + r.merging, r.refusedNoted = false, false + answer := oneLine(f.Words) + if r.RefusedSince == nil { + r.RefusedSince = &now + } + if !outOfDateRefusal(answer) { + r.Urgent = true // never a warning again for this head + } + if answer != r.Answer { + h.Logf("[mesh-delivery] %s is approved, and the merge tool did not merge it (asked again at the next verdict, Approve or composed check): %s", d.ID, + answer) + } + r.Answer = answer + h.keep(d) +} + +// settleMerges works out, under the lock, what a merge owes its key and the source once the bridge has spoken: a refused +// member written refused, and a group whose members are all merged written merged and moved onto source.main in one write. +// The writing happens outside the lock. +func (h *Holder) settleMerges() []func() { + var out []func() + done := map[string]bool{} + ids := make([]string, 0, len(h.deliveries)) + for id := range h.deliveries { + ids = append(ids, id) + } + sort.Strings(ids) + for _, id := range ids { + d := h.deliveries[id] + r := d.Ready + if r == nil || r.Approved == nil || r.AskID == "" || len(r.Heads) == 0 { + continue + } + name, key, approved := askGroup(d), r.AskID, h.approvedOf(r.Heads) + switch { + case r.Merged == nil && r.RefusedSince != nil && !r.refusedNoted && r.Answer != "" && r.Answer != "merged": + r.refusedNoted = true + target := d.ID + out = append(out, func() { + h.mu.Lock() + value, ok := h.mergeValue(name, key, approved, target, sdk.MergeRefused) + h.mu.Unlock() + if ok { + if err := h.Merges.Put(name, value); err != nil { + h.Logf("[mesh-delivery] the refusal of %s could not be written to its merge: %v", target, err) + } + } + }) + case r.Merged != nil && !r.SourceMoved && !done[key]: + if !h.allMerged(r.Heads) { + continue + } + done[key] = true + heads := map[string]string{} + var members []string + for hid := range r.Heads { + m := h.deliveries[hid] + commit := m.MergedAs + if m.Ready != nil && m.Ready.MergeCommit != "" { + commit = m.Ready.MergeCommit + } + heads[m.Repository] = commit + members = append(members, hid) + } + sort.Strings(members) + last := h.deliveries[members[len(members)-1]] + out = append(out, func() { h.finish(name, key, approved, last.ID, heads, members) }) + } + } + return out +} + +// allMerged is whether every delivery an Approve covers is merged. +func (h *Holder) allMerged(heads map[string]string) bool { + for id := range heads { + m := h.deliveries[id] + if m == nil || (m.MergedAs == "" && (m.Ready == nil || m.Ready.Merged == nil)) { + return false + } + } + return true +} + +// finish writes a group's merge as all merged and moves source.main to the merge commits in one compare-and-set write: the +// group is one revision of the mesh's source (ADR 0437 §1, §8). +func (h *Holder) finish(name, key string, approved map[string]string, last string, heads map[string]string, members []string) { + h.mu.Lock() + value, ok := h.mergeValue(name, key, approved, last, sdk.MergeMerged) + h.mu.Unlock() + var err error + if ok { + err = h.Merges.Put(name, value) + } + if err == nil { + err = h.Source.Move(func(prev *sdk.Source) (sdk.Source, bool) { + next := sdk.Source{Group: name, Heads: map[string]string{}} + if prev != nil { + for repository, head := range prev.Heads { + next.Heads[repository] = head + } + } + for repository, commit := range heads { + next.Heads[repository] = commit + } + return next, prev == nil || !reflect.DeepEqual(prev.Heads, next.Heads) + }) + } + h.mu.Lock() + defer h.mu.Unlock() + if err != nil { + h.Logf("[mesh-delivery] group %s merged, and its place in the source cannot be written yet: %v; asked again in %s", name, err, askAgainAfter) + h.again("source.main, moved once every member is merged") + return + } + for _, id := range members { + if m := h.deliveries[id]; m != nil && m.Ready != nil { + m.Ready.SourceMoved = true + h.keep(m) + } + } + h.Logf("[mesh-delivery] group %s is merged whole: source.main holds %s", name, strings.Join(members, ", ")) +} diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/operator_test.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/operator_test.go index 6c312b21..6bc6d482 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/operator_test.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/operator_test.go @@ -1,101 +1,110 @@ package main import ( + "fmt" "sync" "time" "git.novox.be/novox/mesh-sdk/go/asks" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" ) -// fakeOperator is the operator channel: it keeps the asks made, and answers as the test says. In auto mode it -// approves every ask the moment it is read (the tests of other behaviour, which are not about the Approve). +// fakeOperator is the gate's bucket asks and the teller behind it: it keeps the asks written, and answers as the test says +// (the teller's answer under the ask's key). In auto mode it approves every ask the moment it is read (the tests of other +// behaviour, which are not about the Approve). An ask is known here by its key (mesh-sdk go/delivery MergeAskKey), as the +// delivery keeps it. type fakeOperator struct { mu sync.Mutex now func() time.Time auto bool asked []asks.Ask + byKey map[string]sdk.Ask + order []string + seen map[string]bool cancelled []string state map[string]string - warrant map[string]*asks.Warrant + answers map[string]*sdk.Answer refuse error } func newFakeOperator(now func() time.Time) *fakeOperator { - return &fakeOperator{now: now, state: map[string]string{}, warrant: map[string]*asks.Warrant{}} + return &fakeOperator{now: now, byKey: map[string]sdk.Ask{}, seen: map[string]bool{}, state: map[string]string{}, answers: map[string]*sdk.Answer{}} } -func (o *fakeOperator) Ask(a asks.Ask) error { +func (o *fakeOperator) Ask(key string, a sdk.Ask) error { o.mu.Lock() defer o.mu.Unlock() if o.refuse != nil { return o.refuse } - if err := a.Check(o.now()); err != nil { + if err := a.Ask.Check(o.now()); err != nil { return err } - o.asked = append(o.asked, a) - o.state[a.ID] = "open" + if _, err := sdk.EncodeAsk(a); err != nil { + return err + } + if o.state[key] == "open" && sdk.SameAsk(o.byKey[key], a) { + return nil // the same ask is there, as kvOperator leaves it + } + o.asked = append(o.asked, a.Ask) + o.byKey[key] = a + if _, known := o.seen[key]; !known { + o.order = append(o.order, key) + o.seen[key] = true + } + o.state[key] = "open" + delete(o.answers, key) return nil } -func (o *fakeOperator) Cancel(id string) error { +func (o *fakeOperator) Cancel(key string) error { o.mu.Lock() defer o.mu.Unlock() - o.cancelled = append(o.cancelled, id) - o.state[id] = "cancelled" + o.cancelled = append(o.cancelled, key) + o.state[key] = "cancelled" return nil } -func (o *fakeOperator) Where(id string) (string, *asks.Warrant, error) { +func (o *fakeOperator) Where(key string) (*sdk.Answer, error) { o.mu.Lock() defer o.mu.Unlock() - if o.auto && o.state[id] == "open" { - o.answerLocked(id, OptionApprove) + if o.auto && o.state[key] == "open" { + o.answerLocked(key, OptionApprove) } - return o.state[id], o.warrant[id], nil + return o.answers[key], nil } -func (o *fakeOperator) ask(id string) asks.Ask { - for _, a := range o.asked { - if a.ID == id { - return a - } +// ask is the router's ask at a key, as the gate composed it. +func (o *fakeOperator) ask(key string) asks.Ask { + if a, ok := o.byKey[key]; ok { + return a.Ask } - panic("no ask " + id) + panic("no ask " + key) } -func (o *fakeOperator) answerLocked(id, option string) { - a := o.ask(id) +func (o *fakeOperator) answerLocked(key, option string) { + a := o.ask(key) opt, _ := a.Option(option) - o.state[id] = "answered" - o.warrant[id] = &asks.Warrant{Ask: id, Asker: Module, Outcome: asks.OutcomeChosen, Option: option, Label: opt.Label, - Level: opt.Level, By: &asks.Person{Who: asks.Operator, Kind: "telegram", Verified: "user id verified"}, - At: o.now(), AskDigest: a.Digest()} + o.state[key] = "answered" + o.answers[key] = &sdk.Answer{Option: option, Warrant: fmt.Sprintf("%s@%s", a.ID, o.now().UTC().Format(time.RFC3339)), + Words: "the operator, via telegram, chose " + opt.Label} } -// answer is the operator's choice, proven by the router. -func (o *fakeOperator) answer(id, option string) { +// answer is the operator's choice, proven to the teller by the router and written by it. +func (o *fakeOperator) answer(key, option string) { o.mu.Lock() defer o.mu.Unlock() - o.answerLocked(id, option) + o.answerLocked(key, option) } -// end is an ask that closed without a choice. -func (o *fakeOperator) end(id string) { - o.mu.Lock() - defer o.mu.Unlock() - o.state[id] = "expired" - o.warrant[id] = &asks.Warrant{Ask: id, Asker: Module, Outcome: asks.OutcomeExpired, At: o.now()} -} - -// open is the ids of the asks waiting for an answer, in the order asked. +// open is the keys of the asks waiting for an answer, in the order asked. func (o *fakeOperator) open() []string { o.mu.Lock() defer o.mu.Unlock() var out []string - for _, a := range o.asked { - if o.state[a.ID] == "open" { - out = append(out, a.ID) + for _, k := range o.order { + if o.state[k] == "open" { + out = append(out, k) } } return out 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 7ca17df5..c4f1041d 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ports.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ports.go @@ -85,11 +85,6 @@ type Forge interface { // OpenPulls is every open pull request, with what its base requires and what its head has (novox/hq issue // 347). Reads only. OpenPulls() ([]OpenPull, error) - // Merge asks the forge's merge tool to merge a pull request at the head named (gitea_merge_pull_request, novox/hq - // issue 722): the one path every merge takes. approved are the heads the operator's Approve covers, each - // `owner/repo@` (ADR 0421 §9); the tool refuses a pull request whose head differs. Its refusal is the - // error, in its words. - Merge(owner, repo string, number int, head string, approved []string) error // Comment adds a comment to a pull request. Comment(owner, repo string, number int, body string) error } @@ -386,11 +381,6 @@ func (f toolForge) Statuses(owner, repo string, number int, commit, base string) return &c, nil } -func (f toolForge) Merge(owner, repo string, number int, head string, approved []string) error { - return f.tool("gitea_merge_pull_request", map[string]any{"owner": owner, "repo": repo, "number": number, "head": head, - "approved_heads": approved}) -} - func (f toolForge) Comment(owner, repo string, number int, body string) error { return f.tool("gitea_add_comment", map[string]any{"owner": owner, "repo": repo, "number": number, "body": body}) } diff --git a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go index 4a8a6fcd..f5ff6be6 100644 --- a/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go +++ b/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/ready.go @@ -7,6 +7,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/asks" + sdk "git.novox.be/novox/mesh-sdk/go/delivery" ) // The Approve is asked as soon as the checks pass (novox/hq ADR 0421, issue 1022). @@ -57,8 +58,15 @@ type ReadyRun struct { Tries int `json:"tries,omitempty"` LastTry time.Time `json:"last_try"` Answer string `json:"answer,omitempty"` - // merging is whether a merge ask is on its way, outside the lock: not asked twice meanwhile. Not kept. + // merging is whether the member is handed to the forge bridge and not merged or refused yet: not handed twice + // meanwhile. Not kept. merging bool + // refusedNoted is whether the refusal was written into the group's merge key. Not kept. + refusedNoted bool + // MergeCommit is the commit the forge bridge reported the merge as, and SourceMoved whether source.main holds it + // (merges.go). + MergeCommit string `json:"merge_commit,omitempty"` + SourceMoved bool `json:"source_moved,omitempty"` // OwnAskedAt is when its own check was asked after the merge tool refused it for a pending status of its own (issue // 1049): once for the head this run is for. OwnAskedAt time.Time `json:"own_asked_at,omitempty"` @@ -205,18 +213,17 @@ func (h *Holder) readyAsks(now time.Time) []func() { if !r.LastTry.IsZero() { continue // refused: asked again at the next event that decides whether it may merge (mergeAgainNow), with no timer } - if g := h.groups[d.Group]; d.Group != "" && g != nil { - // A group is merged in one pass, in its order, and stops at the first refusal (issue 1275). - if a := h.groupPass(g, d, now); a != nil { - asks = append(asks, a) - } - continue + // A group is merged one member at a time, handed to the forge bridge, and the pass stops at the first refusal (ADR + // 0441 §3, §5); a lone pull request is a group of one. + g := h.groups[d.Group] + if d.Group == "" { + g = nil + } + if a := h.handTurn(g, d, now); a != nil { + asks = append(asks, a) } - r.Tries++ - r.LastTry, r.merging = now, true - h.keep(d) - asks = append(asks, h.askMerge(d.ID, r.Heads)) } + asks = append(asks, h.settleMerges()...) asks = append(h.cancels, asks...) h.cancels = nil return asks @@ -336,72 +343,6 @@ func (h *Holder) groupTurnToMerge(d *Delivery) string { return "" } -// askMerge asks the forge's merge tool to merge a head the operator approved, and keeps its answer. The tool is handed -// the heads the Approve covers, each `owner/repo@` (novox/hq ADR 0421 §9), and refuses a pull request whose -// head differs from the one approved. -func (h *Holder) askMerge(id string, heads map[string]string) func() { - return func() { h.mergeOne(id, heads) } -} - -// mergeOne asks the merge tool for one approved head and keeps its answer; whether it merged. -func (h *Holder) mergeOne(id string, heads map[string]string) bool { - h.mu.Lock() - d := h.deliveries[id] - if d == nil || d.Ready == nil { - h.mu.Unlock() - return false - } - owner, repo, number, head := d.Owner(), d.Repo(), d.Number, d.Commit - if sha := heads[id]; sha != "" { - head = sha - } - h.mu.Unlock() - err := h.Forge.Merge(owner, repo, number, head, approvedHeads(heads)) - h.mu.Lock() - defer h.mu.Unlock() - d = h.deliveries[id] - if d == nil || d.Ready == nil { - return false - } - r, now := d.Ready, h.Now() - r.merging = false - if err == nil { - // Issue 1025: the announcement is owed, kept with the delivery, so a death right after the merge - // still says it once. - d.Owed = append(d.Owed, Effect{Kind: EffectEmit, Event: "merged", Line: string(mustJSON(mergedEvent(d, head))), Since: now}) - r.Merged, r.Answer, r.RefusedSince = &now, "merged", nil - h.Logf("[mesh-delivery] %s merged on the operator's Approve at %s, its check passed against its trunk's head", id, - shortOf(head, 12)) - h.keep(d) - return true - } - answer := oneLine(err.Error()) - if r.RefusedSince == nil { - r.RefusedSince = &now - } - if !outOfDateRefusal(answer) { - r.Urgent = true // never a warning again for this head - } - if answer != r.Answer { - h.Logf("[mesh-delivery] %s is approved, and the merge tool did not merge it (asked again at the next verdict, Approve or composed check): %s", id, - answer) - } - r.Answer = answer - h.keep(d) - return false -} - -// approvedHeads are the heads an Approve covers as the merge tool is handed them: `owner/repo@`, sorted. -func approvedHeads(heads map[string]string) []string { - var out []string - for id, sha := range heads { - repository, _, _ := strings.Cut(id, "@") - out = append(out, repository+"@"+sha) - } - sort.Strings(out) - return out -} - // sayOn comments on a delivery's pull request. func (h *Holder) sayOn(d *Delivery, body string) func() { owner, repo, number, id := d.Owner(), d.Repo(), d.Number, d.ID @@ -468,6 +409,10 @@ func mergedEvent(d *Delivery, head string) map[string]any { "ask_id": "", "option": "", "heads": heads} if r.Approved != nil { ev["ask_id"], ev["option"] = r.AskID, OptionApprove + if _, _, _, err := sdk.ParseAskKey(r.AskID); err == nil { + // The ask of the gate: the router knows it by the id its key gives, and the teller asked it (ADR 0437 §6). + ev["ask_id"], ev["asker"] = sdk.AskID(r.AskID), sdk.TellerModule + } // an Approve carried over from today's module keeps its router id, and the asker the controller takes by default } return ev }