From 07c13b0b9cce688da80c32aa3b0a0548c7c17f16 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 23:32:42 +0100 Subject: [PATCH] B1 (builder): the builder's own buckets are named from its credential's module MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The versions, bounds, reserved, publish-queue, acted, toolchains-heard and instances buckets were constants naming mesh-delivery-builder; ownBucketsOf takes them from the module the credential names. Builds: design 47 G A-296, §428 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-builder/buildrun/names.go | 2 +- .../buildrun/toolchains.go | 5 +--- .../cmd/mesh-delivery-builder/buckets.go | 22 +++++++++++++++ .../mesh-delivery-builder/integration_test.go | 10 +++---- .../cmd/mesh-delivery-builder/main.go | 28 ++++++++++--------- .../cmd/mesh-delivery-builder/names_test.go | 7 +++-- .../mesh-delivery-builder/publishbounds.go | 4 +-- .../cmd/mesh-delivery-builder/readscall.go | 2 +- .../cmd/mesh-delivery-builder/reserve.go | 2 +- .../cmd/mesh-delivery-builder/says.go | 20 +++++++------ .../cmd/mesh-delivery-builder/says_test.go | 2 +- .../cmd/mesh-delivery-builder/stepbounds.go | 2 +- .../stepbounds_integration_test.go | 7 +++-- .../cmd/mesh-delivery-builder/stepsread.go | 2 +- .../cmd/mesh-delivery-builder/unput.go | 2 +- .../cmd/mesh-delivery-builder/versions.go | 2 +- .../internal/versions/claim.go | 2 +- .../internal/versions/publish.go | 3 +- .../internal/versions/republish.go | 12 ++++---- .../internal/versions/reserve.go | 2 -- .../internal/versions/versions.go | 5 ++-- .../internal/versions/versions_test.go | 2 +- 22 files changed, 83 insertions(+), 62 deletions(-) create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/buckets.go diff --git a/modules/mesh-delivery-builder/buildrun/names.go b/modules/mesh-delivery-builder/buildrun/names.go index 2410e40b..5db2d792 100644 --- a/modules/mesh-delivery-builder/buildrun/names.go +++ b/modules/mesh-delivery-builder/buildrun/names.go @@ -7,5 +7,5 @@ package buildrun func StepsBucket(module string) string { return module + "_steps" } // ToolchainsBucket is the bucket of a build module's state toolchains, which the checker and the indexer read (the builder -// hears the toolchain-reported events and keeps ToolchainsHeardBucket instead; design 47 G §492, §604). +// hears the toolchain-reported events and keeps the state ToolchainsHeardState instead; design 47 G §492, §604). func ToolchainsBucket(module string) string { return module + "_toolchains" } diff --git a/modules/mesh-delivery-builder/buildrun/toolchains.go b/modules/mesh-delivery-builder/buildrun/toolchains.go index 755dfe4a..85e144c5 100644 --- a/modules/mesh-delivery-builder/buildrun/toolchains.go +++ b/modules/mesh-delivery-builder/buildrun/toolchains.go @@ -9,10 +9,7 @@ import ( // ToolchainsHeardState is the builder's old state section of the toolchains it heard (design 47 G §604, §546): one key // ., the toolchain as the build seat said it. -const ( - ToolchainsHeardState = "toolchains-heard" - ToolchainsHeardBucket = "mesh-delivery-builder_" + ToolchainsHeardState -) +const ToolchainsHeardState = "toolchains-heard" // ToolchainSet holds each node's toolchain of a language, as the builder heard it (design 47 G §604). Safe for use from // the consumer and from a build at once. diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/buckets.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/buckets.go new file mode 100644 index 00000000..083d6993 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/buckets.go @@ -0,0 +1,22 @@ +package main + +import ( + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "mesh-delivery-builder/internal/versions" +) + +// ownBuckets are the buckets of the builder's own states, each `_` of the module its credential names, never +// a constant (design 47 G A-296, §428). +type ownBuckets struct { + versions, bounds, reserved, queue, acted, toolchainsHeard, instances string +} + +func ownBucketsOf(module string) ownBuckets { + return ownBuckets{ + versions: builds.StateBucket(module, versions.State), bounds: builds.StateBucket(module, versions.BoundsState), + reserved: builds.StateBucket(module, versions.ReservedState), queue: builds.StateBucket(module, versions.PublishQueueState), + acted: builds.StateBucket(module, versions.ActedState), toolchainsHeard: builds.StateBucket(module, buildrun.ToolchainsHeardState), + instances: builds.StateBucket(module, instancesState), + } +} 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 aaf40603..557106b6 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 @@ -264,7 +264,7 @@ func startBedWith(t *testing.T, closed bool) *bed { t.Fatal(err) } } - if kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: buildrun.ToolchainsHeardBucket, History: 1}); err != nil { + if kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: builds.StateBucket(builds.BuilderModule, buildrun.ToolchainsHeardState), History: 1}); err != nil { t.Fatal(err) } else { b.heard = kv @@ -294,9 +294,9 @@ func startBedWith(t *testing.T, closed bool) *bed { } viewstest.Serve(t, ctx, js, ctl, testAssigner, "assignments", testAssigner+".assignments", readers...) b.builds = mk(builds.BuildsBucket(builds.BuilderModule), builds.BuilderBuildsHistory) - b.instances = mk(instancesBucket, 1) + b.instances = mk(builds.StateBucket(builds.BuilderModule, instancesState), 1) // The state bounds, whose keys have each its own time to live: an ask's step bounds (design 47 G A-77, A-118) are kept in it. - if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: versions.BoundsBucket, History: 1, LimitMarkerTTL: time.Second}); err != nil { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: builds.StateBucket(builds.BuilderModule, versions.BoundsState), History: 1, LimitMarkerTTL: time.Second}); err != nil { t.Fatal(err) } b.out = &saidLog{} @@ -618,7 +618,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{b: process.AwaitBucket(b.ctx, b.js, nil, versions.Bucket, "", t.Logf)}} + store := versions.Versions{Store: awaitedVersions{b: process.AwaitBucket(b.ctx, b.js, nil, builds.StateBucket(builds.BuilderModule, versions.State), "", t.Logf)}} numbers := make(chan string, 8) var claims sync.WaitGroup for i := range 8 { @@ -1121,7 +1121,7 @@ func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *regis } return kv } - r.versions, r.findings, r.reasks = mk(versions.Bucket), mk(builds.StateBucket(builds.BuilderModule, mcseat.StateFindings)), mk(builds.StateBucket(builds.BuilderModule, build.ReAsksState)) + r.versions, r.findings, r.reasks = mk(builds.StateBucket(builds.BuilderModule, versions.State)), mk(builds.StateBucket(builds.BuilderModule, mcseat.StateFindings)), mk(builds.StateBucket(builds.BuilderModule, build.ReAsksState)) mk(builds.StateBucket(builds.BuilderModule, versions.FloorWaitsState)) // EVENTS with two seats' outcomes, and the forge bridge's stream (a branch's end): each with the builder's durable consumer. for stream, subjects := range map[string][]string{ 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 844d2028..7e5219cf 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -118,6 +118,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro half.branches = branches // The build modules are no names (design 47 G A-82): the holders of the build seats, the seat list's definitions and the // catalog half's claims, found at each use. + // The part's own buckets are `_` of the credential's module (design 47 G A-296). + own := ownBucketsOf(cred.Module) seatGate := newSeatGate() // The part's one holder function (A-298): the seat list's callers and the views found by their seat (§717) read it. holders := branches.SeatHolders(ctx) @@ -156,7 +158,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // (the seat verb held, novox/hq issue 1937) before it leaves the key done, on its calls connection. // The builder numbers each trunk build at its claim (novox/hq design 51 *The versions*): its state versions, bound once // the bus holds it; the floor's three sources asked as its own account. - versionsKV := process.AwaitBucket(ctx, js, conn, versions.Bucket, "no trunk build is numbered: each is built unnumbered and published by its digest", logger.Printf) + versionsKV := process.AwaitBucket(ctx, js, conn, own.versions, "no trunk build is numbered: each is built unnumbered and published by its digest", logger.Printf) found := &findings{module: cred.Module, w: fw, logf: logger.Printf, now: time.Now} floorWaits := &versions.FloorWaits{Holder: versions.HolderOf(branches), Logf: logger.Printf, Said: found.floorSaid(context.WithoutCancel(ctx))} // The names waiting on their floor are kept in the state floor-waits (DLV-B4): a restarted builder takes them in, so @@ -170,7 +172,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Names: versions.NamesFrom(versions.EntriesOf(branches))} // 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) + pubBoundsKV := process.AwaitBucket(ctx, js, conn, own.bounds, "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 @@ -217,8 +219,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro waitsBucket.Ready(func(kv jetstream.KeyValue) { b.Waits.(*waitsKept).start(ctx, kv) }) // 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) + reservedKV := process.AwaitBucket(ctx, js, conn, own.reserved, "no version is reserved: builds are made unreserved", logger.Printf) + reserved := newReservations(awaitedReserved{awaitedVersions{b: reservedKV, bucket: own.reserved}}, calls.Call, pubBoundsKV.KV, found, b.Reconsider, logger.Printf) // Each artifact the holder says published at a reserved version is said as the seat mesh-builds' package-published (design 47 G §52). reserved.say = func(ctx context.Context, p meshbuilds.PackagePublished) error { return emitSeatEvent(context.WithoutCancel(ctx), js, cred, meshbuilds.SubjectPackagePublished, "package-published-"+p.Module+"-"+p.Version+"-"+p.Artifact, p) @@ -234,16 +236,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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), + queueKV := process.AwaitBucket(ctx, js, conn, own.queue, "no version is published again on an event or by builder_publish_again", logger.Printf) + actedKV := process.AwaitBucket(ctx, js, conn, own.acted, "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: own.queue}}}, + Acted: awaitedReserved{awaitedVersions{b: actedKV, bucket: own.acted}}, 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}} + retakeActed := awaitedReserved{awaitedVersions{b: actedKV, bucket: own.acted}} asker := &operatorask.Asker[oc.Warrant]{Module: cred.Module, Call: calls.Call, Record: calls.Ask, Publish: publishVia(ctx, js), Subscribe: decided.hear(logger.Printf), Contract: operatorask.Contract{Seat: oc.Seat, StateAsks: oc.StateAsks}, AskSubject: oc.AcceptSubjectAsk, DecidedSubject: oc.SubjectDecided} @@ -282,10 +284,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // controller delivers from this builder's keys only while an instance says it writes references. A bucket not there // yet (the manifest's state not made) is said, and the controller's plan waits for the keys, named (no build-agent // is asked since R4g, ADR 0524 §7). - switch kv := process.OptionalBucket(ctx, js, instancesBucket, "the controller is not told that this builder writes references until it starts again", logger.Printf); { + switch kv := process.OptionalBucket(ctx, js, own.instances, "the controller is not told that this builder writes references until it starts again", logger.Printf); { case kv == nil: default: - says := newSayer(ctx, kv, cred.Node, seats.Modules, logger.Printf) + says := newSayer(ctx, kv, own.instances, cred.Node, seats.Modules, logger.Printf) b.Answered = says.Answered defer func() { stopping, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) @@ -301,7 +303,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro heardTools := newToolchainsHeard(tools, b.NodesChanged, func(ctx context.Context, e meshbuilds.ToolchainAgreed) error { return emitSeatEvent(context.WithoutCancel(ctx), js, cred, meshbuilds.SubjectToolchainAgreed, "toolchain-agreed-"+e.Language+"-"+e.Version, e) }, logger) - process.AwaitBucket(ctx, js, conn, buildrun.ToolchainsHeardBucket, "the toolchains heard are kept in memory only, and builder_toolchains shows none", logger.Printf). + process.AwaitBucket(ctx, js, conn, own.toolchainsHeard, "the toolchains heard are kept in memory only, and builder_toolchains shows none", logger.Printf). Ready(func(kv jetstream.KeyValue) { heardTools.bind(ctx, kvHeard{kvKeeper{ctx: ctx, kv: kv}}) heardTools.readRunning(ctx, js) @@ -311,7 +313,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } if w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()); err != nil { - process.NotWatched(logger.Printf, buildrun.ToolchainsHeardBucket, err, "builder_toolchains_drop is not heard until the builder starts again") + process.NotWatched(logger.Printf, own.toolchainsHeard, err, "builder_toolchains_drop is not heard until the builder starts again") } else { go func() { errs <- follow(ctx, w, dropped) }() } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/names_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/names_test.go index 8a3e6a0d..e37af399 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/names_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/names_test.go @@ -44,8 +44,11 @@ func TestTheNamesAreTheManifests(t *testing.T) { } } for want, got := range map[string]string{ - "mesh-build-go_steps": buildrun.StepsBucket("mesh-build-go"), - buildrun.ToolchainsHeardBucket: "mesh-delivery-builder_toolchains-heard", + "mesh-build-go_steps": buildrun.StepsBucket("mesh-build-go"), + ownBucketsOf("acme-builder").toolchainsHeard: "acme-builder_toolchains-heard", + ownBucketsOf("acme-builder").bounds: "acme-builder_bounds", + ownBucketsOf("acme-builder").versions: "acme-builder_versions", + ownBucketsOf("acme-builder").instances: "acme-builder_instances", } { if want != got { t.Errorf("a bucket is %q, want %q", got, want) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go index 9f5cf00e..cb9628e1 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/publishbounds.go @@ -33,7 +33,7 @@ func (p *publishes) arm(ctx context.Context, e versions.Entry) { kv := p.bounds() var err error if kv == nil { - err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket) + err = fmt.Errorf("%s is not on the bus yet", versions.BoundsState) } else if _, err = kv.Create(ctx, versions.PublishBoundKey(e.Name, e.Version), versions.EncodeBound("the publish of "+e.Name+"@"+e.Version+" to "+e.Seat, time.Now().Add(p.bound)), jetstream.KeyTTL(p.bound)); errors.Is(err, jetstream.ErrKeyExists) { err = nil } @@ -120,7 +120,7 @@ func watchBounds(ctx context.Context, kv jetstream.KeyValue, each func(jetstream const consequence = "a publish no holder answers, and a node holding the old form seven days, are not raised until the builder starts again" pw, err := settle(ctx, kv, each) if err != nil { - process.NotWatched(logf, versions.BoundsBucket, err, consequence) + process.NotWatched(logf, versions.BoundsState, err, consequence) w.NotHeard(versions.BoundsState, consequence, err) return } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go index 2cd14157..cf4192a2 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/readscall.go @@ -55,7 +55,7 @@ func (r *engineReads) callsOf(f *findings, bounds func() jetstream.KeyValue, hea Arm: func(ctx context.Context, node string, ends time.Time) error { kv := bounds() if kv == nil { - return errors.New(versions.BoundsBucket + " is not on the bus yet") + return errors.New(versions.BoundsState + " is not on the bus yet") } left := max(time.Until(ends), time.Second) _, err := kv.Create(ctx, enginereads.BoundPrefix+node, versions.EncodeBound("the manifest-reads call to "+node, ends), jetstream.KeyTTL(left)) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve.go index 1582b99d..6fd3ced9 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/reserve.go @@ -103,7 +103,7 @@ 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) + err = fmt.Errorf("%s is not on the bus yet", versions.BoundsState) } 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 } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says.go index 242cacbe..600e023e 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says.go @@ -16,7 +16,7 @@ import ( // says so; since R4g it asks no build-agent, and a module whose key is not yet done waits, named (ADR 0524 §7). The bucket is read by the controller alone, so its name and shape // are not in mesh-sdk. const ( - instancesBucket = "mesh-delivery-builder_instances" + instancesState = "instances" writesReferences = `{"v":1,"writes":["references"]}` ) @@ -30,17 +30,19 @@ type sayStore interface { type sayer struct { ctx context.Context store sayStore - node string - need func() []string - logf func(string, ...any) + // bucket is the bucket of instances, `_instances` of the credential's module, for the log lines. + bucket string + node string + need func() []string + logf func(string, ...any) mu sync.Mutex answered map[string]bool said bool } -func newSayer(ctx context.Context, store sayStore, node string, need func() []string, logf func(string, ...any)) *sayer { - return &sayer{ctx: ctx, store: store, node: node, need: need, logf: logf, answered: map[string]bool{}} +func newSayer(ctx context.Context, store sayStore, bucket, node string, need func() []string, logf func(string, ...any)) *sayer { + return &sayer{ctx: ctx, store: store, bucket: bucket, node: node, need: need, logf: logf, answered: map[string]bool{}} } // Answered is told the build module whose step answered a reference (build.Builder.Answered). @@ -57,11 +59,11 @@ func (s *sayer) Answered(module string) { } } if _, err := s.store.Put(s.ctx, s.node, []byte(writesReferences)); err != nil { - s.logf("saying in %s that this builder writes references: %v; said again at the next reference", instancesBucket, err) + s.logf("saying in %s that this builder writes references: %v; said again at the next reference", s.bucket, err) return } s.said = true - s.logf("every build module answered a reference: said in %s that this builder writes references", instancesBucket) + s.logf("every build module answered a reference: said in %s that this builder writes references", s.bucket) } // unsay takes the say away, if it was made: a builder replaced by one that does not write references leaves none. @@ -72,7 +74,7 @@ func (s *sayer) unsay(ctx context.Context) error { return nil } if err := s.store.Delete(ctx, s.node); err != nil { - return fmt.Errorf("taking away the say in %s: %w", instancesBucket, err) + return fmt.Errorf("taking away the say in %s: %w", s.bucket, err) } s.said = false return nil diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says_test.go index ce0f9f90..e8c15df6 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/says_test.go @@ -24,7 +24,7 @@ func (s saysOf) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt // one; stopped, it says nothing any more. func TestTheBuilderSaysItWritesReferencesOnceEveryBuildModuleAnsweredOne(t *testing.T) { s := saysOf{} - say := newSayer(t.Context(), s, "shanks", func() []string { return []string{"mesh-build-go", "mesh-build-node", "mesh-build-image"} }, t.Logf) + say := newSayer(t.Context(), s, "mesh-delivery-builder_instances", "shanks", func() []string { return []string{"mesh-build-go", "mesh-build-node", "mesh-build-image"} }, t.Logf) say.Answered("mesh-build-go") say.Answered("mesh-build-node") if v, ok := s["shanks"]; ok { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go index 1ea37585..767dcc8f 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds.go @@ -31,7 +31,7 @@ type stepBounds struct { func (s *stepBounds) Arm(ctx context.Context, key, why string, bound time.Duration) error { kv := s.bounds() if kv == nil { - return fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket) + return fmt.Errorf("%s is not on the bus yet", versions.BoundsState) } ends := s.now().Add(bound) value := versions.EncodeBound(why, ends) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds_integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds_integration_test.go index 9aad7b9f..6cec6f38 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds_integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepbounds_integration_test.go @@ -4,6 +4,7 @@ package main import ( "encoding/json" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "sync" "testing" "time" @@ -20,7 +21,7 @@ import ( // it once. func TestTwoWaitersOnOneAskArmItsStepBoundOnce(t *testing.T) { b := startBed(t) - kv, err := b.js.KeyValue(b.ctx, versions.BoundsBucket) + kv, err := b.js.KeyValue(b.ctx, builds.StateBucket(builds.BuilderModule, versions.BoundsState)) if err != nil { t.Fatal(err) } @@ -79,11 +80,11 @@ func TestTwoWaitersOnOneAskArmItsStepBoundOnce(t *testing.T) { if e.Revision() != last.Revision()+1 || !v.Ends.Equal(clock().Add(bound)) || e.Operation() != jetstream.KeyValuePut { t.Fatalf("movement %d: revision %d (was %d), ends %s, %s: the bound does not count from the step's movement", move+1, e.Revision(), last.Revision(), v.Ends, e.Operation()) } - msg, err := b.js.Stream(b.ctx, "KV_"+versions.BoundsBucket) + msg, err := b.js.Stream(b.ctx, "KV_"+builds.StateBucket(builds.BuilderModule, versions.BoundsState)) if err != nil { t.Fatal(err) } - raw, err := msg.GetLastMsgForSubject(b.ctx, "$KV."+versions.BoundsBucket+"."+key) + raw, err := msg.GetLastMsgForSubject(b.ctx, "$KV."+builds.StateBucket(builds.BuilderModule, versions.BoundsState)+"."+key) if err != nil || raw.Header.Get("Nats-TTL") == "" { t.Fatalf("movement %d: the key written again keeps no time to live (%v, %v)", move+1, raw, err) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go index 378a8db2..6ce90f94 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stepsread.go @@ -45,7 +45,7 @@ func (s *stepsRead) unanswered(module, askID string, fault error) { ctx := context.Background() var err error if kv := s.bounds(); kv == nil { - err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket) + err = fmt.Errorf("%s is not on the bus yet", versions.BoundsState) } 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 } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/unput.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/unput.go index 19817b41..ffd2e778 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/unput.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/unput.go @@ -39,7 +39,7 @@ const unputCallee = "the put of its asks on the build seats and the write of its func (u *unputs) left(ctx context.Context, key, why string) { var err error if kv := u.bounds(); kv == nil { - err = fmt.Errorf("%s is not on the bus yet", versions.BoundsBucket) + err = fmt.Errorf("%s is not on the bus yet", versions.BoundsState) } else if _, err = kv.Create(ctx, versions.UnputBoundKey(key), versions.EncodeBound("the build key "+key+" left building unput: "+why, u.now().Add(u.bound)), jetstream.KeyTTL(u.bound)); errors.Is(err, jetstream.ErrKeyExists) { err = nil } 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 c0976448..2d30aacb 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/versions.go @@ -28,7 +28,7 @@ type awaitedVersions struct { bucket string } -var errVersionsMissing = errors.New("the state " + versions.Bucket + " is not on the bus yet") +var errVersionsMissing = errors.New("the state " + versions.State + " is not on the bus yet") func (s awaitedVersions) kv() (jetstream.KeyValue, error) { if kv := s.b.KV(); kv != nil { diff --git a/modules/mesh-delivery-builder/internal/versions/claim.go b/modules/mesh-delivery-builder/internal/versions/claim.go index 063e8abe..b5411323 100644 --- a/modules/mesh-delivery-builder/internal/versions/claim.go +++ b/modules/mesh-delivery-builder/internal/versions/claim.go @@ -174,7 +174,7 @@ func (n *Numberer) Number(ctx context.Context, c BuildClaim) (Numbering, error) if err != nil && kind == KindPackage { // A build that cannot read the published packages refuses, naming the unread state, never building against a // guess (ADR 0666 §15; design 51 *The versions*, line 287). - out.Refused = fmt.Sprintf("%s: refused: the builder's state %s (bucket %s) cannot be read, so its published versions are not known: %v", name, State, Bucket, err) + out.Refused = fmt.Sprintf("%s: refused: the builder's state %s cannot be read, so its published versions are not known: %v", name, State, err) return out, nil } if err != nil { diff --git a/modules/mesh-delivery-builder/internal/versions/publish.go b/modules/mesh-delivery-builder/internal/versions/publish.go index 82fe99dd..8751ed50 100644 --- a/modules/mesh-delivery-builder/internal/versions/publish.go +++ b/modules/mesh-delivery-builder/internal/versions/publish.go @@ -21,8 +21,7 @@ const ( // BoundsState is the builder's state of its expiring keys, each with its own time to live: a publish's bound // (PublishBoundKey), a node's seven days holding the old form's window (OldFormBoundKey) and a build key left // building unput (UnputBoundKey). Its end, the bus's marker heard on the watch, is what acts. - BoundsState = "bounds" - BoundsBucket = "mesh-delivery-builder_bounds" + BoundsState = "bounds" // PublishBound is the build module's own wait for the holder's word (design 51 *How a publish ends*: at most // 10 minutes), so a version unanswered past it is one the holder never answered. PublishBound = 10 * time.Minute diff --git a/modules/mesh-delivery-builder/internal/versions/republish.go b/modules/mesh-delivery-builder/internal/versions/republish.go index 4a361b56..1939a23c 100644 --- a/modules/mesh-delivery-builder/internal/versions/republish.go +++ b/modules/mesh-delivery-builder/internal/versions/republish.go @@ -24,13 +24,11 @@ import ( const ( // PublishQueueState is the builder's old state section of the queues, capped at PublishQueueMiB. - PublishQueueState = "publish-queue" - PublishQueueBucket = "mesh-delivery-builder_" + PublishQueueState - PublishQueueMiB = 4 + PublishQueueState = "publish-queue" + PublishQueueMiB = 4 // ActedState holds, one key per subject, the last stream sequence the builder acted on; capped at ActedMiB. - ActedState = "acted" - ActedBucket = "mesh-delivery-builder_" + ActedState - ActedMiB = 1 + ActedState = "acted" + ActedMiB = 1 // PublishQueueTool is builder_publish_queue. PublishQueueTool = "builder_publish_queue" // DefaultRepublishMax is the setting republish-max-versions unless set (§633). @@ -368,7 +366,7 @@ func (r *Republisher) enqueue(ctx context.Context, v ReservedVersion, holder str raw, _ := json.Marshal(QueuedPublish{V: 1, Holder: holder, Module: v.Module, Version: v.Version, Artifacts: arts, Queued: r.queuedAt()}) used, err := r.Queue.Bytes(ctx) if err != nil { - return false, fmt.Errorf("the size of %s cannot be read: %w", PublishQueueBucket, err) + return false, fmt.Errorf("the size of the state %s cannot be read: %w", PublishQueueState, err) } if used+uint64(len(raw)) > PublishQueueMiB<<20 { if r.Unreleased != nil { diff --git a/modules/mesh-delivery-builder/internal/versions/reserve.go b/modules/mesh-delivery-builder/internal/versions/reserve.go index fbe2e814..eef5f337 100644 --- a/modules/mesh-delivery-builder/internal/versions/reserve.go +++ b/modules/mesh-delivery-builder/internal/versions/reserve.go @@ -25,8 +25,6 @@ import ( 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" diff --git a/modules/mesh-delivery-builder/internal/versions/versions.go b/modules/mesh-delivery-builder/internal/versions/versions.go index d84367d2..b04dd7ca 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions.go +++ b/modules/mesh-delivery-builder/internal/versions/versions.go @@ -19,10 +19,9 @@ import ( "time" ) -// State is the builder state's name; Bucket the bucket the bus keeps it in. +// State is the builder state's name; its bucket is `_versions` of the credential's module. const ( - State = "versions" - Bucket = "mesh-delivery-builder_versions" + State = "versions" // Cap is the bucket's cap in bytes (64 MiB); FullPercent the share of it past which the builder writes its finding. Cap = 64 << 20 FullPercent = 80 diff --git a/modules/mesh-delivery-builder/internal/versions/versions_test.go b/modules/mesh-delivery-builder/internal/versions/versions_test.go index 30b9f44c..da2ec6d5 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions_test.go +++ b/modules/mesh-delivery-builder/internal/versions/versions_test.go @@ -456,7 +456,7 @@ func TestTheBuildersVersions(t *testing.T) { // is refused, naming the state; a bundle's is built as before (ADR 0617 §1), the error said. st := n.Versions.Store.(*mem) st.down = errors.New("nats: no responders available for request") - if got, err := n.Number(ctx, pkg); err != nil || !strings.Contains(got.Refused, "state versions (bucket mesh-delivery-builder_versions) cannot be read") || len(got.Versions) != 0 { + if got, err := n.Number(ctx, pkg); err != nil || !strings.Contains(got.Refused, "state versions cannot be read") || len(got.Versions) != 0 { t.Errorf("a package with its versions unread = %+v, %v; want refused naming the state", got, err) } if got, err := n.Number(ctx, c); err == nil || got.Refused != "" {