DLV-B4: the builder hears the four registry seats' published, refused and removed, and keeps its floor waits in the state floor-waits
Each outcome is a seat event <seat>.<event> on the shared events stream (the lead's ruling of 12:4x), heard on the builder's events consumer through its consumes; the subjects are mesh-contracts'. A restarted builder takes its floor waits in again, so its silent sources' events still wake them; builder_floor_waits reads them. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -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, <owner>/<name>"}}},
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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 `<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
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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"}
|
||||
|
||||
@@ -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 `<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)
|
||||
}
|
||||
|
||||
@@ -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 },
|
||||
|
||||
Reference in New Issue
Block a user