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

Checker and builder wait through mesh-sdk's HearEnds, so a deaf watch loses no step end in either

Browse Source 
The checker's HeardEnds moved to mesh-sdk buildrun (fix/lost-check,
6aeb401): both modules pin it, and the builder, which waits on the same
Asker, now reads an ask's step at each step end of its build module too.
checker_ask_again and checker_status's lost stay in the checker.

Tests: one row in the builder's ask table (its watch deaf, read at a
nudge, done with no new ask); counts unchanged.
This commit is contained in:
jschoubben
2026-10-28 20:31:04 +01:00
parent 6f079f13a8
commit cb46ff8735
10 changed files with 47 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go
+11 -2
View File
@@ -92,7 +92,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, delivery.AskLanguages, dead, logger.Printf,
buildrun.SayHearing(ctx, hearingKV, logger.Printf))
defer hearing.Stop()
conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again() })
// Each ask's step is heard by its own watch and, at every step end of its build module and every reconnect, read
// (buildrun.HearEnds): a watch that goes deaf loses no end (the lost check of 2026-10-28).
var buildModules []string
for _, l := range delivery.AskLanguages {
if m, err := delivery.BuildModuleOf(l); err == nil {
buildModules = append(buildModules, m)
}
}
heard := buildrun.HearEnds(ctx, asks, asks, buildModules, logger.Printf)
conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") })
// The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs
// (the seat verb held, novox/hq issue 1937) before it leaves the key done.
calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
@@ -102,7 +111,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
defer calls.Close()
b := &build.Builder{
Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: asks, Dead: dead, Hearing: hearing}, Bound: bound},
Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound},
Held: heldOnTheBus{js},
// An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound.
Gone: heldOnTheBus{js},
modules/mesh-delivery-builder/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-builder
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-builder/go.sum
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 h1:ycu/sNieTdLpD3bzhZOCXdzcoqXUOxPRmqWV019REpo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529 h1:Xx5mcDPWcRuPelUxdjvwkmVcfyGww0rRajA9JjFA/hk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-delivery-builder/internal/build/run_test.go
+26 -3
View File
@@ -691,6 +691,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
byHand bool
always bool // Gone says every ask cannot be answered, not only the first
dead bool // DEAD_LETTERS keeps the first ask: the seat gave it up (issue 734)
deaf bool // the ask's own watch hears nothing though its step ends (the lost check of 2026-10-28)
puts int
done bool
asking int // the asking the event says, 0 when none is said
@@ -739,6 +740,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
{name: "given up by its stream: asked again under a new id and done from it", puts: 2, done: true, asking: 2},
{name: "asked again by hand: the same, with no check bound", byHand: true, puts: 2, done: true, asking: 2},
{name: "given up into DEAD_LETTERS: its wait ends failed with the words and it is asked again at once", dead: true, puts: 2, done: true, asking: 2},
{name: "its step ended while its watch was deaf: read at its build module's next step end, done with no new ask", deaf: true, puts: 1, done: true},
{name: "never answerable: failed after its askings", askings: 2, always: true, puts: 2},
{name: "taken over at its last asking: failed, not asked again", askings: 3, always: true, step: "asked mesh-build-go again (asking 3): x", puts: 0},
} {
@@ -829,7 +831,12 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
if c.dead {
r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: buildrun.LostAskWaiter{Inner: r.asks, Dead: dead}, Bound: 2 * time.Second}
}
if !c.byHand && !c.dead {
var heard *buildrun.HeardEnds
if c.deaf {
heard = &buildrun.HeardEnds{Inner: silentWatch{}, Read: r.asks}
r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: heard, Bound: 2 * time.Second}
}
if !c.byHand && !c.dead && !c.deaf {
gone.why = func(a delivery.StepAsk) string {
if c.always || a.ID == first || first == "" {
return "it waits first in line while a node took a later ask"
@@ -851,7 +858,16 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
}
}
}
if c.done {
if c.deaf {
r.asks.End(r.answer(waitForAsks(t, r, 1)[0]))
for deadline := time.Now().Add(2 * time.Second); heard.Waiting() == 0; time.Sleep(time.Millisecond) {
if time.Now().After(deadline) {
t.Fatal("no ask is waited on")
}
}
heard.Nudge(buildrun.GoBuildModule) // a step of mesh-build-go ended
}
if c.done && !c.deaf {
asks := waitForAsks(t, r, c.puts)
if first == "" && !c.byHand {
first = asks[0].ID
@@ -876,7 +892,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
mu.Lock()
defer mu.Unlock()
// Note 62: the old message of an ask asked again because its stream gave it up is dropped from the stream.
if gone.mu.Lock(); c.done && !c.byHand && !c.dead && (len(gone.dropped) != 1 || gone.dropped[0] != first) {
if gone.mu.Lock(); c.done && !c.byHand && !c.dead && !c.deaf && (len(gone.dropped) != 1 || gone.dropped[0] != first) {
t.Errorf("dropped %v from the stream, want the old ask %s", gone.dropped, first)
}
gone.mu.Unlock()
@@ -1086,3 +1102,10 @@ func TestATakeOverThatCannotReadTheStreamWaitsForTheAsk(t *testing.T) {
t.Fatalf("the key = %+v after %d asks, want done from the ask's step, none put", v, len(r.asks.Asks()))
}
}
// silentWatch is a watch of an ask's step that stays open and hears nothing.
type silentWatch struct{}
func (silentWatch) WatchAsk(context.Context, string, string) (<-chan delivery.Step, error) {
return make(chan delivery.Step), nil
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go
+2 -3
View File
@@ -145,9 +145,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf))
defer hearing.Stop()
// Each ask's step is heard by its own watch and, at every step end of its build module and every reconnect, read
// (check.HeardEnds): a watch that goes deaf loses no end (the lost check of 2026-10-28).
heard := &check.HeardEnds{Inner: asks, Read: askSteps{js: js}, Logf: logger.Printf}
hearStepEnds(ctx, js, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, heard, logger.Printf)
// (buildrun.HearEnds): a watch that goes deaf loses no end (the lost check of 2026-10-28).
heard := buildrun.HearEnds(ctx, asks, asks, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, logger.Printf)
conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") })
waiter := buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}
ch := &check.Checker{
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/stepends.go
-83
View File
@@ -1,83 +0,0 @@
package main
import (
"context"
"fmt"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/process"
"mesh-delivery-checker/internal/check"
)
// askSteps reads an ask's step keys now, from its build module's steps state (check.StepReader): a read, not a watch.
type askSteps struct{ js jetstream.JetStream }
func (a askSteps) AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error) {
pattern, err := delivery.AskStepWatch(askID)
if err != nil {
return nil, err
}
kv, err := a.js.KeyValue(ctx, delivery.StepsBucket(module))
if err != nil {
return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
}
lister, err := kv.ListKeysFiltered(ctx, pattern)
if err != nil {
return nil, err
}
var out []delivery.Step
for key := range lister.Keys() {
e, err := kv.Get(ctx, key)
if err != nil {
continue // deleted between the list and the read
}
if step, err := delivery.DecodeStep(e.Value()); err == nil {
out = append(out, step)
}
}
return out, nil
}
// hearStepEnds nudges every ask of a build module the checker waits on each time a step of that module ends (any key of
// its steps state written not running): an ask whose own watch went deaf reads its end then (check.HeardEnds). One watch
// per build module, updates only; one that closes is opened again while it delivered something since it was opened, and
// said and left otherwise. No timer.
func hearStepEnds(ctx context.Context, js process.BucketOpener, modules []string, heard *check.HeardEnds, logf func(string, ...any)) {
for _, module := range modules {
kv := process.OptionalBucket(ctx, js, delivery.StepsBucket(module), "a check whose own watch of its step goes deaf is read only at a reconnect or by hand", logf)
if kv == nil {
continue
}
go func() {
for ctx.Err() == nil {
w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
if err != nil {
logf("the steps of %s cannot be watched, so a check whose own watch goes deaf is read only at a reconnect or by hand: %v", module, err)
return
}
heardAny := false
for e := range w.Updates() {
if e == nil || e.Operation() != jetstream.KeyValuePut {
continue
}
heardAny = true
if step, err := delivery.DecodeStep(e.Value()); err == nil && step.State != delivery.StepRunning {
heard.Nudge(module)
}
}
_ = w.Stop()
if ctx.Err() == nil {
heard.Nudge(module) // a step may have ended while the watch was closing
if !heardAny {
logf("the watch of the steps of %s closed having heard nothing, so it is left: a check whose own watch goes deaf is read only at a reconnect or by hand", module)
return
}
logf("the watch of the steps of %s closed; opened again", module)
}
}
}()
}
}
modules/mesh-delivery-checker/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-checker
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-checker/go.sum
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388 h1:ycu/sNieTdLpD3bzhZOCXdzcoqXUOxPRmqWV019REpo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028174713-38dbe1d3a388/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529 h1:Xx5mcDPWcRuPelUxdjvwkmVcfyGww0rRajA9JjFA/hk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261028192930-6aeb40109529/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-delivery-checker/internal/check/heardends.go
-144
View File
@@ -1,144 +0,0 @@
package check
import (
"context"
"sync"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
)
// **A step's end is never only its own watch's to hear** (the lost check of 2026-10-28: mesh-catalog #553's verdict
// stayed running after its step mvsgthyjed5786 had passed on shanks at 19:09:31Z; the watch of that one key, opened at
// 18:46:05Z, heard neither the step's start nor its end and never closed, while a watch opened at 18:58:48Z heard its own
// step end). HeardEnds stands between the asker (buildrun.Asker) and the watch of each ask's step key: when it is
// nudged, every ask it waits on reads its step keys now (a read, not the watch) and hands what it reads to its wait, so a
// step that ended is judged even when its own watch went deaf. A watch that closes under a wait is opened again, at most
// reopenTries times. It is nudged on an event (Nudge): any step of the same build module ending (the checker's one watch of
// each steps state), a reconnect to the bus, and the hand (checker_ask_again's verdict). It holds no timer.
// StepReader reads, now, every step a build module's steps state holds for an ask: its keys `*.<ask id>`.
type StepReader interface {
AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error)
}
// reopenTries is how often a wait opens its ask's watch again after it closed, before the wait ends failed.
const reopenTries = 3
// HeardEnds is an AskWaiter over Inner that also reads an ask's steps when nudged. Safe for concurrent use.
type HeardEnds struct {
Inner buildrun.AskWaiter
Read StepReader
Logf func(format string, args ...any)
mu sync.Mutex
waiting map[string]*waitedAsk // by ask id
}
type waitedAsk struct {
module string
nudge chan struct{}
}
func (h *HeardEnds) logf(format string, args ...any) {
if h.Logf != nil {
h.Logf(format, args...)
}
}
// WatchAsk is AskWaiter's: what Inner's watch hands, and, at each nudge, what the ask's keys hold now.
func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
in, err := h.Inner.WatchAsk(ctx, module, askID)
if err != nil {
return nil, err
}
w := &waitedAsk{module: module, nudge: make(chan struct{}, 1)}
h.mu.Lock()
if h.waiting == nil {
h.waiting = map[string]*waitedAsk{}
}
h.waiting[askID] = w
h.mu.Unlock()
out := make(chan delivery.Step)
go func() {
defer close(out)
defer func() {
h.mu.Lock()
if h.waiting[askID] == w {
delete(h.waiting, askID)
}
h.mu.Unlock()
}()
send := func(s delivery.Step) bool {
select {
case out <- s:
return true
case <-ctx.Done():
return false
}
}
reopened := 0
for {
select {
case <-ctx.Done():
return
case s, ok := <-in:
if !ok {
if ctx.Err() != nil || reopened >= reopenTries {
h.logf("the watch of the ask %s's step on %s closed %d times; its wait ends", askID, module, reopened+1)
return
}
reopened++
if in, err = h.Inner.WatchAsk(ctx, module, askID); err != nil {
h.logf("the watch of the ask %s's step on %s closed and cannot be opened again, so its wait ends: %v", askID, module, err)
return
}
h.logf("the watch of the ask %s's step on %s closed; opened again (%d of %d)", askID, module, reopened, reopenTries)
continue
}
if !send(s) {
return
}
case <-w.nudge:
if h.Read == nil {
continue
}
steps, err := h.Read.AskSteps(ctx, module, askID)
if err != nil {
h.logf("the steps of the ask %s on %s cannot be read: %v", askID, module, err)
continue
}
for _, s := range steps {
if s.State != delivery.StepRunning {
h.logf("the ask %s's step on %s ended %s, read at a nudge: its watch had not handed it over", askID, s.Node, s.State)
}
if !send(s) {
return
}
}
}
}
}()
return out, nil
}
// Nudge has every ask waited on of module read its steps now; module "" nudges every ask.
func (h *HeardEnds) Nudge(module string) {
h.mu.Lock()
defer h.mu.Unlock()
for _, w := range h.waiting {
if module == "" || w.module == module {
select {
case w.nudge <- struct{}{}:
default: // a nudge is pending already
}
}
}
}
// Waiting is how many asks are waited on now.
func (h *HeardEnds) Waiting() int {
h.mu.Lock()
defer h.mu.Unlock()
return len(h.waiting)
}
modules/mesh-delivery-checker/internal/check/lostask_test.go
+2 -2
View File
@@ -269,7 +269,7 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) {
t.Run(row.name, func(t *testing.T) {
b := newBench(t)
var asked func() int
var heard *HeardEnds
var heard *buildrun.HeardEnds
var tr *troubled
if row.lose > 0 {
l := losing(b, row.lose)
@@ -277,7 +277,7 @@ func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) {
} else {
tr = &troubled{FakeAsks: &buildrun.FakeAsks{}, answer: b.answer, restarts: row.restarts, deaf: row.deaf,
deafIDs: map[string]bool{}, steps: map[string]delivery.Step{}}
heard = &HeardEnds{Inner: tr, Read: tr}
heard = &buildrun.HeardEnds{Inner: tr, Read: tr}
b.ex.Asks = buildrun.Asker{Publisher: tr, Waiter: heard, Bound: 5 * time.Second}
asked = func() int { return len(tr.Asks()) }
}