novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

mesh-delivery-builder: reads held's answer and the artifacts-gone body in the seat mesh-artifact-sweep's generated types, not go/delivery's (B2f; ADR 0571 §2, ADR 0634 §2; issue 1966)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
jschoubben
2026-11-05 15:59:45 +01:00
parent 7fcd2d0b64
commit 1286a24926
8 changed files with 78 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go
+8 -6
View File
@@ -12,6 +12,7 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -35,7 +36,7 @@ func (h *heldVerb) serve(t *testing.T, b *bed) {
References []string `json:"references"`
}
_ = json.Unmarshal(m.Data, &args)
a := delivery.HeldAnswer{Store: "store.test:5100", Held: []string{}, Gone: []string{}}
a := sweep.HeldAnswer{Store: "store.test:5100", Held: []string{}, Gone: []string{}}
h.mu.Lock()
for _, r := range args.References {
h.asked = append(h.asked, r)
@@ -65,11 +66,11 @@ func (h *heldVerb) asks() int {
// from the manifest's consumes.
func eventsConsumer(t *testing.T, b *bed, node string) {
t.Helper()
if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{sweep.SubjectArtifactsGone}}); err != nil {
t.Fatal(err)
}
if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder",
AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{sweep.SubjectArtifactsGone}}); err != nil {
t.Fatal(err)
}
}
@@ -168,12 +169,13 @@ func TestArtifactsGoneLeadsTheBuilderToBuildTheKeyAgain(t *testing.T) {
collected = true
mu.Unlock()
key, _ := delivery.BuilderBuildKey("ca-trust", fp)
body, err := delivery.EncodeArtifactsGone(delivery.ArtifactsGone{Module: "ca-trust", Key: key, Fingerprint: fp,
Gone: []string{ref}, At: time.Now().UTC()})
at := time.Now().UTC()
body, err := json.Marshal(sweep.ArtifactsGone{Event: sweep.EventArtifactsGone, Module: "ca-trust", Key: key, Fingerprint: fp,
Gone: []string{ref}, At: &at})
if err != nil {
t.Fatal(err)
}
if _, err := b.js.Publish(b.ctx, delivery.ArtifactsGoneSubject, body); err != nil {
if _, err := b.js.Publish(b.ctx, sweep.SubjectArtifactsGone, body); err != nil {
t.Fatal(err)
}
v := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool {
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go
+2 -1
View File
@@ -28,6 +28,7 @@ import (
runner "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry"
"git.novox.be/novox/mesh-contracts/go/seats/ociartifactregistry"
sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep"
"git.novox.be/novox/mesh-sdk/go/buildstep"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -1191,7 +1192,7 @@ func (b *bed) registry(t *testing.T, node string, seatVersions ...string) *regis
mk(versions.FloorWaitsBucket)
// 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{
eventsStream: {delivery.ArtifactsGoneSubject, ociartifactregistry.SubjectPublished, ociartifactregistry.SubjectRefused, ociartifactregistry.SubjectRemoved,
eventsStream: {sweep.SubjectArtifactsGone, ociartifactregistry.SubjectPublished, ociartifactregistry.SubjectRefused, ociartifactregistry.SubjectRemoved,
gopackageregistry.SubjectPublished, gopackageregistry.SubjectRefused, gopackageregistry.SubjectRemoved},
forgeStream: {process.ModuleStreamSubject(delivery.ForgeBridgeModule, ">")},
} {
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+2 -1
View File
@@ -25,6 +25,7 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/process"
"git.novox.be/novox/mesh-sdk/go/views"
@@ -329,7 +330,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
floor := floorWake{b: b, waits: floorWaits, logf: logger.Printf}
// And the four registry seats' published, refused and removed (DLV-B4), which move a numbered version.
seats := seatHearer{n: numberer, b: b, logf: logger.Printf, pub: pubs}
hear(eventsStream, delivery.ArtifactsGoneEvent+", what wakes a floor, the registry seats' words and manifest-reads-changed are not heard until the builder starts again", func(m jetstream.Msg) {
hear(eventsStream, sweep.EventArtifactsGone+", what wakes a floor, the registry seats' words and manifest-reads-changed are not heard until the builder starts again", func(m jetstream.Msg) {
if reads.heard(m, logger.Printf) || floor.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
return
}
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go
+2 -1
View File
@@ -21,6 +21,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/sdk/enginereads"
"git.novox.be/novox/mesh-delivery/sdk/forms"
@@ -223,7 +224,7 @@ func aFloorsWakeIsHeard(t *testing.T) {
if !f.heard(context.Background(), cleared) || !strings.Contains(logs.String(), "the floor of m/img is asked again: condition-cleared") {
t.Fatalf("condition-cleared: log %q", logs.String())
}
if f.heard(context.Background(), &wakeMsg{subject: delivery.ArtifactsGoneSubject}) {
if f.heard(context.Background(), &wakeMsg{subject: sweep.SubjectArtifactsGone}) {
t.Fatal("artifacts-gone was taken as a floor's wake")
}
// DLV-B4: a registry seat's outcome on the same consumer moves a version; a published wakes a name whose seat was
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go
+53 -4
View File
@@ -9,8 +9,8 @@ import (
"github.com/nats-io/nats.go/jetstream"
sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-builder/internal/build"
)
@@ -35,7 +35,7 @@ func storedVia(call toolCaller) func(ctx context.Context, references []string) (
if err != nil {
return nil, fmt.Errorf("asking held: %w", err)
}
a, err := delivery.DecodeHeldAnswer(raw, references)
a, err := decodeHeld(raw, references)
if err != nil {
return nil, err
}
@@ -54,13 +54,62 @@ func storedVia(call toolCaller) func(ctx context.Context, references []string) (
// heardGone takes one artifacts-gone event in: acknowledged, then handed to the builder, which asks the store and builds
// the key again. A body that is not one is said and acknowledged, so it is not handed over again.
func heardGone(ctx context.Context, b *build.Builder, m jetstream.Msg, logf func(string, ...any)) {
e, err := delivery.DecodeArtifactsGone(m.Data())
e, err := decodeGone(m.Data())
if ackErr := m.Ack(); ackErr != nil {
logf("acknowledging %s: %v", m.Subject(), ackErr)
}
if err != nil {
logf("%s holds what is not %s, left: %v", m.Subject(), delivery.ArtifactsGoneEvent, err)
logf("%s holds what is not %s, left: %v", m.Subject(), sweep.EventArtifactsGone, err)
return
}
b.ArtifactsGone(ctx, e)
}
// decodeHeld reads the seat's answer to held: the verb answers {output, ok, answer}, the answer being the seat's
// HeldAnswer. A reference asked that the answer names in none of held, gone and unknown is an error: an answer that
// leaves one out would let a key with a gone blob be reused.
func decodeHeld(raw []byte, asked []string) (sweep.HeldAnswer, error) {
var wrapped struct {
Output string `json:"output"`
OK bool `json:"ok"`
Answer json.RawMessage `json:"answer"`
}
if err := json.Unmarshal(raw, &wrapped); err != nil {
return sweep.HeldAnswer{}, fmt.Errorf("the answer of %s is not {output, ok, answer}: %w", SeatHeld, err)
}
if !wrapped.OK || len(wrapped.Answer) == 0 {
return sweep.HeldAnswer{}, fmt.Errorf("%s did not answer: %s", SeatHeld, wrapped.Output)
}
var a sweep.HeldAnswer
if err := json.Unmarshal(wrapped.Answer, &a); err != nil {
return sweep.HeldAnswer{}, fmt.Errorf("the answer of %s is not {held, gone}: %w", SeatHeld, err)
}
named := map[string]bool{}
for _, r := range a.Held {
named[r] = true
}
for _, r := range a.Gone {
named[r] = true
}
for r := range a.Unknown {
named[r] = true
}
for _, r := range asked {
if !named[r] {
return sweep.HeldAnswer{}, fmt.Errorf("the answer of %s names nothing of %s", SeatHeld, r)
}
}
return a, nil
}
// decodeGone reads the event's body, refusing one that names no module, fingerprint or reference.
func decodeGone(raw []byte) (sweep.ArtifactsGone, error) {
var e sweep.ArtifactsGone
if err := json.Unmarshal(raw, &e); err != nil {
return sweep.ArtifactsGone{}, fmt.Errorf("%s is not one: %w", sweep.EventArtifactsGone, err)
}
if e.Event != sweep.EventArtifactsGone || e.Module == "" || e.Fingerprint == "" || len(e.Gone) == 0 {
return sweep.ArtifactsGone{}, fmt.Errorf("%s names no module, fingerprint or gone reference: %s", sweep.EventArtifactsGone, raw)
}
return e, nil
}
modules/mesh-delivery-builder/go.mod
+4
View File
@@ -44,3 +44,7 @@ require (
)
replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk
require git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep v0.0.0-00010101000000-000000000000
replace git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep => ../../contracts/go/seats/meshartifactsweep
modules/mesh-delivery-builder/internal/build/gone_test.go
+3 -2
View File
@@ -9,6 +9,7 @@ import (
"sync"
"testing"
sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -184,14 +185,14 @@ func TestArtifactsGoneMakesTheBuilderAskAndBuildAgain(t *testing.T) {
held.mu.Lock()
held.gone[goneRef] = true // collected after the builder's look; the controller's take-in found it
held.mu.Unlock()
r.b.ArtifactsGone(context.Background(), delivery.ArtifactsGone{Event: delivery.ArtifactsGoneEvent, Module: "m",
r.b.ArtifactsGone(context.Background(), sweep.ArtifactsGone{Event: sweep.EventArtifactsGone, Module: "m",
Fingerprint: "src2:" + strings.Repeat("9", 64), Gone: []string{goneRef}})
r.b.Wait()
if r.st.revs[r.key] != rev || !log.has("a fingerprint the module does not have now") {
t.Fatalf("an event for another fingerprint moved the key (revision %d, was %d): %v", r.st.revs[r.key], rev, log.lines)
}
r.b.ArtifactsGone(context.Background(), delivery.ArtifactsGone{Event: delivery.ArtifactsGoneEvent, Module: "m",
r.b.ArtifactsGone(context.Background(), sweep.ArtifactsGone{Event: sweep.EventArtifactsGone, Module: "m",
Key: r.key, Fingerprint: fp1, Gone: []string{goneRef}})
r.b.Wait()
v := valueOf(t, r.st, r.key)
modules/mesh-delivery-builder/internal/build/run.go
+4 -3
View File
@@ -13,6 +13,7 @@ import (
"sync"
"time"
sweep "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshartifactsweep"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/sdk/forms"
"git.novox.be/novox/mesh-sdk/go/buildstep"
@@ -281,7 +282,7 @@ func (b *Builder) storedOnce(key string) func(ctx context.Context, references []
// that the store does not hold what the module's key names. The module is looked at again, which asks the store (ClaimOptions.Stored) and
// builds the key again; an event for a fingerprint the module no longer has is said and left. The event only wakes the
// builder: what is built rests on the store's answer, never on the event alone.
func (b *Builder) ArtifactsGone(ctx context.Context, e delivery.ArtifactsGone) {
func (b *Builder) ArtifactsGone(ctx context.Context, e sweep.ArtifactsGone) {
b.init()
b.mu.Lock()
m, ok := b.known[e.Module]
@@ -295,11 +296,11 @@ func (b *Builder) ArtifactsGone(ctx context.Context, e delivery.ArtifactsGone) {
b.mu.Unlock()
if !ok || m.FP2 != e.Fingerprint {
b.logf("%s: heard %s for %s, a fingerprint the module does not have now: nothing to build", e.Module,
delivery.ArtifactsGoneEvent, e.Fingerprint)
sweep.EventArtifactsGone, e.Fingerprint)
return
}
b.logf("%s: heard %s (%s): asking the artifact store and building the key again", e.Module,
delivery.ArtifactsGoneEvent, strings.Join(e.Gone, ", "))
sweep.EventArtifactsGone, strings.Join(e.Gone, ", "))
b.consider(ctx, e.Module, m)
}