A-181: the assigner reads the builder's build ends from their stream, kept last per subject, and reads no builder bucket
The builder's `build.ended` stream declares last-per-subject; the grants part composes it one message per subject and its readers read whole; the assigner reads it whole at its start then follows it (consumes `build.ended.**`), so what was built and a failed build's error survive its restart. world.built, buildNeeds, manifestOf and the put verbs read the ends; the manifest reads no builder bucket; assigner_put's fingerprint-prefix form goes. Builds: design 47 G A-181, A-162, §556, §703 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -1,26 +1,25 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
"mesh-delivery-assigner/internal/assignment"
|
||||
)
|
||||
|
||||
// What a want names of a build is heard, never read from the builder's state (design 47 G §379, §526, A-94, A-96; ADR 0596 §2): the
|
||||
// builder's event `build.ended` (a done build's parts by digest, its keeping node and its branch) and the catalog's
|
||||
// `version-released` (a released version and each part's digest, with the source revision it was built from), each kept as heard.
|
||||
|
||||
// buildEndedSubjects is the builder's event build.ended on its own stream, as the manifest's consumes names it
|
||||
// (`mesh-delivery-builder.build.ended`), each module and fingerprint after it.
|
||||
var buildEndedSubjects = process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEvent+".>")
|
||||
|
||||
// builtEnds are the done builds' ends heard, by module and fingerprint (hearBuildEnded).
|
||||
type builtEnds struct {
|
||||
mu sync.Mutex
|
||||
@@ -115,21 +114,46 @@ func (r *releasedVersions) of(module string, source int64) (assign.KeptBuild, bo
|
||||
return k, ok
|
||||
}
|
||||
|
||||
// hearBuildsAndReleases registers the hub's takers of build.ended and version-released, each waking the pass; before the hub starts.
|
||||
func hearBuildsAndReleases(hub *process.EventHub, ends *builtEnds, released *releasedVersions, wake func(), logf func(string, ...any)) error {
|
||||
taker := func(put func([]byte) error) func(jetstream.Msg) {
|
||||
return func(m jetstream.Msg) {
|
||||
if err := put(m.Data()); err != nil {
|
||||
logf("%s is not kept: %v", m.Subject(), err)
|
||||
}
|
||||
if err := m.Ack(); err != nil {
|
||||
logf("acknowledging %s: %v", m.Subject(), err)
|
||||
}
|
||||
wake()
|
||||
// hearReleases registers the hub's taker of version-released, waking the pass; before the hub starts.
|
||||
func hearReleases(hub *process.EventHub, released *releasedVersions, wake func(), logf func(string, ...any)) error {
|
||||
return hub.Handle(meshcatalog.SubjectVersionReleased, func(m jetstream.Msg) {
|
||||
if err := released.put(m.Data()); err != nil {
|
||||
logf("%s is not kept: %v", m.Subject(), err)
|
||||
}
|
||||
if err := m.Ack(); err != nil {
|
||||
logf("acknowledging %s: %v", m.Subject(), err)
|
||||
}
|
||||
wake()
|
||||
})
|
||||
}
|
||||
|
||||
// buildsStreamName is the builder's stream of build ends as the bus names it: kept last per subject (one end per module and
|
||||
// fingerprint, its manifest's `last-per-subject`, design 47 G A-181, §703).
|
||||
var buildsStreamName = catalogue.StreamBusName(builds.BuildsStream)
|
||||
|
||||
// hearBuilds reads the builder's stream of build ends whole at the start and then follows it (design 47 G A-181, §556): what was
|
||||
// built, and a failed build's error, survive the assigner's restart, and it reads none of the builder's buckets. The read ends before
|
||||
// it returns, so the first pass knows every build; a stream it cannot hear is said, and no want names a build until the assigner
|
||||
// starts again.
|
||||
func hearBuilds(ctx context.Context, js jetstream.JetStream, ends *builtEnds, wake func(), findings assign.Findings, logf func(string, ...any)) {
|
||||
acct := process.Account{Node: thisNode, Module: assignment.Module}
|
||||
nh := notHeardWriter(findings, logf)
|
||||
cc, err := acct.HearStream(ctx, js, buildsStreamName, func(m jetstream.Msg) {
|
||||
if err := ends.put(m.Data()); err != nil {
|
||||
logf("%s is not kept: %v", m.Subject(), err)
|
||||
}
|
||||
if err := m.Ack(); err != nil {
|
||||
logf("acknowledging %s: %v", m.Subject(), err)
|
||||
}
|
||||
wake()
|
||||
}, logf)
|
||||
if err != nil {
|
||||
nh.NotHeard(buildsStreamName, "what the builder built is not heard until the assigner starts again, so no want names a build's parts or its keeping node and no put goes on", err)
|
||||
return
|
||||
}
|
||||
if err := hub.Handle(buildEndedSubjects, taker(ends.put)); err != nil {
|
||||
return err
|
||||
}
|
||||
return hub.Handle(meshcatalog.SubjectVersionReleased, taker(released.put))
|
||||
nh.Heard(buildsStreamName)
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
cc.Stop()
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
)
|
||||
|
||||
// endDone is the builder's `build.ended` of a done build of module at fp heard by w, naming the digest of its manifest (design 47 G
|
||||
// A-181: what was built is heard, never read from the builder's bucket).
|
||||
func endDone(t testing.TB, w *world, module, fp, manifest string) {
|
||||
t.Helper()
|
||||
data, err := json.Marshal(builds.BuildEnded{Module: module, Fingerprint: fp, Outcome: builds.BuildDone, Manifest: manifest})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := w.ends.put(data); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
@@ -186,6 +186,15 @@ func startBed(t *testing.T, leaveOut string) *bed {
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, nodeengine.StreamRunning, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The builder's stream of build ends, kept last per subject (its manifest's `last-per-subject`, design 47 G A-181), which the assigner
|
||||
// reads whole at its start and follows through its durable, as the grants part makes it (from the last of each subject).
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buildsStreamName, Subjects: []string{process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEvent+".>")}, MaxMsgsPerSubject: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateOrUpdateConsumer(ctx, buildsStreamName, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy,
|
||||
FilterSubjects: []string{process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEvent+".>")}, DeliverPolicy: jetstream.DeliverLastPerSubjectPolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The assigner's own stream of its findings' events (design 47 G §383), which its finding writer says into.
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNER_FINDINGS", Subjects: []string{"mesh.stream.mesh-delivery-assigner.event.finding-written.*", "mesh.stream.mesh-delivery-assigner.event.finding-cleared.*", "mesh.stream.mesh-delivery-assigner.event.findings-stated.*"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -414,6 +423,21 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
|
||||
}
|
||||
|
||||
// putJSON writes a value into one of the bed's buckets.
|
||||
// putBuild says the builder's `build.ended` of the build a builder's key and value (the old form of a done build, as the tests write it) name,
|
||||
// on the builder's stream (design 47 G A-181): the assigner hears what was built, it reads no bucket of the builder.
|
||||
func (b *bed) putBuild(t *testing.T, key string, value []byte) {
|
||||
t.Helper()
|
||||
module, fp, err := builds.ParseBuilderBuildKey(key)
|
||||
built, derr := builds.DecodeBuilderBuild(value)
|
||||
if err != nil || derr != nil {
|
||||
t.Fatalf("%s: %v %v", key, err, derr)
|
||||
}
|
||||
raw, _ := json.Marshal(builds.BuildEnded{Module: module, Fingerprint: fp, Outcome: built.State, Error: built.Error, Manifest: built.Manifest, Needs: built.Needs})
|
||||
if _, err := b.js.Publish(b.ctx, process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEventName(module, fp)), raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (b *bed) putJSON(t *testing.T, bucket, key string, value []byte) {
|
||||
t.Helper()
|
||||
if _, err := b.buckets[bucket].Put(b.ctx, key, value); err != nil {
|
||||
@@ -529,7 +553,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bkey, _ := builds.BuilderBuildKey("web", fNew)
|
||||
b.putJSON(t, buckets.Builds, bkey, built)
|
||||
b.putBuild(t, bkey, built)
|
||||
for _, fp := range []string{fNew, fOld} {
|
||||
gkey, _ := grantsstate.PartGrantKey("web", fp)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
@@ -679,7 +703,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
for _, fp := range fps {
|
||||
built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: row.digest})
|
||||
bkey, _ := builds.BuilderBuildKey(module, fp)
|
||||
b.putJSON(t, buckets.Builds, bkey, built)
|
||||
b.putBuild(t, bkey, built)
|
||||
gkey, _ := grantsstate.PartGrantKey(module, fp)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
b.putJSON(t, buckets.Grants, gkey, grant)
|
||||
@@ -755,7 +779,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bkey, _ := builds.BuilderBuildKey("forms", fp)
|
||||
b.putJSON(t, buckets.Builds, bkey, built)
|
||||
b.putBuild(t, bkey, built)
|
||||
gkey, _ := grantsstate.PartGrantKey("forms", fp)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
b.putJSON(t, buckets.Grants, gkey, grant)
|
||||
@@ -979,7 +1003,7 @@ func TestTheAssignerReplacedResumesWithNothingLostOrRepeated(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bkey, _ := builds.BuilderBuildKey(name, fp)
|
||||
b.putJSON(t, buckets.Builds, bkey, built)
|
||||
b.putBuild(t, bkey, built)
|
||||
gkey, _ := grantsstate.PartGrantKey(name, fp)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
b.putJSON(t, buckets.Grants, gkey, grant)
|
||||
@@ -1022,7 +1046,7 @@ func (b *bed) fixture(t *testing.T, name, fp, manifest string) {
|
||||
b.putModule(t, name, mod)
|
||||
built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest})
|
||||
bkey, _ := builds.BuilderBuildKey(name, fp)
|
||||
b.putJSON(t, buckets.Builds, bkey, built)
|
||||
b.putBuild(t, bkey, built)
|
||||
gkey, _ := grantsstate.PartGrantKey(name, fp)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
b.putJSON(t, buckets.Grants, gkey, grant)
|
||||
@@ -1939,3 +1963,52 @@ func serveCatalog(t *testing.T, b *bed, nc *nats.Conn) {
|
||||
}
|
||||
subscribed(t, nc)
|
||||
}
|
||||
|
||||
// A-181: the assigner reads the builder's stream of build ends whole at its start (the last of each fingerprint, a failed one's error
|
||||
// among them) though its durable is made after them, then follows it: what was built survives its restart, and it reads no builder bucket.
|
||||
func TestTheAssignerReadsTheBuildersEndsWholeAtItsStart(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
thisNode = "n1"
|
||||
b.putBuild(t, mustBuildKey(t, "web", "src2:"+strings.Repeat("a", 64)), mustEncodeBuild(t, builds.BuilderBuild{State: builds.BuildFailed, Error: "first try"}))
|
||||
b.putBuild(t, mustBuildKey(t, "web", "src2:"+strings.Repeat("a", 64)), mustEncodeBuild(t, builds.BuilderBuild{State: builds.BuildDone, Manifest: strings.Repeat("d", 64)}))
|
||||
b.putBuild(t, mustBuildKey(t, "api", "src2:"+strings.Repeat("b", 64)), mustEncodeBuild(t, builds.BuilderBuild{State: builds.BuildFailed, Error: "transport fault: x"}))
|
||||
var ends builtEnds
|
||||
woken := make(chan struct{}, 8)
|
||||
hearBuilds(b.ctx, b.js, &ends, func() { woken <- struct{}{} }, assign.Findings{}, t.Logf)
|
||||
if e, ok := ends.endOf("web", "src2:"+strings.Repeat("a", 64)); !ok || e.Outcome != builds.BuildDone || e.Manifest != strings.Repeat("d", 64) {
|
||||
t.Fatalf("the read of the stream left web's end as %+v, %v: want the last, done", e, ok)
|
||||
}
|
||||
if e, ok := ends.endOf("api", "src2:"+strings.Repeat("b", 64)); !ok || e.Outcome != builds.BuildFailed || e.Error != "transport fault: x" {
|
||||
t.Fatalf("the read of the stream left api's end as %+v, %v", e, ok)
|
||||
}
|
||||
b.putBuild(t, mustBuildKey(t, "api", "src2:"+strings.Repeat("c", 64)), mustEncodeBuild(t, builds.BuilderBuild{State: builds.BuildDone, Manifest: strings.Repeat("d", 64)}))
|
||||
deadline := time.After(10 * time.Second)
|
||||
for {
|
||||
if _, ok := ends.endOf("api", "src2:"+strings.Repeat("c", 64)); ok {
|
||||
return
|
||||
}
|
||||
select {
|
||||
case <-woken:
|
||||
case <-deadline:
|
||||
t.Fatal("an end said after the read was not followed")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func mustBuildKey(t *testing.T, module, fp string) string {
|
||||
t.Helper()
|
||||
k, err := builds.BuilderBuildKey(module, fp)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return k
|
||||
}
|
||||
|
||||
func mustEncodeBuild(t *testing.T, b builds.BuilderBuild) []byte {
|
||||
t.Helper()
|
||||
raw, err := builds.EncodeBuilderBuild(b)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
@@ -63,7 +63,6 @@ func main() {
|
||||
var readBuckets = []string{
|
||||
buckets.Grants,
|
||||
buckets.Answers,
|
||||
buckets.Builds,
|
||||
buckets.Presence,
|
||||
buckets.Have,
|
||||
buckets.Provisions,
|
||||
@@ -326,7 +325,6 @@ func (e emitting) Emit(eventType string, body any) error { return e.Conn.Emit(ev
|
||||
var watched = []string{
|
||||
buckets.Grants,
|
||||
buckets.Answers,
|
||||
buckets.Builds,
|
||||
plc.AssignmentsBucket,
|
||||
plc.ProposalsBucket,
|
||||
plc.CandidatesBucket,
|
||||
@@ -579,9 +577,10 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
logger.Printf("a pipeline's end or a registration is not heard, so the wants keep the trunk-only form until the assigner starts again: %v", err)
|
||||
}
|
||||
}
|
||||
if err := hearBuildsAndReleases(hub, &w.ends, &w.released, wake, logger.Printf); err != nil {
|
||||
logger.Printf("a build's end or a version's release is not heard, so no want names a build's parts, its keeping node or its version until the assigner starts again: %v", err)
|
||||
if err := hearReleases(hub, &w.released, wake, logger.Printf); err != nil {
|
||||
logger.Printf("a version's release is not heard, so no want names its version until the assigner starts again: %v", err)
|
||||
}
|
||||
hearBuilds(ctx, js, &w.ends, wake, findings, logger.Printf)
|
||||
hearReadsChanged(ctx, js, hub, reads, findings, gate.EventsLive, logger.Printf)
|
||||
hearRunning(ctx, js, w, wake, findings, logger.Printf)
|
||||
}
|
||||
|
||||
@@ -18,7 +18,6 @@ import (
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
wdc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
|
||||
mdc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
|
||||
mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
|
||||
@@ -560,18 +559,14 @@ func putOfTheBus(t *testing.T) {
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: mdigest, Repository: "novox/mesh-catalog", Commit: strings.Repeat("c", 40)})
|
||||
putModuleRaw(p.w, trunkBranch, "nats", mod)
|
||||
p.w.manifests[mdigest] = parsed{module: "nats", manifest: catalogue.Manifest{Module: "nats", Provides: []catalogue.Offer{{Name: "mesh-bus"}}}}
|
||||
bkey, _ := builds.BuilderBuildKey("nats", fp)
|
||||
built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest})
|
||||
world(buckets.Builds, bkey, built)
|
||||
endDone(t, p.w, "nats", fp, mdigest)
|
||||
hkey, _ := nodeengineclient.HaveKey("novox", "nats")
|
||||
before, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64), Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
|
||||
world(buckets.Have, hkey, before)
|
||||
|
||||
buildKV := mem()
|
||||
buildKV.values[bkey] = built
|
||||
// Issue 1966: newest is asked of the indexer's first view of modules, never read from its bus state.
|
||||
indexer := &fakeIndexer{views: map[string]map[string][]byte{"mesh-delivery-indexer.modules": {"mesh-catalog": mod, "nats": mod}}}
|
||||
s := newPutTools(memStore{putsState: putsKV, builds.BuilderModule + ".builds": buildKV}, func(key string, body any) (json.RawMessage, error) {
|
||||
s := newPutTools(memStore{putsState: putsKV}, func(key string, body any) (json.RawMessage, error) {
|
||||
if raw, ok, err := indexer.answer(key, body); ok {
|
||||
return raw, err
|
||||
}
|
||||
@@ -582,10 +577,10 @@ func putOfTheBus(t *testing.T) {
|
||||
answerWithin = 0
|
||||
defer func() { answerWithin = was }()
|
||||
if _, err := s.doPut(map[string]any{"node": "novox", "module": "nats", "version": "fd497ab8ffff", "why": "x"}, "shanks.node-tools"); err == nil {
|
||||
t.Error("a prefix of a fingerprint the builder never built was taken")
|
||||
t.Error("a prefix of a fingerprint was taken: a put names the whole fingerprint (A-181)")
|
||||
}
|
||||
if got, err := s.version("nats", "fd497ab80000"); err != nil || got != fp {
|
||||
t.Errorf("the prefix of the fingerprint built reads %q, %v", got, err)
|
||||
if _, err := s.version("nats", "fd497ab80000"); err == nil {
|
||||
t.Error("the prefix of the fingerprint built was read: a put names the whole fingerprint (A-181)")
|
||||
}
|
||||
if got, err := s.version("nats", "newest"); err != nil || got != fp || indexer.asks["mesh-delivery-indexer.modules"] != 2 {
|
||||
t.Errorf("newest reads %q, %v, asking %d pages of the indexer's first view (want its 2)", got, err, indexer.asks["mesh-delivery-indexer.modules"])
|
||||
@@ -1407,9 +1402,7 @@ func aWantWaitsOnTheVerbEdgesTheGraphKeeps(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
p.w.manifests[digest] = parsed{module: "server", manifest: parsedServer}
|
||||
build, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: digest})
|
||||
key, _ := builds.BuilderBuildKey("server", fp("f"))
|
||||
p.w.put(buckets.Builds, key, build)
|
||||
endDone(t, p.w, "server", fp("f"), digest)
|
||||
running := nodeengine.HaveByNodeModuleValue{Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}
|
||||
at := func(f string) nodeengine.HaveByNodeModuleValue { h := running; h.Fp = f; return h }
|
||||
served := p.served(g, assign.Nodes{Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{
|
||||
|
||||
@@ -20,7 +20,6 @@ import (
|
||||
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
|
||||
@@ -154,9 +153,7 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) {
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: mdigest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
|
||||
putModuleRaw(p.w, trunkBranch, "web", mod)
|
||||
p.w.manifests[mdigest] = parsed{module: "web", manifest: catalogue.Manifest{Module: "web"}}
|
||||
bkey, _ := builds.BuilderBuildKey("web", fp)
|
||||
built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest})
|
||||
p.w.put(buckets.Builds, bkey, built)
|
||||
endDone(t, p.w, "web", fp, mdigest)
|
||||
gkey, _ := grantsstate.PartGrantKey("web", fp)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
p.w.put(buckets.Grants, gkey, grant)
|
||||
@@ -251,9 +248,7 @@ func TestACommitOnBetaMovesOnlyTheNodeThatChoseBeta(t *testing.T) {
|
||||
mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: f, Manifest: mdigest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
|
||||
putModuleRaw(p.w, branch, "web", mod)
|
||||
p.w.manifests[mdigest] = parsed{module: "web", manifest: catalogue.Manifest{Module: "web"}}
|
||||
bkey, _ := builds.BuilderBuildKey("web", f)
|
||||
built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest})
|
||||
p.w.put(buckets.Builds, bkey, built)
|
||||
endDone(t, p.w, "web", f, mdigest)
|
||||
gkey, _ := grantsstate.PartGrantKey("web", f)
|
||||
grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{})
|
||||
p.w.put(buckets.Grants, gkey, grant)
|
||||
|
||||
@@ -14,7 +14,6 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
|
||||
mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate"
|
||||
@@ -184,14 +183,8 @@ func (p *passer) putWorld(node, module string, put puts.Put, manifest catalogue.
|
||||
w.Assigned = true
|
||||
}
|
||||
}
|
||||
if bkey, err := builds.BuilderBuildKey(module, put.FP); err == nil {
|
||||
if raw := p.w.value(buckets.Builds, bkey); raw != nil {
|
||||
if b, err := builds.DecodeBuilderBuild(raw); err == nil {
|
||||
w.Build, w.BuildWords = b.State, b.Error
|
||||
} else {
|
||||
w.Build = "unreadable: " + err.Error()
|
||||
}
|
||||
}
|
||||
if e, ok := p.w.ends.endOf(module, put.FP); ok { // the build's end heard (A-181): none heard, no state
|
||||
w.Build, w.BuildWords = e.Outcome, e.Error
|
||||
}
|
||||
w.Composed, w.Granted = p.composedFor(node, module, put.FP)
|
||||
if d := w.Composed; d != nil {
|
||||
|
||||
@@ -17,7 +17,6 @@ import (
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
|
||||
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/catalogkey"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
@@ -72,14 +71,11 @@ func (s *putTools) hear(key string, value []byte) {
|
||||
s.mu.Unlock()
|
||||
}
|
||||
|
||||
// fullFP is a fingerprint of the new form, whole; prefix at least its first twelve hex digits.
|
||||
var (
|
||||
fullFP = regexp.MustCompile(`^src2:[0-9a-f]{64}$`)
|
||||
prefix = regexp.MustCompile(`^(src2:)?[0-9a-f]{12,63}$`)
|
||||
)
|
||||
// fullFP is a fingerprint of the new form, whole.
|
||||
var fullFP = regexp.MustCompile(`^src2:[0-9a-f]{64}$`)
|
||||
|
||||
// version reads the version a put names: newest (the indexer's fingerprint at the trunk), a whole fingerprint, or the start
|
||||
// of one the builder built, which must name one build alone.
|
||||
// version reads the version a put names: newest (the indexer's fingerprint at the trunk) or a whole fingerprint (design 47 G A-181:
|
||||
// a put names the whole fingerprint, the builder's builds being no bucket to list).
|
||||
func (s *putTools) version(module, version string) (string, error) {
|
||||
v := strings.ToLower(strings.TrimSpace(version))
|
||||
switch {
|
||||
@@ -103,29 +99,8 @@ func (s *putTools) version(module, version string) (string, error) {
|
||||
return m.Fp2, nil
|
||||
case fullFP.MatchString(v):
|
||||
return v, nil
|
||||
case prefix.MatchString(v):
|
||||
start := strings.TrimPrefix(v, meshcatalogclient.SourcePrefix2)
|
||||
keys, err := s.st.keys(builds.BuilderModule + ".builds")
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the builder's builds cannot be listed: %w", err)
|
||||
}
|
||||
var found []string
|
||||
for _, k := range keys {
|
||||
m, fp, err := builds.ParseBuilderBuildKey(k)
|
||||
if err == nil && m == module && strings.HasPrefix(strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2), start) {
|
||||
found = append(found, fp)
|
||||
}
|
||||
}
|
||||
switch len(found) {
|
||||
case 0:
|
||||
return "", fmt.Errorf("the builder built no %s whose fingerprint starts %s: give it whole", module, start)
|
||||
case 1:
|
||||
return found[0], nil
|
||||
}
|
||||
sort.Strings(found)
|
||||
return "", fmt.Errorf("%s starts %d fingerprints of %s the builder built (%s): give more of it", start, len(found), module, strings.Join(found, ", "))
|
||||
}
|
||||
return "", fmt.Errorf("%q is no version: newest, a fingerprint (src2: and 64 hex digits) or at least its first 12 hex digits", version)
|
||||
return "", fmt.Errorf("%q is no version: newest, a fingerprint (src2: and 64 hex digits) whole", version)
|
||||
}
|
||||
|
||||
// putAnswer is what assigner_put and assigner_puts answer of one put.
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"mesh-delivery-assigner/internal/buckets"
|
||||
"slices"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
|
||||
@@ -94,22 +93,14 @@ func (p *passer) servesOf(module, fp string) ([]string, bool) {
|
||||
return graph.ServesOf(m), true
|
||||
}
|
||||
|
||||
// manifestOf is the manifest of module's build at fp (the builder's build names its digest in the indexer's manifests),
|
||||
// and whether it could be read.
|
||||
// manifestOf is the manifest of module's build at fp (the build's end names its digest, A-181, which the indexer's manifests
|
||||
// hold), and whether it could be read.
|
||||
func (p *passer) manifestOf(module, fp string) (catalogue.Manifest, bool) {
|
||||
key, err := builds.BuilderBuildKey(module, fp)
|
||||
if err != nil {
|
||||
e, ok := p.w.ends.endOf(module, fp)
|
||||
if !ok || e.Outcome != builds.BuildDone || e.Manifest == "" {
|
||||
return catalogue.Manifest{}, false
|
||||
}
|
||||
raw := p.w.value(buckets.Builds, key)
|
||||
if raw == nil {
|
||||
return catalogue.Manifest{}, false
|
||||
}
|
||||
b, err := builds.DecodeBuilderBuild(raw)
|
||||
if err != nil || b.Manifest == "" {
|
||||
return catalogue.Manifest{}, false
|
||||
}
|
||||
got := p.w.manifest(b.Manifest)
|
||||
got := p.w.manifest(e.Manifest)
|
||||
if got.err != "" {
|
||||
return catalogue.Manifest{}, false
|
||||
}
|
||||
|
||||
@@ -459,20 +459,10 @@ func (w *world) has(bucket, key string) bool {
|
||||
return ok
|
||||
}
|
||||
|
||||
// built says whether the builder's build of a module at a fingerprint is done.
|
||||
// built says whether the builder's build of a module at a fingerprint is done, from its `build.ended` heard (design 47 G A-181).
|
||||
func (w *world) built(module, fp string) bool {
|
||||
key, err := builds.BuilderBuildKey(module, fp)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
w.mu.Lock()
|
||||
value, ok := w.values[buckets.Builds][key]
|
||||
w.mu.Unlock()
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
b, err := builds.DecodeBuilderBuild(value)
|
||||
return err == nil && b.State == builds.BuildDone
|
||||
e, ok := w.ends.endOf(module, fp)
|
||||
return ok && e.Outcome == builds.BuildDone
|
||||
}
|
||||
|
||||
// buildStanding says whether the builder's `build.ended` of module at fp said the build failed, and its error whole (design 47 G
|
||||
@@ -484,18 +474,13 @@ func (w *world) buildStanding(module, fp string) (failed bool, words string) {
|
||||
return false, ""
|
||||
}
|
||||
|
||||
// buildNeeds are the reference forms the builder's done build of module at fp is written in (the build key's needs, ADR
|
||||
// 0618 §1): nil when it is not done or names none.
|
||||
// buildNeeds are the reference forms the builder's done build of module at fp is written in (the build's end says them, ADR
|
||||
// 0618 §1, A-181): nil when it is not done or names none.
|
||||
func (w *world) buildNeeds(module, fp string) []string {
|
||||
key, err := builds.BuilderBuildKey(module, fp)
|
||||
if err != nil {
|
||||
return nil
|
||||
if e, ok := w.ends.endOf(module, fp); ok && e.Outcome == builds.BuildDone {
|
||||
return e.Needs
|
||||
}
|
||||
b, err := builds.DecodeBuilderBuild(w.value(buckets.Builds, key))
|
||||
if err != nil || b.State != builds.BuildDone {
|
||||
return nil
|
||||
}
|
||||
return b.Needs
|
||||
return nil
|
||||
}
|
||||
|
||||
// recordReads is what node's record `<node>.node` states its engine reads (ADR 0614 §4), and whether the assigner holds
|
||||
|
||||
@@ -7,7 +7,6 @@ package buckets
|
||||
import (
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
watchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
||||
delivery "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
|
||||
wants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
|
||||
@@ -66,8 +65,6 @@ const (
|
||||
TellerModule = "mesh-delivery-teller"
|
||||
// Answers holds the teller's answer to each ask.
|
||||
Answers = TellerModule + "_answers"
|
||||
// Builds is the builder's.
|
||||
Builds = builds.BuilderBuildsBucket
|
||||
// Grants is the grants part's.
|
||||
Grants = grantsstate.PartGrantsBucket
|
||||
)
|
||||
|
||||
@@ -10,7 +10,7 @@ func TestBucketsAreTheEarlierNames(t *testing.T) {
|
||||
"Modules": {Modules, "mesh-delivery-indexer_modules"}, "Manifests": {Manifests, "mesh-delivery-indexer_manifests"},
|
||||
"Presence": {Presence, "mesh-delivery-watchman_presence"}, "Source": {Source, "mesh-delivery-gate_source"},
|
||||
"Merges": {Merges, "mesh-delivery-gate_merges"}, "Answers": {Answers, "mesh-delivery-teller_answers"},
|
||||
"Builds": {Builds, "mesh-delivery-builder_builds"}, "Grants": {Grants, "mesh-grants_grants"},
|
||||
"Grants": {Grants, "mesh-grants_grants"},
|
||||
"Wants": {Wants, "mesh-delivery-assigner_wants"}, "Differences": {Differences, "mesh-delivery-assigner_differences"},
|
||||
"Asks": {Asks, "mesh-delivery-assigner_asks"}, "Puts": {Puts, "mesh-delivery-assigner_puts"},
|
||||
"Policies": {Policies, "mesh-delivery-assigner_policies"}, "Provisions": {Provisions, "mesh-provisioner_provisions"}, "TellerModule": {TellerModule, "mesh-delivery-teller"},
|
||||
|
||||
@@ -46,14 +46,15 @@ func TestManifest(t *testing.T) {
|
||||
// every node's published compositions to compare them, and the gate's source and merges, for the release that carries a build (ADR 0601 §1).
|
||||
// DLV-F2r (ADR 0615 §1, ADR 0631 §1): the indexer's manifests, modules and module graph (whose verb edges a want waits
|
||||
// on, ADR 0614 §2, §3) are read through its first views and views family, none of the indexer's states by reads.
|
||||
want := []string{"mesh-delivery-builder.builds", "mesh-delivery-gate.merges", "mesh-delivery-gate.source",
|
||||
// A-181: what was built is heard from the builder's stream of ends, so the builder's builds bucket is no read.
|
||||
want := []string{"mesh-delivery-gate.merges", "mesh-delivery-gate.source",
|
||||
"mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed",
|
||||
"mesh-provisioner.provisions", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
t.Errorf("reads = %v, want exactly %v", reads, want)
|
||||
}
|
||||
// the buckets the program names are the manifest's reads, `<module>.<state>` as `<module>_<state>`
|
||||
named := []string{buckets.Builds, buckets.Merges, buckets.Source, buckets.Answers, buckets.Presence, buckets.Grants,
|
||||
named := []string{buckets.Merges, buckets.Source, buckets.Answers, buckets.Presence, buckets.Grants,
|
||||
buckets.ComposedRead, "mesh-provisioner_provisions", buckets.Have}
|
||||
slices.Sort(named)
|
||||
for i, r := range want {
|
||||
@@ -130,7 +131,7 @@ func TestManifest(t *testing.T) {
|
||||
}
|
||||
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events;
|
||||
// issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears.
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildEndedEvent, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) {
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildsStreamEvents, buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) {
|
||||
t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes)
|
||||
}
|
||||
if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
|
||||
|
||||
@@ -30,7 +30,6 @@
|
||||
}
|
||||
],
|
||||
"reads": [
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-gate.merges",
|
||||
"mesh-delivery-gate.source",
|
||||
"mesh-delivery-teller.answers",
|
||||
@@ -53,7 +52,7 @@
|
||||
"mesh-catalog.version-released",
|
||||
"mesh-catalog.seat-definition-changed",
|
||||
"mesh-catalog.seat-definition-removed",
|
||||
"mesh-delivery-builder.build.ended",
|
||||
"mesh-delivery-builder.build.ended.**",
|
||||
"mesh-delivery-teller.ask-ended",
|
||||
"mesh-pipeline-runner.pipeline-ended",
|
||||
"mesh-pipeline-runner.group-ended",
|
||||
|
||||
@@ -53,10 +53,11 @@ type manifest struct {
|
||||
Consumes []string `json:"consumes"`
|
||||
State []state `json:"state"`
|
||||
Streams []struct {
|
||||
Name string `json:"name"`
|
||||
Events []string `json:"events"`
|
||||
MaxAgeDays int `json:"max-age-days"`
|
||||
MaxMiB int `json:"max-mib"`
|
||||
Name string `json:"name"`
|
||||
Events []string `json:"events"`
|
||||
MaxAgeDays int `json:"max-age-days"`
|
||||
MaxMiB int `json:"max-mib"`
|
||||
LastPerSubject bool `json:"last-per-subject"`
|
||||
} `json:"streams"`
|
||||
Logs []struct {
|
||||
Name string `json:"name"`
|
||||
@@ -177,9 +178,9 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
// Design 47 G §672: the toolchain waits' stream, kept 7 days within 16 MiB, and the bound's setting, 60 minutes.
|
||||
// Design 47 G A-30, §383: and its own findings' events, which its finding writer says.
|
||||
if len(m.Streams) != 3 || m.Streams[0].Name != "waits" || !reflect.DeepEqual(m.Streams[0].Events, []string{"wait-held.**", "wait-ended.**"}) || m.Streams[0].MaxAgeDays != 7 || m.Streams[0].MaxMiB != 16 ||
|
||||
m.Streams[1].Name != builds.BuildsStream || !reflect.DeepEqual(m.Streams[1].Events, []string{builds.BuildsStreamEvents}) || m.Streams[1].MaxAgeDays != 7 ||
|
||||
m.Streams[1].Name != builds.BuildsStream || !reflect.DeepEqual(m.Streams[1].Events, []string{builds.BuildsStreamEvents}) || m.Streams[1].MaxAgeDays != 7 || !m.Streams[1].LastPerSubject ||
|
||||
m.Streams[2].Name != "mesh-delivery-builder-findings" || !reflect.DeepEqual(m.Streams[2].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) {
|
||||
t.Errorf("streams = %+v, want waits with wait-held.** and wait-ended.**, 7 days, 16 MiB, the builds' stream (build.ended.**, 7 days) and the findings' stream", m.Streams)
|
||||
t.Errorf("streams = %+v, want waits with wait-held.** and wait-ended.**, 7 days, 16 MiB, the builds' stream (build.ended.**, 7 days, kept last per subject: design 47 G A-181) and the findings' stream", m.Streams)
|
||||
}
|
||||
if c, ok := m.Config["toolchain-wait-minutes"]; !ok || c.Default != float64(60) || len(m.Config) != 2 || m.Config["republish-max-versions"].Default != float64(50) {
|
||||
t.Errorf("config = %+v, want toolchain-wait-minutes of 60 and republish-max-versions of 50 (design 47 G §588, §633)", m.Config)
|
||||
|
||||
@@ -130,7 +130,7 @@
|
||||
],
|
||||
"streams": [
|
||||
{ "name": "waits", "events": ["wait-held.**", "wait-ended.**"], "max-age-days": 7, "max-mib": 16 },
|
||||
{ "name": "mesh-delivery-builder-builds", "events": ["build.ended.**"], "max-age-days": 7, "max-mib": 16 },
|
||||
{ "name": "mesh-delivery-builder-builds", "events": ["build.ended.**"], "max-age-days": 7, "max-mib": 16, "last-per-subject": true },
|
||||
{ "name": "mesh-delivery-builder-findings", "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], "max-age-days": 7, "max-mib": 16 }
|
||||
],
|
||||
"config": {
|
||||
|
||||
@@ -60,6 +60,9 @@ type ModuleStream struct {
|
||||
// History says the stream keeps its history (novox/hq ADR 0592 §6, `retention: history`): every message, with no
|
||||
// maximum age, a new one refused when it is full rather than its oldest dropped.
|
||||
History bool
|
||||
// LastPerSubject says the stream keeps only the last message of each subject (`last-per-subject`, design 47 G §703): a part that
|
||||
// follows it reads it whole at each start (§556).
|
||||
LastPerSubject bool
|
||||
}
|
||||
|
||||
// Subjects are the subjects the stream takes.
|
||||
@@ -104,6 +107,14 @@ func (s ModuleStream) Stream() Stream {
|
||||
out.Why += "; heard only by the consumers its trusted config value grants"
|
||||
}
|
||||
}
|
||||
if s.LastPerSubject {
|
||||
out.MaxMsgsPerSubject = 1
|
||||
out.Why = fmt.Sprintf("%s's own stream kept last per subject (design 47 G §703): the newest message of each subject, at most %d MiB, "+
|
||||
"read whole by the part that follows it at each start (§556)", s.Module, s.MaxMiB)
|
||||
if s.Restricted {
|
||||
out.Why += "; heard only by the consumers its trusted config value grants"
|
||||
}
|
||||
}
|
||||
if s.WorkQueue {
|
||||
out.Retention = RetentionWorkQueue
|
||||
out.DiscardNew = true
|
||||
@@ -262,8 +273,9 @@ func heardGrants(p Principal, hears []Heard) []string {
|
||||
"$JS.API.CONSUMER.INFO."+h.Stream+"."+durable,
|
||||
"$JS.API.CONSUMER.MSG.NEXT."+h.Stream+"."+durable,
|
||||
"$JS.ACK."+h.Stream+"."+durable+".>")
|
||||
if strings.HasPrefix(h.Stream, seatdef.KeptStreamPrefix) {
|
||||
// A seat's kept event is heard through process.HearSeatEvents, which reads the stream's own info to see whether it is
|
||||
if strings.HasPrefix(h.Stream, seatdef.KeptStreamPrefix) || h.LastPerSubject {
|
||||
// A seat's kept event, and a module's own stream kept last per subject, are heard through process.HearSeatEvents and
|
||||
// process.HearStream, which reads the stream's own info to see whether it is
|
||||
// kept last per subject (design 47 G §641, A-19): the info of this stream alone.
|
||||
out = append(out, "$JS.API.STREAM.INFO."+h.Stream)
|
||||
}
|
||||
|
||||
@@ -30,6 +30,10 @@ func TestAReaderOfASeatsKeptEventIsGrantedItsStreamInfo(t *testing.T) {
|
||||
if !slices.Contains(kept, "$JS.API.STREAM.INFO.SEAT_mesh-delivery_owed") {
|
||||
t.Errorf("a reader of a seat's kept event is not granted its stream's info: %v", kept)
|
||||
}
|
||||
own := heardGrants(p, []Heard{{Stream: "BUILDER_BUILDS", Filters: []string{"f"}, LastPerSubject: true}})
|
||||
if !slices.Contains(own, "$JS.API.STREAM.INFO.BUILDER_BUILDS") {
|
||||
t.Errorf("a reader of a module's stream kept last per subject is not granted its info: %v", own)
|
||||
}
|
||||
other := heardGrants(p, []Heard{{Stream: "mesh-x-events", Filters: []string{"f"}}})
|
||||
if slices.Contains(other, "$JS.API.STREAM.INFO.mesh-x-events") {
|
||||
t.Errorf("a reader of a module's stream is granted its info: %v", other)
|
||||
|
||||
@@ -25,3 +25,20 @@ func TestAStarEmitterBeforeAFindingEventReachesEachDeclaringStream(t *testing.T)
|
||||
t.Fatalf("with no event defined, heard %+v", hears)
|
||||
}
|
||||
}
|
||||
|
||||
// Design 47 G §703, §556, A-181: a stream a manifest declares kept last per subject is asked of the bus so (one message per subject),
|
||||
// and what hears it is composed to read it whole.
|
||||
func TestAStreamKeptLastPerSubjectIsComposedAndHeardSo(t *testing.T) {
|
||||
shelf := map[string]catalogue.Manifest{
|
||||
"builder": {Module: "builder", Streams: []catalogue.StreamDeclaration{{Name: "builder-builds", Events: []string{"build.ended.**"}, LastPerSubject: true}}},
|
||||
}
|
||||
none := func(string, string) []any { return nil }
|
||||
reader := catalogue.Manifest{Module: "reader", Consumes: []string{"builder.build.ended.**"}}
|
||||
hears, _, _ := hearsOf(reader, shelf, "ace", none, reader.Consumes, nil)
|
||||
if len(hears) != 1 || !hears[0].LastPerSubject {
|
||||
t.Fatalf("heard %+v, want one stream read whole", hears)
|
||||
}
|
||||
if got := streamsOf(shelf["builder"]); len(got) != 1 || got[0].Stream().MaxMsgsPerSubject != 1 {
|
||||
t.Fatalf("composed %+v, want one message per subject", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -434,7 +434,7 @@ func streamsOf(m catalogue.Manifest) []buscompose.ModuleStream {
|
||||
}
|
||||
out = append(out, buscompose.ModuleStream{Module: m.Module, Name: s.BusName(), Events: events,
|
||||
MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil,
|
||||
WorkQueue: s.Retention == catalogue.StreamRetentionWorkQueue, Duplicates: s.Duplicates, History: s.KeepsHistory()})
|
||||
WorkQueue: s.Retention == catalogue.StreamRetentionWorkQueue, Duplicates: s.Duplicates, History: s.KeepsHistory(), LastPerSubject: s.LastPerSubject})
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -510,7 +510,7 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node
|
||||
parts, touches, onEvents := catalogue.Hearing(reading, declared, node, values, defined)
|
||||
var out []buscompose.Heard
|
||||
for _, p := range parts {
|
||||
h := buscompose.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter, Restricted: p.Stream.Grant != nil}
|
||||
h := buscompose.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter, Restricted: p.Stream.Grant != nil, LastPerSubject: p.Stream.LastPerSubject}
|
||||
for _, pattern := range p.Patterns {
|
||||
h.Filters = append(h.Filters, buscompose.ModuleStreamSubject(p.Emitter, buscompose.WirePattern(pattern)))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user