Merge branch 'feat/47-G-builder6' into feat/47-G-join8

This commit is contained in:
2026-11-12 00:12:54 +01:00
49 changed files with 3246 additions and 333 deletions
@@ -0,0 +1,79 @@
package main
import (
"context"
"encoding/json"
"fmt"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"mesh-delivery-builder/internal/build"
)
// catalogWithin is how long a call waits for the catalog half's first view of one branch.
var catalogWithin = 30 * time.Second
// buildTool is builder_build (design 47 G §589): a module's build keys whole.
func buildTool(r *build.BuildReader) stdio.Tool {
return stdio.Tool{Name: build.BuildTool,
Description: "A module's build keys, whole, newest first (at most 20): per key its fingerprint (whole, as the build-input fingerprint the catalog graph " +
"entry carries it), state (queued, building, done or failed), revision and since when, the step and its node, the error whole, event-builds " +
"(the revisions events started since the last a person started; after three the operator is asked) and the open ask; and newest, the watched " +
"branches whose newest catalog graph entry of the module names that fingerprint, read from the catalog half. A module none of whose keys the newest " +
"entry names says so, naming the entry's fingerprint. With fingerprint, that key alone. Which node wants which fingerprint is the assigner's " +
"to answer (the seat mesh-wants). Read-only.",
Input: map[string]any{"type": "object", "required": []string{"module"}, "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "the module"},
"fingerprint": map[string]any{"type": "string", "description": "one key: its fingerprint, src2:<hex>"}}},
Run: func(args map[string]any) (any, error) {
module, _ := args["module"].(string)
fp, _ := args["fingerprint"].(string)
return r.Build(module, fp)
}}
}
// newestVia reads the fingerprint of a module's newest catalog graph entry on each of its watched branches, from the catalog
// half through the seat mesh-catalog (the verb registrations for the branches, the verb revision for each branch's view), as
// the delivery view's tools read it: once per call (design 47 G §532, §589, A-87).
func newestVia(ask meshcatalogclient.Call) func(module string) (map[string]string, error) {
return func(module string) (map[string]string, error) {
raw, err := ask(meshcatalogclient.RegistrationsVerb, meshcatalog.RegistrationsInput{Module: module})
if err != nil {
return nil, fmt.Errorf("the registrations of %s: %w", module, err)
}
answer, said, ok := buscalls.Unwrap(raw)
if !ok {
return nil, fmt.Errorf("%s answered: %s", meshcatalogclient.RegistrationsVerb, said)
}
var regs meshcatalog.RegistrationsAnswer
if err := json.Unmarshal(answer, &regs); err != nil {
return nil, fmt.Errorf("%s answered no registrations: %w", meshcatalogclient.RegistrationsVerb, err)
}
out := map[string]string{}
for _, reg := range regs.Registrations {
if reg.Module != module {
continue
}
for _, branch := range reg.Branches {
c, err := meshcatalogclient.NewCatalog(ask, branch)
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(context.Background(), catalogWithin)
err = c.Start(ctx)
cancel()
if err != nil {
return nil, fmt.Errorf("the catalog half of %s has no first view: %w", branch, err)
}
if e, ok := c.Entry(module); ok && e.Fp != "" {
out[branch] = e.Fp
}
}
}
return out, nil
}
}
@@ -0,0 +1,32 @@
package main
import (
"encoding/json"
"strings"
"testing"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// Design 47 G §589, A-87: the tools bundle reads each watched branch's newest catalog graph entry of the module from the catalog
// half itself: the verb registrations for the branches, the verb revision for each branch's view.
func TestNewestIsReadFromTheCatalogHalfOfEachWatchedBranch(t *testing.T) {
fp := map[string]string{"main": "src2:" + strings.Repeat("1", 64), "feature": "src2:" + strings.Repeat("2", 64)}
ask := func(key string, body any) (json.RawMessage, error) {
switch key {
case meshcatalogclient.RegistrationsVerb:
return json.Marshal(meshcatalog.RegistrationsAnswer{Registrations: []meshcatalog.Registration{{Module: "m", Branches: []string{"main", "feature"}}}})
case meshcatalogclient.RevisionVerb:
in := body.(meshcatalog.RevisionInput)
return json.Marshal(meshcatalog.RevisionAnswer{Revision: &meshcatalog.CatalogRevision{Revision: 3, Kind: meshcatalogclient.KindTrunk},
Entries: []meshcatalog.CatalogEntry{{Module: "m", Fp: fp[in.Branch]}, {Module: "z", Fp: "src2:" + strings.Repeat("9", 64)}}})
}
t.Fatalf("asked %s", key)
return nil, nil
}
got, err := newestVia(ask)("m")
if err != nil || len(got) != 2 || got["main"] != fp["main"] || got["feature"] != fp["feature"] {
t.Fatalf("newest = %v, %v, want each branch's entry fingerprint whole", got, err)
}
}
@@ -54,17 +54,9 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, q
if a.Refused {
return nil, errors.New(a.Answer)
}
return map[string]any{"key": key, "name": a.Name, "do": a.Do, "version": a.Version, "answer": a.Answer}, nil
return map[string]any{"key": key, "name": a.Name, "answer": a.Answer}, nil
}
nameFP := map[string]any{"type": "object", "required": []string{"name", "fp", "why"}, "properties": map[string]any{
"name": map[string]any{"type": "string", "description": "the numbered name, <module>/<artifact>, as builder_versions lists it"},
"fp": map[string]any{"type": "string", "description": "the fingerprint, src2:<hex>"},
"why": map[string]any{"type": "string", "description": "why, in plain words"}}}
return []stdio.Tool{
{Name: versions.PublishAgainTool, Description: versions.PublishAgainDescription, Input: nameFP,
Run: func(args map[string]any) (any, error) {
return handAct(build.NewPublishAgain(str(args, "name"), str(args, "fp"), str(args, "why"), stdio.CallerOf(args), now()))
}},
{Name: versions.FloorAgainTool, Description: versions.FloorAgainDescription,
Input: map[string]any{"type": "object", "required": []string{"name", "why"}, "properties": map[string]any{
"name": map[string]any{"type": "string", "description": "the numbered name, <module>/<artifact>"},
@@ -164,15 +156,16 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, q
"fingerprint, source revision, repository, directory, commit and ref, the build node, when it ended, its outcome (done, failed or " +
"queued), the manifest built, the bases it was built against and each artifact made with its digest and reference; a failed " +
"build carries its error and the step it was at. more says the " +
"log held more than one answer carries: ask again from the last build's time. What the operator's command nox builds reads. Read-only.",
"log held more than one answer carries: ask again from the last build's time. With module, that module's builds only. What the operator's command nox builds reads. Read-only.",
Input: map[string]any{"type": "object", "required": []string{"since"}, "properties": map[string]any{
"since": map[string]any{"type": "string", "description": "RFC 3339 time; the builds that ended at or after it"}}},
"since": map[string]any{"type": "string", "description": "RFC 3339 time; the builds that ended at or after it"},
"module": map[string]any{"type": "string", "description": "only this module's builds"}}},
Run: func(args map[string]any) (any, error) {
at, err := time.Parse(time.RFC3339Nano, str(args, "since"))
if err != nil {
return nil, fmt.Errorf("since %q is not an RFC 3339 time: %w", str(args, "since"), err)
}
return build.BuildsSince(since, at)
return build.BuildsSince(since, at, str(args, "module"))
}},
{Name: build.AnnounceAgainTool,
Description: "Announce a done build again (take in again, compose again): the builder writes the done key of module and fingerprint " +
@@ -371,7 +364,11 @@ func main() {
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, fr, ofr)
tools = append(tools, buildTool(&build.BuildReader{Builds: read(stdio.State("builds")), Entry: entryOf(stdio.State("builds")), Asked: read(stdio.State("asked")),
Newest: newestVia(stdio.Ask), Steps: r}))
tools = append(tools, reservedTools(rr, buildAgain, time.Now, meshhandactsclient.Recorder{Ask: stdio.Ask})...)
qs, as := stdio.State(versions.PublishQueueState), stdio.State(versions.ActedState)
tools = append(tools, republishTools(readerOf(qs), readerOf(as), buildAgain, time.Now, meshhandactsclient.Recorder{Ask: stdio.Ask})...)
ws := stdio.State(build.WaitsState)
wr := versions.Reader{Keys: ws.Keys, Get: func(key string) (json.RawMessage, bool, error) {
e, err := ws.Get(key)
@@ -400,3 +397,14 @@ func main() {
os.Exit(1)
}
}
// entryOf reads a key with its revision and when it was written.
func entryOf(s stdio.KeptState) func(key string) (json.RawMessage, uint64, time.Time, error) {
return func(key string) (json.RawMessage, uint64, time.Time, error) {
e, err := s.Get(key)
if err != nil || e == nil {
return nil, 0, time.Time{}, err
}
return e.Value, e.Revision, written(s, key, e.Revision), nil
}
}
@@ -0,0 +1,65 @@
package main
import (
"encoding/json"
"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"
)
// readerOf is a kept state as the versions package's reader.
func readerOf(s stdio.KeptState) versions.Reader {
return versions.Reader{Keys: s.Keys, Get: func(key string) (json.RawMessage, bool, error) {
e, err := s.Get(key)
if err != nil || e == nil {
return nil, false, err
}
return e.Value, true, nil
}}
}
// republishTools are builder_publish_again and builder_publish_queue (design 47 G §588). The publish again is a hand act:
// it is recorded in the hand-act log first, then written to reasks for the builder, which owns the queue, to answer
// with the versions it queued.
func republishTools(queue, acted 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.PublishAgainTool, Description: versions.PublishAgainDescription,
Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "only this module's reserved versions"},
"version": map[string]any{"type": "string", "description": "with module: only this reserved version, MAJOR.MINOR.PATCH"},
"why": map[string]any{"type": "string", "description": "why, in plain words"}}},
Run: func(args map[string]any) (any, error) {
key, value, err := build.NewRepublish(str(args, "module"), str(args, "version"), str(args, "why"), stdio.CallerOf(args), now())
if err != nil {
return nil, err
}
if _, err := acts.Record(meshhandactsclient.Request{What: fmt.Sprintf("published reserved versions again by hand (module %q, version %q)", str(args, "module"), str(args, "version")),
Why: str(args, "why"), Cause: versions.PublishAgainTool, Verb: "publish-again"}); err != nil {
return nil, fmt.Errorf("the hand act was not recorded through mesh-hand-acts, so nothing was queued: %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, "queued": a.Queued, "answer": a.Answer}, nil
}},
{Name: versions.PublishQueueTool, Description: versions.PublishQueueDescription,
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) { return versions.AnswerQueue(queue, acted) }},
}
}
@@ -115,7 +115,7 @@ func (h heldOnTheBus) seen(ctx context.Context, a buildstep.StepAsk, later bool)
// stream is the build seat's stream of language, its name and the subject of kind on it.
func (h heldOnTheBus) stream(ctx context.Context, language, kind string) (jetstream.Stream, string, string, error) {
subject, err := buildstep.AskSubject(language, kind)
subject, err := buildstep.AskSubjectBy(language, kind, buildstep.AskerBuilder)
if err != nil {
return nil, "", "", err
}
@@ -286,7 +286,7 @@ func (h heldOnTheBus) Queue(ctx context.Context, language string, limit int) (bu
return q, err
}
for _, kind := range buildstep.AskKinds {
subject, _ := buildstep.AskSubject(language, kind)
subject, _ := buildstep.AskSubjectBy(language, kind, buildstep.AskerBuilder)
var ahead []build.Queued
if err := onSubject(ctx, stream, subject, func(m build.Queued) bool {
ahead = append(ahead, m)
@@ -165,6 +165,18 @@ func (f *fakeBuildModules) serve(t *testing.T, module string, nodes ...string) {
if err := runner.DeclareAskWorkers(f.ctx, f.js, language, 30*time.Second); err != nil {
t.Fatal(err)
}
// Design 47 G §600: each queue's worker takes an ask in both forms, main's and the builder's (accept.<queue>.builder); the
// build modules' helper above filters main's alone, so the worker is declared again with both.
stream, _ := sdkstep.AskStream(language)
for _, kind := range sdkstep.AskKinds {
worker, _ := sdkstep.AskWorker(language, kind)
main, _ := sdkstep.AskSubject(language, kind)
byBuilder, _ := sdkstep.AskSubjectBy(language, kind, sdkstep.AskerBuilder)
if _, err := f.js.CreateOrUpdateConsumer(f.ctx, stream, jetstream.ConsumerConfig{Durable: worker, FilterSubjects: []string{main, byBuilder},
AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: 5}); err != nil {
t.Fatal(err)
}
}
put := func(key string, value any) (uint64, error) {
raw, _ := value.(json.RawMessage)
return kv.Put(f.ctx, key, raw)
@@ -586,18 +598,10 @@ func TestBuilderMakesADoneKeyWithTheDigests(t *testing.T) { // 1607/AC7
if got := reg.numberedOf("netcheck/tools-go"); len(got) != 0 {
t.Errorf("version.numbered of netcheck/tools-go said %v, want none", got)
}
// DLV-B10: builder_publish_again of a published, held build is refused: nothing says a version numbered (§349).
for _, a := range []string{"tools-go", "tools-typescript"} {
reg.published(t, "netcheck/"+a, "0.1.5", digestOf(a))
}
reg.awaitOf(t, "netcheck/tools-go", func(es []versions.Entry) bool { return len(es) == 1 && es[0].State == versions.Published })
key, again, err := build.NewPublishAgain("netcheck/tools-go", fp, "the holder lost it", "me@shanks", time.Now())
if err != nil {
t.Fatal(err)
}
if a := reg.answered(t, key, again); !a.Refused {
t.Errorf("builder_publish_again = %+v, want a refusal", a)
}
// A numbered version a holder will refuse below, written to the state as the builder numbers one.
refusable, _ := json.Marshal(versions.Entry{Name: "netcheck/tools-go", Kind: versions.KindBundle, Series: "0.1", Version: "0.1.6", Module: "netcheck",
Fingerprint: "src2:" + strings.Repeat("6", 64), Seat: ociartifactregistry.Seat, State: versions.Numbered, At: time.Now().UTC()})
@@ -1084,7 +1088,7 @@ func TestARestartedBuilderPutsNoSecondAsk(t *testing.T) {
held := heldOnTheBus{b.js}
goAsk := sdkstep.StepAsk{ID: v.Asks[sdkstep.AskGo], Language: sdkstep.AskGo, Kind: sdkstep.AskDelivery}
stale := sdkstep.StepAsk{ID: "mvo53jcm70b5f7", Language: sdkstep.AskGo, Kind: sdkstep.AskDelivery}
subject, _ := sdkstep.AskSubject(sdkstep.AskGo, sdkstep.AskDelivery)
subject, _ := sdkstep.AskSubjectBy(sdkstep.AskGo, sdkstep.AskDelivery, sdkstep.AskerBuilder)
if _, err := b.js.Publish(b.ctx, subject, []byte("{}"), jetstream.WithMsgID(stale.ID)); err != nil {
t.Fatal(err)
}
@@ -7,8 +7,10 @@ package main
import (
"cmp"
"context"
"errors"
"fmt"
"io"
"io/fs"
"log"
"maps"
"os"
@@ -22,6 +24,8 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
@@ -136,7 +140,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
tools := &buildrun.ToolchainSet{}
// Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot
// (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity.
asks := buildrun.JetStreamAsks{JS: js}
// The read of an ask's steps is a kept call (design 47 G §568): a fault ends no key, and past five minutes unanswered it is a
// finding; its bound is a key of the state bounds, bound once the bus holds it (set below, with the findings).
stepReads := &stepsRead{bounds: func() jetstream.KeyValue { return nil }, bound: meshconditionsclient.CallUnansweredAfter, now: time.Now, logf: logger.Printf}
asks := buildrun.JetStreamAsks{JS: js, Unanswered: stepReads.unanswered, Answered: stepReads.answered}
// The asks the build seats gave up on (issue 734): heard from DEAD_LETTERS, each ends the wait for its step at once,
// and the lane asks its part again (Builder.waitOnce) instead of waiting for the check bound.
// Heard in the background, and again at a reconnect or an ask while it is not; whether it is heard is kept in the
@@ -177,6 +184,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// A publish that ends badly is a finding (design 51 *Health and conditions*): a holder's refused, a version no holder
// answered within its bound (a key of the state bounds with its own time to live), a seat's outcome that does not read.
pubBoundsKV := process.AwaitBucket(ctx, js, conn, versions.BoundsBucket, "a publish no holder answers, and a node holding the old form seven days, are not raised", logger.Printf)
stepReads.bounds, stepReads.findings = func() jetstream.KeyValue { return pubBoundsKV.KV() }, found
pubs := &publishes{bounds: pubBoundsKV.KV, versions: numberer.Versions, findings: found, bound: versions.PublishBound, logf: logger.Printf}
numberer.Publishing = pubs.arm
b := &build.Builder{
@@ -200,13 +208,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
b.Numbers = build.VersionNumbers{N: numberer}
// The builder's toolchain waits (design 47 G §672): its state waits, and wait-held and wait-ended on its stream waits; its
// bound is its setting toolchain-wait-minutes.
waitBound := defaultToolchainWait
if v := strings.TrimSpace(getenv("MESH_BUILDER_TOOLCHAIN_WAIT_MINUTES")); v != "" {
n, err := strconv.Atoi(v)
if err != nil || n < 1 {
return fmt.Errorf("MESH_BUILDER_TOOLCHAIN_WAIT_MINUTES %q is not a number of minutes", v)
}
waitBound = time.Duration(n) * time.Minute
waitBound, err := toolchainWaitOf(getenv("MESH_BUILDER_TOOLCHAIN_WAIT_FILE"))
if err != nil {
return err
}
waitsBucket := process.AwaitBucket(ctx, js, conn, build.WaitsBucket, "the toolchain waits are kept in memory only, and builder_waits shows none", logger.Printf)
b.Waits = newWaitsKept(waitsBucket, func(ctx context.Context, event, id string, body any) error {
@@ -221,17 +225,32 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// 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, DropReserved: reserved.dropped, Done: doneVia(b), Reconsider: b.Reconsider,
BuildAgain: func(ctx context.Context, module, fp, why, by string) (string, bool) {
r := b.BuildAgain(ctx, module, fp, why, by, buildAgainBound)
if r.Refused != "" {
return r.Refused, true
}
if r.Failed != "" {
return "built again, and it failed before its asks were put: " + r.Failed, false
}
return fmt.Sprintf("built again as revision %d", r.Revision), false
}}
acts := versions.Acts{Numberer: numberer, DropReserved: reserved.dropped, Reconsider: b.Reconsider}
// A reserved version left unpublished is published again, by the builder on an event or by a person's tool (design 47 G
// §588): its states publish-queue and acted, bound once the bus holds them.
republishMax, err := republishMaxOf(getenv("MESH_BUILDER_REPUBLISH_MAX_FILE"))
if err != nil {
return err
}
queueKV := process.AwaitBucket(ctx, js, conn, versions.PublishQueueBucket, "no version is published again on an event or by builder_publish_again", logger.Printf)
actedKV := process.AwaitBucket(ctx, js, conn, versions.ActedBucket, "no version is published again on an event or by builder_publish_again", logger.Printf)
republish := &versions.Republisher{Reserved: reserved.r, Queue: awaitedQueue{awaitedReserved{awaitedVersions{b: queueKV, bucket: versions.PublishQueueBucket}}},
Acted: awaitedReserved{awaitedVersions{b: actedKV, bucket: versions.ActedBucket}}, Entry: entryVia(calls.Call), Send: sendVia(calls.Call),
Unreleased: reserved.unreleasedWords, Max: republishMax, Logf: logger.Printf}
acts.Republish = republish.Republish
again := &republishing{rp: republish}
// The operator is asked once a key's event builds are spent (design 47 G §568): the builder's asker on the seat operator-channel.
decided := &decisions{subject: oc.SubjectDecided(cred.Module)}
retakeActed := awaitedReserved{awaitedVersions{b: actedKV, bucket: versions.ActedBucket}}
operator := &build.OperatorAsk{Builder: b, Acted: retakeActed, Ctx: ctx, Logf: logger.Printf,
Client: &occ.Client{Module: cred.Module, State: calls.State("asked"), Publish: publishVia(ctx, js),
Record: recordVia(calls.Call, calls.Ask), Subscribe: decided.hear(logger.Printf)}}
if err := operator.Listen(); err != nil {
return fmt.Errorf("hearing the operator's answers: %w", err)
}
retake := &retaking{rt: &build.Retaker{Builder: b, Acted: retakeActed, Entry: entryVia(calls.Call), Ask: operator.Ask, Logf: logger.Printf}}
queueKV.Ready(again.bound(ctx, js, logger.Printf))
actedKV.Ready(again.bound(ctx, js, logger.Printf))
// A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base
// revision (the gate's source state). The source's revisions are watched with their history (a direct get of one revision
// is not what a read grants): a revision heard after the candidate that names it builds that candidate then. Where it
@@ -376,6 +395,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// 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, res: reserved}
events := func(m jetstream.Msg) {
if again.heard(ctx, m, logger.Printf) { // the publish again's own subjects, left for the bus to give again when its state cannot take them
return
}
if decided.heard(m) { // the operator's answer to an ask about a key (§568)
return
}
if retake.heard(ctx, m, logger.Printf) { // a key failed on a transport fault, built again when its cause passed (§568)
return
}
if reads.heard(m, logger.Printf) || heardTools.Reported(ctx, m) || heardTools.Running(ctx, m) || 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
}
@@ -422,6 +450,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
unput.expired(ctx, e)
return
}
if _, _, ok := versions.StepsBoundOf(e.Key()); ok {
stepReads.expired(ctx, e)
return
}
if node, ok := enginereads.BoundOf(e.Key()); ok {
if e.Operation() == jetstream.KeyValuePurge { // its five minutes ended
reads.expiredCall(node)
@@ -572,3 +604,28 @@ func (s *sourceHeads) head(_ context.Context, rev uint64, repository string) (st
}
return "", nil
}
// toolchainWaitOf reads the toolchain wait's setting, toolchain-wait-minutes, from the file the manifest declares with the
// content ${config:toolchain-wait-minutes} (design 47 G §672, A-71), at the builder's start: no file, or an empty one, is
// the default; a content that is no number of minutes is an error naming the file.
func toolchainWaitOf(path string) (time.Duration, error) {
if path == "" {
return defaultToolchainWait, nil
}
raw, err := os.ReadFile(path)
if errors.Is(err, fs.ErrNotExist) {
return defaultToolchainWait, nil
}
if err != nil {
return 0, err
}
v := strings.TrimSpace(string(raw))
if v == "" {
return defaultToolchainWait, nil
}
n, err := strconv.Atoi(v)
if err != nil || n < 1 {
return 0, fmt.Errorf("%s holds %q, which is not a number of minutes", path, v)
}
return time.Duration(n) * time.Minute, nil
}
@@ -58,7 +58,7 @@ func reAsked(ctx context.Context, b askAgainer, acts versions.Acts, kv jetstream
released(ctx, b, kv, e, r, logf)
return
}
if r.Mode == build.ModePublishAgain || r.Mode == build.ModeFloorAgain || r.Mode == build.ModeOldFormDrop || r.Mode == build.ModeReservedDrop {
if r.Mode == build.ModeRepublish || 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
@@ -192,10 +192,15 @@ func emitBuildEnded(ctx context.Context, js jetstream.JetStream, cred process.Ac
// versionActed answers a reasks value of builder_publish_again, builder_floor_again or builder_old_form's drop, a hand act said in the log.
func versionActed(ctx context.Context, acts versions.Acts, kv jetstream.KeyValue, e jetstream.KeyValueEntry, r build.ReAsk, logf func(string, ...any)) {
var err error
if r.Mode == build.ModePublishAgain {
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
if r.Mode == build.ModeRepublish {
if acts.Republish == nil {
err = errors.New("this builder publishes nothing again")
} else {
r.Queued, err = acts.Republish(ctx, r.Module, r.Version)
if err == nil {
r.Answer = "queued " + fmt.Sprint(len(r.Queued)) + " versions to publish again: " + strings.Join(r.Queued, ", ") + "; builder_publish_queue shows them"
}
}
} else if r.Mode == build.ModeReservedDrop {
if acts.DropReserved == nil {
err = errors.New("this builder keeps no reservations")
@@ -0,0 +1,146 @@
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/fs"
"os"
"strconv"
"strings"
"sync/atomic"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"mesh-delivery-builder/internal/versions"
)
// The builder's publish again (design 47 G §588, A-73, A-76): the wiring of versions.Republisher to the bus.
// awaitedQueue is the state publish-queue as versions.QueueStore: bound once the bus holds it; its bytes are the cap's.
type awaitedQueue struct{ awaitedReserved }
func (s awaitedQueue) Bytes(ctx context.Context) (uint64, error) {
kv, err := s.kv()
if err != nil {
return 0, err
}
st, err := kv.Status(ctx)
if err != nil {
return 0, err
}
return st.Bytes(), nil
}
// entryVia reads a catalog graph entry by its fingerprint through the seat mesh-catalog's verb entry.
func entryVia(call toolCaller) func(ctx context.Context, fingerprint string) (meshcatalog.CatalogEntry, error) {
return func(_ context.Context, fingerprint string) (meshcatalog.CatalogEntry, error) {
raw, err := call("seat:"+meshcatalog.Seat+"."+meshcatalog.VerbEntry, meshcatalog.EntryInput{Fingerprint: fingerprint})
if err != nil {
return meshcatalog.CatalogEntry{}, err
}
var a meshcatalog.EntryAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return meshcatalog.CatalogEntry{}, fmt.Errorf("the catalog answered what is no entry: %w", err)
}
return a.Entry, nil
}
}
// sendVia sends a publish to the registry seat with the digest and no parts (A-73): the holder takes the bytes from the
// artifact store, addressed by digest. The four seats' publish take one wire form; an answer that lacks parts, or a fault of
// the call, is the holder's refusal, said in its words (not stored when the holder lacks the bytes).
func sendVia(call toolCaller) func(ctx context.Context, p versions.Publish) error {
return func(_ context.Context, p versions.Publish) error {
raw, err := call("seat:"+p.Seat+"."+versions.VerbPublish, publishBody(p))
if err != nil {
return fmt.Errorf("%s's publish of %s %s: %w", p.Seat, p.Name, p.Version, err)
}
var a gopackageregistry.PublishAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return fmt.Errorf("%s answered what is no publish answer: %w", p.Seat, err)
}
if len(a.Lacks) > 0 {
return fmt.Errorf("%s%s holds %d of the parts of %s %s", versions.NotStoredWords, p.Seat, len(a.Lacks), p.Name, p.Version)
}
return nil
}
}
// publishBody is the seat's publish input for p: the digest, its kind and name and version, no parts.
func publishBody(p versions.Publish) gopackageregistry.PublishInput {
return gopackageregistry.PublishInput{Commit: p.Commit, Digest: p.Digest, Fingerprint: p.Fingerprint, Kind: p.Kind, Name: p.Name, Transfer: p.Transfer, Version: p.Version}
}
// republishMaxOf reads the setting republish-max-versions from the file the manifest declares (design 47 G §633, A-71's
// form): no file or an empty one is the default; a content that is no positive number is an error naming the file.
func republishMaxOf(path string) (int, error) {
if path == "" {
return versions.DefaultRepublishMax, nil
}
raw, err := os.ReadFile(path)
if errors.Is(err, fs.ErrNotExist) {
return versions.DefaultRepublishMax, nil
}
if err != nil {
return 0, err
}
v := strings.TrimSpace(string(raw))
if v == "" {
return versions.DefaultRepublishMax, nil
}
n, err := strconv.Atoi(v)
if err != nil || n < 1 {
return 0, fmt.Errorf("%s holds %q, which is not a number of versions", path, v)
}
return n, nil
}
// republishing starts the publish again once both its states are bound: the queue is taken up again and the last running
// of every node's module is read (§556), acted on where its sequence is newer than the kept one.
type republishing struct {
rp *versions.Republisher
ready atomic.Int32
}
func (g *republishing) bound(ctx context.Context, js jetstream.JetStream, logf func(string, ...any)) func(jetstream.KeyValue) {
return func(jetstream.KeyValue) {
if g.ready.Add(1) != 2 {
return
}
if err := g.rp.Resume(ctx); err != nil {
logf("the queued publishes are not taken up again: %v", err)
}
err := lastPerSubject(ctx, js, nodeengine.StreamRunning, nodeengine.SubjectRunningAll, func(subject string, data []byte, seq uint64) {
if _, err := g.rp.Heard(ctx, subject, data, seq); err != nil {
logf("the running %s is not acted on at the start: %v", subject, err)
}
})
if err != nil {
logf("the running of the node-engine seat is not read for the publish again at the start: %v", err)
}
}
}
// heard hands a message of the events consumer to the publish again: one of its subjects whose state cannot take it is left
// for the bus to give again (Nak); every message goes on to the other parts.
func (g *republishing) heard(ctx context.Context, m jetstream.Msg, logf func(string, ...any)) (stop bool) {
meta, err := m.Metadata()
if err != nil {
return false
}
handled, err := g.rp.Heard(ctx, m.Subject(), m.Data(), meta.Sequence.Stream)
if handled && err != nil {
logf("%v; heard again", err)
if nakErr := m.Nak(); nakErr != nil {
logf("leaving %s for the bus to give again: %v", m.Subject(), nakErr)
}
return true
}
return false
}
@@ -0,0 +1,79 @@
package main
import (
"context"
"encoding/json"
"errors"
"os"
"path/filepath"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry"
"mesh-delivery-builder/internal/build"
"mesh-delivery-builder/internal/versions"
)
// TestRepublishSend: design 47 G §588, A-73: a publish again is the registry seat's publish with the digest and no parts; the
// four seats take one wire form, a call's fault or an answer that lacks parts is the holder's refusal in its words, and its
// setting is read from the file the manifest declares.
func TestRepublishSend(t *testing.T) {
p := versions.Publish{Seat: versions.SeatArtifacts, Kind: "bundle", Name: "m/prog", Version: "0.1.3", Digest: "sha256:aa", Fingerprint: "src2:m", Commit: "c", Transfer: "t"}
want, _ := json.Marshal(publishBody(p))
if strings.Contains(string(want), "parts") {
t.Errorf("the publish names parts: %s", want)
}
for name, v := range map[string]any{"container image": containerimageregistry.PublishInput{Digest: "sha256:aa"}, "oci": ociartifactregistry.PublishInput{Digest: "sha256:aa"},
"go": gopackageregistry.PublishInput{Digest: "sha256:aa"}, "npm": npmpackageregistry.PublishInput{Digest: "sha256:aa"}} {
var ref, got map[string]any
a, _ := json.Marshal(gopackageregistry.PublishInput{Digest: "sha256:aa"})
b, _ := json.Marshal(v)
_ = json.Unmarshal(a, &ref)
_ = json.Unmarshal(b, &got)
if len(ref) != len(got) {
t.Errorf("the %s seat's publish input is %s, the go seat's %s: one wire form is sent to all four", name, b, a)
}
}
var asked string
for name, c := range map[string]struct {
answer string
err error
want string
}{
"held": {`{"held":true,"transfer":"t"}`, nil, ""},
"lacks": {`{"lacks":["sha256:bb"],"transfer":"t"}`, nil, versions.NotStoredWords},
"a fault": {"", errors.New("no responders"), "no responders"},
"no sense": {`[`, nil, "no publish answer"},
} {
send := sendVia(func(key string, body any) (json.RawMessage, error) {
asked = key
return json.RawMessage(c.answer), c.err
})
err := send(context.Background(), p)
if asked != "seat:oci-artifact-registry.publish" || (c.want == "") != (err == nil) || (err != nil && !strings.Contains(err.Error(), c.want)) {
t.Errorf("%s: asked %q, error %v, want %q", name, asked, err, c.want)
}
}
dir := t.TempDir()
for content, want := range map[string]int{"": 50, "7": 7} {
path := filepath.Join(dir, "max")
_ = os.WriteFile(path, []byte(content), 0o600)
if got, err := republishMaxOf(path); err != nil || got != want {
t.Errorf("republishMaxOf(%q) = %d, %v, want %d", content, got, err, want)
}
}
_ = os.WriteFile(filepath.Join(dir, "bad"), []byte("0"), 0o600)
if _, err := republishMaxOf(filepath.Join(dir, "bad")); err == nil {
t.Error("a setting of 0 is accepted")
}
if _, _, err := build.NewRepublish("", "0.1.3", "why", "me", versionsNow()); err == nil {
t.Error("a version with no module is accepted")
}
}
func versionsNow() time.Time { return time.Unix(1, 0) }
@@ -125,6 +125,11 @@ func (rv *reservations) unreleased(ctx context.Context, v versions.ReservedVersi
needs: "builder_reserved shows the reservation; builder_publish_again publishes the missing artifacts under it; builder_reserved_drop drops it."})
}
// unreleasedWords raises version-unreleased about a version with the words of a publish again that ended badly.
func (rv *reservations) unreleasedWords(ctx context.Context, v versions.ReservedVersion, why string) {
rv.unreleased(ctx, v, why)
}
// 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 {
@@ -0,0 +1,124 @@
package main
import (
"context"
"encoding/json"
"fmt"
mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
stdio "git.novox.be/novox/mesh-sdk/go"
"github.com/nats-io/nats.go/jetstream"
"mesh-delivery-builder/internal/build"
)
// The builder's retake of a key failed on a transport fault, on the node-engine seat's events (design 47 G §568, ADR 0673): the
// wiring of build.Retaker to the events consumer.
type retaking struct{ rt *build.Retaker }
// heard hands a message of the events consumer to the retake: one of its subjects whose state cannot take it is left for the bus to
// give again (Nak); every message goes on to the other parts.
func (g *retaking) heard(ctx context.Context, m jetstream.Msg, logf func(string, ...any)) (stop bool) {
meta, err := m.Metadata()
if err != nil {
return false
}
handled, err := g.rt.Heard(ctx, m.Subject(), m.Data(), meta.Sequence.Stream)
if handled && err != nil {
logf("%v; heard again", err)
if nakErr := m.Nak(); nakErr != nil {
logf("leaving %s for the bus to give again: %v", m.Subject(), nakErr)
}
return true
}
return false
}
// The builder's three bus adapters for the operator-channel client (design 47 G §568), as the teller has them: its own ask
// published, the router's record read through the seat, its own `decided` heard through the module's hub.
// publishVia publishes an accept of the seat, de-duplicated by id: the same id twice is one message.
func publishVia(ctx context.Context, js jetstream.JetStream) func(subject string, body any, id string) error {
return func(subject string, body any, id string) error {
data, err := json.Marshal(body)
if err != nil {
return err
}
var opts []jetstream.PublishOpt
if id != "" {
opts = append(opts, jetstream.WithMsgID(id))
}
_, err = js.Publish(ctx, subject, data, opts...)
return err
}
}
// routerBucket is the bucket of the router's asks: the state `asks` of the seat operator-channel (the contract's StateAsks) in
// the bucket of the seat's holder on the seat list, which the seat mesh-module-assigner's verb assignments answers with seats
// (design 47 G A-74); no module's name is in the program.
func routerBucket(call func(key string, body any) (json.RawMessage, error)) (string, error) {
raw, err := call("seat:"+mma.Seat+"."+mma.VerbAssignments, mma.AssignmentsInput{Seats: mma.AssignmentsInputSeatsTrue})
if err != nil {
return "", fmt.Errorf("who holds the seat %s is not asked of the seat %s: %w", oc.Seat, mma.Seat, err)
}
var held struct {
Seats []mma.SeatHolders `json:"seats"`
}
if err := json.Unmarshal(raw, &held); err != nil {
return "", fmt.Errorf("the answer of %s.%s is not its seat's: %w", mma.Seat, mma.VerbAssignments, err)
}
for _, s := range held.Seats {
if s.Seat == oc.Seat && len(s.Holders) > 0 {
return s.Holders[0].Module + "_" + oc.StateAsks, nil
}
}
return "", fmt.Errorf("no module holds the seat %s on the seat list, so the router's record of an ask is not read", oc.Seat)
}
// recordVia reads the router's record of one of this module's asks: a get of `<module>.<id>` in the bucket the seat's holder
// keeps (record is process.Conn.Ask).
func recordVia(call func(key string, body any) (json.RawMessage, error), record func(bucket, id string) (*stdio.StateEntry, error)) func(id string) (*stdio.StateEntry, error) {
return func(id string) (*stdio.StateEntry, error) {
bucket, err := routerBucket(call)
if err != nil {
return nil, err
}
return record(bucket, id)
}
}
// decisions holds the handler of the router's `decided` event for this module: the module's one hub has the pattern
// `mesh.seat.*.event.>`, which covers it, so the events handler hands the message here (a hub refuses two patterns that overlap).
type decisions struct {
subject string
handle func(jetstream.Msg)
}
// hear is the client's Subscribe: it keeps the handler of the warrants the router says to this module.
func (d *decisions) hear(logf func(string, ...any)) func(pattern string, handle func(stdio.Envelope) error) error {
return func(pattern string, handle func(stdio.Envelope) error) error {
d.handle = func(m jetstream.Msg) {
if err := handle(stdio.Envelope{Key: pattern, Body: m.Data()}); err != nil {
logf("a warrant could not be handled and is given again: %v", err)
if nakErr := m.Nak(); nakErr != nil {
logf("leaving %s for the bus to give again: %v", m.Subject(), nakErr)
}
return
}
if err := m.Ack(); err != nil {
logf("a warrant was handled but could not be acknowledged: %v", err)
}
}
return nil
}
}
// heard takes the message when it is this module's `decided`.
func (d *decisions) heard(m jetstream.Msg) bool {
if d.handle == nil || m.Subject() != d.subject {
return false
}
d.handle(m)
return true
}
@@ -0,0 +1,95 @@
package main
import (
"context"
"errors"
"fmt"
"sync"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"mesh-delivery-builder/internal/versions"
)
// **The builder's read of an ask's steps is a kept call, unanswered past five minutes a finding** (design 47 G §568, A-77, ADR
// 0668 §3): a transport fault opening or taking up the watch of `steps` ends no build key; the read is made again once the bus
// answers. The first fault arms a key of the state bounds (StepsBoundKey) with its own time to live; the bus's marker of its end,
// heard on the watch of the state, raises call-unanswered about the ask while the read is still unanswered; the answer deletes
// the key and clears the finding. No timer.
type stepsRead struct {
bounds func() jetstream.KeyValue
findings *findings
bound time.Duration
now func() time.Time
logf func(string, ...any)
mu sync.Mutex
open map[string]string // the asks whose read is unanswered, by ask id: the last fault
}
// unanswered is told each fault of the read of the ask's steps on module: the first arms the bound.
func (s *stepsRead) unanswered(module, askID string, fault error) {
s.mu.Lock()
if s.open == nil {
s.open = map[string]string{}
}
_, armed := s.open[askID]
s.open[askID] = fault.Error()
s.mu.Unlock()
if armed {
return
}
ctx := context.Background()
var err error
if kv := s.bounds(); kv == nil {
err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket)
} else if _, err = kv.Create(ctx, versions.StepsBoundKey(module, askID), versions.EncodeBound("the read of the steps of the ask "+askID+" on "+module, s.now().Add(s.bound)), jetstream.KeyTTL(s.bound)); errors.Is(err, jetstream.ErrKeyExists) {
err = nil
}
if err != nil {
s.logf("the bound of the read of the steps of the ask %s is not written, so it is raised now: %v", askID, err)
s.raise(ctx, module, askID, fault.Error(), s.now())
}
}
// answered ends the read's bound and clears its finding.
func (s *stepsRead) answered(module, askID string) {
s.mu.Lock()
_, was := s.open[askID]
delete(s.open, askID)
s.mu.Unlock()
if !was {
return
}
ctx := context.Background()
if kv := s.bounds(); kv != nil {
if err := kv.Delete(ctx, versions.StepsBoundKey(module, askID)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
s.logf("the bound of the read of the steps of the ask %s is not deleted: %v", askID, err)
}
}
s.findings.clear(ctx, meshconditionsclient.KindCallUnanswered, meshconditionsclient.ScopeCall, askID)
}
// expired is told each change of the state bounds: the end of a steps bound raises call-unanswered while its read is unanswered.
func (s *stepsRead) expired(ctx context.Context, e jetstream.KeyValueEntry) {
module, askID, ok := versions.StepsBoundOf(e.Key())
if !ok || e.Operation() != jetstream.KeyValuePurge {
return
}
s.mu.Lock()
fault, open := s.open[askID]
s.mu.Unlock()
if !open {
return
}
s.raise(ctx, module, askID, fault, s.now().Add(-s.bound))
}
func (s *stepsRead) raise(ctx context.Context, module, askID, fault string, since time.Time) {
o := callUnanswered(askID, "the read of the steps of "+module, "", fault, "builder."+meshconditionsclient.KindCallUnanswered, since)
o.needs = "the bus answers the read again by itself; builder_stuck shows the key waiting on the ask."
s.findings.raise(ctx, o)
}
@@ -0,0 +1,43 @@
package main
import (
"context"
"errors"
"strings"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient"
"mesh-delivery-builder/internal/versions"
)
// Design 47 G §568, A-77: the first fault of the read of an ask's steps arms a bound key with its own time to live; its end
// while the read is unanswered raises call-unanswered about the ask; the answer deletes the bound and clears the finding.
func TestTheStepsReadIsCallUnansweredAfterItsBound(t *testing.T) {
st := &memFindings{vals: map[string][]byte{}}
f := &findings{w: st, logf: t.Logf, now: func() time.Time { return time.Unix(0, 0) }}
marks := &memBounds{keys: map[string]bool{}}
s := &stepsRead{bounds: func() jetstream.KeyValue { return marks }, findings: f, bound: meshconditionsclient.CallUnansweredAfter,
now: func() time.Time { return time.Unix(600, 0) }, logf: t.Logf}
ask := "mvl0frv18d1958"
callKey := meshconditionsclient.Key(meshconditionsclient.ScopeCall, ask, meshconditionsclient.KindCallUnanswered)
s.unanswered("mesh-build-go", ask, errors.New("nats: no responders available for request"))
if !marks.keys[versions.StepsBoundKey("mesh-build-go", ask)] || st.vals[callKey] != nil {
t.Fatalf("a first fault: bound %v, finding %v; want the bound armed, no finding yet", marks.keys, st.vals[callKey] != nil)
}
s.expired(context.Background(), purged{key: versions.StepsBoundKey("mesh-build-go", ask)})
if got, err := decodeFinding(st.vals[callKey]); err != nil || got.Kind != meshconditionsclient.KindCallUnanswered || !strings.Contains(got.Summary, "the steps of mesh-build-go") || !strings.Contains(got.Summary, ask) {
t.Fatalf("the bound's end: the finding = %+v, %v", got, err)
}
s.answered("mesh-build-go", ask)
if marks.keys[versions.StepsBoundKey("mesh-build-go", ask)] || st.vals[callKey] != nil {
t.Errorf("the read answered: bound %v, finding left %v", marks.keys, st.vals[callKey] != nil)
}
s.expired(context.Background(), purged{key: versions.StepsBoundKey("mesh-build-go", ask)})
if st.vals[callKey] != nil {
t.Error("a bound's end after the answer raised call-unanswered")
}
}
@@ -254,8 +254,8 @@ func (h *toolchainsHeard) agree(ctx context.Context) {
}
// lastPerSubject reads the last message of every subject of a stream kept last per subject, through an ordered consumer
// made for that read (design 47 G §556), and hands each to each; it answers when the last of them was handed.
func lastPerSubject(ctx context.Context, js jetstream.JetStream, stream, filter string, each func(subject string, data []byte)) error {
// made for that read (design 47 G §556), and hands each to each with its stream sequence; it answers when the last of them was handed.
func lastPerSubject(ctx context.Context, js jetstream.JetStream, stream, filter string, each func(subject string, data []byte, sequence uint64)) error {
cons, err := js.OrderedConsumer(ctx, stream, jetstream.OrderedConsumerConfig{FilterSubjects: []string{filter}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy})
if err != nil {
return fmt.Errorf("reading %s: %w", stream, err)
@@ -278,14 +278,18 @@ func lastPerSubject(ctx context.Context, js jetstream.JetStream, stream, filter
if err != nil {
return fmt.Errorf("reading %s: %w", stream, err)
}
each(m.Subject(), m.Data())
seq := uint64(0)
if meta, err := m.Metadata(); err == nil {
seq = meta.Sequence.Stream
}
each(m.Subject(), m.Data(), seq)
}
return nil
}
// readRunning is the start's read of every node's module's last running (§556), before the consumer's.
func (h *toolchainsHeard) readRunning(ctx context.Context, js jetstream.JetStream) {
if err := lastPerSubject(ctx, js, nodeengine.StreamRunning, nodeengine.SubjectRunningAll, func(_ string, data []byte) { h.running(ctx, data) }); err != nil {
if err := lastPerSubject(ctx, js, nodeengine.StreamRunning, nodeengine.SubjectRunningAll, func(_ string, data []byte, _ uint64) { h.running(ctx, data) }); err != nil {
h.logf("the running of the node-engine seat is not read at the start, so a build module stopped while this builder was away is forgotten at its next running: %v", err)
}
}
@@ -0,0 +1,39 @@
package main
import (
"os"
"path/filepath"
"testing"
"time"
)
// A-71: the toolchain wait's setting is read at the start from the file the manifest declares, never from the environment.
func TestToolchainWaitIsReadFromItsFile(t *testing.T) {
dir := t.TempDir()
n := 0
file := func(content string) string {
n++
p := filepath.Join(dir, "w"+string(rune('a'+n)))
if err := os.WriteFile(p, []byte(content), 0o600); err != nil {
t.Fatal(err)
}
return p
}
for _, c := range []struct {
name, path string
want time.Duration
bad bool
}{
{"no file named", "", defaultToolchainWait, false},
{"file absent", filepath.Join(dir, "absent"), defaultToolchainWait, false},
{"empty", file(" \n"), defaultToolchainWait, false},
{"minutes", file("15\n"), 15 * time.Minute, false},
{"not a number", file("soon"), 0, true},
{"zero", file("0"), 0, true},
} {
got, err := toolchainWaitOf(c.path)
if (err != nil) != c.bad || got != c.want {
t.Errorf("%s: %v, %v; want %v, error %v", c.name, got, err, c.want, c.bad)
}
}
}
@@ -13,7 +13,6 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-builder/internal/build"
@@ -306,22 +305,3 @@ func (f *floorWake) wake(ctx context.Context, kind, seat, module string) {
func conditionModules(id string) []string {
return strings.FieldsFunc(id, func(r rune) bool { return r == '/' || r == '.' || r == ':' })
}
// doneVia says whether a build key is done, read from the builder's own builds state: the build's file is held.
func doneVia(b *build.Builder) func(ctx context.Context, module, fp string) (bool, error) {
return func(ctx context.Context, module, fp string) (bool, error) {
key, err := builds.BuilderBuildKey(module, fp)
if err != nil {
return false, err
}
raw, _, found, err := b.Claimer.Store.Get(ctx, key)
if err != nil || !found {
return false, err
}
cur, err := builds.DecodeBuilderBuild(raw)
if err != nil {
return false, err
}
return cur.State == builds.BuildDone, nil
}
}