From 9cd38f59ffe5fe045c931e0a6284749125385cac Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:53:02 +0100 Subject: [PATCH] A-181: the assigner reads the builder's build ends from their stream, kept last per subject, and reads no builder bucket MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../cmd/mesh-delivery-assigner/buildended.go | 62 +++++++++----- .../buildends_helper_test.go | 21 +++++ .../integration_test.go | 83 +++++++++++++++++-- .../cmd/mesh-delivery-assigner/main.go | 7 +- .../cmd/mesh-delivery-assigner/pass_test.go | 19 ++--- .../mesh-delivery-assigner/pipelines_test.go | 9 +- .../cmd/mesh-delivery-assigner/puts.go | 11 +-- .../cmd/mesh-delivery-assigner/tools.go | 35 ++------ .../cmd/mesh-delivery-assigner/verbs.go | 19 ++--- .../cmd/mesh-delivery-assigner/world.go | 31 ++----- .../internal/buckets/buckets.go | 3 - .../internal/buckets/buckets_test.go | 2 +- .../mesh-delivery-assigner/manifest_test.go | 7 +- modules/mesh-delivery-assigner/module.json | 3 +- .../mesh-delivery-builder/manifest_test.go | 13 +-- modules/mesh-delivery-builder/module.json | 2 +- .../internal/buscompose/modstreams.go | 16 +++- .../internal/buscompose/orderedread_test.go | 4 + .../internal/grants/defined_test.go | 17 ++++ .../mesh-grants/internal/grants/records.go | 4 +- 20 files changed, 224 insertions(+), 144 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildends_helper_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go index d918b4eb..f670eb63 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go @@ -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() + }() } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildends_helper_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildends_helper_test.go new file mode 100644 index 00000000..0f12d2a2 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildends_helper_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index ba7ce20e..1670fa6f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -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 +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index baa43787..2c583b55 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 48a1e5f4..9eb18aea 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -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{ diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go index 19a2d4a4..e3e92cd0 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go @@ -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) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index dab41719..27ad5b4a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -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 { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 7664c4fb..7a5ad112 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -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. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/verbs.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/verbs.go index ec642a9c..35a3c35f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/verbs.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/verbs.go @@ -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 } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 88590412..3acf053b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -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` states its engine reads (ADR 0614 §4), and whether the assigner holds diff --git a/modules/mesh-delivery-assigner/internal/buckets/buckets.go b/modules/mesh-delivery-assigner/internal/buckets/buckets.go index 0cf5be2d..30d65c63 100644 --- a/modules/mesh-delivery-assigner/internal/buckets/buckets.go +++ b/modules/mesh-delivery-assigner/internal/buckets/buckets.go @@ -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 ) diff --git a/modules/mesh-delivery-assigner/internal/buckets/buckets_test.go b/modules/mesh-delivery-assigner/internal/buckets/buckets_test.go index 0b6de994..9b092c1e 100644 --- a/modules/mesh-delivery-assigner/internal/buckets/buckets_test.go +++ b/modules/mesh-delivery-assigner/internal/buckets/buckets_test.go @@ -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"}, diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 94ad01ef..947184bc 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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, `.` as `_` - 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", diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index fe290015..49058b99 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -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", diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index f7498cb2..e09f37f8 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -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) diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index a60b5e69..e27cc4b4 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -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": { diff --git a/modules/mesh-grants/internal/buscompose/modstreams.go b/modules/mesh-grants/internal/buscompose/modstreams.go index 0c724157..4b32fff5 100644 --- a/modules/mesh-grants/internal/buscompose/modstreams.go +++ b/modules/mesh-grants/internal/buscompose/modstreams.go @@ -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) } diff --git a/modules/mesh-grants/internal/buscompose/orderedread_test.go b/modules/mesh-grants/internal/buscompose/orderedread_test.go index 8c7192f0..96fb5fc9 100644 --- a/modules/mesh-grants/internal/buscompose/orderedread_test.go +++ b/modules/mesh-grants/internal/buscompose/orderedread_test.go @@ -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) diff --git a/modules/mesh-grants/internal/grants/defined_test.go b/modules/mesh-grants/internal/grants/defined_test.go index aafa7126..85aa2f3a 100644 --- a/modules/mesh-grants/internal/grants/defined_test.go +++ b/modules/mesh-grants/internal/grants/defined_test.go @@ -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) + } +} diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index 133dc6b4..2e26b50b 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -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))) }