Merge branch 'feat/47-G-build' into feat/47-G-build-contracts

This commit is contained in:
2026-11-11 11:45:50 +01:00
25 changed files with 1429 additions and 33 deletions
@@ -1128,10 +1128,10 @@ func aReleasesApproveAnswersTheMoveOfItsBuildAlone(t *testing.T) {
p.logf = t.Logf
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: commit, Src: 7, FpSrc: c.fpSrc, Path: meshcatalogclient.RootPath})
p.w.put(buckets.Modules, "web", mod)
src, _ := mdc.EncodeSource(mdseat.SourceByMainValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}})
p.w.putEntry(buckets.Source, mdc.SourceMainKey, src, 7, time.Now())
later, _ := mdc.EncodeSource(mdseat.SourceByMainValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}})
p.w.putEntry(buckets.Source, mdc.SourceMainKey, later, 8, time.Now()) // a later merge leaves revision 7's release
src, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}})
p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), src, 7, time.Now())
later, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}})
p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), later, 8, time.Now()) // a later merge leaves revision 7's release
if c.mergedAt != "" {
merge, err := mdc.EncodeMerge(mdseat.MergesByGroupValue{Group: "feat/web", Ask: "merge.feat_web.0123456789ab", Pulls: []mdseat.MergesByGroupValuePullsItem{
{Repository: "novox/web", Number: 3, Head: strings.Repeat("3", 40), State: string(mdc.MergeMerged), Commit: c.mergedAt}}})
@@ -315,6 +315,9 @@ func (s *wantsTools) tools() []stdio.Tool {
}
var out []stdio.Tool
for _, v := range meshwants.Verbs() {
if run[v] == nil {
continue // a verb of the seat this module does not serve yet: revision, entry, difference (AS11), input (AS5)
}
out = append(out, stdio.Tool{Name: meshwants.Seat + "." + v, Run: run[v]})
}
return out
@@ -86,7 +86,7 @@ func (w *world) putEntry(bucket, key string, value []byte, revision uint64, at t
if bucket == plc.AssignmentsBucket {
_, _ = w.assignmentKeys.Put(key, value, revision, false) // put took it already; this is its revision
}
if bucket == buckets.Source && key == mdc.SourceMainKey {
if bucket == buckets.Source && key == mdc.SourceKey(trunkBranch) {
if s, err := mdc.DecodeSource(value); err == nil && s.Group != "" {
w.sourceGroups[revision] = s.Group
}
@@ -575,3 +575,6 @@ func (w *world) unplacedNodes() []string {
sort.Strings(out)
return out
}
// trunkBranch is the branch of the trunk-only want, kept until the first registrations (design 47 G §95, §353).
const trunkBranch = "main"
@@ -9,6 +9,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
"mesh-delivery-assigner/internal/puts"
"reflect"
"slices"
"strings"
"testing"
@@ -431,7 +432,7 @@ func TestEachRuleHoldsTheWantBack(t *testing.T) {
before["g14"] = mwc.Want{Fp: fNew, Composed: dig(fNew)}
}
out := Wants(m, n, before, nil)
if got := out.Wants["g14"]; got != c.want {
if got := out.Wants["g14"]; !reflect.DeepEqual(got, c.want) {
t.Errorf("%s: g14's want = %+v, want %+v", c.name, got, c.want)
}
if got := out.Rules["g14"].Rule; got != c.rule {
@@ -409,7 +409,16 @@ func main() {
}
return e.Value, true, nil
}}
rs := stdio.State(versions.ReservedState)
rr := versions.Reader{Keys: rs.Keys, Get: func(key string) (json.RawMessage, bool, error) {
e, err := rs.Get(key)
if err != nil || e == nil {
return nil, false, err
}
return e.Value, true, nil
}}
tools := Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(builds.BuilderLog).Since, time.Now, vr, tags, tr, fr, ofr)
tools = append(tools, reservedTools(rr, buildAgain, time.Now, meshhandactsclient.Recorder{Ask: stdio.Ask})...)
// The builder's expiring keys (state bounds): what each bounds and when it ends.
tools = append(tools, stdio.Tool{Name: versions.BoundsTool, Description: versions.BoundsDescription,
Input: map[string]any{"type": "object", "properties": map[string]any{}},
@@ -0,0 +1,62 @@
package main
import (
"errors"
"fmt"
"strings"
"time"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
stdio "git.novox.be/novox/mesh-sdk/go"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
)
// reservedTools are builder_reserved, builder_reserved_drop and the seat mesh-builds' verb reserved (design 47 G section
// 369), served as tools of the bundle. The drop is a hand act: it is recorded in the hand-act log first, then written to
// reasks for the builder, which owns the reservations, to answer.
func reservedTools(rr versions.Reader, ask queueAsker, now func() time.Time, acts meshhandactsclient.Recorder) []stdio.Tool {
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
return []stdio.Tool{
{Name: versions.ReservedTool, Description: versions.ReservedDescription,
Input: map[string]any{"type": "object", "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "only this module's reservations"}}},
Run: func(args map[string]any) (any, error) { return versions.AnswerReserved(rr, str(args, "module")) }},
{Name: versions.ReservedVerb, Description: "The seat mesh-builds' verb reserved: a module's reserved version with each artifact's digest once built " +
"(design 47 G section 369), for a registry holder deciding whether to publish. Read-only.",
Input: map[string]any{"type": "object", "required": []string{"module"}, "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "the module"},
"version": map[string]any{"type": "string", "description": "the version, when asked for a particular one"}}},
Run: func(args map[string]any) (any, error) {
return versions.AnswerReservedVerb(rr, str(args, "module"), str(args, "version"))
}},
{Name: versions.ReservedDropTool, Description: versions.ReservedDropDescription,
Input: map[string]any{"type": "object", "required": []string{"module", "series", "why"}, "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "the module"},
"series": map[string]any{"type": "string", "description": "its MAJOR.MINOR"},
"why": map[string]any{"type": "string", "description": "why, in plain words"}}},
Run: func(args map[string]any) (any, error) {
key, value, err := build.NewReservedDrop(str(args, "module"), str(args, "series"), str(args, "why"), stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
if _, err := acts.Record(meshhandactsclient.Request{What: fmt.Sprintf("dropped the reservation of %s %s by hand", str(args, "module"), str(args, "series")),
Why: str(args, "why"), Cause: versions.ReservedDropTool, Verb: "reserved-drop"}); err != nil {
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was dropped: %w", err)
}
raw, err := ask(key, value)
if err != nil {
return nil, err
}
a, err := build.DecodeReAsk(raw)
if err != nil {
return nil, err
}
if a.Refused {
return nil, errors.New(a.Answer)
}
return map[string]any{"key": key, "answer": a.Answer}, nil
}},
}
}
@@ -623,7 +623,7 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
}
// Running-testing note (review round 1): two claims never get one number on the bus's own compare-and-set, as
// the builder's state versions keeps them: eight claims of one name at once, eight versions.
store := versions.Versions{Store: awaitedVersions{process.AwaitBucket(b.ctx, b.js, nil, versions.Bucket, "", t.Logf)}}
store := versions.Versions{Store: awaitedVersions{b: process.AwaitBucket(b.ctx, b.js, nil, versions.Bucket, "", t.Logf)}}
numbers := make(chan string, 8)
var claims sync.WaitGroup
for i := range 8 {
@@ -163,7 +163,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
floorWaits.Keep(kvKeeper{ctx: ctx, kv: kv}, heldFloorWaits(ctx, kv, logger.Printf))
})
// A package is numbered under its name in its ecosystem, read from the indexer's packages state (design 32).
numberer := &versions.Numberer{Versions: versions.Versions{Store: awaitedVersions{versionsKV}}, Floor: floorVia(calls.Call), Waits: floorWaits,
numberer := &versions.Numberer{Versions: versions.Versions{Store: awaitedVersions{b: versionsKV}}, Floor: floorVia(calls.Call), Waits: floorWaits,
Names: versions.NamesIn(indexerPackages{packages}),
Said: func(nb builds.VersionNumbered) {
if err := emitNumbered(context.WithoutCancel(ctx), js, cred, nb); err != nil {
@@ -202,10 +202,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
boundsKV := process.AwaitBucket(ctx, js, conn, versions.TagAskBoundsBucket, "an unanswered tag ask is not narrated", logger.Printf)
bounds := &tagBounds{asks: tagAsksKV.KV, bounds: boundsKV.KV, bound: tagAskBound, waits: floorWaits.Waiting, logf: logger.Printf, now: time.Now,
log: func(e builds.BuildLogEntry) { b.Log(e) }}
// The versions a build reserves at its start (design 47 G §369): its state reserved, and the catalog's versions-filled
// and version-released heard; nothing is reserved before the fill.
reservedKV := process.AwaitBucket(ctx, js, conn, versions.ReservedBucket, "no version is reserved: builds are made unreserved", logger.Printf)
reserved := newReservations(awaitedReserved{awaitedVersions{b: reservedKV, bucket: versions.ReservedBucket}}, calls.Call, pubBoundsKV.KV, found, b.Reconsider, logger.Printf)
b.Reserved = reserved
// A state bound later numbers the builds looked at without it.
versionsKV.Ready(func(jetstream.KeyValue) { b.Reconsider(ctx) })
// The versions' hand acts (builder_publish_again, builder_floor_again) come through reasks, as builder_build_again.
acts := versions.Acts{Numberer: numberer, Done: doneVia(b), Reconsider: b.Reconsider, Remove: removeVia(calls.Call),
acts := versions.Acts{Numberer: numberer, DropReserved: reserved.dropped, Done: doneVia(b), Reconsider: b.Reconsider, Remove: removeVia(calls.Call),
BuildAgain: func(ctx context.Context, module, fp, why, by string) (string, bool) {
r := b.BuildAgain(ctx, module, fp, why, by, buildAgainBound)
if r.Refused != "" {
@@ -349,9 +354,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// condition keeper's condition-cleared.
floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf}
// And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version.
seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs}
seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs, res: reserved}
events := func(m jetstream.Msg) {
if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || reserved.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
return
}
heardGone(ctx, b, m, logger.Printf)
@@ -407,6 +412,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
})
pubBoundsKV.Ready(func(kv jetstream.KeyValue) {
each := func(e jetstream.KeyValueEntry) {
if reserved.ended(ctx, e) {
return
}
if _, ok := versions.UnputBoundOf(e.Key()); ok {
unput.expired(ctx, e)
return
@@ -53,7 +53,7 @@ func reAsked(ctx context.Context, b askAgainer, acts versions.Acts, kv jetstream
announcedAgain(ctx, b, kv, e, r, logf)
return
}
if r.Mode == build.ModePublishAgain || r.Mode == build.ModeFloorAgain || r.Mode == build.ModeOldFormDrop {
if r.Mode == build.ModePublishAgain || r.Mode == build.ModeFloorAgain || r.Mode == build.ModeOldFormDrop || r.Mode == build.ModeReservedDrop {
// A version's hand act (design 51 *Tools of the builder*) may build again, so it is answered beside the watch.
go versionActed(ctx, acts, kv, e, r, logf)
return
@@ -163,6 +163,12 @@ func versionActed(ctx context.Context, acts versions.Acts, kv jetstream.KeyValue
var ag versions.Again
ag, r.Answer, err = acts.PublishAgain(ctx, r.Name, r.FP, r.Why, r.By, r.At)
r.Do, r.Version = ag.Do, ag.Entry.Version
} else if r.Mode == build.ModeReservedDrop {
if acts.DropReserved == nil {
err = errors.New("this builder keeps no reservations")
} else {
r.Answer, err = acts.DropReserved(ctx, r.Name, r.FP, r.By, r.Why, r.At)
}
} else if r.Mode == build.ModeOldFormDrop {
if acts.DropOldForm == nil {
err = errors.New("this builder keeps no old-form window")
@@ -0,0 +1,208 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"sync/atomic"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
)
// The reservations (design 47 G §369): the builder's state reserved, the catalog's versions it numbers from, and the
// catalog's version-released and versions-filled it hears.
// awaitedReserved is the state reserved as versions.ReserveStore: bound once the bus holds it.
type awaitedReserved struct{ awaitedVersions }
func (s awaitedReserved) Delete(ctx context.Context, key string) error {
kv, err := s.kv()
if err != nil {
return err
}
if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
return err
}
return nil
}
// reservations is the builder's reserving: the Reserver, the fill the catalog said, the bounds of a built reservation's
// publish and the findings.
type reservations struct {
r *versions.Reserver
filled atomic.Bool
bounds func() jetstream.KeyValue
found *findings
bound time.Duration
wake func(context.Context)
logf func(string, ...any)
}
// newReservations makes the reserving over store; the released versions of a module are asked of the seat mesh-catalog's
// verb versions through call (design 47 G §348), the only source numbering reads them from.
func newReservations(store versions.ReserveStore, call toolCaller, bounds func() jetstream.KeyValue, found *findings, wake func(context.Context), logf func(string, ...any)) *reservations {
rv := &reservations{bounds: bounds, found: found, bound: versions.UnreleasedBound, wake: wake, logf: logf}
rv.r = &versions.Reserver{Store: store, Filled: rv.filled.Load, ReleasedOf: func(_ context.Context, module string) ([]string, error) {
raw, err := call("seat:"+meshcatalog.Seat+".versions", meshcatalog.VersionsInput{Module: module})
if err != nil {
return nil, err
}
var a meshcatalog.VersionsAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return nil, fmt.Errorf("the catalog's versions of %s answered what is no list of versions: %w", module, err)
}
var out []string
for _, v := range a.Versions {
out = append(out, v.Version)
}
return out, nil
}}
return rv
}
var _ build.Reservations = (*reservations)(nil)
func (rv *reservations) Filled() bool { return rv.filled.Load() }
func (rv *reservations) Reserve(ctx context.Context, module, series string, artifacts []string) (versions.ReservedVersion, error) {
return rv.r.Reserve(ctx, module, series, artifacts)
}
// Built records the digests; once every artifact has one, the publish's bound starts (a key of the state bounds with its
// own time to live): its end with the reservation standing is version-unreleased.
func (rv *reservations) Built(ctx context.Context, module, version string, digests map[string]string) error {
for ref, d := range digests {
if _, err := rv.r.Built(ctx, module, version, ref, d); err != nil {
return err
}
}
cur, found, err := rv.r.Of(ctx, module, version)
if err != nil || !found || cur.BuiltAt.IsZero() {
return err
}
rv.arm(ctx, cur)
return nil
}
func (rv *reservations) arm(ctx context.Context, v versions.ReservedVersion) {
kv := rv.bounds()
var err error
if kv == nil {
err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket)
} else if _, err = kv.Create(ctx, versions.ReservedBoundKey(v.Module, v.Series), versions.EncodeBound("the release of "+v.Module+" "+v.Version, time.Now().Add(rv.bound)), jetstream.KeyTTL(rv.bound)); errors.Is(err, jetstream.ErrKeyExists) {
err = nil
}
if err != nil {
rv.logf("the release bound of %s %s is not written, so it is raised unreleased now: %v", v.Module, v.Version, err)
rv.unreleased(ctx, v, fmt.Sprintf("its bound could not be written (%v)", err))
}
}
// unreleased raises version-unreleased about the module, naming the version and the artifacts missing.
func (rv *reservations) unreleased(ctx context.Context, v versions.ReservedVersion, why string) {
missing := v.Missing()
what := "an artifact is unpublished"
if len(missing) > 0 {
what = strings.Join(missing, ", ") + " unpublished"
}
rv.found.raise(ctx, finding{kind: versions.VersionUnreleased, scope: meshconditionsclient.ScopeModule, id: v.Module,
summary: fmt.Sprintf("%s %s is built and not released: %s (%s)", v.Module, v.Version, what, why),
headline: "A built version is not released",
explanation: "The builder reserved " + v.Module + " " + v.Version + " and its build ended, and the catalog has not said version-released: " + what + ".",
resolved: "The catalog says version-released for " + v.Module + " " + v.Version + ".",
needs: "builder_reserved shows the reservation; builder_publish_again publishes the missing artifacts under it; builder_reserved_drop drops it."})
}
// ended is told each change of the state bounds: a reservation's bound whose time ended raises version-unreleased while
// the reservation stands built.
func (rv *reservations) ended(ctx context.Context, e jetstream.KeyValueEntry) bool {
key, ok := versions.ReservedBoundOf(e.Key())
if !ok {
return false
}
if e.Operation() != jetstream.KeyValuePurge {
return true
}
v, found, err := rv.r.At(ctx, key)
if err != nil || !found || v.BuiltAt.IsZero() {
return true
}
rv.unreleased(ctx, v, fmt.Sprintf("nothing within %s of its build", rv.bound))
return true
}
// dropBound ends a reservation's bound.
func (rv *reservations) dropBound(ctx context.Context, key string) {
if kv := rv.bounds(); kv != nil {
if err := kv.Delete(ctx, versions.ReservedBoundFor(key)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
rv.logf("the release bound of %s is not ended: %v", key, err)
}
}
}
// heard takes the catalog's version-released and versions-filled in, and a registry seat's published (published): a
// message that is none of them is left to the others. A state that cannot take one leaves it for the bus to give again.
func (rv *reservations) heard(ctx context.Context, m jetstream.Msg) bool {
switch m.Subject() {
case meshcatalog.SubjectVersionsFilled:
rv.filled.Store(true)
rv.logf("the catalog said versions-filled: versions are reserved from now on")
if rv.wake != nil {
rv.wake(ctx)
}
case meshcatalog.SubjectVersionReleased:
var ev meshcatalog.VersionReleasedEvent
if err := json.Unmarshal(m.Data(), &ev); err != nil || ev.Module == "" || ev.Version == "" {
rv.logf("%s holds what is no version-released, left: %v", m.Subject(), err)
break
}
gone, err := rv.r.Released(ctx, ev.Module, ev.Version)
if err != nil {
rv.logf("%v; heard again", err)
if nakErr := m.Nak(); nakErr != nil {
rv.logf("leaving %s for the bus to give again: %v", m.Subject(), nakErr)
}
return true
}
for _, k := range gone {
rv.dropBound(ctx, k)
}
rv.found.clear(ctx, versions.VersionUnreleased, meshconditionsclient.ScopeModule, ev.Module)
default:
return false
}
if err := m.Ack(); err != nil {
rv.logf("acknowledging %s: %v", m.Subject(), err)
}
return true
}
// published marks the artifacts of the reservation a registry seat's published names (its version and digest).
func (rv *reservations) published(ctx context.Context, o versions.Outcome) {
if _, err := rv.r.Published(ctx, o.Version, o.Digest); err != nil {
rv.logf("a published of %s@%s is not recorded in the reservations: %v", o.Name, o.Version, err)
}
}
// dropped is builder_reserved_drop: the reservation is deleted and its bound ended.
func (rv *reservations) dropped(ctx context.Context, module, series, by, why string, at time.Time) (string, error) {
ok, err := rv.r.Drop(ctx, module, series)
if err != nil {
return "", err
}
if !ok {
return "", fmt.Errorf("%s holds no reservation of %s", module, series)
}
rv.dropBound(ctx, versions.ReservedKey(module, series))
rv.found.clear(ctx, versions.VersionUnreleased, meshconditionsclient.ScopeModule, module)
return fmt.Sprintf("the reservation of %s %s is dropped (%s, %s)", module, series, by, why), nil
}
@@ -0,0 +1,131 @@
package main
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"mesh-delivery-builder/internal/versions"
)
// memReserve is the state reserved in memory.
type memReserve struct {
vals map[string][]byte
revs map[string]uint64
}
func (m *memReserve) Get(_ context.Context, key string) ([]byte, uint64, bool, error) {
v, ok := m.vals[key]
return v, m.revs[key], ok, nil
}
func (m *memReserve) Create(_ context.Context, key string, v []byte) (uint64, error) {
if _, ok := m.vals[key]; ok {
return 0, versions.ErrConflict
}
m.vals[key], m.revs[key] = v, 1
return 1, nil
}
func (m *memReserve) Update(_ context.Context, key string, v []byte, rev uint64) (uint64, error) {
if m.revs[key] != rev {
return 0, versions.ErrConflict
}
m.vals[key], m.revs[key] = v, rev+1
return rev + 1, nil
}
func (m *memReserve) KeysOf(_ context.Context, filter string) ([]string, error) {
var out []string
for k := range m.vals {
if filter == versions.AllFilter || strings.HasPrefix(k, strings.TrimSuffix(filter, "*")) {
out = append(out, k)
}
}
return out, nil
}
func (m *memReserve) Delete(_ context.Context, key string) error {
delete(m.vals, key)
return nil
}
// boundsKV keeps the keys created and deleted in the state bounds.
type boundsKV struct {
jetstream.KeyValue
keys map[string]bool
}
func (b *boundsKV) Create(_ context.Context, key string, _ []byte, _ ...jetstream.KVCreateOpt) (uint64, error) {
b.keys[key] = true
return 1, nil
}
func (b *boundsKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
delete(b.keys, key)
return nil
}
// TestTheCatalogsWordsMoveTheReservations: design 47 G §369, §372 (test 102) — versions-filled opens the reserving and wakes
// the builds that wait; a built reservation arms its bound and its end raises version-unreleased naming the artifacts
// missing; the catalog's version-released deletes the key and the bound and clears the finding; builder_reserved_drop drops one.
func TestTheCatalogsWordsMoveTheReservations(t *testing.T) {
ctx := context.Background()
st := &memFindings{vals: map[string][]byte{}}
found := &findings{store: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }}
bounds := &boundsKV{keys: map[string]bool{}}
woke := 0
call := func(addr string, body any) (json.RawMessage, error) {
if addr != "seat:mesh-catalog.versions" {
t.Errorf("the released versions were asked of %q", addr)
}
return json.Marshal(meshcatalog.VersionsAnswer{Module: "m", Versions: []meshcatalog.VersionEntry{{Module: "m", Version: "1.4.2"}}})
}
rv := newReservations(&memReserve{vals: map[string][]byte{}, revs: map[string]uint64{}}, call, func() jetstream.KeyValue { return bounds }, found,
func(context.Context) { woke++ }, t.Logf)
if _, err := rv.Reserve(ctx, "m", "1.4", []string{"m:a"}); err == nil || rv.Filled() {
t.Fatalf("a version was reserved before versions-filled (%v)", err)
}
filled := &wakeMsg{subject: meshcatalog.SubjectVersionsFilled, data: []byte(`{"v":1}`)}
if !rv.heard(ctx, filled) || !filled.acked || !rv.Filled() || woke != 1 {
t.Fatalf("versions-filled: acked %v, filled %v, woke %d", filled.acked, rv.Filled(), woke)
}
res, err := rv.Reserve(ctx, "m", "1.4", []string{"m:a", "m:b"})
if err != nil || res.Version != "1.4.3" {
t.Fatalf("reserved %q (%v), want 1.4.3 after the catalog's 1.4.2", res.Version, err)
}
if err := rv.Built(ctx, "m", "1.4.3", map[string]string{"m:a": "sha256:a", "m:b": "sha256:b"}); err != nil {
t.Fatal(err)
}
boundKey := versions.ReservedBoundKey("m", "1.4")
if !bounds.keys[boundKey] {
t.Fatalf("the publish's bound is not armed: %v", bounds.keys)
}
rv.published(ctx, versions.Outcome{Name: "m-a", Version: "1.4.3", Digest: "sha256:a"})
if !rv.ended(ctx, purged{key: boundKey}) {
t.Fatal("a reservation's bound was not taken in")
}
key := meshconditionsclient.Key(meshconditionsclient.ScopeModule, "m", versions.VersionUnreleased)
got, err := meshconditionsclient.DecodeFinding(st.vals[key])
if err != nil || !strings.Contains(got.Summary, "m:b unpublished") || strings.Contains(got.Summary, "m:a unpublished") || !strings.Contains(got.Summary, "1.4.3") {
t.Fatalf("finding = %+v (%v), want version-unreleased naming m:b of 1.4.3", got, err)
}
released := &wakeMsg{subject: meshcatalog.SubjectVersionReleased, data: []byte(`{"v":1,"module":"m","version":"1.4.3","source":1,"artifacts":[]}`)}
if !rv.heard(ctx, released) || !released.acked {
t.Fatalf("version-released was not taken in (acked %v)", released.acked)
}
if all, _ := rv.r.All(ctx); len(all) != 0 || bounds.keys[boundKey] || len(st.vals) != 0 {
t.Errorf("after version-released: reservations %v, bounds %v, findings %v; want none", all, bounds.keys, st.vals)
}
if _, err := rv.Reserve(ctx, "m", "1.4", []string{"m:a"}); err != nil {
t.Fatal(err)
}
if answer, err := rv.dropped(ctx, "m", "1.4", "op", "never released", time.Unix(0, 0)); err != nil || !strings.Contains(answer, "dropped") {
t.Errorf("builder_reserved_drop: %q (%v)", answer, err)
}
if _, err := rv.dropped(ctx, "m", "1.4", "op", "again", time.Unix(0, 0)); err == nil {
t.Error("a reservation that is gone was dropped again")
}
}
@@ -50,6 +50,8 @@ type seatHearer struct {
logf func(string, ...any)
// pub raises and clears the findings of a publish that ended badly; nil raises none.
pub *publishes
// res marks a reservation's artifacts published; nil marks none.
res *reservations
}
// heard takes a message of the events consumer in when it is a registry seat's outcome, and answers whether it was.
@@ -90,6 +92,9 @@ func (h *seatHearer) heard(ctx context.Context, m jetstream.Msg) bool {
if h.pub != nil {
h.pub.heard(ctx, se.seat, se.event, o)
}
if h.res != nil && se.event == versions.HeardPublished {
h.res.published(ctx, o)
}
h.ack(m)
if len(wake) > 0 {
h.logf("the floor of %s is asked again: %s said %s", strings.Join(wake, ", "), se.seat, se.event)
@@ -29,7 +29,11 @@ const (
// awaitedVersions is the state versions as versions.Store: bound once the bus holds it (the grants part makes it, issue
// 1790); until then every use is an error, so a trunk build is built unnumbered and said (build.Builder.number).
type awaitedVersions struct{ b *process.AwaitedBucket }
type awaitedVersions struct {
b *process.AwaitedBucket
// bucket names the state in the error of one not on the bus yet; empty is the versions'.
bucket string
}
var errVersionsMissing = errors.New("the state " + versions.Bucket + " is not on the bus yet")
@@ -37,6 +41,9 @@ func (s awaitedVersions) kv() (jetstream.KeyValue, error) {
if kv := s.b.KV(); kv != nil {
return kv, nil
}
if s.bucket != "" {
return nil, errors.New("the state " + s.bucket + " is not on the bus yet")
}
return nil, errVersionsMissing
}
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"strconv"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
@@ -140,3 +141,59 @@ func (b *Builder) PackageNamed(ctx context.Context, module string) {
b.consider(ctx, module, m)
}
}
// Reservations reserves the version of a trunk build at its start and records each artifact's digest (design 47 G §369).
// Nil reserves nothing.
type Reservations interface {
// Filled says the catalog said versions-filled: nothing is reserved before it.
Filled() bool
// Reserve reserves module's next version in series for the artifacts (their references, <module>:<artifact>).
Reserve(ctx context.Context, module, series string, artifacts []string) (versions.ReservedVersion, error)
// Built records the digests of the artifacts built under the reserved version, by reference.
Built(ctx context.Context, module, version string, digests map[string]string) error
}
// reserves says whether a look at a trunk build is one that reserves: the build has a series of the form MAJOR.MINOR
// (design 47 G §368) and artifacts, and its key is not done already.
func (b *Builder) reserves(ctx context.Context, module string, m meshcatalog.ModulesByModuleValue, plan Plan) bool {
if b.Reserved == nil || len(plan.Tasks) == 0 {
return false
}
if strings.Count(plan.Series, ".") != 1 {
return false
}
return b.doneDigests(ctx, module, m.Fp2) == nil
}
// waitFill holds a build that would reserve until the catalog says versions-filled, saying wait-held naming the fill once
// (design 47 G §369, §372). It answers whether the build waits.
func (b *Builder) waitFill(module string, m meshcatalog.ModulesByModuleValue) bool {
if b.Reserved.Filled() {
return false
}
if _, said := b.fillWaits.LoadOrStore(module+"."+m.Fp2, true); !said {
line := module + " at " + m.Fp2 + " is not built yet: wait-held, the catalog has not said versions-filled, and nothing is reserved before it"
b.logf("%s", line)
if b.Log != nil {
b.Log(builds.BuildLogEntry{Event: builds.BuildSaidEvent, Module: module, FP: m.Fp2, Commit: m.Commit, Node: b.Node,
At: time.Now().UTC(), Line: line, Why: "wait-held"})
}
}
return true
}
// reserve reserves the version of the build that starts; a reservation that cannot be made is said and the build goes on
// unreserved, as an unnumbered build does.
func (b *Builder) reserve(ctx context.Context, module string, plan Plan) (versions.ReservedVersion, bool) {
refs := make([]string, 0, len(plan.Tasks))
for _, t := range plan.Tasks {
refs = append(refs, module+":"+t.Artifact)
}
res, err := b.Reserved.Reserve(ctx, module, plan.Series, refs)
if err != nil {
b.logf("%s: no version is reserved for this build: %v", module, err)
return versions.ReservedVersion{}, false
}
b.logf("%s: version %s is reserved for this build", module, res.Version)
return res, true
}
@@ -35,7 +35,11 @@ type Task struct {
}
// Plan is what a manifest asks the builder to build, by artifact name.
type Plan struct{ Tasks []Task }
type Plan struct {
Tasks []Task
// Series is the manifest's MAJOR.MINOR (design 47 G §368): the series its build reserves a version in (§369).
Series string
}
// Modules are the build modules the plan calls, sorted.
func (p Plan) Modules() []string {
@@ -89,7 +93,7 @@ func PlanOf(m catalogue.Manifest) (Plan, error) {
}
artifacts := append([]catalogue.Artifact(nil), m.Build.Artifacts...)
sort.Slice(artifacts, func(i, j int) bool { return artifacts[i].Name < artifacts[j].Name })
var p Plan
p := Plan{Series: m.Version}
for _, a := range artifacts {
switch {
case a.Kind == catalogue.ArtifactArchive:
@@ -0,0 +1,104 @@
package build
import (
"context"
"slices"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"mesh-delivery-builder/internal/versions"
)
// fakeReservations answers a fill and keeps what the builder reserved and recorded.
type fakeReservations struct {
filled bool
reserved [][]string
series []string
built map[string]string
}
func (f *fakeReservations) Filled() bool { return f.filled }
func (f *fakeReservations) Reserve(_ context.Context, module, series string, artifacts []string) (versions.ReservedVersion, error) {
f.reserved, f.series = append(f.reserved, artifacts), append(f.series, series)
return versions.ReservedVersion{Module: module, Series: series, Version: series + ".5"}, nil
}
func (f *fakeReservations) Built(_ context.Context, _, version string, digests map[string]string) error {
f.built = map[string]string{version: ""}
for ref, d := range digests {
f.built[ref] = d
}
return nil
}
// TestABuildReservesItsVersionAtItsStartOnceTheCatalogFilled: design 47 G §369, §372 (test 102) — nothing is reserved before
// versions-filled and the build waits saying wait-held naming the fill; after it the build reserves one version in the
// manifest's MAJOR.MINOR for its artifacts and records their digests; a done key reserves nothing again; a manifest whose
// version is no MAJOR.MINOR reserves nothing.
func TestABuildReservesItsVersionAtItsStartOnceTheCatalogFilled(t *testing.T) {
r := newRig(t, goBundle)
r.mf.byDigest[mdigest('a')] = catalogue.Manifest{Module: "m", Version: "0.1", Build: &catalogue.Build{Artifacts: []catalogue.Artifact{goBundle}}}
res := &fakeReservations{}
r.b.Reserved = res
var said []builds.BuildLogEntry
r.b.Log = func(e builds.BuildLogEntry) { said = append(said, e) }
r.b.Module(context.Background(), "m", moduleValue(t, "modules/m"))
r.b.Wait()
if _, ok := r.st.vals[r.key]; ok || len(res.reserved) != 0 {
t.Fatalf("before versions-filled the build was claimed (%v) or reserved %v", ok, res.reserved)
}
var waits []string
for _, e := range said {
if e.Event == builds.BuildSaidEvent && e.Why == "wait-held" {
waits = append(waits, e.Line)
}
}
if len(waits) != 1 || !slices.ContainsFunc(waits, func(l string) bool { return strings.Contains(l, "versions-filled") }) {
t.Fatalf("wait-held lines = %q, want one naming versions-filled", waits)
}
res.filled = true
v := r.consider(t, "modules/m")
if v.State != builds.BuildDone || len(res.reserved) != 1 || !slices.Equal(res.reserved[0], []string{"m:tool"}) || res.series[0] != "0.1" {
t.Fatalf("build %+v reserved %v in %v, want done and one reservation of m:tool in 0.1", v.State, res.reserved, res.series)
}
if res.built["m:tool"] != digest || len(res.built) != 2 {
t.Errorf("recorded %v, want the digest of m:tool under the reserved version", res.built)
}
r.consider(t, "modules/m")
if len(res.reserved) != 1 {
t.Errorf("a done key reserved again: %v", res.reserved)
}
// A manifest whose version is no MAJOR.MINOR reserves nothing and is built as before.
r2 := newRig(t, goBundle)
res2 := &fakeReservations{filled: true}
r2.b.Reserved = res2
if v := r2.consider(t, "modules/m"); v.State != builds.BuildDone || len(res2.reserved) != 0 {
t.Errorf("a manifest with no MAJOR.MINOR: state %s, reserved %v", v.State, res2.reserved)
}
}
// TestADropOfAReservationIsAReasksValueOfItsOwnKey: design 47 G §369 — builder_reserved_drop asks the builder through
// reasks, under a key of the reservation (never a build key), refusing a module with no MAJOR.MINOR or no why.
func TestADropOfAReservationIsAReasksValueOfItsOwnKey(t *testing.T) {
at := time.Unix(0, 0)
key, data, err := NewReservedDrop("m", "1.4", "never released", "op@x", at)
if err != nil || key != "reserved."+versions.ReservedKey("m", "1.4") {
t.Fatalf("key %q (%v)", key, err)
}
if r, err := DecodeReAsk(data); err != nil || r.Mode != ModeReservedDrop || r.Name != "m" || r.FP != "1.4" || r.By != "op@x" {
t.Errorf("value = %+v (%v)", r, err)
}
for name, args := range map[string][3]string{"no module": {"", "1.4", "why"}, "no series": {"m", "1", "why"}, "three numbers": {"m", "1.4.2", "why"}, "no why": {"m", "1.4", " "}} {
if _, _, err := NewReservedDrop(args[0], args[1], args[2], "op", at); err == nil {
t.Errorf("%s: a drop was made", name)
}
}
}
@@ -95,6 +95,8 @@ type Builder struct {
// Numbers numbers each trunk build at its claim and records what it was built with (design 51 *The versions*); nil
// numbers nothing.
Numbers Numbers
// Reserved reserves the version of a trunk build at its start (design 47 G §369); nil reserves nothing.
Reserved Reservations
// Tags answers a check's build's tag from the kept tag asks; nil numbers no check's build.
Tags TagOf
// Forms answers what each node's engine states it reads (ADR 0618 §2, ADR 0624 §1): present, the nodes whose engine
@@ -103,16 +105,18 @@ type Builder struct {
// its single reference, and its needs. Nil: no node states anything, and the first forms are kept.
Forms func() (present, counted map[string][]string)
once sync.Once
sem chan struct{}
wg sync.WaitGroup
mu sync.Mutex
known map[string]meshcatalog.ModulesByModuleValue // by modules key
inflight map[string]bool // by build key
again map[string]meshcatalog.ModulesByModuleValue // by build key: the newest value that arrived while the key was being built
plans map[string]planned // by manifest digest
cands map[string]meshcatalog.CandidatesByGroupValue // by candidates key (candidates.go)
via map[string]via // by build key: a candidate's fetch (candidates.go)
once sync.Once
sem chan struct{}
wg sync.WaitGroup
mu sync.Mutex
// fillWaits are the builds that said wait-held for the fill (module.fingerprint), so it is said once.
fillWaits sync.Map
known map[string]meshcatalog.ModulesByModuleValue // by modules key
inflight map[string]bool // by build key
again map[string]meshcatalog.ModulesByModuleValue // by build key: the newest value that arrived while the key was being built
plans map[string]planned // by manifest digest
cands map[string]meshcatalog.CandidatesByGroupValue // by candidates key (candidates.go)
via map[string]via // by build key: a candidate's fetch (candidates.go)
// outOfStep are, by build key, the out-of-step files a build's lanes answered, until its end entry takes them.
outOfStep map[string][]string
trunkFP map[string]bool // by build key: a trunk module value carries this fingerprint
@@ -491,6 +495,10 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
return
}
}
reserving := trunk && tag == "" && planErr == nil && !nothing && b.reserves(ctx, module, m, plan)
if reserving && b.waitFill(module, m) {
return
}
// A key built again by hand is taken over as it stands too: one left building by a look that ended without its write
// (issue 1966) is the hand tool's to finish.
cl, err := b.Claimer.Claim(ctx, module, m.Fp2, m.Manifest, ClaimOptions{TakeOver: first || byHand, ByHand: byHand, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
@@ -525,6 +533,11 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
}
}
}
var reserved versions.ReservedVersion
hasReserved := false
if reserving && cl.Build.State != builds.BuildDone {
reserved, hasReserved = b.reserve(ctx, module, plan)
}
var result builds.BuilderBuild
if nothing {
// A manifest with no artifact is built by building nothing: its key is done with no artifact, claimed and written as
@@ -580,6 +593,15 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
b.logf("%s: %v", cl.Key, err)
}
}
if result.State == builds.BuildDone && hasReserved {
digests := map[string]string{}
for _, a := range result.Artifacts {
digests[module+":"+a.Name] = a.Digest
}
if err := b.Reserved.Built(ctx, module, reserved.Version, digests); err != nil {
b.logf("%s: the digests of %s are not recorded: %v", cl.Key, reserved.Version, err)
}
}
if result.State == builds.BuildFailed {
b.logf("%s: failed: %s", cl.Key, result.Error)
}
@@ -120,6 +120,27 @@ func NewOldFormDrop(node, why, by string, at time.Time) (string, []byte, error)
return "old-form." + node, data, err
}
// ModeReservedDrop is the Mode of a reasks value of builder_reserved_drop (design 47 G section 369): a module's reservation
// of one MAJOR.MINOR dropped by hand; Name is the module and FP the MAJOR.MINOR.
const ModeReservedDrop = "reserved-drop"
// NewReservedDrop is the value builder_reserved_drop writes: the key `reserved.<module>.<MAJOR.MINOR>` (never a build key)
// and the value.
func NewReservedDrop(module, series, why, by string, at time.Time) (string, []byte, error) {
module, series = strings.TrimSpace(module), strings.TrimSpace(series)
if module == "" {
return "", nil, errors.New("module is required")
}
if strings.Count(series, ".") != 1 {
return "", nil, fmt.Errorf("series %q is not MAJOR.MINOR", series)
}
if strings.TrimSpace(why) == "" {
return "", nil, errors.New("say why the reservation is dropped (why)")
}
data, err := json.Marshal(ReAsk{V: 1, Why: strings.TrimSpace(why), By: by, At: at.UTC(), Mode: ModeReservedDrop, Name: module, FP: series})
return "reserved." + versions.ReservedKey(module, series), data, err
}
// ModeBuild is the Mode of a reasks value that asks the key to be built again, not its waiting asks to be asked again.
const ModeBuild = "build"
@@ -217,6 +217,8 @@ type Acts struct {
Remove func(ctx context.Context, seat string, e Entry) (RemoveOutcome, error)
// DropOldForm drops a retired node from the old form's count by hand (builder_old_form): its answer, or why not.
DropOldForm func(ctx context.Context, node, by, why string, at time.Time) (string, error)
// DropReserved drops a module's reservation of one MAJOR.MINOR by hand (builder_reserved_drop): its answer, or why not.
DropReserved func(ctx context.Context, module, series, by, why string, at time.Time) (string, error)
}
// TagAsked answers one unanswered value of tag-asks. An untag calls remove on each of the branch's tagged versions'
@@ -0,0 +1,450 @@
package versions
import (
"context"
"encoding/json"
"errors"
"fmt"
"sort"
"strconv"
"strings"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
)
// **The reservation is one key per module and MAJOR.MINOR** (design 47 G §369, §52): the builder keeps in its old state
// section `reserved` the version its build reserved and each artifact's digest once built, written by compare-and-set at
// the build's start. A newer build of the module takes the next after the higher of the highest released and the one
// reserved, so two builds never share a version, and a superseded build's version is never released (a gap in the
// patches). The key is deleted when the catalog says `version-released` for it, or when a version of another MAJOR.MINOR
// of the module is released (the manifest moved on). This is no list of released versions: numbering reads those only from
// the catalog (§348), handed to Reserve by the caller.
const (
// ReservedState is the builder's old state section of the reservations.
ReservedState = "reserved"
// ReservedBucket is its bucket.
ReservedBucket = "mesh-delivery-builder_" + ReservedState
// ReservedTool is builder_reserved, ReservedDropTool builder_reserved_drop.
ReservedTool = "builder_reserved"
ReservedDropTool = "builder_reserved_drop"
// UnreleasedBound is the publish's bound: past it a built reservation with an artifact unpublished is the finding
// version-unreleased (§369).
UnreleasedBound = 10 * time.Minute
// VersionUnreleased is that finding.
VersionUnreleased = "version-unreleased"
)
// ReservedVersion is the value of a reservation key.
type ReservedVersion struct {
V int `json:"v"`
Module string `json:"module"`
Series string `json:"series"` // MAJOR.MINOR
Version string `json:"version"` // MAJOR.MINOR.PATCH
// Artifacts are the module's artifacts at the version: the digest once built, empty before.
Artifacts map[string]string `json:"artifacts"`
// BuiltAt is when the build ended with every digest known; the publish's bound counts from it.
BuiltAt time.Time `json:"built-at,omitzero"`
// Published are the artifacts a registry seat said published at this version with the built digest.
Published map[string]bool `json:"published,omitempty"`
At time.Time `json:"at"`
}
// ReservedKey is the key of module's reservation of series (MAJOR.MINOR).
func ReservedKey(module, series string) string { return Key(module, series) }
// ErrNotFilled is what Reserve answers before the catalog says versions-filled: nothing is reserved until then (§369).
var ErrNotFilled = errors.New("versions-filled is not heard: nothing is reserved before the catalog's versions are filled")
// ReserveStore is the reservations' bucket: compare-and-set, and deletion.
type ReserveStore interface {
Store
Delete(ctx context.Context, key string) error
}
// Reserver reserves versions.
type Reserver struct {
Store ReserveStore
// ReleasedOf are the patch versions (MAJOR.MINOR.PATCH) the catalog holds released for module, from its verb versions.
ReleasedOf func(ctx context.Context, module string) ([]string, error)
// Filled says the catalog said versions-filled.
Filled func() bool
Now func() time.Time
}
func (r *Reserver) now() time.Time {
if r.Now != nil {
return r.Now().UTC()
}
return time.Now().UTC()
}
// SplitVersion is a MAJOR.MINOR.PATCH version's series and patch.
func SplitVersion(version string) (series string, patch int, err error) {
i := strings.LastIndex(version, ".")
if i < 0 || strings.Count(version, ".") != 2 {
return "", 0, fmt.Errorf("%q is not MAJOR.MINOR.PATCH", version)
}
patch, err = strconv.Atoi(version[i+1:])
if err != nil || patch < 0 {
return "", 0, fmt.Errorf("%q is not MAJOR.MINOR.PATCH", version)
}
return version[:i], patch, nil
}
// Reserve reserves the next version of module's series for a build that starts, for artifacts (their names), and answers
// it. The next patch is the one after the higher of the highest released and the one reserved, 0 when there is neither. A
// conflicting write (another build reserved first) is read again and the next taken.
func (r *Reserver) Reserve(ctx context.Context, module, series string, artifacts []string) (ReservedVersion, error) {
if r.Filled != nil && !r.Filled() {
return ReservedVersion{}, ErrNotFilled
}
released, err := r.ReleasedOf(ctx, module)
if err != nil {
return ReservedVersion{}, fmt.Errorf("the released versions of %s cannot be read: %w", module, err)
}
high := -1
for _, v := range released {
if s, p, err := SplitVersion(v); err == nil && s == series && p > high {
high = p
}
}
key := ReservedKey(module, series)
for {
cur, rev, found, err := r.get(ctx, key)
if err != nil {
return ReservedVersion{}, err
}
next := high
if found {
if _, p, err := SplitVersion(cur.Version); err == nil && p > next {
next = p
}
}
out := ReservedVersion{V: 1, Module: module, Series: series, Version: fmt.Sprintf("%s.%d", series, next+1), Artifacts: map[string]string{}, At: r.now()}
for _, a := range artifacts {
out.Artifacts[a] = ""
}
raw, _ := json.Marshal(out)
if found {
_, err = r.Store.Update(ctx, key, raw, rev)
} else {
_, err = r.Store.Create(ctx, key, raw)
}
if errors.Is(err, ErrConflict) {
continue
}
if err != nil {
return ReservedVersion{}, fmt.Errorf("%s cannot be reserved: %w", key, err)
}
return out, nil
}
}
func (r *Reserver) get(ctx context.Context, key string) (ReservedVersion, uint64, bool, error) {
raw, rev, found, err := r.Store.Get(ctx, key)
if err != nil || !found {
return ReservedVersion{}, 0, false, err
}
var v ReservedVersion
if err := json.Unmarshal(raw, &v); err != nil {
return ReservedVersion{}, 0, false, fmt.Errorf("%s holds what is not a reservation: %w", key, err)
}
return v, rev, true, nil
}
// Built records artifact's digest under the reserved version of module's series; once every artifact has one the build is
// ended (BuiltAt). A reservation moved on to another version (a newer build took the next) is left alone and answers false.
func (r *Reserver) Built(ctx context.Context, module, version, artifact, digest string) (bool, error) {
series, _, err := SplitVersion(version)
if err != nil {
return false, err
}
key := ReservedKey(module, series)
for {
cur, rev, found, err := r.get(ctx, key)
if err != nil || !found || cur.Version != version {
return false, err
}
cur.Artifacts[artifact] = digest
if cur.BuiltAt.IsZero() && allSet(cur.Artifacts) {
cur.BuiltAt = r.now()
}
raw, _ := json.Marshal(cur)
if _, err := r.Store.Update(ctx, key, raw, rev); errors.Is(err, ErrConflict) {
continue
} else if err != nil {
return false, fmt.Errorf("%s cannot be written: %w", key, err)
}
return true, nil
}
}
func allSet(m map[string]string) bool {
for _, d := range m {
if d == "" {
return false
}
}
return true
}
// Released takes the catalog's version-released in: the key of that version is deleted, and every key of module's other
// series (the manifest moved on). A key holding a newer reserved version of the same series stays. It answers the keys
// deleted.
func (r *Reserver) Released(ctx context.Context, module, version string) ([]string, error) {
series, _, err := SplitVersion(version)
if err != nil {
return nil, err
}
keys, err := r.Store.KeysOf(ctx, NameFilter(module))
if err != nil {
return nil, fmt.Errorf("the reservations of %s cannot be listed: %w", module, err)
}
var gone []string
for _, k := range keys {
cur, _, found, err := r.get(ctx, k)
if err != nil {
return gone, err
}
if !found || (cur.Series == series && cur.Version != version) {
continue
}
if err := r.Store.Delete(ctx, k); err != nil {
return gone, fmt.Errorf("%s cannot be deleted: %w", k, err)
}
gone = append(gone, k)
}
sort.Strings(gone)
return gone, nil
}
// Unreleased are the reservations whose build ended more than UnreleasedBound before now with an artifact unpublished:
// the finding version-unreleased, naming the module, the version and the artifacts missing (published says whether an
// artifact is). A reservation whose build has not ended is not one.
type Unreleased struct {
Module, Version string
Missing []string
}
// Overdue answers the reservations past the bound with an artifact not published.
func (r *Reserver) Overdue(ctx context.Context, published func(module, version, artifact string) bool) ([]Unreleased, error) {
keys, err := r.Store.KeysOf(ctx, AllFilter)
if err != nil {
return nil, fmt.Errorf("the reservations cannot be listed: %w", err)
}
sort.Strings(keys)
var out []Unreleased
for _, k := range keys {
cur, _, found, err := r.get(ctx, k)
if err != nil {
return nil, err
}
if !found || cur.BuiltAt.IsZero() || r.now().Sub(cur.BuiltAt) < UnreleasedBound {
continue
}
var missing []string
for _, a := range cur.Missing() {
if published == nil || !published(cur.Module, cur.Version, a) {
missing = append(missing, a)
}
}
if len(missing) > 0 {
out = append(out, Unreleased{Module: cur.Module, Version: cur.Version, Missing: missing})
}
}
return out, nil
}
// All are the reservations, sorted by key: builder_reserved.
func (r *Reserver) All(ctx context.Context) ([]ReservedVersion, error) {
keys, err := r.Store.KeysOf(ctx, AllFilter)
if err != nil {
return nil, fmt.Errorf("the reservations cannot be listed: %w", err)
}
sort.Strings(keys)
var out []ReservedVersion
for _, k := range keys {
cur, _, found, err := r.get(ctx, k)
if err != nil {
return nil, err
}
if found {
out = append(out, cur)
}
}
return out, nil
}
// Drop deletes module's reservation of series on a person's word (builder_reserved_drop); the caller writes the hand-act
// log. It answers whether there was one.
func (r *Reserver) Drop(ctx context.Context, module, series string) (bool, error) {
key := ReservedKey(module, series)
_, _, found, err := r.get(ctx, key)
if err != nil || !found {
return false, err
}
return true, r.Store.Delete(ctx, key)
}
// Published takes a registry seat's published in: the artifacts of a reservation at version (a leading v is the Go form)
// whose built digest it names are marked published. It answers the keys it marked.
func (r *Reserver) Published(ctx context.Context, version, digest string) ([]string, error) {
version = strings.TrimPrefix(version, "v")
if version == "" || digest == "" {
return nil, nil
}
keys, err := r.Store.KeysOf(ctx, AllFilter)
if err != nil {
return nil, fmt.Errorf("the reservations cannot be listed: %w", err)
}
sort.Strings(keys)
var marked []string
for _, k := range keys {
again:
cur, rev, found, err := r.get(ctx, k)
if err != nil {
return marked, err
}
if !found || cur.Version != version {
continue
}
changed := false
for a, d := range cur.Artifacts {
if d == digest && !cur.Published[a] {
if cur.Published == nil {
cur.Published = map[string]bool{}
}
cur.Published[a], changed = true, true
}
}
if !changed {
continue
}
raw, _ := json.Marshal(cur)
if _, err := r.Store.Update(ctx, k, raw, rev); errors.Is(err, ErrConflict) {
goto again
} else if err != nil {
return marked, fmt.Errorf("%s cannot be written: %w", k, err)
}
marked = append(marked, k)
}
return marked, nil
}
// Of is module's reservation of series, and whether there is one; with an empty series the module's highest.
func (r *Reserver) Of(ctx context.Context, module, version string) (ReservedVersion, bool, error) {
all, err := r.All(ctx)
if err != nil {
return ReservedVersion{}, false, err
}
var out ReservedVersion
found := false
for _, v := range all {
if v.Module != module || (version != "" && v.Version != version) {
continue
}
if !found {
out, found = v, true
continue
}
_, p, _ := SplitVersion(v.Version)
_, q, _ := SplitVersion(out.Version)
if v.Series > out.Series || (v.Series == out.Series && p > q) {
out = v
}
}
return out, found, nil
}
const reservedBoundPrefix = "reserved."
// ReservedBoundKey is the bound of a reservation's publish in the state bounds (key-ttl): `reserved.<reserved key>`, armed
// when its build ended with every digest known. Its end while the reservation stands is the finding version-unreleased.
func ReservedBoundKey(module, series string) string {
return ReservedBoundFor(ReservedKey(module, series))
}
// ReservedBoundFor is the bound of the reservation under key.
func ReservedBoundFor(key string) string { return reservedBoundPrefix + key }
// ReservedBoundOf is the reserved key a reservation's bound names, and whether key is one.
func ReservedBoundOf(key string) (string, bool) { return strings.CutPrefix(key, reservedBoundPrefix) }
// readOnlyReserve is the reservations read by the tools bundle, which writes nothing.
type readOnlyReserve struct{ readStore }
func (readOnlyReserve) Delete(context.Context, string) error { return errReadOnly }
// AnswerReserved is builder_reserved: every reservation, or one module's.
func AnswerReserved(r Reader, module string) (map[string]any, error) {
rs := &Reserver{Store: readOnlyReserve{readStore{r}}}
all, err := rs.All(context.Background())
if err != nil {
return nil, err
}
out := []ReservedVersion{}
for _, v := range all {
if module == "" || v.Module == module {
out = append(out, v)
}
}
return map[string]any{"reserved": out}, nil
}
// AnswerReservedVerb is the seat mesh-builds' verb reserved: module's reserved version (the given one, else the highest)
// with each artifact's digest, an artifact not built yet with none.
func AnswerReservedVerb(r Reader, module, version string) (meshbuilds.ReservedAnswer, error) {
module = strings.TrimSpace(module)
if module == "" {
return meshbuilds.ReservedAnswer{}, errors.New("module is required")
}
rs := &Reserver{Store: readOnlyReserve{readStore{r}}}
v, found, err := rs.Of(context.Background(), module, strings.TrimSpace(version))
if err != nil {
return meshbuilds.ReservedAnswer{}, err
}
if !found {
return meshbuilds.ReservedAnswer{}, fmt.Errorf("%s has no reserved version%s", module, map[bool]string{true: " " + version, false: ""}[version != ""])
}
out := meshbuilds.ReservedAnswer{Module: v.Module, Version: v.Version, Artifacts: []meshbuilds.ReservedAnswerArtifactsItem{}}
names := make([]string, 0, len(v.Artifacts))
for a := range v.Artifacts {
names = append(names, a)
}
sort.Strings(names)
for _, a := range names {
out.Artifacts = append(out.Artifacts, meshbuilds.ReservedAnswerArtifactsItem{Artifact: a, Digest: v.Artifacts[a]})
}
return out, nil
}
// The descriptions of the reservation's tools.
const (
ReservedDescription = "The versions the builder reserved (its state reserved, design 47 G section 369): one per module and MAJOR.MINOR, " +
"the version its build reserved at its start, each artifact's digest once built, when the build ended and which artifacts a registry " +
"seat said published. With module, that module's only. A key goes when the catalog says version-released for it. Read-only."
ReservedDropDescription = "Drop a module's reservation of one MAJOR.MINOR by hand (design 47 G section 369), when its version will never be " +
"released; the next build takes the next after the highest released. A hand act: it is recorded in the hand-act log."
// ReservedVerb is the seat mesh-builds' verb served as a tool of the same name.
ReservedVerb = meshbuilds.VerbReserved
)
// Missing are the artifacts of v no registry seat said published, sorted.
func (v ReservedVersion) Missing() []string {
var out []string
for a := range v.Artifacts {
if !v.Published[a] {
out = append(out, a)
}
}
sort.Strings(out)
return out
}
// At is the reservation under key, and whether there is one.
func (r *Reserver) At(ctx context.Context, key string) (ReservedVersion, bool, error) {
v, _, found, err := r.get(ctx, key)
return v, found, err
}
@@ -0,0 +1,158 @@
package versions
import (
"encoding/json"
"context"
"errors"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
"reflect"
"testing"
"time"
)
type reserveMem struct {
*mem
onUpdate func() // runs once before the next Update
}
func (m *reserveMem) Delete(_ context.Context, key string) error {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.vals, key)
delete(m.revs, key)
return nil
}
func (m *reserveMem) Update(ctx context.Context, key string, value []byte, rev uint64) (uint64, error) {
if f := m.onUpdate; f != nil {
m.onUpdate = nil
f()
}
return m.mem.Update(ctx, key, value, rev)
}
// TestReservation: design 47 G §369 (test 102): two builds close together reserve two versions, the key goes on
// version-released, a built reservation with an artifact unpublished past the bound is unreleased, and nothing is reserved
// before versions-filled.
func TestReservation(t *testing.T) {
ctx := context.Background()
now := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC)
released := []string{"0.1.18", "0.1.19", "0.2.7"}
filled := true
store := &reserveMem{mem: newMem()}
r := &Reserver{Store: store, Filled: func() bool { return filled }, Now: func() time.Time { return now },
ReleasedOf: func(context.Context, string) ([]string, error) { return released, nil }}
arts := []string{"m:a", "m:b"}
filled = false
if _, err := r.Reserve(ctx, "m", "0.1", arts); !errors.Is(err, ErrNotFilled) {
t.Fatalf("before versions-filled: %v, want ErrNotFilled", err)
}
if got, _ := r.All(ctx); len(got) != 0 {
t.Fatalf("a reservation was written before the fill: %v", got)
}
filled = true
first, err := r.Reserve(ctx, "m", "0.1", arts)
if err != nil || first.Version != "0.1.20" {
t.Fatalf("first build reserves %q (%v), want 0.1.20 after the highest released 0.1.19", first.Version, err)
}
second, err := r.Reserve(ctx, "m", "0.1", arts)
if err != nil || second.Version != "0.1.21" {
t.Fatalf("second build reserves %q (%v), want 0.1.21: never the first's", second.Version, err)
}
// Two builds at once: the rival reserves between this one's read and write, and this one takes the next.
store.onUpdate = func() {
if _, err := r.Reserve(ctx, "m", "0.1", arts); err != nil {
t.Error(err)
}
}
third, err := r.Reserve(ctx, "m", "0.1", arts)
if err != nil || third.Version != "0.1.23" {
t.Fatalf("a build that lost the compare-and-set reserves %q (%v), want 0.1.23", third.Version, err)
}
if v, _ := r.Reserve(ctx, "m", "0.3", arts); v.Version != "0.3.0" {
t.Errorf("a series with nothing released reserves %q, want 0.3.0", v.Version)
}
// The artifacts' digests, then the bound.
for _, a := range arts {
if ok, err := r.Built(ctx, "m", "0.1.23", a, "sha256:"+a); err != nil || !ok {
t.Fatalf("built %s: %v %v", a, ok, err)
}
}
if ok, _ := r.Built(ctx, "m", "0.1.22", "m:a", "x"); ok {
t.Error("a digest was written under a version a newer build superseded")
}
published := func(_, _, a string) bool { return a == "m:a" }
if got, _ := r.Overdue(ctx, published); len(got) != 0 {
t.Fatalf("unreleased before the bound: %v", got)
}
now = now.Add(UnreleasedBound)
got, err := r.Overdue(ctx, published)
want := []Unreleased{{Module: "m", Version: "0.1.23", Missing: []string{"m:b"}}}
if err != nil || !reflect.DeepEqual(got, want) {
t.Fatalf("past the bound: %v (%v), want %v", got, err, want)
}
// version-released: its key goes, and the other series' keys of the module (the manifest moved on).
if gone, err := r.Released(ctx, "m", "0.1.23"); err != nil || len(gone) != 2 {
t.Fatalf("released 0.1.23 deleted %v (%v), want the 0.1 key and the 0.3 key", gone, err)
}
if all, _ := r.All(ctx); len(all) != 0 {
t.Errorf("reservations left: %v", all)
}
if again, _ := r.Reserve(ctx, "m", "0.1", arts); again.Version != "0.1.20" {
t.Logf("with the catalog list unchanged the next is %s", again.Version)
}
if ok, _ := r.Drop(ctx, "m", "0.1"); !ok {
t.Error("builder_reserved_drop found no reservation")
}
}
// TestReservationIsReadAndMarkedPublished: design 47 G §369 — the reservation is read by builder_reserved and by the seat's
// verb reserved with each artifact's digest, a registry seat's published marks the artifacts of that version and digest, and
// the missing artifacts it names are the unpublished ones.
func TestReservationIsReadAndMarkedPublished(t *testing.T) {
ctx := context.Background()
store := &reserveMem{mem: newMem()}
r := &Reserver{Store: store, ReleasedOf: func(context.Context, string) ([]string, error) { return []string{"1.4.2"}, nil }}
v, err := r.Reserve(ctx, "m", "1.4", []string{"m:a", "m:b"})
if err != nil || v.Version != "1.4.3" {
t.Fatalf("reserved %q (%v), want 1.4.3", v.Version, err)
}
if _, err := r.Built(ctx, "m", "1.4.3", "m:a", "sha256:aa"); err != nil {
t.Fatal(err)
}
rd := Reader{
Keys: func() ([]string, error) { return store.KeysOf(ctx, AllFilter) },
Get: func(key string) (json.RawMessage, bool, error) {
raw, _, ok, err := store.Get(ctx, key)
return raw, ok, err
}}
ans, err := AnswerReservedVerb(rd, "m", "")
want := meshbuilds.ReservedAnswer{Module: "m", Version: "1.4.3", Artifacts: []meshbuilds.ReservedAnswerArtifactsItem{{Artifact: "m:a", Digest: "sha256:aa"}, {Artifact: "m:b"}}}
if err != nil || !reflect.DeepEqual(ans, want) {
t.Fatalf("reserved answers %+v (%v), want %+v", ans, err, want)
}
if _, err := AnswerReservedVerb(rd, "m", "1.4.9"); err == nil {
t.Error("a version never reserved was answered")
}
if got, err := AnswerReserved(rd, "other"); err != nil || len(got["reserved"].([]ReservedVersion)) != 0 {
t.Errorf("builder_reserved of another module = %v (%v), want none", got, err)
}
if marked, err := r.Published(ctx, "v1.4.3", "sha256:aa"); err != nil || len(marked) != 1 {
t.Fatalf("a published of 1.4.3 with a built digest marked %v (%v)", marked, err)
}
if marked, _ := r.Published(ctx, "1.4.3", "sha256:other"); len(marked) != 0 {
t.Errorf("a published with another digest marked %v", marked)
}
cur, _, _ := r.At(ctx, ReservedKey("m", "1.4"))
if got := cur.Missing(); !reflect.DeepEqual(got, []string{"m:b"}) {
t.Errorf("missing = %v, want m:b", got)
}
if key, ok := ReservedBoundOf(ReservedBoundKey("m", "1.4")); !ok || key != ReservedKey("m", "1.4") || ReservedBoundFor(key) != ReservedBoundKey("m", "1.4") {
t.Errorf("the bound key does not name the reservation back: %q %v", key, ok)
}
}
+17 -8
View File
@@ -38,9 +38,14 @@ type artifact struct {
}
type manifest struct {
Module string `json:"module"`
Version string `json:"version"`
Slug string `json:"slug"`
Module string `json:"module"`
Version string `json:"version"`
Slug string `json:"slug"`
Claims []struct {
Name string `json:"name"`
Scope string `json:"scope"`
Serves []string `json:"serves"`
} `json:"claims"`
Tools []string `json:"tools"`
Emits []string `json:"emits"`
Consumes []string `json:"consumes"`
@@ -97,7 +102,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
if m.Module != "mesh-delivery-builder" {
t.Errorf("module = %q", m.Module)
}
if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.BuilderBuildsHistory
if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reserved", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // builds.BuilderBuildsHistory
t.Errorf("state = %+v, want %+v", m.State, want)
}
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
@@ -127,7 +132,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
// The watchdogs seat's presence says a node's machine is heard (issue 1966, ADR 0596 §2, ADR 0668 §3).
if want := []string{"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-catalog.versions", "seat:mesh-hand-acts.hand-act", "seat:mesh-watchdogs.presence", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
"seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
}
@@ -139,7 +144,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
// DLV-M1: the indexer's and the assigner's views families of the states it follows.
if want := []string{"container-image-registry.kept", "container-image-registry.published", "container-image-registry.refused", "container-image-registry.removed",
"go-package-registry.kept", "go-package-registry.published", "go-package-registry.refused", "go-package-registry.removed",
"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-catalog.manifest-accepted",
"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-catalog.manifest-accepted", "mesh-catalog.version-released", "mesh-catalog.versions-filled",
"mesh-conditions.condition-cleared", "mesh-delivery-assigner.views.assignments", "mesh-delivery-checker.tag-wanted", "mesh-delivery-forge-bridge.forge.**",
"mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules",
"mesh-delivery-indexer.views.packages", "node-engine.manifest-reads-changed",
@@ -155,7 +160,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
}
// 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again.
if want := []string{"builder_announce_again", "builder_ask_again", "builder_bounds", "builder_build_again", "builder_builds_since", "builder_engine_reads", "builder_first_view_again", "builder_first_views",
"builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_seat_queue", "builder_stuck", "builder_tag", "builder_tag_asks", "builder_untag", "builder_versions"}; !reflect.DeepEqual(sorted(m.Tools), want) {
"builder_floor_again", "builder_floor_waits", "builder_old_form", "builder_publish_again", "builder_reserved", "builder_reserved_drop", "builder_seat_queue", "builder_stuck", "builder_tag", "builder_tag_asks", "builder_untag", "builder_versions", "reserved"}; !reflect.DeepEqual(sorted(m.Tools), want) {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
// Design 51 *Events of the builder*: numbered and old-form-closed on the family version, apart from the builds' ends.
@@ -166,7 +171,11 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
if len(m.Logs) != 1 || m.Logs[0].Name != builds.BuilderLog || m.Logs[0].MaxAgeDays != builds.BuilderLogDays || m.Logs[0].MaxMiB != builds.BuilderLogMiB {
t.Errorf("logs = %+v, want %s kept %d days in %d MiB", m.Logs, builds.BuilderLog, builds.BuilderLogDays, builds.BuilderLogMiB)
}
for _, k := range []string{"streams", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} {
// Design 47 G §52, §369: the seat mesh-builds, held once for the mesh, serving reserved.
if len(m.Claims) != 1 || m.Claims[0].Name != "mesh-builds" || m.Claims[0].Scope != "mesh" || !reflect.DeepEqual(m.Claims[0].Serves, []string{"reserved"}) {
t.Errorf("claims = %+v, want mesh-builds, scope mesh, serving reserved", m.Claims)
}
for _, k := range []string{"streams", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} {
if _, ok := top[k]; ok {
t.Errorf("the manifest declares %q: the builder has none", k)
}
+16
View File
@@ -2,6 +2,15 @@
"module": "mesh-delivery-builder",
"version": "1",
"slug": "builder",
"claims": [
{
"name": "mesh-builds",
"scope": "mesh",
"serves": [
"reserved"
]
}
],
"tools": [
"builder_announce_again",
"builder_ask_again",
@@ -13,12 +22,15 @@
"builder_floor_waits",
"builder_old_form",
"builder_publish_again",
"builder_reserved",
"builder_reserved_drop",
"builder_seat_queue",
"builder_stuck",
"builder_tag",
"builder_tag_asks",
"builder_untag",
"builder_versions",
"reserved",
"builder_first_views",
"builder_first_view_again"
],
@@ -40,6 +52,8 @@
"mesh-build-go.toolchain.reported",
"mesh-build-node.toolchain.reported",
"mesh-catalog.manifest-accepted",
"mesh-catalog.version-released",
"mesh-catalog.versions-filled",
"mesh-conditions.condition-cleared",
"mesh-delivery-checker.tag-wanted",
"mesh-delivery-forge-bridge.forge.**",
@@ -66,6 +80,7 @@
"seat:go-package-registry.remove",
"seat:go-package-registry.versions",
"seat:mesh-artifact-sweep.held",
"seat:mesh-catalog.versions",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-watchdogs.presence",
"seat:node-build-go.drop",
@@ -92,6 +107,7 @@
{ "name": "old-form", "history": 1 },
{ "name": "bounds", "history": 1, "key-ttl": true },
{ "name": "queues", "history": 1 },
{ "name": "reserved", "history": 1 },
{ "name": "reasks", "history": 1 },
{ "name": "tag-ask-bounds", "history": 1, "key-ttl": true },
{ "name": "tag-asks", "history": 1 },
@@ -5,6 +5,7 @@ import (
"errors"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-sdk/go/keys"
@@ -191,4 +192,37 @@ func TestForge(t *testing.T) {
}
}
})
// Design 47 G §164: an owed act's outcome names its seat, its id, one outcome and words for every outcome but
// performed; an act the bridge cannot decode may name no id.
t.Run("owed ended", func(t *testing.T) {
at := time.Date(2026, 11, 9, 20, 0, 0, 0, time.UTC)
ok := OwedEnded{Of: "mesh-delivery", ID: "g1.view.1", Outcome: OwedPerformed, At: at}
raw, err := EncodeOwedEnded(ok)
if back, derr := DecodeOwedEnded(raw); err != nil || derr != nil || back.V != Version || back.ID != ok.ID {
t.Fatalf("round trip: %s %v %v", raw, err, derr)
}
if _, err := DecodeOwedEnded([]byte(`{"v":99,"of":"x","id":"a","outcome":"performed","at":"2026-11-09T20:00:00Z"}`)); err == nil {
t.Errorf("an unknown version was read")
}
for _, c := range []struct {
name string
edit func(*OwedEnded)
want string
}{
{"no seat", func(e *OwedEnded) { e.Of = "" }, "no seat"},
{"an outcome of no kind", func(e *OwedEnded) { e.Outcome = "done" }, "none of"},
{"no id", func(e *OwedEnded) { e.ID = "" }, "no id"},
{"a refusal with no words", func(e *OwedEnded) { e.Outcome = OwedRefused }, "no words"},
{"dropped with no reason", func(e *OwedEnded) { e.Outcome = OwedDropped }, "no words"},
{"no time", func(e *OwedEnded) { e.At = time.Time{} }, "no time"},
{"undecodable with no id is said", func(e *OwedEnded) { e.ID, e.Outcome, e.Words = "", OwedUndecodable, "not json" }, ""},
} {
e := ok
c.edit(&e)
_, err := EncodeOwedEnded(e)
if (c.want == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.want)) {
t.Errorf("%s: %v, want %q", c.name, err, c.want)
}
}
})
}
@@ -0,0 +1,76 @@
package forgemsg
import (
"encoding/json"
"fmt"
"slices"
"time"
)
// OwedEndedEvent is the name of an owed act's outcome on the forge bridge's stream `forge` (design 47 G §164): the
// bridge says it for each act the gate or the checker said as the event `owed`. Its two parts are no repository's
// forge event (`forge.<owner>.<name>.<kind>`).
const OwedEndedEvent = "owed.ended"
// The outcomes of an owed act (design 47 G §164, §168): performed, refused (the forge's own words), could not be
// performed (the forge did not answer within the call's bound, with the error), undecodable (an act the bridge cannot
// read, with why) and, from the tool `bridge_owed_drop`, dropped.
const (
OwedPerformed = "performed"
OwedRefused = "refused"
OwedCouldNotBePerformed = "could not be performed"
OwedUndecodable = "undecodable"
OwedDropped = "dropped"
)
// OwedOutcomes are the outcomes an owed.ended says, one of them.
var OwedOutcomes = []string{OwedPerformed, OwedRefused, OwedCouldNotBePerformed, OwedUndecodable, OwedDropped}
// OwedEnded is an owed act's outcome (design 47 G §164): the act's id and one outcome, with the words of any outcome
// but performed. Of is the seat that said the act (mesh-delivery for the gate's, mesh-delivery-checker for the
// checker's), so each hears its own ids (§164). An act the bridge cannot decode may name no id.
type OwedEnded struct {
V int `json:"v"`
Of string `json:"of"`
ID string `json:"id,omitempty"`
Outcome string `json:"outcome"`
Words string `json:"words,omitempty"`
At time.Time `json:"at"`
}
func (e OwedEnded) check() error {
switch {
case e.Of == "":
return fmt.Errorf("an owed act's outcome names no seat that said it")
case !slices.Contains(OwedOutcomes, e.Outcome):
return fmt.Errorf("the outcome %q of an owed act is none of %v", e.Outcome, OwedOutcomes)
case e.ID == "" && e.Outcome != OwedUndecodable:
return fmt.Errorf("the outcome %s of an owed act names no id", e.Outcome)
case e.Outcome != OwedPerformed && e.Words == "":
return fmt.Errorf("the outcome %s of the owed act %s says no words", e.Outcome, e.ID)
case e.At.IsZero():
return fmt.Errorf("the outcome of the owed act %s says no time", e.ID)
}
return nil
}
// EncodeOwedEnded is e as JSON at Version; it refuses what check refuses.
func EncodeOwedEnded(e OwedEnded) ([]byte, error) {
if err := e.check(); err != nil {
return nil, err
}
e.V = Version
return json.Marshal(e)
}
// DecodeOwedEnded reads an owed act's outcome; it refuses an unknown version and what EncodeOwedEnded refuses.
func DecodeOwedEnded(data []byte) (OwedEnded, error) {
var e OwedEnded
if err := decode(data, &e); err != nil {
return OwedEnded{}, err
}
if err := e.check(); err != nil {
return OwedEnded{}, err
}
return e, nil
}