Layouts of the indexer's taken and the view's reach, and the delivery seat's settle verb, so whether a merge is on the machines can be derived (issue 1280, ADR 0556)
This commit is contained in:
@@ -75,6 +75,19 @@ func deliveryVerbs() []Verb {
|
||||
Input: schema(map[string]string{"id": "the delivery's id", "why": "the condition it answers"},
|
||||
[]string{"id", "why"})},
|
||||
deliveryRemovedInR6("retire-history"),
|
||||
{Name: "settle", Description: "Ask the operator to settle deliveries held because their merge reached the " +
|
||||
"trunk without a passing check: one ask on the operator's channel for all of them, as an Approve is asked. " +
|
||||
"The verb moves nothing; on the operator's answer each goes to merged, and to delivered once every node " +
|
||||
"runs a build containing it (novox/hq ADR 0556). Refused for a delivery not held, or held for another " +
|
||||
"reason, naming why.",
|
||||
Input: map[string]any{"type": "object", "additionalProperties": false, "required": []string{"ids", "why"},
|
||||
"properties": map[string]any{
|
||||
"ids": map[string]any{"type": "array", "items": map[string]any{"type": "string"}, "minItems": 1,
|
||||
"description": "the held deliveries' ids (owner/repository@commit)"},
|
||||
"why": map[string]any{"type": "string", "description": "why they may be settled, shown to the operator"},
|
||||
}},
|
||||
// Added after the seat's release 6 holder shipped: optional until the gate serves it everywhere.
|
||||
Optional: true},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -86,7 +86,7 @@ func TestTheDeliverySeatHasNoPlanVerbs(t *testing.T) {
|
||||
if _, ok := rows["mark-ready"]; ok {
|
||||
t.Errorf("mark-ready is still promised")
|
||||
}
|
||||
for _, kept := range []string{"deliveries", "show", "groups", "checks", "table", "recheck", "close", "stalled"} {
|
||||
for _, kept := range []string{"deliveries", "show", "groups", "checks", "table", "recheck", "close", "stalled", "settle"} {
|
||||
if v, ok := rows[kept]; !ok || strings.HasPrefix(v.Description, "Removed") {
|
||||
t.Errorf("%s is not kept (ADR 0539 clause 17)", kept)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,154 @@
|
||||
package delivery
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// How far a merge reached (novox/hq ADR 0556, issue 1280). The indexer says, per repository, the last merge it took in
|
||||
// (TakenBucket); the delivery view derives, per repository, how far each of its modules reached on the nodes that want it
|
||||
// (ReachBucket); the gate, the one writer of a delivery's record, watches the view's state and moves a merged delivery to
|
||||
// delivered when every module its merge changed runs, on every node that wants it, at a build containing the merge.
|
||||
|
||||
// TakenBucket is the indexer's state `taken`: one key per repository (RepositoryKey), the last merge it took in.
|
||||
const TakenBucket = IndexerModule + "_taken"
|
||||
|
||||
// ReachBucket is the delivery view's state `reach`: one key per repository (RepositoryKey), how far its modules reached.
|
||||
const ReachBucket = ViewModule + "_reach"
|
||||
|
||||
// RepositoryKey is a repository's key in TakenBucket and ReachBucket: its forge name `owner/name` (RepositoryName), in
|
||||
// lower case as the forge compares it, escaped as a key part (EscapeName). An empty spelling, or one with no owner, is
|
||||
// refused.
|
||||
func RepositoryKey(repository string) (string, error) {
|
||||
name := strings.ToLower(RepositoryName(repository))
|
||||
if owner, repo, ok := strings.Cut(name, "/"); !ok || owner == "" || repo == "" {
|
||||
return "", fmt.Errorf("delivery: %q names no repository as owner/name", repository)
|
||||
}
|
||||
return EscapeName(name), nil
|
||||
}
|
||||
|
||||
// Taken is the value of a taken key: the repository, the merge commit the indexer last took in (applied its message
|
||||
// whole), and when.
|
||||
type Taken struct {
|
||||
V int `json:"v"`
|
||||
Repository string `json:"repository"`
|
||||
Merge string `json:"merge"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// EncodeTaken is t as JSON at Version; a value with no repository or merge is refused.
|
||||
func EncodeTaken(t Taken) ([]byte, error) {
|
||||
if t.Repository == "" || t.Merge == "" {
|
||||
return nil, fmt.Errorf("delivery: a taken value names its repository and merge")
|
||||
}
|
||||
t.V = Version
|
||||
return json.Marshal(t)
|
||||
}
|
||||
|
||||
// DecodeTaken reads a taken value; it refuses an unknown version (ErrUnknownVersion) and one with no merge.
|
||||
func DecodeTaken(data []byte) (Taken, error) {
|
||||
var t Taken
|
||||
if err := decode(data, &t); err != nil {
|
||||
return t, err
|
||||
}
|
||||
if t.Repository == "" || t.Merge == "" {
|
||||
return t, fmt.Errorf("delivery: a taken value names no repository or merge")
|
||||
}
|
||||
return t, nil
|
||||
}
|
||||
|
||||
// Reach is the value of a reach key: how far the modules of one repository reached on the nodes. Indexed is the last
|
||||
// merge the indexer took in for it (Taken.Merge), empty while it took in none. A repository the indexer holds no module
|
||||
// of (a library) has no Modules.
|
||||
type Reach struct {
|
||||
V int `json:"v"`
|
||||
Repository string `json:"repository"`
|
||||
Indexed string `json:"indexed,omitempty"`
|
||||
Modules map[string]ModuleReach `json:"modules,omitempty"`
|
||||
}
|
||||
|
||||
// ModuleReach is how far one module reached. FP is its fingerprint at the indexer now; Nodes how many nodes want it
|
||||
// (a want not removed); Changes its fingerprints the view knows of, each with the merge commit it first appeared at,
|
||||
// newest first. Floor is the oldest first-appearance commit among the fingerprints its nodes run, set only when every
|
||||
// node's want is settled (its have equal to the want, running and healthy) and every fingerprint they run is among
|
||||
// Changes; otherwise Waiting says why there is none. A module no node wants has neither.
|
||||
type ModuleReach struct {
|
||||
FP string `json:"fp,omitempty"`
|
||||
Nodes int `json:"nodes"`
|
||||
Changes []Change `json:"changes,omitempty"`
|
||||
Floor string `json:"floor,omitempty"`
|
||||
Waiting string `json:"waiting,omitempty"`
|
||||
}
|
||||
|
||||
// Change is one fingerprint of a module and the merge commit it first appeared at (Module.Commit).
|
||||
type Change struct {
|
||||
Commit string `json:"commit"`
|
||||
FP string `json:"fp"`
|
||||
}
|
||||
|
||||
// MostChanges is how many changes of a module a reach value keeps: as many revisions as the indexer's modules bucket keeps.
|
||||
const MostChanges = ModulesHistory
|
||||
|
||||
func (r Reach) check() error {
|
||||
if r.Repository == "" {
|
||||
return fmt.Errorf("delivery: a reach value names no repository")
|
||||
}
|
||||
for name, m := range r.Modules {
|
||||
if name == "" {
|
||||
return fmt.Errorf("delivery: a reach value names a module with no name")
|
||||
}
|
||||
if m.Nodes < 0 {
|
||||
return fmt.Errorf("delivery: the reach of %s counts %d nodes", name, m.Nodes)
|
||||
}
|
||||
if m.Floor != "" && m.Waiting != "" {
|
||||
return fmt.Errorf("delivery: the reach of %s has a floor and waits", name)
|
||||
}
|
||||
if len(m.Changes) > MostChanges {
|
||||
return fmt.Errorf("delivery: the reach of %s keeps %d changes, more than %d", name, len(m.Changes), MostChanges)
|
||||
}
|
||||
for _, c := range m.Changes {
|
||||
if c.Commit == "" || c.FP == "" {
|
||||
return fmt.Errorf("delivery: a change of %s names no commit or fingerprint", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// EncodeReach is r as JSON at Version; a value that does not check is refused.
|
||||
func EncodeReach(r Reach) ([]byte, error) {
|
||||
if err := r.check(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
r.V = Version
|
||||
return json.Marshal(r)
|
||||
}
|
||||
|
||||
// DecodeReach reads a reach value; it refuses an unknown version (ErrUnknownVersion) and one that does not check.
|
||||
func DecodeReach(data []byte) (Reach, error) {
|
||||
var r Reach
|
||||
if err := decode(data, &r); err != nil {
|
||||
return r, err
|
||||
}
|
||||
return r, r.check()
|
||||
}
|
||||
|
||||
// SameReach says whether two reach values are the same, so the view writes only when they differ.
|
||||
func SameReach(a, b Reach) bool {
|
||||
if a.Repository != b.Repository || a.Indexed != b.Indexed || len(a.Modules) != len(b.Modules) {
|
||||
return false
|
||||
}
|
||||
for _, name := range slices.Sorted(maps.Keys(a.Modules)) {
|
||||
x, ok := b.Modules[name]
|
||||
y := a.Modules[name]
|
||||
if !ok || x.FP != y.FP || x.Nodes != y.Nodes || x.Floor != y.Floor || x.Waiting != y.Waiting ||
|
||||
!slices.Equal(x.Changes, y.Changes) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestSourceKeys(t *testing.T) { // 1200/AC1
|
||||
@@ -196,7 +197,23 @@ func TestSourceValuesRoundTrip(t *testing.T) { // 1200/AC3
|
||||
t.Errorf("forge %s round trip: %+v %v", f.Kind, got, err)
|
||||
}
|
||||
}
|
||||
// 1280: the taken and reach values (ADR 0556).
|
||||
tk := Taken{Repository: "novox/mesh-sdk", Merge: "cccc", At: time.Date(2026, 10, 26, 20, 0, 0, 0, time.UTC)}
|
||||
data, err = EncodeTaken(tk)
|
||||
check("taken", data, err)
|
||||
tk.V = Version
|
||||
if got, err := DecodeTaken(data); err != nil || got != tk {
|
||||
t.Errorf("taken round trip: %+v %v", got, err)
|
||||
}
|
||||
r := sampleReach()
|
||||
data, err = EncodeReach(r)
|
||||
check("reach", data, err)
|
||||
if got, err := DecodeReach(data); err != nil || !SameReach(got, r) {
|
||||
t.Errorf("reach round trip: %+v %v", got, err)
|
||||
}
|
||||
for name, dec := range map[string]func([]byte) error{
|
||||
"taken": func(b []byte) error { _, err := DecodeTaken(b); return err },
|
||||
"reach": func(b []byte) error { _, err := DecodeReach(b); return err },
|
||||
"source": func(b []byte) error { _, err := DecodeSource(b); return err },
|
||||
"candidate": func(b []byte) error { _, err := DecodeCandidate(b); return err },
|
||||
"module": func(b []byte) error { _, err := DecodeModule(b); return err },
|
||||
@@ -282,6 +299,40 @@ func TestSourceValuesRefused(t *testing.T) { // 1200/AC4
|
||||
wantErr("unknown kind", err, "closed")
|
||||
_, err = EncodeForge(Forge{Kind: ForgePull, Repository: "nope", Number: 1, Head: "h", Action: "opened"})
|
||||
wantErr("bad repository", err, "nope")
|
||||
|
||||
// 1280: taken and reach values that do not hold (ADR 0556).
|
||||
_, err = EncodeTaken(Taken{Repository: "a/b"})
|
||||
wantErr("taken without merge", err, "merge")
|
||||
r := sampleReach()
|
||||
r.Repository = ""
|
||||
_, err = EncodeReach(r)
|
||||
wantErr("reach without repository", err, "repository")
|
||||
r = sampleReach()
|
||||
m := r.Modules["gitea"]
|
||||
m.Waiting = "gitea on ace does not yet run src2:b"
|
||||
r.Modules["gitea"] = m
|
||||
_, err = EncodeReach(r)
|
||||
wantErr("reach with a floor that waits", err, "gitea", "floor")
|
||||
r = sampleReach()
|
||||
m = r.Modules["gitea"]
|
||||
m.Changes = append(m.Changes, Change{Commit: "dddd"})
|
||||
r.Modules["gitea"] = m
|
||||
_, err = EncodeReach(r)
|
||||
wantErr("reach with a change naming no fingerprint", err, "gitea")
|
||||
r = sampleReach()
|
||||
m = r.Modules["gitea"]
|
||||
m.Changes = make([]Change, MostChanges+1)
|
||||
r.Modules["gitea"] = m
|
||||
_, err = EncodeReach(r)
|
||||
wantErr("reach keeping too many changes", err, "more than")
|
||||
}
|
||||
|
||||
// sampleReach is a reach value of one repository with a module at its floor and a library-like empty one.
|
||||
func sampleReach() Reach {
|
||||
return Reach{V: Version, Repository: "novox/mesh-catalog", Indexed: "cccc", Modules: map[string]ModuleReach{
|
||||
"gitea": {FP: "src2:a", Nodes: 1, Floor: "cccc", Changes: []Change{{Commit: "cccc", FP: "src2:a"}, {Commit: "bbbb", FP: "src2:b"}}},
|
||||
"zsh": {FP: "src2:z", Changes: []Change{{Commit: "aaaa", FP: "src2:z"}}},
|
||||
}}
|
||||
}
|
||||
|
||||
func TestSourceSame(t *testing.T) { // 1200/AC5
|
||||
@@ -314,14 +365,38 @@ func TestSourceSame(t *testing.T) { // 1200/AC5
|
||||
if SameForge(f, g) {
|
||||
t.Error("different forge messages are the same")
|
||||
}
|
||||
if !SameReach(sampleReach(), sampleReach()) {
|
||||
t.Error("equal reach values differ")
|
||||
}
|
||||
for name, change := range map[string]func(*Reach){
|
||||
"indexed": func(r *Reach) { r.Indexed = "dddd" },
|
||||
"floor": func(r *Reach) { m := r.Modules["gitea"]; m.Floor = "bbbb"; r.Modules["gitea"] = m },
|
||||
"changes": func(r *Reach) { m := r.Modules["zsh"]; m.Changes = nil; r.Modules["zsh"] = m },
|
||||
"modules": func(r *Reach) { delete(r.Modules, "zsh") },
|
||||
} {
|
||||
r := sampleReach()
|
||||
change(&r)
|
||||
if SameReach(r, sampleReach()) {
|
||||
t.Errorf("reach values differing in %s are the same", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSourceBuckets(t *testing.T) {
|
||||
if SourceBucket != "mesh-delivery-gate_source" || CandidatesBucket != "mesh-delivery-indexer_candidates" ||
|
||||
ModulesBucket != "mesh-delivery-indexer_modules" || SourceMainKey != "main" ||
|
||||
SourceHistory != 64 || CandidatesHistory != 10 || ModulesHistory != 20 {
|
||||
SourceHistory != 64 || CandidatesHistory != 10 || ModulesHistory != 20 ||
|
||||
TakenBucket != "mesh-delivery-indexer_taken" || ReachBucket != "mesh-delivery-view_reach" {
|
||||
t.Error("a bucket or history differs from the decided layout")
|
||||
}
|
||||
// 1280: a repository keys taken and reach by its forge name, in lower case, from either spelling (ADR 0556).
|
||||
for in, want := range map[string]string{"novox/Mesh-SDK": "novox/mesh-sdk", "http://h:3000/novox/mesh-sdk.git": "novox/mesh-sdk",
|
||||
"a/b.c": "a/b=2ec", "": "", "nope": "", "/x": ""} {
|
||||
got, err := RepositoryKey(in)
|
||||
if (want == "") != (err != nil) || got != want {
|
||||
t.Errorf("RepositoryKey(%q) = %q, %v; want %q", in, got, err, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestForgeDirs(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user