From ff6c038d595dc1f2dc5e65efd3f348facf9e006f Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 19:29:59 +0100 Subject: [PATCH] =?UTF-8?q?The=20assigner's=20composition=20check=20compos?= =?UTF-8?q?es=20over=20the=20input=20the=20nodes=20compose=20from,=20read?= =?UTF-8?q?=20with=20frombus=20once=20the=20catalogue=20places=20them=20(i?= =?UTF-8?q?ssue=201126,=20ADR=200547=20=C2=A72,=20A2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/mesh-delivery-assigner/pass_test.go | 16 ++++ .../cmd/mesh-delivery-assigner/world.go | 79 +++++++++++++++++++ .../internal/assign/compose.go | 7 ++ .../internal/assign/composer.go | 15 +++- .../internal/assign/composer_test.go | 20 +++++ 5 files changed, 135 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 321f3dd..b0f52e3 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -781,6 +781,10 @@ func TestAPairOneSourceNamesAloneGetsNoWantAndOneCondition(t *testing.T) { if got, _ := p.w.assigned(); !slices.Equal(got["g14"], []string{"a", "b"}) || len(p.sayPlacement(p.w.placementDiffers())) != 0 || len(said) != 0 { t.Fatalf("before any placement: assigned %v, said %v", got, said) } + none := func(context.Context, string) ([]byte, error) { return nil, errNotThere } + if m := p.w.mesh(context.Background(), none); m.Input != nil || m.InputUnread != "" { + t.Errorf("before any placement the check composes over %+v (%q)", m.Input, m.InputUnread) + } place("g14", "a", "c") got, _ := p.w.assigned() frozen := p.sayPlacement(p.w.placementDiffers()) @@ -795,6 +799,18 @@ func TestAPairOneSourceNamesAloneGetsNoWantAndOneCondition(t *testing.T) { if frozen := p.sayPlacement(p.w.placementDiffers()); len(frozen) != 0 || len(said) != 4 || said[2].State != asks.StateCleared || said[3].State != asks.StateCleared { t.Errorf("both agree again: frozen %v, said %+v", frozen, said) } + + // 1126 (ADR 0547 §2): placed, the check composes over the input the nodes compose from (frombus), once each placed node + // wrote its record; until then it says why and composes the assignments alone. + if m := p.w.mesh(context.Background(), none); m.Input != nil || !strings.Contains(m.InputUnread, "g14") { + t.Errorf("a placed node with no record: input %+v, unread %q", m.Input, m.InputUnread) + } + rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "g14", TunnelKey: "k"}}) + p.w.put(delivery.HaveBucket, "g14.node", rec) + m := p.w.mesh(context.Background(), none) + if m.Input == nil || len(m.Input.Nodes) != 1 || !slices.Equal(m.Input.Nodes[0].Assigned, []string{"a", "b", "c"}) || m.Input.Nodes[0].Place.Key != "k" { + t.Errorf("the check's input from the bus: %+v (%q)", m.Input, m.InputUnread) + } } // 1126 (ADR 0547 §2 to §4): a node the catalogue places composes its own: its wants name the build alone and its own want diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 821cd89..22fb1d4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -14,6 +14,8 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/provider" @@ -258,9 +260,86 @@ func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh { } m.Manifests[name] = p.manifest } + // Once the catalogue places the nodes, the check composes over the input they compose from (ADR 0547 §2, ADR 0549 + // §3): one composer's input for the check and the nodes. An input that cannot be read is said, and the nodes and their + // assignments alone are composed, as before any placement. + if _, seeded := w.placed(); seeded { + in, err := frombus.Input(ctx, worldReader{w: w, manifests: manifests}, frombus.Mesh{}, frombus.Scope{Every: true}) + if err != nil { + m.InputUnread = err.Error() + } else { + m.Input = &in + } + } return m } +// worldReader is what the assigner watched as frombus reads the bus: the indexer's placement and modules, each node's +// record in its own have key, the manifests by digest; the check reads no config value, own secret or credential (the +// composition it asks resolves, and stands in for what it does not carry). +type worldReader struct { + w *world + manifests getter +} + +func (r worldReader) Placements(context.Context) (map[string]delivery.Placement, error) { + out := map[string]delivery.Placement{} + for key, value := range r.w.read(delivery.PlacementBucket) { + node, err := delivery.ParsePlacementKey(key) + if err != nil { + continue + } + if p, err := delivery.DecodePlacement(value); err == nil && p.Node == node { + out[node] = p + } + } + return out, nil +} + +func (r worldReader) Records(context.Context) (map[string]delivery.NodeRecord, error) { + out := map[string]delivery.NodeRecord{} + for key, value := range r.w.read(delivery.HaveBucket) { + node, _, isNode, err := delivery.ParseHaveKey(key) + if err != nil || !isNode { + continue + } + if h, err := delivery.DecodeHaveNode(value); err == nil && h.Record != nil { + out[node] = *h.Record + } + } + return out, nil +} + +func (r worldReader) Modules(context.Context) (map[string]delivery.Module, error) { + return r.w.trunk(), nil +} + +func (r worldReader) Manifest(ctx context.Context, digest string) ([]byte, error) { + key, err := delivery.ManifestsKey(digest) + if err != nil { + return nil, err + } + raw, err := r.manifests(ctx, key) + if err != nil { + return nil, err + } + m, err := delivery.DecodeManifest(raw) + if err != nil { + return nil, err + } + return []byte(m.JSON), nil +} + +func (worldReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil } + +func (worldReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { + return nil, nil +} + +func (worldReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) { + return nil, nil, nil +} + // change is a candidate's change for the composition check, or why the composition waits for it. func (w *world) change(ctx context.Context, manifests getter, c delivery.Candidate, trunk map[string]delivery.Module) (assign.Change, string) { ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, diff --git a/modules/mesh-delivery-assigner/internal/assign/compose.go b/modules/mesh-delivery-assigner/internal/assign/compose.go index b1f26e4..cb788b4 100644 --- a/modules/mesh-delivery-assigner/internal/assign/compose.go +++ b/modules/mesh-delivery-assigner/internal/assign/compose.go @@ -35,6 +35,13 @@ type Mesh struct { // Digests is every trunk module's manifest digest in the indexer's manifests bucket, readable or not: what a // composer at another mesh-sdk reads (issue 1622). Digests map[string]string + // Input is the rest of the mesh as the nodes compose from it, filled by mesh-sdk's frombus from the catalogue's + // placement and the nodes' records (novox/hq ADR 0547 §2, ADR 0549 §3, issue 1126), so the check and the nodes share + // one composer's input; nil before any placement is read, when the nodes and their assignments alone are composed. + Input *compose.Input + // InputUnread says why that input could not be read from the bus, when it could not: the check then composes the + // nodes and their assignments alone, and says so in its warnings. + InputUnread string } // Change is a group's change as the indexer's candidate holds it: the manifest of every module it moves or adds, those diff --git a/modules/mesh-delivery-assigner/internal/assign/composer.go b/modules/mesh-delivery-assigner/internal/assign/composer.go index b8ccb02..8c60ce8 100644 --- a/modules/mesh-delivery-assigner/internal/assign/composer.go +++ b/modules/mesh-delivery-assigner/internal/assign/composer.go @@ -42,8 +42,15 @@ func HeadJob(m Mesh, c Change) (delivery.ComposeJob, error) { for _, name := range c.Removed { delete(change, name) } - req := composer.Request{V: composer.Version, Input: plainInput(m.Assigned), Depth: composer.DepthResolve, - HoldingsFromClaims: true, Machines: Touched(m, c), + in, fromClaims := plainInput(m.Assigned), true + if m.Input != nil { + // The input the nodes compose from, without its catalogue: the cases carry the manifests (issue 1126). + in = *m.Input + in.Catalogue = nil + fromClaims = len(in.Holdings) == 0 + } + req := composer.Request{V: composer.Version, Input: in, Depth: composer.DepthResolve, + HoldingsFromClaims: fromClaims, Machines: Touched(m, c), Cases: []composer.Case{{Name: CaseTrunk, Manifests: trunk}, {Name: CaseChange, Manifests: change}}} return job(SDKRepository, c.SDKHead, SDKDir, req, "the mesh-sdk head of "+c.Group) } @@ -195,6 +202,10 @@ func FromComposer(m Mesh, c Change, head Ran, readers []Reader, took time.Durati readerFails, readerWarnings := judgeReaders(c, change, readers) fails = append(fails, readerFails...) v.Warnings = append(v.Warnings, readerWarnings...) + if m.InputUnread != "" { + v.Warnings = append(v.Warnings, "composed over the nodes and their assignments alone: the input the nodes compose from "+ + "cannot be read from the bus ("+m.InputUnread+")") + } return finish(v, m, c, machines, fails) } diff --git a/modules/mesh-delivery-assigner/internal/assign/composer_test.go b/modules/mesh-delivery-assigner/internal/assign/composer_test.go index 4b709c8..2bc3258 100644 --- a/modules/mesh-delivery-assigner/internal/assign/composer_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/composer_test.go @@ -8,6 +8,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" "git.novox.be/novox/mesh-sdk/go/composer" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -100,6 +101,25 @@ func TestAGroupWithAnSDKMemberAsksTheComposerAtItsHead(t *testing.T) { if !slices.Equal(req.Machines, []string{"g14"}) || req.Depth != composer.DepthResolve || !req.HoldingsFromClaims { t.Errorf("request = machines %v, depth %q, holdings from claims %v", req.Machines, req.Depth, req.HoldingsFromClaims) } + + // 1126 (ADR 0547 §2, ADR 0549 §3): once the catalogue places the nodes, the check composes over the input the nodes + // compose from (frombus), its catalogue left to the cases, and the seats held as placement says. + m := digestedMesh() + m.Input = &compose.Input{Catalogue: map[string]catalogue.Manifest{"web": {Module: "web"}}, + Nodes: []compose.Node{{Name: "g14", Assigned: []string{"web"}, Place: compose.Place{Endpoint: "g14.example:51820"}}}, + Holdings: []catalogue.Held{{Claim: "web-seat", Scope: "mesh", Node: "g14", Module: "web"}}} + placedJob, err := HeadJob(m, c) + if err != nil { + t.Fatal(err) + } + placed, err := composer.Decode(placedJob.Request) + if err != nil { + t.Fatal(err) + } + if placed.Input.Catalogue != nil || len(placed.Input.Nodes) != 1 || placed.Input.Nodes[0].Place.Endpoint != "g14.example:51820" || + len(placed.Input.Holdings) != 1 || placed.HoldingsFromClaims { + t.Errorf("the placed check's request = %+v, holdings from claims %v", placed.Input, placed.HoldingsFromClaims) + } } // 1622/AC2, AC3: a catalogue member using a field only the SDK head reads passes with the head's composer, and the