Find the delivery seat's holder by its claim, and record each build's fingerprint inputs (design 47 R3b: issues 1210-1216) #394

Merged
mesh-tools merged 8 commits from feat/delivery-seat-by-claim into main 2026-10-19 10:28:30 +00:00
54 changed files with 2280 additions and 150 deletions
+1
View File
@@ -432,6 +432,7 @@ func answer(ctx context.Context, publisher builder.Publisher, on, workspace stri
}
result.Against = built.Against
result.SourceFingerprint = built.Source
result.SourceInputs = built.SourceInputs
// What the walk did instead of building, or why it built (novox/hq ADR 0350 §3, §4).
result.ReuseKey = built.ReuseKey
if built.Reused != nil {
+10 -5
View File
@@ -367,7 +367,7 @@ func buildsEvery(p inventory.Plan, modules []inventory.Entry) bool {
// ownPath says a merge touches a module on the controller's own path, whose walk waits for nobody's word.
func ownPath(touches []inventory.Entry) bool {
for _, e := range touches {
if _, own := onTheControllersPath[e.Manifest.Module]; own {
if onTheOwnPath(e.Manifest.Module, e.Manifest) {
return true
}
}
@@ -945,7 +945,11 @@ func cutUnrelated(ctx context.Context, open *stores, batches, started, waiting [
var read map[string][]inventory.ReadRepository
var edges []inventory.Edge
judged := false
openWalks, err := openWalksOf(ctx, inv, started)
onPath, err := ownPathNow(ctx, inv)
if err != nil {
return err
}
openWalks, err := openWalksOf(ctx, inv, started, onPath)
if err != nil {
return err
}
@@ -978,7 +982,7 @@ func cutUnrelated(ctx context.Context, open *stores, batches, started, waiting [
var fold, keep []inventory.Plan
if !b.OwnPath() {
for _, w := range waiting {
ww, err := openWalkOf(ctx, inv, w)
ww, err := openWalkOf(ctx, inv, w, onPath)
if err != nil {
return err
}
@@ -1013,7 +1017,7 @@ func cutUnrelated(ctx context.Context, open *stores, batches, started, waiting [
if !cut.Open() {
continue
}
w, err := openWalkOf(ctx, inv, cut)
w, err := openWalkOf(ctx, inv, cut, onPath)
if err != nil {
return err
}
@@ -1391,8 +1395,9 @@ func classOfModules(modules []string, entries []inventory.Entry) string {
}
}
}
onPath := ownPathOf(entries)
for _, m := range modules {
if _, own := onTheControllersPath[m]; own || resolvers[m] {
if _, own := onPath[m]; own || resolvers[m] {
return inventory.ClassCore
}
}
+1
View File
@@ -153,6 +153,7 @@ func buildFrom(result link.BuildResult) inventory.Build {
Path: result.Path,
// What it was made from (novox/hq issue 280): two builds with one are one build.
SourceFingerprint: result.SourceFingerprint,
SourceInputs: result.SourceInputs,
// And its reuse key (novox/hq ADR 0350 §1), which the chain names (ADR 0408 §1).
ReuseKey: result.ReuseKey,
}
@@ -317,7 +317,7 @@ func TestAModuleMovesOnTheOperatorsApproveOfTheMergeThatMovedIt(t *testing.T) {
keptDeliveryMerge = func(_ context.Context, owner, repo string, number int, head string) (deliveryMerge, bool, error) {
return a.merge, !c.unkept && owner == "novox" && repo == "mesh-host" && number == 7 && head == a.merge.HeadSHA, nil
}
routerAskOf = func(_ context.Context, _ catalogueReader, id string) (*asks.Ask, *asks.Warrant, error) {
routerAskOf = func(_ context.Context, _ catalogueReader, _, id string) (*asks.Ask, *asks.Warrant, error) {
if id != a.merge.AskID {
t.Errorf("%s: the router was asked for %q", c.what, id)
}
+11
View File
@@ -2,10 +2,12 @@ package main
import (
"bytes"
"reflect"
"strings"
"testing"
"time"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
@@ -66,3 +68,12 @@ func TestBuildsSaysTheSourceFingerprint(t *testing.T) {
t.Errorf("the other lines changed:\n%s\nwant\n%s", plain.String(), want)
}
}
// 1215/AC4: what the builder said of the inputs is what the controller keeps of the build.
func TestTheControllerKeepsTheInputsABuilderSaid(t *testing.T) {
in := &sdkdelivery.SourceInputs{Module: "mailu", Tree: "t", Toolchains: map[string]string{"code": "ts sha256:aa 01"}}
kept := buildFrom(link.BuildResult{ID: "build-1", Repository: "r", Commit: "c", SourceFingerprint: "src1:x", SourceInputs: in})
if !reflect.DeepEqual(kept.SourceInputs, in) {
t.Errorf("the controller keeps %+v, the builder said %+v", kept.SourceInputs, in)
}
}
+44
View File
@@ -0,0 +1,44 @@
package main
import (
"reflect"
"testing"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// 1212/AC1 and AC2: the check request names the pulls bucket of each module claiming the delivery seat, in the order
// of the claimants' names.
func TestACheckNamesThePullsBucketsOfTheSeatsClaimants(t *testing.T) {
for name, c := range map[string]struct {
entries []inventory.Entry
want []string
}{
"only the gate": {[]inventory.Entry{claimingTheSeat("mesh-delivery-gate"), namedOnly("mesh-delivery")}, []string{"mesh-delivery-gate_pulls"}},
"only today's": {[]inventory.Entry{claimingTheSeat("mesh-delivery"), namedOnly("gitea")}, []string{"mesh-delivery_pulls"}},
"both, name order": {[]inventory.Entry{claimingTheSeat("mesh-delivery-gate"), claimingTheSeat("mesh-delivery")}, []string{"mesh-delivery_pulls", "mesh-delivery-gate_pulls"}},
"none": {[]inventory.Entry{namedOnly("gitea")}, nil},
} {
if got := pullsBucketsOf(c.entries); !reflect.DeepEqual(got, c.want) {
t.Errorf("%s: %v, want %v", name, got, c.want)
}
}
}
// 1212/AC1 and AC2: the check a request carries names those buckets (the wiring, not only the helper).
func TestTheCheckRequestCarriesThePullsBuckets(t *testing.T) {
for name, c := range map[string]struct {
entries []inventory.Entry
want []string
}{
"only the gate": {[]inventory.Entry{claimingTheSeat("mesh-delivery-gate")}, []string{"mesh-delivery-gate_pulls"}},
"only today's": {[]inventory.Entry{claimingTheSeat("mesh-delivery")}, []string{"mesh-delivery_pulls"}},
"both, name order": {[]inventory.Entry{claimingTheSeat("mesh-delivery-gate"), claimingTheSeat("mesh-delivery")}, []string{"mesh-delivery_pulls", "mesh-delivery-gate_pulls"}},
} {
got := checkOf(link.PullUpdated{Owner: "novox", Repo: "a", Number: 7}, checkScope{}, c.entries, nil, nil, nil, nil)
if !reflect.DeepEqual(got.Pulls, c.want) || got.Owner != "novox" || got.Number != 7 {
t.Errorf("%s: %+v, want pulls %v", name, got, c.want)
}
}
}
+21 -4
View File
@@ -12,6 +12,7 @@ import (
"strings"
"time"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
@@ -305,13 +306,19 @@ func checkRequestFor(ctx context.Context, open *stores, p link.PullUpdated, scop
Held: heldBy(ctx),
Seats: seatBases(ctx),
Source: sourceOnSeat(source),
Check: &link.CheckRequest{Owner: p.Owner, Repo: p.Repo, Number: p.Number, Base: p.Base, Trunk: p.BaseSHA, Ready: p.Ready,
Paths: p.Paths, Beside: beside, Modules: scope.Modules, Dependents: scope.Dependents, New: scope.New,
Manifests: scope.Manifests, Judge: scope.Judge, Builds: builds, Named: namedForTheCheck(ctx, p, builds),
Running: runningBy(ctx)},
Check: checkOf(p, scope, entries, beside, builds, namedForTheCheck(ctx, p, builds), runningBy(ctx)),
}, nil
}
// checkOf is the check a request carries: the pull request, what the change touches, and the buckets of the pull
// requests' keys of the modules claiming the delivery seat (novox/hq ADR 0437 §3, issue 1212).
func checkOf(p link.PullUpdated, scope checkScope, entries []inventory.Entry, beside map[string]link.CheckedOut,
builds []link.CheckBuild, named []link.CheckedBuild, running map[string]string) *link.CheckRequest {
return &link.CheckRequest{Pulls: pullsBucketsOf(entries), Owner: p.Owner, Repo: p.Repo, Number: p.Number, Base: p.Base,
Trunk: p.BaseSHA, Ready: p.Ready, Paths: p.Paths, Beside: beside, Modules: scope.Modules, Dependents: scope.Dependents,
New: scope.New, Manifests: scope.Manifests, Judge: scope.Judge, Builds: builds, Named: named, Running: running}
}
// besideOf is what a check reads beside its own repository, by the directory it is found under: each core repository
// by the module the mesh builds from it, at besideRefs. running is the commit the mesh runs of a module.
func besideOf(entries []inventory.Entry, running func(module string) string,
@@ -661,3 +668,13 @@ func sayChecked(ctx context.Context, c link.Checked) {
fmt.Printf("%s: the verdict could not be said, so the pull request is not told it: %v\n", c.ID, err)
}
}
// pullsBucketsOf are the buckets of the pull requests' keys a check names for the build agent: one per module
// claiming the delivery seat, in the order of their names (novox/hq ADR 0437 §3, issue 1212).
func pullsBucketsOf(entries []inventory.Entry) []string {
var out []string
for _, m := range claimantsOf(entries, catalogue.DeliverySeat) {
out = append(out, broker.BucketName(m, "pulls"))
}
return out
}
+56 -17
View File
@@ -42,13 +42,53 @@ import (
// `delivery order` (a group's order from the graph), `delivery check` (a group's heads composed), `delivery
// go`, `delivery stop` and `delivery walks`.
// onTheControllersPath are the modules whose walk never waits for the delivery's owner, and what each is.
// onTheControllersPath are the modules, by name, whose walk never waits for the delivery's owner, and what each is.
// The delivery's owner is not among them: it is whichever module claims the seat (ownPathOf, novox/hq ADR 0437 §3).
var onTheControllersPath = map[string]string{
"mesh-controller": "the controller",
"mesh-host": "the node-engine",
"node-tools": "the node tools",
"nats": "the bus",
catalogue.DeliverySeat: "the delivery's owner",
"mesh-controller": "the controller",
"mesh-host": "the node-engine",
"node-tools": "the node tools",
"nats": "the bus",
}
// claimantsOf are the modules among the catalogued entries that claim a seat, each once, in name order.
func claimantsOf(entries []inventory.Entry, seat string) []string {
var out []string
for _, e := range entries {
if e.Manifest.ClaimsSeat(seat) && !slices.Contains(out, e.Manifest.Module) {
out = append(out, e.Manifest.Module)
}
}
sort.Strings(out)
return out
}
// ownPathOf is the controller's own path among these entries: the modules by name, and the modules that claim
// the delivery seat, which is how the delivery's owner is found (novox/hq ADR 0437 §3).
func ownPathOf(entries []inventory.Entry) map[string]string {
own := make(map[string]string, len(onTheControllersPath)+1)
for m, what := range onTheControllersPath {
own[m] = what
}
for _, m := range claimantsOf(entries, catalogue.DeliverySeat) {
own[m] = "the delivery's owner"
}
return own
}
// onTheOwnPath is whether one module, with its manifest at hand, is on the controller's own path.
func onTheOwnPath(module string, m catalogue.Manifest) bool {
_, named := onTheControllersPath[module]
return named || m.ClaimsSeat(catalogue.DeliverySeat)
}
// ownPathNow is ownPathOf the catalogue as it is.
func ownPathNow(ctx context.Context, inv *inventory.Inventory) (map[string]string, error) {
entries, err := inv.Catalogued(ctx)
if err != nil {
return nil, err
}
return ownPathOf(entries), nil
}
// deliverySeatHeld is whether a module claiming the delivery seat is assigned somewhere: the seat has a
@@ -69,8 +109,9 @@ func awaitsFor(entries []inventory.Entry, modules []string) *inventory.PlanDeliv
if !deliverySeatHeld(entries) {
return nil
}
onPath := ownPathOf(entries)
for _, m := range modules {
if _, own := onTheControllersPath[m]; own {
if _, own := onPath[m]; own {
return nil
}
}
@@ -327,16 +368,17 @@ func stopInBatch(ctx context.Context, inv *inventory.Inventory, b *inventory.Pla
// the stop's one act. The kind is read from what the merges moved; the stopped walk itself is no walk a cut folds.
func batchToWalkAgain(ctx context.Context, inv *inventory.Inventory, stopping string, again []inventory.BatchedMerge,
now time.Time) (inventory.Plan, []*inventory.Plan, []inventory.MergeMove, error) {
own := false
for _, n := range movesOf(again) {
if _, on := onTheControllersPath[n]; on {
own = true
}
}
entries, read, edges, err := theGraph(ctx, inv)
if err != nil {
return inventory.Plan{}, nil, nil, err
}
own := false
onPath := ownPathOf(entries)
for _, n := range movesOf(again) {
if _, on := onPath[n]; on {
own = true
}
}
joins, _, _, err := relatedBatches(ctx, inv, again, own, stopping, entries, read, edges)
if err != nil {
return inventory.Plan{}, nil, nil, err
@@ -935,7 +977,7 @@ func deliveryCommand(ctx context.Context, args []string) error {
walks[i].Phases = walks[i].WalkPhases(now, appliedFrom(ctx, inv))
}
return answer(map[string]any{"held": deliverySeatHeld(entries), "walks": walks,
"own-path": sortedKeysOf(ownPathWords())})
"own-path": sortedKeysOf(ownPathOf(entries))})
}
return fmt.Errorf("delivery %s: plan, order, check, go, stop, backups-again or walks", sub)
}
@@ -1006,9 +1048,6 @@ func recordWalkPhases(ctx context.Context, inv *inventory.Inventory, now time.Ti
return nil
}
// ownPathWords is the controller's own path as words, for an answer.
func ownPathWords() map[string]string { return onTheControllersPath }
// theGraph is what the planner reads.
func theGraph(ctx context.Context, inv *inventory.Inventory) ([]inventory.Entry, map[string][]inventory.ReadRepository,
[]inventory.Edge, error) {
@@ -51,7 +51,7 @@ func TestTheApprovedFirstBuildPassesTheGuardOfAVerbAndIsQueued(t *testing.T) {
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
return a.merge, true, nil
}
routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) {
routerAskOf = func(context.Context, catalogueReader, string, string) (*asks.Ask, *asks.Warrant, error) {
return &a.ask, &a.warrant, nil
}
t.Cleanup(func() { keptDeliveryMerge, routerAskOf = wasKept, wasAsk })
@@ -94,7 +94,7 @@ func TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegis
following := following{open: open}
server := link.ConnectNats(js, nil, noListener{})
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, kept: following.resumeFirstBuilds}); err != nil {
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: theDeliveryClaims, kept: following.resumeFirstBuilds}); err != nil {
t.Fatal(err)
}
if err := broker.AssertServingConsumers(js); err != nil {
@@ -110,7 +110,7 @@ func TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegis
word := func() {
t.Helper()
body, _ := json.Marshal(a.merge)
if _, err := js.Context().Publish(broker.DeliveryMergedSubject, body); err != nil {
if _, err := js.Context().Publish(broker.DeliveryMergedSubjectOf(theDelivery), body); err != nil {
t.Fatal(err)
}
becomes(t, 15*time.Second, "the word being kept", func() bool {
+1 -1
View File
@@ -58,7 +58,7 @@ func TestAFirstMergeIntoANewRepositoryAsksItsModulesBuildsOnlyOnTheApprove(t *te
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
return a.merge, !c.unkept, nil
}
routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) {
routerAskOf = func(context.Context, catalogueReader, string, string) (*asks.Ask, *asks.Warrant, error) {
return &a.ask, &a.warrant, nil
}
dirs, _, err := firstBuildsOf(t.Context(), nil, m, entries, nil)
+2 -1
View File
@@ -23,6 +23,7 @@ func TestAChangePlanSaysTheClassOfWhatItMoves(t *testing.T) {
resolver := entry("dnsmasq", "modules/dnsmasq")
resolver.Manifest.Claims = []catalogue.Claim{{Name: resolverSeat}}
delivery := entry(catalogue.DeliverySeat, "modules/mesh-delivery")
delivery.Manifest.Claims = []catalogue.Claim{{Name: catalogue.DeliverySeat}}
webApp := entry("photos", "modules/photos")
entries := []inventory.Entry{gitea, resolver, delivery, webApp}
// photos stands on gitea: a change to gitea moves photos as its dependent.
@@ -44,7 +45,7 @@ func TestAChangePlanSaysTheClassOfWhatItMoves(t *testing.T) {
{"a leaf and a core module together: the slowest-budgeted", []string{"modules/gitea/index.ts", "modules/dnsmasq/module.json"}, inventory.ClassCore},
{"nothing any module's build reads", []string{"README.md"}, "none"},
{"a new leaf module alone", []string{"modules/weather/module.json"}, inventory.ClassLeaf},
{"a new module on the controller's own path, by its own class", []string{"next/mesh-delivery/module.json"}, inventory.ClassCore},
{"a new module on the controller's own path, by its own class", []string{"next/mesh-host/module.json"}, inventory.ClassCore},
{"a new leaf module beside a core one moved", []string{"modules/weather/module.json", "modules/dnsmasq/module.json"}, inventory.ClassCore},
} {
if got := plan(c.paths...).Class; got != c.want {
+2 -2
View File
@@ -69,7 +69,7 @@ var (
// own path, or one holding the mesh's resolver. Every other module is a leaf. Its class sets the module's
// delivery budget; how many readings its gate takes is gateRuleOf's.
func coreModule(module string, m catalogue.Manifest) bool {
if _, own := onTheControllersPath[module]; own {
if onTheOwnPath(module, m) {
return true
}
for _, c := range m.Claims {
@@ -106,7 +106,7 @@ func gateRuleOf(module string, m catalogue.Manifest) gateRule {
case coreComponent(module) == lease.ComponentNodeTools:
return gateRule{passes: 1, witnessed: true, said: "the tool runner's rule: one reading once it answers, " +
"after the node-engine's witness window (novox/hq ADR 0351)"}
case module == catalogue.DeliverySeat:
case m.ClaimsSeat(catalogue.DeliverySeat):
return gateRule{passes: 1, said: "mesh-delivery's rule: one reading (novox/hq ADR 0351)"}
case coreModule(module, m):
return gateRule{passes: gatePasses, within: gateWithin, said: fmt.Sprintf("the core's rule: %d readings, "+
+12 -3
View File
@@ -39,7 +39,7 @@ func TestEachModuleHasItsGateRule(t *testing.T) {
}{
{"mesh-controller", nil, 1, 0, true, "the controller's rule"},
{"node-tools", nil, 1, 0, true, "the tool runner's rule"},
{catalogue.DeliverySeat, nil, 1, 0, false, "mesh-delivery's rule"},
{catalogue.DeliverySeat, []catalogue.Claim{{Name: catalogue.DeliverySeat}}, 1, 0, false, "mesh-delivery's rule"},
{"mesh-host", nil, 3, 2 * time.Minute, false, "the core's rule"},
{"dnsmasq", []catalogue.Claim{{Name: resolverSeat}}, 3, 2 * time.Minute, false, "the core's rule"},
{"nats", nil, 3, 2 * time.Minute, false, "the core's rule"},
@@ -56,7 +56,12 @@ func TestEachModuleHasItsGateRule(t *testing.T) {
// Every module that was core stays core for its delivery budget and its walk: only its readings change.
func TestTheCoreStaysTheCore(t *testing.T) {
for _, m := range []string{"mesh-controller", "mesh-host", "node-tools", "nats", catalogue.DeliverySeat} {
if !coreModule(m, catalogue.Manifest{Module: m}) {
// The delivery's owner is core by the claim of the seat, not by its name (novox/hq ADR 0437 §3, issue 1210).
claims := []catalogue.Claim(nil)
if m == catalogue.DeliverySeat {
claims = []catalogue.Claim{{Name: catalogue.DeliverySeat}}
}
if !coreModule(m, catalogue.Manifest{Module: m, Claims: claims}) {
t.Errorf("%s is no longer judged as the core for its class", m)
}
}
@@ -78,7 +83,11 @@ func aGateOn(module, node string, since time.Time) *gateOn {
}
func (o *gateOn) read(at time.Time, h health, why string) {
r := gateRuleOf(o.module, catalogue.Manifest{Module: o.module})
m := catalogue.Manifest{Module: o.module}
if o.module == catalogue.DeliverySeat { // today's module, which claims the seat
m.Claims = []catalogue.Claim{{Name: catalogue.DeliverySeat}}
}
r := gateRuleOf(o.module, m)
floor := gateFloor(o.module, r, *o.g.Since, []string{o.node}, o.g.Applied, o.holder)
tallyGate(o.g, gateJudging{modules: []string{o.module}, reading: map[string]health{o.module: h},
worst: h, why: why, rules: map[string]gateRule{o.module: r}, floors: map[string]time.Time{o.module: floor}}, at)
@@ -0,0 +1,37 @@
package main
import (
"context"
"errors"
"testing"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
// 1211/AC1 to AC3: the controller keeps a merge announced by a module that claims the delivery seat, whatever its
// name, and takes no word from one that does not (novox/hq ADR 0437 §3). With no bus held, a word that passes the
// claim is held for the bus (ErrTryAgain); a word that does not is refused before the bus is asked.
func TestOnlyAModuleClaimingTheSeatIsTheDeliverysWordOnAMerge(t *testing.T) {
entries := []inventory.Entry{claimingTheSeat("mesh-delivery-gate"), namedOnly("gitea"), namedOnly("mesh-delivery")}
keeper := keepingDeliveryMerges{
js: func() *broker.JetStream { return nil },
catalog: func(context.Context) ([]inventory.Entry, error) { return entries, nil },
}
body := []byte(`{"owner":"novox","repo":"r","number":1,"head_sha":"abc","ask_id":"a","option":"approve"}`)
for emitter, passes := range map[string]bool{
"mesh-delivery-gate": true, // the claimant, under another name
"gitea": false, // claims no seat
"mesh-delivery": false, // named so, claiming no seat here
} {
err := keeper.KeepDeliveryMerge(context.Background(), emitter, body)
if passes != errors.Is(err, link.ErrTryAgain) || (!passes && !errors.Is(err, link.ErrNotTheDeliverysWord)) {
t.Errorf("%s: %v (passes the claim: %v)", emitter, err, passes)
}
}
entries = []inventory.Entry{claimingTheSeat("mesh-delivery")}
if err := keeper.KeepDeliveryMerge(context.Background(), "mesh-delivery", body); !errors.Is(err, link.ErrTryAgain) {
t.Errorf("today's module, claiming the seat: %v", err)
}
}
+53 -6
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"slices"
"sort"
"strings"
@@ -12,8 +13,10 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/asks"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
)
@@ -44,6 +47,28 @@ type deliveryMerge struct {
AskID string `json:"ask_id"`
Option string `json:"option"`
Heads map[string]string `json:"heads"`
// Asker is the module that asked the ask, whose warrant is read from the router under that name (novox/hq ADR
// 0437 §6): empty is today's module, mesh-delivery.
Asker string `json:"asker,omitempty"`
}
// askerOf is the asker whose warrant the merge is read under: the one the announcement names, mesh-delivery when it
// names none.
func (d deliveryMerge) askerOf() string {
if d.Asker == "" {
return theDelivery
}
return d.Asker
}
// askerAccepted is whether the controller takes a warrant from this asker for a merge: the teller, and mesh-delivery
// until the gate takes the seat over (novox/hq ADR 0437 §6). Any other registers no move.
func askerAccepted(asker string) error {
if asker == theDelivery || asker == sdkdelivery.TellerModule {
return nil
}
return fmt.Errorf("the merge names the asker %q, and the controller takes a warrant only from %s and %s",
asker, sdkdelivery.TellerModule, theDelivery)
}
// parseDeliveryMerge reads an announcement, refusing one that names no pull request, head or ask.
@@ -79,6 +104,10 @@ func mergeAct(ask, option string, heads map[string]string) asks.Act {
// proved by the channel it came through, of this very ask as it was published (digest), bound to the heads the
// announcement names, and those heads hold the head merged. Answers why it holds, or why not.
func movedOnApprove(d deliveryMerge, ask *asks.Ask, w *asks.Warrant) (string, error) {
asker := d.askerOf()
if err := askerAccepted(asker); err != nil {
return "", err
}
switch {
case ask == nil || w == nil:
return "", errors.New("the router holds no closed record of the ask the merge was made on")
@@ -87,7 +116,7 @@ func movedOnApprove(d deliveryMerge, ask *asks.Ask, w *asks.Warrant) (string, er
case d.Option != optionApprove:
return "", fmt.Errorf("the merge was made on the option %q, not an Approve", d.Option)
}
option, err := w.For(theDelivery, *ask)
option, err := w.For(asker, *ask)
if err != nil {
return "", err
}
@@ -115,12 +144,26 @@ func movedOnApprove(d deliveryMerge, ask *asks.Ask, w *asks.Warrant) (string, er
// keepingDeliveryMerges keeps each merge mesh-delivery announces in the controller's bucket for them.
type keepingDeliveryMerges struct {
js func() *broker.JetStream
// catalog is the catalogue, whose modules claiming the delivery seat are the only ones whose word on a merge is
// kept (novox/hq ADR 0437 §3).
catalog func(ctx context.Context) ([]inventory.Entry, error)
// kept, when set, is told each merge once it is kept: a first merge waiting for the word asks its builds then
// (novox/hq ADR 0432, issue 1146).
kept func(ctx context.Context, d deliveryMerge) error
}
func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, body []byte) error {
func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, emitter string, body []byte) error {
if k.catalog == nil {
return fmt.Errorf("%w: the catalogue is not read", link.ErrTryAgain)
}
entries, err := k.catalog(ctx)
if err != nil {
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
}
if !slices.Contains(claimantsOf(entries, catalogue.DeliverySeat), emitter) {
return fmt.Errorf("%w: %q is not among %v", link.ErrNotTheDeliverysWord, emitter,
claimantsOf(entries, catalogue.DeliverySeat))
}
d, err := parseDeliveryMerge(body)
if err != nil {
return err
@@ -172,9 +215,9 @@ var keptDeliveryMerge = func(ctx context.Context, owner, repo string, number int
return d, err == nil, err
}
// routerAskOf is the router's record of one of mesh-delivery's asks: the ask as published and its warrant. A
// routerAskOf is the router's record of one of the delivery's asks, under its asker: the ask as published and its warrant. A
// variable so a test can hand it one.
var routerAskOf = func(ctx context.Context, inv catalogueReader, id string) (*asks.Ask, *asks.Warrant, error) {
var routerAskOf = func(ctx context.Context, inv catalogueReader, asker, id string) (*asks.Ask, *asks.Warrant, error) {
serving := servingBus.Load()
if serving == nil {
return nil, nil, errors.New("only the serving controller reads the router's record")
@@ -186,7 +229,7 @@ var routerAskOf = func(ctx context.Context, inv catalogueReader, id string) (*as
if bucket == "" {
return nil, nil, errors.New("no module declares the operator channel's records")
}
_, ask, w, err := readRouterEntry(ctx, serving.Conn(), bucket, theDelivery, id)
_, ask, w, err := readRouterEntry(ctx, serving.Conn(), bucket, asker, id)
return ask, w, err
}
@@ -267,7 +310,11 @@ func approvedAnnounced(ctx context.Context, inv catalogueReader, owner, name str
if !kept {
return "", fmt.Errorf("%w: %s/%s#%d at %.8s", errNotAnnounced, owner, name, k.Number, k.HeadSHA)
}
ask, w, err := routerAskOf(ctx, inv, d.AskID)
// The warrant is read under the asker the announcement names, and under none the controller does not take.
if err := askerAccepted(d.askerOf()); err != nil {
return "", fmt.Errorf("%w: %w", errWordNotAnApprove, err)
}
ask, w, err := routerAskOf(ctx, inv, d.askerOf(), d.AskID)
if err != nil {
return "", err
}
+139
View File
@@ -0,0 +1,139 @@
package main
import (
"context"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/asks"
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
)
// The controller reads a merge's warrant under the asker the announcement names, and accepts only the teller and
// today's mesh-delivery (novox/hq ADR 0437 §6, issue 1213). The teller's name is the SDK's.
// askedBy is an approved merge whose announcement names an asker ("" names none) and whose warrant is that asker's.
func askedBy(t *testing.T, named, warrantOf string) anApprovedMerge {
t.Helper()
a := approvedMerge(t)
a.merge.Asker = named
a.warrant.Asker = warrantOf
return a
}
// 1213/AC1: an announcement naming the teller, and the teller's warrant of the operator's verified Approve of the
// merge's heads: the move is registered (the warrant is read from the router under the teller).
func TestAMergeNamingTheTellerIsMovedOnTheTellersWarrant(t *testing.T) {
a := askedBy(t, sdkdelivery.TellerModule, sdkdelivery.TellerModule)
if _, err := movedOnApprove(a.merge, &a.ask, &a.warrant); err != nil {
t.Fatalf("the teller's warrant of the operator's Approve was refused: %v", err)
}
was := routerAskOf
t.Cleanup(func() { routerAskOf = was })
wasKept := keptDeliveryMerge
t.Cleanup(func() { keptDeliveryMerge = wasKept })
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
return a.merge, true, nil
}
var readUnder string
routerAskOf = func(_ context.Context, _ catalogueReader, asker, _ string) (*asks.Ask, *asks.Warrant, error) {
readUnder = asker
return &a.ask, &a.warrant, nil
}
if _, err := approvedAnnounced(t.Context(), nil, "novox", "mesh-controller", announcedPull(7, a.merge.HeadSHA)); err != nil {
t.Fatalf("the merge was not moved on the teller's warrant: %v", err)
}
if readUnder != sdkdelivery.TellerModule {
t.Errorf("the warrant was read under %q", readUnder)
}
}
// 1213/AC2: an announcement naming mesh-delivery, or none (today's module): read under mesh-delivery, registered.
func TestAMergeNamingNoAskerIsMovedOnMeshDeliverysWarrant(t *testing.T) {
for _, named := range []string{"", theDelivery} {
a := askedBy(t, named, theDelivery)
if _, err := movedOnApprove(a.merge, &a.ask, &a.warrant); err != nil {
t.Errorf("asker %q: refused: %v", named, err)
}
}
was := routerAskOf
wasKept := keptDeliveryMerge
t.Cleanup(func() { routerAskOf, keptDeliveryMerge = was, wasKept })
a := askedBy(t, "", theDelivery)
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
return a.merge, true, nil
}
var readUnder string
routerAskOf = func(_ context.Context, _ catalogueReader, asker, _ string) (*asks.Ask, *asks.Warrant, error) {
readUnder = asker
return &a.ask, &a.warrant, nil
}
if _, err := approvedAnnounced(t.Context(), nil, "novox", "mesh-controller", announcedPull(7, a.merge.HeadSHA)); err != nil {
t.Fatal(err)
}
if readUnder != theDelivery {
t.Errorf("with no asker named the warrant was read under %q, not %q", readUnder, theDelivery)
}
}
// 1213/AC3: any other asker registers no move, and the reason names the asker refused; the router is not asked.
func TestAMergeNamingAnyOtherAskerMovesNothingAndNamesIt(t *testing.T) {
was := routerAskOf
wasKept := keptDeliveryMerge
t.Cleanup(func() { routerAskOf, keptDeliveryMerge = was, wasKept })
for _, other := range []string{sdkdelivery.GateModule, "mesh-controller", "unknown-module"} {
a := askedBy(t, other, other) // even a warrant of that very asker
_, err := movedOnApprove(a.merge, &a.ask, &a.warrant)
if err == nil || !strings.Contains(err.Error(), other) {
t.Errorf("asker %q: %v", other, err)
}
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
return a.merge, true, nil
}
routerAskOf = func(context.Context, catalogueReader, string, string) (*asks.Ask, *asks.Warrant, error) {
t.Errorf("the router was read under the refused asker %q", other)
return &a.ask, &a.warrant, nil
}
if _, err := approvedAnnounced(t.Context(), nil, "novox", "mesh-controller", announcedPull(7, a.merge.HeadSHA)); err == nil ||
!strings.Contains(err.Error(), other) {
t.Errorf("asker %q, approved as announced: %v", other, err)
}
}
}
// 1213/AC4: a warrant whose own asker differs from the asker read, or whose heads differ from the head merged, moves nothing.
func TestAWarrantOfAnotherAskerOrOtherHeadsMovesNothing(t *testing.T) {
a := askedBy(t, sdkdelivery.TellerModule, theDelivery) // read under the teller, but the warrant says mesh-delivery
if why, err := movedOnApprove(a.merge, &a.ask, &a.warrant); err == nil {
t.Errorf("a warrant of another asker was accepted: %s", why)
}
a = askedBy(t, "", sdkdelivery.TellerModule) // today's module named, the teller's warrant
if why, err := movedOnApprove(a.merge, &a.ask, &a.warrant); err == nil {
t.Errorf("the teller's warrant under mesh-delivery was accepted: %s", why)
}
a = askedBy(t, sdkdelivery.TellerModule, sdkdelivery.TellerModule)
a.merge.HeadSHA = "d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4"
if why, err := movedOnApprove(a.merge, &a.ask, &a.warrant); err == nil {
t.Errorf("a head the Approve did not cover was accepted: %s", why)
}
}
// 1213/AC5: the announcement's asker is read and kept with the rest of it, under the JSON name asker.
func TestTheAnnouncementsAskerIsReadAndKept(t *testing.T) {
body := `{"owner":"novox","repo":"r","number":1,"head_sha":"abc","ask_id":"a","option":"approve","asker":"mesh-delivery-teller"}`
d, err := parseDeliveryMerge([]byte(body))
if err != nil || d.Asker != sdkdelivery.TellerModule {
t.Fatalf("read %+v, %v", d, err)
}
none, err := parseDeliveryMerge([]byte(strings.Replace(body, `,"asker":"mesh-delivery-teller"`, "", 1)))
if err != nil || none.Asker != "" {
t.Fatalf("an announcement naming none read as %+v, %v", none, err)
}
}
// announcedPull is the forge's announcement of a merge of a pull request at a head.
func announcedPull(number int, head string) keptMerge {
var k keptMerge
k.Number, k.HeadSHA = number, head
return k
}
+2 -2
View File
@@ -74,7 +74,7 @@ func TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTh
}
server := link.ConnectNats(js, nil, noListener{})
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream}); err != nil {
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: theDeliveryClaims}); err != nil {
t.Fatal(err)
}
// The events consumer exists, from now on, before the word is published: Receive would make it too late.
@@ -90,7 +90,7 @@ func TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTh
// mesh-delivery says it merged the pull request, and the controller keeps what it said.
body, _ := json.Marshal(a.merge)
if _, err := js.Context().Publish(broker.DeliveryMergedSubject, body); err != nil {
if _, err := js.Context().Publish(broker.DeliveryMergedSubjectOf(theDelivery), body); err != nil {
t.Fatal(err)
}
becomes(t, 15*time.Second, "the merge being kept", func() bool {
@@ -93,7 +93,7 @@ func TestANewModuleIsRegisteredOnTheOperatorsApproveOfItsMergeAndRefusedOtherwis
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
return a.merge, !c.unkept, nil
}
routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) {
routerAskOf = func(context.Context, catalogueReader, string, string) (*asks.Ask, *asks.Warrant, error) {
return &a.ask, &a.warrant, nil
}
if c.forge != nil {
@@ -149,7 +149,7 @@ func TestASecondNewModuleFromARepositoryIsJudgedAlongsideAndAsksNoWarrant(t *tes
t.Error("1146/AC5: the announced merges were asked for a module judged alongside")
return deliveryMerge{}, false, nil
}
routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) {
routerAskOf = func(context.Context, catalogueReader, string, string) (*asks.Ask, *asks.Warrant, error) {
t.Error("1146/AC5: the router was asked for a module judged alongside")
return nil, nil, nil
}
+2 -2
View File
@@ -75,7 +75,7 @@ func TestANewModuleFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheB
}
server := link.ConnectNats(js, nil, noListener{})
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream}); err != nil {
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: theDeliveryClaims}); err != nil {
t.Fatal(err)
}
// The events consumer exists, from now on, before the word is published: Receive would make it too late.
@@ -91,7 +91,7 @@ func TestANewModuleFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheB
// mesh-delivery says it merged the pull request, and the controller keeps what it said.
body, _ := json.Marshal(a.merge)
if _, err := js.Context().Publish(broker.DeliveryMergedSubject, body); err != nil {
if _, err := js.Context().Publish(broker.DeliveryMergedSubjectOf(theDelivery), body); err != nil {
t.Fatal(err)
}
becomes(t, 15*time.Second, "the merge being kept", func() bool {
@@ -0,0 +1,93 @@
package main
import (
"context"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
)
// The controller finds the delivery's owner by the claim of the seat mesh-delivery, never by the module's name
// (novox/hq ADR 0437 §3, issue 1210). The gate is a module named mesh-delivery-gate; this is only a fixture name.
func claimingTheSeat(module string) inventory.Entry {
return inventory.Entry{Manifest: catalogue.Manifest{Module: module,
Claims: []catalogue.Claim{{Name: catalogue.DeliverySeat, Scope: catalogue.ScopeMesh}}}, On: []string{"anchor"}}
}
func namedOnly(module string) inventory.Entry {
return inventory.Entry{Manifest: catalogue.Manifest{Module: module}, On: []string{"anchor"}}
}
// 1210/AC1: a module claiming the seat is on the controller's own path whatever its name, awaiting the delivery's
// word as today's module does; a module named mesh-delivery that claims no seat is not on it.
func TestTheModuleClaimingTheSeatIsOnTheControllersOwnPath(t *testing.T) {
gate := []inventory.Entry{claimingTheSeat("mesh-delivery-gate"), namedOnly("mesh-delivery")}
own := ownPathOf(gate)
if _, on := own["mesh-delivery-gate"]; !on {
t.Error("the module claiming the seat is not on the controller's own path")
}
if _, on := own["mesh-delivery"]; on {
t.Error("a module named mesh-delivery that claims no seat is on the controller's own path")
}
if d := awaitsFor(gate, []string{"gitea", "plex"}); d == nil || d.Awaits != catalogue.DeliverySeat {
t.Errorf("an ordinary walk awaits %+v with the gate holding the seat", d)
}
if d := awaitsFor(gate, []string{"mesh-delivery-gate", "gitea"}); d != nil {
t.Errorf("a walk moving the gate awaits %+v; the gate cannot gate itself", d)
}
if d := awaitsFor(gate, []string{"mesh-delivery", "gitea"}); d == nil {
t.Error("a walk moving a module named mesh-delivery that claims no seat starts at once")
}
if !ownWalk(walkOf("novox/one", "notes", "mesh-delivery-gate"), own) || ownWalk(walkOf("novox/one", "notes", "mesh-delivery"), own) {
t.Error("a walk is on the own path by the claim of the module it plans")
}
if ownPath([]inventory.Entry{claimingTheSeat("mesh-delivery-gate")}) != true || ownPath([]inventory.Entry{namedOnly("mesh-delivery")}) {
t.Error("a merge touches the own path by the claim of the module it touches")
}
if classOfModules([]string{"mesh-delivery-gate"}, gate) != inventory.ClassCore {
t.Error("the module claiming the seat is not core")
}
}
// 1210/AC2: the merge gate's rule of the module claiming the seat is one reading; a module claiming no delivery
// seat gets its ordinary rule.
func TestTheModuleClaimingTheSeatGetsTheDeliverysGateRule(t *testing.T) {
withTheGatesTimes(t)
claimant := claimingTheSeat("mesh-delivery-gate").Manifest
r := gateRuleOf("mesh-delivery-gate", claimant)
if r.passes != 1 || r.within != 0 || !strings.HasPrefix(r.said, "mesh-delivery's rule: one reading (novox/hq ADR 0351)") {
t.Errorf("the gate's rule is %+v", r)
}
r = gateRuleOf("mesh-delivery", catalogue.Manifest{Module: "mesh-delivery"})
if !strings.HasPrefix(r.said, "a leaf module's rule") {
t.Errorf("a module named mesh-delivery claiming no seat gets %q", r.said)
}
}
// 1210/AC3: with both today's module and the gate claiming the seat (R3d's group), both are on the own path and
// both get the delivery's rule.
func TestBothClaimantsAreOnTheOwnPathAndGetTheRule(t *testing.T) {
withTheGatesTimes(t)
both := []inventory.Entry{claimingTheSeat("mesh-delivery"), claimingTheSeat("mesh-delivery-gate")}
own := ownPathOf(both)
for _, m := range []string{"mesh-delivery", "mesh-delivery-gate"} {
if _, on := own[m]; !on {
t.Errorf("%s is not on the own path", m)
}
if r := gateRuleOf(m, claimingTheSeat(m).Manifest); r.passes != 1 || !strings.HasPrefix(r.said, "mesh-delivery's rule") {
t.Errorf("%s gets %+v", m, r)
}
if d := awaitsFor(both, []string{m, "gitea"}); d != nil {
t.Errorf("a walk moving %s awaits %+v", m, d)
}
}
}
// theDeliveryClaims is a catalogue whose one module claiming the seat is today's mesh-delivery, for the tests that
// announce a merge from it.
func theDeliveryClaims(context.Context) ([]inventory.Entry, error) {
return []inventory.Entry{claimingTheSeat("mesh-delivery")}, nil
}
+10 -6
View File
@@ -155,7 +155,7 @@ func newerOpenPlan(p inventory.Plan, plans []inventory.Plan) (inventory.Plan, bo
}
// retryRefusal is why a plan cannot be retried, or nothing.
func retryRefusal(p inventory.Plan, plans []inventory.Plan) error {
func retryRefusal(p inventory.Plan, plans []inventory.Plan, onPath map[string]string) error {
switch {
case p.State == inventory.PlanDone:
return fmt.Errorf("%s is done; there is nothing to retry", p.ID)
@@ -170,7 +170,7 @@ func retryRefusal(p inventory.Plan, plans []inventory.Plan) error {
return fmt.Errorf("%s supersedes it: a newer merge of %s (%s at %s) is open, and retrying %s would build "+
"what that one replaced", q.ID, q.Repository, q.ID, short(q.Commit), p.ID)
}
return oneWalkAtATime(p, plans)
return oneWalkAtATime(p, plans, onPath)
}
stopped := stoppedRollouts(p)
if len(stopped) == 0 {
@@ -194,20 +194,20 @@ func retryRefusal(p inventory.Plan, plans []inventory.Plan) error {
"the older build back", m, q.ID, q.State, q.Repository, short(q.Commit), p.ID)
}
}
return oneWalkAtATime(p, plans)
return oneWalkAtATime(p, plans, onPath)
}
// oneWalkAtATime refuses a retry while an open walk, started or waiting for its word, is related to it (novox/hq
// ADR 0276, ADR 0407 decision 4, issue 973): a walk retried beside a related one would be two walks on the same
// things at once. A walk related to none may be retried beside them. The retried walk is judged by the modules it
// plans, which already hold their dependents, and the repositories it carries.
func oneWalkAtATime(p inventory.Plan, plans []inventory.Plan) error {
func oneWalkAtATime(p inventory.Plan, plans []inventory.Plan, onPath map[string]string) error {
mine := footprintOfWalk(p, nil)
for _, q := range plans {
if q.ID == p.ID || !q.Open() || q.Release != nil {
continue
}
if ownWalk(p) || ownWalk(q) {
if ownWalk(p, onPath) || ownWalk(q, onPath) {
return fmt.Errorf("%s is open (%s) and one of them is on the controller's own path: one walk at a time — "+
"retry %s once it ended", q.ID, q.Named(), p.ID)
}
@@ -327,7 +327,11 @@ func retryPlan(ctx context.Context, open *stores, id string) (string, error) {
}
failed = toRetry(&p, recorded, byID)
}
if err := retryRefusal(p, plans); err != nil {
onPath, err := ownPathNow(ctx, inv)
if err != nil {
return "", err
}
if err := retryRefusal(p, plans, onPath); err != nil {
return "", err
}
if again := unjudgedAtGate(p); len(failed) == 0 && len(again) > 0 {
+1 -1
View File
@@ -209,7 +209,7 @@ func serve(ctx context.Context) (err error) {
}
// And the merges mesh-delivery made, kept a week with the ask each was made on: what registers a module that
// moved between repositories on the operator's Approve (novox/hq ADR 0420 §7, issue 1025).
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream,
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: open.inventory.Catalogued,
kept: following{open: open}.resumeFirstBuilds}); err != nil {
return err
}
+4 -4
View File
@@ -68,7 +68,7 @@ func TestRetryRefusesWhatItCannotResume(t *testing.T) {
{plan("plan-o", inventory.PlanBuilding, at), nil, "still building"},
{failed, []inventory.Plan{plan("plan-2", inventory.PlanRolling, at.Add(time.Hour))}, "plan-2 supersedes it"},
} {
err := retryRefusal(c.p, c.others)
err := retryRefusal(c.p, c.others, ownPathOf(nil))
if err == nil || !strings.Contains(err.Error(), c.says) {
t.Errorf("%s: %v, wanted it to say %q", c.p.ID, err, c.says)
}
@@ -76,12 +76,12 @@ func TestRetryRefusesWhatItCannotResume(t *testing.T) {
stopped := failed
stopped.Modules = map[string]*inventory.PlanModule{"a": {State: "built"}}
stopped.Note = "a stopped at its first machine"
if err := retryRefusal(stopped, nil); err == nil || !strings.Contains(err.Error(), "nothing in tier 0") {
if err := retryRefusal(stopped, nil, ownPathOf(nil)); err == nil || !strings.Contains(err.Error(), "nothing in tier 0") {
t.Errorf("a plan with nothing failed to build was retried: %v", err)
}
// A failed or done plan does not supersede it.
if err := retryRefusal(failed, []inventory.Plan{plan("plan-4", inventory.PlanFailed, at.Add(time.Hour)),
plan("plan-5", inventory.PlanDone, at.Add(time.Hour))}); err != nil {
plan("plan-5", inventory.PlanDone, at.Add(time.Hour))}, ownPathOf(nil)); err != nil {
t.Errorf("refused for a plan that does not supersede it: %v", err)
}
// Another branch's open walk, or an older one, does not supersede it, and is one walk open: one at a time
@@ -89,7 +89,7 @@ func TestRetryRefusesWhatItCannotResume(t *testing.T) {
other := plan("plan-3", inventory.PlanBuilding, at.Add(time.Hour))
other.Branch = "release"
for _, open := range []inventory.Plan{other, plan("plan-0", inventory.PlanBuilding, at.Add(-time.Hour))} {
if err := retryRefusal(failed, []inventory.Plan{open}); err == nil || !strings.Contains(err.Error(), "one walk at a time") {
if err := retryRefusal(failed, []inventory.Plan{open}, ownPathOf(nil)); err == nil || !strings.Contains(err.Error(), "one walk at a time") {
t.Errorf("retried beside the open walk %s: %v", open.ID, err)
}
}
+6 -6
View File
@@ -144,12 +144,12 @@ func relatedOpen(cand footprint, candOwn bool, open []openWalk, edges []inventor
// ownWalk says a walk is on the controller's own path: its batch says so, or it plans a module of that path
// (issue 973).
func ownWalk(p inventory.Plan) bool {
func ownWalk(p inventory.Plan, onPath map[string]string) bool {
if p.OwnPath() {
return true
}
for name := range p.Modules {
if _, own := onTheControllersPath[name]; own {
if _, own := onPath[name]; own {
return true
}
}
@@ -173,19 +173,19 @@ func startedWalks(plans []inventory.Plan, id string, created time.Time) []invent
// openWalkOf is an open walk as a change is judged against it: what it reaches, from its planned modules and the
// merges the record holds for it.
func openWalkOf(ctx context.Context, inv *inventory.Inventory, p inventory.Plan) (openWalk, error) {
func openWalkOf(ctx context.Context, inv *inventory.Inventory, p inventory.Plan, onPath map[string]string) (openWalk, error) {
kept, err := inv.MergesOf(ctx, p.ID)
if err != nil {
return openWalk{}, err
}
return openWalk{id: p.ID, fp: footprintOfWalk(p, kept), own: ownWalk(p)}, nil
return openWalk{id: p.ID, fp: footprintOfWalk(p, kept), own: ownWalk(p, onPath)}, nil
}
// openWalksOf are openWalkOf for each of plans.
func openWalksOf(ctx context.Context, inv *inventory.Inventory, plans []inventory.Plan) ([]openWalk, error) {
func openWalksOf(ctx context.Context, inv *inventory.Inventory, plans []inventory.Plan, onPath map[string]string) ([]openWalk, error) {
var out []openWalk
for _, p := range plans {
w, err := openWalkOf(ctx, inv, p)
w, err := openWalkOf(ctx, inv, p, onPath)
if err != nil {
return nil, err
}
+6 -6
View File
@@ -119,19 +119,19 @@ func TestABatchWhoseMovesAreNotKnownIsRelatedToEverything(t *testing.T) {
// A walk is on the controller's own path when its batch says so or it plans a module of that path (issue 973).
func TestAWalkOnTheControllersPathIsKnownByItsFlagOrItsModules(t *testing.T) {
if ownWalk(walkOf("novox/one", "notes")) {
if ownWalk(walkOf("novox/one", "notes"), ownPathOf(nil)) {
t.Error("an ordinary walk read as on the controller's own path")
}
flagged := walkOf("novox/one", "notes")
flagged.Delivery = &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: true}}
if !ownWalk(flagged) {
if !ownWalk(flagged, ownPathOf(nil)) {
t.Error("a walk whose batch is own-path read as ordinary")
}
var module string
for m := range onTheControllersPath {
module = m
}
if !ownWalk(walkOf("novox/one", "notes", module)) {
if !ownWalk(walkOf("novox/one", "notes", module), ownPathOf(nil)) {
t.Errorf("a walk planning %s read as ordinary", module)
}
}
@@ -168,15 +168,15 @@ func TestARetryWaitsOnlyForAnOpenWalkItIsRelatedTo(t *testing.T) {
p.ID, p.State = id, inventory.PlanBuilding
return p
}
if err := oneWalkAtATime(failed, []inventory.Plan{open("plan-2", "novox/two", "app")}); err != nil {
if err := oneWalkAtATime(failed, []inventory.Plan{open("plan-2", "novox/two", "app")}, ownPathOf(nil)); err != nil {
t.Errorf("refused beside an unrelated walk: %v", err)
}
err := oneWalkAtATime(failed, []inventory.Plan{open("plan-3", "novox/three", "notes")})
err := oneWalkAtATime(failed, []inventory.Plan{open("plan-3", "novox/three", "notes")}, ownPathOf(nil))
if err == nil || !strings.Contains(err.Error(), "plan-3") || !strings.Contains(err.Error(), "both move notes") {
t.Errorf("a retry beside a walk moving the same module: %v", err)
}
failed.Delivery = &inventory.PlanDelivery{Batch: &inventory.PlanBatch{Own: true}}
if err := oneWalkAtATime(failed, []inventory.Plan{open("plan-2", "novox/two", "app")}); err == nil {
if err := oneWalkAtATime(failed, []inventory.Plan{open("plan-2", "novox/two", "app")}, ownPathOf(nil)); err == nil {
t.Error("an own-path retry went beside an open walk")
}
}
+6 -2
View File
@@ -701,11 +701,15 @@ func startedBeside(ctx context.Context, inv *inventory.Inventory, p inventory.Pl
if err != nil {
return "", "", err
}
open, err := openWalksOf(ctx, inv, started)
onPath, err := ownPathNow(ctx, inv)
if err != nil {
return "", "", err
}
mine, err := openWalkOf(ctx, inv, p)
open, err := openWalksOf(ctx, inv, started, onPath)
if err != nil {
return "", "", err
}
mine, err := openWalkOf(ctx, inv, p, onPath)
if err != nil {
return "", "", err
}
+1 -1
View File
@@ -3,7 +3,7 @@ module github.com/novox/mesh-controller
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9
github.com/jackc/pgx/v5 v5.10.0
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
+2
View File
@@ -2,6 +2,8 @@ git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5 h1:E5stfIeCN6uYg
git.novox.be/novox/mesh-host v0.0.0-20261018171808-9d9e3a46c4f5/go.mod h1:4wyoCr6IIIIarEUuROThWDj7DAvJvIidqMjWcpSnOwE=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4 h1:yYDxFRJoH59kiCoEulD+FBwctyf9CrtKIxlEAsVL39Q=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9 h1:qHpN/fpTDqijfMVoepxUT6mPkfcydb4LiPKsIokv6iM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+74
View File
@@ -0,0 +1,74 @@
package broker
import "testing"
// The operator-channel seat as the messenger declares it once it takes a module's condition: copied from
// mesh-catalog modules/messenger/module.json at 5b235a6e (issue 1217), the seat's accepts, emits and by-caller verbs.
func operatorChannelWithCondition() Seat {
return Seat{Name: "operator-channel", Scope: "mesh", Accepts: []string{"ask", "cancel", "condition"},
Emits: []string{"decided"}, Serves: []string{"open", "history", "notify", "tell", "asks"},
ByCaller: []string{"ask", "cancel", "condition", "decided"}, Records: []string{"asks"}}
}
// 1216/AC1: a module that uses the seat may publish its condition under its own name.
func TestAModuleUsingTheOperatorChannelMayPublishItsConditionUnderItsOwnName(t *testing.T) {
x := perms(t, Principal{Kind: KindModule, Node: "anchor", Module: "web-app", Uses: []Seat{operatorChannelWithCondition()}})
if !allowed(x.Publish, "mesh.seat.operator-channel.accept.condition.web-app") {
t.Errorf("web-app may not say a condition as itself: %v", x.Publish)
}
}
// 1216/AC2: it may not publish a condition under another module's name, nor on the wildcard.
func TestAModuleMayNotSayAConditionUnderAnotherName(t *testing.T) {
x := perms(t, Principal{Kind: KindModule, Node: "anchor", Module: "web-app", Uses: []Seat{operatorChannelWithCondition()}})
for _, subject := range []string{
"mesh.seat.operator-channel.accept.condition.other-app",
"mesh.seat.operator-channel.accept.condition.*",
"mesh.seat.operator-channel.accept.condition.>",
"mesh.seat.operator-channel.accept.condition",
} {
if allowed(x.Publish, subject) {
t.Errorf("web-app may publish %s", subject)
}
}
}
// 1216/AC3: the seat's holder may take accept.condition.<any module> from the work queue SEAT_OPERATOR_CHANNEL.
func TestTheHolderOfTheOperatorChannelTakesEveryModulesCondition(t *testing.T) {
seat := operatorChannelWithCondition()
holder := perms(t, Principal{Kind: KindModule, Node: "anchor", Module: "messenger", Holds: []Seat{seat}})
if !allowed(holder.Subscribe, "mesh.seat.operator-channel.accept.condition.web-app") ||
!allowed(holder.Subscribe, "mesh.seat.operator-channel.accept.condition.other-app") {
t.Errorf("the holder does not take accept.condition.*: %v", holder.Subscribe)
}
if !allowed(holder.Publish, "$JS.API.CONSUMER.MSG.NEXT.SEAT_OPERATOR_CHANNEL.SEAT_OPERATOR_CHANNEL_worker") {
t.Errorf("the holder may not pull the seat's work queue: %v", holder.Publish)
}
users, _ := Users(Records{Nodes: []string{"anchor"}, Assigned: map[string][]Declared{"anchor": {
{Module: "messenger", Holds: []Seat{seat}},
{Module: "web-app", Uses: []Seat{seat}},
}}})
streams, workers := SeatTrafficObjects(users)
kept := false
for _, s := range streams {
if s.Name == "SEAT_OPERATOR_CHANNEL" {
for _, subject := range s.Subjects {
kept = kept || subjectMatches(subject, "mesh.seat.operator-channel.accept.condition.web-app")
}
}
}
if !kept {
t.Error("the work queue SEAT_OPERATOR_CHANNEL does not keep accept.condition.<module>")
}
taken := false
for _, w := range workers {
if w.Name == "SEAT_OPERATOR_CHANNEL_worker" {
for _, f := range w.Filters {
taken = taken || subjectMatches(f, "mesh.seat.operator-channel.accept.condition.web-app")
}
}
}
if !taken {
t.Error("the seat's worker does not take accept.condition.<module>")
}
}
+12 -2
View File
@@ -448,8 +448,18 @@ var ControllerFollows = []string{
// the heads that Approve covered, kept a week to read the warrant a moved module's registration needs.
var ControllerHears = []string{DeliveryMergedSubject}
// DeliveryMergedSubject is where mesh-delivery says it merged a pull request.
var DeliveryMergedSubject = moduleEventSubject("mesh-delivery", "merged")
// DeliveryMergedEvent is the event a module says when it merged a pull request.
const DeliveryMergedEvent = "merged"
// DeliveryMergedSubject is where the delivery's owner says it merged a pull request: one wildcard subject, the
// module that claims the delivery seat being whichever the catalogue says (novox/hq ADR 0437 §3). Who said it is
// read from the subject, and whether it claims the seat is the controller's to check.
var DeliveryMergedSubject = moduleEventSubject("*", DeliveryMergedEvent)
// DeliveryMergedSubjectOf is where one module says it merged a pull request.
func DeliveryMergedSubjectOf(module string) string {
return moduleEventSubject(module, DeliveryMergedEvent)
}
// AsksSeat is the seat an ask is made on and its warrant heard from (novox/hq ADR 0259): the router's.
const AsksSeat = "operator-channel"
+19
View File
@@ -9,6 +9,8 @@ import (
"slices"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// What a build says it was made from, as files (novox/hq ADR 0267), and what a build compiling a Go
@@ -202,3 +204,20 @@ func TestAGoBundleSaysItsClosureAndAnUntoldImageNothing(t *testing.T) {
t.Fatalf("an image compiling nothing it was told of said a build source: %+v", got.Sources)
}
}
// 1215/AC1 (compiled here, run by the merge check; the unit level has 1215/AC1 over the same inputs): a build's
// result carries every input of its fingerprint, and the SDK function over them gives it again.
func TestABuildSaysTheInputsItsFingerprintWasComputedFrom(t *testing.T) {
got, _, _ := buildTheProxy(t, aSharedRepository("one", "1"), false)
if got.SourceInputs == nil {
t.Fatal("the result carries no source inputs")
}
in := *got.SourceInputs
if in.Module != "route-proxy" || in.Tree == "" || in.Contexts["server"] == "" {
t.Errorf("the inputs miss the module, its tree or the server's context: %+v", in)
}
fp, none := delivery.SourceFingerprint(in)
if fp == "" || none != "" || fp != got.Source {
t.Errorf("the inputs give %q (%q); the build recorded %q", fp, none, got.Source)
}
}
+6 -1
View File
@@ -21,6 +21,7 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/catalogue"
)
@@ -91,6 +92,10 @@ type Result struct {
// (novox/hq issue 280).
Source string
// SourceInputs are what Source was computed from (novox/hq ADR 0437 §5, issue 1215), set also where the source
// does not pin the build, so the reason is visible: with them delivery.SourceFingerprint gives Source again.
SourceInputs *delivery.SourceInputs
// ReuseKey is the build's reuse key (novox/hq ADR 0350 §1): its source fingerprint's parts with the builder's
// version, and ReuseParts those parts; empty, with NoKey saying why, where the source does not pin the build.
ReuseKey string
@@ -491,7 +496,7 @@ func build(ctx context.Context, run Runner, publish Publisher,
// The reuse key (novox/hq ADR 0350 §1): what a check keeps its build under, and what a walk takes one by.
key := keyOf(src, manifest, Version)
return Result{Manifest: resolved, Commit: commit, Built: built,
Against: against(within, manifest, stoodOn), Read: readBy(manifest), Source: fingerprint,
Against: against(within, manifest, stoodOn), Read: readBy(manifest), Source: fingerprint, SourceInputs: resultInputs(src),
Trunk: trunk, OnTrunk: onTrunk, Branches: branches, Sources: sources,
ReuseKey: key.Key, ReuseParts: key.Parts, NoKey: key.Why, Reused: reused, Why: builtWhy,
Predefined: listed}, nil
+30 -27
View File
@@ -10,6 +10,7 @@ import (
"sort"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/catalogue"
)
@@ -36,7 +37,7 @@ import (
// (novox/hq ADR 0097 refuses the unpinned bases; what a RUN step downloads is not seen here).
// sourcePrefix names the fingerprint's form, so a later form is never compared equal to this one.
const sourcePrefix = "src1:"
const sourcePrefix = delivery.SourcePrefix
// sourceInputs collects what one build was made from.
type sourceInputs struct {
@@ -262,40 +263,42 @@ func (s *sourceInputs) notPinned(why string) {
}
}
// sourceFingerprint is the one function the controller's builder and the indexer both call (novox/hq ADR 0437 §5);
// a variable only so a test can see it called.
var sourceFingerprint = delivery.SourceFingerprint
// inputs are what the build's fingerprint is of, as the SDK names them.
func (s *sourceInputs) inputs() delivery.SourceInputs {
if s == nil {
return delivery.SourceInputs{}
}
return delivery.SourceInputs{
Module: s.module, Tree: s.tree, Bases: s.bases,
Contexts: s.contexts, Toolchains: s.toolchains, Unpinned: s.unpinned,
}
}
// resultInputs are the inputs a build's result records; nil for a build with no inputs to speak of.
func resultInputs(s *sourceInputs) *delivery.SourceInputs {
if s == nil {
return nil
}
in := s.inputs()
return &in
}
// fingerprint is the build's source fingerprint, or empty when the source does not pin the build.
func (s *sourceInputs) fingerprint() string {
if s == nil || s.unpinned != "" || s.tree == "" {
if s == nil {
return ""
}
var lines []string
lines = append(lines, "module "+s.module, "tree "+s.tree)
bases := map[string]bool{}
for _, b := range s.bases {
bases[referenceDigest(b)] = true
}
for b := range bases {
lines = append(lines, "base "+b)
}
for a, t := range s.contexts {
lines = append(lines, "context "+a+" "+t)
}
for a, t := range s.toolchains {
lines = append(lines, "toolchain "+a+" "+t)
}
// The module and its tree first, the rest in a fixed order.
sort.Strings(lines[2:])
sum := sha256.Sum256([]byte(strings.Join(lines, "\n")))
return sourcePrefix + hex.EncodeToString(sum[:])
fp, _ := sourceFingerprint(s.inputs())
return fp
}
// referenceDigest is a reference's digest — `sha256:…` — so the registry address it was copied into does not
// enter the fingerprint; the reference itself when it carries none.
func referenceDigest(reference string) string {
if _, digest, pinned := strings.Cut(reference, "@"); pinned && digest != "" {
return digest
}
return reference
}
func referenceDigest(reference string) string { return delivery.ReferenceDigest(reference) }
// gitTree is the git tree of a directory of a clone at its checked-out commit: the whole tree for an
// empty path.
+125
View File
@@ -0,0 +1,125 @@
package builder
import (
"encoding/json"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 1214/AC1: the inputs of three builds the live builder recorded a source fingerprint for, copied from mesh-sdk
// go/delivery/fingerprint_test.go (TestFingerprintLiveBuilds), where each value is captured, never written by hand.
func TestTheBuildersFingerprintEqualsWhatTheLiveBuilderRecorded(t *testing.T) {
const ts = "typescript sha256:4b81682df5665c1f461d3f720a3a0fd7593f4aa71e4fc272dbfdc4bbd235b489 20125bd590041690"
for _, c := range []struct {
name string
in sourceInputs
recorded string
}{
{"mailu", sourceInputs{
module: "mailu",
tree: "cfc6fa8ff5fc1e220132e51ade653b9d3f387b3e",
bases: []string{"sha256:25f3cfeaceca14921366af4d1240b56457ef46273bdb508c7b0e8f469f6fd228"},
toolchains: map[string]string{"code": ts},
}, "src1:539f2e64000d436c572ef53b40e524012455945f8c5a1c6bb173ee0707e88ef4"},
{"radarr", sourceInputs{
module: "radarr",
tree: "42b9c918b5af75cba2bbc75845012e7f64e16df9",
toolchains: map[string]string{
"code": ts,
"tools-go": "go sha256:2054591ca034b68323d87ec887ba34f56819cb3364034360c0777da4b314dd54 0995d85edc9a6a23",
},
}, "src1:b3c150d39c395b99dd68974a48490e9a42467aa5d486d0c604d6ef1e8e9c910a"},
{"build-agent", sourceInputs{
module: "build-agent",
tree: "11c0b4f90cae0d6e4ef9cc48cbfdf9f6dbe52bca",
bases: []string{
"sha256:d9e853e87e55526f6b2917df91a2115c36dd7c696a35be12163d44e6e2a4b6bc",
"sha256:8ac98ca534ac3f51e1f420a1dd2c15e74c75cfa0f23f3ad27eb5d7236c349a0c",
},
contexts: map[string]string{"server": "narrow:d51ca744fd11af59d616c858f76c27b3e58e16e11fbdc0c720eb9d7d19d55fac"},
}, "src1:e846efe73716b78e80860baaac96153de2cb76551325753838ca245d2160a3d7"},
} {
if got := c.in.fingerprint(); got != c.recorded {
t.Errorf("%s: fingerprint = %q, the live builder recorded %q", c.name, got, c.recorded)
}
}
}
// 1214/AC2: a build with no tree, or one that resolves packages when built, gets no fingerprint.
func TestABuildWithoutATreeOrUnpinnedGetsNoFingerprint(t *testing.T) {
if (&sourceInputs{module: "app"}).fingerprint() != "" {
t.Error("a build with no tree has a fingerprint")
}
s := &sourceInputs{module: "app", tree: "aaaa"}
s.notPinned("it resolves packages from the mesh's registry at build time")
if got := s.fingerprint(); got != "" {
t.Errorf("an unpinned build has a fingerprint: %s", got)
}
if got := orNoTree(""); got != "its tree was not named" {
t.Errorf("the reason for no tree = %q", got)
}
if got := orNoTree(s.unpinned); got != s.unpinned {
t.Errorf("the reason for unpinned = %q", got)
}
}
// 1214/AC3: the builder has no hash of its own: its fingerprint is what delivery.SourceFingerprint says, whatever
// that is (a stand-in proves it is called), and the reuse key does not move with it.
func TestTheBuilderFingerprintsWithTheSDKFunction(t *testing.T) {
was := sourceFingerprint
t.Cleanup(func() { sourceFingerprint = was })
var got delivery.SourceInputs
sourceFingerprint = func(in delivery.SourceInputs) (string, string) {
got = in
return "src1:stand-in", ""
}
s := &sourceInputs{module: "m", tree: "t", bases: []string{"x@sha256:a"}, contexts: map[string]string{"c": "ct"}, toolchains: map[string]string{"c": "go"}}
if fp := s.fingerprint(); fp != "src1:stand-in" {
t.Fatalf("fingerprint = %q, not the SDK function's", fp)
}
if got.Module != "m" || got.Tree != "t" || len(got.Bases) != 1 || got.Contexts["c"] != "ct" || got.Toolchains["c"] != "go" {
t.Errorf("the SDK function was given %+v", got)
}
}
// 1215/AC1: the inputs a result records, read back, give the fingerprint the build recorded (the live builds of
// 1214/AC1), through encoding/json as the build record keeps them.
func TestTheInputsAResultRecordsGiveItsFingerprintAgain(t *testing.T) {
s := &sourceInputs{module: "build-agent", tree: "11c0b4f90cae0d6e4ef9cc48cbfdf9f6dbe52bca",
bases: []string{
"sha256:d9e853e87e55526f6b2917df91a2115c36dd7c696a35be12163d44e6e2a4b6bc",
"sha256:8ac98ca534ac3f51e1f420a1dd2c15e74c75cfa0f23f3ad27eb5d7236c349a0c",
},
contexts: map[string]string{"server": "narrow:d51ca744fd11af59d616c858f76c27b3e58e16e11fbdc0c720eb9d7d19d55fac"}}
raw, err := json.Marshal(resultInputs(s))
if err != nil {
t.Fatal(err)
}
var back delivery.SourceInputs
if err := json.Unmarshal(raw, &back); err != nil {
t.Fatal(err)
}
want := "src1:e846efe73716b78e80860baaac96153de2cb76551325753838ca245d2160a3d7"
if fp, none := delivery.SourceFingerprint(back); fp != want || none != "" {
t.Errorf("the inputs read back give %q (%q), want %q", fp, none, want)
}
}
// 1215/AC3: a build that gets no fingerprint still has its inputs recorded, so the reason is visible.
func TestABuildWithNoFingerprintStillSaysItsInputs(t *testing.T) {
s := newSourceInputs("app")
s.tree = "aaaa"
s.notPinned("it resolves packages from the mesh's registry at build time")
in := resultInputs(s)
if in == nil || in.Module != "app" || in.Tree != "aaaa" || in.Unpinned == "" {
t.Fatalf("the inputs of an unpinned build: %+v", in)
}
if fp, none := delivery.SourceFingerprint(*in); fp != "" || none != in.Unpinned {
t.Errorf("the inputs give %q (%q)", fp, none)
}
noTree := resultInputs(newSourceInputs("app"))
if noTree == nil || noTree.Module != "app" || noTree.Tree != "" {
t.Errorf("the inputs of a build with no tree: %+v", noTree)
}
}
+16 -13
View File
@@ -10,6 +10,8 @@ package chain
import (
"encoding/json"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// Ref names a build and the commit it was made from: one end of a change.
@@ -97,19 +99,20 @@ type Build struct {
Path string `json:"path,omitempty"`
Manifest json.RawMessage `json:"manifest,omitempty"`
// ManifestDigest names the manifest in the manifests bucket (ManifestKey); a record that has it keeps no Manifest.
ManifestDigest string `json:"manifest-digest,omitempty"`
Against []string `json:"against,omitempty"`
Mirrored []string `json:"mirrored,omitempty"`
Read []ReadRepository `json:"read,omitempty"`
Sources []Source `json:"sources,omitempty"`
SourceFingerprint string `json:"source-fingerprint,omitempty"`
Failed string `json:"failed,omitempty"`
Made []Artifact `json:"made,omitempty"`
Asked time.Time `json:"asked,omitzero"`
At time.Time `json:"at,omitzero"`
ReuseKey string `json:"reuse-key,omitempty"`
Published *Published `json:"published,omitempty"`
Request *Request `json:"request,omitempty"`
ManifestDigest string `json:"manifest-digest,omitempty"`
Against []string `json:"against,omitempty"`
Mirrored []string `json:"mirrored,omitempty"`
Read []ReadRepository `json:"read,omitempty"`
Sources []Source `json:"sources,omitempty"`
SourceFingerprint string `json:"source-fingerprint,omitempty"`
SourceInputs *delivery.SourceInputs `json:"source-inputs,omitempty"`
Failed string `json:"failed,omitempty"`
Made []Artifact `json:"made,omitempty"`
Asked time.Time `json:"asked,omitzero"`
At time.Time `json:"at,omitzero"`
ReuseKey string `json:"reuse-key,omitempty"`
Published *Published `json:"published,omitempty"`
Request *Request `json:"request,omitempty"`
}
// Commit is one repository's commit a rollout carries.
+3
View File
@@ -5,6 +5,7 @@ import (
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/novox/mesh-controller/internal/chain"
)
@@ -51,6 +52,8 @@ type Build struct {
// SourceFingerprint is what the build was made from, hashed, as its builder said it (novox/hq
// issue 280); empty from a builder that predates it, or where the source does not pin the build.
SourceFingerprint string
// SourceInputs are what it was computed from (novox/hq ADR 0437 §5); nil from a build recorded before they were.
SourceInputs *delivery.SourceInputs
// Failed is the builder's own words, empty when it worked.
Failed string
Made []Artifact
+2 -2
View File
@@ -23,7 +23,7 @@ import (
// toChainBuild is a build as the bucket keeps it.
func toChainBuild(b Build) chain.Build {
c := chain.Build{ID: b.ID, Repository: b.Repository, Ref: b.Ref, Module: b.Module, Commit: b.Commit, On: b.On,
Path: b.Path, Against: b.Against, Mirrored: b.Mirrored, SourceFingerprint: b.SourceFingerprint,
Path: b.Path, Against: b.Against, Mirrored: b.Mirrored, SourceFingerprint: b.SourceFingerprint, SourceInputs: b.SourceInputs,
ManifestDigest: b.ManifestDigest, Failed: b.Failed, Asked: b.Asked.UTC(), At: b.At.UTC(), ReuseKey: b.ReuseKey, Published: b.Published}
if len(b.Manifest) > 0 {
c.Manifest = json.RawMessage(b.Manifest)
@@ -46,7 +46,7 @@ func toChainBuild(b Build) chain.Build {
// fromChainBuild is a build as the mesh reads it.
func fromChainBuild(c chain.Build) Build {
b := Build{ID: c.ID, Repository: c.Repository, Ref: c.Ref, Module: c.Module, Commit: c.Commit, On: c.On, Path: c.Path,
Manifest: []byte(c.Manifest), ManifestDigest: c.ManifestDigest, Against: c.Against, Mirrored: c.Mirrored, SourceFingerprint: c.SourceFingerprint,
Manifest: []byte(c.Manifest), ManifestDigest: c.ManifestDigest, Against: c.Against, Mirrored: c.Mirrored, SourceFingerprint: c.SourceFingerprint, SourceInputs: c.SourceInputs,
Failed: c.Failed, Asked: c.Asked, At: c.At, ReuseKey: c.ReuseKey, Published: c.Published}
if len(b.Manifest) == 0 {
b.Manifest = nil
+54
View File
@@ -5,9 +5,11 @@ import (
"encoding/json"
"fmt"
"reflect"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-controller/internal/chain"
@@ -129,3 +131,55 @@ func TestTheBusBeingAwayIsNotNowLikeTheStoreBeingAway(t *testing.T) {
}
}
}
// 1215/AC1 and AC4: the inputs of a build's fingerprint are kept in the build's record, through the bucket's bytes, unchanged.
func TestABuildKeepsTheInputsOfItsSourceFingerprint(t *testing.T) {
in := &delivery.SourceInputs{Module: "notes", Tree: "t1", Bases: []string{"r/b@sha256:aa"},
Contexts: map[string]string{"server": "narrow:ff"}, Toolchains: map[string]string{"code": "go sha256:cc 0123"}}
fp, _ := delivery.SourceFingerprint(*in)
b := Build{ID: "build-1", Module: "notes", Commit: "c1", SourceFingerprint: fp, SourceInputs: in}
raw, err := json.Marshal(toChainBuild(b))
if err != nil {
t.Fatal(err)
}
var back chain.Build
if err := json.Unmarshal(raw, &back); err != nil {
t.Fatal(err)
}
got := fromChainBuild(back)
if !reflect.DeepEqual(got.SourceInputs, in) {
t.Fatalf("the inputs came back as %+v, not %+v", got.SourceInputs, in)
}
if again, _ := delivery.SourceFingerprint(*got.SourceInputs); again != got.SourceFingerprint {
t.Errorf("the inputs read back give %q, the record says %q", again, got.SourceFingerprint)
}
}
// 1215/AC3: a build with no fingerprint keeps its inputs, so the reason is visible.
func TestABuildWithNoFingerprintKeepsItsInputs(t *testing.T) {
in := &delivery.SourceInputs{Module: "notes", Tree: "t1", Unpinned: "it resolves packages when built"}
raw, _ := json.Marshal(toChainBuild(Build{ID: "build-1", Module: "notes", SourceInputs: in}))
var back chain.Build
if err := json.Unmarshal(raw, &back); err != nil {
t.Fatal(err)
}
if got := fromChainBuild(back); got.SourceFingerprint != "" || !reflect.DeepEqual(got.SourceInputs, in) {
t.Errorf("read back as %q with %+v", got.SourceFingerprint, got.SourceInputs)
}
}
// 1215/AC2: a record written before the inputs were kept reads as before, with none and no error.
func TestABuildRecordFromBeforeTheInputsReadsAsBefore(t *testing.T) {
var back chain.Build
old := `{"id":"build-1","module":"notes","commit":"c1","source-fingerprint":"src1:aa","reuse-key":"rk"}`
if err := json.Unmarshal([]byte(old), &back); err != nil {
t.Fatal(err)
}
got := fromChainBuild(back)
if got.SourceInputs != nil || got.SourceFingerprint != "src1:aa" || got.ReuseKey != "rk" {
t.Errorf("an old record reads as %+v", got)
}
if raw, _ := json.Marshal(toChainBuild(got)); strings.Contains(string(raw), "source-inputs") {
t.Errorf("an old record is written with inputs: %s", raw)
}
}
+13 -4
View File
@@ -5,6 +5,8 @@ import (
"strconv"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// Asking a machine to build a module, and hearing what came out.
@@ -140,10 +142,14 @@ type PullRef struct {
// CheckRequest is what a merge check needs beyond the repository and its head.
type CheckRequest struct {
Owner string `json:"owner"`
Repo string `json:"repo"`
Number int `json:"number,omitempty"`
Base string `json:"base,omitempty"`
// Pulls are the buckets of the pull requests' keys, one per module claiming the delivery seat, in the order of
// their names (novox/hq ADR 0437 §3): the build agent, which holds no catalogue, reads and watches the key in
// them. Empty from a controller from before it, and then mesh-delivery's.
Pulls []string `json:"pulls,omitempty"`
Owner string `json:"owner"`
Repo string `json:"repo"`
Number int `json:"number,omitempty"`
Base string `json:"base,omitempty"`
// Trunk is the base branch's head the check is asked against, as the forge announced it with the head (novox/hq ADR
// 0406 §4): with the head, what the pull request's key says the current check is for, and what a waiting request
// is dropped by and a running check stopped by when the key moves.
@@ -410,6 +416,9 @@ type BuildResult struct {
// byte-reproducible. Empty from a builder that predates it, or where the source does not pin the
// build; then only identical artifacts make a rebuild no move.
SourceFingerprint string `json:"source-fingerprint,omitempty"`
// SourceInputs are what SourceFingerprint was computed from (novox/hq ADR 0437 §5, issue 1215), as encoding/json
// writes the SDK's type; also set where the source does not pin the build.
SourceInputs *delivery.SourceInputs `json:"source-inputs,omitempty"`
// Delivery is the how-manyth time the queue handed this ask out (BuildStart.Delivery): more than 1
// says the outcome is from a run started again after an earlier taker left it (novox/hq issue 594).
+1 -1
View File
@@ -439,7 +439,7 @@ func (m *natsMachine) judgedAlready(ctx context.Context, r BuildRequest) string
return ""
}
if c.Group == "" {
key, found, err := ReadPullKey(ctx, m.js.Conn(), c.Owner, c.Repo, c.Number)
key, found, err := ReadPullKey(ctx, m.js.Conn(), PullBucketsOf(c), c.Owner, c.Repo, c.Number)
if err != nil {
fmt.Fprintf(os.Stderr, "could not read the key of %s/%s#%d, so its check runs: %v\n", c.Owner, c.Repo, c.Number, err)
} else if found && key.Judged(r.Ref, c.Trunk) {
+73 -17
View File
@@ -91,28 +91,79 @@ func (k PullKey) Judged(head, trunk string) bool {
return judgedBy && sameCommit(k.Head, head) && sameCommit(k.Trunk, trunk) && trunk != ""
}
// ReadPullKey reads a pull request's key from mesh-delivery's bucket: false when it has none.
func ReadPullKey(ctx context.Context, conn *nats.Conn, owner, repo string, number int) (PullKey, bool, error) {
// PullBucketsOf are the buckets a check's pull request's key is looked for in: the ones its request names, and
// mesh-delivery's when it names none (a controller from before the request named them).
func PullBucketsOf(c *CheckRequest) []string {
if c == nil || len(c.Pulls) == 0 {
return []string{broker.PullsBucket}
}
return c.Pulls
}
// ReadPullKey reads a pull request's key from the named buckets in turn: the first that holds it answers, a bucket that
// does not exist is passed over, and false when none holds it.
func ReadPullKey(ctx context.Context, conn *nats.Conn, buckets []string, owner, repo string, number int) (PullKey, bool, error) {
api, err := jetstream.New(conn)
if err != nil {
return PullKey{}, false, err
}
kv, err := api.KeyValue(ctx, broker.PullsBucket)
if err != nil {
return PullKey{}, false, fmt.Errorf("the pull requests' keys: %w", err)
return readPullKeyIn(ctx, buckets, func(ctx context.Context, bucket string) (func(context.Context, string) ([]byte, error), error) {
kv, err := api.KeyValue(ctx, bucket)
if err != nil {
return nil, err
}
return func(ctx context.Context, key string) ([]byte, error) {
e, err := kv.Get(ctx, key)
if err != nil {
return nil, err
}
return e.Value(), nil
}, nil
}, owner, repo, number)
}
// readPullKeyIn is ReadPullKey over buckets opened by open.
func readPullKeyIn(ctx context.Context, buckets []string,
open func(ctx context.Context, bucket string) (func(context.Context, string) ([]byte, error), error),
owner, repo string, number int) (PullKey, bool, error) {
for _, bucket := range buckets {
get, err := open(ctx, bucket)
if errors.Is(err, jetstream.ErrBucketNotFound) {
continue
}
if err != nil {
return PullKey{}, false, fmt.Errorf("the pull requests' keys in %s: %w", bucket, err)
}
value, err := get(ctx, PullKeyOf(owner, repo, number))
if errors.Is(err, jetstream.ErrKeyNotFound) {
continue
}
if err != nil {
return PullKey{}, false, err
}
var key PullKey
if err := json.Unmarshal(value, &key); err != nil {
return PullKey{}, false, fmt.Errorf("the key of %s/%s#%d is not a pull request's: %w", owner, repo, number, err)
}
return key, true, nil
}
e, err := kv.Get(ctx, PullKeyOf(owner, repo, number))
if errors.Is(err, jetstream.ErrKeyNotFound) {
return PullKey{}, false, nil
return PullKey{}, false, nil
}
// firstBucketThatExists is the first of the buckets that exists, as exists says; an error other than a missing bucket
// stands.
func firstBucketThatExists(ctx context.Context, buckets []string, exists func(ctx context.Context, bucket string) error) (string, error) {
for _, bucket := range buckets {
err := exists(ctx, bucket)
if errors.Is(err, jetstream.ErrBucketNotFound) {
continue
}
if err != nil {
return "", err
}
return bucket, nil
}
if err != nil {
return PullKey{}, false, err
}
var key PullKey
if err := json.Unmarshal(e.Value(), &key); err != nil {
return PullKey{}, false, fmt.Errorf("the key of %s/%s#%d is not a pull request's: %w", owner, repo, number, err)
}
return key, true, nil
return "", fmt.Errorf("none of the pull requests' buckets %v exists", buckets)
}
// WatchPull calls moved once when the pull request's key moves on from what this check was asked for — its head or its
@@ -129,7 +180,12 @@ func WatchPull(ctx context.Context, conn *nats.Conn, r BuildRequest, moved func(
if err != nil {
return nil, err
}
kv, err := api.KeyValue(ctx, broker.PullsBucket)
var kv jetstream.KeyValue
_, err = firstBucketThatExists(ctx, PullBucketsOf(c), func(ctx context.Context, bucket string) error {
var err error
kv, err = api.KeyValue(ctx, bucket)
return err
})
if err != nil {
return nil, fmt.Errorf("the pull requests' keys: %w", err)
}
+102
View File
@@ -0,0 +1,102 @@
package link
import (
"bytes"
"context"
"log"
"strings"
"testing"
"github.com/novox/mesh-controller/internal/broker"
)
// The controller hears a merge's announcement from whichever module claims the delivery seat, so the subject is a
// wildcard and the emitter is read from it (novox/hq ADR 0437 §3, issue 1211).
type keptMerges struct {
emitters []string
refuse bool
}
func (k *keptMerges) KeepDeliveryMerge(_ context.Context, emitter string, _ []byte) error {
if k.refuse {
return ErrNotTheDeliverysWord
}
k.emitters = append(k.emitters, emitter)
return nil
}
func mergeSays(t *testing.T, in *fakeInbound, to *settled, subject string) Control {
t.Helper()
m := in.sends(t, to, KindDeliveryMerged, map[string]any{"owner": "novox"}).(*fakeControl)
m.subject = subject
return m
}
// 1211/AC1: a merged announcement on the subject of the module claiming the seat is read as the delivery's word and
// kept, as today's module's is (3: unchanged).
func TestAMergedAnnouncementIsKeptWhicheverModuleSaidIt(t *testing.T) {
s, in := serving()
kept := &keptMerges{}
if err := s.KeepsDeliveryMerges(kept); err != nil {
t.Fatal(err)
}
for _, module := range []string{"mesh-delivery-gate", "mesh-delivery"} { // 1211/AC1, 1211/AC3
subject := "mesh.mod." + module + ".event.merged"
if kind, ok := kindOfSubject(subject); !ok || kind != KindDeliveryMerged {
t.Errorf("%s decodes to %q %v", subject, kind, ok)
}
if got, ok := DeliveryMergedEmitter(subject); !ok || got != module {
t.Errorf("%s: emitter %q %v", subject, got, ok)
}
to := &settled{}
s.act(t.Context(), mergeSays(t, in, to, subject))
if !to.acked {
t.Errorf("%s: not taken: %+v", module, to)
}
}
if len(kept.emitters) != 2 || kept.emitters[0] != "mesh-delivery-gate" || kept.emitters[1] != "mesh-delivery" {
t.Errorf("the keeper was told the emitters %v", kept.emitters)
}
for _, subject := range []string{"mesh.mod.a.b.event.merged", "mesh.mod.x.event.merged.more", "mesh.seat.x.event.merged", "mesh.mod..event.merged"} {
if _, ok := DeliveryMergedEmitter(subject); ok {
t.Errorf("%s read as a merge announcement", subject)
}
}
}
// 1211/AC2: an announcement from a module that claims no delivery seat is not the delivery's word: nothing is kept
// and the log names the module that said it.
func TestAMergedAnnouncementFromAModuleNotClaimingTheSeatIsNamedAndNotKept(t *testing.T) {
s, in := serving()
var said bytes.Buffer
s.log = log.New(&said, "", 0)
if err := s.KeepsDeliveryMerges(&keptMerges{refuse: true}); err != nil {
t.Fatal(err)
}
to := &settled{}
s.act(t.Context(), mergeSays(t, in, to, "mesh.mod.gitea.event.merged"))
if !to.rejected || to.acked || len(in.held) != 0 {
t.Errorf("settled %+v, held %d", to, len(in.held))
}
if !strings.Contains(said.String(), "gitea") {
t.Errorf("the log does not name the module that said it: %q", said.String())
}
}
// 1211/AC4: the controller's consumer filter hears merged from any module, on one wildcard subject.
func TestTheControllerHearsMergedFromAnyModuleOnOneSubject(t *testing.T) {
if broker.DeliveryMergedSubject != "mesh.mod.*.event.merged" {
t.Errorf("the subject is %q", broker.DeliveryMergedSubject)
}
found := false
for _, s := range broker.ControllerHears {
found = found || s == broker.DeliveryMergedSubject
}
if !found {
t.Errorf("the controller hears %v", broker.ControllerHears)
}
if got := broker.DeliveryMergedSubjectOf("mesh-delivery-gate"); got != "mesh.mod.mesh-delivery-gate.event.merged" {
t.Errorf("a module's own subject is %q", got)
}
}
+113
View File
@@ -0,0 +1,113 @@
package link
import (
"context"
"encoding/json"
"errors"
"reflect"
"strings"
"testing"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
)
// The build agent reads and watches the pull requests' keys in the buckets the check request names (novox/hq ADR 0437
// §3, issue 1212): the controller knows the seat's claimants, the build agent holds no catalogue.
// 1212/AC4: a request naming no bucket reads mesh-delivery's, as before; the field is optional on the wire.
func TestARequestNamingNoPullsBucketUsesMeshDeliverys(t *testing.T) {
if got := PullBucketsOf(&CheckRequest{}); !reflect.DeepEqual(got, []string{broker.PullsBucket}) {
t.Errorf("buckets %v", got)
}
if got := PullBucketsOf(nil); !reflect.DeepEqual(got, []string{broker.PullsBucket}) {
t.Errorf("buckets of no check %v", got)
}
named := &CheckRequest{Pulls: []string{"mesh-delivery-gate_pulls"}}
if got := PullBucketsOf(named); !reflect.DeepEqual(got, named.Pulls) {
t.Errorf("buckets %v", got)
}
raw := mustJSON(t, CheckRequest{Owner: "o"})
if contains(string(raw), "pulls") {
t.Errorf("an empty field is written: %s", raw)
}
}
// opened is a set of buckets by name, each holding keys; a bucket not in it does not exist.
func opened(buckets map[string]map[string]string, opens *[]string) func(context.Context, string) (func(context.Context, string) ([]byte, error), error) {
return func(_ context.Context, bucket string) (func(context.Context, string) ([]byte, error), error) {
*opens = append(*opens, bucket)
keys, ok := buckets[bucket]
if !ok {
return nil, jetstream.ErrBucketNotFound
}
return func(_ context.Context, key string) ([]byte, error) {
v, ok := keys[key]
if !ok {
return nil, jetstream.ErrKeyNotFound
}
return []byte(v), nil
}, nil
}
}
// 1212/AC3: the buckets are read in turn, the first holding the key answers, a missing bucket is passed over, and
// none holding it is no key.
func TestThePullKeyIsReadFromTheNamedBucketsInTurn(t *testing.T) {
key := PullKeyOf("novox", "a", 7)
both := map[string]map[string]string{
"mesh-delivery_pulls": {key: `{"head":"old"}`},
"mesh-delivery-gate_pulls": {key: `{"head":"new"}`},
}
var opens []string
got, found, err := readPullKeyIn(t.Context(), []string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, opened(both, &opens), "novox", "a", 7)
if err != nil || !found || got.Head != "new" || len(opens) != 1 {
t.Errorf("first holding answers: %+v %v %v, opened %v", got, found, err, opens)
}
opens = nil
only := map[string]map[string]string{"mesh-delivery_pulls": {key: `{"head":"old"}`}, "mesh-delivery-gate_pulls": {}}
got, found, err = readPullKeyIn(t.Context(), []string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, opened(only, &opens), "novox", "a", 7)
if err != nil || !found || got.Head != "old" || len(opens) != 2 {
t.Errorf("a bucket without the key is passed over: %+v %v %v, opened %v", got, found, err, opens)
}
opens = nil
got, found, err = readPullKeyIn(t.Context(), []string{"gone_pulls", "mesh-delivery_pulls"}, opened(only, &opens), "novox", "a", 7)
if err != nil || !found || got.Head != "old" {
t.Errorf("a missing bucket is passed over: %+v %v %v", got, found, err)
}
_, found, err = readPullKeyIn(t.Context(), []string{"gone_pulls", "mesh-delivery-gate_pulls"}, opened(only, &opens), "novox", "a", 8)
if err != nil || found {
t.Errorf("none holding it: found %v, %v", found, err)
}
boom := errors.New("bus away")
_, _, err = readPullKeyIn(t.Context(), []string{"x"}, func(context.Context, string) (func(context.Context, string) ([]byte, error), error) {
return nil, boom
}, "novox", "a", 7)
if !errors.Is(err, boom) {
t.Errorf("a real failure is an error: %v", err)
}
// The watch takes the first named bucket that exists.
opens = nil
bucket, err := firstBucketThatExists(t.Context(), []string{"gone_pulls", "mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, func(_ context.Context, b string) error {
opens = append(opens, b)
if b == "gone_pulls" {
return jetstream.ErrBucketNotFound
}
return nil
})
if err != nil || bucket != "mesh-delivery-gate_pulls" || len(opens) != 2 {
t.Errorf("watch bucket %q %v, tried %v", bucket, err, opens)
}
}
func mustJSON(t *testing.T, v any) string {
t.Helper()
raw, err := json.Marshal(v)
if err != nil {
t.Fatal(err)
}
return string(raw)
}
func contains(s, sub string) bool { return strings.Contains(s, sub) }
+18 -2
View File
@@ -296,8 +296,6 @@ func kindOfSubject(subject string) (string, bool) {
return KindPullUpdated, true
case broker.DecidedSubject:
return KindDecided, true
case broker.DeliveryMergedSubject:
return KindDeliveryMerged, true
case BuildOutcome(), BuildOutcomeOf(TheBuildMachineBefore):
// A build's outcome is the role's event now, so it arrives on the events stream rather than
// the control branch — and is acted on by the same handler, because what the controller does
@@ -305,12 +303,30 @@ func kindOfSubject(subject string) (string, bool) {
// runs (ADR 0190): the old builder still answers on the retired seat until it is unassigned.
return KindBuilt, true
}
if _, ok := DeliveryMergedEmitter(subject); ok {
return KindDeliveryMerged, true
}
if _, ok := ProvisionerEmitter(subject); ok {
return KindProvisioner, true
}
return "", false
}
// DeliveryMergedEmitter is the module a merge's announcement came from, read from its subject
// (`mesh.mod.<module>.event.merged`); false for any other subject. The controller's own follow pattern, with `*`
// for the module, decodes too.
func DeliveryMergedEmitter(subject string) (string, bool) {
rest, ok := strings.CutPrefix(subject, "mesh.mod.")
if !ok {
return "", false
}
module, event, ok := strings.Cut(rest, ".event.")
if !ok || module == "" || strings.Contains(module, ".") || event != broker.DeliveryMergedEvent {
return "", false
}
return module, true
}
// ProvisionerEmitter is the module a provider's standing event came from, read from its subject
// (`mesh.mod.<module>.event.provisioner.<failing|recovered|retirement>`); false for any other subject. The
// controller's own follow pattern, with `*` for the module, decodes too.
+16 -2
View File
@@ -159,10 +159,16 @@ func (s *Server) Decides(d Decider) error {
// DeliveryMergeKeeper is what the controller does with mesh-delivery's word that it merged a pull request (novox/hq
// ADR 0420 §7): keep it, to be read when a moved module's build is taken in.
//
// The emitter is the module the announcement's subject names; a keeper that finds it does not claim the delivery seat
// answers ErrNotTheDeliverysWord (novox/hq ADR 0437 §3).
type DeliveryMergeKeeper interface {
KeepDeliveryMerge(ctx context.Context, body []byte) error
KeepDeliveryMerge(ctx context.Context, emitter string, body []byte) error
}
// ErrNotTheDeliverysWord is a keeper's answer to a merge's announcement from a module that claims no delivery seat.
var ErrNotTheDeliverysWord = errors.New("the module claims no delivery seat")
// KeepsDeliveryMerges says what to do about mesh-delivery's merges, and asks for them to be delivered.
func (s *Server) KeepsDeliveryMerges(k DeliveryMergeKeeper) error {
if err := s.inbound.Also(KindDeliveryMerged); err != nil {
@@ -730,7 +736,15 @@ func (s *Server) deliveryMerged(ctx context.Context, m Control) {
_ = m.Took()
return
}
err := s.deliveryMerges.KeepDeliveryMerge(ctx, m.Body())
emitter, _ := DeliveryMergedEmitter(m.Subject())
err := s.deliveryMerges.KeepDeliveryMerge(ctx, emitter, m.Body())
if errors.Is(err, ErrNotTheDeliverysWord) {
// A merged event of a module that claims no delivery seat: not the delivery's word, so nothing is kept (and
// handing it back would not change that).
s.log.Printf("a word on a merge from %q is not the delivery's, and nothing is kept: %v", emitter, err)
_ = m.Drop()
return
}
switch s.decide(ctx, m, "mesh-delivery's word on a merge", "", "", err) {
case Hold:
return
+35
View File
@@ -0,0 +1,35 @@
package link
import (
"encoding/json"
"reflect"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// 1215/AC4: the builder's result on the wire carries the inputs, under source-inputs, and reads back unchanged.
func TestABuildResultCarriesTheSourceInputs(t *testing.T) {
in := &delivery.SourceInputs{Module: "m", Tree: "t", Bases: []string{"x@sha256:a"},
Contexts: map[string]string{"c": "narrow:1"}, Toolchains: map[string]string{"c": "go sha256:b 01"}, Unpinned: "why"}
raw, err := json.Marshal(BuildResult{ID: "build-1", SourceInputs: in})
if err != nil {
t.Fatal(err)
}
var wire map[string]json.RawMessage
if err := json.Unmarshal(raw, &wire); err != nil || wire["source-inputs"] == nil {
t.Fatalf("the result on the wire has no source-inputs: %s", raw)
}
var back BuildResult
if err := json.Unmarshal(raw, &back); err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(back.SourceInputs, in) {
t.Errorf("read back as %+v, not %+v", back.SourceInputs, in)
}
raw, _ = json.Marshal(BuildResult{ID: "build-2"})
wire = nil
if err := json.Unmarshal(raw, &wire); err != nil || wire["source-inputs"] != nil {
t.Errorf("a result with none writes source-inputs: %s", raw)
}
}
+1 -1
View File
@@ -9,7 +9,7 @@ a second place to clean besides the catalogue itself.
mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile
mesh-host 9d9e3a46c4f5943cfffb413d962ce48b75e2fd39 examples/foundation-first-node.lock
mesh-sdk 238440d6f8c46a8dd1d829f4313d7b035a6ca847 conformance/events/module-event.json
mesh-sdk aa68b67389c964ea3e573bc50630c77d6518cdab conformance/events/module-event.json
To move them, from this repository's root, with the two repositories checked out beside it:
+149
View File
@@ -0,0 +1,149 @@
package asks
import (
"encoding/json"
"errors"
"fmt"
"reflect"
"strings"
)
// ConditionSubject is where a module says a condition to the operator: a by-caller accept of the operator-channel
// seat, `mesh.seat.operator-channel.accept.condition.<publisher>` (novox/hq ADR 0437 §9). The router reads the
// verb at the fifth part and the caller at the sixth, so the publisher is one token, the module's own name; it is
// refused when empty or when it holds a dot, a wildcard or whitespace.
func ConditionSubject(publisher string) (string, error) {
if publisher == "" {
return "", errors.New("asks: the condition's publisher is empty")
}
if strings.ContainsAny(publisher, ".*> \t\r\n") || strings.ContainsFunc(publisher, func(r rune) bool { return r <= ' ' }) {
return "", fmt.Errorf("asks: the condition's publisher %q is not one token: it holds a dot, a wildcard or whitespace", publisher)
}
return "mesh.seat." + Seat + ".accept.condition." + publisher, nil
}
// ConditionVersion is the version of the condition value this package writes and the only one it reads.
const ConditionVersion = 1
// ErrUnknownConditionVersion is what DecodeCondition returns for a value whose version is missing or not
// ConditionVersion. Test it with errors.Is.
var ErrUnknownConditionVersion = errors.New("asks: unknown condition version")
// The words a condition's explanation opens with (novox/hq ADR 0253 §1).
const (
NothingToDo = "Nothing for you to do."
NeedsYou = "Needs you: "
)
// The severities and states of a condition.
const (
SeverityUrgent = "urgent"
SeverityWarning = "warning"
StateRaised = "raised"
StateCleared = "cleared"
)
// Scope is what a condition is about, under the controller's names.
type Scope struct {
Scope string `json:"scope"`
ID string `json:"id"`
Machine string `json:"machine,omitempty"`
}
// Action is one answer a notification offers, under the controller's names.
type Action struct {
Label string `json:"label"`
Verb string `json:"verb"`
Machine string `json:"machine,omitempty"`
Arguments map[string]string `json:"arguments,omitempty"`
Level Level `json:"level,omitempty"`
}
// Condition is a condition a module says to the operator, under the names of the controller's conditions: its key,
// severity, subject, state (raised or cleared), summary and the plain words of ADR 0253. A cleared condition needs
// only its key, subject and state. The router alone chooses the channels.
type Condition struct {
V int `json:"v"`
Key string `json:"key"`
Severity string `json:"severity,omitempty"`
Subject Scope `json:"subject"`
State string `json:"state"`
Summary string `json:"summary,omitempty"`
Headline string `json:"headline,omitempty"`
Needs string `json:"needs,omitempty"`
Explanation string `json:"explanation,omitempty"`
Resolved string `json:"resolved,omitempty"`
Actions []Action `json:"actions,omitempty"`
}
func (c Condition) check() error {
if c.Key == "" {
return errors.New("asks: a condition has no key")
}
if c.Subject.Scope == "" || c.Subject.ID == "" {
return errors.New("asks: a condition's subject needs its scope and id")
}
switch c.State {
case StateCleared:
return nil
case StateRaised:
default:
return fmt.Errorf("asks: the condition's state %q is not raised or cleared", c.State)
}
switch c.Severity {
case SeverityUrgent, SeverityWarning:
default:
return fmt.Errorf("asks: the condition's severity %q is not urgent or warning", c.Severity)
}
if c.Headline == "" {
return errors.New("asks: a raised condition has no headline")
}
if n := len([]rune(c.Headline)); n > HeadlineLength {
return fmt.Errorf("asks: the condition's headline is %d characters, more than %d", n, HeadlineLength)
}
verdict := NothingToDo
if c.Needs != "" {
verdict = NeedsYou + c.Needs
}
if !strings.HasPrefix(c.Explanation, verdict) {
return fmt.Errorf("asks: the condition's explanation must open with %q", verdict)
}
return nil
}
// EncodeCondition is c as JSON at ConditionVersion; it refuses what ADR 0253 and ADR 0437 §9 refuse.
func EncodeCondition(c Condition) ([]byte, error) {
if err := c.check(); err != nil {
return nil, err
}
c.V = ConditionVersion
return json.Marshal(c)
}
// DecodeCondition reads a condition; it refuses an unknown version (ErrUnknownConditionVersion) and what
// EncodeCondition refuses.
func DecodeCondition(data []byte) (Condition, error) {
var probe struct {
V *int `json:"v"`
}
if err := json.Unmarshal(data, &probe); err != nil {
return Condition{}, fmt.Errorf("asks: the condition is not JSON: %w", err)
}
if probe.V == nil {
return Condition{}, fmt.Errorf("%w: no version in the value, this reader reads %d", ErrUnknownConditionVersion, ConditionVersion)
}
if *probe.V != ConditionVersion {
return Condition{}, fmt.Errorf("%w: the value is version %d, this reader reads %d", ErrUnknownConditionVersion, *probe.V, ConditionVersion)
}
var c Condition
if err := json.Unmarshal(data, &c); err != nil {
return Condition{}, fmt.Errorf("asks: the condition does not fit its shape: %w", err)
}
if err := c.check(); err != nil {
return Condition{}, err
}
return c, nil
}
// SameCondition says whether two conditions are the same value.
func SameCondition(a, b Condition) bool { return reflect.DeepEqual(a, b) }
+13 -1
View File
@@ -18,7 +18,19 @@
// question 9 item 7): a writer writes a new version only once every reader's have shows a version that reads it. A
// writer writes only when the new value differs from the one before (the Same functions).
//
// Standard library only.
// **Source, candidates, modules and the forge** (issue 1200; ADR 0437 §5, release 3): the gate's bucket `source` with
// its one key `main` (every repository's merged head and the group that moved it), the indexer's `candidates.<group>`
// and `modules.<module>` (a module's source fingerprint, or none and why), and the forge bridge's messages named
// `forge.<owner>.<name>.<kind>` (source.go).
//
// **Merges, asks, owed and answers** (issue 1201; ADR 0437 §6, §8): the gate's `merges.<group>`, its asks
// `merge.<group>.<heads hash>` (the router's ask whole), its owed keys `<repository>.<number>.<head>.<act>` that the
// forge bridge performs, and the teller's answer under the ask's own key (gate.go).
//
// **The source fingerprint** (issue 1203; ADR 0437 §5): one function of named inputs that the controller's builder and
// the indexer both call, so they agree byte for byte (fingerprint.go).
//
// Standard library only, and this module's own asks package.
package delivery
import (
@@ -0,0 +1,63 @@
package delivery
import (
"crypto/sha256"
"encoding/hex"
"sort"
"strings"
)
// SourceInputs are what a module's source fingerprint is of (novox/hq ADR 0437 §5): the module, the git tree of
// its directory, its bases (references, of which only the digest counts), the root tree of each repository its
// artifacts read (by artifact), each artifact's toolchain line, and the reason the source does not pin the build
// when it does not (a build that resolves packages when built). A context is the value the builder records for it:
// a repository's root tree, or for an image that compiles a Go program the `narrow:` hash of the build source it was
// handed (mesh-controller internal/builder/builder.go:1146-1156).
type SourceInputs struct {
Module string
Tree string
Bases []string
Contexts map[string]string
Toolchains map[string]string
Unpinned string
}
// SourceFingerprint is the module's source fingerprint, `src1:` and the hex sha256 of its inputs' lines, or none
// and the reason when the source does not pin the build: Unpinned when set, "no tree" when Tree is empty. It is the
// one function the controller's builder and the indexer both call, so they agree byte for byte. The lines are
// "module <m>" and "tree <t>", then "base <digest>" (once each), "context <artifact> <tree>" and "toolchain
// <artifact> <line>" sorted, all joined by a newline.
func SourceFingerprint(in SourceInputs) (fp, none string) {
if in.Unpinned != "" {
return "", in.Unpinned
}
if in.Tree == "" {
return "", "no tree"
}
lines := []string{"module " + in.Module, "tree " + in.Tree}
bases := map[string]bool{}
for _, b := range in.Bases {
bases[ReferenceDigest(b)] = true
}
for b := range bases {
lines = append(lines, "base "+b)
}
for a, t := range in.Contexts {
lines = append(lines, "context "+a+" "+t)
}
for a, t := range in.Toolchains {
lines = append(lines, "toolchain "+a+" "+t)
}
sort.Strings(lines[2:])
sum := sha256.Sum256([]byte(strings.Join(lines, "\n")))
return SourcePrefix + hex.EncodeToString(sum[:]), ""
}
// ReferenceDigest is a reference's digest, `sha256:...`, so the registry address a base was copied into does not
// enter the fingerprint; the reference itself when it carries none.
func ReferenceDigest(reference string) string {
if _, digest, pinned := strings.Cut(reference, "@"); pinned && digest != "" {
return digest
}
return reference
}
+404
View File
@@ -0,0 +1,404 @@
package delivery
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"reflect"
"regexp"
"strconv"
"strings"
"git.novox.be/novox/mesh-sdk/go/asks"
)
// The gate's buckets and the teller's (ADR 0437 §6, §8). The gate writes merges (each pull request's expected
// head, its state and each merge commit), its asks (`merge.<group>.<heads hash>`, the router's ask whole) and owed
// (one key per act it owes the forge, which the forge bridge performs once per revision). The teller writes its
// answer under the ask's own key in its bucket answers.
const (
MergesBucket = GateModule + "_merges"
AsksBucket = GateModule + "_asks"
OwedBucket = GateModule + "_owed"
AnswersBucket = TellerModule + "_answers"
MergesHistory = 10
AsksHistory = 5
OwedHistory = 5
AnswersHistory = 5
)
// AskKindMerge is the kind of the gate's ask for a group's merge, the first part of its key.
const AskKindMerge = "merge"
// The acts the gate owes the forge, the last part of an owed key.
const (
OwedStatus = "status"
OwedView = "view"
OwedNote = "note"
OwedComment = "comment"
)
// The states of a pull request in a merge.
type MergeState string
// The states a merge's pull request is in.
const (
MergeMerging MergeState = "merging"
MergeMerged MergeState = "merged"
MergeRefused MergeState = "refused"
)
// The states a status the gate owes the forge may carry.
var statusStates = []string{"pending", "success", "failure", "error", "warning"}
var hashOK = regexp.MustCompile(`^[0-9a-f]{12}$`)
// MergesKey is the key, in MergesBucket, of a group: `<group>`.
func MergesKey(group string) (string, error) {
if err := checkName("group", group); err != nil {
return "", err
}
return group, nil
}
// MergeAskKey is the key, in AsksBucket, of the ask for merging a group at these heads: `merge.<group>.<heads
// hash>`. The teller's answer has the same key in AnswersBucket (AnswerKey).
func MergeAskKey(group string, heads []Head) (string, error) {
if err := checkName("group", group); err != nil {
return "", err
}
hash, err := HeadsHash(heads)
if err != nil {
return "", err
}
return AskKindMerge + "." + group + "." + hash, nil
}
// ParseAskKey is the inverse of MergeAskKey: the kind, the group and the heads hash.
func ParseAskKey(key string) (kind, group, hash string, err error) {
parts := strings.Split(key, ".")
if len(parts) != 3 || parts[0] != AskKindMerge {
return "", "", "", fmt.Errorf("delivery: %q is not an ask key, merge.<group>.<heads hash>", key)
}
if err := checkName("group", parts[1]); err != nil {
return "", "", "", fmt.Errorf("delivery: %q is not an ask key: %w", key, err)
}
if !hashOK.MatchString(parts[2]) {
return "", "", "", fmt.Errorf("delivery: %q is not an ask key: %q is not 12 hex digits", key, parts[2])
}
return parts[0], parts[1], parts[2], nil
}
// AnswerKey is the key, in AnswersBucket, of the teller's answer to the ask at askKey: the same string.
func AnswerKey(askKey string) string { return askKey }
// AskID is the id the teller gives the router for the ask at askKey. The router's ids refuse dots (asks.UsableID),
// so in the readable part a dot becomes `-` and any other byte outside letters, digits, `_` and `-` becomes `_`; that part is cut to fit, and every id ends in `-` and
// the first 12 hex digits of the sha256 of the whole key, so two keys never give one id. It cannot be read back:
// the teller keeps the key beside it.
func AskID(askKey string) string {
const most = 64 // the router's bound on an id
const tail = 13 // "-" and 12 hex digits
readable := []byte(askKey)
for i, c := range readable {
switch {
case c >= 'A' && c <= 'Z', c >= 'a' && c <= 'z', c >= '0' && c <= '9':
case c == '.':
readable[i] = '-'
default: // a slash, an equals sign or any other byte the router's id refuses
readable[i] = '_'
}
}
if len(readable) > 0 && (readable[0] == '-' || readable[0] == '_') {
readable[0] = 'k' // an id starts with a letter or digit
}
if len(readable) > most-tail {
readable = readable[:most-tail]
}
sum := sha256.Sum256([]byte(askKey))
return string(readable) + "-" + hex.EncodeToString(sum[:])[:12]
}
func checkOwedAct(act string) (kind, context string, err error) {
kind, context, hasContext := strings.Cut(act, "=")
switch kind {
case OwedStatus:
if !hasContext {
return "", "", fmt.Errorf("delivery: the owed act %q is a status without its context, status=<context>", act)
}
if err := checkName("status context", context); err != nil {
return "", "", err
}
return kind, context, nil
case OwedView, OwedNote, OwedComment:
if hasContext {
return "", "", fmt.Errorf("delivery: the owed act %q takes no context", act)
}
return kind, "", nil
}
return "", "", fmt.Errorf("delivery: the owed act %q is not status=<context>, view, note or comment", act)
}
// ParseOwedAct splits an owed act into its kind and, for a status, its context.
func ParseOwedAct(act string) (kind, context string, err error) { return checkOwedAct(act) }
// OwedKey is the key, in OwedBucket, of an act the gate owes the forge on a pull request at a head:
// `<repository>.<number>.<head>.<act>`, the act being `status=<context>` (mesh/merge-gate), `view`, `note` or
// `comment`.
func OwedKey(repository string, number int, head, act string) (string, error) {
if err := checkRepository(repository); err != nil {
return "", err
}
if number <= 0 {
return "", fmt.Errorf("delivery: an owed key needs the pull request number, not %d", number)
}
if err := checkName("head", head); err != nil {
return "", err
}
if _, _, err := checkOwedAct(act); err != nil {
return "", err
}
return repository + "." + strconv.Itoa(number) + "." + head + "." + act, nil
}
// ParseOwedKey is the inverse of OwedKey; act is whole, as `status=<context>` for a status.
func ParseOwedKey(key string) (repository string, number int, head, act string, err error) {
parts := strings.Split(key, ".")
if len(parts) != 4 {
return "", 0, "", "", fmt.Errorf("delivery: %q is not an owed key, <repository>.<number>.<head>.<act>", key)
}
number, convErr := strconv.Atoi(parts[1])
if convErr != nil {
return "", 0, "", "", fmt.Errorf("delivery: %q is not an owed key: %q is not a pull request number", key, parts[1])
}
if _, err := OwedKey(parts[0], number, parts[2], parts[3]); err != nil {
return "", 0, "", "", fmt.Errorf("delivery: %q is not an owed key: %w", key, err)
}
return parts[0], number, parts[2], parts[3], nil
}
// Merge is the value of a merges key: the ask approved and, for each pull request, the head the approval expects
// and how far its merge has come.
type Merge struct {
V int `json:"v"`
Group string `json:"group"`
Ask string `json:"ask"`
Pulls []MergePull `json:"pulls"`
}
// MergePull is one pull request of a merge: its expected head, its state, the merge commit once merged, the
// forge's words when refused.
type MergePull struct {
Head
State MergeState `json:"state"`
Commit string `json:"commit,omitempty"`
Words string `json:"words,omitempty"`
}
func (m Merge) check() error {
if err := checkName("group", m.Group); err != nil {
return err
}
if m.Ask == "" {
return fmt.Errorf("delivery: a merge names no ask")
}
if len(m.Pulls) == 0 {
return fmt.Errorf("delivery: a merge names no pull request")
}
for _, p := range m.Pulls {
who := fmt.Sprintf("%s#%d", p.Repository, p.Number)
switch p.State {
case MergeMerging:
case MergeMerged:
if p.Commit == "" {
return fmt.Errorf("delivery: the merged pull request %s has no merge commit", who)
}
case MergeRefused:
if p.Words == "" {
return fmt.Errorf("delivery: the refused pull request %s has no words of the forge", who)
}
default:
return fmt.Errorf("delivery: the state %q of %s is not merging, merged or refused", p.State, who)
}
}
return nil
}
// Ask is the value of an asks key: the router's ask whole as the gate composes it, so the teller publishes it
// unchanged, and the group and heads it was composed for.
type Ask struct {
V int `json:"v"`
Group string `json:"group"`
Heads []Head `json:"heads"`
Ask asks.Ask `json:"ask"`
}
func (a Ask) check() error {
key, err := MergeAskKey(a.Group, a.Heads)
if err != nil {
return fmt.Errorf("delivery: an ask needs its group and at least one pull request's head: %w", err)
}
if want := AskID(key); a.Ask.ID != want {
return fmt.Errorf("delivery: the ask's id %q is not %q, the id its group and heads give", a.Ask.ID, want)
}
if len(a.Ask.Options) == 0 {
return fmt.Errorf("delivery: the ask has no option")
}
return nil
}
// Answer is the teller's answer to an ask: the option chosen, the warrant's id and its words.
type Answer struct {
V int `json:"v"`
Option string `json:"option"`
Warrant string `json:"warrant"`
Words string `json:"words"`
}
func (a Answer) check() error {
for _, f := range []struct{ name, value string }{{"option", a.Option}, {"warrant", a.Warrant}, {"words", a.Words}} {
if f.value == "" {
return fmt.Errorf("delivery: an answer has no %s", f.name)
}
}
return nil
}
// Owed is the value of an owed key: what the forge bridge performs. A status needs Context and State, a view and a
// comment need Text, a note needs Text and Ref (the notes ref, mesh-plan).
type Owed struct {
V int `json:"v"`
Act string `json:"act"`
Context string `json:"context,omitempty"`
State string `json:"state,omitempty"`
Text string `json:"text,omitempty"`
Ref string `json:"ref,omitempty"`
}
func (o Owed) check() error {
switch o.Act {
case OwedStatus:
if o.Context == "" {
return fmt.Errorf("delivery: an owed status has no context")
}
for _, s := range statusStates {
if o.State == s {
return nil
}
}
return fmt.Errorf("delivery: the owed status state %q is not %s", o.State, strings.Join(statusStates, ", "))
case OwedView, OwedComment:
if o.Text == "" {
return fmt.Errorf("delivery: an owed %s has no text", o.Act)
}
case OwedNote:
if o.Text == "" {
return fmt.Errorf("delivery: an owed note has no text")
}
if o.Ref == "" {
return fmt.Errorf("delivery: an owed note has no ref")
}
default:
return fmt.Errorf("delivery: the owed act %q is not status, view, note or comment", o.Act)
}
return nil
}
// EncodeMerge is m as JSON at Version; it refuses a merged pull request without its merge commit, a refused one
// without the forge's words and a state outside the list.
func EncodeMerge(m Merge) ([]byte, error) {
if err := m.check(); err != nil {
return nil, err
}
m.V = Version
return json.Marshal(m)
}
// DecodeMerge reads a merge; it refuses an unknown version (ErrUnknownVersion) and what EncodeMerge refuses.
func DecodeMerge(data []byte) (Merge, error) {
var m Merge
if err := decode(data, &m); err != nil {
return Merge{}, err
}
if err := m.check(); err != nil {
return Merge{}, err
}
return m, nil
}
// EncodeAsk is a as JSON at Version; it refuses an ask whose id is not the one its group and heads give, or with
// no option.
func EncodeAsk(a Ask) ([]byte, error) {
if err := a.check(); err != nil {
return nil, err
}
a.V = Version
return json.Marshal(a)
}
// DecodeAsk reads an ask; it refuses an unknown version (ErrUnknownVersion) and what EncodeAsk refuses.
func DecodeAsk(data []byte) (Ask, error) {
var a Ask
if err := decode(data, &a); err != nil {
return Ask{}, err
}
if err := a.check(); err != nil {
return Ask{}, err
}
return a, nil
}
// EncodeAnswer is a as JSON at Version; it refuses an answer missing its option, warrant or words.
func EncodeAnswer(a Answer) ([]byte, error) {
if err := a.check(); err != nil {
return nil, err
}
a.V = Version
return json.Marshal(a)
}
// DecodeAnswer reads an answer; it refuses an unknown version (ErrUnknownVersion) and what EncodeAnswer refuses.
func DecodeAnswer(data []byte) (Answer, error) {
var a Answer
if err := decode(data, &a); err != nil {
return Answer{}, err
}
if err := a.check(); err != nil {
return Answer{}, err
}
return a, nil
}
// EncodeOwed is o as JSON at Version; it refuses an act missing what it needs.
func EncodeOwed(o Owed) ([]byte, error) {
if err := o.check(); err != nil {
return nil, err
}
o.V = Version
return json.Marshal(o)
}
// DecodeOwed reads an owed act; it refuses an unknown version (ErrUnknownVersion) and what EncodeOwed refuses.
func DecodeOwed(data []byte) (Owed, error) {
var o Owed
if err := decode(data, &o); err != nil {
return Owed{}, err
}
if err := o.check(); err != nil {
return Owed{}, err
}
return o, nil
}
// SameMerge says whether two merges are the same value, so a writer writes only when they differ.
func SameMerge(a, b Merge) bool { return reflect.DeepEqual(a, b) }
// SameAsk says whether two asks are the same value.
func SameAsk(a, b Ask) bool { return reflect.DeepEqual(a, b) }
// SameAnswer says whether two answers are the same value.
func SameAnswer(a, b Answer) bool { return a == b }
// SameOwed says whether two owed acts are the same value.
func SameOwed(a, b Owed) bool { return a == b }
+407
View File
@@ -0,0 +1,407 @@
package delivery
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"reflect"
"sort"
"strings"
)
// The modules of the parts release 3 adds (novox/hq ADR 0437): their buckets are their own state, `<module>_<name>`.
const (
GateModule = "mesh-delivery-gate"
IndexerModule = "mesh-delivery-indexer"
ForgeBridgeModule = "mesh-delivery-forge-bridge"
TellerModule = "mesh-delivery-teller"
)
// The gate's source bucket and the indexer's candidates and modules buckets (ADR 0437 §5). The source bucket has
// one key, SourceMainKey: every repository's merged head and the group that moved it, written by the gate alone by
// compare-and-set; its revision is the mesh's source revision. The indexer writes the other two.
const (
SourceBucket = GateModule + "_source"
SourceMainKey = "main"
SourceHistory = 64
CandidatesBucket = IndexerModule + "_candidates"
// CandidatesHistory is the revisions the candidates bucket keeps per key.
CandidatesHistory = 10
ModulesBucket = IndexerModule + "_modules"
// ModulesHistory is the revisions the modules bucket keeps per key.
ModulesHistory = 20
)
// ForgeStream is the forge bridge's stream as its manifest declares it; the broker derives the real names.
const ForgeStream = "forge"
// The kinds of forge message, the last part of a forge event's name.
const (
ForgePull = "pull"
ForgeMerged = "merged"
ForgeRefused = "refused"
)
// The actions of a pull message.
const (
PullOpened = "opened"
PullUpdated = "updated"
PullClosed = "closed"
)
// SourcePrefix opens every source fingerprint.
const SourcePrefix = "src1:"
// CandidatesKey is the key, in CandidatesBucket, of a group: `<group>`, the branch name the group's pull requests
// share (a slash is allowed); a lone pull request is a group of one under its branch name.
func CandidatesKey(group string) (string, error) {
if err := checkName("group", group); err != nil {
return "", err
}
return group, nil
}
// ParseCandidatesKey is the inverse of CandidatesKey.
func ParseCandidatesKey(key string) (group string, err error) {
if group, err = CandidatesKey(key); err != nil {
return "", fmt.Errorf("delivery: %q is not a candidates key: %w", key, err)
}
return group, nil
}
// ModulesKey is the key, in ModulesBucket, of a module: `<module>`.
func ModulesKey(module string) (string, error) {
if err := checkName("module", module); err != nil {
return "", err
}
return module, nil
}
// ParseModulesKey is the inverse of ModulesKey.
func ParseModulesKey(key string) (module string, err error) {
if module, err = ModulesKey(key); err != nil {
return "", fmt.Errorf("delivery: %q is not a modules key: %w", key, err)
}
return module, nil
}
// checkRepository refuses a repository that is not `<owner>/<name>`: exactly one slash, each part a valid name.
func checkRepository(repository string) error {
owner, name, ok := strings.Cut(repository, "/")
if !ok || strings.Contains(name, "/") {
return fmt.Errorf("delivery: the repository %q is not <owner>/<name>, with exactly one slash", repository)
}
if err := checkName("repository owner", owner); err != nil {
return fmt.Errorf("delivery: the repository %q: %w", repository, err)
}
if err := checkName("repository name", name); err != nil {
return fmt.Errorf("delivery: the repository %q: %w", repository, err)
}
if strings.Contains(owner, "/") {
return fmt.Errorf("delivery: the repository %q is not <owner>/<name>, with exactly one slash", repository)
}
return nil
}
func checkForgeKind(kind string) error {
switch kind {
case ForgePull, ForgeMerged, ForgeRefused:
return nil
}
return fmt.Errorf("delivery: the forge kind %q is not pull, merged or refused", kind)
}
// ForgeEvent is the name of a forge message of the given kind: `forge.<owner>.<name>.<kind>`.
func ForgeEvent(repository, kind string) (string, error) {
if err := checkRepository(repository); err != nil {
return "", err
}
if err := checkForgeKind(kind); err != nil {
return "", err
}
owner, name, _ := strings.Cut(repository, "/")
return "forge." + owner + "." + name + "." + kind, nil
}
// ParseForgeEvent is the inverse of ForgeEvent.
func ParseForgeEvent(event string) (repository, kind string, err error) {
parts := strings.Split(event, ".")
if len(parts) != 4 || parts[0] != "forge" {
return "", "", fmt.Errorf("delivery: %q is not a forge event, forge.<owner>.<name>.<kind>", event)
}
repository = parts[1] + "/" + parts[2]
if _, err := ForgeEvent(repository, parts[3]); err != nil {
return "", "", fmt.Errorf("delivery: %q is not a forge event: %w", event, err)
}
return repository, parts[3], nil
}
// Head is one pull request at one commit: the repository as `<owner>/<name>` and the full sha.
type Head struct {
Repository string `json:"repository"`
Number int `json:"number"`
Head string `json:"head"`
}
// HeadsHash is the first 12 hex digits of the sha256 over the lines `<repository>#<number>=<head>`, sorted and
// joined by a newline: the same for every order of heads, different when any head changes. An empty list is refused.
func HeadsHash(heads []Head) (string, error) {
if len(heads) == 0 {
return "", fmt.Errorf("delivery: a heads hash needs at least one pull request")
}
lines := make([]string, len(heads))
for i, h := range heads {
lines[i] = fmt.Sprintf("%s#%d=%s", h.Repository, h.Number, h.Head)
}
sort.Strings(lines)
sum := sha256.Sum256([]byte(strings.Join(lines, "\n")))
return hex.EncodeToString(sum[:])[:12], nil
}
// Source is the value of SourceMainKey: every repository's merged head (repository to full sha, not only the
// group's) and the group that moved it.
type Source struct {
V int `json:"v"`
Group string `json:"group"`
Heads map[string]string `json:"heads"`
}
func (s Source) check() error {
if len(s.Heads) == 0 {
return fmt.Errorf("delivery: a source names no repository")
}
for repository, head := range s.Heads {
if repository == "" {
return fmt.Errorf("delivery: a source names a repository with no name")
}
if head == "" {
return fmt.Errorf("delivery: the source's head of %s is empty", repository)
}
}
return nil
}
// Candidate is the value of a candidates key: the group's pull requests and heads, the heads hash, the base (the
// source revision it was previewed onto) and the modules each pull request touches. It carries no fingerprint
// until release 4 (ADR 0437 §5).
type Candidate struct {
V int `json:"v"`
Group string `json:"group"`
Base uint64 `json:"base"`
HeadsHash string `json:"heads"`
Pulls []CandidatePull `json:"pulls"`
}
// CandidatePull is one pull request of a candidate and the modules it touches.
type CandidatePull struct {
Head
Modules []string `json:"modules"`
}
func (c Candidate) check() error {
if len(c.Pulls) == 0 {
return fmt.Errorf("delivery: a candidate names no pull request")
}
heads := make([]Head, len(c.Pulls))
for i, p := range c.Pulls {
heads[i] = p.Head
}
want, _ := HeadsHash(heads)
if c.HeadsHash != want {
return fmt.Errorf("delivery: the candidate's heads hash %q is not the hash %q of its pull requests' heads", c.HeadsHash, want)
}
return nil
}
// Module is the value of a modules key: the module's source fingerprint, or none and why (a build that resolves
// packages when built gets none, said here), and where it was computed: Repository and Commit are the trunk merge,
// Src the source revision then (0 before the gate writes source.main). Exactly one of FP and None is set.
type Module struct {
V int `json:"v"`
FP string `json:"fp,omitempty"`
None string `json:"none,omitempty"`
Src uint64 `json:"src,omitempty"`
Repository string `json:"repository"`
Commit string `json:"commit"`
}
func (m Module) check() error {
switch {
case m.FP != "" && m.None != "":
return fmt.Errorf("delivery: a module has both a fingerprint and a reason for none")
case m.FP == "" && m.None == "":
return fmt.Errorf("delivery: a module has neither a fingerprint nor a reason for none")
case m.FP != "" && !strings.HasPrefix(m.FP, SourcePrefix):
return fmt.Errorf("delivery: the module's fingerprint %q does not start with %s", m.FP, SourcePrefix)
}
return nil
}
// Forge is a message of the forge bridge's stream, trimmed to repository, branch, pull request and head. A pull
// message says Action; a merged message carries the merge commit, the tree of every directory holding a
// module.json and the repository's root tree; a refused message names the owed act it answers and the forge's own
// words.
type Forge struct {
V int `json:"v"`
Kind string `json:"kind"`
Repository string `json:"repository"`
Branch string `json:"branch,omitempty"`
Number int `json:"number,omitempty"`
Head string `json:"head,omitempty"`
Action string `json:"action,omitempty"`
Merge string `json:"merge,omitempty"`
Trees map[string]string `json:"trees,omitempty"`
Root string `json:"root,omitempty"`
Act string `json:"act,omitempty"`
Words string `json:"words,omitempty"`
}
func (f Forge) check() error {
if err := checkRepository(f.Repository); err != nil {
return err
}
if err := checkForgeKind(f.Kind); err != nil {
return err
}
switch f.Kind {
case ForgePull:
if f.Number <= 0 {
return fmt.Errorf("delivery: a pull message needs its pull request number")
}
if f.Head == "" {
return fmt.Errorf("delivery: a pull message needs its head")
}
switch f.Action {
case PullOpened, PullUpdated, PullClosed:
default:
return fmt.Errorf("delivery: the pull action %q is not opened, updated or closed", f.Action)
}
case ForgeMerged:
if f.Number <= 0 {
return fmt.Errorf("delivery: a merged message needs its pull request number")
}
if f.Head == "" {
return fmt.Errorf("delivery: a merged message needs its head")
}
if f.Merge == "" {
return fmt.Errorf("delivery: a merged message needs its merge commit")
}
if f.Root == "" {
return fmt.Errorf("delivery: a merged message needs the repository's root tree")
}
if len(f.Trees) == 0 {
return fmt.Errorf("delivery: a merged message needs the tree of at least one directory holding a module.json")
}
for dir, tree := range f.Trees {
if tree == "" {
return fmt.Errorf("delivery: a merged message's tree of the directory %q is empty", dir)
}
}
case ForgeRefused:
if f.Act == "" {
return fmt.Errorf("delivery: a refused message needs the act it answers")
}
if f.Words == "" {
return fmt.Errorf("delivery: a refused message needs the forge's words")
}
}
return nil
}
// EncodeSource is s as JSON at Version; it refuses a source with no repository or an empty head.
func EncodeSource(s Source) ([]byte, error) {
if err := s.check(); err != nil {
return nil, err
}
s.V = Version
return json.Marshal(s)
}
// DecodeSource reads a source; it refuses an unknown version (ErrUnknownVersion) and what EncodeSource refuses.
func DecodeSource(data []byte) (Source, error) {
var s Source
if err := decode(data, &s); err != nil {
return Source{}, err
}
if err := s.check(); err != nil {
return Source{}, err
}
return s, nil
}
// EncodeCandidate is c as JSON at Version; it refuses a heads hash that is not the hash of its pull requests' heads.
func EncodeCandidate(c Candidate) ([]byte, error) {
if err := c.check(); err != nil {
return nil, err
}
c.V = Version
return json.Marshal(c)
}
// DecodeCandidate reads a candidate; it refuses an unknown version (ErrUnknownVersion) and what EncodeCandidate refuses.
func DecodeCandidate(data []byte) (Candidate, error) {
var c Candidate
if err := decode(data, &c); err != nil {
return Candidate{}, err
}
if err := c.check(); err != nil {
return Candidate{}, err
}
return c, nil
}
// EncodeModule is m as JSON at Version; it refuses both or neither of a fingerprint and a reason, and a
// fingerprint not of the form src1:.
func EncodeModule(m Module) ([]byte, error) {
if err := m.check(); err != nil {
return nil, err
}
m.V = Version
return json.Marshal(m)
}
// DecodeModule reads a module; it refuses an unknown version (ErrUnknownVersion) and what EncodeModule refuses.
func DecodeModule(data []byte) (Module, error) {
var m Module
if err := decode(data, &m); err != nil {
return Module{}, err
}
if err := m.check(); err != nil {
return Module{}, err
}
return m, nil
}
// EncodeForge is f as JSON at Version; it refuses a message missing what its kind needs.
func EncodeForge(f Forge) ([]byte, error) {
if err := f.check(); err != nil {
return nil, err
}
f.V = Version
return json.Marshal(f)
}
// DecodeForge reads a forge message; it refuses an unknown version (ErrUnknownVersion) and what EncodeForge refuses.
func DecodeForge(data []byte) (Forge, error) {
var f Forge
if err := decode(data, &f); err != nil {
return Forge{}, err
}
if err := f.check(); err != nil {
return Forge{}, err
}
return f, nil
}
// SameSource says whether two sources are the same value, so a writer writes only when they differ.
func SameSource(a, b Source) bool { return reflect.DeepEqual(a, b) }
// SameCandidate says whether two candidates are the same value.
func SameCandidate(a, b Candidate) bool { return reflect.DeepEqual(a, b) }
// SameModule says whether two modules are the same value.
func SameModule(a, b Module) bool { return a == b }
// SameForge says whether two forge messages are the same value.
func SameForge(a, b Forge) bool { return reflect.DeepEqual(a, b) }
+2 -2
View File
@@ -1,5 +1,5 @@
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4
## explicit; go 1.22
# git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019091011-aa68b67389c9
## explicit; go 1.26.0
git.novox.be/novox/mesh-sdk/go/asks
git.novox.be/novox/mesh-sdk/go/delivery
git.novox.be/novox/mesh-sdk/go/seal