From 5d33692d7f1dbc67b28bb7f35a3f72a75be4afe2 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 23:04:14 +0100 Subject: [PATCH] Assigner: the pass keeps an assignment graph revision per catalog revision taken in, first MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pass computes the assignment graph revision (assigngraph.Compute) from the catalog halves it follows, the placements, the first branch of each registration and each node's capability reading, keeps it in the objects assignment-revisions through a bus adapter modelled on the indexer's, says assignment-revision, and holds the kept revision and its catalog source for the wants. Builds: design 47 G A-67, §292, G-5 "the wants follow the trunk's assignment graph" Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-assigner/graphkept.go | 97 +++++++++++++++++++ .../mesh-delivery-assigner/graphkept_test.go | 54 +++++++++++ .../cmd/mesh-delivery-assigner/halves.go | 13 +++ .../cmd/mesh-delivery-assigner/main.go | 4 + .../cmd/mesh-delivery-assigner/objects.go | 89 +++++++++++++++++ .../cmd/mesh-delivery-assigner/pass.go | 3 + 6 files changed, 260 insertions(+) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go new file mode 100644 index 00000000..87f1c6c4 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept.go @@ -0,0 +1,97 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "slices" + + mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + + "mesh-delivery-assigner/internal/assigngraph" +) + +// graphKeeper is the pass's assignment graph (design 47 G §292, A-67): per catalog revision taken in it computes the revision +// from the catalog halves it follows, the placements, the first branch of each registration and each node's capability reading, +// keeps it in the objects assignment-revisions and says the seat mesh-wants' event assignment-revision; the wants of the pass +// that follows are written from the revision it holds. +type graphKeeper struct { + objects assigngraph.Objects + // halves is every followed branch's catalog graph revision held now. + halves func() map[string]mcc.Half + // last is the revision last computed, so a pass whose inputs did not change keeps and says nothing. + last meshwants.AssignmentRevision + has bool + // kept is the revision the wants follow, and kept's source the trunk's source revision it carries; ok is false until one is kept. + kept meshwants.AssignmentRevision + source int64 + ok bool + said string +} + +// keepGraph computes the revision of this pass and keeps it when the pass's inputs differ from the last computed. +func (p *passer) keepGraph(ctx context.Context) { + g := p.graph + if g == nil || g.objects == nil || g.halves == nil { + return + } + in := assigngraph.Inputs{Halves: g.halves(), Placements: p.w.assignments(), Capabilities: p.capabilityDigests()} + // With no pipeline ends kept, the wants are the trunk-only form (pass.go): a placement naming no branch runs the trunk's. + in.FirstBranch = func(string) string { return trunkBranch } + if p.ends != nil { + in.FirstBranch = func(module string) string { return p.ends.branchOf(module, "") } + } + c := assigngraph.Compute(in) + for _, u := range c.Uncomputed { + if g.said != u { + g.said = u + p.logf("the assignment graph leaves out %s", u) + } + } + if g.has && g.ok && assigngraph.Same(g.last, c.Revision) { + return + } + rev, fresh, err := assigngraph.Keep(ctx, g.objects, c, meshwants.EventAssignmentRevision, p.clock()) + if err != nil { + p.logf("the assignment graph revision is not kept, so no want is said as wanted until it is: %v", err) + return + } + g.last, g.has, g.kept, g.ok = c.Revision, true, rev, true + g.source = 0 + for _, cat := range rev.Catalogs { + if cat.Branch == trunkBranch { + g.source = cat.Source + } + } + if !fresh { + return + } + id := fmt.Sprintf("%s-%s-%d", meshwants.EventAssignmentRevision, rev.Kind, rev.Revision) + if err := p.wantsSay(meshwants.EventAssignmentRevision, assigngraph.Event(rev), id); err != nil { + p.logf("the assignment revision %d is kept but not said: %v", rev.Revision, err) + } +} + +// capabilityDigests is the digest of each node's capability reading, the statement its engine made, by node. +func (p *passer) capabilityDigests() []meshwants.AssignmentRevisionCapabilitiesItem { + if p.reads == nil { + return nil + } + known := p.w.knownNodes() + slices.Sort(known) + var out []meshwants.AssignmentRevisionCapabilitiesItem + for _, node := range known { + if !p.reads.Known(node) { + continue + } + reads := slices.Clone(p.reads.Reads([]string{node})[node]) + slices.Sort(reads) + raw, _ := json.Marshal(reads) + sum := sha256.Sum256(raw) + out = append(out, meshwants.AssignmentRevisionCapabilitiesItem{Node: node, Digest: "sha256:" + hex.EncodeToString(sum[:])}) + } + return out +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go new file mode 100644 index 00000000..49dec5d5 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/graphkept_test.go @@ -0,0 +1,54 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + + "mesh-delivery-assigner/internal/buckets" +) + +// said is one event the passer said on the seat mesh-wants. +type said struct { + event, id string + body json.RawMessage +} + +// Design 47 G A-67, §292: the pass keeps an assignment graph revision per catalog revision taken in, before it writes any want, and +// says assignment-revision once; the same inputs keep and say nothing more. +func TestThePassKeepsAnAssignmentGraphRevisionPerCatalogRevision(t *testing.T) { + ctx := context.Background() + at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + kv := &memKV{values: map[string][]byte{}} + p := newPasser(kvSet{buckets.Wants: kv}, at) + p.logf = t.Logf + var events []said + p.publishWants = func(event string, body any, id string) error { + raw, _ := json.Marshal(body) + events = append(events, said{event, id, raw}) + return nil + } + placeOn(t, p.w, "ace", "web") + half := mcc.Half{Revision: meshcatalog.CatalogRevision{Branch: trunkBranch, Kind: "trunk", Revision: 4, Source: 9, + Entries: []meshcatalog.EntryRef{{Module: "web", Digest: "sha256:" + strings.Repeat("a", 64)}}}, + Entries: map[string]meshcatalog.CatalogEntry{"web": {Module: "web", Fp: fpTest}}} + objects := memObjects{} + p.graph = &graphKeeper{objects: objects, halves: func() map[string]mcc.Half { return map[string]mcc.Half{trunkBranch: half} }} + p.keepGraph(ctx) + if len(events) != 1 || events[0].event != meshwants.EventAssignmentRevision || len(objects) == 0 { + t.Fatalf("the pass kept %d objects and said %+v", len(objects), events) + } + p.keepGraph(ctx) // the same inputs: nothing kept, nothing said + if len(events) != 1 { + t.Fatalf("a pass of the same inputs said %d events", len(events)) + } + if g := p.graph; !g.ok || g.kept.Revision != 1 || g.source != 9 { + t.Errorf("kept %+v, source %d", g.kept, g.source) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go index b57d1575..40fd4ee2 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go @@ -110,3 +110,16 @@ func fromEntry(e meshcatalog.CatalogEntry) (meshcatalog.ModulesByModuleValue, gr n.Serves = graph.ServesOf(man) return v, n, &parsed{module: e.Module, manifest: man} } + +// views is the catalog graph revision held now of each followed branch whose first view is answered. +func (h *halves) views() map[string]mcc.Half { + h.mu.Lock() + defer h.mu.Unlock() + out := make(map[string]mcc.Half, len(h.cats)) + for branch, c := range h.cats { + if half, ok := c.View(); ok { + out[branch] = half + } + } + return out +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 6e324839..ca6e6d0e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -12,6 +12,7 @@ import ( "io" "log" "maps" + "mesh-delivery-assigner/internal/assigngraph" "mesh-delivery-assigner/internal/assignment" "mesh-delivery-assigner/internal/buckets" "os" @@ -542,6 +543,9 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } if js != nil { p.publishWants = wantsPublisher(ctx, js, thisNode) + if iv != nil && iv.h != nil { + p.graph = &graphKeeper{objects: newBusObjects(js, process.Account{Node: thisNode, Module: assignment.Module}, assigngraph.RevisionsObjects), halves: iv.h.views} + } } // The node-engine seat's manifest-reads-changed, for one node (ADR 0618 §4): that node is asked again, with no timer. // Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go new file mode 100644 index 00000000..9844df61 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/objects.go @@ -0,0 +1,89 @@ +package main + +import ( + "bytes" + "context" + "errors" + "fmt" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-assigner/internal/assigngraph" +) + +// busObjects is the assigner's own object bucket assignment-revisions in the old `objects` section (novox/hq ADR 0619 §2), +// which the grants part creates, as the passer's process reaches it: opened at its first use and kept; until the bucket exists +// every call is an error saying so, and the next call opens it again (design 47 G A-67, modelled on the indexer's busObjects). +type busObjects struct { + js jetstream.JetStream + module string + local string + mu sync.Mutex + o jetstream.ObjectStore +} + +// newBusObjects is the object bucket of this local name of the module the account runs as. +func newBusObjects(js jetstream.JetStream, cred process.Account, local string) *busObjects { + return &busObjects{js: js, module: cred.Module, local: local} +} + +func (b *busObjects) open(ctx context.Context) (jetstream.ObjectStore, error) { + b.mu.Lock() + defer b.mu.Unlock() + if b.o != nil { + return b.o, nil + } + o, err := b.js.ObjectStore(ctx, catalogue.ObjectsBucket(b.module, b.local)) + if errors.Is(err, jetstream.ErrBucketNotFound) || errors.Is(err, jetstream.ErrStreamNotFound) { + return nil, fmt.Errorf("the object bucket %s is not on the bus yet (the grants part creates it from the manifest's objects): %w", b.local, process.ErrNotYet) + } + if err != nil { + return nil, fmt.Errorf("the object bucket %s is not open: %w", b.local, err) + } + b.o = o + return o, nil +} + +func (b *busObjects) Put(ctx context.Context, name string, data []byte) error { + o, err := b.open(ctx) + if err != nil { + return err + } + _, err = o.Put(ctx, jetstream.ObjectMeta{Name: name}, bytes.NewReader(data)) + return err +} + +func (b *busObjects) Get(ctx context.Context, name string) ([]byte, error) { + o, err := b.open(ctx) + if err != nil { + return nil, err + } + data, err := o.GetBytes(ctx, name) + if errors.Is(err, jetstream.ErrObjectNotFound) { + return nil, assigngraph.ErrNoObject + } + return data, err +} + +func (b *busObjects) Names(ctx context.Context) ([]string, error) { + o, err := b.open(ctx) + if err != nil { + return nil, err + } + infos, err := o.List(ctx) + if errors.Is(err, jetstream.ErrNoObjectsFound) { + return []string{}, nil + } + if err != nil { + return nil, err + } + names := make([]string, 0, len(infos)) + for _, i := range infos { + names = append(names, i.Name) + } + return names, nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index bad14a3f..f122cfe1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -101,6 +101,8 @@ type passer struct { // wait-held and wait-ended. Nil: no wait is made. outside *outsideActs publishWants publisher + // graph keeps the assignment graph revision the wants follow (graphkept.go, design 47 G §292, A-67); nil: none is kept. + graph *graphKeeper } // wantsSay says one event of the seat mesh-wants; with no bus to say it on, nothing is said. @@ -202,6 +204,7 @@ func (p *passer) clock() time.Time { func (p *passer) pass(ctx context.Context) { p.assignmentsSeat(ctx) + p.keepGraph(ctx) p.wants(ctx) p.puts(ctx) p.nodeWants(ctx)