Merge b2/registry (B2e: window, buildstep ask, buildrun, gate codecs) into b7/registry

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-05 15:58:00 +01:00
16 changed files with 1925 additions and 66 deletions
@@ -278,3 +278,35 @@ func CheckPreview(p checker.PreviewsByRepositoryHeadTrunkValue) error {
}
return nil
}
// VerdictKey is the key, in the state verdicts, of a group's check at its heads: `<escaped group>.<heads hash>`, the
// group escaped so a branch such as `fix/1.2` keys, the heads hash as the mesh-delivery client's HeadsHash gives it
// (12 lowercase hex digits).
func VerdictKey(group, headsHash string) (string, error) {
if group == "" {
return "", fmt.Errorf("meshdeliverycheckerclient: the group name is empty")
}
if !isHex12(headsHash) {
return "", fmt.Errorf("meshdeliverycheckerclient: %q is not a heads hash, 12 lowercase hex digits", headsHash)
}
return keys.EscapeName(group) + "." + headsHash, nil
}
// ParseVerdictKey is the inverse of VerdictKey.
func ParseVerdictKey(key string) (group, headsHash string, err error) {
i := strings.LastIndex(key, ".")
if i < 0 {
return "", "", fmt.Errorf("meshdeliverycheckerclient: %q is not a verdict key, <group>.<heads hash>", key)
}
part, headsHash := key[:i], key[i+1:]
if !isHex12(headsHash) {
return "", "", fmt.Errorf("meshdeliverycheckerclient: %q is not a verdict key: %q is not a heads hash, 12 lowercase hex digits", key, headsHash)
}
if part == "" || strings.Contains(part, ".") {
return "", "", fmt.Errorf("meshdeliverycheckerclient: %q is not a verdict key: the group part is empty or holds a dot", key)
}
if group, err = keys.UnescapeName(part); err != nil {
return "", "", fmt.Errorf("meshdeliverycheckerclient: %q is not a verdict key: %w", key, err)
}
return group, headsHash, nil
}
@@ -125,3 +125,27 @@ func previewIsRefused(t *testing.T) {
t.Errorf("a head in conflict is refused: %v", err)
}
}
// A verdict's key as go/delivery's CheckerVerdictKey wrote it (golden): the group escaped, the heads hash after the
// last dot.
func TestVerdictKeyAsGoDeliveryWroteIt(t *testing.T) {
key, err := VerdictKey("fix/1.2", "0123456789ab")
if err != nil || key != "fix/1=2e2.0123456789ab" {
t.Fatalf("%q %v", key, err)
}
group, heads, err := ParseVerdictKey(key)
if err != nil || group != "fix/1.2" || heads != "0123456789ab" {
t.Fatalf("%q %q %v", group, heads, err)
}
for _, bad := range []string{"nodot", "g.XYZ", ".0123456789ab", "a.b.0123456789ab", "g=zz.0123456789ab"} {
if _, _, err := ParseVerdictKey(bad); err == nil {
t.Errorf("%q parses", bad)
}
}
if _, err := VerdictKey("", "0123456789ab"); err == nil {
t.Error("an empty group keys")
}
if _, err := VerdictKey("g", "0123"); err == nil {
t.Error("a short hash keys")
}
}
@@ -0,0 +1,407 @@
package meshdeliveryclient
import (
"bytes"
"encoding/json"
"fmt"
"reflect"
"strconv"
"strings"
"time"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// The values of the seat's states `source`, `merges`, `asks`, `owed` and `turns`, in the seat's types: each is stamped
// with Version when encoded, read strictly (an unknown field refused) and its version checked when decoded, and its rules
// held on both sides, so the gate that writes them and the readers (the forge bridge, the teller, the assigner, the view)
// agree. What the schema cannot say is written here once (ADR 0625 §4).
// Version is the layout version of the seat's values.
const Version = 1
// MergeState is the state of a pull request in a merge.
type MergeState string
// The states a merge's pull request is in.
const (
MergeMerging MergeState = "merging"
MergeMerged MergeState = "merged"
MergeRefused MergeState = "refused"
)
// SourceMainKey is the one key of the state source.
const SourceMainKey = "main"
func encodeValue(v any, check error, stamp func()) ([]byte, error) {
if check != nil {
return nil, check
}
stamp()
return json.Marshal(v)
}
// ApprovedKey is the key, in a merge's approved heads, of a pull request: `<repository>@<number>`.
func ApprovedKey(repository string, number int) string {
return repository + "@" + strconv.Itoa(number)
}
// TurnsKey is the key, in the state turns, of a repository: `<owner>.<name>`, lower-case and escaped.
func TurnsKey(repository string) (string, error) {
repository = strings.ToLower(repository)
part, err := escapeRepository(repository)
if err != nil {
return "", err
}
return strings.Replace(part, "/", ".", 1), nil
}
// ParseTurnsKey is the inverse of TurnsKey; a key that does not name a lower-case <owner>/<name> is refused.
func ParseTurnsKey(key string) (string, error) {
owner, name, ok := strings.Cut(key, ".")
if !ok || owner == "" || name == "" || strings.Contains(name, ".") {
return "", fmt.Errorf("meshdeliveryclient: %q is not a turns key, <owner>.<name>", key)
}
repository, err := unescapeRepository(owner + "/" + name)
if err != nil {
return "", fmt.Errorf("meshdeliveryclient: %q is not a turns key: %w", key, err)
}
if again, err := TurnsKey(repository); err != nil || again != key {
return "", fmt.Errorf("meshdeliveryclient: %q is not a turns key: it does not name a lowercase <owner>/<name>", key)
}
return repository, nil
}
// ----- source
// CheckSource refuses a source naming no repository, a repository with no name or an empty head.
func CheckSource(s seat.SourceByMainValue) error {
if len(s.Heads) == 0 {
return fmt.Errorf("meshdeliveryclient: a source names no repository")
}
for repository, head := range s.Heads {
if repository == "" {
return fmt.Errorf("meshdeliveryclient: a source names a repository with no name")
}
if head == "" {
return fmt.Errorf("meshdeliveryclient: the source's head of %s is empty", repository)
}
}
return nil
}
// EncodeSource is s as the value of the key main.
func EncodeSource(s seat.SourceByMainValue) ([]byte, error) {
return encodeValue(&s, CheckSource(s), func() { s.V = Version })
}
// DecodeSource reads the value of the key main.
func DecodeSource(data []byte) (seat.SourceByMainValue, error) {
if err := keys.CheckVersion(data); err != nil {
return seat.SourceByMainValue{}, err
}
s, err := seat.DecodeSourceByMainValue(data)
if err != nil {
return s, fmt.Errorf("meshdeliveryclient: a source does not fit its shape: %w", err)
}
return s, CheckSource(s)
}
// SameSource says whether two sources are the same.
func SameSource(a, b seat.SourceByMainValue) bool { return reflect.DeepEqual(a, b) }
// ----- merge
// CheckMerge refuses a merge naming no group, ask or pull request, a pull request outside the approved heads or at
// another head than the Approve was for, and a state that lacks what it needs: merged its commit, refused the forge's words.
func CheckMerge(m seat.MergesByGroupValue) error {
if m.Group == "" {
return fmt.Errorf("meshdeliveryclient: a merge names no group")
}
if m.Ask == "" {
return fmt.Errorf("meshdeliveryclient: a merge names no ask")
}
if len(m.Pulls) == 0 {
return fmt.Errorf("meshdeliveryclient: a merge names no pull request")
}
for key, sha := range m.Approved {
if sha == "" {
return fmt.Errorf("meshdeliveryclient: the approved head of %s is empty", key)
}
}
for _, p := range m.Pulls {
who := fmt.Sprintf("%s#%d", p.Repository, p.Number)
if len(m.Approved) > 0 {
sha, ok := m.Approved[ApprovedKey(p.Repository, int(p.Number))]
if !ok {
return fmt.Errorf("meshdeliveryclient: the pull request %s is handed to the merge but is not among the approved heads", who)
}
if sha != p.Head {
return fmt.Errorf("meshdeliveryclient: the pull request %s is handed at head %q but the Approve was for %q", who, p.Head, sha)
}
}
switch MergeState(p.State) {
case MergeMerging:
case MergeMerged:
if p.Commit == "" {
return fmt.Errorf("meshdeliveryclient: the merged pull request %s has no merge commit", who)
}
case MergeRefused:
if p.Words == "" {
return fmt.Errorf("meshdeliveryclient: the refused pull request %s has no words of the forge", who)
}
default:
return fmt.Errorf("meshdeliveryclient: the state %q of %s is not merging, merged or refused", p.State, who)
}
}
return nil
}
// EncodeMerge is m as the value of a group's key.
func EncodeMerge(m seat.MergesByGroupValue) ([]byte, error) {
return encodeValue(&m, CheckMerge(m), func() { m.V = Version })
}
// DecodeMerge reads the value of a group's key.
func DecodeMerge(data []byte) (seat.MergesByGroupValue, error) {
if err := keys.CheckVersion(data); err != nil {
return seat.MergesByGroupValue{}, err
}
m, err := seat.DecodeMergesByGroupValue(data)
if err != nil {
return m, fmt.Errorf("meshdeliveryclient: a merge does not fit its shape: %w", err)
}
return m, CheckMerge(m)
}
// SameMerge says whether two merges are the same.
func SameMerge(a, b seat.MergesByGroupValue) bool { return reflect.DeepEqual(a, b) }
// ----- ask
// HeadsOf are an ask's heads as Heads (for HeadsHash and MergeAskKey).
func HeadsOf(items []seat.AsksByMergeGroupHeadsValueHeadsItem) []Head {
out := make([]Head, len(items))
for i, h := range items {
out[i] = Head{Repository: h.Repository, Number: int(h.Number), Head: h.Head}
}
return out
}
// CheckAsk refuses an ask with no group or heads, a negative count, or an ask that is no object.
func CheckAsk(a seat.AsksByMergeGroupHeadsValue) error {
if a.Asking < 0 {
return fmt.Errorf("meshdeliveryclient: the ask's asking %d is negative", a.Asking)
}
if _, err := MergeAskKey(a.Group, HeadsOf(a.Heads)); err != nil {
return fmt.Errorf("meshdeliveryclient: an ask needs its group and at least one pull request's head: %w", err)
}
raw, err := json.Marshal(a.Ask)
var fields map[string]json.RawMessage
if err != nil || len(bytes.TrimSpace(raw)) == 0 || json.Unmarshal(raw, &fields) != nil || fields == nil {
return fmt.Errorf("meshdeliveryclient: the value carries no router's ask")
}
return nil
}
// EncodeAsk is a as the value of a merge ask's key; a.Ask may be a json.RawMessage.
func EncodeAsk(a seat.AsksByMergeGroupHeadsValue) ([]byte, error) {
return encodeValue(&a, CheckAsk(a), func() { a.V = Version })
}
// DecodeAsk reads the value of a merge ask's key; its Ask is the router's ask as a decoded object.
func DecodeAsk(data []byte) (seat.AsksByMergeGroupHeadsValue, error) {
if err := keys.CheckVersion(data); err != nil {
return seat.AsksByMergeGroupHeadsValue{}, err
}
a, err := seat.DecodeAsksByMergeGroupHeadsValue(data)
if err != nil {
return a, fmt.Errorf("meshdeliveryclient: an ask does not fit its shape: %w", err)
}
return a, CheckAsk(a)
}
// SameAsk says whether two asks are the same (the router's ask by its content).
func SameAsk(a, b seat.AsksByMergeGroupHeadsValue) bool { return reflect.DeepEqual(a, b) }
// ----- owed
// EncodeOwed is o as the value of an owed key, after CheckOwed.
func EncodeOwed(o seat.OwedByRepositoryNumberHeadActValue) ([]byte, error) {
return encodeValue(&o, CheckOwed(o), func() { o.V = Version })
}
// DecodeOwed reads the value of an owed key.
func DecodeOwed(data []byte) (seat.OwedByRepositoryNumberHeadActValue, error) {
if err := keys.CheckVersion(data); err != nil {
return seat.OwedByRepositoryNumberHeadActValue{}, err
}
o, err := seat.DecodeOwedByRepositoryNumberHeadActValue(data)
if err != nil {
return o, fmt.Errorf("meshdeliveryclient: an owed value does not fit its shape: %w", err)
}
return o, CheckOwed(o)
}
// SameOwed says whether two owed values are the same.
func SameOwed(a, b seat.OwedByRepositoryNumberHeadActValue) bool { return a == b }
// ----- turns
// CheckTurn refuses a turn whose repository is not lower-case, that names no delivery, or whose deliveries repeat a heads
// hash, lack their times, are out of the order of the Approve, or name a wait without a verdict key.
func CheckTurn(t seat.TurnsByRepositoryValue) error {
repository := strings.ToLower(t.Repository)
if repository != t.Repository {
return fmt.Errorf("meshdeliveryclient: the turn's repository %q is not lowercase", t.Repository)
}
if _, err := escapeRepository(repository); err != nil {
return err
}
if len(t.Waiting) == 0 {
return fmt.Errorf("meshdeliveryclient: the turn of %s names no delivery: its key is deleted instead", t.Repository)
}
seen := map[string]bool{}
for i, w := range t.Waiting {
if w.Group == "" {
return fmt.Errorf("meshdeliveryclient: a delivery in the turn of %s names no group", t.Repository)
}
if !hashOK.MatchString(w.Heads) {
return fmt.Errorf("meshdeliveryclient: the heads hash %q in the turn of %s is not 12 lowercase hex digits", w.Heads, t.Repository)
}
if seen[w.Heads] {
return fmt.Errorf("meshdeliveryclient: the turn of %s names the heads %s twice", t.Repository, w.Heads)
}
seen[w.Heads] = true
if w.Approved.IsZero() || w.Since.IsZero() {
return fmt.Errorf("meshdeliveryclient: the delivery %s in the turn of %s says no time of its Approve or of its wait", w.Group, t.Repository)
}
if i > 0 && w.Approved.Before(t.Waiting[i-1].Approved) {
return fmt.Errorf("meshdeliveryclient: the turn of %s is not in the order of the Approve: %s was approved before %s", t.Repository, w.Group, t.Waiting[i-1].Group)
}
if w.WaitsFor != "" {
if _, _, err := parseVerdictKey(w.WaitsFor); err != nil {
return fmt.Errorf("meshdeliveryclient: the delivery %s in the turn of %s waits for what is not a verdict key: %w", w.Group, t.Repository, err)
}
} else if (w.WaitsSince != nil && !w.WaitsSince.IsZero()) || w.Step != "" {
return fmt.Errorf("meshdeliveryclient: the delivery %s in the turn of %s says when and where a check runs, and names no check", w.Group, t.Repository)
}
}
return nil
}
// parseVerdictKey is the checker's verdict key `<group>.<heads hash>` (the checker's client has the same rule over its
// own seat; a client uses its own seat's contract alone).
func parseVerdictKey(key string) (group, heads string, err error) {
i := strings.LastIndex(key, ".")
if i < 0 || !hashOK.MatchString(key[i+1:]) || i == 0 || strings.Contains(key[:i], ".") {
return "", "", fmt.Errorf("%q is not a verdict key, <group>.<heads hash>", key)
}
if group, err = keys.UnescapeName(key[:i]); err != nil {
return "", "", err
}
return group, key[i+1:], nil
}
// TurnHolds says whether a delivery at these heads is not in the turn while the turn names others: the delivery's
// check or merge holds back for the one before it.
func TurnHolds(t seat.TurnsByRepositoryValue, heads string) bool {
if len(t.Waiting) == 0 {
return false
}
for _, w := range t.Waiting {
if w.Heads == heads {
return false
}
}
return true
}
// EncodeTurn is t as the value of a repository's key.
func EncodeTurn(t seat.TurnsByRepositoryValue) ([]byte, error) {
if err := CheckTurn(t); err != nil {
return nil, err
}
t.V = Version
return json.Marshal(t)
}
// DecodeTurn reads the value of a repository's key.
func DecodeTurn(data []byte) (seat.TurnsByRepositoryValue, error) {
if err := keys.CheckVersion(data); err != nil {
return seat.TurnsByRepositoryValue{}, err
}
t, err := seat.DecodeTurnsByRepositoryValue(data)
if err != nil {
return t, fmt.Errorf("meshdeliveryclient: a turn does not fit its shape: %w", err)
}
return t, CheckTurn(t)
}
// SameTurn says whether two turns are the same, a time by the instant it names.
func SameTurn(a, b seat.TurnsByRepositoryValue) bool {
if len(a.Waiting) != len(b.Waiting) || a.Repository != b.Repository || a.V != b.V {
return false
}
for i := range a.Waiting {
x, y := a.Waiting[i], b.Waiting[i]
if !x.Approved.Equal(y.Approved) || !x.Since.Equal(y.Since) || !sameTime(x.WaitsSince, y.WaitsSince) {
return false
}
x.Approved, y.Approved, x.Since, y.Since, x.WaitsSince, y.WaitsSince = time.Time{}, time.Time{}, time.Time{}, time.Time{}, nil, nil
if !reflect.DeepEqual(x, y) {
return false
}
}
return true
}
func sameTime(a, b *time.Time) bool {
if a == nil || a.IsZero() {
return b == nil || b.IsZero()
}
return b != nil && a.Equal(*b)
}
// Answer is the operator's answer to a merge ask as the gate acts on it: the option chosen, the warrant that proves who
// answered, and the words said. It is the teller's state `answers` under the ask's key (the teller's own record; this is
// its layout for the gate that reads it).
type Answer struct {
V int `json:"v"`
Option string `json:"option"`
Warrant string `json:"warrant"`
Words string `json:"words"`
}
// CheckAnswer refuses an answer lacking its option, warrant or words.
func CheckAnswer(a Answer) error {
for _, f := range []struct{ name, value string }{{"option", a.Option}, {"warrant", a.Warrant}, {"words", a.Words}} {
if f.value == "" {
return fmt.Errorf("meshdeliveryclient: an answer has no %s", f.name)
}
}
return nil
}
// EncodeAnswer is a as the value of an answer's key.
func EncodeAnswer(a Answer) ([]byte, error) {
return encodeValue(&a, CheckAnswer(a), func() { a.V = Version })
}
// DecodeAnswer reads the value of an answer's key.
func DecodeAnswer(data []byte) (Answer, error) {
if err := keys.CheckVersion(data); err != nil {
return Answer{}, err
}
var a Answer
dec := json.NewDecoder(bytes.NewReader(data))
dec.DisallowUnknownFields()
if err := dec.Decode(&a); err != nil {
return Answer{}, fmt.Errorf("meshdeliveryclient: an answer does not fit its shape: %w", err)
}
return a, CheckAnswer(a)
}
// SameAnswer says whether two answers are the same.
func SameAnswer(a, b Answer) bool { return a == b }
@@ -0,0 +1,186 @@
package meshdeliveryclient
import (
"encoding/json"
"errors"
"reflect"
"strings"
"testing"
"time"
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// sameJSON says whether two documents hold the same values (the generated types write their fields in alphabetical
// order, the old package's in declaration order: the bytes differ, the documents do not).
func sameJSON(t *testing.T, got []byte, want string) {
t.Helper()
var g, w any
if err := json.Unmarshal(got, &g); err != nil {
t.Fatalf("%s: %v", got, err)
}
if err := json.Unmarshal([]byte(want), &w); err != nil {
t.Fatal(err)
}
if !reflect.DeepEqual(g, w) {
t.Errorf("got %s\nwant %s", got, want)
}
}
// The goldens are the output of go/delivery's Encode* before it was deleted; each value must encode to the same
// document and decode back to what it was.
func TestGateValuesEncodeAsGoDeliveryDid(t *testing.T) {
ap := time.Date(2026, 11, 5, 10, 0, 0, 0, time.UTC)
wait := ap.Add(2 * time.Minute)
for _, c := range []struct {
name string
encode func() ([]byte, error)
decode func([]byte) (any, error)
value any
golden string
}{
{"source", func() ([]byte, error) {
return EncodeSource(seat.SourceByMainValue{Group: "fix/1.2", Heads: map[string]string{"novox/a": "aaaa", "novox/b": "bbbb"}})
}, func(b []byte) (any, error) { return DecodeSource(b) },
seat.SourceByMainValue{V: 1, Group: "fix/1.2", Heads: map[string]string{"novox/a": "aaaa", "novox/b": "bbbb"}},
`{"v":1,"group":"fix/1.2","heads":{"novox/a":"aaaa","novox/b":"bbbb"}}`},
{"merge", func() ([]byte, error) {
return EncodeMerge(seat.MergesByGroupValue{Group: "g", Ask: "merge.g.0123456789ab", Approved: map[string]string{"novox/a@7": "h1"},
Pulls: []seat.MergesByGroupValuePullsItem{{Repository: "novox/a", Number: 7, Head: "h1", State: "merged", Commit: "c1"}}})
}, func(b []byte) (any, error) { return DecodeMerge(b) },
seat.MergesByGroupValue{V: 1, Group: "g", Ask: "merge.g.0123456789ab", Approved: map[string]string{"novox/a@7": "h1"},
Pulls: []seat.MergesByGroupValuePullsItem{{Repository: "novox/a", Number: 7, Head: "h1", State: "merged", Commit: "c1"}}},
`{"v":1,"group":"g","ask":"merge.g.0123456789ab","approved":{"novox/a@7":"h1"},"pulls":[{"repository":"novox/a","number":7,"head":"h1","state":"merged","commit":"c1"}]}`},
{"ask", func() ([]byte, error) {
return EncodeAsk(seat.AsksByMergeGroupHeadsValue{Group: "g", Asking: 2, Ask: json.RawMessage(`{"headline":"x"}`),
Heads: []seat.AsksByMergeGroupHeadsValueHeadsItem{{Repository: "novox/a", Number: 7, Head: "h1"}}})
}, func(b []byte) (any, error) { return DecodeAsk(b) },
seat.AsksByMergeGroupHeadsValue{V: 1, Group: "g", Asking: 2, Ask: map[string]any{"headline": "x"},
Heads: []seat.AsksByMergeGroupHeadsValueHeadsItem{{Repository: "novox/a", Number: 7, Head: "h1"}}},
`{"v":1,"group":"g","heads":[{"repository":"novox/a","number":7,"head":"h1"}],"ask":{"headline":"x"},"asking":2}`},
{"owed", func() ([]byte, error) {
return EncodeOwed(seat.OwedByRepositoryNumberHeadActValue{Act: OwedStatus, Context: StatusContextCheck, Target: "t", State: "pending"})
}, func(b []byte) (any, error) { return DecodeOwed(b) },
seat.OwedByRepositoryNumberHeadActValue{V: 1, Act: OwedStatus, Context: StatusContextCheck, Target: "t", State: "pending"},
`{"v":1,"act":"status","context":"mesh/check","target":"t","state":"pending"}`},
{"turn", func() ([]byte, error) {
return EncodeTurn(seat.TurnsByRepositoryValue{Repository: "novox/a", Waiting: []seat.TurnsByRepositoryValueWaitingItem{{
Group: "g", Heads: "0123456789ab", Ask: "k", Approved: ap, Since: ap.Add(time.Minute), WaitsFor: "g.0123456789ab", WaitsSince: &wait, Step: "s"}}})
}, func(b []byte) (any, error) { return DecodeTurn(b) },
seat.TurnsByRepositoryValue{V: 1, Repository: "novox/a", Waiting: []seat.TurnsByRepositoryValueWaitingItem{{
Group: "g", Heads: "0123456789ab", Ask: "k", Approved: ap, Since: ap.Add(time.Minute), WaitsFor: "g.0123456789ab", WaitsSince: &wait, Step: "s"}}},
`{"v":1,"repository":"novox/a","waiting":[{"group":"g","heads":"0123456789ab","ask":"k","approved":"2026-11-05T10:00:00Z","since":"2026-11-05T10:01:00Z","waits-for":"g.0123456789ab","waits-since":"2026-11-05T10:02:00Z","step":"s"}]}`},
{"answer", func() ([]byte, error) { return EncodeAnswer(Answer{Option: "merge", Warrant: "w", Words: "ok"}) },
func(b []byte) (any, error) { return DecodeAnswer(b) }, Answer{V: 1, Option: "merge", Warrant: "w", Words: "ok"},
`{"v":1,"option":"merge","warrant":"w","words":"ok"}`},
} {
raw, err := c.encode()
if err != nil {
t.Fatalf("%s: %v", c.name, err)
}
sameJSON(t, raw, c.golden)
back, err := c.decode([]byte(c.golden))
if err != nil || !reflect.DeepEqual(back, c.value) {
t.Errorf("%s decoded %+v, %v; want %+v", c.name, back, err, c.value)
}
}
}
func TestGateKeysAsGoDeliveryHadThem(t *testing.T) {
for in, want := range map[string]string{"novox/Mesh-sdk": "novox.mesh-sdk", "novox/mesh-sdk": "novox.mesh-sdk", "novox/mesh.x": "novox.mesh=2ex"} {
got, err := TurnsKey(in)
if err != nil || got != want {
t.Errorf("TurnsKey(%q) = %q, %v; want %q", in, got, err, want)
}
}
if r, err := ParseTurnsKey("novox.mesh=2ex"); err != nil || r != "novox/mesh.x" {
t.Errorf("%q %v", r, err)
}
for _, bad := range []string{"novox", "novox.Mesh", "a.b.c", ".x", "novox.mesh-sdk."} {
if _, err := ParseTurnsKey(bad); err == nil {
t.Errorf("%q parses as a turns key", bad)
}
}
if _, err := TurnsKey("novox"); err == nil {
t.Error("a repository with no slash keys")
}
if k := ApprovedKey("novox/a", 7); k != "novox/a@7" {
t.Error(k)
}
if SourceMainKey != seat.SourceByMainKey() {
t.Error("the source's key is the seat's")
}
}
func TestGateValuesRefuseWhatTheirRulesRefuse(t *testing.T) {
ap := time.Date(2026, 11, 5, 10, 0, 0, 0, time.UTC)
pull := seat.MergesByGroupValuePullsItem{Repository: "novox/a", Number: 7, Head: "h1", State: "merging"}
merge := func(edit func(*seat.MergesByGroupValue)) error {
m := seat.MergesByGroupValue{Group: "g", Ask: "k", Pulls: []seat.MergesByGroupValuePullsItem{pull}}
edit(&m)
_, err := EncodeMerge(m)
return err
}
turn := func(edit func(*seat.TurnsByRepositoryValue)) error {
tv := seat.TurnsByRepositoryValue{Repository: "novox/a", Waiting: []seat.TurnsByRepositoryValueWaitingItem{{Group: "g", Heads: "0123456789ab", Approved: ap, Since: ap}}}
edit(&tv)
_, err := EncodeTurn(tv)
return err
}
for name, c := range map[string]struct {
err error
want string
}{
"merge: no group": {merge(func(m *seat.MergesByGroupValue) { m.Group = "" }), "no group"},
"merge: no pull": {merge(func(m *seat.MergesByGroupValue) { m.Pulls = nil }), "no pull request"},
"merge: not approved": {merge(func(m *seat.MergesByGroupValue) { m.Approved = map[string]string{"novox/b@1": "x"} }), "not among the approved"},
"merge: other head": {merge(func(m *seat.MergesByGroupValue) { m.Approved = map[string]string{"novox/a@7": "x"} }), "Approve was for"},
"merge: merged no commit": {merge(func(m *seat.MergesByGroupValue) { m.Pulls[0].State = "merged" }), "no merge commit"},
"merge: refused no words": {merge(func(m *seat.MergesByGroupValue) { m.Pulls[0].State = "refused" }), "no words"},
"merge: unknown state": {merge(func(m *seat.MergesByGroupValue) { m.Pulls[0].State = "lost" }), "not merging, merged or refused"},
"turn: upper case": {turn(func(v *seat.TurnsByRepositoryValue) { v.Repository = "Novox/a" }), "not lowercase"},
"turn: empty": {turn(func(v *seat.TurnsByRepositoryValue) { v.Waiting = nil }), "names no delivery"},
"turn: bad hash": {turn(func(v *seat.TurnsByRepositoryValue) { v.Waiting[0].Heads = "xyz" }), "12 lowercase hex"},
"turn: no time": {turn(func(v *seat.TurnsByRepositoryValue) { v.Waiting[0].Since = time.Time{} }), "no time"},
"turn: step with no wait": {turn(func(v *seat.TurnsByRepositoryValue) { v.Waiting[0].Step = "s" }), "names no check"},
"turn: not a verdict key": {turn(func(v *seat.TurnsByRepositoryValue) { v.Waiting[0].WaitsFor = "nope" }), "not a verdict key"},
} {
if c.err == nil || !strings.Contains(c.err.Error(), c.want) {
t.Errorf("%s: %v, want %q", name, c.err, c.want)
}
}
if _, err := EncodeSource(seat.SourceByMainValue{Group: "g"}); err == nil {
t.Error("an empty source encodes")
}
if _, err := EncodeAsk(seat.AsksByMergeGroupHeadsValue{Group: "g", Ask: "text", Heads: []seat.AsksByMergeGroupHeadsValueHeadsItem{{Repository: "novox/a", Number: 1, Head: "h"}}}); err == nil {
t.Error("an ask that is no object encodes")
}
if _, err := EncodeAnswer(Answer{Option: "merge"}); err == nil {
t.Error("an answer with no warrant encodes")
}
if _, err := DecodeSource([]byte(`{"v":2,"group":"g","heads":{"a":"b"}}`)); !errors.Is(err, keys.ErrUnknownVersion) {
t.Errorf("another version: %v", err)
}
if _, err := DecodeMerge([]byte(`{"v":1,"group":"g","ask":"k","pulls":[],"extra":1}`)); err == nil {
t.Error("an unknown field decodes")
}
}
func TestTurnHoldsAndSameTurn(t *testing.T) {
ap := time.Date(2026, 11, 5, 10, 0, 0, 0, time.UTC)
tv := seat.TurnsByRepositoryValue{V: 1, Repository: "novox/a", Waiting: []seat.TurnsByRepositoryValueWaitingItem{{Group: "g", Heads: "0123456789ab", Approved: ap, Since: ap}}}
if TurnHolds(tv, "0123456789ab") || !TurnHolds(tv, "ba9876543210") || TurnHolds(seat.TurnsByRepositoryValue{}, "x") {
t.Error("a turn holds a delivery that is not in it while others wait")
}
other := tv
other.Waiting = []seat.TurnsByRepositoryValueWaitingItem{tv.Waiting[0]}
other.Waiting[0].Approved = ap.In(time.FixedZone("x", 3600))
if !SameTurn(tv, other) {
t.Error("the same instant in another zone is another turn")
}
other.Waiting[0].Since = ap.Add(time.Second)
if SameTurn(tv, other) {
t.Error("another instant is the same turn")
}
}
+9 -8
View File
@@ -8,6 +8,7 @@ import (
"encoding/json"
"errors"
"strings"
"time"
)
// Seat is the seat's name.
@@ -433,16 +434,16 @@ type OwedByRepositoryNumberHeadActValue struct {
// TurnsByRepositoryValueWaitingItem is a type of the contract.
type TurnsByRepositoryValueWaitingItem struct {
// when (RFC 3339 time)
Approved string `json:"approved"`
Ask string `json:"ask,omitempty"`
Group string `json:"group"`
Heads string `json:"heads"`
Approved time.Time `json:"approved"`
Ask string `json:"ask,omitempty"`
Group string `json:"group"`
Heads string `json:"heads"`
// when (RFC 3339 time)
Since string `json:"since"`
Step string `json:"step,omitempty"`
WaitsFor string `json:"waits-for,omitempty"`
Since time.Time `json:"since"`
Step string `json:"step,omitempty"`
WaitsFor string `json:"waits-for,omitempty"`
// when (RFC 3339 time)
WaitsSince string `json:"waits-since,omitempty"`
WaitsSince *time.Time `json:"waits-since,omitempty"`
}
// TurnsByRepositoryValue is one value of the state turns under the key <repository>.
+3
View File
@@ -529,10 +529,12 @@
},
"approved": {
"type": "string",
"format": "date-time",
"description": "when (RFC 3339 time)"
},
"since": {
"type": "string",
"format": "date-time",
"description": "when (RFC 3339 time)"
},
"waits-for": {
@@ -540,6 +542,7 @@
},
"waits-since": {
"type": "string",
"format": "date-time",
"description": "when (RFC 3339 time)"
},
"step": {
@@ -0,0 +1,21 @@
package builds
import (
"os/exec"
"strings"
"testing"
)
// The builder's published value is read by modules that never touch the bus (mesh-host's composer tool): nothing it
// imports reaches the bus library.
func TestThePackageReachesNoBus(t *testing.T) {
out, err := exec.Command("go", "list", "-deps", ".").Output()
if err != nil {
t.Fatalf("go list: %v", err)
}
for _, p := range strings.Fields(string(out)) {
if strings.HasPrefix(p, "github.com/nats-io/") {
t.Errorf("builds depends on %s", p)
}
}
}
+52
View File
@@ -0,0 +1,52 @@
package buildrun
import (
"encoding/json"
"sync"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// keys is a steps state in memory: every value put, in order, per key, and a signal for each end.
type keys struct {
mu sync.Mutex
puts map[string][]buildstep.Step
ended chan string
}
func newKeys() *keys { return &keys{puts: map[string][]buildstep.Step{}, ended: make(chan string, 16)} }
func (k *keys) put(key string, value any) (uint64, error) {
step, err := buildstep.DecodeStep(value.(json.RawMessage))
if err != nil {
return 0, err
}
k.mu.Lock()
k.puts[key] = append(k.puts[key], step)
n := uint64(len(k.puts[key]))
k.mu.Unlock()
if step.State != buildstep.StepRunning {
k.ended <- key
}
return n, nil
}
func (k *keys) last(key string) buildstep.Step {
k.mu.Lock()
defer k.mu.Unlock()
s := k.puts[key]
return s[len(s)-1]
}
func (k *keys) waitEnd(t *testing.T) string {
t.Helper()
select {
case key := <-k.ended:
return key
case <-time.After(3 * time.Second):
t.Fatal("the step never ended on its key")
return ""
}
}
@@ -1,10 +1,11 @@
package buildstep
package buildrun
import (
"context"
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"os"
"strconv"
"strings"
@@ -20,7 +21,7 @@ const RestartReason = "the build module restarted while it ran"
// for this process's, which may still run it, so EndOne refuses it.
var Boot = time.Now()
// Owner is this process as a step's owner (Step.Owner): `<boot id>.<pid>.<start>`, the node's boot id
// Owner is this process as a step's owner (buildstep.Step.Owner): `<boot id>.<pid>.<start>`, the node's boot id
// (/proc/sys/kernel/random/boot_id), the pid and Boot in milliseconds, base 36. The start keeps two processes apart
// that a reboot or a pid namespace gave the same pid. A running step whose owner is another is a previous process's.
var Owner = ownerOf(bootID, os.Getpid(), Boot)
@@ -84,15 +85,15 @@ func ScanOf(s stdio.KeptState) Scan {
// has the tool's name). An ask's key is the stream's to settle, never a restart's: the puller leaves an ask it was
// running unacknowledged so the stream redelivers it, and the asker's wait takes the first end it reads on the key.
// A key a gone process owns is deleted, never ended (EndStale); one with no owner is ended only once the asker's bound
// has passed (StepBound), when it is nobody's.
func isAsk(key string, s Step) bool {
_, _, err := ParseAskStepKey(key)
// has passed (buildstep.StepBound), when it is nobody's.
func isAsk(key string, s buildstep.Step) bool {
_, _, err := buildstep.ParseAskStepKey(key)
return err == nil && strings.HasSuffix(s.Tool, " ask")
}
// AskWhat says what an ask is, in one line: its kind, and its repository, head and levels, or its module and
// fingerprint.
func AskWhat(a Ask) string {
func AskWhat(a buildstep.Ask) string {
short := func(s string) string {
if len(s) > 12 {
return s[:12]
@@ -122,9 +123,9 @@ func AskWhat(a Ask) string {
}
// endRunning writes step ended failed with why, only while its key is still at revision.
func endRunning(scan Scan, key string, step Step, revision uint64, why string) error {
step.State, step.Reason, step.Ended, step.Result = StepFailed, why, now().UTC(), nil
raw, err := EncodeStep(step)
func endRunning(scan Scan, key string, step buildstep.Step, revision uint64, why string) error {
step.State, step.Reason, step.Ended, step.Result = buildstep.StepFailed, why, now().UTC(), nil
raw, err := buildstep.EncodeStep(step)
if err != nil {
return err
}
@@ -134,7 +135,7 @@ func endRunning(scan Scan, key string, step Step, revision uint64, why string) e
// EndStale ends this node's running steps that no process of this module runs any more, and answers their ids. Only keys
// of node are read; a value that is no step is left alone.
// - A hand-started step with an owner (Step.Owner) that is not this process (Owner) is ended failed with
// - A hand-started step with an owner (buildstep.Step.Owner) that is not this process (Owner) is ended failed with
// RestartReason at once, whatever its age: its process is gone and nothing redelivers it, so nothing else would
// write its end (issue 1890, note 21).
// - A pulled ask's step with an owner that is not this process is deleted, not ended (novox/hq ADR 0525 §6): the
@@ -146,7 +147,7 @@ func endRunning(scan Scan, key string, step Step, revision uint64, why string) e
// - A step owned by this process is left: it may run now. A redelivered ask begun on this node is owned by it.
// - A step with no owner (written by an older build) keeps the earlier rule: a hand-started step started before
// `before` (the start of this process) is ended; a pulled ask is left to the stream's redelivery (isAsk) until its
// start is older than StepBound, and then ended "past the step bound: its asker stopped waiting at <time>".
// start is older than buildstep.StepBound, and then ended "past the step bound: its asker stopped waiting at <time>".
//
// Each end and delete is logged, naming the step, its owner and why. A key written by another writer since it was read
// (a redelivered ask's start, a real end) is skipped.
@@ -157,7 +158,7 @@ func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any
}
var ended []string
for _, k := range keys {
n, _, perr := ParseStepKey(k)
n, _, perr := buildstep.ParseStepKey(k)
if perr != nil || n != node {
continue
}
@@ -168,8 +169,8 @@ func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any
if len(raw) == 0 {
continue
}
step, err := DecodeStep(raw)
if err != nil || step.State != StepRunning {
step, err := buildstep.DecodeStep(raw)
if err != nil || step.State != buildstep.StepRunning {
continue
}
why, said := staleWhy(k, step, before)
@@ -204,17 +205,17 @@ func EndStale(scan Scan, node string, before time.Time, logf func(string, ...any
// staleWhy is why a running step is ended (or, for an ask with an owner, deleted) by EndStale and what the log says of
// its owner, or "" to leave it.
func staleWhy(key string, step Step, before time.Time) (why, said string) {
func staleWhy(key string, step buildstep.Step, before time.Time) (why, said string) {
switch {
case step.Owner == Owner:
return "", ""
case step.Owner != "":
return RestartReason, "its owner " + step.Owner + " is not this process " + Owner
case isAsk(key, step):
if !step.Started.Add(StepBound).Before(now()) {
if !step.Started.Add(buildstep.StepBound).Before(now()) {
return "", ""
}
return "past the step bound: its asker stopped waiting at " + step.Started.Add(StepBound).UTC().Format("2006-01-02T15:04:05Z"),
return "past the step bound: its asker stopped waiting at " + step.Started.Add(buildstep.StepBound).UTC().Format("2006-01-02T15:04:05Z"),
"it names no owner (an older build wrote it)"
case step.Started.Before(before):
return RestartReason, "it names no owner (an older build wrote it) and started before this process"
@@ -225,45 +226,45 @@ func staleWhy(key string, step Step, before time.Time) (why, said string) {
// EndOne ends this node's running step id (or the key <node>.<id>) as failed with why. A step that is not running is
// refused, and so is one this process owns (Owner; for a step with no owner, one started since Boot: it may still run,
// whether a pulled ask or a step a tool started), one live reports as held, and one whose key changed since it was read.
func EndOne(scan Scan, node, id, why string, live func(id string) bool) (Step, error) {
func EndOne(scan Scan, node, id, why string, live func(id string) bool) (buildstep.Step, error) {
id = strings.TrimSpace(id)
if n, rest, ok := strings.Cut(id, "."); ok {
if n != node {
return Step{}, fmt.Errorf("the step %q is of node %q, this module runs on %q", id, n, node)
return buildstep.Step{}, fmt.Errorf("the step %q is of node %q, this module runs on %q", id, n, node)
}
id = rest
}
if strings.TrimSpace(why) == "" {
return Step{}, fmt.Errorf("say why the step is ended (why)")
return buildstep.Step{}, fmt.Errorf("say why the step is ended (why)")
}
key, err := StepKey(node, id)
key, err := buildstep.StepKey(node, id)
if err != nil {
return Step{}, err
return buildstep.Step{}, err
}
raw, rev, err := scan.Get(key)
if err != nil {
return Step{}, err
return buildstep.Step{}, err
}
if len(raw) == 0 {
return Step{}, fmt.Errorf("no step %s in the steps state", key)
return buildstep.Step{}, fmt.Errorf("no step %s in the steps state", key)
}
step, err := DecodeStep(raw)
step, err := buildstep.DecodeStep(raw)
if err != nil {
return Step{}, err
return buildstep.Step{}, err
}
if step.State != StepRunning {
return Step{}, fmt.Errorf("the step %s is %s, not running", key, step.State)
if step.State != buildstep.StepRunning {
return buildstep.Step{}, fmt.Errorf("the step %s is %s, not running", key, step.State)
}
if step.Owner == Owner || (step.Owner == "" && !step.Started.Before(Boot)) || (live != nil && live(id)) {
return Step{}, fmt.Errorf("the step %s was started by this process and may run now; only a step a previous process left is ended", key)
return buildstep.Step{}, fmt.Errorf("the step %s was started by this process and may run now; only a step a previous process left is ended", key)
}
if err := endRunning(scan, key, step, rev, why); err != nil {
if errors.Is(err, stdio.ErrChanged) {
return Step{}, fmt.Errorf("the step %s changed while it was being ended; read it again", key)
return buildstep.Step{}, fmt.Errorf("the step %s changed while it was being ended; read it again", key)
}
return Step{}, err
return buildstep.Step{}, err
}
step.State, step.Reason = StepFailed, why
step.State, step.Reason = buildstep.StepFailed, why
return step, nil
}
@@ -316,7 +317,7 @@ func EndStaleAtStart(ctx context.Context, scan Scan, node string, before time.Ti
// EndStaleRound is EndStale as the module's periodic round runs it: at once, and only when the runtime serves (a round
// before that is skipped, the start's EndStaleAtStart and the next round cover it). It ends the steps a previous
// process left and the asks past StepBound.
// process left and the asks past buildstep.StepBound.
func EndStaleRound(scan Scan, node string, logf func(string, ...any)) {
select {
case <-stdio.Served():
@@ -327,3 +328,6 @@ func EndStaleRound(scan Scan, node string, logf func(string, ...any)) {
logf("ending the steps left running: %v", err)
}
}
// now is the clock; a test sets it.
var now = time.Now
@@ -1,8 +1,9 @@
package buildstep
package buildrun
import (
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"strings"
"testing"
"time"
@@ -30,7 +31,7 @@ func scanOf(k *keys, race func(key string)) Scan {
if !ok {
return nil, 0, nil
}
raw, err := EncodeStep(s[len(s)-1])
raw, err := buildstep.EncodeStep(s[len(s)-1])
return raw, uint64(len(s)), err
},
Delete: func(key string, revision uint64) error {
@@ -70,25 +71,25 @@ func TestEndStaleAndEndOne(t *testing.T) {
defer func() { Boot = oldBoot }()
gone := ownerOf("b00t", 4242, boot.Add(-time.Hour))
put := func(node, id, tool, state string, started time.Time, owner string) {
key, _ := StepKey(node, id)
_, _ = k.put(key, json.RawMessage(mustStep(t, Step{ID: id, Tool: tool, Node: node, State: state, Started: started, Owner: owner})))
key, _ := buildstep.StepKey(node, id)
_, _ = k.put(key, json.RawMessage(mustStep(t, buildstep.Step{ID: id, Tool: tool, Node: node, State: state, Started: started, Owner: owner})))
}
put("ace", "old1", "build_go_fetch", StepRunning, boot.Add(-time.Hour), "")
put("ace", "mjk3l9x0abcdef", "check ask", StepRunning, boot.Add(-time.Hour), "")
put("ace", "mjk3l9x0abcdee", "check ask", StepRunning, time.Now().Add(-5*time.Minute), "")
put("ace", "mjk3l9x0abcded", "check ask", StepRunning, time.Now().Add(-5*time.Minute), gone) // note 21's ghost
put("ace", "gone2", "build_go_test", StepRunning, boot.Add(time.Second), gone)
put("ace", "mine1", "check ask", StepRunning, boot.Add(-time.Hour), Owner)
put("ace", "raced1", "build_go_test", StepRunning, boot.Add(-time.Hour), "")
put("ace", "new1", "build_go_fetch", StepRunning, boot.Add(time.Second), "")
put("ace", "done1", "build_go_fetch", StepPassed, boot.Add(-time.Hour), gone)
put("g14", "other1", "build_go_fetch", StepRunning, boot.Add(-time.Hour), gone)
put("ace", "old1", "build_go_fetch", buildstep.StepRunning, boot.Add(-time.Hour), "")
put("ace", "mjk3l9x0abcdef", "check ask", buildstep.StepRunning, boot.Add(-time.Hour), "")
put("ace", "mjk3l9x0abcdee", "check ask", buildstep.StepRunning, time.Now().Add(-5*time.Minute), "")
put("ace", "mjk3l9x0abcded", "check ask", buildstep.StepRunning, time.Now().Add(-5*time.Minute), gone) // note 21's ghost
put("ace", "gone2", "build_go_test", buildstep.StepRunning, boot.Add(time.Second), gone)
put("ace", "mine1", "check ask", buildstep.StepRunning, boot.Add(-time.Hour), Owner)
put("ace", "raced1", "build_go_test", buildstep.StepRunning, boot.Add(-time.Hour), "")
put("ace", "new1", "build_go_fetch", buildstep.StepRunning, boot.Add(time.Second), "")
put("ace", "done1", "build_go_fetch", buildstep.StepPassed, boot.Add(-time.Hour), gone)
put("g14", "other1", "build_go_fetch", buildstep.StepRunning, boot.Add(-time.Hour), gone)
// raced1 is ended by its own run between the read and the write.
raced := false
scan := scanOf(k, func(key string) {
if key == "ace.raced1" && !raced {
raced = true
put("ace", "raced1", "build_go_test", StepPassed, boot.Add(-time.Hour), "")
put("ace", "raced1", "build_go_test", buildstep.StepPassed, boot.Add(-time.Hour), "")
}
})
var said []string
@@ -100,18 +101,18 @@ func TestEndStaleAndEndOne(t *testing.T) {
t.Errorf("the ghost ask's key is still held: %+v", k.last("ace.mjk3l9x0abcded"))
}
for _, key := range []string{"ace.old1", "ace.gone2"} {
if s := k.last(key); s.State != StepFailed || s.Reason != RestartReason || s.Ended.IsZero() {
if s := k.last(key); s.State != buildstep.StepFailed || s.Reason != RestartReason || s.Ended.IsZero() {
t.Errorf("%s = %+v", key, s)
}
}
if s := k.last("ace.mjk3l9x0abcdef"); s.State != StepFailed || !strings.HasPrefix(s.Reason, "past the step bound: its asker stopped waiting at ") {
if s := k.last("ace.mjk3l9x0abcdef"); s.State != buildstep.StepFailed || !strings.HasPrefix(s.Reason, "past the step bound: its asker stopped waiting at ") {
t.Errorf("the old ask = %+v", s)
}
if log := strings.Join(said, "\n"); !strings.Contains(log, "ace.mjk3l9x0abcded") || !strings.Contains(log, "deleted, left to the stream's redelivery") || !strings.Contains(log, "its owner "+gone+" is not this process "+Owner) {
t.Errorf("the log does not name the ghost and its owner:\n%s", log)
}
for key, state := range map[string]string{"ace.mjk3l9x0abcdee": StepRunning, "ace.mine1": StepRunning, "ace.raced1": StepPassed,
"ace.new1": StepRunning, "ace.done1": StepPassed, "g14.other1": StepRunning} {
for key, state := range map[string]string{"ace.mjk3l9x0abcdee": buildstep.StepRunning, "ace.mine1": buildstep.StepRunning, "ace.raced1": buildstep.StepPassed,
"ace.new1": buildstep.StepRunning, "ace.done1": buildstep.StepPassed, "g14.other1": buildstep.StepRunning} {
if s := k.last(key); s.State != state {
t.Errorf("%s = %s, want %s", key, s.State, state)
}
@@ -123,14 +124,14 @@ func TestEndStaleAndEndOne(t *testing.T) {
t.Errorf("ended %s with why %q", c.id, c.why)
}
}
put("ace", "old2", "build_go_fetch", StepRunning, boot.Add(time.Second), gone)
put("ace", "old2", "build_go_fetch", buildstep.StepRunning, boot.Add(time.Second), gone)
if _, err := EndOne(scan, "ace", "old2", "x", func(string) bool { return true }); err == nil {
t.Error("ended a held step")
}
if _, err := EndOne(scan, "ace", "ace.old2", "stuck for good", nil); err != nil {
t.Fatal(err)
}
if s := k.last("ace.old2"); s.State != StepFailed || s.Reason != "stuck for good" || s.Owner != gone {
if s := k.last("ace.old2"); s.State != buildstep.StepFailed || s.Reason != "stuck for good" || s.Owner != gone {
t.Errorf("old2 = %+v", s)
}
if a, b := ownerOf("b", 1, boot), ownerOf("b", 1, boot.Add(time.Second)); a == b {
@@ -138,9 +139,9 @@ func TestEndStaleAndEndOne(t *testing.T) {
}
}
func mustStep(t *testing.T, s Step) []byte {
func mustStep(t *testing.T, s buildstep.Step) []byte {
t.Helper()
raw, err := EncodeStep(s)
raw, err := buildstep.EncodeStep(s)
if err != nil {
t.Fatal(err)
}
@@ -149,8 +150,8 @@ func mustStep(t *testing.T, s Step) []byte {
// 1890: AskWhat names the repository, head and levels of a check ask and the module of a build ask.
func TestAskWhat(t *testing.T) {
var check, build Ask
for raw, into := range map[string]*Ask{
var check, build buildstep.Ask
for raw, into := range map[string]*buildstep.Ask{
`{"kind":"check","check":{"fetch":{"repository":"novox/mesh-sdk","commit":"0123456789abcdef0123"},"levels":["unit","integration"]}}`: &check,
`{"kind":"delivery","build":{"module":"mesh-x","fp":"abcdef0123456789"}}`: &build,
} {
+622
View File
@@ -0,0 +1,622 @@
package buildstep
import (
"encoding/json"
"fmt"
"path"
"slices"
"sort"
"strings"
"time"
meshkeys "git.novox.be/novox/mesh-sdk/go/keys"
)
// **The build ask, once.** The three build seats (node-build-go, node-build-image, node-build-node) accept the same ask,
// defined once as the part `build-step` of the Delivery contracts. The generator writes one Go type per seat, so the
// seat packages hold three structurally equal StepAsks; this file is the one neutral form the shared runner and the
// askers use, and AskOf is the one conversion from any seat's (a seat's value and this form are the same JSON, held
// equal by the part's fixtures: the contracts' tests decode each in both). This package imports no seat's contract
// (novox/hq ADR 0634 §6).
// The kinds of ask: the seat's accepted kinds, in the order a holder pulls them.
const (
AskDelivery = "delivery"
AskCheck = "check"
AskRecheck = "recheck"
)
// AskKinds are the accepted kinds of a build seat, in the order a holder pulls them (ADR 0536 §4).
var AskKinds = []string{AskDelivery, AskCheck, AskRecheck}
// The languages of the build seats: each is one build module's (mesh-build-<language>) and one seat's
// (node-build-<language>).
const (
AskGo = "go"
AskNode = "node"
AskImage = "image"
)
// AskLanguages are the build seats' languages.
var AskLanguages = []string{AskGo, AskNode, AskImage}
// NeedNodeBuildNode is the need of a check whose tests run through the Node build seat's holder on the same node
// (issue 890): a node that holds no Node build module declines the ask.
const NeedNodeBuildNode = "node-build-node"
// AskNeeds are the needs an ask may name.
var AskNeeds = []string{NeedNodeBuildNode}
// CheckNeed refuses a need an ask may not name.
func CheckNeed(need string) error {
if !slices.Contains(AskNeeds, need) {
return fmt.Errorf("buildstep: %q is not a need an ask may name (%s)", need, strings.Join(AskNeeds, ", "))
}
return nil
}
// Lacks are the needs has does not say its node has; a nil has lacks all.
func Lacks(needs []string, has func(need string) bool) []string {
var out []string
for _, n := range needs {
if has == nil || !has(n) {
out = append(out, n)
}
}
return out
}
// AskRuns is how many times a holder runs one ask before it ends refused; AskMaxDeliver is the redelivery bound that
// leaves one more for the terminal step (ADR 0525 §5).
const (
AskRuns = 3
AskMaxDeliver = AskRuns + 2
)
// StepAskVersion is the layout version of an ask.
const StepAskVersion = 1
// The levels a check ask names.
const (
LevelUnit = "unit"
LevelIntegration = "integration"
)
// The registry seats a build publishes through (ADR 0605 §1).
var RegistrySeats = []string{"npm-package-registry", "go-package-registry", "container-image-registry", "oci-artifact-registry"}
// NodeNamesArg and NodeNamesEnv carry the node names a check's tests run against.
const (
NodeNamesArg = "node-names"
NodeNamesEnv = "MESH_NODE_NAMES"
)
func checkAskKind(kind string) error {
if !slices.Contains(AskKinds, kind) {
return fmt.Errorf("buildstep: the ask kind %q is not delivery, check or recheck", kind)
}
return nil
}
func checkAskLanguage(language string) error {
if !slices.Contains(AskLanguages, language) {
return fmt.Errorf("buildstep: the ask language %q is not go, node or image", language)
}
return nil
}
// BuildSeat is the seat of a language's build module.
func BuildSeat(language string) (string, error) {
if err := checkAskLanguage(language); err != nil {
return "", err
}
return "node-build-" + language, nil
}
// BuildModuleOf is the module of a language's build seat.
func BuildModuleOf(language string) (string, error) {
if err := checkAskLanguage(language); err != nil {
return "", err
}
return "mesh-build-" + language, nil
}
// AskSubject is the subject a build seat accepts a kind of ask on: the seat's accept subject.
func AskSubject(language, kind string) (string, error) {
seat, err := BuildSeat(language)
if err != nil {
return "", err
}
if err := checkAskKind(kind); err != nil {
return "", err
}
return "mesh.seat." + seat + ".accept." + kind, nil
}
// ParseAskSubject is the inverse of AskSubject.
func ParseAskSubject(subject string) (language, kind string, err error) {
rest, ok := strings.CutPrefix(subject, "mesh.seat.node-build-")
if !ok {
return "", "", fmt.Errorf("buildstep: %q is not a build seat's accept subject", subject)
}
language, kind, ok = strings.Cut(rest, ".accept.")
if !ok {
return "", "", fmt.Errorf("buildstep: %q is not a build seat's accept subject", subject)
}
if want, err := AskSubject(language, kind); err != nil || want != subject {
return "", "", fmt.Errorf("buildstep: %q is not a build seat's accept subject: %v", subject, err)
}
return language, kind, nil
}
// AskStream is the work-queue stream of a language's build seat.
func AskStream(language string) (string, error) {
seat, err := BuildSeat(language)
if err != nil {
return "", err
}
return "SEAT_" + strings.ToUpper(strings.ReplaceAll(seat, "-", "_")), nil
}
// AskWorker is the shared pull worker of a kind on a language's stream.
func AskWorker(language, kind string) (string, error) {
stream, err := AskStream(language)
if err != nil {
return "", err
}
if err := checkAskKind(kind); err != nil {
return "", err
}
return stream + "_" + strings.ToUpper(kind) + "_worker", nil
}
// AskStepKey is the key of the step of the ask askID on node.
func AskStepKey(node, askID string) (string, error) {
if err := CheckAskID(askID); err != nil {
return "", err
}
return StepKey(node, askID)
}
// AskStepWatch is the key pattern of an ask's step on any node.
func AskStepWatch(askID string) (string, error) {
if err := CheckAskID(askID); err != nil {
return "", err
}
return "*." + askID, nil
}
// StepAsk is one ask of a build module: the part's `StepAsk`.
type StepAsk struct {
V int `json:"v"`
ID string `json:"id"`
Kind string `json:"kind"`
Language string `json:"language"`
Asker string `json:"asker"`
At time.Time `json:"at,omitzero"`
Build *BuildJob `json:"build,omitempty"`
Check *CheckJob `json:"check,omitempty"`
Compose *ComposeJob `json:"compose,omitempty"`
// Needs are what a check's tests need of the holder's node (CheckNeed); a holder lacking one declines the ask.
Needs []string `json:"needs,omitempty"`
// Packages pin the mesh packages a build or a check builds against.
Packages PackageVersions `json:"packages,omitempty"`
}
// Fetch is what a build module's fetch tool is asked, inside an ask: the part's `Fetch`.
type Fetch struct {
V int `json:"v,omitempty"`
Repository string `json:"repository"`
Commit string `json:"commit"`
Ref string `json:"ref,omitempty"`
Base string `json:"base,omitempty"`
Dir string `json:"dir,omitempty"`
Paths []string `json:"paths,omitempty"`
Packages PackageVersions `json:"packages,omitempty"`
Beside []FetchBeside `json:"beside,omitempty"`
}
// FetchBeside is a sibling repository placed beside the fetched tree.
type FetchBeside struct {
Repository string `json:"repository"`
Commit string `json:"commit"`
}
// ToolchainUsed is the toolchain a build ran with.
type ToolchainUsed struct {
Language string `json:"language"`
Recipe string `json:"recipe"`
Version string `json:"version"`
}
// AskArtifact is one artifact of a build ask: built from the fetched tree, or copied from an upstream pin (From).
type AskArtifact struct {
Artifact string `json:"artifact"`
From string `json:"from,omitempty"`
Seat string `json:"seat,omitempty"`
Version string `json:"version,omitempty"`
}
// BuildJob is a trunk fingerprint's build.
type BuildJob struct {
Module string `json:"module"`
Path string `json:"path,omitempty"`
Fp string `json:"fp,omitempty"`
Fetch *Fetch `json:"fetch,omitempty"`
Artifacts []AskArtifact `json:"artifacts"`
Publish bool `json:"publish,omitempty"`
Toolchain *ToolchainUsed `json:"toolchain,omitempty"`
}
// CheckJob is the check of one or both levels of a fetched tree.
type CheckJob struct {
Fetch Fetch `json:"fetch"`
Levels []string `json:"levels"`
Kept bool `json:"kept,omitempty"`
Runner string `json:"runner,omitempty"`
RunnerWhen string `json:"runner_when,omitempty"`
Run string `json:"run,omitempty"`
Packages []string `json:"packages,omitempty"`
NodeNames []string `json:"node-names,omitempty"`
}
// ComposeJob asks one compose-candidate build.
type ComposeJob struct {
Fetch Fetch `json:"fetch"`
Dir string `json:"dir"`
Package string `json:"package"`
Request json.RawMessage `json:"request"`
Reader string `json:"reader,omitempty"`
}
// AskOf is any seat's ask (or this form) as the neutral StepAsk: the one conversion, since the three seats' types
// are the part's one layout. A value that does not fit the layout is refused.
func AskOf[T any](seatAsk T) (StepAsk, error) {
raw, err := json.Marshal(seatAsk)
if err != nil {
return StepAsk{}, fmt.Errorf("buildstep: an ask does not encode: %w", err)
}
return DecodeStepAsk(raw)
}
// AskAs is the neutral ask as a seat's type T (the inverse of AskOf).
func AskAs[T any](a StepAsk) (T, error) {
var out T
raw, err := json.Marshal(a)
if err != nil {
return out, err
}
err = json.Unmarshal(raw, &out)
return out, err
}
func (a StepAsk) check() error {
if err := CheckAskID(a.ID); err != nil {
return err
}
if err := checkAskKind(a.Kind); err != nil {
return err
}
if err := checkAskLanguage(a.Language); err != nil {
return err
}
if err := meshkeys.CheckName("asker", a.Asker); err != nil {
return err
}
if a.At.IsZero() {
return fmt.Errorf("buildstep: the ask %s names no time", a.ID)
}
jobs := 0
for _, set := range []bool{a.Build != nil, a.Check != nil, a.Compose != nil} {
if set {
jobs++
}
}
switch {
case jobs != 1:
return fmt.Errorf("buildstep: the ask %s holds a build, a check or a compose, exactly one", a.ID)
case a.Compose != nil && (a.Kind != AskCheck || a.Language != AskGo):
return fmt.Errorf("buildstep: the ask %s is a compose, which the Go seat's check stream alone carries", a.ID)
case a.Build != nil && a.Kind == AskRecheck:
return fmt.Errorf("buildstep: the ask %s is a build on the recheck stream, which carries checks alone", a.ID)
case a.Check != nil && a.Kind == AskDelivery:
return fmt.Errorf("buildstep: the ask %s is a check on the delivery stream, which carries trunk builds alone", a.ID)
case a.Check != nil && a.Language == AskImage:
return fmt.Errorf("buildstep: the ask %s is a check of an image, which has no tests", a.ID)
case len(a.Needs) > 0 && a.Check == nil:
return fmt.Errorf("buildstep: the ask %s names needs and is no check, whose tests alone need a node", a.ID)
case len(a.Packages) > 0 && a.Compose != nil:
return fmt.Errorf("buildstep: the ask %s is a compose and names package versions, which a compose does not take", a.ID)
}
if err := a.Packages.Check(); err != nil {
return fmt.Errorf("the ask %s: %w", a.ID, err)
}
for _, n := range a.Needs {
if err := CheckNeed(n); err != nil {
return fmt.Errorf("the ask %s: %w", a.ID, err)
}
}
switch {
case a.Build != nil:
return a.Build.check(a.ID)
case a.Compose != nil:
return a.Compose.check(a.ID)
}
return a.Check.check(a.ID)
}
func (c ComposeJob) check(id string) error {
if err := c.Fetch.check(id); err != nil {
return err
}
if c.Dir == "" || c.Dir != path.Clean(c.Dir) || strings.HasPrefix(c.Dir, "/") || c.Dir == ".." || strings.HasPrefix(c.Dir, "../") {
return fmt.Errorf("buildstep: the compose ask %s names no clean directory inside the repository (%q)", id, c.Dir)
}
if c.Package == "" {
return fmt.Errorf("buildstep: the compose ask %s names no package to build", id)
}
if !json.Valid(c.Request) || len(c.Request) == 0 || c.Request[0] != '{' {
return fmt.Errorf("buildstep: the compose ask %s holds no request object", id)
}
return nil
}
func (b BuildJob) check(id string) error {
if err := meshkeys.CheckName("module", b.Module); err != nil {
return err
}
if len(b.Artifacts) == 0 {
return fmt.Errorf("buildstep: the build ask %s names no artifact", id)
}
fetch := false
for _, a := range b.Artifacts {
if err := meshkeys.CheckName("artifact", a.Artifact); err != nil {
return err
}
if a.Seat != "" && !slices.Contains(RegistrySeats, a.Seat) {
return fmt.Errorf("buildstep: the build ask %s publishes %s through %q, which is no registry seat (%s)", id,
a.Artifact, a.Seat, strings.Join(RegistrySeats, ", "))
}
if a.Version != "" && a.Seat == "" {
return fmt.Errorf("buildstep: the build ask %s gives %s the version %s and names no seat to publish it through",
id, a.Artifact, a.Version)
}
fetch = fetch || a.From == ""
}
if fetch {
if b.Path == "" {
return fmt.Errorf("buildstep: the build ask %s names no directory (path) for the module %s", id, b.Module)
}
if b.Fetch == nil {
return fmt.Errorf("buildstep: the ask %s fetches no repository <owner>/<name>", id)
}
return b.Fetch.check(id)
}
return nil
}
func (c CheckJob) check(id string) error {
if len(c.Levels) == 0 {
return fmt.Errorf("buildstep: the check ask %s names no level", id)
}
for _, l := range c.Levels {
if l != LevelUnit && l != LevelIntegration {
return fmt.Errorf("buildstep: the level %q is not %s or %s", l, LevelUnit, LevelIntegration)
}
}
for _, p := range c.Packages {
if p == "" || strings.ContainsAny(p, " \t\n") {
return fmt.Errorf("buildstep: the check ask %s names a package %q that is no import path", id, p)
}
}
seen := map[string]bool{}
for _, n := range c.NodeNames {
if err := meshkeys.CheckName("node", n); err != nil || strings.Contains(n, ",") || seen[n] {
return fmt.Errorf("buildstep: the check ask %s names the node %q, which is no node name or named twice", id, n)
}
seen[n] = true
}
return c.Fetch.check(id)
}
func (f Fetch) check(id string) error {
if f.Repository == "" || !strings.Contains(f.Repository, "/") {
return fmt.Errorf("buildstep: the ask %s fetches no repository <owner>/<name>", id)
}
if !askCommitOK.MatchString(f.Commit) {
return fmt.Errorf("buildstep: the ask %s fetches no full commit", id)
}
return nil
}
// EncodeStepAsk is a as the bytes the asker puts on a build seat's stream: stamped with the version and UTC, checked.
func EncodeStepAsk(a StepAsk) ([]byte, error) {
a.V = StepAskVersion
a.At = a.At.UTC()
if err := a.check(); err != nil {
return nil, err
}
return json.Marshal(a)
}
// DecodeStepAsk reads an ask, refusing another version, an unknown shape or a malformed ask.
func DecodeStepAsk(data []byte) (StepAsk, error) {
var probe struct {
V *int `json:"v"`
}
if err := json.Unmarshal(data, &probe); err != nil {
return StepAsk{}, fmt.Errorf("buildstep: an ask is not JSON: %w", err)
}
if probe.V == nil || *probe.V != StepAskVersion {
return StepAsk{}, fmt.Errorf("%w: an ask of version %v, this reader reads %d", ErrUnknownVersion, probe.V, StepAskVersion)
}
var a StepAsk
if err := json.Unmarshal(data, &a); err != nil {
return StepAsk{}, fmt.Errorf("buildstep: an ask does not fit its shape: %w", err)
}
if err := a.check(); err != nil {
return StepAsk{}, err
}
return a, nil
}
// PackageNeeds is what a package of a check needs of a node.
type PackageNeeds struct {
Package string `json:"package"`
Needs []string `json:"needs"`
}
// StepDeclined is the result of a step a holder declined: its node lacks what the ask needs.
type StepDeclined struct {
Declined bool `json:"declined"`
Needs []string `json:"needs"`
Lacks []string `json:"lacks"`
}
// DeclinedOf reads a refused step's result as a decline.
func DeclinedOf(step Step) (StepDeclined, bool) {
if step.State != StepRefused || len(step.Result) == 0 {
return StepDeclined{}, false
}
var d StepDeclined
if err := json.Unmarshal(step.Result, &d); err != nil || !d.Declined {
return StepDeclined{}, false
}
return d, true
}
// UnmetNeeds are the needs the levels' unmet packages name, sorted, once each.
func UnmetNeeds(levels []LevelAnswer) []string {
seen := map[string]bool{}
var out []string
for _, l := range levels {
for _, u := range l.Unmet {
for _, n := range u.Needs {
if !seen[n] {
seen[n] = true
out = append(out, n)
}
}
}
}
sort.Strings(out)
return out
}
// BuildAnswer is what a build ask's step answers.
type BuildAnswer struct {
Artifacts []AskBuilt `json:"artifacts"`
Toolchain *ToolchainUsed `json:"toolchain,omitempty"`
Imports []string `json:"imports,omitempty"`
OutOfStep []string `json:"out-of-step,omitempty"`
}
// AskBuilt is one artifact a build made.
type AskBuilt struct {
Artifact string `json:"artifact"`
Digest string `json:"digest"`
Reference string `json:"reference,omitempty"`
}
// CheckAnswer is what a check ask's step answers.
type CheckAnswer struct {
Fetch Fetch `json:"fetch"`
Levels []LevelAnswer `json:"levels,omitempty"`
}
// ComposeAnswer is what a compose ask's step answers.
type ComposeAnswer struct {
Built bool `json:"built"`
Key string `json:"key,omitempty"`
Verdict json.RawMessage `json:"verdict,omitempty"`
}
// LevelAnswer is one level of a check.
type LevelAnswer struct {
Level string `json:"level"`
Judged string `json:"judged,omitempty"`
Digest *LevelDigest `json:"digest,omitempty"`
Kept []KeptTaken `json:"kept,omitempty"`
Tests []json.RawMessage `json:"tests,omitempty"`
Unmet []PackageNeeds `json:"unmet,omitempty"`
}
// LevelDigest is the digest of a level's tests by package.
type LevelDigest struct {
Level string `json:"level,omitempty"`
Packages []PackageDigest `json:"packages,omitempty"`
Beside []string `json:"beside,omitempty"`
WithoutTests []string `json:"without_tests,omitempty"`
JudgedBy string `json:"judged_by,omitempty"`
Needs []PackageNeeds `json:"needs,omitempty"`
}
// PackageDigest is the digest of one package's tests.
type PackageDigest struct {
Package string `json:"package"`
Digest string `json:"digest"`
}
// KeptTaken is a package whose kept verdict a level took instead of running its tests.
type KeptTaken struct {
Package string `json:"package"`
Digest string `json:"digest"`
Tests int `json:"tests"`
}
// UnmarshalJSON reads the packages as the list or as the older map by package.
func (d *LevelDigest) UnmarshalJSON(raw []byte) error {
var v struct {
Level string `json:"level"`
Packages json.RawMessage `json:"packages"`
Beside []string `json:"beside"`
WithoutTests []string `json:"without_tests"`
JudgedBy string `json:"judged_by"`
Needs []PackageNeeds `json:"needs"`
}
if err := json.Unmarshal(raw, &v); err != nil {
return err
}
var pkgs []PackageDigest
if p := strings.TrimSpace(string(v.Packages)); p != "" && p != "null" {
if strings.HasPrefix(p, "{") {
var m map[string]string
if err := json.Unmarshal(v.Packages, &m); err != nil {
return fmt.Errorf("a digest's packages: %w", err)
}
for pkg, digest := range m {
pkgs = append(pkgs, PackageDigest{Package: pkg, Digest: digest})
}
} else if err := json.Unmarshal(v.Packages, &pkgs); err != nil {
return fmt.Errorf("a digest's packages: %w", err)
}
}
sort.Slice(pkgs, func(i, j int) bool { return pkgs[i].Package < pkgs[j].Package })
*d = LevelDigest{Level: v.Level, Packages: pkgs, Beside: v.Beside, WithoutTests: v.WithoutTests, JudgedBy: v.JudgedBy, Needs: v.Needs}
return nil
}
// NeedsOf are the needs of a package in the digest.
func (d *LevelDigest) NeedsOf(pkg string) []string {
if d == nil {
return nil
}
for _, n := range d.Needs {
if n.Package == pkg {
return n.Needs
}
}
return nil
}
// Map is the digest as digest by package.
func (d *LevelDigest) Map() map[string]string {
if d == nil || len(d.Packages) == 0 {
return nil
}
m := make(map[string]string, len(d.Packages))
for _, p := range d.Packages {
m[p.Package] = p.Digest
}
return m
}
+261
View File
@@ -0,0 +1,261 @@
package buildstep
import (
"bytes"
"encoding/json"
"errors"
"os"
"path/filepath"
"strings"
"testing"
"time"
)
const (
testCommit = "0123456789abcdef0123456789abcdef01234567"
testAskID = "m2kx9a1b3c4d5e"
)
func goodBuild() StepAsk {
return StepAsk{ID: testAskID, Kind: AskDelivery, Language: AskGo, Asker: "mesh-delivery-builder",
At: time.Date(2026, 11, 5, 10, 0, 0, 0, time.UTC),
Build: &BuildJob{Module: "nats", Path: "modules/nats", Fetch: &Fetch{Repository: "novox/mesh-catalog", Commit: testCommit},
Artifacts: []AskArtifact{{Artifact: "go"}}}}
}
func goodCheck() StepAsk {
return StepAsk{ID: testAskID, Kind: AskCheck, Language: AskGo, Asker: "mesh-delivery-checker",
At: time.Date(2026, 11, 5, 10, 0, 0, 0, time.UTC),
Check: &CheckJob{Fetch: Fetch{Repository: "novox/mesh-catalog", Commit: testCommit}, Levels: []string{LevelUnit}}}
}
// The ask the three build seats share is one neutral form: it round-trips, and every ill-formed ask is refused,
// naming what is wrong (the rules of the old delivery.StepAsk, now the shared runner's).
func TestAskCheckAndRoundTrip(t *testing.T) {
for _, a := range []StepAsk{goodBuild(), goodCheck()} {
raw, err := EncodeStepAsk(a)
if err != nil {
t.Fatal(err)
}
back, err := DecodeStepAsk(raw)
if err != nil || back.ID != a.ID || back.V != StepAskVersion {
t.Fatalf("round trip: %+v, %v", back, err)
}
}
for _, c := range []struct {
name string
edit func(*StepAsk)
want string
}{
{"no id", func(a *StepAsk) { a.ID = "x" }, "not an ask id"},
{"unknown kind", func(a *StepAsk) { a.Kind = "build" }, "ask kind"},
{"unknown language", func(a *StepAsk) { a.Language = "rust" }, "ask language"},
{"no time", func(a *StepAsk) { a.At = time.Time{} }, "names no time"},
{"recheck build", func(a *StepAsk) { a.Kind = AskRecheck }, "recheck stream"},
{"two jobs", func(a *StepAsk) { a.Check = goodCheck().Check }, "exactly one"},
{"needs on a build", func(a *StepAsk) { a.Needs = []string{NeedNodeBuildNode} }, "names needs"},
{"unknown registry seat", func(a *StepAsk) { a.Build.Artifacts[0].Seat = "somewhere" }, "no registry seat"},
{"version without seat", func(a *StepAsk) { a.Build.Artifacts[0].Version = "1.0.0" }, "names no seat"},
{"no path", func(a *StepAsk) { a.Build.Path = "" }, "names no directory"},
{"short commit", func(a *StepAsk) { a.Build.Fetch.Commit = "abc" }, "full commit"},
{"bad package pin", func(a *StepAsk) { a.Packages = PackageVersions{"m": "main"} }, "neither a full commit"},
} {
a := goodBuild()
c.edit(&a)
if _, err := EncodeStepAsk(a); err == nil || !strings.Contains(err.Error(), c.want) {
t.Errorf("%s: %v, want %q", c.name, err, c.want)
}
}
chk := goodCheck()
chk.Check.Levels = []string{"nightly"}
if _, err := EncodeStepAsk(chk); err == nil || !strings.Contains(err.Error(), "nightly") {
t.Errorf("a level: %v", err)
}
chk = goodCheck()
chk.Language = AskImage
if _, err := EncodeStepAsk(chk); err == nil || !strings.Contains(err.Error(), "no tests") {
t.Errorf("an image check: %v", err)
}
if _, err := DecodeStepAsk([]byte(`{"v":2}`)); !errors.Is(err, ErrUnknownVersion) {
t.Errorf("another version: %v", err)
}
}
// The seat's value and the neutral form are the same JSON: AskOf takes any seat's type (here its shape, with a
// pointer time), AskAs gives one back.
func TestAskOfAnySeatsType(t *testing.T) {
type seatAsk struct {
V int64 `json:"v,omitempty"`
ID string `json:"id"`
Kind string `json:"kind"`
Lang string `json:"language"`
Asker string `json:"asker"`
At *time.Time `json:"at,omitempty"`
Check *struct {
Fetch Fetch `json:"fetch"`
Levels []string `json:"levels"`
} `json:"check,omitempty"`
}
at := time.Date(2026, 11, 5, 10, 0, 0, 0, time.UTC)
s := seatAsk{V: 1, ID: testAskID, Kind: AskCheck, Lang: AskGo, Asker: "mesh-delivery-checker", At: &at}
s.Check = &struct {
Fetch Fetch `json:"fetch"`
Levels []string `json:"levels"`
}{Fetch{Repository: "novox/mesh-catalog", Commit: testCommit}, []string{LevelUnit}}
a, err := AskOf(s)
if err != nil || a.Check == nil || !a.At.Equal(at) || a.Check.Fetch.Commit != testCommit {
t.Fatalf("%+v %v", a, err)
}
back, err := AskAs[seatAsk](a)
if err != nil || back.ID != s.ID || !back.At.Equal(at) {
t.Fatalf("%+v %v", back, err)
}
s.Check = nil
if _, err := AskOf(s); err == nil {
t.Error("an ask with no job is accepted")
}
}
func TestAskSubjectsStreamsAndKeys(t *testing.T) {
for _, l := range AskLanguages {
for _, k := range AskKinds {
subj, err := AskSubject(l, k)
if err != nil {
t.Fatal(err)
}
gl, gk, err := ParseAskSubject(subj)
if err != nil || gl != l || gk != k {
t.Errorf("%s: %s %s %v", subj, gl, gk, err)
}
}
}
if got, _ := AskWorker(AskGo, AskCheck); got != "SEAT_NODE_BUILD_GO_CHECK_worker" {
t.Errorf("worker %q", got)
}
if got, _ := AskStream(AskImage); got != "SEAT_NODE_BUILD_IMAGE" {
t.Errorf("stream %q", got)
}
if _, _, err := ParseAskSubject("mesh.seat.node-build-go.accept.nothing"); err == nil {
t.Error("an unknown kind parses")
}
if _, err := AskSubject("rust", AskCheck); err == nil {
t.Error("an unknown language has a subject")
}
if k, err := AskStepKey("ace", testAskID); err != nil || k != "ace."+testAskID {
t.Errorf("%q %v", k, err)
}
if w, _ := AskStepWatch(testAskID); w != "*."+testAskID {
t.Errorf("watch %q", w)
}
if k, _ := PauseKey("ace"); k != "paused.ace" {
t.Errorf("pause key %q", k)
}
if k, _ := CapacityKey("ace"); k != "capacity.ace" {
t.Errorf("capacity key %q", k)
}
if _, err := CapacityKey("a.b"); err == nil {
t.Error("a dotted node keys")
}
}
func TestDeclinesNeedsAndLevelDigest(t *testing.T) {
if got := Lacks([]string{NeedNodeBuildNode}, nil); len(got) != 1 {
t.Errorf("a nil has lacks all: %v", got)
}
if got := Lacks([]string{NeedNodeBuildNode}, func(string) bool { return true }); len(got) != 0 {
t.Errorf("%v", got)
}
step := Step{State: StepRefused, Result: json.RawMessage(`{"declined":true,"needs":["node-build-node"],"lacks":["node-build-node"]}`)}
if d, ok := DeclinedOf(step); !ok || d.Lacks[0] != NeedNodeBuildNode {
t.Errorf("%+v %v", d, ok)
}
step.State = StepFailed
if _, ok := DeclinedOf(step); ok {
t.Error("a failed step is a decline")
}
var l LevelAnswer
raw := `{"level":"unit","digest":{"packages":{"b":"2","a":"1"},"needs":[{"package":"a","needs":["node-build-node"]}]},"unmet":[{"package":"a","needs":["node-build-node"]}]}`
if err := json.Unmarshal([]byte(raw), &l); err != nil {
t.Fatal(err)
}
if p := l.Digest.Packages; len(p) != 2 || p[0].Package != "a" || l.Digest.Map()["b"] != "2" || l.Digest.NeedsOf("a")[0] != NeedNodeBuildNode {
t.Errorf("%+v", l.Digest)
}
if got := UnmetNeeds([]LevelAnswer{l, l}); len(got) != 1 {
t.Errorf("%v", got)
}
}
func TestPauseCapacityAndComposerKey(t *testing.T) {
raw, err := EncodePause(Pause{Paused: true, By: "operator@shanks", At: time.Date(2026, 11, 5, 10, 0, 0, 0, time.UTC)})
if err != nil {
t.Fatal(err)
}
if p, err := DecodePause(raw); err != nil || !p.Paused || p.V != Version {
t.Fatalf("%+v %v", p, err)
}
if _, err := DecodePause([]byte(`{"v":9}`)); !errors.Is(err, ErrUnknownVersion) {
t.Errorf("%v", err)
}
craw, _ := EncodeCapacity(Capacity{Effective: 2, Configured: 2, Why: "fits"})
if c, err := DecodeCapacity(craw); err != nil || c.Effective != 2 {
t.Fatalf("%+v %v", c, err)
}
tc := ToolchainUsed{Language: "go", Version: "1.26.0", Recipe: "1"}
if k := ComposerKey("novox/r", testCommit, "d", "p", tc); k != ComposerKey("novox/r", testCommit, "d", "p", tc) || !strings.HasPrefix(k, "composer-") || len(k) != len("composer-")+16 {
t.Errorf("composer key %q", k)
}
if ComposerKey("novox/r", testCommit, "d", "p", tc) == ComposerKey("novox/r", testCommit, "d2", "p", tc) {
t.Error("the directory is not in the key")
}
}
// The part `build-step` is one layout for the three build seats: each seat's fixtures read strictly in the neutral
// forms (an unknown field, or a field this package lacks, fails). Where the generator cannot share a part, this is
// what holds the three seats' types and this form equal.
func TestTheThreeSeatsFixturesFitTheNeutralForms(t *testing.T) {
strict := func(t *testing.T, raw json.RawMessage, into any) {
t.Helper()
dec := json.NewDecoder(bytes.NewReader(raw))
dec.DisallowUnknownFields()
if err := dec.Decode(into); err != nil {
t.Error(err)
}
}
seen := 0
for _, seat := range []string{"node-build-go", "node-build-image", "node-build-node"} {
raw, err := os.ReadFile(filepath.Join("..", "..", "contracts", "seats", seat, "fixtures", "states.json"))
if err != nil {
t.Fatal(err)
}
var fixtures []struct {
Name string `json:"name"`
Of string `json:"of"`
Given json.RawMessage `json:"given"`
}
if err := json.Unmarshal(raw, &fixtures); err != nil {
t.Fatal(err)
}
for _, f := range fixtures {
t.Run(seat+"/"+f.Name, func(t *testing.T) {
switch {
case strings.HasPrefix(f.Of, "accept:"):
seen++
strict(t, f.Given, &StepAsk{})
case strings.HasPrefix(f.Of, "state:pauses/paused."):
seen++
strict(t, f.Given, &Pause{})
case strings.HasPrefix(f.Of, "state:pauses/capacity."):
seen++
strict(t, f.Given, &Capacity{})
case strings.HasPrefix(f.Of, "state:steps/"):
seen++
strict(t, f.Given, &Step{})
}
})
}
}
if seen < 12 {
t.Errorf("only %d fixtures read", seen)
}
}
+33
View File
@@ -0,0 +1,33 @@
package buildstep
import (
"go/parser"
"go/token"
"path/filepath"
"strconv"
"strings"
"testing"
)
// The layouts, keys and codecs of this package are read by modules that never touch the bus (a composer's tool, the
// builder's published package): it imports neither the bus library nor mesh-sdk's runtime (the root package), whose
// part that runs is package buildrun's.
func TestThePackageIsFreeOfTheBus(t *testing.T) {
files, _ := filepath.Glob("*.go")
fset := token.NewFileSet()
for _, f := range files {
if strings.HasSuffix(f, "_test.go") {
continue
}
parsed, err := parser.ParseFile(fset, f, nil, parser.ImportsOnly)
if err != nil {
t.Fatal(err)
}
for _, imp := range parsed.Imports {
path, _ := strconv.Unquote(imp.Path.Value)
if strings.HasPrefix(path, "github.com/nats-io/") || path == "git.novox.be/novox/mesh-sdk/go" {
t.Errorf("%s imports %s, which reaches the bus", f, path)
}
}
}
}
+107
View File
@@ -0,0 +1,107 @@
package buildstep
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"strings"
"time"
)
// The states `pauses` of the build seats, whose three seats hold the same layouts (the part `build-step`): a node's
// pause under `paused.<node>`, its capacity under `capacity.<node>`.
// PausesState is the name of the state.
const PausesState = "pauses"
// PauseKey is the key of a node's pause.
func PauseKey(node string) (string, error) {
if err := checkNode(node); err != nil {
return "", err
}
return "paused." + node, nil
}
// CapacityKey is the key of a node's capacity.
func CapacityKey(node string) (string, error) {
if err := checkNode(node); err != nil {
return "", err
}
return "capacity." + node, nil
}
// Pause is whether a node takes no new ask, and who said so.
type Pause struct {
V int `json:"v"`
Paused bool `json:"paused"`
By string `json:"by,omitempty"`
At time.Time `json:"at,omitzero"`
}
// EncodePause is p as a pauses value.
func EncodePause(p Pause) ([]byte, error) {
p.V = Version
p.At = p.At.UTC()
return json.Marshal(p)
}
// DecodePause reads a pause, refusing another version.
func DecodePause(data []byte) (Pause, error) {
var p Pause
if err := decodeVersioned(data, &p, func() int { return p.V }); err != nil {
return Pause{}, err
}
return p, nil
}
// Capacity is how many steps a node runs at once, and why.
type Capacity struct {
V int `json:"v"`
Effective int `json:"effective"`
Configured int `json:"configured"`
ConfiguredFrom string `json:"configured-from,omitempty"`
QuotaCores float64 `json:"quota-cores,omitempty"`
QuotaFrom string `json:"quota-from"`
CoresPerStep float64 `json:"cores-per-step"`
ByCPU int `json:"by-cpu,omitempty"`
BuildHours string `json:"build-hours,omitempty"`
OutsideHours bool `json:"outside-hours,omitempty"`
BuildHoursFault string `json:"build-hours-fault,omitempty"`
Ineligible string `json:"ineligible,omitempty"`
Why string `json:"why"`
At time.Time `json:"at,omitzero"`
}
// EncodeCapacity is c as a pauses value.
func EncodeCapacity(c Capacity) ([]byte, error) {
c.V = Version
c.At = c.At.UTC()
return json.Marshal(c)
}
// DecodeCapacity reads a capacity, refusing another version.
func DecodeCapacity(data []byte) (Capacity, error) {
var c Capacity
if err := decodeVersioned(data, &c, func() int { return c.V }); err != nil {
return Capacity{}, err
}
return c, nil
}
func decodeVersioned(data []byte, into any, version func() int) error {
if err := json.Unmarshal(data, into); err != nil {
return fmt.Errorf("buildstep: value does not fit its shape: %w", err)
}
if v := version(); v != Version {
return fmt.Errorf("%w: a value of version %d, this reader reads %d", ErrUnknownVersion, v, Version)
}
return nil
}
// ComposerKey is the key a composer build is kept under: the fetched commit, the directory, the package and the
// toolchain it ran with.
func ComposerKey(repository, commit, dir, pkg string, toolchain ToolchainUsed) string {
sum := sha256.Sum256([]byte(strings.Join([]string{repository, commit, dir, pkg, toolchain.Version, toolchain.Recipe}, "\n")))
return "composer-" + hex.EncodeToString(sum[:8])
}
+64
View File
@@ -0,0 +1,64 @@
package window
import (
"slices"
"sort"
"time"
)
// WindowHeld is the part of a window every window keeper shares (design 51: distribution's old seat name and direct
// writes, the builder's old reference form): whether it is open, the nodes holding it and since when, and when it last
// closed and who held it then. A keeper embeds it in its own record, so its JSON is the record's.
type WindowHeld struct {
Open bool `json:"open"`
Holding []string `json:"holding,omitempty"`
Since map[string]time.Time `json:"since,omitempty"`
Closed time.Time `json:"closed,omitzero"`
// LastHolding are the nodes that held it last, said when it closes.
LastHolding []string `json:"last_holding,omitempty"`
}
// StepWindow is the one rule every window keeper steps its window by (security-upgrade review F3, the second of a
// kind): named is every node the read named, dropped ones included, and holding those of them that hold the window at
// now. A read that names no node judges nothing, as a first view's empty answer at revision 0 is no empty state (ADR
// 0615): an empty node_haves or a placement not yet read never closes a window, and the window stays as it was. A read
// naming nodes, none holding, closes an open window (closed true, once); one naming a holder opens it, each holder's
// since kept from w. overdue are the holders whose since is at least findingAfter before now (the keeper's finding,
// cleared for every other node); a read naming no node keeps w's holders' overdue as they stand at now.
func StepWindow(w WindowHeld, named, holding []string, now time.Time, findingAfter time.Duration) (next WindowHeld, closed bool, overdue []string) {
now = now.UTC()
if len(named) == 0 {
next = w
for _, n := range w.Holding {
if since, ok := w.Since[n]; ok && now.Sub(since) >= findingAfter {
overdue = append(overdue, n)
}
}
return next, false, overdue
}
if len(holding) == 0 {
next = WindowHeld{Closed: w.Closed, LastHolding: w.LastHolding}
if w.Open {
next.Closed, next.LastHolding = now, slices.Clone(w.Holding)
if next.LastHolding == nil {
next.LastHolding = []string{}
}
return next, true, nil
}
return next, false, nil
}
held := slices.Clone(holding)
sort.Strings(held)
next = WindowHeld{Open: true, Holding: held, Since: map[string]time.Time{}, Closed: w.Closed, LastHolding: w.LastHolding}
for _, n := range held {
since, had := w.Since[n]
if !had {
since = now
}
next.Since[n] = since
if now.Sub(since) >= findingAfter {
overdue = append(overdue, n)
}
}
return next, false, overdue
}
+41
View File
@@ -0,0 +1,41 @@
package window
import (
"slices"
"testing"
"time"
)
// The one rule every window keeper steps by (design 51; security-upgrade review F3): a read naming no node judges
// nothing, so an empty node_haves or an unread placement never closes a window.
func TestStepWindow(t *testing.T) {
now := time.Date(2026, 11, 4, 17, 0, 0, 0, time.UTC)
week := 7 * 24 * time.Hour
long := now.Add(-8 * 24 * time.Hour)
open := WindowHeld{Open: true, Holding: []string{"n1"}, Since: map[string]time.Time{"n1": long}}
for _, c := range []struct {
name string
w WindowHeld
named, holding []string
open, closed bool
holds, overdue []string
}{
{"a read naming no node leaves an open window open, its holder still overdue", open, nil, nil, true, false, []string{"n1"}, []string{"n1"}},
{"a read naming no node leaves a closed window closed", WindowHeld{Closed: long}, nil, nil, false, false, nil, nil},
{"a read naming nodes, none holding, closes an open window once", open, []string{"n1", "n2"}, nil, false, true, nil, nil},
{"a closed window a read finds unheld stays closed, said no more", WindowHeld{Closed: long, LastHolding: []string{"n1"}}, []string{"n1"}, nil, false, false, nil, nil},
{"a new holder opens it, since now, not overdue", WindowHeld{}, []string{"n1", "n2"}, []string{"n2"}, true, false, []string{"n2"}, nil},
{"a holder keeps its since and is overdue past the bound", open, []string{"n1", "n2"}, []string{"n2", "n1"}, true, false, []string{"n1", "n2"}, []string{"n1"}},
} {
next, closed, overdue := StepWindow(c.w, c.named, c.holding, now, week)
if next.Open != c.open || closed != c.closed || !slices.Equal(next.Holding, c.holds) || !slices.Equal(overdue, c.overdue) {
t.Errorf("%s: open %v closed %v holding %v overdue %v", c.name, next.Open, closed, next.Holding, overdue)
}
if closed && (!next.Closed.Equal(now) || !slices.Equal(next.LastHolding, c.w.Holding)) {
t.Errorf("%s: closed at %v, last held by %v", c.name, next.Closed, next.LastHolding)
}
if !c.open && !closed && !next.Closed.Equal(c.w.Closed) {
t.Errorf("%s: its last close moved to %v", c.name, next.Closed)
}
}
}