§716: the gate keeps a group's members from the seat git's pull events in deliveries and groups; the switch runs once (A-281)

The gate consumes pull-updated and invokes open-pulls and pull; it writes deliveries and groups
under their key forms, deletes them as a take-in or a closed pull request leaves them, copies a
delivery's base onto the record's member, answers groups from records and the group key together,
puts a missed event right at its start, and writes switch.done once the switch has run.

Builds: design 47 G §716, A-279, A-236, A-281

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