diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index c40f7be1..9f4e12e4 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -29,7 +29,7 @@ type queueAsker func(key string, value []byte) (json.RawMessage, error) // the builder's read, buildAgain writes a reasks value of mode build and waits for the builder's answer. // tags writes a tag-asks value and waits for the builder's answer; ta reads tag-asks. func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, queue, buildAgain queueAsker, since build.LogSince, now func() time.Time, vs versions.Reader, - tags queueAsker, ta versions.Reader) []stdio.Tool { + tags queueAsker, ta, fw versions.Reader) []stdio.Tool { str := func(args map[string]any, k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) } // handAct writes a reasks value of a versions hand act and answers the builder's answer, or its refusal as an error. handAct := func(key string, value []byte, err error) (any, error) { @@ -99,6 +99,9 @@ func Tools(r build.StuckReader, askAgain func(key string, value []byte) error, q Input: map[string]any{"type": "object", "properties": map[string]any{ "repository": map[string]any{"type": "string", "description": "only this repository's asks, /"}}}, Run: func(args map[string]any) (any, error) { return versions.AnswerTagAsks(ta, str(args, "repository")) }}, + {Name: versions.FloorWaitsTool, Description: versions.FloorWaitsDescription, + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { return versions.AnswerFloorWaits(fw) }}, {Name: versions.VersionsTool, Description: versions.VersionsDescription, Input: map[string]any{"type": "object", "required": []string{"name"}, "properties": map[string]any{ "name": map[string]any{"type": "string", "description": "the package, image, bundle or archive name"}, @@ -362,7 +365,15 @@ func main() { } return e.Value, true, nil }} - if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now, vr, tags, tr)); err != nil { + fs := stdio.State(versions.FloorWaitsState) + fr := versions.Reader{Keys: fs.Keys, Get: func(key string) (json.RawMessage, bool, error) { + e, err := fs.Get(key) + if err != nil || e == nil { + return nil, false, err + } + return e.Value, true, nil + }} + if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now, vr, tags, tr, fr)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err) os.Exit(1) } 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 439037ce..0ef44baa 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -113,7 +113,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The builder numbers each trunk build at its claim (novox/hq design 51 *The versions*): its state versions, bound once // the bus holds it; the floor's three sources asked as its own account; each new number said as version.numbered. versionsKV := process.AwaitBucket(ctx, js, conn, versions.Bucket, "no trunk build is numbered: each is built unnumbered and published by its digest", logger.Printf) - floorWaits := &versions.FloorWaits{} + floorWaits := &versions.FloorWaits{Logf: logger.Printf} + // The names waiting on their floor are kept in the state floor-waits (DLV-B4): a restarted builder takes them in, so + // their silent sources' events still wake them; until the state is bound they are kept here only. + process.AwaitBucket(ctx, js, conn, versions.FloorWaitsBucket, "the names waiting on their floor are kept in memory only, and builder_floor_waits shows none", logger.Printf). + Ready(func(kv jetstream.KeyValue) { + floorWaits.Keep(kvKeeper{ctx: ctx, kv: kv}, heldFloorWaits(ctx, kv, logger.Printf)) + }) numberer := &versions.Numberer{Versions: versions.Versions{Store: awaitedVersions{versionsKV}}, Floor: floorVia(calls.Call), Waits: floorWaits, Said: func(nb versions.NumberedBody) { if err := emitNumbered(context.WithoutCancel(ctx), js, cred, nb); err != nil { @@ -229,8 +235,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The same consumer carries what wakes a floor (design 51 *The floor*): the indexer's manifest-accepted and the // condition keeper's condition-cleared. 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} cc, err := process.HearEvents(cons, func(m jetstream.Msg) { - if floor.heard(ctx, m) { + if floor.heard(ctx, m) || seats.heard(ctx, m) { return } heardGone(ctx, b, m, logger.Printf) 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 16305538..01bdd7a0 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 @@ -11,6 +11,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" + "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" @@ -83,11 +84,16 @@ type wakeMsg struct { subject string data []byte acked bool + naked bool } func (m *wakeMsg) Subject() string { return m.subject } func (m *wakeMsg) Data() []byte { return m.data } func (m *wakeMsg) Ack() error { m.acked = true; return nil } +func (m *wakeMsg) Nak() error { m.naked = true; return nil } +func (m *wakeMsg) Headers() nats.Header { + return nil +} func aFloorsWakeIsHeard(t *testing.T) { var logs bytes.Buffer @@ -107,4 +113,34 @@ func aFloorsWakeIsHeard(t *testing.T) { if f.heard(context.Background(), &wakeMsg{subject: delivery.ArtifactsGoneSubject}) { 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 + // silent; one the state cannot take is left for the bus to give again. + logs.Reset() + waits.Wait("m/pkg", versions.SeatGoPackages, "", versions.Floor{Silent: []string{versions.SourceSeat}}, nil, time.Unix(0, 0)) + h := &seatHearer{n: &versions.Numberer{Versions: versions.Versions{Store: noVersions{}}, Waits: waits}, b: &build.Builder{}, logf: log.New(&logs, "", 0).Printf} + byDigest := &wakeMsg{subject: "mesh.seat.go-package-registry.event.published", data: []byte(`{"name":"m/other","digest":"sha256:aa","kind":"package","holder":"gitea"}`)} + if !h.heard(context.Background(), byDigest) || !byDigest.acked || !strings.Contains(logs.String(), "the floor of m/pkg is asked again: go-package-registry said published") { + t.Fatalf("published: acked %v, log %q", byDigest.acked, logs.String()) + } + refused := &wakeMsg{subject: "mesh.seat.oci-artifact-registry.event.refused", data: []byte(`{"name":"m/program","version":"0.1.3","digest":"sha256:bb","reason":"bytes differ"}`)} + if !h.heard(context.Background(), refused) || refused.acked || !refused.naked { + t.Fatalf("a refusal the state cannot take: acked %v, naked %v, log %q", refused.acked, refused.naked, logs.String()) + } + if h.heard(context.Background(), &wakeMsg{subject: manifestAcceptedSubject}) { + t.Fatal("manifest-accepted was taken as a seat's outcome") + } } + +// noVersions is a state versions that holds nothing. +type noVersions struct{} + +func (noVersions) Get(context.Context, string) ([]byte, uint64, bool, error) { + return nil, 0, false, nil +} +func (noVersions) Create(context.Context, string, []byte) (uint64, error) { + return 0, errors.New("read only") +} +func (noVersions) Update(context.Context, string, []byte, uint64) (uint64, error) { + return 0, errors.New("read only") +} +func (noVersions) KeysOf(context.Context, string) ([]string, error) { return nil, nil } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go new file mode 100644 index 00000000..58ca0421 --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/seats.go @@ -0,0 +1,130 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "strings" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/containerimageregistry" + "git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry" + "git.novox.be/novox/mesh-contracts/go/seats/npmpackageregistry" + "git.novox.be/novox/mesh-contracts/go/seats/ociartifactregistry" + + "mesh-delivery-builder/internal/build" + "mesh-delivery-builder/internal/versions" +) + +// seatEvent is one registry seat's event the builder hears: the seat and the event's name. +type seatEvent struct{ seat, event string } + +// seatEvents are the four registry seats' published, refused and removed, by subject (design 51 *How a publish ends*, +// DLV-B4): each a seat event `.` on the shared events stream, heard on the builder's durable consumer of +// it through the manifest's consumes. The subjects are the seats' contracts' (mesh-contracts). +var seatEvents = map[string]seatEvent{ + containerimageregistry.SubjectPublished: {containerimageregistry.Seat, containerimageregistry.EventPublished}, + containerimageregistry.SubjectRefused: {containerimageregistry.Seat, containerimageregistry.EventRefused}, + containerimageregistry.SubjectRemoved: {containerimageregistry.Seat, containerimageregistry.EventRemoved}, + ociartifactregistry.SubjectPublished: {ociartifactregistry.Seat, ociartifactregistry.EventPublished}, + ociartifactregistry.SubjectRefused: {ociartifactregistry.Seat, ociartifactregistry.EventRefused}, + ociartifactregistry.SubjectRemoved: {ociartifactregistry.Seat, ociartifactregistry.EventRemoved}, + gopackageregistry.SubjectPublished: {gopackageregistry.Seat, gopackageregistry.EventPublished}, + gopackageregistry.SubjectRefused: {gopackageregistry.Seat, gopackageregistry.EventRefused}, + gopackageregistry.SubjectRemoved: {gopackageregistry.Seat, gopackageregistry.EventRemoved}, + npmpackageregistry.SubjectPublished: {npmpackageregistry.Seat, npmpackageregistry.EventPublished}, + npmpackageregistry.SubjectRefused: {npmpackageregistry.Seat, npmpackageregistry.EventRefused}, + npmpackageregistry.SubjectRemoved: {npmpackageregistry.Seat, npmpackageregistry.EventRemoved}, +} + +// seatHearer takes the registry seats' outcomes in: each moves a numbered version (versions.Numberer.Heard), and a +// `published` that wakes a name waiting on its floor looks at every module again. +type seatHearer struct { + n *versions.Numberer + b *build.Builder + logf func(string, ...any) +} + +// heard takes a message of the events consumer in when it is a registry seat's outcome, and answers whether it was. +// One the state could not take is left unacknowledged (Nak), so the bus gives it again; one that is no outcome is said +// and dropped. +func (h *seatHearer) heard(ctx context.Context, m jetstream.Msg) bool { + se, ok := seatEvents[m.Subject()] + if !ok { + return false + } + var o versions.Outcome + if err := json.Unmarshal(m.Data(), &o); err != nil || o.Name == "" { + h.logf("%s holds what is no outcome of %s, left: %v", m.Subject(), se.seat, err) + h.ack(m) + return true + } + id := "" + if m.Headers() != nil { + id = m.Headers().Get("x-event-id") + } + wake, err := h.n.Heard(ctx, se.seat, se.event, id, o) + if err != nil { + h.logf("%v; heard again", err) + if nakErr := m.Nak(); nakErr != nil { + h.logf("leaving %s for the bus to give again: %v", m.Subject(), nakErr) + } + return true + } + h.ack(m) + if len(wake) > 0 { + h.logf("the floor of %s is asked again: %s said %s", strings.Join(wake, ", "), se.seat, se.event) + h.b.Reconsider(ctx) + } + return true +} + +func (h *seatHearer) ack(m jetstream.Msg) { + if err := m.Ack(); err != nil { + h.logf("acknowledging %s: %v", m.Subject(), err) + } +} + +// kvKeeper is the state floor-waits as versions.FloorKeeper. +type kvKeeper struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (k kvKeeper) Put(key string, value []byte) error { + _, err := k.kv.Put(k.ctx, key, value) + return err +} + +func (k kvKeeper) Delete(key string) error { + err := k.kv.Delete(k.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil + } + return err +} + +// heldFloorWaits are the waits floor-waits holds at the builder's start; a value that is no wait is said and left. +func heldFloorWaits(ctx context.Context, kv jetstream.KeyValue, logf func(string, ...any)) []versions.FloorWait { + l, err := kv.ListKeys(ctx) + if err != nil { + logf("%s cannot be listed, so a restart's waits are asked at their next claim: %v", versions.FloorWaitsState, err) + return nil + } + defer func() { _ = l.Stop() }() + var out []versions.FloorWait + for k := range l.Keys() { + e, err := kv.Get(ctx, k) + if err != nil { + continue + } + w, err := versions.DecodeFloorWait(e.Value()) + if err != nil { + logf("%s.%s holds what is no floor wait, left: %v", versions.FloorWaitsState, k, err) + continue + } + out = append(out, w) + } + return out +} diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index 3131c6f3..ab72384a 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -11,6 +11,7 @@ require ( require git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep v0.0.0-20261102193506-d221c633799a require ( + git.novox.be/novox/mesh-contracts/go v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun v0.0.0-00010101000000-000000000000 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect diff --git a/modules/mesh-delivery-builder/internal/versions/floor.go b/modules/mesh-delivery-builder/internal/versions/floor.go index f88403fc..62aab55a 100644 --- a/modules/mesh-delivery-builder/internal/versions/floor.go +++ b/modules/mesh-delivery-builder/internal/versions/floor.go @@ -2,6 +2,8 @@ package versions import ( "context" + "encoding/json" + "errors" "fmt" "slices" "sort" @@ -107,17 +109,85 @@ func (w FloorWait) Finding() string { w.Name, w.Since.UTC().Format(time.RFC3339), strings.Join(parts, ", ")) } +// The builder's state floor-waits (DLV-B4): each name waiting on its floor, keyed by its escaped name, so a restarted +// builder still knows which names its silent sources' events wake; builder_floor_waits reads it. +const ( + FloorWaitsState = "floor-waits" + FloorWaitsBucket = "mesh-delivery-builder_floor-waits" + FloorWaitsTool = "builder_floor_waits" +) + +// FloorWaitKey is a wait's key in floor-waits. +func FloorWaitKey(name string) string { return EscapeToken(name) } + +// FloorKeeper keeps the waits on the bus: Put a wait's value, Delete its key. +type FloorKeeper interface { + Put(key string, value []byte) error + Delete(key string) error +} + // FloorWaits are the names waiting on their floor. Wake answers which names to ask again for an event, Answered drops -// a name once all three answered; the builder's finding for each is the wait's Finding. +// a name once all three answered; the builder's finding for each is the wait's Finding. Once Kept, every change is +// written to the state floor-waits as well. type FloorWaits struct { - mu sync.Mutex - waits map[string]FloorWait + mu sync.Mutex + waits map[string]FloorWait + keeper FloorKeeper + // Logf says a wait that could not be kept; nil says nothing. + Logf func(string, ...any) +} + +// Keep binds the state: the waits it holds that are not known here are taken in (a restarted builder's), every wait +// known here is written to it, and every change from now on is. +func (fw *FloorWaits) Keep(k FloorKeeper, held []FloorWait) { + fw.mu.Lock() + if fw.waits == nil { + fw.waits = map[string]FloorWait{} + } + for _, w := range held { + if _, had := fw.waits[w.Name]; !had && w.Name != "" { + fw.waits[w.Name] = w + } + } + fw.keeper = k + all := make([]FloorWait, 0, len(fw.waits)) + for _, w := range fw.waits { + all = append(all, w) + } + fw.mu.Unlock() + for _, w := range all { + fw.put(k, w) + } +} + +func (fw *FloorWaits) put(k FloorKeeper, w FloorWait) { + if k == nil { + return + } + data, err := json.Marshal(w) + if err == nil { + err = k.Put(FloorWaitKey(w.Name), data) + } + if err != nil && fw.Logf != nil { + fw.Logf("the floor wait of %s is not kept in %s (a restart forgets it until its next claim): %v", w.Name, FloorWaitsState, err) + } +} + +// DecodeFloorWait reads one value of floor-waits. +func DecodeFloorWait(raw []byte) (FloorWait, error) { + var w FloorWait + if err := json.Unmarshal(raw, &w); err != nil { + return FloorWait{}, err + } + if w.Name == "" { + return FloorWait{}, errors.New("a floor wait names no name") + } + return w, nil } // Wait records that name's floor is not known. func (fw *FloorWaits) Wait(name, seat, holder string, f Floor, why map[string]string, now time.Time) FloorWait { fw.mu.Lock() - defer fw.mu.Unlock() if fw.waits == nil { fw.waits = map[string]FloorWait{} } @@ -127,18 +197,73 @@ func (fw *FloorWaits) Wait(name, seat, holder string, f Floor, why map[string]st } w.Name, w.Seat, w.Holder, w.Silent, w.Why = name, seat, holder, slices.Clone(f.Silent), why fw.waits[name] = w + k := fw.keeper + fw.mu.Unlock() + fw.put(k, w) return w } // Answered drops name: its floor is known. It says whether the name waited, so its finding is cleared. func (fw *FloorWaits) Answered(name string) bool { fw.mu.Lock() - defer fw.mu.Unlock() _, had := fw.waits[name] delete(fw.waits, name) + k := fw.keeper + fw.mu.Unlock() + if had && k != nil { + if err := k.Delete(FloorWaitKey(name)); err != nil && fw.Logf != nil { + fw.Logf("the floor wait of %s is not cleared from %s: %v", name, FloorWaitsState, err) + } + } return had } +// FloorWaitsAnswer is what builder_floor_waits answers: every name waiting on its floor, oldest first, with its finding. +type FloorWaitsAnswer struct { + Waits []FloorWaitRead `json:"waits"` +} + +// FloorWaitRead is one wait as builder_floor_waits shows it. +type FloorWaitRead struct { + FloorWait + Finding string `json:"finding"` +} + +// AnswerFloorWaits is builder_floor_waits, read from the state floor-waits. +func AnswerFloorWaits(r Reader) (FloorWaitsAnswer, error) { + keys, err := r.Keys() + if err != nil { + return FloorWaitsAnswer{}, err + } + out := FloorWaitsAnswer{Waits: []FloorWaitRead{}} + for _, k := range keys { + raw, ok, err := r.Get(k) + if err != nil { + return FloorWaitsAnswer{}, err + } + if !ok { + continue + } + w, err := DecodeFloorWait(raw) + if err != nil { + return FloorWaitsAnswer{}, fmt.Errorf("%s.%s: %w", FloorWaitsState, k, err) + } + out.Waits = append(out.Waits, FloorWaitRead{FloorWait: w, Finding: w.Finding()}) + } + sort.Slice(out.Waits, func(i, j int) bool { + if !out.Waits[i].Since.Equal(out.Waits[j].Since) { + return out.Waits[i].Since.Before(out.Waits[j].Since) + } + return out.Waits[i].Name < out.Waits[j].Name + }) + return out, nil +} + +// FloorWaitsDescription is builder_floor_waits' description. +const FloorWaitsDescription = "The names waiting on their floor (the builder's state floor-waits, design 51 *The floor*), oldest first: " + + "each name, its registry seat, the sources still silent and why, since when, and the finding the builder says for it. " + + "Nothing of a waiting name is numbered until all three sources answer; builder_floor_again asks one again by hand. Read-only." + // Wake answers the names to ask their floor again on an event: kind one of the Wake constants; seat the seat a // `published` came on; module the module whose condition cleared. A name is woken only when the event says something // of a source it waits on; WakeByHand with a name wakes that name alone. diff --git a/modules/mesh-delivery-builder/internal/versions/versions_test.go b/modules/mesh-delivery-builder/internal/versions/versions_test.go index a2eef4bd..34870e74 100644 --- a/modules/mesh-delivery-builder/internal/versions/versions_test.go +++ b/modules/mesh-delivery-builder/internal/versions/versions_test.go @@ -72,6 +72,20 @@ func (m *mem) KeysOf(_ context.Context, filter string) ([]string, error) { return out, nil } +// keeperMem is the state floor-waits in memory. +type keeperMem map[string][]byte + +func (k keeperMem) Put(key string, v []byte) error { k[key] = v; return nil } +func (k keeperMem) Delete(key string) error { delete(k, key); return nil } +func (k keeperMem) keys() ([]string, error) { + var out []string + for key := range k { + out = append(out, key) + } + return out, nil +} +func (k keeperMem) get(key string) (json.RawMessage, bool, error) { v, ok := k[key]; return v, ok, nil } + func known(seat, tags, reqs []string) Floor { return Floor{Seat: seat, Tags: tags, Requirements: reqs} } // TestTheBuildersVersions holds design 51 *The versions*, *The floor* and *Tagged versions* (DLV-B1, DLV-B2, DLV-B3): @@ -274,6 +288,22 @@ func TestTheBuildersVersions(t *testing.T) { if !fw.Answered("m/program") || fw.Answered("m/program") || len(fw.Waiting()) != 1 { t.Error("Answered drops the wait once") } + // DLV-B4: the waits are kept in floor-waits; a restarted builder takes them in, and its sources' events still wake them. + kept := keeperMem{} + restarted := &FloorWaits{} + restarted.Keep(kept, []FloorWait{{Name: "m/held", Seat: SeatImages, Silent: []string{SourceTags}, Since: time.Unix(5, 0)}}) + if got := restarted.Wake(WakeForge, "", "", ""); !reflect.DeepEqual(got, []string{"m/held"}) { + t.Errorf("a wait held in floor-waits woke %v after a restart, want m/held", got) + } + restarted.Wait("m/new", SeatNPM, "", Floor{Silent: []string{SourceSeat}}, nil, time.Unix(9, 0)) + ans, err := AnswerFloorWaits(Reader{Keys: kept.keys, Get: kept.get}) + if err != nil || len(ans.Waits) != 2 || ans.Waits[0].Name != "m/held" || ans.Waits[1].Name != "m/new" || !strings.Contains(ans.Waits[1].Finding, "seat did not answer") { + t.Fatalf("builder_floor_waits = %+v, %v", ans, err) + } + restarted.Answered("m/held") + if _, ok := kept[FloorWaitKey("m/held")]; ok { + t.Error("an answered floor stays in floor-waits") + } }) t.Run("old form", func(t *testing.T) { forms := []string{"ref:artifact-store", "ref:registry"} diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index cc78f53c..f0ec38cc 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -93,7 +93,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if m.Module != "mesh-delivery-builder" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory + if want := []state{{Name: "builds", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory t.Errorf("state = %+v, want %+v", m.State, want) } // 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and @@ -118,8 +118,13 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 // starts; and by the seat mesh-artifact-sweep's artifacts-gone, a blob the sweep found gone (1937/AC4, 1955/L7), the // controller's no longer (1126/B9, ADR 0590). // Design 51 *The floor*: the indexer's manifest-accepted and the condition keeper's condition-cleared wake a silent floor. - if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", - "mesh-conditions.condition-cleared", "mesh-delivery-indexer.manifest-accepted"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + // DLV-B4: the four registry seats' outcomes, each a seat event `.` on the shared events stream. + if want := []string{"container-image-registry.published", "container-image-registry.refused", "container-image-registry.removed", + "go-package-registry.published", "go-package-registry.refused", "go-package-registry.removed", + "mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", + "mesh-conditions.condition-cleared", "mesh-delivery-indexer.manifest-accepted", + "npm-package-registry.published", "npm-package-registry.refused", "npm-package-registry.removed", + "oci-artifact-registry.published", "oci-artifact-registry.refused", "oci-artifact-registry.removed"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -129,7 +134,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 t.Errorf("secrets-owner = %q", m.SecretsOwner) } // 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again. - if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_floor_again", + if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_floor_again", "builder_floor_waits", "builder_publish_again", "builder_seat_queue", "builder_stuck", "builder_tag", "builder_tag_asks", "builder_untag", "builder_versions"}; !reflect.DeepEqual(sorted(m.Tools), want) { t.Errorf("tools = %v, want %v", m.Tools, want) } diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 87c4b61a..2f48305a 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -8,6 +8,7 @@ "builder_build_again", "builder_builds_since", "builder_floor_again", + "builder_floor_waits", "builder_publish_again", "builder_seat_queue", "builder_stuck", @@ -22,11 +23,23 @@ "version.old-form-closed" ], "consumes": [ + "container-image-registry.published", + "container-image-registry.refused", + "container-image-registry.removed", + "go-package-registry.published", + "go-package-registry.refused", + "go-package-registry.removed", "mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-conditions.condition-cleared", - "mesh-delivery-indexer.manifest-accepted" + "mesh-delivery-indexer.manifest-accepted", + "npm-package-registry.published", + "npm-package-registry.refused", + "npm-package-registry.removed", + "oci-artifact-registry.published", + "oci-artifact-registry.refused", + "oci-artifact-registry.removed" ], "invokes": [ "mesh-delivery-forge-bridge.bridge_tags", @@ -43,6 +56,7 @@ ], "state": [ { "name": "builds", "history": 5 }, + { "name": "floor-waits", "history": 1 }, { "name": "hearing", "history": 1 }, { "name": "instances", "history": 1 }, { "name": "queues", "history": 1 },