Assigner: the pass keeps an assignment graph revision per catalog revision taken in, first

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
2026-11-11 23:04:14 +01:00
parent c2a3f3553f
commit 5d33692d7f
6 changed files with 260 additions and 0 deletions
@@ -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
+}
@@ -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)
+	}
+}
@@ -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
+}
@@ -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
@@ -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
+}
@@ -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)