From 74da8116db492e273e9f1b20a5c130ee03e88345 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:44:47 +0100 Subject: [PATCH 1/3] BL2: the build reserves its version at its start; reserved state, tools, verb and findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The builder keeps one key per module and MAJOR.MINOR in its state reserved, written by compare-and-set when a trunk build is claimed; nothing is reserved, and the build says wait-held, before the catalog says versions-filled; the key goes on version-released; a built reservation unpublished past ten minutes is the finding version-unreleased; builder_reserved, builder_reserved_drop (hand act, recorded) and the seat mesh-builds' verb reserved. Builds: design 47 G §369, §372, §52, §85 (reserved) Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-builder-tools/main.go | 9 + .../mesh-delivery-builder-tools/reserved.go | 62 +++ .../mesh-delivery-builder/integration_test.go | 2 +- .../cmd/mesh-delivery-builder/main.go | 16 +- .../cmd/mesh-delivery-builder/reasks.go | 8 +- .../cmd/mesh-delivery-builder/reserve.go | 208 ++++++++ .../cmd/mesh-delivery-builder/reserve_test.go | 131 +++++ .../cmd/mesh-delivery-builder/seats.go | 5 + .../cmd/mesh-delivery-builder/versions.go | 9 +- .../internal/build/numbers.go | 57 +++ .../internal/build/plan.go | 8 +- .../internal/build/reserve_test.go | 104 ++++ .../internal/build/run.go | 42 +- .../internal/build/stuck.go | 21 + .../internal/versions/again.go | 2 + .../internal/versions/reserve.go | 450 ++++++++++++++++++ .../internal/versions/reserve_test.go | 158 ++++++ .../mesh-delivery-builder/manifest_test.go | 25 +- modules/mesh-delivery-builder/module.json | 16 + 19 files changed, 1306 insertions(+), 27 deletions(-) create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/reserved.go create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve.go create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve_test.go create mode 100644 modules/mesh-delivery-builder/internal/build/reserve_test.go create mode 100644 modules/mesh-delivery-builder/internal/versions/reserve.go create mode 100644 modules/mesh-delivery-builder/internal/versions/reserve_test.go diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index 002ecd3d..1411f36c 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -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{}}, diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/reserved.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/reserved.go new file mode 100644 index 00000000..8e3d342c --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/reserved.go @@ -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 + }}, + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index 9c949810..6d270415 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -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 { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 5d40a676..659e7caa 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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 diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go index 8d7c8d3c..bbcba722 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reasks.go @@ -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") diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve.go new file mode 100644 index 00000000..bb41bbd8 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve.go @@ -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 +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve_test.go new file mode 100644 index 00000000..d1a04e2f --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve_test.go @@ -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") + } +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go index d6029f81..e3bff8b8 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go @@ -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) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go index 40d50d4e..8fa2946d 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go @@ -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 } diff --git a/modules/mesh-delivery-builder/internal/build/numbers.go b/modules/mesh-delivery-builder/internal/build/numbers.go index 6cf6250f..f1b9cef5 100644 --- a/modules/mesh-delivery-builder/internal/build/numbers.go +++ b/modules/mesh-delivery-builder/internal/build/numbers.go @@ -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, :). + 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 +} diff --git a/modules/mesh-delivery-builder/internal/build/plan.go b/modules/mesh-delivery-builder/internal/build/plan.go index 11c678d5..bb9b1999 100644 --- a/modules/mesh-delivery-builder/internal/build/plan.go +++ b/modules/mesh-delivery-builder/internal/build/plan.go @@ -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: diff --git a/modules/mesh-delivery-builder/internal/build/reserve_test.go b/modules/mesh-delivery-builder/internal/build/reserve_test.go new file mode 100644 index 00000000..cc1fc2dd --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/reserve_test.go @@ -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) + } + } +} diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index a7711f09..f5f8bf3a 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -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) } diff --git a/modules/mesh-delivery-builder/internal/build/stuck.go b/modules/mesh-delivery-builder/internal/build/stuck.go index e5ef2e86..77860c0f 100644 --- a/modules/mesh-delivery-builder/internal/build/stuck.go +++ b/modules/mesh-delivery-builder/internal/build/stuck.go @@ -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..` (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" diff --git a/modules/mesh-delivery-builder/internal/versions/again.go b/modules/mesh-delivery-builder/internal/versions/again.go index 9e77235b..2f3e814d 100644 --- a/modules/mesh-delivery-builder/internal/versions/again.go +++ b/modules/mesh-delivery-builder/internal/versions/again.go @@ -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' diff --git a/modules/mesh-delivery-builder/internal/versions/reserve.go b/modules/mesh-delivery-builder/internal/versions/reserve.go new file mode 100644 index 00000000..36b02425 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/reserve.go @@ -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.`, 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 +} diff --git a/modules/mesh-delivery-builder/internal/versions/reserve_test.go b/modules/mesh-delivery-builder/internal/versions/reserve_test.go new file mode 100644 index 00000000..c02c0c96 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/versions/reserve_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index dae603fa..d03eb00c 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -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) } diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 9ab32e47..39a0cc6b 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -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 }, From d813ce7e3976b973f9d752d6d94da83db2bf1ef9 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:45:19 +0100 Subject: [PATCH 2/3] FB4a: the forge message forms of an owed act's outcome (owed.ended) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gate's developer needs OwedEnded and its outcomes; fields are the ones §164 and §168 name (id, outcome, words, the seat that said the act, time). Builds: design 47 G §164, §168 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../forgemsg/forge_test.go | 34 +++++++++ .../forgemsg/owed.go | 76 +++++++++++++++++++ 2 files changed, 110 insertions(+) create mode 100644 modules/mesh-delivery-forge-bridge/forgemsg/owed.go diff --git a/modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go b/modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go index a94f4779..18c75b4c 100644 --- a/modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go +++ b/modules/mesh-delivery-forge-bridge/forgemsg/forge_test.go @@ -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) + } + } + }) } diff --git a/modules/mesh-delivery-forge-bridge/forgemsg/owed.go b/modules/mesh-delivery-forge-bridge/forgemsg/owed.go new file mode 100644 index 00000000..b8400a36 --- /dev/null +++ b/modules/mesh-delivery-forge-bridge/forgemsg/owed.go @@ -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...`). +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 +} From 154505f451ae9da7f6a8ad2e2ee0f58031cca04d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 11:45:37 +0100 Subject: [PATCH 3/3] Assigner: repair against the new contracts (source key per branch, want with digests, seat verbs not yet served) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §15, §95, §353, §379 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../cmd/mesh-delivery-assigner/pass_test.go | 8 ++++---- .../cmd/mesh-delivery-assigner/wants_tools.go | 3 +++ .../cmd/mesh-delivery-assigner/world.go | 5 ++++- .../mesh-delivery-assigner/internal/assign/wants_test.go | 3 ++- 4 files changed, 13 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 5b077f78..0bc60e9a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -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}}}) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go index 02a0d387..034e9b15 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go @@ -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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index cf2b67f4..893d02f5 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -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" diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go index d15823b1..1c38e3a7 100644 --- a/modules/mesh-delivery-assigner/internal/assign/wants_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -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 {