Stop watching the node-engine's have in the checker: a check's asks wait in the build seats' streams, which redeliver a lost node's ask, so no check fails for want of a node; presence stays for an unheard forge and the assigner's presence (issue 1912)

This commit is contained in:
2026-10-23 16:41:34 +02:00
parent 84b2c92ccf
commit 37860d46a0
5 changed files with 35 additions and 33 deletions
@@ -197,7 +197,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
tools := &buildrun.ToolchainSet{}
presence := &buildrun.PresenceSet{}
have := &buildrun.HaveSet{}
trunks := &check.Trunks{}
// A check is one ask per pull request and build module on the build seats' check stream, pulled by a build node with a
// free slot (novox/hq ADR 0525 §1, §3): the checker chooses no node and starts no step.
@@ -279,8 +278,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
defer cc.Stop()
}
errs := make(chan error, 8)
// The watchman's presence says which nodes are asleep. Where its bucket is not there yet no node is online, so nothing
// is checked; the checker says so and is started again with the watchman.
// The watchman's presence: a node coming back online asks again a check whose forge was not heard (UnheardWords), and
// the assigner's own presence says whether a composition is being made (compositionWatch, ADR 0537 §4). A check is no
// longer held for a node: its asks wait in the build seats' streams, which redeliver a lost node's ask (ADR 0525 §6,
// issue 1912), so the node-engine's have is not watched here.
switch kv, err := js.KeyValue(ctx, delivery.PresenceBucket); {
case errors.Is(err, jetstream.ErrBucketNotFound):
logger.Printf("the bucket %s does not exist yet: no node counts as online until the checker starts again", delivery.PresenceBucket)
@@ -294,27 +295,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go func() {
errs <- follow(ctx, pw, func(e jetstream.KeyValueEntry) {
if presence.Put(e.Key(), e.Value(), !opOf(e)) {
ch.NodesChanged(ctx) // a node came online or went: a check that waited for one may go on
}
})
}()
}
// The node-engine's have says whether a node's build module runs and is healthy: a node whose build module is not is not
// chosen. Where its bucket is not there yet no build module counts as healthy, so nothing is checked.
switch kv, err := js.KeyValue(ctx, delivery.HaveBucket); {
case errors.Is(err, jetstream.ErrBucketNotFound):
logger.Printf("the bucket %s does not exist yet: no build module counts as healthy until the checker starts again", delivery.HaveBucket)
case err != nil:
return fmt.Errorf("the bucket %s: %w", delivery.HaveBucket, err)
default:
hw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) { have.Put(e.Key(), e.Value(), !opOf(e)) })
if err != nil {
return err
}
go func() {
errs <- follow(ctx, hw, func(e jetstream.KeyValueEntry) {
if have.Put(e.Key(), e.Value(), !opOf(e)) {
ch.NodesChanged(ctx)
ch.NodesChanged(ctx) // a node came online or went: a check whose forge was not heard may now be heard
}
})
}()
@@ -0,0 +1,21 @@
package main
import (
"os"
"strings"
"testing"
)
// 1912: with checks as asks on the build seats' streams, the stream redelivers a lost node's ask (ADR 0525 §6), so the
// checker program opens no node-engine have to start again a check that failed for want of a node.
func TestTheCheckerWatchesNoHave(t *testing.T) {
raw, err := os.ReadFile("main.go")
if err != nil {
t.Fatal(err)
}
for _, s := range []string{"delivery.HaveBucket", "buildrun.HaveSet"} {
if strings.Contains(string(raw), s) {
t.Errorf("the checker program still reads %s", s)
}
}
}
@@ -14,10 +14,6 @@ import (
// DefaultBound is how long a check may run before it is ended failed naming the step it was stuck at (ADR 0484, issue 1678).
const DefaultBound = 30 * time.Minute
// NodesWords is in the Words of a failure that is the build nodes' (none qualifies, the toolchains disagree): such a verdict
// is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8).
const NodesWords = "build node qualifies"
// UnheardWords end the Words of a failure because a part of the mesh the check reads was not heard (the forge, issue 1624):
// like a failure of the nodes, it is not the change's, and is judged again when the nodes change.
const UnheardWords = "not heard; judged again when the nodes change"
@@ -307,9 +303,11 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
return &claimed{rev}, nil
}
// IsNodesFailure says whether a verdict failed because no build node could take the check.
// IsNodesFailure says whether a verdict failed because a part of the mesh the check reads was not heard (UnheardWords):
// it is judged again when the nodes change. No check fails for want of a build node any more: its asks wait in the build
// seats' streams (ADR 0525 §6, issue 1912).
func IsNodesFailure(v delivery.Verdict) bool {
return v.State == delivery.VerdictFail && (strings.Contains(v.Failed.Words, NodesWords) || strings.Contains(v.Failed.Words, UnheardWords))
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, UnheardWords)
}
// try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends
@@ -208,11 +208,13 @@ func TestACheckPastItsBoundFailsNamingTheStuckStep(t *testing.T) { // N6/AC3
}
}
func TestNoQualifyingNodeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // N6/AC4
// A check whose forge was not heard is judged again when the nodes change (issue 1912: no check fails for want of a build
// node any more; its asks wait in the streams).
func TestAnUnheardForgeFailsWithTheReasonAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // N6/AC4
ctx := context.Background()
j := &judgeFunc{f: func(_ context.Context, _ *Case, n int) (Outcome, error) {
if n == 1 {
return Outcome{Verdict: failure("shanks", "no go "+NodesWords+": none is online")}, nil
return Outcome{Verdict: failure("the forge", "novox/mesh-delivery#7: go.mod cannot be read: no responders (the forge was "+UnheardWords+")")}, nil
}
return Outcome{Verdict: passing()}, nil
}}
@@ -39,7 +39,7 @@ func TestCheckerStatusNamesTheWaitingChecksByKind(t *testing.T) { // 1897/AC4
ctx := context.Background()
j := &judgeFunc{f: func(_ context.Context, c *Case, _ int) (Outcome, error) {
if c.Group == group(2) {
return Outcome{Verdict: failure("checker", "no "+NodesWords+": every node is offline")}, nil
return Outcome{Verdict: failure("the forge", "go.mod cannot be read: no responders (the forge was "+UnheardWords+")")}, nil
}
return Outcome{Verdict: passing()}, nil
}}