Merge branch 'b2/registry' into b7f/registry
This commit is contained in:
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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>.
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
} {
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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])
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user