Ask the Approve through the gate's asks, hand merges to the bridge one member at a time, and move source.main once a group is merged (issue 1348)

This commit is contained in:
2026-10-20 00:45:02 +00:00
parent dd462a80e2
commit beafa55d97
17 changed files with 1220 additions and 443 deletions
@@ -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: "<id>=<sha>" 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
}
@@ -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)
}
@@ -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, `<repository>@<full sha>` 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) }
@@ -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 (`<module>.<name>` is the bucket `<module>_<name>`): 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.<module>.event.<name>, mesh.seat.<seat>.event.<name>).
if _, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "EVENTS",
Subjects: []string{"mesh.mod.gitea.event.>", "mesh.seat." + ControllerSeat + ".event.>", "mesh.seat." + asks.Seat + ".event.>"}}); 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"
@@ -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()
}
}
@@ -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) {
@@ -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()
@@ -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:
@@ -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
@@ -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)
@@ -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.<group>.<heads hash>: %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@<first 12 characters>` 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))
}
}
@@ -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
@@ -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)
@@ -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.<group>` 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: `<repository>@<number>` 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, ", "))
}
@@ -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
@@ -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@<full sha>` (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})
}
@@ -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@<full sha>` (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@<full sha>`, 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
}