Merge pull request 'The checker and the builder use it; checker_ask_again asks a lost check again; the checker's integration bed writes a node record (issue 734's kind)' (#103) from fix/lost-check into main
This commit was merged in pull request #103.
This commit is contained in:
@@ -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,
|
hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, delivery.AskLanguages, dead, logger.Printf,
|
||||||
buildrun.SayHearing(ctx, hearingKV, logger.Printf))
|
buildrun.SayHearing(ctx, hearingKV, logger.Printf))
|
||||||
defer hearing.Stop()
|
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 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.
|
// (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))
|
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()
|
defer calls.Close()
|
||||||
b := &build.Builder{
|
b := &build.Builder{
|
||||||
Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
|
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},
|
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.
|
// 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},
|
Gone: heldOnTheBus{js},
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ module mesh-delivery-builder
|
|||||||
go 1.26.0
|
go 1.26.0
|
||||||
|
|
||||||
require (
|
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.20261028193502-f91340fe82ef
|
||||||
github.com/nats-io/nats-server/v2 v2.11.17
|
github.com/nats-io/nats-server/v2 v2.11.17
|
||||||
github.com/nats-io/nats.go v1.54.0
|
github.com/nats-io/nats.go v1.54.0
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -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.20261028193502-f91340fe82ef h1:ggwHXCvAOhBCKAY+ErehKEhu/IQTS4poJBRDokGYnDc=
|
||||||
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.20261028193502-f91340fe82ef/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
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=
|
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||||
|
|||||||
@@ -691,6 +691,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
|
|||||||
byHand bool
|
byHand bool
|
||||||
always bool // Gone says every ask cannot be answered, not only the first
|
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)
|
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
|
puts int
|
||||||
done bool
|
done bool
|
||||||
asking int // the asking the event says, 0 when none is said
|
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: "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: "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: "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: "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},
|
{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 {
|
if c.dead {
|
||||||
r.b.Asks = buildrun.Asker{Publisher: r.asks, Waiter: buildrun.LostAskWaiter{Inner: r.asks, Dead: dead}, Bound: 2 * time.Second}
|
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 {
|
gone.why = func(a delivery.StepAsk) string {
|
||||||
if c.always || a.ID == first || first == "" {
|
if c.always || a.ID == first || first == "" {
|
||||||
return "it waits first in line while a node took a later ask"
|
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)
|
asks := waitForAsks(t, r, c.puts)
|
||||||
if first == "" && !c.byHand {
|
if first == "" && !c.byHand {
|
||||||
first = asks[0].ID
|
first = asks[0].ID
|
||||||
@@ -876,7 +892,7 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
|
|||||||
mu.Lock()
|
mu.Lock()
|
||||||
defer mu.Unlock()
|
defer mu.Unlock()
|
||||||
// Note 62: the old message of an ask asked again because its stream gave it up is dropped from the stream.
|
// 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)
|
t.Errorf("dropped %v from the stream, want the old ask %s", gone.dropped, first)
|
||||||
}
|
}
|
||||||
gone.mu.Unlock()
|
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()))
|
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
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
// Command mesh-delivery-checker-tools serves the checker's read tools (novox/hq ADR 0484 clause 13, issue 1686): the runtime
|
// Command mesh-delivery-checker-tools serves the checker's read tools (novox/hq ADR 0484 clause 13, issue 1686): the runtime
|
||||||
// launches it for a call, it reads the checker's verdicts and kept states and the indexer's candidates through the checker's
|
// launches it for a call, it reads the checker's verdicts and kept states and the indexer's candidates through the checker's
|
||||||
// state and reads, and writes nothing. The checker is the other program: it serves no tool (ADR 0437 §4).
|
// state and reads. It writes only the tries (the try's tools) and, asking a check again by hand, one verdict. The checker is the other program: it serves no tool (ADR 0437 §4).
|
||||||
package main
|
package main
|
||||||
|
|
||||||
import (
|
import (
|
||||||
@@ -235,6 +235,24 @@ func Tools(r check.Reader) []stdio.Tool {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// RecoveryTools are the tools that steer a check by hand (tools-are-how-a-module-is-spoken-to): ask a check again whose
|
||||||
|
// wait judged nothing (the lost check of 2026-10-28). They write the checker's verdicts through w.
|
||||||
|
func RecoveryTools(r check.Reader, w check.VerdictEntry) []stdio.Tool {
|
||||||
|
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s }
|
||||||
|
return []stdio.Tool{{Name: check.AskAgainTool,
|
||||||
|
Description: "Ask a candidate's check again by hand, whose verdict stays running while nothing runs for it (" + check.StatusReadTool + " names it " +
|
||||||
|
"under lost): its verdict is ended as asked again, saying why, and the checker, which hears its verdicts, stops the wait that judged nothing " +
|
||||||
|
"and asks the check anew, whole. Refused for a group with no candidate, a verdict that is not running (a pass or failure stands until a push or " +
|
||||||
|
"a trunk move), and while a step of its pull requests can still run on a build module (its end judges it). The checker reads a step that " +
|
||||||
|
"ended while its watch was deaf by itself at the next step end of its build module; this is for what that does not reach.",
|
||||||
|
Input: map[string]any{"type": "object", "required": []string{"group", "why"}, "properties": map[string]any{
|
||||||
|
"group": map[string]any{"type": "string", "description": "the delivery group's branch name"},
|
||||||
|
"why": map[string]any{"type": "string", "description": "why it is asked again, in words; kept in the verdict it ends"}}},
|
||||||
|
Run: func(args map[string]any) (any, error) {
|
||||||
|
return r.AskAgain(w, str(args, "group"), str(args, "why"))
|
||||||
|
}}}
|
||||||
|
}
|
||||||
|
|
||||||
// stepBound is the checker's step bound as the checker reads it (MESH_CHECKER_STEP_BOUND), its default when unset or not a
|
// stepBound is the checker's step bound as the checker reads it (MESH_CHECKER_STEP_BOUND), its default when unset or not a
|
||||||
// duration.
|
// duration.
|
||||||
func stepBound() time.Duration {
|
func stepBound() time.Duration {
|
||||||
@@ -249,7 +267,8 @@ func main() {
|
|||||||
PreviewState: read(stdio.State("previews")),
|
PreviewState: read(stdio.State("previews")),
|
||||||
StepStates: map[string]check.StateRead{"mesh-build-go": read(stdio.State("mesh-build-go.steps")), "mesh-build-node": read(stdio.State("mesh-build-node.steps"))},
|
StepStates: map[string]check.StateRead{"mesh-build-go": read(stdio.State("mesh-build-go.steps")), "mesh-build-node": read(stdio.State("mesh-build-node.steps"))},
|
||||||
HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound(),
|
HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound(),
|
||||||
TurnState: read(stdio.State("mesh-delivery-gate.turns")), HearingState: read(stdio.State(buildrun.HearingState))}
|
TurnState: read(stdio.State("mesh-delivery-gate.turns")), HearingState: read(stdio.State(buildrun.HearingState)),
|
||||||
|
AskState: read(stdio.State(check.AsksState))}
|
||||||
r.Runs = runsOf{read(stdio.State("mesh-delivery-indexer.modules"))} // the node-engine at the commit the mesh runs, as the checker reads it
|
r.Runs = runsOf{read(stdio.State("mesh-delivery-indexer.modules"))} // the node-engine at the commit the mesh runs, as the checker reads it
|
||||||
tries := stdio.State(check.TriesState)
|
tries := stdio.State(check.TriesState)
|
||||||
source := stdio.State("mesh-delivery-gate.source")
|
source := stdio.State("mesh-delivery-gate.source")
|
||||||
@@ -277,7 +296,20 @@ func main() {
|
|||||||
return err
|
return err
|
||||||
}}
|
}}
|
||||||
trying := Trying{Tries: store, Ender: store, Trunks: trunks}
|
trying := Trying{Tries: store, Ender: store, Trunks: trunks}
|
||||||
if err := stdio.Serve("", logged(append(Tools(r), TryTools(trying)...))); err != nil {
|
verdicts := stdio.State("verdicts")
|
||||||
|
written := stateStore{entry: func(key string) (json.RawMessage, uint64, error) {
|
||||||
|
reads.Add(1)
|
||||||
|
e, err := verdicts.Get(key)
|
||||||
|
if err != nil || e == nil {
|
||||||
|
return nil, 0, err
|
||||||
|
}
|
||||||
|
return e.Value, e.Revision, nil
|
||||||
|
}, update: func(key string, value []byte, revision uint64) error {
|
||||||
|
_, err := verdicts.Update(key, json.RawMessage(value), revision)
|
||||||
|
return err
|
||||||
|
}}
|
||||||
|
tools := append(append(Tools(r), TryTools(trying)...), RecoveryTools(r, written)...)
|
||||||
|
if err := stdio.Serve("", logged(tools)); err != nil {
|
||||||
fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err)
|
fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err)
|
||||||
os.Exit(1)
|
os.Exit(1)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -23,6 +23,14 @@ func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
|
|||||||
if !slices.Equal(names, want) {
|
if !slices.Equal(names, want) {
|
||||||
t.Errorf("tools = %v, want %v", names, want)
|
t.Errorf("tools = %v, want %v", names, want)
|
||||||
}
|
}
|
||||||
|
// The recovery tool beside them asks a lost check again; with nothing to write it refuses.
|
||||||
|
recovery := RecoveryTools(check.Reader{}, nil)
|
||||||
|
if len(recovery) != 1 || recovery[0].Name != check.AskAgainTool || recovery[0].Description == "" {
|
||||||
|
t.Errorf("recovery tools = %+v", recovery)
|
||||||
|
}
|
||||||
|
if _, err := recovery[0].Run(map[string]any{"group": "feat/x", "why": "lost"}); err == nil {
|
||||||
|
t.Errorf("%s answered with no state to write", check.AskAgainTool)
|
||||||
|
}
|
||||||
// With nothing to read from a tool says so and changes nothing.
|
// With nothing to read from a tool says so and changes nothing.
|
||||||
for _, tl := range Tools(check.Reader{}) {
|
for _, tl := range Tools(check.Reader{}) {
|
||||||
if tl.Name == check.StatusReadTool {
|
if tl.Name == check.StatusReadTool {
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
|
|
||||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/process"
|
||||||
|
|
||||||
"mesh-delivery-checker/internal/check"
|
"mesh-delivery-checker/internal/check"
|
||||||
)
|
)
|
||||||
@@ -130,3 +131,55 @@ func (m manifestReader) Manifest(ctx context.Context, digest string) (catalogue.
|
|||||||
func parseManifest(raw string) (catalogue.Manifest, error) {
|
func parseManifest(raw string) (catalogue.Manifest, error) {
|
||||||
return catalogue.ParseManifestForInputs([]byte(raw))
|
return catalogue.ParseManifestForInputs([]byte(raw))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// awaitedKV is a bucket that may appear after the start (process.AwaitBucket) as a check.KV: a call while it is not on the
|
||||||
|
// bus fails saying so.
|
||||||
|
type awaitedKV struct{ b *process.AwaitedBucket }
|
||||||
|
|
||||||
|
func (a awaitedKV) kv() (kvStore, error) {
|
||||||
|
kv := a.b.KV()
|
||||||
|
if kv == nil {
|
||||||
|
return kvStore{}, errors.New("the bucket is not on the bus yet")
|
||||||
|
}
|
||||||
|
return kvStore{kv}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a awaitedKV) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) {
|
||||||
|
s, err := a.kv()
|
||||||
|
if err != nil {
|
||||||
|
return nil, 0, false, err
|
||||||
|
}
|
||||||
|
return s.Get(ctx, key)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a awaitedKV) Create(ctx context.Context, key string, value []byte) (uint64, error) {
|
||||||
|
s, err := a.kv()
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return s.Create(ctx, key, value)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a awaitedKV) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) {
|
||||||
|
s, err := a.kv()
|
||||||
|
if err != nil {
|
||||||
|
return 0, err
|
||||||
|
}
|
||||||
|
return s.Update(ctx, key, value, revision)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a awaitedKV) Keys(ctx context.Context) ([]string, error) {
|
||||||
|
s, err := a.kv()
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return s.Keys(ctx)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (a awaitedKV) Delete(ctx context.Context, key string) error {
|
||||||
|
s, err := a.kv()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return s.Delete(ctx, key)
|
||||||
|
}
|
||||||
|
|||||||
@@ -89,6 +89,8 @@ type fakeBuildModules struct {
|
|||||||
// beside is, by workspace, the siblings the fetch placed in it as `<owner/repo> <commit>`, sorted: a digest echoes them
|
// beside is, by workspace, the siblings the fetch placed in it as `<owner/repo> <commit>`, sorted: a digest echoes them
|
||||||
// (1929/AC2; mesh-sdk buildstep.besideOf), or the level runs whole and keeps nothing.
|
// (1929/AC2; mesh-sdk buildstep.besideOf), or the level runs whole and keeps nothing.
|
||||||
beside map[string][]string
|
beside map[string][]string
|
||||||
|
// names are the nodes' names the last check ask carried, for the names check (ADR 0567 §12).
|
||||||
|
names []string
|
||||||
}
|
}
|
||||||
|
|
||||||
func (f *fakeBuildModules) count(tool string) int {
|
func (f *fakeBuildModules) count(tool string) int {
|
||||||
@@ -191,7 +193,14 @@ func (f *fakeBuildModules) serve(t *testing.T, module string, kept jetstream.Key
|
|||||||
}
|
}
|
||||||
p := &buildstep.Puller{Source: &buildstep.JetStreamAsks{JS: f.js, Language: language}, Language: language, Node: "n1", Put: put,
|
p := &buildstep.Puller{Source: &buildstep.JetStreamAsks{JS: f.js, Language: language}, Language: language, Node: "n1", Put: put,
|
||||||
Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond,
|
Capacity: func() int { return 2 }, PullWait: 200 * time.Millisecond,
|
||||||
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { return buildstep.RunAsk(ctx, steps, a) }}
|
Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
|
||||||
|
if a.Check != nil {
|
||||||
|
f.mu.Lock()
|
||||||
|
f.names = a.Check.NodeNames
|
||||||
|
f.mu.Unlock()
|
||||||
|
}
|
||||||
|
return buildstep.RunAsk(ctx, steps, a)
|
||||||
|
}}
|
||||||
go func() { _ = p.Pull(f.ctx) }()
|
go func() { _ = p.Pull(f.ctx) }()
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -334,6 +343,7 @@ func startBed(t *testing.T) *bed {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
b.setOnline(t, "n1", true)
|
b.setOnline(t, "n1", true)
|
||||||
|
b.setNode(t, "n1")
|
||||||
for _, module := range []string{"mesh-build-go", "mesh-build-node"} {
|
for _, module := range []string{"mesh-build-go", "mesh-build-node"} {
|
||||||
b.setHealth(t, "n1", module)
|
b.setHealth(t, "n1", module)
|
||||||
}
|
}
|
||||||
@@ -366,6 +376,20 @@ func (b *bed) setOnline(t *testing.T, node string, online bool) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// setNode writes the node's record in the node-engine's have, as the node-engine does, so the checker names it in each
|
||||||
|
// check ask (ADR 0567 §12); with no record the checker asks no check.
|
||||||
|
func (b *bed) setNode(t *testing.T, node string) {
|
||||||
|
t.Helper()
|
||||||
|
data, err := delivery.EncodeNodeRecord(delivery.NodeRecord{V: 1, Node: node})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
key, _ := delivery.HaveNodeKey(node)
|
||||||
|
if _, err := b.buckets[delivery.HaveBucket].Put(b.ctx, key, data); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (b *bed) setHealth(t *testing.T, node, module string) {
|
func (b *bed) setHealth(t *testing.T, node, module string) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
data, err := delivery.EncodeHave(delivery.Have{Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
data, err := delivery.EncodeHave(delivery.Have{Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
||||||
@@ -493,6 +517,12 @@ func TestCheckerJudgesACandidateAndOwesItsStatus(t *testing.T) { // N6/AC5, 1300
|
|||||||
if v.Composition == nil || v.Composition.State != delivery.CompositionNotRun || len(v.Machines) != 1 || v.Machines[0] != "n1" {
|
if v.Composition == nil || v.Composition.State != delivery.CompositionNotRun || len(v.Machines) != 1 || v.Machines[0] != "n1" {
|
||||||
t.Errorf("composition %+v machines %v", v.Composition, v.Machines)
|
t.Errorf("composition %+v machines %v", v.Composition, v.Machines)
|
||||||
}
|
}
|
||||||
|
b.fake.mu.Lock()
|
||||||
|
names := b.fake.names
|
||||||
|
b.fake.mu.Unlock()
|
||||||
|
if len(names) != 1 || names[0] != "n1" {
|
||||||
|
t.Errorf("the check ask named the nodes %v, want [n1] from the have's node record", names)
|
||||||
|
}
|
||||||
o := b.owed(t, 7, head1, "success")
|
o := b.owed(t, 7, head1, "success")
|
||||||
if o.Context != delivery.StatusContextCheck || !strings.HasPrefix(o.Text, "pass 111111111111 on aaaaaaaaaaaa") || !strings.Contains(o.Text, "unit 2 run") {
|
if o.Context != delivery.StatusContextCheck || !strings.HasPrefix(o.Text, "pass 111111111111 on aaaaaaaaaaaa") || !strings.Contains(o.Text, "unit 2 run") {
|
||||||
t.Errorf("owed = %+v", o)
|
t.Errorf("owed = %+v", o)
|
||||||
@@ -797,7 +827,8 @@ func TestTheCheckerSaysOwedReadableOnTheBus(t *testing.T) { // R2-1
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
if said.Module != "mesh-delivery-checker" || said.Node != "n1" || !said.Keeps("owed") || !said.Readable("owed") ||
|
if said.Module != "mesh-delivery-checker" || said.Node != "n1" || !said.Keeps("owed") || !said.Readable("owed") ||
|
||||||
!said.Readable("previews") || !said.Readable("verdicts") || !said.Readable("kept") || !said.Readable("tries") || said.Readable("hearing") || len(said.States) != 6 {
|
!said.Readable("previews") || !said.Readable("verdicts") || !said.Readable("kept") || !said.Readable("tries") || said.Readable("hearing") ||
|
||||||
t.Fatalf("the checker said %+v, want its six states with kept, owed, previews, tries and verdicts readable, tries by the delivery view (1929/AC4; 1887 A5b; build_usage)", said)
|
!said.Keeps("asks") || said.Readable("asks") || len(said.States) != 7 {
|
||||||
|
t.Fatalf("the checker said %+v, want its seven states, asks and hearing its own, with kept, owed, previews, tries and verdicts readable, tries by the delivery view (1929/AC4; 1887 A5b; build_usage)", said)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -69,6 +69,8 @@ var keptStates = []announce.State{
|
|||||||
// design limit 1); the checker alone writes them (ADR 0444 clause 5).
|
// design limit 1); the checker alone writes them (ADR 0444 clause 5).
|
||||||
// Whether DEAD_LETTERS is heard (issue 734): read by the checker's own status tool alone.
|
// Whether DEAD_LETTERS is heard (issue 734): read by the checker's own status tool alone.
|
||||||
{Name: "hearing"},
|
{Name: "hearing"},
|
||||||
|
// The asks each running check put (check.AsksState): read by the checker's own checker_ask_again alone.
|
||||||
|
{Name: "asks"},
|
||||||
{Name: "kept", Readable: true},
|
{Name: "kept", Readable: true},
|
||||||
{Name: "owed", Readable: true},
|
{Name: "owed", Readable: true},
|
||||||
{Name: "previews", Readable: true},
|
{Name: "previews", Readable: true},
|
||||||
@@ -144,9 +146,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
|||||||
hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, []string{delivery.AskGo, delivery.AskNode}, dead,
|
hearing := buildrun.NewDeadLetterHearing(ctx, buildrun.JetStreamDeadLetters{JS: js}, []string{delivery.AskGo, delivery.AskNode}, dead,
|
||||||
logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf))
|
logger.Printf, buildrun.SayHearing(ctx, hearingKV, logger.Printf))
|
||||||
defer hearing.Stop()
|
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
|
||||||
waiter := buildrun.LostAskWaiter{Inner: asks, Dead: dead, Hearing: hearing}
|
// (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}
|
||||||
|
asksKV := process.AwaitBucket(ctx, js, conn, check.AsksBucket, "checker_ask_again refuses every check, as it cannot tell whether its ask waits in a stream", logger.Printf)
|
||||||
ch := &check.Checker{
|
ch := &check.Checker{
|
||||||
|
Asks: awaitedKV{asksKV},
|
||||||
Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]},
|
Verdicts: kvStore{buckets[delivery.CheckerVerdictsBucket]},
|
||||||
Owed: kvStore{buckets[delivery.CheckerOwedBucket]},
|
Owed: kvStore{buckets[delivery.CheckerOwedBucket]},
|
||||||
Trunks: trunks,
|
Trunks: trunks,
|
||||||
@@ -331,6 +338,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
}()
|
}()
|
||||||
|
// The checker's own verdicts, updates only: one checker_ask_again wrote asks its check again (Checker.VerdictWritten).
|
||||||
|
vw, err := buckets[delivery.CheckerVerdictsBucket].WatchAll(ctx, jetstream.UpdatesOnly())
|
||||||
|
if err != nil {
|
||||||
|
return fmt.Errorf("the checker's verdicts cannot be watched, so no check can be asked again by hand: %w", err)
|
||||||
|
}
|
||||||
|
go func() {
|
||||||
|
errs <- follow(ctx, vw, func(e jetstream.KeyValueEntry) {
|
||||||
|
if opOf(e) {
|
||||||
|
ch.VerdictWritten(ctx, e.Key(), e.Value())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}()
|
||||||
logger.Printf("started on %s", cred.Node)
|
logger.Printf("started on %s", cred.Node)
|
||||||
go func() { errs <- follow(ctx, w, each) }()
|
go func() { errs <- follow(ctx, w, each) }()
|
||||||
select {
|
select {
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ module mesh-delivery-checker
|
|||||||
go 1.26.0
|
go 1.26.0
|
||||||
|
|
||||||
require (
|
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.20261028193502-f91340fe82ef
|
||||||
github.com/nats-io/nats-server/v2 v2.11.17
|
github.com/nats-io/nats-server/v2 v2.11.17
|
||||||
github.com/nats-io/nats.go v1.54.0
|
github.com/nats-io/nats.go v1.54.0
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -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.20261028193502-f91340fe82ef h1:ggwHXCvAOhBCKAY+ErehKEhu/IQTS4poJBRDokGYnDc=
|
||||||
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.20261028193502-f91340fe82ef/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
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=
|
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||||
|
|||||||
@@ -0,0 +1,198 @@
|
|||||||
|
package check
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/json"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"strings"
|
||||||
|
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
|
)
|
||||||
|
|
||||||
|
// **A check is asked again by hand** (the lost check of 2026-10-28; tools-are-how-a-module-is-spoken-to: every recovery
|
||||||
|
// has a tool). checker_ask_again ends a candidate's running verdict, whose wait judged nothing, with AskedAgainWords and
|
||||||
|
// the why; the checker hears it on its watch of its verdicts (Checker.VerdictWritten), ends the check of those heads
|
||||||
|
// running in it, and claims and asks it anew, a whole check (IsEnded). It is refused while a step of the candidate's pull
|
||||||
|
// requests can still run (checker_status's running_steps): that step's end judges it; and while an ask the check put is
|
||||||
|
// still in its build seat's stream, taken by no node (it has no step key yet): asking again then would run the check twice
|
||||||
|
// (issue 1791's kind). The checker keeps the asks each running check put in its state asks (AsksState, by verdict key).
|
||||||
|
// An ask is not put again under its own id instead: the stream de-duplicates an id within its window, so the copy waiting
|
||||||
|
// stays and the new one is dropped, and past the window both would run; nothing takes a put ask back (buildrun.Asker).
|
||||||
|
|
||||||
|
// AsksState is the checker's state of the asks each running check put, by verdict key (CheckAsks); AsksBucket its bucket.
|
||||||
|
const (
|
||||||
|
AsksState = "asks"
|
||||||
|
AsksBucket = delivery.CheckerModule + "_" + AsksState
|
||||||
|
)
|
||||||
|
|
||||||
|
// CheckAsks are the asks a check put, in order, as the state asks keeps them while it runs; its end deletes the key.
|
||||||
|
type CheckAsks struct {
|
||||||
|
Asks []string `json:"asks"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// AsksKV is the state asks: a KV whose key a check's end deletes, so the state holds only the checks that run.
|
||||||
|
type AsksKV interface {
|
||||||
|
KV
|
||||||
|
Delete(ctx context.Context, key string) error
|
||||||
|
}
|
||||||
|
|
||||||
|
// forgetAsks deletes the asks a check put at its end; one that cannot be deleted is said, and the next check of the same
|
||||||
|
// heads writes the key anew.
|
||||||
|
func (ch *Checker) forgetAsks(ctx context.Context, key string) {
|
||||||
|
if err := ch.Asks.Delete(ctx, key); err != nil {
|
||||||
|
ch.logf("%s: the asks it put could not be forgotten at its end: %v", key, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// recordAsks keeps the asks a check put under its verdict key; nil Asks keeps nothing. A record that cannot be written
|
||||||
|
// is said: checker_ask_again then refuses the check, as it cannot tell whether its ask waits in a stream.
|
||||||
|
func (ch *Checker) recordAsks(ctx context.Context, key string, asks []string) {
|
||||||
|
if ch.Asks == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
data, err := json.Marshal(CheckAsks{Asks: asks})
|
||||||
|
if err == nil {
|
||||||
|
_, err = putOnce(ctx, ch.Asks, key, data, func(cur []byte) bool { return string(cur) == string(data) })
|
||||||
|
}
|
||||||
|
if err != nil {
|
||||||
|
ch.logf("%s: the asks it put cannot be kept (%v), so %s refuses it", key, err, AskAgainTool)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// AskAgainTool asks a candidate's check again by hand.
|
||||||
|
const AskAgainTool = "checker_ask_again"
|
||||||
|
|
||||||
|
// AskedAgainWords open the words of a verdict AskAgainTool wrote; the why follows.
|
||||||
|
const AskedAgainWords = EndedWords + "asked again by hand (" + AskAgainTool + "): "
|
||||||
|
|
||||||
|
// errAskedAgain is the cause a running check is ended with when its verdict was asked again by hand.
|
||||||
|
var errAskedAgain = errors.New("its verdict was asked again by hand (" + AskAgainTool + ")")
|
||||||
|
|
||||||
|
// IsAskedAgain says whether a verdict was written by AskAgainTool.
|
||||||
|
func IsAskedAgain(v delivery.Verdict) bool {
|
||||||
|
return IsEnded(v) && strings.HasPrefix(v.Failed.Words, AskedAgainWords)
|
||||||
|
}
|
||||||
|
|
||||||
|
// VerdictEntry is the verdicts state as the ask-again tool writes it: a key's value and revision (nil and 0 for none),
|
||||||
|
// and a compare-and-set.
|
||||||
|
type VerdictEntry interface {
|
||||||
|
Entry(key string) (json.RawMessage, uint64, error)
|
||||||
|
Update(key string, value []byte, revision uint64) error
|
||||||
|
}
|
||||||
|
|
||||||
|
// AskedAgain is the answer of AskAgainTool.
|
||||||
|
type AskedAgain struct {
|
||||||
|
Key string `json:"key"`
|
||||||
|
Was string `json:"was"`
|
||||||
|
Said string `json:"said"`
|
||||||
|
Note string `json:"note"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// AskAgain ends the running verdict of the group's candidate as asked again by hand, saying why. It refuses a group with
|
||||||
|
// no candidate, a verdict that is not running (a pass or a failure stands until a push or a trunk move), one a step of
|
||||||
|
// whose pull requests can still run, and one whose asks are not known or one of which no node took yet.
|
||||||
|
func (r Reader) AskAgain(w VerdictEntry, group, why string) (AskedAgain, error) {
|
||||||
|
why = strings.TrimSpace(why)
|
||||||
|
if group == "" || why == "" {
|
||||||
|
return AskedAgain{}, fmt.Errorf("give the group and why it is asked again")
|
||||||
|
}
|
||||||
|
if w == nil {
|
||||||
|
return AskedAgain{}, fmt.Errorf("the verdicts state cannot be written here")
|
||||||
|
}
|
||||||
|
c, ok := r.candidate(group)
|
||||||
|
if !ok {
|
||||||
|
return AskedAgain{}, fmt.Errorf("no candidate is held for the group %q", group)
|
||||||
|
}
|
||||||
|
key, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
|
||||||
|
if err != nil {
|
||||||
|
return AskedAgain{}, err
|
||||||
|
}
|
||||||
|
raw, rev, err := w.Entry(key)
|
||||||
|
if err != nil {
|
||||||
|
return AskedAgain{}, fmt.Errorf("%s cannot be read: %w", key, err)
|
||||||
|
}
|
||||||
|
if len(raw) == 0 {
|
||||||
|
return AskedAgain{}, fmt.Errorf("%s holds no verdict: the checker claims its check when it looks at the candidate (%s)", key, StatusReadTool)
|
||||||
|
}
|
||||||
|
v, err := delivery.DecodeVerdict(raw)
|
||||||
|
if err != nil {
|
||||||
|
return AskedAgain{}, fmt.Errorf("%s holds what is not a verdict: %w", key, err)
|
||||||
|
}
|
||||||
|
if v.State != delivery.VerdictRunning {
|
||||||
|
return AskedAgain{}, fmt.Errorf("%s is %s, not running: a verdict stands until a push or a trunk move asks it again", key, v.State)
|
||||||
|
}
|
||||||
|
steps := r.readSteps()
|
||||||
|
for _, rs := range steps.running {
|
||||||
|
if asksOf(c, rs.What) {
|
||||||
|
return AskedAgain{}, fmt.Errorf("a step of it runs: %s %s on %s since %s (%s); its end judges it", rs.Module, rs.ID, rs.Node, rs.Since, rs.What)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if r.AskState.Get == nil {
|
||||||
|
return AskedAgain{}, fmt.Errorf("the checker's asks are not readable here, so whether its ask waits in a build seat's stream cannot be told")
|
||||||
|
}
|
||||||
|
kept, err := r.AskState.Get(key)
|
||||||
|
var put CheckAsks
|
||||||
|
if err == nil && len(kept) > 0 {
|
||||||
|
err = json.Unmarshal(kept, &put)
|
||||||
|
}
|
||||||
|
if err != nil || len(put.Asks) == 0 {
|
||||||
|
return AskedAgain{}, fmt.Errorf("the checker names no ask of %s (%v), so whether one waits in a build seat's stream cannot be told; asking again could run it twice", key, err)
|
||||||
|
}
|
||||||
|
taken := map[string]bool{}
|
||||||
|
for _, s := range steps.all {
|
||||||
|
taken[s.step.ID] = true
|
||||||
|
}
|
||||||
|
for _, id := range put.Asks {
|
||||||
|
if !taken[id] {
|
||||||
|
return AskedAgain{}, fmt.Errorf("its ask %s waits in its build seat's stream, taken by no node yet: asking again would run it twice; "+
|
||||||
|
"the seats' backlog is in mesh-delivery-view.build_usage", id)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
words := AskedAgainWords + why
|
||||||
|
ended := delivery.Verdict{Group: v.Group, Heads: v.Heads, Trunk: v.Trunk, State: delivery.VerdictFail,
|
||||||
|
Failed: delivery.FailedFirst{Machine: "checker", Words: words}}
|
||||||
|
data, err := delivery.EncodeVerdict(ended)
|
||||||
|
if err != nil {
|
||||||
|
return AskedAgain{}, err
|
||||||
|
}
|
||||||
|
if err := w.Update(key, data, rev); err != nil {
|
||||||
|
return AskedAgain{}, fmt.Errorf("%s changed while it was asked again, so it was not; read it again: %w", key, err)
|
||||||
|
}
|
||||||
|
return AskedAgain{Key: key, Was: v.State, Said: words,
|
||||||
|
Note: "the checker hears it on its watch of its verdicts, ends the wait that judged nothing and asks the check anew, whole; " +
|
||||||
|
VerdictsReadTool + " shows it running again"}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// VerdictWritten takes a change of the checker's verdicts in (its watch, updates only): a verdict AskAgainTool wrote ends
|
||||||
|
// the check of those heads running in this process, which then considers them again; with none running, the candidate
|
||||||
|
// that holds them is considered at once. Any other verdict is the checker's own, and left.
|
||||||
|
func (ch *Checker) VerdictWritten(ctx context.Context, key string, value []byte) {
|
||||||
|
v, err := delivery.DecodeVerdict(value)
|
||||||
|
if err != nil || !IsAskedAgain(v) {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
ch.init()
|
||||||
|
ch.mu.Lock()
|
||||||
|
if l, ok := ch.live[key]; ok {
|
||||||
|
l.end(errAskedAgain)
|
||||||
|
ch.mu.Unlock()
|
||||||
|
ch.logf("%s: %s; the check running for it is ended and asked again", key, v.Failed.Words)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
var group string
|
||||||
|
var cand delivery.Candidate
|
||||||
|
for k, c := range ch.known {
|
||||||
|
if vk, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash); err == nil && vk == key {
|
||||||
|
group, cand = k, c
|
||||||
|
break
|
||||||
|
}
|
||||||
|
}
|
||||||
|
ch.mu.Unlock()
|
||||||
|
if group == "" {
|
||||||
|
ch.logf("%s: %s, but no candidate holds these heads, so nothing is asked", key, v.Failed.Words)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
ch.logf("%s: %s; asked again", key, v.Failed.Words)
|
||||||
|
ch.consider(ctx, group, cand)
|
||||||
|
}
|
||||||
@@ -79,6 +79,8 @@ type Checker struct {
|
|||||||
}
|
}
|
||||||
// Bound is how long one check may run; none when 0 (DefaultBound).
|
// Bound is how long one check may run; none when 0 (DefaultBound).
|
||||||
Bound time.Duration
|
Bound time.Duration
|
||||||
|
// Asks keeps the asks each running check put, by verdict key (AsksState), for checker_ask_again; nil keeps none.
|
||||||
|
Asks AsksKV
|
||||||
// Link is the base of the address a status points at; the verdict key follows.
|
// Link is the base of the address a status points at; the verdict key follows.
|
||||||
Link string
|
Link string
|
||||||
// Announce says a status was written (EventOwedWritten); nil says nothing.
|
// Announce says a status was written (EventOwedWritten); nil says nothing.
|
||||||
@@ -328,7 +330,11 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
|
|||||||
delete(ch.again, vkey)
|
delete(ch.again, vkey)
|
||||||
delete(ch.live, vkey)
|
delete(ch.live, vkey)
|
||||||
ch.mu.Unlock()
|
ch.mu.Unlock()
|
||||||
|
again := errors.Is(context.Cause(run), errAskedAgain)
|
||||||
end(nil)
|
end(nil)
|
||||||
|
if again && ctx.Err() == nil {
|
||||||
|
ch.consider(ctx, key, ch.candidate(key, c)) // asked again by hand (AskAgainTool): claimed anew
|
||||||
|
}
|
||||||
}()
|
}()
|
||||||
for {
|
for {
|
||||||
// Every check is asked at once: the build seats' streams hold the backlog, a new head's check and a check after a
|
// Every check is asked at once: the build seats' streams hold the backlog, a new head's check and a check after a
|
||||||
@@ -711,10 +717,20 @@ func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
began := ch.now()
|
began := ch.now()
|
||||||
|
if cs.Watch == nil && ch.Asks != nil {
|
||||||
|
cs.Watch = func(_ string, step *delivery.Step) {
|
||||||
|
if step == nil { // put: kept before its wait starts
|
||||||
|
ch.recordAsks(ctx, cs.Key, cs.Asks())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
defer ch.forgetAsks(ctx, cs.Key)
|
||||||
|
}
|
||||||
ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
|
ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
|
||||||
out := ch.judge(run, cs)
|
out := ch.judge(run, cs)
|
||||||
// An ask the build seat gave up on judged nothing (issue 734): the check is asked again under the same claim, its status
|
// An ask the build seat gave up on, or whose step the build module's next process ended because it was replaced under
|
||||||
// pending, and after LostAskTries it fails as not heard, judged again when the nodes change.
|
// it, judged nothing (issue 734, IsLostAsk): the check is asked again under the same claim, its status pending, and after
|
||||||
|
// LostAskTries it fails as not heard, judged again when the nodes change. A step that ended while its watch was deaf is
|
||||||
|
// read at the next nudge (HeardEnds) and judged.
|
||||||
for try := 1; IsLostAsk(out.Verdict) && run.Err() == nil; try++ {
|
for try := 1; IsLostAsk(out.Verdict) && run.Err() == nil; try++ {
|
||||||
if try >= LostAskTries {
|
if try >= LostAskTries {
|
||||||
out.Verdict.Failed.Words = strings.TrimSpace(out.Verdict.Failed.Words + "\nasked " + fmt.Sprint(LostAskTries) +
|
out.Verdict.Failed.Words = strings.TrimSpace(out.Verdict.Failed.Words + "\nasked " + fmt.Sprint(LostAskTries) +
|
||||||
@@ -724,8 +740,8 @@ func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) {
|
|||||||
ch.logf("%s: %s; asked again (%d of %d)", cs.Key, out.Verdict.Failed.Words, try+1, LostAskTries)
|
ch.logf("%s: %s; asked again (%d of %d)", cs.Key, out.Verdict.Failed.Words, try+1, LostAskTries)
|
||||||
out = ch.judge(run, cs)
|
out = ch.judge(run, cs)
|
||||||
}
|
}
|
||||||
if ctx.Err() != nil {
|
if ctx.Err() != nil || errors.Is(context.Cause(run), errAskedAgain) {
|
||||||
return
|
return // the checker stops, or the verdict was asked again by hand, which wrote it: its next check claims it anew
|
||||||
}
|
}
|
||||||
var why superseded
|
var why superseded
|
||||||
if errors.As(context.Cause(run), &why) {
|
if errors.As(context.Cause(run), &why) {
|
||||||
|
|||||||
@@ -60,3 +60,11 @@ func (m *memKV) Keys(context.Context) ([]string, error) {
|
|||||||
sort.Strings(keys)
|
sort.Strings(keys)
|
||||||
return keys, nil
|
return keys, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (m *memKV) Delete(_ context.Context, key string) error {
|
||||||
|
m.mu.Lock()
|
||||||
|
defer m.mu.Unlock()
|
||||||
|
delete(m.vals, key)
|
||||||
|
delete(m.revs, key)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package check
|
|||||||
import (
|
import (
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||||
)
|
)
|
||||||
@@ -19,9 +20,13 @@ import (
|
|||||||
// judged again when the nodes change.
|
// judged again when the nodes change.
|
||||||
const LostAskTries = 3
|
const LostAskTries = 3
|
||||||
|
|
||||||
// IsLostAsk says whether a verdict failed because the build seat gave its ask up (buildrun.LostWords).
|
// IsLostAsk says whether a verdict failed because its ask judged nothing: the build seat gave it up (buildrun.LostWords), or
|
||||||
|
// the build module that ran its step was replaced under it, and its next process ended the step it left running
|
||||||
|
// (buildstep.RestartReason; a delivery of the build module, as on 2026-10-28). Neither is the change's failure: the check
|
||||||
|
// is asked again, as for a given-up ask.
|
||||||
func IsLostAsk(v delivery.Verdict) bool {
|
func IsLostAsk(v delivery.Verdict) bool {
|
||||||
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, buildrun.LostWords)
|
return v.State == delivery.VerdictFail && (strings.Contains(v.Failed.Words, buildrun.LostWords) ||
|
||||||
|
strings.Contains(v.Failed.Words, buildstep.RestartReason))
|
||||||
}
|
}
|
||||||
|
|
||||||
// IsNotTaken says whether a verdict failed because no build node took its ask within the bound (buildrun.NotTakenWords).
|
// IsNotTaken says whether a verdict failed because no build node took its ask within the bound (buildrun.NotTakenWords).
|
||||||
|
|||||||
@@ -8,6 +8,7 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"git.novox.be/novox/mesh-sdk/go/buildstep"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||||
)
|
)
|
||||||
@@ -141,30 +142,206 @@ func TestAnAskGivenUpOrNeverTakenEndsItsTryAsErrorAndItsHeadIsAskedAnew(t *testi
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// Issue 734: a pull request's check whose ask the build seat gave up on is asked again under its claim, its status
|
// troubled is the build seat as a check meets it when its step's end does not reach its wait: the first `restarts` asks'
|
||||||
// pending, and judged; one given up every time fails as not heard and is judged again when the nodes change.
|
// steps are ended by the build module's next process (buildstep.RestartReason), and the first `deaf` asks' watches hear
|
||||||
func TestACheckWhoseAskWasGivenUpIsAskedAgain(t *testing.T) {
|
// nothing though their steps end (the lost check of 2026-10-28). Its steps state is read through AskSteps and Steps.
|
||||||
|
type troubled struct {
|
||||||
|
*buildrun.FakeAsks
|
||||||
|
answer func(delivery.StepAsk) delivery.Step
|
||||||
|
|
||||||
|
mu sync.Mutex
|
||||||
|
restarts, deaf int
|
||||||
|
deafIDs map[string]bool
|
||||||
|
steps map[string]delivery.Step
|
||||||
|
}
|
||||||
|
|
||||||
|
func (tr *troubled) PublishAsk(ctx context.Context, subject, id string, data []byte) error {
|
||||||
|
if err := tr.FakeAsks.PublishAsk(ctx, subject, id, data); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
a, err := delivery.DecodeStepAsk(data)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
step := tr.answer(a)
|
||||||
|
step.What = buildstep.AskWhat(a)
|
||||||
|
tr.mu.Lock()
|
||||||
|
if tr.restarts > 0 {
|
||||||
|
tr.restarts--
|
||||||
|
step.State, step.Reason, step.Result = delivery.StepFailed, buildstep.RestartReason, nil
|
||||||
|
}
|
||||||
|
deaf := tr.deaf > 0
|
||||||
|
if deaf {
|
||||||
|
tr.deaf--
|
||||||
|
tr.deafIDs[id] = true
|
||||||
|
}
|
||||||
|
tr.steps[id] = step
|
||||||
|
tr.mu.Unlock()
|
||||||
|
if !deaf {
|
||||||
|
tr.Write(step)
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (tr *troubled) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
|
||||||
|
tr.mu.Lock()
|
||||||
|
deaf := tr.deafIDs[askID]
|
||||||
|
tr.mu.Unlock()
|
||||||
|
if deaf {
|
||||||
|
return make(chan delivery.Step), nil // open, and silent for good
|
||||||
|
}
|
||||||
|
return tr.FakeAsks.WatchAsk(ctx, module, askID)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (tr *troubled) AskSteps(_ context.Context, _, askID string) ([]delivery.Step, error) {
|
||||||
|
tr.mu.Lock()
|
||||||
|
defer tr.mu.Unlock()
|
||||||
|
if s, ok := tr.steps[askID]; ok {
|
||||||
|
return []delivery.Step{s}, nil
|
||||||
|
}
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Steps is the steps state as the read tools read it, with running in place of every step the hand has not let end.
|
||||||
|
func (tr *troubled) Steps(running bool) StateRead {
|
||||||
|
tr.mu.Lock()
|
||||||
|
defer tr.mu.Unlock()
|
||||||
|
vals := map[string]json.RawMessage{}
|
||||||
|
for id, s := range tr.steps {
|
||||||
|
if running {
|
||||||
|
s.State, s.Ended, s.Started = delivery.StepRunning, time.Time{}, time.Now()
|
||||||
|
}
|
||||||
|
raw, _ := delivery.EncodeStep(s)
|
||||||
|
vals[s.Node+"."+id] = raw
|
||||||
|
}
|
||||||
|
return StateRead{Keys: func() ([]string, error) {
|
||||||
|
var out []string
|
||||||
|
for k := range vals {
|
||||||
|
out = append(out, k)
|
||||||
|
}
|
||||||
|
return out, nil
|
||||||
|
}, Get: func(k string) (json.RawMessage, error) { return vals[k], nil }}
|
||||||
|
}
|
||||||
|
|
||||||
|
// kvRead is a bucket in memory as the read tools read it, live.
|
||||||
|
func kvRead(kv *memKV) StateRead {
|
||||||
|
return StateRead{Keys: func() ([]string, error) { return kv.Keys(context.Background()) },
|
||||||
|
Get: func(k string) (json.RawMessage, error) {
|
||||||
|
v, _, _, err := kv.Get(context.Background(), k)
|
||||||
|
return v, err
|
||||||
|
}}
|
||||||
|
}
|
||||||
|
|
||||||
|
// asVerdicts is the rig's verdicts as the ask-again tool reads and writes them.
|
||||||
|
type asVerdicts struct{ kv *memKV }
|
||||||
|
|
||||||
|
func (a asVerdicts) Entry(key string) (json.RawMessage, uint64, error) {
|
||||||
|
v, rev, _, err := a.kv.Get(context.Background(), key)
|
||||||
|
return v, rev, err
|
||||||
|
}
|
||||||
|
func (a asVerdicts) Update(key string, value []byte, revision uint64) error {
|
||||||
|
_, err := a.kv.Update(context.Background(), key, value, revision)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Issue 734 and the lost check of 2026-10-28: a check whose ask judged nothing is asked again, and one whose step ended
|
||||||
|
// unheard is judged. Given up by the build seat once, or its step ended by the build module's next process after a
|
||||||
|
// delivery replaced it, the check is asked again under its claim and passes; given up every time it fails loudly as not
|
||||||
|
// heard, judged again when the nodes change. A step whose watch went deaf is read at the next step end of its build module
|
||||||
|
// (a nudge) and judged with no new ask; one no event reaches is named lost by checker_status, refused by checker_ask_again
|
||||||
|
// while its step still runs, and asked again by it once its step ended.
|
||||||
|
func TestACheckWhoseAskJudgedNothingIsAskedAgainOrRead(t *testing.T) {
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
for _, row := range []struct {
|
for _, row := range []struct {
|
||||||
lose int
|
name string
|
||||||
state string
|
lose, restarts, deaf int
|
||||||
asks int
|
how string // "nudge": a step of the build module ends; "hand": checker_ask_again
|
||||||
|
state string
|
||||||
|
asks int
|
||||||
}{
|
}{
|
||||||
{1, delivery.VerdictPass, 2},
|
{name: "given up once", lose: 1, state: delivery.VerdictPass, asks: 2},
|
||||||
{LostAskTries, delivery.VerdictFail, LostAskTries},
|
{name: "given up every time", lose: LostAskTries, state: delivery.VerdictFail, asks: LostAskTries},
|
||||||
|
{name: "its build module replaced under it", restarts: 1, state: delivery.VerdictPass, asks: 2},
|
||||||
|
{name: "replaced under every ask", restarts: LostAskTries, state: delivery.VerdictFail, asks: LostAskTries},
|
||||||
|
{name: "its watch deaf, a step of its build module ends", deaf: 1, how: "nudge", state: delivery.VerdictPass, asks: 1},
|
||||||
|
{name: "its watch deaf, asked again by hand", deaf: 1, how: "hand", state: delivery.VerdictPass, asks: 2},
|
||||||
|
{name: "its watch deaf, asked again by hand after its build module replaced it", deaf: 1, restarts: 1, how: "hand", state: delivery.VerdictPass, asks: 2},
|
||||||
} {
|
} {
|
||||||
b := newBench(t)
|
t.Run(row.name, func(t *testing.T) {
|
||||||
l := losing(b, row.lose)
|
b := newBench(t)
|
||||||
r := newRig(t, b.ex)
|
var asked func() int
|
||||||
cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
|
var heard *buildrun.HeardEnds
|
||||||
r.ch.Candidate(ctx, "feat%2Fx", cand)
|
var tr *troubled
|
||||||
r.ch.Wait()
|
if row.lose > 0 {
|
||||||
v := r.verdict(t, keyOf(t, "feat/x", cand))
|
l := losing(b, row.lose)
|
||||||
if v.State != row.state || len(l.Asks()) != row.asks {
|
asked = func() int { return len(l.Asks()) }
|
||||||
t.Errorf("lose %d: verdict %s (%s) after %d asks, want %s after %d", row.lose, v.State, v.Failed.Words, len(l.Asks()), row.state, row.asks)
|
} else {
|
||||||
}
|
tr = &troubled{FakeAsks: &buildrun.FakeAsks{}, answer: b.answer, restarts: row.restarts, deaf: row.deaf,
|
||||||
if row.state == delivery.VerdictFail && (!IsLostAsk(v) || !IsNodesFailure(v)) {
|
deafIDs: map[string]bool{}, steps: map[string]delivery.Step{}}
|
||||||
t.Errorf("a check given up every time = %q, want lost and judged again when the nodes change", v.Failed.Words)
|
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()) }
|
||||||
|
}
|
||||||
|
r := newRig(t, b.ex)
|
||||||
|
asksKept := newMemKV()
|
||||||
|
r.ch.Asks = asksKept
|
||||||
|
cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
|
||||||
|
key := keyOf(t, "feat/x", cand)
|
||||||
|
r.ch.Candidate(ctx, "feat%2Fx", cand)
|
||||||
|
if row.deaf > 0 {
|
||||||
|
for deadline := time.Now().Add(2 * time.Second); heard.Waiting() == 0 && time.Now().Before(deadline); {
|
||||||
|
time.Sleep(2 * time.Millisecond)
|
||||||
|
}
|
||||||
|
if v := r.verdict(t, key); v.State != delivery.VerdictRunning {
|
||||||
|
t.Fatalf("a check whose step ended unheard = %s, want running until it is read", v.State)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
switch row.how {
|
||||||
|
case "nudge":
|
||||||
|
heard.Nudge(buildrun.NodeBuildModule) // another build module's step: not this one's to read
|
||||||
|
heard.Nudge(buildrun.GoBuildModule)
|
||||||
|
case "hand":
|
||||||
|
cands := newMemKV()
|
||||||
|
ckey, _ := delivery.CandidatesKey("feat/x")
|
||||||
|
_, _ = cands.Create(ctx, ckey, cand)
|
||||||
|
none := StateRead{Keys: func() ([]string, error) { return nil, nil }, Get: func(string) (json.RawMessage, error) { return nil, nil }}
|
||||||
|
rd := Reader{CandidateState: kvRead(cands), VerdictState: kvRead(r.verdicts), AskState: kvRead(asksKept),
|
||||||
|
StepStates: map[string]StateRead{buildrun.GoBuildModule: none}} // no node took its ask yet
|
||||||
|
if _, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "its watch went deaf"); err == nil || !strings.Contains(err.Error(), "waits in its build seat's stream") {
|
||||||
|
t.Errorf("asked again while its ask waits in the stream (issue 1791's kind): %v", err)
|
||||||
|
}
|
||||||
|
rd.StepStates[buildrun.GoBuildModule] = tr.Steps(true)
|
||||||
|
if _, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "its watch went deaf"); err == nil || !strings.Contains(err.Error(), "a step of it runs") {
|
||||||
|
t.Errorf("asked again while its step runs: %v", err)
|
||||||
|
}
|
||||||
|
rd.StepStates[buildrun.GoBuildModule] = tr.Steps(false)
|
||||||
|
if st, _ := rd.Status(); !strings.Contains(st.Lost["feat/x"], AskAgainTool) {
|
||||||
|
t.Errorf("status lost = %v, want feat/x named with %s", st.Lost, AskAgainTool)
|
||||||
|
}
|
||||||
|
got, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "its watch went deaf")
|
||||||
|
if err != nil || got.Key != key || !strings.HasPrefix(got.Said, AskedAgainWords) {
|
||||||
|
t.Fatalf("asked again = %+v, %v", got, err)
|
||||||
|
}
|
||||||
|
raw, _, _, _ := r.verdicts.Get(ctx, key)
|
||||||
|
r.ch.VerdictWritten(ctx, key, raw) // the checker's watch of its verdicts hands it over
|
||||||
|
}
|
||||||
|
r.ch.Wait()
|
||||||
|
v := r.verdict(t, key)
|
||||||
|
if v.State != row.state || asked() != row.asks {
|
||||||
|
t.Errorf("verdict %s (%s) after %d asks, want %s after %d", v.State, v.Failed.Words, asked(), row.state, row.asks)
|
||||||
|
}
|
||||||
|
if _, _, kept, _ := asksKept.Get(ctx, key); kept {
|
||||||
|
t.Errorf("the asks of the ended check are still kept under %s", key)
|
||||||
|
}
|
||||||
|
if row.state == delivery.VerdictFail && (!IsLostAsk(v) || !IsNodesFailure(v)) {
|
||||||
|
t.Errorf("a check that judged nothing every time = %q, want lost and judged again when the nodes change", v.Failed.Words)
|
||||||
|
}
|
||||||
|
if row.how == "hand" {
|
||||||
|
rd := Reader{VerdictState: kvRead(r.verdicts), CandidateState: kvRead(newMemKV())}
|
||||||
|
if _, err := rd.AskAgain(asVerdicts{r.verdicts}, "feat/x", "again"); err == nil {
|
||||||
|
t.Error("a judged check was asked again")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -47,7 +47,9 @@ type Reader struct {
|
|||||||
Runs interface{ CommitOf(module string) string }
|
Runs interface{ CommitOf(module string) string }
|
||||||
// HearingState is the checker's state hearing: whether it hears DEAD_LETTERS (issue 734).
|
// HearingState is the checker's state hearing: whether it hears DEAD_LETTERS (issue 734).
|
||||||
HearingState StateRead
|
HearingState StateRead
|
||||||
Now func() time.Time
|
// AskState is the checker's state asks: the asks each running check put (AsksState), which checker_ask_again reads.
|
||||||
|
AskState StateRead
|
||||||
|
Now func() time.Time
|
||||||
}
|
}
|
||||||
|
|
||||||
// VerdictRead is a verdict as stored with the full report of the status of each pull request it judged (Report), when the
|
// VerdictRead is a verdict as stored with the full report of the status of each pull request it judged (Report), when the
|
||||||
@@ -306,6 +308,8 @@ type RunningStep struct {
|
|||||||
ID string `json:"id"`
|
ID string `json:"id"`
|
||||||
Tool string `json:"tool"`
|
Tool string `json:"tool"`
|
||||||
Since string `json:"since,omitempty"`
|
Since string `json:"since,omitempty"`
|
||||||
|
// What says what the step is (buildstep.AskWhat): an ask's kind, repository@head and levels.
|
||||||
|
What string `json:"what,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// LeftStep is a key a build module left running that no step behind it runs any more, and why.
|
// LeftStep is a key a build module left running that no step behind it runs any more, and why.
|
||||||
@@ -331,6 +335,9 @@ type Status struct {
|
|||||||
// CannotBeJudged are the candidates no pull request of which can be judged, by group, with why: the gate's source.main
|
// CannotBeJudged are the candidates no pull request of which can be judged, by group, with why: the gate's source.main
|
||||||
// names no trunk head of their repositories (issue 1422). They are judged once it does; until then their status is error.
|
// names no trunk head of their repositories (issue 1422). They are judged once it does; until then their status is error.
|
||||||
CannotBeJudged map[string]string `json:"cannot_be_judged"`
|
CannotBeJudged map[string]string `json:"cannot_be_judged"`
|
||||||
|
// Lost are the candidates whose verdict is running while no step of theirs can still run and their newest step ended, by
|
||||||
|
// group, with that step: nothing will judge them (the lost check of 2026-10-28); checker_ask_again asks one again.
|
||||||
|
Lost map[string]string `json:"lost"`
|
||||||
// DeadLetters says whether the checker hears the asks the build seats give up on (issue 734): deaf, a try or check
|
// DeadLetters says whether the checker hears the asks the build seats give up on (issue 734): deaf, a try or check
|
||||||
// whose ask is given up waits until its bound.
|
// whose ask is given up waits until its bound.
|
||||||
DeadLetters DeadLettersHeard `json:"dead_letters"`
|
DeadLetters DeadLettersHeard `json:"dead_letters"`
|
||||||
@@ -394,63 +401,11 @@ func heldBy(t *Turns, c delivery.Candidate, heads map[string]string) string {
|
|||||||
|
|
||||||
// Status reads the build modules' steps and the candidates against their verdicts.
|
// Status reads the build modules' steps and the candidates against their verdicts.
|
||||||
func (r Reader) Status() (Status, error) {
|
func (r Reader) Status() (Status, error) {
|
||||||
st := Status{RunningByNode: map[string]int{}}
|
st := Status{RunningByNode: map[string]int{}, Lost: map[string]string{}}
|
||||||
mods := make([]string, 0, len(r.StepStates))
|
read := r.readSteps()
|
||||||
for m := range r.StepStates {
|
st.Running, st.LeftRunning, st.Notes = read.running, read.left, read.notes
|
||||||
mods = append(mods, m)
|
for _, rs := range st.Running {
|
||||||
}
|
st.RunningByNode[rs.Node]++
|
||||||
sort.Strings(mods)
|
|
||||||
st.Running, st.LeftRunning = []RunningStep{}, []LeftStep{}
|
|
||||||
now, bound := time.Now(), r.StepBound
|
|
||||||
if r.Now != nil {
|
|
||||||
now = r.Now()
|
|
||||||
}
|
|
||||||
if bound <= 0 {
|
|
||||||
bound = DefaultStepBound
|
|
||||||
}
|
|
||||||
for _, m := range mods {
|
|
||||||
sr := r.StepStates[m]
|
|
||||||
if sr.Keys == nil || sr.Get == nil {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
keys, err := sr.Keys()
|
|
||||||
if err != nil {
|
|
||||||
st.Notes = append(st.Notes, "the steps of "+m+" cannot be listed: "+err.Error())
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
// A step that runs started lately: the newest ids are read first, at most MaxReads of them.
|
|
||||||
newestFirst(keys)
|
|
||||||
keys, cut := capKeys(keys, "")
|
|
||||||
if cut != "" {
|
|
||||||
st.Notes = append(st.Notes, "the steps of "+m+": read the newest "+fmt.Sprint(MaxReads)+" of its keys; a step older than those is not shown")
|
|
||||||
}
|
|
||||||
steps := make([]delivery.Step, len(keys))
|
|
||||||
isRunning := make([]bool, len(keys))
|
|
||||||
_ = parallel(len(keys), func(i int) error {
|
|
||||||
raw, err := sr.Get(keys[i])
|
|
||||||
if err != nil || len(raw) == 0 {
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
step, err := delivery.DecodeStep(raw)
|
|
||||||
steps[i], isRunning[i] = step, err == nil && step.State == delivery.StepRunning
|
|
||||||
return nil
|
|
||||||
})
|
|
||||||
for i := range keys {
|
|
||||||
if !isRunning[i] {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
step := steps[i]
|
|
||||||
rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool}
|
|
||||||
if !step.Started.IsZero() {
|
|
||||||
rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z")
|
|
||||||
}
|
|
||||||
if why := r.leftWhy(m, step, now, bound); why != "" {
|
|
||||||
st.LeftRunning = append(st.LeftRunning, LeftStep{RunningStep: rs, Why: why})
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
st.Running = append(st.Running, rs)
|
|
||||||
st.RunningByNode[step.Node]++
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
if r.HaveState.Get == nil {
|
if r.HaveState.Get == nil {
|
||||||
st.Notes = append(st.Notes, "the node-engine's have is not readable here, so a step is not checked against its build module being on its node")
|
st.Notes = append(st.Notes, "the node-engine's have is not readable here, so a step is not checked against its build module being on its node")
|
||||||
@@ -530,7 +485,11 @@ func (r Reader) Status() (Status, error) {
|
|||||||
}
|
}
|
||||||
v, err := delivery.DecodeVerdict(raw)
|
v, err := delivery.DecodeVerdict(raw)
|
||||||
switch {
|
switch {
|
||||||
case err != nil || v.State == delivery.VerdictRunning:
|
case err != nil:
|
||||||
|
case v.State == delivery.VerdictRunning:
|
||||||
|
if why := read.lostWhy(c); why != "" {
|
||||||
|
st.Lost[c.Group] = why
|
||||||
|
}
|
||||||
case IsNodesFailure(v):
|
case IsNodesFailure(v):
|
||||||
st.Waiting[KindNodes] = append(st.Waiting[KindNodes], c.Group)
|
st.Waiting[KindNodes] = append(st.Waiting[KindNodes], c.Group)
|
||||||
case IsUntrunked(v):
|
case IsUntrunked(v):
|
||||||
@@ -551,6 +510,118 @@ func (r Reader) Status() (Status, error) {
|
|||||||
return st, nil
|
return st, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// stepsRead is what the build modules' steps states hold: the steps that can still run, those left running, and every
|
||||||
|
// ask step read, by module, for the candidate it belongs to (stepsOf).
|
||||||
|
type stepsRead struct {
|
||||||
|
running []RunningStep
|
||||||
|
left []LeftStep
|
||||||
|
all []moduleStep
|
||||||
|
notes []string
|
||||||
|
}
|
||||||
|
|
||||||
|
type moduleStep struct {
|
||||||
|
module string
|
||||||
|
step delivery.Step
|
||||||
|
}
|
||||||
|
|
||||||
|
// readSteps reads the newest MaxReads keys of each build module's steps state.
|
||||||
|
func (r Reader) readSteps() stepsRead {
|
||||||
|
out := stepsRead{running: []RunningStep{}, left: []LeftStep{}}
|
||||||
|
mods := make([]string, 0, len(r.StepStates))
|
||||||
|
for m := range r.StepStates {
|
||||||
|
mods = append(mods, m)
|
||||||
|
}
|
||||||
|
sort.Strings(mods)
|
||||||
|
now, bound := time.Now(), r.StepBound
|
||||||
|
if r.Now != nil {
|
||||||
|
now = r.Now()
|
||||||
|
}
|
||||||
|
if bound <= 0 {
|
||||||
|
bound = DefaultStepBound
|
||||||
|
}
|
||||||
|
for _, m := range mods {
|
||||||
|
sr := r.StepStates[m]
|
||||||
|
if sr.Keys == nil || sr.Get == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
keys, err := sr.Keys()
|
||||||
|
if err != nil {
|
||||||
|
out.notes = append(out.notes, "the steps of "+m+" cannot be listed: "+err.Error())
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
// A step that runs started lately: the newest ids are read first, at most MaxReads of them.
|
||||||
|
newestFirst(keys)
|
||||||
|
keys, cut := capKeys(keys, "")
|
||||||
|
if cut != "" {
|
||||||
|
out.notes = append(out.notes, "the steps of "+m+": read the newest "+fmt.Sprint(MaxReads)+" of its keys; a step older than those is not shown")
|
||||||
|
}
|
||||||
|
steps := make([]delivery.Step, len(keys))
|
||||||
|
decoded := make([]bool, len(keys))
|
||||||
|
_ = parallel(len(keys), func(i int) error {
|
||||||
|
raw, err := sr.Get(keys[i])
|
||||||
|
if err != nil || len(raw) == 0 {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
step, err := delivery.DecodeStep(raw)
|
||||||
|
steps[i], decoded[i] = step, err == nil
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
for i := range keys {
|
||||||
|
if !decoded[i] {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
step := steps[i]
|
||||||
|
out.all = append(out.all, moduleStep{module: m, step: step})
|
||||||
|
if step.State != delivery.StepRunning {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool, What: step.What}
|
||||||
|
if !step.Started.IsZero() {
|
||||||
|
rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z")
|
||||||
|
}
|
||||||
|
if why := r.leftWhy(m, step, now, bound); why != "" {
|
||||||
|
out.left = append(out.left, LeftStep{RunningStep: rs, Why: why})
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
out.running = append(out.running, rs)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// asksOf says whether a step is an ask of one of the candidate's pull requests: its What names `<repository>@<head12>`
|
||||||
|
// (buildstep.AskWhat).
|
||||||
|
func asksOf(c delivery.Candidate, what string) bool {
|
||||||
|
for _, p := range c.Pulls {
|
||||||
|
if p.Repository != "" && strings.Contains(what, p.Repository+"@"+short12(p.Head.Head)) {
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
|
||||||
|
// lostWhy says why a candidate whose verdict is running has a check that nothing will judge, or "": no step of its pull
|
||||||
|
// requests can still run, and its newest one ended.
|
||||||
|
func (s stepsRead) lostWhy(c delivery.Candidate) string {
|
||||||
|
for _, rs := range s.running {
|
||||||
|
if asksOf(c, rs.What) {
|
||||||
|
return ""
|
||||||
|
}
|
||||||
|
}
|
||||||
|
var last *delivery.Step
|
||||||
|
for i := range s.all {
|
||||||
|
st := s.all[i].step
|
||||||
|
if st.State != delivery.StepRunning && asksOf(c, st.What) && (last == nil || st.Ended.After(last.Ended)) {
|
||||||
|
last = &s.all[i].step
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if last == nil {
|
||||||
|
return "" // its ask waits in its build seat's stream: no node took it yet
|
||||||
|
}
|
||||||
|
return fmt.Sprintf("its verdict is running, but no step of it runs: its last step %s on %s ended %s at %s and no verdict followed; %s asks it again",
|
||||||
|
last.ID, last.Node, last.State, last.Ended.UTC().Format(time.RFC3339), AskAgainTool)
|
||||||
|
}
|
||||||
|
|
||||||
// DefaultStepBound is how long one check ask's step may run once a build node took it (its key written running; the fetch,
|
// DefaultStepBound is how long one check ask's step may run once a build node took it (its key written running; the fetch,
|
||||||
// the digests and the levels), never counting its wait in the stream; past it the check fails naming the node.
|
// the digests and the levels), never counting its wait in the stream; past it the check fails naming the node.
|
||||||
const DefaultStepBound = 30 * time.Minute
|
const DefaultStepBound = 30 * time.Minute
|
||||||
|
|||||||
@@ -93,7 +93,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
|||||||
t.Errorf("module = %q", m.Module)
|
t.Errorf("module = %q", m.Module)
|
||||||
}
|
}
|
||||||
// delivery.CheckerKeptHistory, CheckerOwedHistory, PreviewsHistory, CheckerVerdictsHistory; the names follow the SDK's buckets.
|
// delivery.CheckerKeptHistory, CheckerOwedHistory, PreviewsHistory, CheckerVerdictsHistory; the names follow the SDK's buckets.
|
||||||
if want := []state{{Name: "hearing", History: 1}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) {
|
if want := []state{{Name: "asks", History: 1}, {Name: "hearing", History: 1}, {Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) {
|
||||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||||
}
|
}
|
||||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
|
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
|
||||||
@@ -124,8 +124,9 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
|||||||
t.Errorf("streams = %+v, want %+v", m.Streams, want)
|
t.Errorf("streams = %+v, want %+v", m.Streams, want)
|
||||||
}
|
}
|
||||||
// The read tools (issue 1686) are served by the tools bundle, not by the checker's process.
|
// The read tools (issue 1686) are served by the tools bundle, not by the checker's process.
|
||||||
// The try's tools (ADR 0501 §5; the end by hand, issue 734) beside them: mesh-controller's try calls them.
|
// The try's tools (ADR 0501 §5; the end by hand, issue 734) beside them: mesh-controller's try calls them. A lost check
|
||||||
if want := []string{"checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_try", "checker_try_end", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) {
|
// is asked again by hand (checker_ask_again).
|
||||||
|
if want := []string{"checker_ask_again", "checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_try", "checker_try_end", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) {
|
||||||
t.Errorf("tools = %v, want %v", m.Tools, want)
|
t.Errorf("tools = %v, want %v", m.Tools, want)
|
||||||
}
|
}
|
||||||
for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} {
|
for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} {
|
||||||
|
|||||||
@@ -10,6 +10,7 @@
|
|||||||
{ "name": "checks", "events": ["preview.**", "owed.**"] }
|
{ "name": "checks", "events": ["preview.**", "owed.**"] }
|
||||||
],
|
],
|
||||||
"state": [
|
"state": [
|
||||||
|
{ "name": "asks", "history": 1 },
|
||||||
{ "name": "hearing", "history": 1 },
|
{ "name": "hearing", "history": 1 },
|
||||||
{ "name": "kept", "history": 1 },
|
{ "name": "kept", "history": 1 },
|
||||||
{ "name": "owed", "history": 5 },
|
{ "name": "owed", "history": 5 },
|
||||||
@@ -38,6 +39,7 @@
|
|||||||
"node-build-node"
|
"node-build-node"
|
||||||
],
|
],
|
||||||
"tools": [
|
"tools": [
|
||||||
|
"checker_ask_again",
|
||||||
"checker_kept",
|
"checker_kept",
|
||||||
"checker_previews",
|
"checker_previews",
|
||||||
"checker_pull",
|
"checker_pull",
|
||||||
|
|||||||
Reference in New Issue
Block a user