Registry milestone: mesh-delivery (issue 1966) #151

Merged
mesh-tools merged 359 commits from feat/registry-milestone into main 2026-11-07 14:19:29 +00:00
10 changed files with 91 additions and 65 deletions
Showing only changes of commit 00903e9149 - Show all commits
@@ -159,8 +159,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
}}
// A publish that ends badly is a finding (design 51 *Health and conditions*): a holder's refused, a version no holder
// answered within its bound (a key of publish-bounds with its own time to live), a seat's outcome that does not read.
pubBoundsKV := process.AwaitBucket(ctx, js, conn, versions.PublishBoundsBucket, "a publish no holder answers is not raised", logger.Printf)
// answered within its bound (a key of the state bounds with its own time to live), a seat's outcome that does not read.
pubBoundsKV := process.AwaitBucket(ctx, js, conn, versions.BoundsBucket, "a publish no holder answers, and a node holding the old form seven days, are not raised", logger.Printf)
pubs := &publishes{bounds: pubBoundsKV.KV, versions: numberer.Versions, findings: found, bound: versions.PublishBound, logf: logger.Printf}
numberer.Publishing = pubs.arm
b := &build.Builder{
@@ -373,7 +373,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
pubBoundsKV.Ready(func(kv jetstream.KeyValue) {
each := func(e jetstream.KeyValueEntry) { pubs.ended(ctx, e) }
if pw, err := settle(ctx, kv, each); err != nil {
process.NotWatched(logger.Printf, versions.PublishBoundsBucket, err, "a publish no holder answers is not raised until the builder starts again")
process.NotWatched(logger.Printf, versions.BoundsBucket, err, "a publish no holder answers, and a node holding the old form seven days, are not raised until the builder starts again")
} else {
go func() { errs <- follow(ctx, pw, each) }()
}
@@ -80,11 +80,13 @@ func (o *oldForm) dropped() map[string]bool {
}
// step moves the window to what counted states.
func (o *oldForm) step(ctx context.Context, counted map[string][]string) {
// named is every node the read named (the placement's nodes before any is dropped), counted the statements of those
// still counted: a read naming no node judges nothing (delivery.StepWindow), so it never closes the window.
func (o *oldForm) step(ctx context.Context, named []string, counted map[string][]string) {
o.mu.Lock()
defer o.mu.Unlock()
before := o.w
next, closed, overdue := o.w.Step(versions.Holding(o.forms, counted, o.w.DroppedSet()), o.now())
next, closed, overdue := o.w.Step(named, versions.Holding(o.forms, counted, o.w.DroppedSet()), o.now())
o.w = next
if !slices.Equal(before.Holding, next.Holding) || before.Open != next.Open || closed {
o.keep(ctx)
@@ -14,7 +14,7 @@ import (
)
// publishes raises and clears the builder's findings of a publish that ended badly (versions/publish.go): a holder's
// refused, a version unanswered past its bound, kept as a key of publish-bounds with its own time to live, and a seat's
// refused, a version unanswered past its bound, kept as a key of the state bounds with its own time to live, and a seat's
// outcome that does not read.
type publishes struct {
bounds func() jetstream.KeyValue
@@ -28,10 +28,10 @@ type publishes struct {
func (p *publishes) arm(ctx context.Context, e versions.Entry) {
kv := p.bounds()
if kv == nil {
p.logf("the publish bound of %s@%s is not written (%s is not on the bus yet), so it is not raised if no holder answers", e.Name, e.Version, versions.PublishBoundsBucket)
p.logf("the publish bound of %s@%s is not written (%s is not on the bus yet), so it is not raised if no holder answers", e.Name, e.Version, versions.BoundsBucket)
return
}
if _, err := kv.Create(ctx, versions.Key(e.Name, e.Version), []byte(`{}`), jetstream.KeyTTL(p.bound)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
if _, err := kv.Create(ctx, versions.PublishBoundKey(e.Name, e.Version), []byte(`{}`), jetstream.KeyTTL(p.bound)); err != nil && !errors.Is(err, jetstream.ErrKeyExists) {
p.logf("the publish bound of %s@%s is not written, so it is not raised if no holder answers: %v", e.Name, e.Version, err)
}
}
@@ -41,7 +41,7 @@ func (p *publishes) arm(ctx context.Context, e versions.Entry) {
func (p *publishes) heard(ctx context.Context, seat, event string, o versions.Outcome) {
if o.Version != "" {
if kv := p.bounds(); kv != nil {
if err := kv.Delete(ctx, versions.Key(o.Name, o.Version)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
if err := kv.Delete(ctx, versions.PublishBoundKey(o.Name, o.Version)); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
p.logf("the publish bound of %s@%s is not ended: %v", o.Name, o.Version, err)
}
}
@@ -78,13 +78,14 @@ func (p *publishes) read(ctx context.Context, seat string) {
p.findings.clear(ctx, versions.SeatOutcomeUnread, health.ScopeModule, seat)
}
// ended is told each change of publish-bounds: a bound whose time ended (the bus's marker) on a version still numbered
// ended is told each change of the state bounds: a bound whose time ended (the bus's marker) on a version still numbered
// raises publish-unanswered.
func (p *publishes) ended(ctx context.Context, e jetstream.KeyValueEntry) {
if e.Operation() != jetstream.KeyValuePurge {
key, ok := versions.PublishBoundOf(e.Key())
if e.Operation() != jetstream.KeyValuePurge || !ok {
return
}
raw, _, found, err := p.versions.Store.Get(ctx, e.Key())
raw, _, found, err := p.versions.Store.Get(ctx, key)
if err != nil || !found {
return
}
@@ -53,7 +53,14 @@ func (r *engineReads) stepWindow() {
ctx = context.Background()
}
_, counted := r.forms()
r.window.step(ctx, counted)
r.window.step(ctx, r.named(), counted)
}
// named are the nodes the placement names, none dropped: what the window's step judges a read by.
func (r *engineReads) named() []string {
r.mu.Lock()
defer r.mu.Unlock()
return slices.Clone(r.counted)
}
// bind keeps the statements in kv and hears it: every node's last statement at the start, a person's ask again and drop
@@ -138,7 +145,7 @@ func (r *engineReads) built() (present, counted map[string][]string) {
if ctx == nil {
ctx = context.Background()
}
r.window.step(ctx, counted)
r.window.step(ctx, r.named(), counted)
}
return present, counted
}
@@ -361,7 +361,7 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) {
if got, err := health.DecodeFinding(st.vals[refusedKey]); err != nil || !strings.Contains(got.Summary, "refused to publish m/program@0.1.3") || !strings.Contains(got.Summary, "bytes differ") {
t.Fatalf("a holder's refused: the finding = %+v, %v", got, err)
}
pubs.ended(context.Background(), purged{key: versions.Key("m/program", "0.1.4")})
pubs.ended(context.Background(), purged{key: versions.PublishBoundKey("m/program", "0.1.4")})
unansweredKey := health.FindingKey(health.ScopeModule, "m/program", versions.PublishUnanswered)
if got, err := health.DecodeFinding(st.vals[unansweredKey]); err != nil || !strings.Contains(got.Summary, "neither published nor refused of m/program@0.1.4") {
t.Fatalf("a bound ended unanswered: the finding = %+v, %v", got, err)
@@ -399,19 +399,24 @@ func aFailedTrunkBuildIsAFinding(t *testing.T) {
reads := &engineReads{keeper: &enginereads.Keeper{}, window: window, counted: []string{"a", "b"}}
newest := []string{forms.FormNeed("oci")}
counted := map[string][]string{"a": newest, "b": nil}
window.step(context.Background(), counted)
window.step(context.Background(), []string{"a", "b"}, counted)
heldKey := health.FindingKey(health.ScopeMachine, "b", versions.OldFormHeld)
if !window.w.Open || !slices.Equal(window.w.Holding, []string{"b"}) || st.vals[heldKey] != nil {
t.Fatalf("one node lacking the newest form: %+v", window.w)
}
// Security-upgrade F3 (review round 1): an empty node_haves read, naming no node, leaves the window open.
window.step(context.Background(), nil, map[string][]string{})
if !window.w.Open || !slices.Equal(window.w.Holding, []string{"b"}) || len(closings) != 0 {
t.Fatalf("an empty read closed the window: %+v, closings %+v", window.w, closings)
}
now = now.Add(versions.FindingAfter)
window.step(context.Background(), counted)
window.step(context.Background(), []string{"a", "b"}, counted)
if got, err := health.DecodeFinding(st.vals[heldKey]); err != nil || !strings.Contains(got.Summary, "the node-engine of b") {
t.Fatalf("seven days held: the finding = %+v, %v", got, err)
}
counted["b"] = newest
window.step(context.Background(), counted)
window.step(context.Background(), counted)
window.step(context.Background(), []string{"a", "b"}, counted)
window.step(context.Background(), []string{"a", "b"}, counted)
if window.w.Open || len(closings) != 1 || !slices.Equal(closings[0].LastHolding, []string{"b"}) || st.vals[heldKey] != nil {
t.Errorf("the node back with the newest form: %+v, closings %+v, finding left %v", window.w, closings, st.vals[heldKey] != nil)
}
@@ -64,16 +64,12 @@ func Holding(fs forms.ArtifactForms, counted map[string][]string, dropped map[st
return out
}
// OldFormWindow is the window's record: whether it is open, who holds it and since when, and the nodes dropped by
// hand; what builder_old_form answers and what `old-form-closed` carries when it closes.
// OldFormWindow is the window's record: whether it is open, who holds it and since when (delivery.WindowHeld, the part
// every window keeper shares), and the nodes dropped by hand; what builder_old_form answers and what `old-form-closed`
// carries when it closes.
type OldFormWindow struct {
Open bool `json:"open"`
Holding []string `json:"holding,omitempty"`
Since map[string]time.Time `json:"since,omitempty"`
Dropped []Dropped `json:"dropped,omitempty"`
Closed time.Time `json:"closed,omitzero"`
// LastHolding are the nodes that held it last, said when it closes.
LastHolding []string `json:"last_holding,omitempty"`
delivery.WindowHeld
Dropped []Dropped `json:"dropped,omitempty"`
}
// Dropped is one retired node dropped from the count by hand (builder_old_form): a hand act with who and why.
@@ -95,31 +91,14 @@ type OldFormClosedBody struct {
// FindingAfter is how long a node holds the window open before it is the builder's finding naming the node.
const FindingAfter = 7 * 24 * time.Hour
// Step moves the window to the nodes holding it at now: it answers the new window, whether it closed in this step (so
// the builder says `old-form-closed` and its log line once), and the nodes holding it past FindingAfter (the finding,
// cleared for every other node).
func (w OldFormWindow) Step(holding []string, now time.Time) (next OldFormWindow, closed bool, overdue []string) {
next = OldFormWindow{Dropped: w.Dropped}
if len(holding) == 0 {
next.Closed, next.LastHolding = w.Closed, w.LastHolding
if w.Open {
next.Closed, next.LastHolding = now.UTC(), w.Holding
return next, true, nil
}
return next, false, nil
}
next.Open, next.Holding, next.Since = true, slices.Clone(holding), map[string]time.Time{}
for _, n := range holding {
since, had := w.Since[n]
if !had {
since = now.UTC()
}
next.Since[n] = since
if now.Sub(since) >= FindingAfter {
overdue = append(overdue, n)
}
}
return next, false, overdue
// Step moves the window by mesh-sdk's one window rule (delivery.StepWindow): named is every counted node the read named,
// those dropped by hand included, and holding those of them holding the window (Holding). A read naming no node (no
// placement read yet, an empty node_haves) judges nothing, so it never closes the window. It answers the new window,
// whether it closed in this step (so the builder says `old-form-closed` and its log line once), and the nodes holding it
// past FindingAfter (the finding, cleared for every other node).
func (w OldFormWindow) Step(named, holding []string, now time.Time) (next OldFormWindow, closed bool, overdue []string) {
held, closed, overdue := delivery.StepWindow(w.WindowHeld, named, holding, now, FindingAfter)
return OldFormWindow{WindowHeld: held, Dropped: w.Dropped}, closed, overdue
}
// Drop drops a retired node from the count by hand; dropping it again changes nothing.
@@ -1,6 +1,9 @@
package versions
import "time"
import (
"strings"
"time"
)
// **A publish that ends badly is loud** (design 51 *Health and conditions*; review round 1): the builder raises the
// finding publish-refused when a holder says refused, and publish-unanswered when a version it built and numbered is
@@ -8,12 +11,15 @@ import "time"
// clears both. A seat's outcome that does not read is the finding seat-outcome-unread about the seat, cleared by its
// next outcome that reads.
//
// The bound is a key of the state publish-bounds per version (Key), written with its own time to live when the build
// is done and the version is still numbered, deleted at the holder's word; the bus's marker of its end, heard on the
// watch, is what raises publish-unanswered while the version is still numbered. No timer.
// The bound is a key of the builder's state bounds per version (PublishBoundKey), written with its own time to live
// when the build is done and the version is still numbered, deleted at the holder's word; the bus's marker of its end,
// heard on the watch, is what raises publish-unanswered while the version is still numbered. No timer.
const (
PublishBoundsState = "publish-bounds"
PublishBoundsBucket = "mesh-delivery-builder_publish-bounds"
// BoundsState is the builder's state of its expiring keys, each with its own time to live: a publish's bound
// (PublishBoundKey) and a node's seven days holding the old form's window (OldFormBoundKey). Its end, the bus's
// marker heard on the watch, is what acts.
BoundsState = "bounds"
BoundsBucket = "mesh-delivery-builder_bounds"
// PublishBound is the build module's own wait for the holder's word (design 51 *How a publish ends*: at most
// 10 minutes), so a version unanswered past it is one the holder never answered.
PublishBound = 10 * time.Minute
@@ -22,3 +28,23 @@ const (
PublishUnanswered = "publish-unanswered"
SeatOutcomeUnread = "seat-outcome-unread"
)
// The prefixes of the keys of BoundsState.
const (
publishBoundPrefix = "publish."
oldFormBoundPrefix = "old-form."
)
// PublishBoundKey is the bound of name's version: `publish.<versions key>`.
func PublishBoundKey(name, version string) string { return publishBoundPrefix + Key(name, version) }
// OldFormBoundKey is the bound of node's seven days holding the old form's window: `old-form.<node>`.
func OldFormBoundKey(node string) string { return oldFormBoundPrefix + EscapeToken(node) }
// PublishBoundOf is the versions key a publish's bound names, and whether key is one; a key that is no publish's
// bound is an old form's (IsOldFormBound) or nothing the builder wrote.
func PublishBoundOf(key string) (string, bool) { return strings.CutPrefix(key, publishBoundPrefix) }
// IsOldFormBound says key is a node's bound of the old form's window: its end steps the window, which raises the
// finding for every node past FindingAfter.
func IsOldFormBound(key string) bool { return strings.HasPrefix(key, oldFormBoundPrefix) }
@@ -335,11 +335,12 @@ func TestTheBuildersVersions(t *testing.T) {
t.Errorf("forms.Current, a node stating nothing and one stating form:seat-reference: Holding = %v, want a", got)
}
t0 := time.Unix(0, 0)
w, closed, overdue := OldFormWindow{}.Step([]string{"c"}, t0)
named := []string{"c", "d"}
w, closed, overdue := OldFormWindow{}.Step(named, []string{"c"}, t0)
if closed || len(overdue) != 0 || !w.Open {
t.Fatalf("Step = %+v %v %v", w, closed, overdue)
}
w, _, overdue = w.Step([]string{"c"}, t0.Add(FindingAfter))
w, _, overdue = w.Step(named, []string{"c"}, t0.Add(FindingAfter))
if !reflect.DeepEqual(overdue, []string{"c"}) {
t.Errorf("seven days held: overdue %v, want c", overdue)
}
@@ -353,7 +354,12 @@ func TestTheBuildersVersions(t *testing.T) {
if _, err := w.Drop("e", "op@shanks", "", t0); err == nil {
t.Error("a drop with no why is taken")
}
w, closed, _ = w.Step(nil, t0.Add(FindingAfter+time.Hour))
// Security-upgrade F3 (review round 1), by mesh-sdk's delivery.StepWindow: a read naming no node (an empty
// node_haves, the placement not read yet) never closes the window, and its holder's seven days still count.
if held, shut, late := w.Step(nil, nil, t0.Add(FindingAfter+time.Minute)); shut || !held.Open || !reflect.DeepEqual(late, []string{"c"}) {
t.Fatalf("a read naming no node: %+v closed=%v overdue %v, want the window open and c overdue", held, shut, late)
}
w, closed, _ = w.Step(named, nil, t0.Add(FindingAfter+time.Hour))
if !closed || w.Open || !reflect.DeepEqual(w.LastHolding, []string{"c"}) || w.Closing().Dropped[0].Node != "d" {
t.Errorf("closing: %+v closed=%v", w, closed)
}
@@ -361,7 +367,7 @@ func TestTheBuildersVersions(t *testing.T) {
if back, err := DecodeOldFormWindow(raw); err != nil || !back.Closed.Equal(w.Closed) {
t.Errorf("the window read back = %+v, %v", back, err)
}
if _, again, _ := w.Step(nil, t0.Add(FindingAfter+2*time.Hour)); again {
if _, again, _ := w.Step(named, nil, t0.Add(FindingAfter+2*time.Hour)); again {
t.Error("a closed window closes again: old-form-closed would be said twice")
}
})
@@ -94,7 +94,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: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "publish-bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
if want := []state{{Name: "builds", History: 5}, {Name: "engine-reads", History: 1}, {Name: "findings", History: 5}, {Name: "floor-waits", History: 1}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "old-form", History: 1}, {Name: "bounds", History: 1, KeyTTL: true}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "tag-ask-bounds", History: 1, KeyTTL: true}, {Name: "tag-asks", History: 1}, {Name: "versions", History: 1}, {Name: "first-views", 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
+1 -1
View File
@@ -81,7 +81,7 @@
{ "name": "hearing", "history": 1 },
{ "name": "instances", "history": 1 },
{ "name": "old-form", "history": 1 },
{ "name": "publish-bounds", "history": 1, "key-ttl": true },
{ "name": "bounds", "history": 1, "key-ttl": true },
{ "name": "queues", "history": 1 },
{ "name": "reasks", "history": 1 },
{ "name": "tag-ask-bounds", "history": 1, "key-ttl": true },