Merge pull request 'Delivery records are true at the source: the layouts for what a repository's merges reached (issue 1280)' (#102) from fix/delivery-records-true into main

go/catalogue/delivery_seat.go

@@ -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 whether it is on the machines " +
+			"is derived as for any merge (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},
 	}
 }
 

go/catalogue/verbs_r6_test.go

@@ -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)
 		}

go/delivery/gate.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"reflect"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 
@@ -87,6 +88,30 @@ func MergeAskKey(group string, heads []Head) (string, error) {
 	return AskKindMerge + "." + part + "." + hash, nil
 }
 
+// SettleGroupPrefix begins the group of a settle ask (novox/hq ADR 0556 clause 5): a person's ask to settle deliveries held
+// because their merge reached the trunk without a passing check. It is kept in the gate's asks under MergeAskKey like a merge
+// ask, so the teller carries it unchanged; its options bind the act AskVerb names for it. Git forbids a colon in a branch
+// name, so no pull request's group (its branch) ever begins with it.
+const SettleGroupPrefix = "settle:"
+
+// SettleGroup is the group of the settle ask of these delivery ids: the prefix and twelve hex digits of the sha256 over the
+// ids, sorted and joined by a newline, so the same deliveries asked again keep one key.
+func SettleGroup(ids []string) string {
+	sorted := append([]string(nil), ids...)
+	sort.Strings(sorted)
+	sum := sha256.Sum256([]byte(strings.Join(sorted, "\n")))
+	return SettleGroupPrefix + hex.EncodeToString(sum[:])[:12]
+}
+
+// AskVerb is the act the options of an ask of the gate bind, by its group: a settle ask's `mesh-delivery.settle`, any other
+// `mesh-delivery.merge`. The gate composes the act with it and the teller checks a warrant against it.
+func AskVerb(group string) string {
+	if strings.HasPrefix(group, SettleGroupPrefix) {
+		return "mesh-delivery.settle"
+	}
+	return "mesh-delivery.merge"
+}
+
 // ParseAskKey is the inverse of MergeAskKey: the kind, the group and the heads hash.
 func ParseAskKey(key string) (kind, group, hash string, err error) {
 	parts := strings.Split(key, ".")

go/delivery/gate_test.go

@@ -77,6 +77,21 @@ func TestAskID(t *testing.T) { // 1201/AC2
 	if AskID(k1) != AskID(k1) {
 		t.Error("the id is not stable")
 	}
+	// 1280 (ADR 0556 clause 5): a settle ask's group is the same for the same ids in any order, keys like any group, and its
+	// options bind mesh-delivery.settle; every other ask binds mesh-delivery.merge.
+	g := SettleGroup([]string{"novox/b@222222222222", "novox/a@111111111111"})
+	if g != SettleGroup([]string{"novox/a@111111111111", "novox/b@222222222222"}) || !strings.HasPrefix(g, SettleGroupPrefix) || len(g) != len(SettleGroupPrefix)+12 {
+		t.Errorf("SettleGroup = %q", g)
+	}
+	if k, err := MergeAskKey(g, gateHeads()); err != nil || !asks.UsableID(AskID(k)) {
+		t.Errorf("the settle ask's key %q: %v", k, err)
+	}
+	for group, want := range map[string]string{g: "mesh-delivery.settle", "feat/x": "mesh-delivery.merge",
+		"settle/foo": "mesh-delivery.merge", "settle": "mesh-delivery.merge"} { // a branch named settle/... is a merge's group
+		if got := AskVerb(group); got != want {
+			t.Errorf("AskVerb(%q) = %q, want %q", group, got, want)
+		}
+	}
 }
 
 func TestOwedKey(t *testing.T) { // 1201/AC3

go/delivery/reach.go

@@ -0,0 +1,167 @@
+package delivery
+
+import (
+	"encoding/json"
+	"fmt"
+	"maps"
+	"slices"
+	"strings"
+	"time"
+)
+
+// How far a repository reached (novox/hq ADR 0556, issue 1280). The indexer says, per repository, the last merge it took in
+// (TakenBucket); the delivery view keeps, per repository, how far each of its modules reached on the nodes that want it, and
+// derives from it the repository's reached head (ReachBucket): the newest head of the repository at which every wanted
+// module runs, on every node, a build at or after its last change. A merge at or before the reached head is on the
+// machines. Nothing records a merge as delivered (ADR 0429 §1): the gate and the tracker derive it from this state.
+
+// 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. Reached is the reached head: the newest head of the repository in the source's history,
+// not past Indexed, at which every module with a node that wants it has its floor at or after its last change at or before
+// that head; ReachedSrc is the source revision that named it. Holds says, in words, what keeps the next head of the
+// repository from being reached (a module on a node, the indexer's lag); empty when the repository's newest head is reached.
+type Reach struct {
+	V          int                    `json:"v"`
+	Repository string                 `json:"repository"`
+	Indexed    string                 `json:"indexed,omitempty"`
+	Modules    map[string]ModuleReach `json:"modules,omitempty"`
+	Reached    string                 `json:"reached,omitempty"`
+	ReachedSrc uint64                 `json:"reached_src,omitempty"`
+	Holds      []string               `json:"holds,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' haves run, a running have
+// counting whether its health check passes or not; it is empty when a node that wants the module has no running have, or
+// runs a fingerprint not among Changes. Waiting names, in words, every node where the module is not settled (its have
+// different from its want, not running, or not healthy), with or without a floor: health is said here and never moves the
+// floor (ADR 0556 clause 2). A module whose floor is empty while a node wants it holds no head of the repository back: a
+// reader takes a merge among its Changes as not yet on the machines, with these words. 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, the merge commit it first appeared at (Module.Commit), and the source revision
+// that named that commit as the repository's head (Module.Src, or the source's history), 0 when not known.
+type Change struct {
+	Commit string `json:"commit"`
+	FP     string `json:"fp"`
+	Src    uint64 `json:"src,omitempty"`
+}
+
+// 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")
+	}
+	if (r.Reached == "") != (r.ReachedSrc == 0) {
+		return fmt.Errorf("delivery: the reach of %s names its reached head and its source revision together", r.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 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) ||
+		a.Reached != b.Reached || a.ReachedSrc != b.ReachedSrc || !slices.Equal(a.Holds, b.Holds) {
+		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
+}

go/delivery/source_test.go

@@ -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,38 @@ 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.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")
+	r = sampleReach()
+	r.ReachedSrc = 0
+	_, err = EncodeReach(r)
+	wantErr("reach with a head and no revision", err, "reached head")
+}
+
+// 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", Reached: "cccc", ReachedSrc: 9, Modules: map[string]ModuleReach{
+		"gitea": {FP: "src2:a", Nodes: 1, Floor: "cccc", Waiting: "gitea on ace runs src2:a and its health check failed", Changes: []Change{{Commit: "cccc", FP: "src2:a", Src: 9}, {Commit: "bbbb", FP: "src2:b", Src: 7}}},
+		"zsh":   {FP: "src2:z", Changes: []Change{{Commit: "aaaa", FP: "src2:z"}}},
+	}}
 }
 
 func TestSourceSame(t *testing.T) { // 1200/AC5
@@ -314,14 +363,40 @@ 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") },
+		"reached": func(r *Reach) { r.Reached, r.ReachedSrc = "bbbb", 7 },
+		"holds":   func(r *Reach) { r.Holds = []string{"gitea on ace does not yet run src2:a"} },
+	} {
+		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) {