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)