diff --git a/cmd/mesh-controller/build_source.go b/cmd/mesh-controller/build_source.go index 11a4ef4a..a7565d8c 100644 --- a/cmd/mesh-controller/build_source.go +++ b/cmd/mesh-controller/build_source.go @@ -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 } diff --git a/cmd/mesh-controller/build_source_test.go b/cmd/mesh-controller/build_source_test.go index 146ffcfe..03ce2bd0 100644 --- a/cmd/mesh-controller/build_source_test.go +++ b/cmd/mesh-controller/build_source_test.go @@ -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) + } + } +} diff --git a/cmd/mesh-controller/moved.go b/cmd/mesh-controller/moved.go new file mode 100644 index 00000000..7c111ce0 --- /dev/null +++ b/cmd/mesh-controller/moved.go @@ -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 +} diff --git a/cmd/mesh-controller/moved_bus_test.go b/cmd/mesh-controller/moved_bus_test.go new file mode 100644 index 00000000..ebf93d7e --- /dev/null +++ b/cmd/mesh-controller/moved_bus_test.go @@ -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) + } + }) + } +} diff --git a/cmd/mesh-controller/moved_test.go b/cmd/mesh-controller/moved_test.go new file mode 100644 index 00000000..1a487280 --- /dev/null +++ b/cmd/mesh-controller/moved_test.go @@ -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) + } + } +} diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 66099d0a..3008cc42 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -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 diff --git a/cmd/mesh-controller/warranted.go b/cmd/mesh-controller/warranted.go index 9f6f918e..17e6e102 100644 --- a/cmd/mesh-controller/warranted.go +++ b/cmd/mesh-controller/warranted.go @@ -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 { diff --git a/internal/broker/controller_buckets.go b/internal/broker/controller_buckets.go index 8aa61828..088f1edf 100644 --- a/internal/broker/controller_buckets.go +++ b/internal/broker/controller_buckets.go @@ -56,8 +56,17 @@ var ( // waiting request for a head and trunk commit a key names drops it. **Written by the build seat's holders, several by // design** — each the key of the check it runs. CheckRunningBucket = BucketName(ControllerSeat, "checks-running") + // DeliveryMergesBucket holds the merges mesh-delivery announced it made, each with the ask whose Approve it merged + // on (novox/hq ADR 0420 §7, issue 1025): per merge a key `///` holding the + // announcement, kept DeliveryMergesKeptFor. What lets the build of a moved module register it from its new + // repository: the controller reads the ask's warrant from the router and accepts only the operator's Approve of + // those heads. Written by the controller alone, from the announcement, when it holds the lease. + DeliveryMergesBucket = BucketName(ControllerSeat, "delivery-merges") ) +// DeliveryMergesKeptFor is how long a merge mesh-delivery announced is kept: a week, as the events are. +const DeliveryMergesKeptFor = 7 * 24 * time.Hour + // The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of // record, one key per record, written from the store by the lease holder only when the record changed, with // one event on the controller's seat per write. Read by whatever names them in `reads` as @@ -141,13 +150,15 @@ func IsControllerBucket(bucket string) bool { return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket || bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket || bucket == AssignmentsBucket || bucket == DeclaredConfigBucket || bucket == PredefinedBucket || - bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket + bucket == MergeWindowBucket || bucket == RolloutSilentBucket || bucket == CheckRunningBucket || + bucket == DeliveryMergesBucket } // ControllerBuckets are the controller's own buckets, in the order they are asserted. func ControllerBuckets() []string { return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket, - AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket} + AssignmentsBucket, DeclaredConfigBucket, PredefinedBucket, MergeWindowBucket, RolloutSilentBucket, CheckRunningBucket, + DeliveryMergesBucket} } // ControllerBucketsAsserter is what raising the controller's buckets needs of a connection. @@ -290,6 +301,20 @@ func (j *JetStream) EnsureControllerBuckets() error { }); err != nil { return fmt.Errorf("asserting bucket %s: %w", CheckRunningBucket, err) } + // **The merges mesh-delivery announced: keys that age** (ADR 0420 §7, issue 1025). One message a key; a week. + if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{ + Bucket: DeliveryMergesBucket, + Description: "the merges mesh-delivery announced it made, each with the ask whose Approve it merged on (novox/hq " + + "ADR 0420): `///` holds the announcement, kept a week; what lets the build " + + "of a moved module register it from its new repository; written by the controller alone", + History: 1, + TTL: DeliveryMergesKeptFor, + MaxValueSize: 16 << 10, + MaxBytes: 16 << 20, + Storage: jetstream.FileStorage, + }); err != nil { + return fmt.Errorf("asserting bucket %s: %w", DeliveryMergesBucket, err) + } // **The record mirrors: no age** (design 54 §5). A key is removed when its record is, by a purge whose // marker lives a day; a key that expired would be a record the bus forgot while the store still holds it. for _, m := range []struct{ bucket, why string }{ diff --git a/internal/broker/nats.go b/internal/broker/nats.go index b73d975e..a1af12fe 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -397,6 +397,11 @@ var DegradableGrants = []DegradableGrant{ Reason: "refused: the controller says a tier sent to a machine only for outsiders, and a send that could not " + "be said is printed and goes on (ADR 0406 §5, issue 974)", }, + { + Subject: "$KV." + DeliveryMergesBucket + ".>", + Reason: "refused: the controller then keeps no merge mesh-delivery announced, so the build of a moved module is " + + "refused as it is without the operator's Approve — failing closed (ADR 0420 §7, issue 1025)", + }, { Subject: "$KV." + RolloutSilentBucket + ".>", Reason: "refused: the controller looks at the plans when a wait for a machine's report is up, by a timer of its own " + diff --git a/internal/broker/streams.go b/internal/broker/streams.go index f8377211..8be30cbb 100644 --- a/internal/broker/streams.go +++ b/internal/broker/streams.go @@ -242,7 +242,7 @@ func eventsTheControllerFollows() []string { out = append(out, subject) } } - return out + return append(out, ControllerHears...) } // DeliverSubjectFor is where a push consumer's messages land. @@ -428,6 +428,19 @@ var ControllerFollows = []string{ DecidedSubject, } +// ControllerHears are events the controller's events consumer takes beside the ones it follows, and that need no +// subscribe grant of their own. **A push consumer delivers onto its deliver subject** (`_DELIVER.controller.>`, already +// granted), so the subject a consumer filters on is never one the controller subscribes: listing this one in +// ControllerFollows would add a subscribe grant the installer's carried user list, which lives in mesh-host, would +// have to carry too (TestTheInstallersFirstUserListIsWhatTheControllerWouldCompose), for nothing the consumer uses. +// +// A merge mesh-delivery made, announced (novox/hq ADR 0420 §7, issue 1025): with the ask whose Approve it merged on and +// 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") + // 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" diff --git a/internal/broker/testdata/composed.conf b/internal/broker/testdata/composed.conf index 32fd9212..d9e24d27 100644 --- a/internal/broker/testdata/composed.conf +++ b/internal/broker/testdata/composed.conf @@ -24,7 +24,7 @@ accounts { jetstream: enabled users = [ { user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: { - publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } + publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller-merges.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_checks-running.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_delivery-merges.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$KV.mesh-controller_merge-window.>", "$KV.mesh-controller_predefined.>", "$KV.mesh-controller_rollout-silent.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.predefined-changed", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.event.sent", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.check.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*", "mesh.seat.operator-channel.tool.tell"] } subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] } allow_responses: { max: 1, ttl: "1m" } } } diff --git a/internal/broker/writers.go b/internal/broker/writers.go index 8dc3c4b0..bf7a5960 100644 --- a/internal/broker/writers.go +++ b/internal/broker/writers.go @@ -130,8 +130,11 @@ var WritersTable = []WriterRow{ // written by the lease holder. {State: "plans and their tiers", Writer: "controller (lease holder), compare-and-set on the plan's revision", KeptIn: "the controller's store; a batch's merge window in key-value " + MergeWindowBucket + - ", a rollout's wait for a machine's report in key-value " + RolloutSilentBucket, Others: "read through plans", - Subjects: append(kvOf(MergeWindowBucket), kvOf(RolloutSilentBucket)...), Writes: isController}, + ", a rollout's wait for a machine's report in key-value " + RolloutSilentBucket + + ", and the merges mesh-delivery announced, for a moved module's registration (ADR 0420 §7), in key-value " + + DeliveryMergesBucket, Others: "read through plans", + Subjects: append(append(kvOf(MergeWindowBucket), kvOf(RolloutSilentBucket)...), kvOf(DeliveryMergesBucket)...), + Writes: isController}, {State: "conditions", Writer: "controller", KeptIn: "key-value " + ConditionsBucket + " (and its history, " + ConditionHistoryBucket + "); how often each was observed, when last and its newest evidence in the " + "controller's store, so a look that changes nothing writes no bucket", Others: "raise or clear only through observations the controller reads", diff --git a/internal/link/contracts.go b/internal/link/contracts.go index 381fdf4f..d5b7aff8 100644 --- a/internal/link/contracts.go +++ b/internal/link/contracts.go @@ -51,6 +51,8 @@ var Contracts = map[string]Contract{ "at the router, and the controller acts on it once, recording that it did under the ask's id — so a word " + "heard again, or late, does nothing more (novox/hq ADR 0259)", Tests: []string{"TestAWarrantIsActedOnOnce"}}, + KindDeliveryMerged: {Unordered: "mesh-delivery's word that it merged a pull request, by the pull request and its head: " + + "kept under that key once, so a word heard again or late rewrites the same value (novox/hq ADR 0420 §7)"}, KindCatchUp: {Unordered: "a catalogue asking what it missed: answered from the record, whenever asked"}, KindProvisioner: {Unordered: "a provider's newest word about a consumer, said again every fifteen minutes " + "while it holds (ADR 0224): the condition keeps the last observed, and S8 says when the words stop. " + diff --git a/internal/link/contracts_test.go b/internal/link/contracts_test.go index 20a5a410..4fde2f1c 100644 --- a/internal/link/contracts_test.go +++ b/internal/link/contracts_test.go @@ -18,6 +18,7 @@ func TestEveryConsumedKindHasAContract(t *testing.T) { subjects := []string{EnrolSubject, BuiltSubject, ReportSubject("anchor"), AliveSubject("anchor"), ToolsAliveSubject("anchor"), HealthSubject("anchor"), "mesh.mod.postgres.event.provisioner.failing"} subjects = append(subjects, broker.ControllerFollows...) + subjects = append(subjects, broker.ControllerHears...) kinds := map[string]bool{} for _, s := range subjects { kind, known := kindOfSubject(s) diff --git a/internal/link/receive.go b/internal/link/receive.go index 0173777a..1585dc20 100644 --- a/internal/link/receive.go +++ b/internal/link/receive.go @@ -47,6 +47,9 @@ const ( // KindDecided is the router's warrant for an ask the controller made, or that ask's end without one // (novox/hq ADR 0259): said to the controller alone, under its own name. KindDecided = "decided" + // KindDeliveryMerged is mesh-delivery saying it merged a pull request, with the ask whose Approve it merged on + // (novox/hq ADR 0420 §7, issue 1025). + KindDeliveryMerged = "delivery-merged" ) // Control is one thing a node or a module said, as the controller must act on it. diff --git a/internal/link/receive_nats.go b/internal/link/receive_nats.go index e6431be6..26c32238 100644 --- a/internal/link/receive_nats.go +++ b/internal/link/receive_nats.go @@ -62,7 +62,8 @@ func Nats(js *broker.JetStream) Inbound { // whatever was asked for — and not at all when nothing was. func (n *natsInbound) Also(kind string) error { switch kind { - case KindModuleMoved, KindCatchUp, KindSourceMoved, KindProvisioner, KindPullUpdated, KindDecided: + case KindModuleMoved, KindCatchUp, KindSourceMoved, KindProvisioner, KindPullUpdated, KindDecided, + KindDeliveryMerged: n.follows[kind] = true return nil default: @@ -301,6 +302,8 @@ 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 diff --git a/internal/link/serve.go b/internal/link/serve.go index c5af8fc4..819b5648 100644 --- a/internal/link/serve.go +++ b/internal/link/serve.go @@ -98,6 +98,8 @@ type Server struct { healths Healths // decider acts on the operator's warrants for what the controller asked (novox/hq ADR 0259). decider Decider + // deliveryMerges keeps mesh-delivery's merges (KeepsDeliveryMerges). + deliveryMerges DeliveryMergeKeeper log *log.Logger // giveUp is how long one message is held for the store; zero means GiveUpAfter. @@ -155,6 +157,21 @@ func (s *Server) Decides(d Decider) error { return nil } +// 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. +type DeliveryMergeKeeper interface { + KeepDeliveryMerge(ctx context.Context, body []byte) error +} + +// 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 { + return err + } + s.deliveryMerges = k + return nil +} + // Answers says what to do about a catalogue's catch-up request, and asks for them to be delivered. func (s *Server) Answers(r Replayer) error { if err := s.inbound.Also(KindCatchUp); err != nil { @@ -229,6 +246,8 @@ func (s *Server) act(ctx context.Context, m Control) { s.pullUpdated(ctx, m) case KindDecided: s.decided(ctx, m) + case KindDeliveryMerged: + s.deliveryMerged(ctx, m) default: // Dropped: a message nothing understands will not be understood on the next attempt // either, and asking for it again would spin. @@ -728,6 +747,24 @@ func (s *Server) decided(ctx context.Context, m Control) { _ = m.Took() } +// deliveryMerged hands mesh-delivery's word on a merge to its keeper; a failure to keep it is held for the store +// like any word that must not be lost, and a word that cannot be read is said and let go. +func (s *Server) deliveryMerged(ctx context.Context, m Control) { + if s.deliveryMerges == nil { + _ = m.Took() + return + } + err := s.deliveryMerges.KeepDeliveryMerge(ctx, m.Body()) + switch s.decide(ctx, m, "mesh-delivery's word on a merge", "", "", err) { + case Hold: + return + } + if err != nil { + s.log.Printf("mesh-delivery's word on a merge could not be kept: %v", err) + } + _ = m.Took() +} + // saysWhatItDid states what a machine now runs, or what it would not take, as a fact on the bus // (novox/hq ADR 0134). // diff --git a/testdata/store-tests b/testdata/store-tests index 61113e7e..a93742a1 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -137,3 +137,6 @@ ./cmd/mesh-controller TestARolloutAdvancesOnEventsAndAReplacedControllerResumesIt ./cmd/mesh-controller TestAWalkLetGoByTheDeliverVerbIsAskedWithNoOtherEvent ./cmd/mesh-controller TestTwoMergesWithinAMinuteLeaveEachOpenPullRequestOneQueuedCheck +./cmd/mesh-controller TestADeliveryGroupMovingAModuleMachinesRunPassesTheGate +./cmd/mesh-controller TestAModuleMovesOnTheOperatorsApproveOfTheMergeThatMovedIt +./cmd/mesh-controller TestAModuleBuiltFromANewRepositoryIsRegisteredOnlyOnTheOperatorsApproveOnTheBus