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

CK2: the checker takes the trunk heads from the catalog half's revision and catalog-revision, not the gate's source (repairs the checker against the new contracts)

Browse Source 
Builds: design 47 G §197, §195, §200

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 12:08:09 +01:00
parent fdc2cec328
commit 21195a06be
12 changed files with 171 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go
+4 -6
View File
@@ -14,7 +14,7 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
stdio "git.novox.be/novox/mesh-sdk/go"
@@ -426,14 +426,12 @@ func main() {
AskState: read(stdio.State(check.AsksState)), BuildState: read(stdio.State("mesh-delivery-builder.builds"))}
r.Runs = runsOf{ownerRead(stdio.Ask, indexerModulesTool)} // the node-engine at the commit the mesh runs, as the checker reads it
tries := stdio.State(check.TriesState)
source := stdio.State("mesh-delivery-gate.source")
trunks := func() (map[string]string, error) {
e, err := source.Get(meshdeliveryclient.SourceMainKey)
if err != nil || e == nil {
raw, err := stdio.Ask(check.TrunkVerb, meshcatalog.RevisionInput{})
if err != nil {
return nil, err
}
s, err := meshdeliveryclient.DecodeSource(e.Value)
return s.Heads, err
return check.TrunkHeads(raw)
}
r.Trunks = trunks
store := stateStore{StateRead: read(tries), create: func(key string, value []byte) error {
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go
+25 -10
View File
@@ -417,9 +417,11 @@ type bed struct {
inputsAnswered atomic.Int32
fake *fakeBuildModules
srvURL string
announce []string
mu sync.Mutex
logs saidLog
// trunk is the catalog half's trunk revision the bed's seat answers `revision` with (design 47 G section 197).
trunkNow *atomic.Value
announce []string
mu sync.Mutex
logs saidLog
}
// saidLog is what the program said, safe for its goroutines and the test's.
@@ -536,9 +538,24 @@ func startBed(t *testing.T) *bed {
t.Fatal(err)
}
}
trunkView := &atomic.Value{}
trunkView.Store(map[string]string{})
for _, subject := range []string{"mesh.seat.mesh-catalog.call.revision", "mesh.seat.mesh-catalog.call.revision.>"} {
if _, err := ctl.Subscribe(subject, func(m *nats.Msg) {
var heads []meshcatalog.Head
for repo, commit := range trunkView.Load().(map[string]string) {
heads = append(heads, meshcatalog.Head{Repository: repo, Commit: commit})
}
body, _ := json.Marshal(meshcatalog.RevisionAnswer{Revision: meshcatalog.CatalogRevision{Kind: "trunk", Revision: 1, Trunk: heads}})
reply, _ := json.Marshal(map[string]any{"result": json.RawMessage(body), "node": "ctl"})
_ = m.Respond(reply)
}); err != nil {
t.Fatal(err)
}
}
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
t.Cleanup(cancel)
b := &bed{ctx: ctx, js: js, buckets: map[string]jetstream.KeyValue{}, tools: map[string]jetstream.KeyValue{}, srvURL: srv.ClientURL()}
b := &bed{trunkNow: trunkView, ctx: ctx, js: js, buckets: map[string]jetstream.KeyValue{}, tools: map[string]jetstream.KeyValue{}, srvURL: srv.ClientURL()}
for module, stream := range toolchainStreams {
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{"mesh.stream." + module + ".event.toolchain.reported"}}); err != nil {
t.Fatal(err)
@@ -589,7 +606,7 @@ func startBed(t *testing.T) *bed {
}
for name, history := range map[string]int{check.CheckerVerdictsBucket: 5, check.CheckerKeptBucket: 1,
check.CheckerOwedBucket: 5, check.PreviewsBucket: 5, check.CandidatesBucket: 10,
check.ModulesBucket: 20, check.ManifestsBucket: 1, check.SourceBucket: 64,
check.ModulesBucket: 20, check.ManifestsBucket: 1,
check.PresenceBucket: 20, check.HaveBucket: 20, check.TriesBucket: 1,
check.TurnsBucket: 5, builds.BuilderBuildsBucket: 1, check.InputsBucket: 1} {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)})
@@ -631,11 +648,9 @@ func startBed(t *testing.T) *bed {
func (b *bed) setTrunk(t *testing.T, heads map[string]string) {
t.Helper()
data, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: heads})
if err != nil {
t.Fatal(err)
}
if _, err := b.buckets[check.SourceBucket].Put(b.ctx, meshdeliveryclient.SourceMainKey, data); err != nil {
b.trunkNow.Store(heads)
body, _ := json.Marshal(meshcatalog.CatalogRevisionEvent{Kind: "trunk", Branch: "main", Revision: 1, V: 1})
if _, err := b.js.Publish(b.ctx, meshcatalog.SubjectCatalogRevision("main"), body); err != nil {
t.Fatal(err)
}
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go
+6 -20
View File
@@ -26,7 +26,6 @@ import (
"mesh-delivery-checker/internal/check"
"git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
@@ -375,26 +374,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}()
}
}
// The gate's source says the trunk head of each repository, which a pull request is merged onto. Where it is not there
// yet no trunk is known, so nothing is checked.
if kv := process.OptionalBucket(ctx, js, check.SourceBucket, "no trunk is known, so nothing is checked until the checker starts again", logger.Printf); kv != nil {
sw, err := settle(ctx, kv, func(e jetstream.KeyValueEntry) {
if e.Key() == meshdeliveryclient.SourceMainKey {
trunks.Put(e.Value(), !opOf(e))
}
})
if err != nil {
process.NotWatched(logger.Printf, check.SourceBucket, err, "no trunk is known, so nothing is checked until the checker starts again")
} else {
go func() {
errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) {
if e.Key() == meshdeliveryclient.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) {
ch.TrunkMoved(ctx) // the trunk moved: its statuses go pending, and a verdict for the old trunk is judged again
}
})
}()
}
// The catalog half says the trunk head of each repository, which a pull request is merged onto (design 47 G §197). Where it
// does not answer yet no trunk is known, so nothing is checked.
firstTrunks, err := followTrunks(hub, caller.Call, trunks, ch.TrunkMoved, logger.Printf)
if err != nil {
return fmt.Errorf("the checker cannot hear the catalog half's revisions: %w", err)
}
firstTrunks(ctx)
// The gate's merge turns (novox/hq ADR 0548 §3): no check is claimed of a repository an approved merge waits on, and the
// delete of its turn claims it. Read before the candidates, so no check starts past a turn written before this start.
// Where the gate's bucket is not there yet it is read as empty, and opened again before each claim until it is there
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/trunk.go
+57
View File
@@ -0,0 +1,57 @@
package main
import (
"context"
"encoding/json"
"fmt"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"mesh-delivery-checker/internal/check"
)
// The checker takes the trunk heads a check merges onto from the catalog half (design 47 G §197), never from the gate's state
// `source`: a first view by the seat mesh-catalog's verb `revision`, which carries the trunk head of every repository, then
// the seat's event `catalog-revision` of a trunk revision, each read again by the verb.
// readTrunks asks the catalog half's newest trunk revision and sets the heads it carries. It says whether they changed. A
// candidate's revision, or an answer that does not read, changes nothing.
func readTrunks(ask func(string, any) (json.RawMessage, error), trunks *check.Trunks) (bool, error) {
raw, err := ask(check.TrunkVerb, meshcatalog.RevisionInput{})
if err != nil {
return false, fmt.Errorf("the catalog half's revision cannot be read: %w", err)
}
heads, err := check.TrunkHeads(raw)
if err != nil {
return false, err
}
return trunks.Set(heads), nil
}
// followTrunks reads the first view and then follows `catalog-revision`: a trunk revision's event reads the heads again, and a
// change of them is the trunk moving (statuses go pending, a verdict for the old trunk is judged again). A read that fails is
// said, and made again at the next event.
func followTrunks(hub interface {
Handle(string, func(jetstream.Msg)) error
}, ask func(string, any) (json.RawMessage, error), trunks *check.Trunks, moved func(context.Context), logf func(string, ...any)) (first func(context.Context), err error) {
run := func(ctx context.Context) {
changed, err := readTrunks(ask, trunks)
if err != nil {
logf("no trunk is known until the catalog half answers, so nothing is checked: %v", err)
return
}
if changed {
moved(ctx)
}
}
err = hub.Handle(meshcatalog.SubjectCatalogRevisionAll, func(m jetstream.Msg) {
var e meshcatalog.CatalogRevisionEvent
if json.Unmarshal(m.Data(), &e) == nil && e.Kind == "trunk" {
run(context.Background())
}
_ = m.Ack()
})
return run, err
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/trunk_test.go
+38
View File
@@ -0,0 +1,38 @@
package main
import (
"encoding/json"
"errors"
"testing"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"mesh-delivery-checker/internal/check"
)
// design 47 G §197: the trunk heads come from the catalog half's trunk revision, by the seat verb revision, and never from the
// gate's source; a candidate's revision and an unanswered verb change nothing.
func TestTrunkHeadsComeFromTheCatalogHalf(t *testing.T) {
answer := func(kind string, heads ...meshcatalog.Head) func(string, any) (json.RawMessage, error) {
return func(key string, _ any) (json.RawMessage, error) {
if key != "seat:mesh-catalog.revision" {
t.Fatalf("asked %s", key)
}
return json.Marshal(meshcatalog.RevisionAnswer{Revision: meshcatalog.CatalogRevision{Kind: kind, Trunk: heads}})
}
}
trunks := &check.Trunks{}
h := meshcatalog.Head{Repository: "r", Commit: "c1"}
if changed, err := readTrunks(answer("trunk", h), trunks); err != nil || !changed || trunks.Heads()["r"] != "c1" {
t.Fatalf("first view: %v %v %v", changed, err, trunks.Heads())
}
if changed, _ := readTrunks(answer("trunk", h), trunks); changed {
t.Fatal("the same heads moved the trunk")
}
if _, err := readTrunks(answer("candidate", meshcatalog.Head{Repository: "r", Commit: "c2"}), trunks); err == nil || trunks.Heads()["r"] != "c1" {
t.Fatalf("a candidate's revision: %v %v", err, trunks.Heads())
}
if _, err := readTrunks(func(string, any) (json.RawMessage, error) { return nil, errors.New("down") }, trunks); err == nil || trunks.Heads()["r"] != "c1" {
t.Fatalf("an unanswered verb: %v %v", err, trunks.Heads())
}
}
modules/mesh-delivery-checker/internal/check/case.go
+4 -11
View File
@@ -234,22 +234,15 @@ func GateHeads(c meshcatalog.CandidatesByGroupValue) string {
// sameTrunk says whether two verdicts' trunk heads are the same.
func sameTrunk(a, b map[string]string) bool { return maps.Equal(a, b) }
// Trunks holds the trunk head of each repository, as the gate's source.main says (a watch puts it in).
// Trunks holds the trunk head of each repository, as the catalog half's trunk revision says (a follower of the seat mesh-catalog's
// verb revision and its event catalog-revision sets it; design 47 G §197).
type Trunks struct {
mu sync.Mutex
heads map[string]string
}
// Put takes a source.main value in and answers whether the heads changed.
func (t *Trunks) Put(value []byte, deleted bool) bool {
var heads map[string]string
if !deleted {
s, err := meshdeliveryclient.DecodeSource(value)
if err != nil {
return false
}
heads = s.Heads
}
// Set takes the trunk heads in and answers whether they changed.
func (t *Trunks) Set(heads map[string]string) bool {
t.mu.Lock()
defer t.mu.Unlock()
if maps.Equal(t.heads, heads) {
modules/mesh-delivery-checker/internal/check/checker_test.go
+1 -6
View File
@@ -14,7 +14,6 @@ import (
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
@@ -107,11 +106,7 @@ func newRig(t testing.TB, j Judge) *rig {
func (r *rig) setTrunks(t testing.TB, heads map[string]string) bool {
t.Helper()
data, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: heads})
if err != nil {
t.Fatal(err)
}
return r.trunks.Put(data, false)
return r.trunks.Set(heads)
}
func (r *rig) verdict(t testing.TB, key string) meshdeliverychecker.VerdictsByGroupHeadsValue {
modules/mesh-delivery-checker/internal/check/examiner_test.go
+2 -8
View File
@@ -25,9 +25,7 @@ import (
gobuild "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
"git.novox.be/novox/mesh-contracts/go/seats/meshconfig"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient"
"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
@@ -880,15 +878,11 @@ func TestTheOutcomeSaysIfEachCandidatesModuleIsBuiltFailedOrWaiting(t *testing.T
}
}
// trunksOf are trunk heads of every repository, as the gate's source names them.
// trunksOf are trunk heads of every repository, as the catalog half's trunk revision names them.
func trunksOf(t *testing.T, heads map[string]string) *Trunks {
t.Helper()
data, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByMainValue{Group: "g", Heads: heads})
if err != nil {
t.Fatal(err)
}
tr := &Trunks{}
tr.Put(data, false)
tr.Set(heads)
return tr
}
modules/mesh-delivery-checker/internal/check/names.go
-1
View File
@@ -22,7 +22,6 @@ const (
// GateModule holds the seat mesh-delivery's states source and turns, which the checker reads (reads: mesh-delivery-gate.source,
// mesh-delivery-gate.turns).
GateModule = "mesh-delivery-gate"
SourceBucket = GateModule + "_" + meshdelivery.StateSource
TurnsBucket = GateModule + "_" + meshdelivery.StateTurns
// IndexerModule holds the seat mesh-catalog's states, which the checker follows by their views (consumes:
modules/mesh-delivery-checker/internal/check/trunkview.go
+28
View File
@@ -0,0 +1,28 @@
package check
import (
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
)
// TrunkVerb is the key of the seat verb the trunk heads are asked by (design 47 G §197).
const TrunkVerb = "seat:" + meshcatalog.Seat + "." + meshcatalog.VerbRevision
// TrunkHeads reads the trunk head of every repository out of the seat mesh-catalog's answer of `revision`. A candidate's
// revision is no trunk's and is refused.
func TrunkHeads(raw json.RawMessage) (map[string]string, error) {
var a meshcatalog.RevisionAnswer
if err := json.Unmarshal(raw, &a); err != nil {
return nil, fmt.Errorf("the catalog half's revision does not read: %w", err)
}
if a.Revision.Kind != "trunk" {
return nil, fmt.Errorf("the catalog half's revision %d is a %q revision, not a trunk's", a.Revision.Revision, a.Revision.Kind)
}
heads := map[string]string{}
for _, h := range a.Revision.Trunk {
heads[h.Repository] = h.Commit
}
return heads, nil
}
modules/mesh-delivery-checker/manifest_test.go
+4 -4
View File
@@ -112,7 +112,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
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-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-gate.turns", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
"mesh-delivery-builder.builds", "mesh-delivery-gate.turns", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
t.Errorf("reads = %v, want %v", m.Reads, want)
}
// DLV-M2 (ADR 0615 §1, ADR 0596): the indexer's candidates, modules, manifests, packages and predefined keys and the
@@ -131,7 +131,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_read_predefined",
"mesh-delivery-indexer.indexer_read_seat_definitions",
"seat:mesh-catalog.seat-definitions", "seat:mesh-config.renames", "seat:mesh-hand-acts.hand-act", "seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
"seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", "seat:mesh-config.renames", "seat:mesh-hand-acts.hand-act", "seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
t.Errorf("invokes = %v, want the forge's reads, the seats' drop and compose-candidate", m.Invokes)
}
if want := []string{"node-build-go", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
@@ -140,7 +140,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.
// And a group's check by the package registry seats' published of its tagged versions (design 51 *Tagged versions*).
if want := []string{"go-package-registry.published", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported",
"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-assigner.views.assignments", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules",
"mesh-catalog.catalog-revision", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "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", "npm-package-registry.published"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want)
}
@@ -202,7 +202,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
}
// D-11 (ADR 0634 §1): the holders and buckets internal/check names, no contract naming a holder, are the manifest's: a state
// the checker reads is a `reads` (<module>.<state>) and a state of the indexer it follows a `consumes` of its views family.
for bucket, read := range map[string]string{check.SourceBucket: "mesh-delivery-gate.source", check.TurnsBucket: "mesh-delivery-gate.turns",
for bucket, read := range map[string]string{check.TurnsBucket: "mesh-delivery-gate.turns",
check.PresenceBucket: "mesh-delivery-watchman.presence", check.HaveBucket: "node-engine.have"} {
if strings.Replace(bucket, "_", ".", 1) != read || !slices.Contains(m.Reads, read) {
t.Errorf("the bucket %s is not the declared read %s", bucket, read)
modules/mesh-delivery-checker/module.json
+2 -1
View File
@@ -16,6 +16,7 @@
],
"consumes": [
"go-package-registry.published",
"mesh-catalog.catalog-revision",
"mesh-catalog.seat-definition-changed",
"mesh-catalog.seat-definition-removed",
"mesh-build-go.toolchain.reported",
@@ -52,7 +53,6 @@
"mesh-build-node.steps",
"mesh-build-node.toolchains",
"mesh-delivery-builder.builds",
"mesh-delivery-gate.source",
"mesh-delivery-gate.turns",
"mesh-delivery-watchman.presence",
"node-engine.have"
@@ -90,6 +90,7 @@
"mesh-delivery-indexer.indexer_read_packages",
"mesh-delivery-indexer.indexer_read_predefined",
"mesh-delivery-indexer.indexer_read_seat_definitions",
"seat:mesh-catalog.revision",
"seat:mesh-catalog.seat-definitions",
"seat:mesh-config.renames",
"seat:mesh-hand-acts.hand-act",