diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go index 58395d16..30195bbc 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/gone_integration_test.go @@ -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 { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index eee4c068..32901723 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -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, ">")}, } { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 0289ec8f..54414870 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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 } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go index 0e53e2f2..cc000338 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/refused_test.go @@ -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 diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go index c5fc876a..7c1a023e 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/stored.go @@ -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 +} diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index 47f4e6af..ea243f5a 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -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 diff --git a/modules/mesh-delivery-builder/internal/build/gone_test.go b/modules/mesh-delivery-builder/internal/build/gone_test.go index 499cdbc9..c114010a 100644 --- a/modules/mesh-delivery-builder/internal/build/gone_test.go +++ b/modules/mesh-delivery-builder/internal/build/gone_test.go @@ -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) diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index fecf91c8..e651e436 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -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) }