B1 (builder): the builder's own buckets are named from its credential's module
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
This commit is contained in:
@@ -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" }
|
||||
|
||||
@@ -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
|
||||
// <node>.<language>, 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.
|
||||
|
||||
@@ -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 `<module>_<state>` 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),
|
||||
}
|
||||
}
|
||||
@@ -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{
|
||||
|
||||
@@ -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 `<module>_<state>` 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) }()
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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, `<module>_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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
+4
-3
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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 `<module>_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
|
||||
|
||||
@@ -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 != "" {
|
||||
|
||||
Reference in New Issue
Block a user