A check that could not run is asked again on the node-engine seat's running of its part (milestone 47/G W6.9b, issue 1373)
The checker consumes node-engine.running (kept last per subject) through its durable consumer and hands each running.<node>.<module> to PartRunning; a body naming no part asks nothing and is said. Naming the not-heard part by its module (the seat git's holder, a build seat's holder) waits on the catalog half's claims (W3.6): until then the part is unnamed and the first running asks the check again. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ Builds: design 47 G §253, §315
This commit is contained in:
@@ -2,8 +2,11 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
)
|
||||
|
||||
@@ -24,3 +27,21 @@ func waitCouldNotRun(ctx context.Context, ch *check.Checker, kept <-chan struct{
|
||||
done()
|
||||
}
|
||||
}
|
||||
|
||||
// partRunner is what a `running` is handed to: check.Checker's PartRunning.
|
||||
type partRunner interface {
|
||||
PartRunning(ctx context.Context, node, module, fingerprint string)
|
||||
}
|
||||
|
||||
// heardRunning takes one node-engine seat's `running.<node>.<module>` in, as the checker's durable consumer of its stream
|
||||
// hands it (module.json consumes node-engine.running): the part named runs at its fingerprint, so every check that could not
|
||||
// run for want of it is asked again once (check.Checker.PartRunning). A body that names no node, module or fingerprint asks
|
||||
// nothing, said in the log.
|
||||
func heardRunning(ctx context.Context, ch partRunner, body []byte, logf func(string, ...any)) {
|
||||
var r nodeengine.RunningEvent
|
||||
if err := json.Unmarshal(body, &r); err != nil || r.Node == "" || r.Module == "" || r.Fingerprint == "" {
|
||||
logf("a running of the node-engine seat that names no node, module or fingerprint asks no check again: %s (%v)", body, err)
|
||||
return
|
||||
}
|
||||
ch.PartRunning(ctx, r.Node, r.Module, r.Fingerprint)
|
||||
}
|
||||
|
||||
@@ -66,4 +66,30 @@ func TestAForgeReadThatReachesNobodyIsUnheard(t *testing.T) { // 1624 C7: "nats:
|
||||
if !errors.As(err, &u) {
|
||||
t.Errorf("err = %v, want it marked as reaching nobody", err)
|
||||
}
|
||||
// Issue 1373, milestone 47/G W6.9: the check that could not run is asked again when the part runs, heard as the
|
||||
// node-engine seat's running.<node>.<module> on the checker's consumer; a running that names no part asks nothing.
|
||||
t.Run("asked again on running", func(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
body string
|
||||
want []string
|
||||
}{
|
||||
{`{"v":1,"node":"novox","module":"gitea","fingerprint":"src2:ab","manifest":"sha256:cd","at":"2026-11-09T20:00:00Z"}`, []string{"novox gitea src2:ab"}},
|
||||
{`{"v":1,"node":"novox","module":"gitea"}`, nil},
|
||||
{`not json`, nil},
|
||||
} {
|
||||
var got []string
|
||||
heardRunning(context.Background(), runnerFunc(func(_ context.Context, node, module, fp string) {
|
||||
got = append(got, node+" "+module+" "+fp)
|
||||
}), []byte(c.body), t.Logf)
|
||||
if strings.Join(got, ",") != strings.Join(c.want, ",") {
|
||||
t.Errorf("%s asked %v, want %v", c.body, got, c.want)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
type runnerFunc func(ctx context.Context, node, module, fingerprint string)
|
||||
|
||||
func (f runnerFunc) PartRunning(ctx context.Context, node, module, fingerprint string) {
|
||||
f(ctx, node, module, fingerprint)
|
||||
}
|
||||
|
||||
@@ -23,6 +23,8 @@ import (
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
|
||||
@@ -354,6 +356,19 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
defer cc.Stop()
|
||||
}
|
||||
// The node-engine seat's running.<node>.<module>, kept last per subject (design 47): a part a check could not hear runs,
|
||||
// and the check is asked again (issue 1373). Its stream and the checker's durable consumer on it are made from the
|
||||
// manifest's consumes; where they are not there yet the checker says so, and a check that could not run waits for its
|
||||
// bound and is said as a finding.
|
||||
if cons, err := cred.EventConsumer(ctx, js, nodeengine.StreamRunning); err != nil {
|
||||
logger.Printf("the consumer %s_%s on %s does not exist yet, so no check that could not run is asked again until the checker starts again: %v", cred.Node, cred.Module, nodeengine.StreamRunning, err)
|
||||
} else {
|
||||
cc, err := process.HearEvents(cons, func(m jetstream.Msg) { heardRunning(ctx, ch, m.Data(), logger.Printf) }, logger.Printf)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer cc.Stop()
|
||||
}
|
||||
errs := make(chan error, 8)
|
||||
// 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
|
||||
|
||||
@@ -141,7 +141,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
||||
// The checker is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it starts.
|
||||
if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported",
|
||||
"mesh-catalog.catalog-revision", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked", "mesh-delivery-assigner.views.assignments", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules",
|
||||
"mesh-delivery-indexer.views.packages", "mesh-delivery-indexer.views.predefined", "mesh-delivery-indexer.views.seat-definitions"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
"mesh-delivery-indexer.views.packages", "mesh-delivery-indexer.views.predefined", "mesh-delivery-indexer.views.seat-definitions", "node-engine.running"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
t.Errorf("consumes = %v, want %v", m.Consumes, want)
|
||||
}
|
||||
if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
|
||||
|
||||
@@ -27,7 +27,8 @@
|
||||
"mesh-delivery-indexer.views.packages",
|
||||
"mesh-delivery-indexer.views.predefined",
|
||||
"mesh-delivery-indexer.views.seat-definitions",
|
||||
"mesh-conditions.findings-asked"
|
||||
"mesh-conditions.findings-asked",
|
||||
"node-engine.running"
|
||||
],
|
||||
"streams": [
|
||||
{ "name": "checks", "events": ["preview.**", "owed.**", "try.**"] },
|
||||
|
||||
Reference in New Issue
Block a user