issue 1025: the operator's Approve of a reviewed group registers a moved module from its new repository (ADR 0420 §7)
This commit is contained in:
@@ -435,6 +435,20 @@ func mayRegisterFrom(ctx context.Context, inv *inventory.Inventory, module strin
|
||||
module, sourceWords(built.Repository, built.Seat))
|
||||
}
|
||||
}
|
||||
// **Or the move the operator approved** (novox/hq ADR 0420 §7, issue 1025): a known module built from another
|
||||
// repository by a merge, whose pull request mesh-delivery merged on the operator's Approve of that head.
|
||||
// A verb alone, or a merge with no such warrant, is refused as before.
|
||||
movedWhy := ""
|
||||
if why != "" && !terminal && !isNew {
|
||||
accepted, err := approvedMove(ctx, inv, forms, buildID, built)
|
||||
switch {
|
||||
case err == nil:
|
||||
movedWhy, why = accepted, ""
|
||||
case !errors.Is(err, errNotAMerge):
|
||||
fmt.Printf("%s: %s is built from another repository, and no approved move covers this build: %v\n",
|
||||
buildID, module, err)
|
||||
}
|
||||
}
|
||||
if why != "" && !terminal {
|
||||
return refuse(why)
|
||||
}
|
||||
@@ -471,7 +485,8 @@ func mayRegisterFrom(ctx context.Context, inv *inventory.Inventory, module strin
|
||||
}
|
||||
// The same repository by the forge's id, not only its name: one deleted and made again is another.
|
||||
recorded := map[string]int64{}
|
||||
if !isNew {
|
||||
// A moved module is registered from the repository the operator approved, not the one it was registered from.
|
||||
if !isNew && movedWhy == "" {
|
||||
id, err := inv.SourceIdentity(ctx, module)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
@@ -492,6 +507,10 @@ func mayRegisterFrom(ctx context.Context, inv *inventory.Inventory, module strin
|
||||
owner, name, m, facts.ID, m, id))
|
||||
}
|
||||
}
|
||||
if movedWhy != "" {
|
||||
fmt.Printf("%s: %s moved from %s to %s/%s — registered at the operator's Approve: %s\n", buildID, module,
|
||||
sourceWords(was.Repository, was.Seat), owner, name, movedWhy)
|
||||
}
|
||||
return facts.ID, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -14,6 +14,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
@@ -382,3 +384,75 @@ func TestAnUnknownIdentityKeepsTheRecordedOne(t *testing.T) {
|
||||
t.Fatalf("the recorded id became %d", id)
|
||||
}
|
||||
}
|
||||
|
||||
// **The operator's Approve of a reviewed group registers a moved module from its new repository** (novox/hq ADR 0420 §7,
|
||||
// issue 1025). sudo is built from the catalogue; a merge into the repository of node-engine's module, announced by
|
||||
// mesh-delivery with the ask it merged on, brings its build from there. Accepted with the operator's Approve of the
|
||||
// merged head; refused with no kept merge, a warrant for other heads, a Decline, a person who is not the operator, and
|
||||
// through a verb.
|
||||
func TestAModuleMovesOnTheOperatorsApproveOfTheMergeThatMovedIt(t *testing.T) {
|
||||
const commit = "c0ffee0123456789"
|
||||
for _, c := range []struct {
|
||||
what string
|
||||
change func(*anApprovedMerge)
|
||||
unkept bool
|
||||
request string // who asked: "merge", or "build" through a verb
|
||||
moved bool
|
||||
}{
|
||||
{what: "the operator's Approve of the merged head", request: "merge", moved: true},
|
||||
{what: "no merge mesh-delivery announced", unkept: true, request: "merge"},
|
||||
{what: "a warrant for other heads", request: "merge", change: func(a *anApprovedMerge) {
|
||||
a.merge.Heads = map[string]string{"novox/mesh-host@aaaaaaaaaaaa": "e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5e5"}
|
||||
}},
|
||||
{what: "a Decline", request: "merge", change: func(a *anApprovedMerge) {
|
||||
a.warrant.Option, a.warrant.Label, a.merge.Option = "decline", "Decline", "decline"
|
||||
}},
|
||||
{what: "a person who is not the operator", request: "merge", change: func(a *anApprovedMerge) { a.warrant.By.Who = "agent" }},
|
||||
{what: "a build asked through a verb", request: "build"},
|
||||
} {
|
||||
open := theCatalogue(t)
|
||||
ctx := t.Context()
|
||||
inv := open.inventory
|
||||
a := approvedMerge(t)
|
||||
a.merge.Owner, a.merge.Repo, a.merge.Number, a.merge.HeadSHA = "novox", "mesh-host", 7, "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1"
|
||||
if c.change != nil {
|
||||
c.change(&a)
|
||||
}
|
||||
wasKept, wasAsk := keptDeliveryMerge, routerAskOf
|
||||
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, _ *inventory.Inventory, id string) (*asks.Ask, *asks.Warrant, error) {
|
||||
if id != a.merge.AskID {
|
||||
t.Errorf("%s: the router was asked for %q", c.what, id)
|
||||
}
|
||||
return &a.ask, &a.warrant, nil
|
||||
}
|
||||
t.Cleanup(func() { keptDeliveryMerge, routerAskOf = wasKept, wasAsk })
|
||||
event, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-host", "base": "main", "number": 7,
|
||||
"head_sha": a.merge.HeadSHA, "merge_commit_sha": commit})
|
||||
if _, err := inv.AddMerge(ctx, inventory.BatchedMerge{Repository: "novox/mesh-host", Branch: "main", Commit: commit,
|
||||
Merged: time.Now(), Heard: time.Now(), Event: event}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
id := fmt.Sprintf("build-%d", time.Now().Add(time.Hour).UnixNano())
|
||||
if err := inv.RecordBuildRequest(ctx, inventory.BuildRequest{ID: id, Repository: "novox/mesh-host", Seat: "git",
|
||||
Path: "modules/sudo", Ref: "main", Commit: commit, For: c.request}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
moved := onTrunk(id, "novox/mesh-host", "git", "modules/sudo", map[string]any{"module": "sudo", "version": "2"})
|
||||
moved.Commit = commit
|
||||
_, _, err := takeIn(ctx, inv, moved)
|
||||
src, _ := inv.SourceOf(ctx, "sudo")
|
||||
switch {
|
||||
case c.moved && err != nil:
|
||||
t.Errorf("%s: the move was refused: %v", c.what, err)
|
||||
case c.moved && src.Repository != "novox/mesh-host":
|
||||
t.Errorf("%s: sudo is built from %q", c.what, src.Repository)
|
||||
case !c.moved && !errors.Is(err, errNotItsSource):
|
||||
t.Errorf("%s: the move was not refused as ADR 0266 has it: %v", c.what, err)
|
||||
case !c.moved && src.Repository != "novox/mesh-catalog":
|
||||
t.Errorf("%s: a refused move changed sudo's source to %q", c.what, src.Repository)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,259 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// The operator's Approve of a delivery group registers a moved module from its new repository (novox/hq ADR 0420
|
||||
// §7, issue 1025). ADR 0266 §44 lets a build's outcome register a known module only from the repository the
|
||||
// catalogue already builds it from. The one exception was a build asked at the controller's terminal. This is a
|
||||
// second, for a move: mesh-delivery announces each merge it makes with the ask whose Approve it merged on; the
|
||||
// controller keeps the announcement, and when the build of a merge of a known module comes from another repository
|
||||
// it reads that ask's warrant from the router and accepts only the operator's Approve of the merged head. A verb
|
||||
// alone, or a merge with no such warrant, never moves a module.
|
||||
|
||||
// theDelivery is the module that merges, and whose asks these are.
|
||||
const theDelivery = "mesh-delivery"
|
||||
|
||||
// optionApprove is the option of mesh-delivery's ask that merges (its OptionApprove).
|
||||
const optionApprove = "approve"
|
||||
|
||||
// deliveryMerge is mesh-delivery's announcement of a merge it made: the pull request and the head it merged, the
|
||||
// group it belongs to, and the ask whose answer it merged on with the option chosen and the heads that answer was
|
||||
// bound to (delivery id → commit).
|
||||
type deliveryMerge struct {
|
||||
Owner string `json:"owner"`
|
||||
Repo string `json:"repo"`
|
||||
Number int `json:"number"`
|
||||
HeadSHA string `json:"head_sha"`
|
||||
Group string `json:"group,omitempty"`
|
||||
AskID string `json:"ask_id"`
|
||||
Option string `json:"option"`
|
||||
Heads map[string]string `json:"heads"`
|
||||
}
|
||||
|
||||
// parseDeliveryMerge reads an announcement, refusing one that names no pull request, head or ask.
|
||||
func parseDeliveryMerge(body []byte) (deliveryMerge, error) {
|
||||
var d deliveryMerge
|
||||
if err := json.Unmarshal(body, &d); err != nil {
|
||||
return d, fmt.Errorf("mesh-delivery's word on a merge cannot be read: %w", err)
|
||||
}
|
||||
if d.Owner == "" || d.Repo == "" || d.Number <= 0 || d.HeadSHA == "" || d.AskID == "" || d.Option == "" {
|
||||
return d, errors.New("mesh-delivery's word on a merge names no pull request, head, ask or option")
|
||||
}
|
||||
return d, nil
|
||||
}
|
||||
|
||||
// deliveryMergeKey is where a merge is kept: bus keys allow no `#` or `@`, so the pull request and its head are
|
||||
// path-like tokens, as the forge spells the repository.
|
||||
func deliveryMergeKey(owner, repo string, number int, head string) string {
|
||||
return fmt.Sprintf("%s/%s/%d/%s", owner, repo, number, head)
|
||||
}
|
||||
|
||||
// mergeAct is what mesh-delivery's answer binds, spelled exactly as mesh-delivery spells it (its approval.go:
|
||||
// mergeAct and headsString): this ask, the option and the heads it covers.
|
||||
func mergeAct(ask, option string, heads map[string]string) asks.Act {
|
||||
var out []string
|
||||
for id, sha := range heads {
|
||||
out = append(out, id+"="+sha)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return asks.Act{"verb": theDelivery + ".merge", "ask": ask, "option": option, "heads": strings.Join(out, ",")}
|
||||
}
|
||||
|
||||
// movedOnApprove judges the warrant of the ask a merge was made on: it holds only if it is the operator's Approve,
|
||||
// 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) {
|
||||
switch {
|
||||
case ask == nil || w == nil:
|
||||
return "", errors.New("the router holds no closed record of the ask the merge was made on")
|
||||
case ask.Who != asks.Operator:
|
||||
return "", fmt.Errorf("the ask %s is for %q to answer, not the operator", ask.ID, ask.Who)
|
||||
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)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if option.ID != optionApprove || w.Option != optionApprove || d.Option != option.ID {
|
||||
return "", fmt.Errorf("the operator chose %q on the ask %s, not an Approve", w.Option, w.Ask)
|
||||
}
|
||||
if w.By == nil || w.By.Who != asks.Operator || w.By.Verified == "" {
|
||||
return "", fmt.Errorf("the ask %s was not answered by the operator on a channel that proves who answered", w.Ask)
|
||||
}
|
||||
if err := option.Performs(mergeAct(d.AskID, option.ID, d.Heads)); err != nil {
|
||||
return "", err
|
||||
}
|
||||
named := false
|
||||
for _, sha := range d.Heads {
|
||||
named = named || sha == d.HeadSHA
|
||||
}
|
||||
if !named {
|
||||
return "", fmt.Errorf("the Approve covers %v, and the head merged, %.8s, is not among them", d.Heads, d.HeadSHA)
|
||||
}
|
||||
return w.Says() + " (ask " + w.Ask + ")", nil
|
||||
}
|
||||
|
||||
// ---- keeping what mesh-delivery announced ----------------------------------------------------------------------
|
||||
|
||||
// keepingDeliveryMerges keeps each merge mesh-delivery announces in the controller's bucket for them.
|
||||
type keepingDeliveryMerges struct {
|
||||
js func() *broker.JetStream
|
||||
}
|
||||
|
||||
func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, body []byte) error {
|
||||
d, err := parseDeliveryMerge(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
js := k.js()
|
||||
if js == nil {
|
||||
return fmt.Errorf("%w: the bus is not held", link.ErrTryAgain)
|
||||
}
|
||||
api, err := jetstream.New(js.Conn())
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
|
||||
}
|
||||
kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket)
|
||||
if err != nil {
|
||||
return fmt.Errorf("%w: the bucket of mesh-delivery's merges: %w", link.ErrTryAgain, err)
|
||||
}
|
||||
if _, err := kv.Put(ctx, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil {
|
||||
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// keptDeliveryMerge is the merge mesh-delivery announced for a pull request's head; false when none was kept.
|
||||
// A variable so a test can hand it one.
|
||||
var keptDeliveryMerge = func(ctx context.Context, owner, repo string, number int, head string) (deliveryMerge, bool, error) {
|
||||
serving := servingBus.Load()
|
||||
if serving == nil {
|
||||
return deliveryMerge{}, false, errors.New("only the serving controller keeps what mesh-delivery announced")
|
||||
}
|
||||
api, err := jetstream.New(serving.Conn())
|
||||
if err != nil {
|
||||
return deliveryMerge{}, false, err
|
||||
}
|
||||
kv, err := api.KeyValue(ctx, broker.DeliveryMergesBucket)
|
||||
if err != nil {
|
||||
return deliveryMerge{}, false, err
|
||||
}
|
||||
e, err := kv.Get(ctx, deliveryMergeKey(owner, repo, number, head))
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return deliveryMerge{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return deliveryMerge{}, false, err
|
||||
}
|
||||
d, err := parseDeliveryMerge(e.Value())
|
||||
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
|
||||
// variable so a test can hand it one.
|
||||
var routerAskOf = func(ctx context.Context, inv *inventory.Inventory, 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")
|
||||
}
|
||||
bucket, err := asksRecords(ctx, inv)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
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)
|
||||
return ask, w, err
|
||||
}
|
||||
|
||||
// readRouterEntry reads the router's record of one asker's ask: its state, the ask as published, and its warrant;
|
||||
// "" when there is none. The controller's grant reaches the JetStream API whole (`$JS.API.>`), so it reads any
|
||||
// asker's record.
|
||||
func readRouterEntry(ctx context.Context, conn *nats.Conn, bucket, asker, ask string) (string, *asks.Ask, *asks.Warrant, error) {
|
||||
reply, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+asker+"."+ask, nil)
|
||||
if err != nil {
|
||||
return "", nil, nil, err
|
||||
}
|
||||
if status := reply.Header.Get("Status"); status != "" {
|
||||
if status == "404" {
|
||||
return "", nil, nil, nil
|
||||
}
|
||||
return "", nil, nil, fmt.Errorf("the router's record could not be read: %s %s", status, reply.Header.Get("Description"))
|
||||
}
|
||||
var rec struct {
|
||||
State string `json:"state"`
|
||||
Ask *asks.Ask `json:"ask"`
|
||||
Warrant *asks.Warrant `json:"warrant"`
|
||||
}
|
||||
if err := json.Unmarshal(reply.Data, &rec); err != nil {
|
||||
return "", nil, nil, fmt.Errorf("the router's record of %s's ask %s cannot be read: %w", asker, ask, err)
|
||||
}
|
||||
return rec.State, rec.Ask, rec.Warrant, nil
|
||||
}
|
||||
|
||||
// errNotAMerge says a build was not asked for a merge, so there is no move to look for.
|
||||
var errNotAMerge = errors.New("the build is not of a merge")
|
||||
|
||||
// approvedMove says whether a build of a known module from another repository is the move the operator approved
|
||||
// (ADR 0420 §7): the build is of a merge, that merge is a pull request mesh-delivery announced it made, and the
|
||||
// ask it made it on holds an Approve by the operator of that head. Answers why it was accepted; an error says why
|
||||
// it was not, and the refusal of ADR 0266 stands.
|
||||
func approvedMove(ctx context.Context, inv *inventory.Inventory, forms *sourceForms, buildID string,
|
||||
built inventory.Source) (string, error) {
|
||||
r, found, err := inv.BuildRequestByID(ctx, buildID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if !found || r.For != "merge" || r.Commit == "" {
|
||||
return "", errNotAMerge
|
||||
}
|
||||
owner, name, onForge := forms.onTheForge(built.Repository, built.Seat)
|
||||
if !onForge {
|
||||
return "", errors.New("the repository is not on the mesh's forge")
|
||||
}
|
||||
merged, known, err := inv.MergeOf(ctx, owner+"/"+name, r.Commit)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if !known {
|
||||
return "", fmt.Errorf("the controller kept no merge of %s/%s at %.8s", owner, name, r.Commit)
|
||||
}
|
||||
k := announcedOf(merged)
|
||||
if k.Number <= 0 || k.HeadSHA == "" {
|
||||
return "", errors.New("the forge's announcement of the merge names no pull request or head")
|
||||
}
|
||||
d, kept, err := keptDeliveryMerge(ctx, owner, name, k.Number, k.HeadSHA)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if !kept {
|
||||
return "", fmt.Errorf("mesh-delivery announced no merge of %s/%s#%d at %.8s", owner, name, k.Number, k.HeadSHA)
|
||||
}
|
||||
ask, w, err := routerAskOf(ctx, inv, d.AskID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
why, err := movedOnApprove(d, ask, w)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return fmt.Sprintf("%s/%s#%d merged at %.8s on %s", owner, name, k.Number, k.HeadSHA, why), nil
|
||||
}
|
||||
@@ -0,0 +1,120 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"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"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// The move whole, against a real bus and the store (novox/hq ADR 0420 §7, issue 1025): mesh-delivery's word that it
|
||||
// merged a pull request is published on the subject the controller consumes and kept in its bucket; the router's
|
||||
// record of the ask is read back from the router's bucket by a direct get; and the build of a module from the
|
||||
// repository that merge was made in is registered from there only when that record holds the operator's Approve of
|
||||
// the merged head. Store-backed: skips without one.
|
||||
func TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheBus(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
what string
|
||||
answer bool // the router holds the operator's Approve of the merged heads
|
||||
moved bool
|
||||
}{
|
||||
{what: "the router holds the operator's Approve", answer: true, moved: true},
|
||||
{what: "the router holds no record of the ask"},
|
||||
} {
|
||||
t.Run(c.what, func(t *testing.T) {
|
||||
open := theCatalogue(t)
|
||||
ctx := t.Context()
|
||||
inv := open.inventory
|
||||
// The module declaring the router's records, as the controller finds them.
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "operator-channel", Version: "1",
|
||||
DefinesSeats: []catalogue.SeatDeclaration{{Name: broker.AsksSeat, Records: []string{"asks"}}}},
|
||||
inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/operator-channel", BuiltFrom: "c0ffee"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
js, err := broker.Dial(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(js.Close)
|
||||
if err := broker.AssertMeshStreams(js); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := js.EnsureControllerBuckets(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
api, err := jetstream.New(js.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
records, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.BucketName("operator-channel", "asks")})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
a := approvedMerge(t)
|
||||
a.merge.Owner, a.merge.Repo, a.merge.Number = "novox", "mesh-host", 7
|
||||
if c.answer {
|
||||
record, _ := json.Marshal(map[string]any{"state": "closed", "ask": a.ask, "warrant": a.warrant})
|
||||
if _, err := records.Put(ctx, theDelivery+"."+a.merge.AskID, record); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
server := link.ConnectNats(js, nil, noListener{})
|
||||
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
serving, stop := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() { defer close(done); _ = server.Serve(serving) }()
|
||||
t.Cleanup(func() { stop(); <-done })
|
||||
servingBus.Store(js)
|
||||
t.Cleanup(func() { servingBus.Store(nil) })
|
||||
|
||||
// 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 {
|
||||
t.Fatal(err)
|
||||
}
|
||||
becomes(t, 15*time.Second, "the merge being kept", func() bool {
|
||||
_, kept, err := keptDeliveryMerge(ctx, "novox", "mesh-host", 7, a.merge.HeadSHA)
|
||||
return err == nil && kept
|
||||
})
|
||||
|
||||
// The forge's merge of that pull request, as the controller heard it, and the build it asked for.
|
||||
const commit = "c0ffee0123456789"
|
||||
event, _ := json.Marshal(map[string]any{"owner": "novox", "repo": "mesh-host", "base": "main", "number": 7,
|
||||
"head_sha": a.merge.HeadSHA, "merge_commit_sha": commit})
|
||||
if _, err := inv.AddMerge(ctx, inventory.BatchedMerge{Repository: "novox/mesh-host", Branch: "main", Commit: commit,
|
||||
Merged: time.Now(), Heard: time.Now(), Event: event}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
id := fmt.Sprintf("build-%d", time.Now().Add(time.Hour).UnixNano())
|
||||
if err := inv.RecordBuildRequest(ctx, inventory.BuildRequest{ID: id, Repository: "novox/mesh-host", Seat: "git",
|
||||
Path: "modules/sudo", Ref: "main", Commit: commit, For: "merge"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
moved := onTrunk(id, "novox/mesh-host", "git", "modules/sudo", map[string]any{"module": "sudo", "version": "2"})
|
||||
moved.Commit = commit
|
||||
_, _, err = takeIn(ctx, inv, moved)
|
||||
src, _ := inv.SourceOf(ctx, "sudo")
|
||||
if c.moved && (err != nil || src.Repository != "novox/mesh-host") {
|
||||
t.Fatalf("the Approved move was not registered from its new repository: %v (built from %q)", err, src.Repository)
|
||||
}
|
||||
if !c.moved && (!errors.Is(err, errNotItsSource) || src.Repository != "novox/mesh-catalog") {
|
||||
t.Fatalf("a move with no Approve was not refused as it is: %v (built from %q)", err, src.Repository)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
)
|
||||
|
||||
// A merge mesh-delivery made, with the ask whose answer it merged on: the heads that answer was bound to, the
|
||||
// ask as the router published it, and the warrant the operator's Approve left (novox/hq ADR 0420 §7, issue 1025).
|
||||
type anApprovedMerge struct {
|
||||
merge deliveryMerge
|
||||
ask asks.Ask
|
||||
warrant asks.Warrant
|
||||
}
|
||||
|
||||
func approvedMerge(t *testing.T) anApprovedMerge {
|
||||
t.Helper()
|
||||
heads := map[string]string{"novox/mesh-controller@aaaaaaaaaaaa": "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1",
|
||||
"novox/mesh-catalog@bbbbbbbbbbbb": "b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2"}
|
||||
m := deliveryMerge{Owner: "novox", Repo: "mesh-controller", Number: 7, HeadSHA: "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1",
|
||||
Group: "g1", AskID: "merge-g1", Option: optionApprove, Heads: heads}
|
||||
bind := func(option string, h map[string]string) string {
|
||||
d, err := asks.ActDigest(mergeAct(m.AskID, option, h))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
at := time.Date(2026, 10, 17, 12, 0, 0, 0, time.UTC)
|
||||
ask := asks.Ask{ID: m.AskID, Headline: "Merge the group", Explanation: "Two pull requests move a module.", Who: asks.Operator,
|
||||
Expires: at.Add(time.Hour), OnExpiry: "asked again",
|
||||
Options: []asks.Option{
|
||||
{ID: optionApprove, Label: "Approve", Does: "merges the group", Level: asks.Approve, Binds: bind(optionApprove, heads)},
|
||||
{ID: "decline", Label: "Decline", Does: "leaves it", Level: asks.Approve, Binds: bind("decline", heads)}}}
|
||||
w := asks.Warrant{Ask: m.AskID, Asker: theDelivery, Outcome: asks.OutcomeChosen, Option: optionApprove, Label: "Approve",
|
||||
Level: asks.Approve, By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "1", Verified: "user id"},
|
||||
At: at, AskDigest: ask.Digest()}
|
||||
return anApprovedMerge{merge: m, ask: ask, warrant: w}
|
||||
}
|
||||
|
||||
// The operator's Approve of the merged head holds, and every other answer does not: a Decline, a person who is not the
|
||||
// operator or not proved by their channel, a warrant for other heads or another ask than the router published, the
|
||||
// merged head outside the heads the Approve covered, and a router that holds nothing.
|
||||
func TestAMovesWarrantHoldsOnlyForTheOperatorsApproveOfTheMergedHead(t *testing.T) {
|
||||
ok := approvedMerge(t)
|
||||
if _, err := movedOnApprove(ok.merge, &ok.ask, &ok.warrant); err != nil {
|
||||
t.Fatalf("the operator's Approve of the merged heads was refused: %v", err)
|
||||
}
|
||||
for _, c := range []struct {
|
||||
what string
|
||||
change func(*anApprovedMerge)
|
||||
}{
|
||||
{"a Decline", func(a *anApprovedMerge) {
|
||||
a.warrant.Option, a.warrant.Label = "decline", "Decline"
|
||||
a.merge.Option = "decline"
|
||||
}},
|
||||
{"no person chose", func(a *anApprovedMerge) { a.warrant.Outcome = asks.OutcomeExpired }},
|
||||
{"a person who is not the operator", func(a *anApprovedMerge) { a.warrant.By.Who = "agent" }},
|
||||
{"a channel that proves nobody", func(a *anApprovedMerge) { a.warrant.By.Verified = "" }},
|
||||
{"other heads than the ask was bound to", func(a *anApprovedMerge) {
|
||||
a.merge.Heads = map[string]string{"novox/mesh-controller@aaaaaaaaaaaa": "c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3c3"}
|
||||
}},
|
||||
{"a head merged that the Approve did not cover", func(a *anApprovedMerge) { a.merge.HeadSHA = "d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4d4" }},
|
||||
{"another ask than the one the warrant answers", func(a *anApprovedMerge) { a.ask.Headline = "Something else" }},
|
||||
{"a warrant of another asker", func(a *anApprovedMerge) { a.warrant.Asker = "claude-code" }},
|
||||
{"an ask for no operator to answer", func(a *anApprovedMerge) { a.ask.Who = "agent" }},
|
||||
{"an Approve given after the ask expired", func(a *anApprovedMerge) { a.warrant.At = a.ask.Expires.Add(time.Minute) }},
|
||||
} {
|
||||
a := approvedMerge(t)
|
||||
c.change(&a)
|
||||
if why, err := movedOnApprove(a.merge, &a.ask, &a.warrant); err == nil {
|
||||
t.Errorf("%s was accepted: %s", c.what, why)
|
||||
}
|
||||
}
|
||||
if _, err := movedOnApprove(ok.merge, nil, &ok.warrant); err == nil {
|
||||
t.Error("a warrant with no ask to check it against was accepted")
|
||||
}
|
||||
if _, err := movedOnApprove(ok.merge, &ok.ask, nil); err == nil {
|
||||
t.Error("an ask with no warrant was accepted")
|
||||
}
|
||||
}
|
||||
|
||||
// mesh-delivery's word is read whole or refused, and kept under a key the bus allows.
|
||||
func TestADeliveryMergesWordIsReadWholeAndKeptUnderABusKey(t *testing.T) {
|
||||
body := `{"owner":"novox","repo":"mesh-controller","number":7,"head_sha":"abc","group":"g","ask_id":"merge-g",` +
|
||||
`"option":"approve","heads":{"novox/mesh-controller@abc":"abc"}}`
|
||||
d, err := parseDeliveryMerge([]byte(body))
|
||||
if err != nil || d.Number != 7 || d.Heads["novox/mesh-controller@abc"] != "abc" {
|
||||
t.Fatalf("read %+v, %v", d, err)
|
||||
}
|
||||
if key := deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA); key != "novox/mesh-controller/7/abc" ||
|
||||
strings.ContainsAny(key, "#@") {
|
||||
t.Errorf("key %q", key)
|
||||
}
|
||||
for _, bad := range []string{`{`, `{"owner":"novox","repo":"r","number":0,"head_sha":"a","ask_id":"x","option":"approve"}`,
|
||||
`{"owner":"novox","repo":"r","number":1,"head_sha":"a","option":"approve"}`} {
|
||||
if _, err := parseDeliveryMerge([]byte(bad)); err == nil {
|
||||
t.Errorf("%s was read", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -202,6 +202,11 @@ func serve(ctx context.Context) (err error) {
|
||||
if err := server.Follows(following{open: open}); err != nil {
|
||||
return err
|
||||
}
|
||||
// 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}); err != nil {
|
||||
return err
|
||||
}
|
||||
// No timer reads the merges back (novox/hq ADR 0407 decision 3, issue 973): the forge's event and the durable
|
||||
// consumer on the bus keep every merge, and one the controller cannot record yet is held and handed over again.
|
||||
// And a catalogue that has just started, asking for what it missed. The same type answers
|
||||
|
||||
@@ -15,7 +15,6 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
@@ -64,24 +63,8 @@ func warrantedAct(asker, ask, caller, state string, w *asks.Warrant) (link.HandA
|
||||
// readRouterRecord reads the router's record of one asker's ask: its state and warrant, or "" when there is none.
|
||||
// The controller's grant reaches the JetStream API whole (`$JS.API.>`), so it reads any asker's record.
|
||||
func readRouterRecord(ctx context.Context, conn *nats.Conn, bucket, asker, ask string) (string, *asks.Warrant, error) {
|
||||
reply, err := conn.RequestWithContext(ctx, "$JS.API.DIRECT.GET.KV_"+bucket+".$KV."+bucket+"."+asker+"."+ask, nil)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
if status := reply.Header.Get("Status"); status != "" {
|
||||
if status == "404" {
|
||||
return "", nil, nil
|
||||
}
|
||||
return "", nil, fmt.Errorf("the router's record could not be read: %s %s", status, reply.Header.Get("Description"))
|
||||
}
|
||||
var rec struct {
|
||||
State string `json:"state"`
|
||||
Warrant *asks.Warrant `json:"warrant"`
|
||||
}
|
||||
if err := json.Unmarshal(reply.Data, &rec); err != nil {
|
||||
return "", nil, fmt.Errorf("the router's record of %s's ask %s cannot be read: %w", asker, ask, err)
|
||||
}
|
||||
return rec.State, rec.Warrant, nil
|
||||
state, _, w, err := readRouterEntry(ctx, conn, bucket, asker, ask)
|
||||
return state, w, err
|
||||
}
|
||||
|
||||
func handActWarrantCommand(ctx context.Context, args []string) error {
|
||||
|
||||
Reference in New Issue
Block a user