diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go new file mode 100644 index 00000000..07829862 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go @@ -0,0 +1,147 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "slices" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + 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/sdk/graph" + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// halves follows the catalog half of each watched branch through the seat mesh-catalog (design 47 G §14, §184, §195): one +// meshcatalogclient.Catalog per branch, a first view of the verb `revision` and then each `catalog-revision` of that branch, +// each entry taken into the world as the value and the graph node the rules read. Of the view the assigner reads only the catalog +// half, never its own output back (§195). +type halves struct { + w *world + call mcc.Call + wake func() + logf func(string, ...any) + ctx context.Context + + mu sync.Mutex + cats map[string]*mcc.Catalog + // started is called with each catalog made, for the part that reads their status. + started func(*mcc.Catalog) + // settled is told once when a branch's first view is answered, for the first pass to wait on (the trunk's alone). + settled chan<- struct{} +} + +// follow makes the catalog of branch and asks its first view, once; true when it was made now. +func (h *halves) follow(branch string) bool { + h.mu.Lock() + if h.cats[branch] != nil { + h.mu.Unlock() + return false + } + c, err := mcc.NewCatalog(h.call, branch) + if err != nil { + h.mu.Unlock() + h.logf("the catalog half of branch %q is not followed: %v", branch, err) + return false + } + c.Follower.Logf = h.logf + c.Follower.Changed = func(module string, entry json.RawMessage, deleted bool, _ time.Time) { + h.changed(branch, module, entry, deleted) + } + if h.cats == nil { + h.cats = map[string]*mcc.Catalog{} + } + h.cats[branch] = c + h.mu.Unlock() + if h.started != nil { + h.started(c) + } + go func() { + if err := c.Start(h.ctx); err != nil { + if h.ctx.Err() == nil { + h.logf("the first view of the catalog half of branch %q is not answered: %v", branch, err) + } + return + } + if branch == trunkBranch && h.settled != nil { + h.settled <- struct{}{} + } + }() + return true +} + +// hear gives a catalog-revision to every catalog: each takes its own branch's and passes over another's. +func (h *halves) hear(m jetstream.Msg) { + h.mu.Lock() + cats := make([]*mcc.Catalog, 0, len(h.cats)) + for _, c := range h.cats { + cats = append(cats, c) + } + h.mu.Unlock() + for _, c := range cats { + if err := c.Hear(h.ctx, m.Data()); err != nil { + h.logf("%s on branch %q: %v", m.Subject(), c.Branch, err) + } + } + _ = m.Ack() +} + +// followers are the first views the catalogs hold. +func (h *halves) followers() []*mcc.Catalog { + h.mu.Lock() + defer h.mu.Unlock() + out := make([]*mcc.Catalog, 0, len(h.cats)) + for _, c := range h.cats { + out = append(out, c) + } + return out +} + +// changed takes one entry of branch's half into the world, or a module taken out of it. +func (h *halves) changed(branch, module string, raw json.RawMessage, deleted bool) { + defer h.wake() + if deleted { + h.w.dropModule(branch, module) + return + } + var e meshcatalog.CatalogEntry + if err := json.Unmarshal(raw, &e); err != nil { + h.logf("the entry of %s on branch %q does not read: %v", module, branch, err) + return + } + v, n, m := fromEntry(e) + h.w.putModule(branch, module, v, n, m) +} + +// fromEntry is a catalog entry as the value and the graph node the rules read, and its manifest parsed once under the digest of +// its words. +func fromEntry(e meshcatalog.CatalogEntry) (meshcatalog.ModulesByModuleValue, graph.Node, *parsed) { + v := meshcatalog.ModulesByModuleValue{Commit: e.Commit, Fp2: e.Fp, Needs: e.Needs, None2: e.None, Packages: e.Packages, + Path: e.Path, Repository: e.Repository, Verdict: e.Verdict} + n := graph.Node{FP: e.Fp, None: e.None, Repository: e.Repository, Path: e.Path, Bases: e.Bases, Provides: e.Provides, Requires: e.Requires, + Awaits: e.Awaits, Invokes: e.Invokes, Serves: e.Serves, LeftOut: e.LeftOut} + for p := range e.Packages { + n.Packages = append(n.Packages, p) + } + slices.Sort(n.Packages) + if len(e.Manifest) == 0 { + return v, n, nil + } + raw, err := json.Marshal(e.Manifest) + if err != nil { + return v, n, nil + } + sum := sha256.Sum256(raw) + v.Manifest = "sha256:" + hex.EncodeToString(sum[:]) + man, err := catalogue.Decode(raw) + if err != nil { + return v, n, &parsed{module: e.Module, err: "it does not decode: " + err.Error()} + } + n.Serves = graph.ServesOf(man) + return v, n, &parsed{module: e.Module, manifest: man} +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 0ce2b023..e10e9312 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + "git.novox.be/novox/mesh-sdk/go/keys" "io" "maps" "mesh-delivery-assigner/internal/assignment" @@ -52,7 +53,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" - "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-assigner/internal/assign" @@ -71,6 +71,17 @@ type bed struct { placed map[string][]string // journal is what the programs the bed starts write in their journal. journal *journal + // cat is the catalog half the seat mesh-catalog serves the assigner (serveCatalog). + cat *catalogBed +} + +// catalogBed is the catalog half of each watched branch as the seat mesh-catalog keeps it: entries by branch and module, a +// revision per branch, and the manifest bodies by digest. +type catalogBed struct { + mu sync.Mutex + entries map[string]map[string]meshcatalog.CatalogEntry + rev map[string]int64 + bodies map[string]string } // journal is a program's journal as the bed hears it: a test awaits a line in it (await) where the line is the only sign @@ -144,7 +155,8 @@ func startBed(t *testing.T, leaveOut string) *bed { } ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) t.Cleanup(cancel) - b := &bed{srv: srv, js: js, ctx: ctx, dir: dir, buckets: map[string]jetstream.KeyValue{}, journal: &journal{grew: make(chan struct{})}} + b := &bed{srv: srv, js: js, ctx: ctx, dir: dir, buckets: map[string]jetstream.KeyValue{}, journal: &journal{grew: make(chan struct{})}, + cat: &catalogBed{entries: map[string]map[string]meshcatalog.CatalogEntry{}, rev: map[string]int64{}, bodies: map[string]string{}}} for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) { if name == leaveOut { continue @@ -171,7 +183,7 @@ func startBed(t *testing.T, leaveOut string) *bed { // event names the engine's machine last, so the filter is the event with every machine (mesh-grants buscompose // machineEvent; the grants part's own test holds the derivation). if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", - FilterSubjects: []string{enginereads.SubjectPrefix + "*", "mesh.mod." + buckets.IndexerModule + ".event.views.>"}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + FilterSubjects: []string{enginereads.SubjectPrefix + "*", "mesh.mod." + buckets.IndexerModule + ".event.views.>", meshcatalog.SubjectCatalogRevisionAll}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } // The operator-channel seat's stream of conditions, which the assigner says its conditions on. @@ -180,6 +192,8 @@ func startBed(t *testing.T, leaveOut string) *bed { } // The indexer as the owner of its manifests and modules (DLV-F2r), with the real owner's side. serveIndexer(t, b, nc) + // The seat mesh-catalog's catalog half (design 47 G §194), which the assigner follows per branch. + serveCatalog(t, b, nc) // The assigner's views family is its ordinary events on EVENTS (ADR 0635 §5); its marks are its state. if kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: marksBucket, History: 1}); err != nil { t.Fatal(err) @@ -422,6 +436,7 @@ func (b *bed) manifest(t *testing.T, module, body string) string { t.Fatal(err) } b.putJSON(t, buckets.Manifests, key, value) + b.cat.bodies[digest] = body return digest } @@ -447,11 +462,11 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { // DLV-F5r (ADR 0621 §3): the indexer's runtime starts together with the assigner and answers its first views only // after the assigner asked; within the bound the assigner writes the wants from what they answer. up := make(chan struct{}) - indexerLate = up + catalogLate = up // ADR 0631 §2: while the indexer does not answer, past the first-view bound the assigner raises the condition naming // the indexer and graph, and reads nothing of the indexer's state; the answer clears it. firstViewBound, firstViewWait = 300*time.Millisecond, 50*time.Millisecond - t.Cleanup(func() { indexerLate, firstViewBound, firstViewWait = nil, 0, 0 }) + t.Cleanup(func() { catalogLate, firstViewBound, firstViewWait = nil, 0, 0 }) for _, name := range append(append([]string{}, readBuckets...), watched...) { if strings.HasPrefix(name, buckets.IndexerModule+"_") { t.Errorf("the assigner reads the indexer's state %s itself", name) @@ -461,7 +476,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { fNew, fOld := meshcatalogclient.SourcePrefix2+strings.Repeat("b", 64), meshcatalogclient.SourcePrefix2+strings.Repeat("a", 64) digest := b.manifest(t, "web", `{"module":"web","version":"1"}`) mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fNew, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - b.putJSON(t, buckets.Modules, "web", mod) + b.putModule(t, "web", mod) built, err := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest}) if err != nil { t.Fatal(err) @@ -493,20 +508,20 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { "seat-holders": {"n1.node-tools"}, "wants": {"n1.mesh-grants"}, }) - graphKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, assignment.Module+"."+buckets.IndexerModule+"."+graph.State, views.UnansweredKind) + graphKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, assignment.Module+"."+meshcatalog.Seat+"."+views.BranchState(meshcatalog.StateCatalogRevisions, trunkBranch), views.UnansweredKind) graphRaised := func() bool { e, err := b.buckets[assign.FindingsBucket].Get(b.ctx, graphKey) - return err == nil && strings.Contains(string(e.Value()), buckets.IndexerModule+".graph") + return err == nil && strings.Contains(string(e.Value()), "mesh-catalog.revision") } for deadline := time.Now().Add(10 * time.Second); !graphRaised(); time.Sleep(50 * time.Millisecond) { if time.Now().After(deadline) { - t.Fatalf("with the indexer down, no condition names its graph (finding %s)", graphKey) + t.Fatalf("with the seat mesh-catalog down, no condition names its revision verb (finding %s)", graphKey) } } close(up) for deadline := time.Now().Add(10 * time.Second); graphRaised(); time.Sleep(50 * time.Millisecond) { if time.Now().After(deadline) { - t.Fatalf("the indexer answered, and the condition naming its graph stands (finding %s)", graphKey) + t.Fatalf("the seat answered, and the condition naming its revision verb stands (finding %s)", graphKey) } } @@ -608,7 +623,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { fp, digest string }{"db": {fpOf('2'), dbDigest}, "app": {fpOf('3'), appDigest}, "tool": {fpOf('4'), toolDigest}} { mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: row.fp, Manifest: row.digest, Repository: "novox/" + module, Commit: strings.Repeat("2", 40), Path: meshcatalogclient.RootPath}) - b.putJSON(t, buckets.Modules, module, mod) + b.putModule(t, module, mod) fps := []string{row.fp} if module == "tool" { fps = append(fps, toolOld) @@ -680,7 +695,7 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { formsDigest := b.manifest(t, "forms", `{"module":"forms","version":"1"}`) formsAt := func(fp string) { mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: formsDigest, Repository: "novox/forms", Commit: strings.Repeat("3", 40), Path: meshcatalogclient.RootPath}) - b.putJSON(t, buckets.Modules, "forms", mod) + b.putModule(t, "forms", mod) } for _, fp := range []string{formsOld, formsNew} { build := builds.BuilderBuild{State: builds.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: formsDigest} @@ -802,7 +817,7 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) { fNew := meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64) digest := b.manifest(t, "web", `{"module":"web","version":"1"}`) mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fNew, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - b.putJSON(t, buckets.Modules, "web", mod) + b.putModule(t, "web", mod) on, _ := presenceValue(true) b.putJSON(t, buckets.Presence, "ace", on) failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fNew, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed, Detail: "connection refused"}) @@ -860,7 +875,7 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) { fNew := meshcatalogclient.SourcePrefix2 + strings.Repeat("d", 64) digest := b.manifest(t, "mail", `{"module":"mail","version":"1","resources":[{"id":"data","type":"directory","path":"/var/lib/mail"}],"data":{"own":[{"id":"mail","path":"${dir:data}","class":"irreplaceable","backup":"copy"}]}}`) mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fNew, Manifest: digest, Repository: "novox/mail", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - b.putJSON(t, buckets.Modules, "mail", mod) + b.putModule(t, "mail", mod) on, _ := presenceValue(true) b.putJSON(t, buckets.Presence, "ace", on) failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fNew, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed, Detail: "connection refused"}) @@ -910,7 +925,7 @@ func TestTheAssignerReplacedResumesWithNothingLostOrRepeated(t *testing.T) { fp := meshcatalogclient.SourcePrefix2 + strings.Repeat(string(rune('a'+i%6)), 63) + fmt.Sprintf("%x", i%16) digest := b.manifest(t, name, fmt.Sprintf(`{"module":%q,"version":"1"}`, name)) mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/" + name, Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - b.putJSON(t, buckets.Modules, name, mod) + b.putModule(t, name, mod) built, err := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest}) if err != nil { t.Fatal(err) @@ -956,7 +971,7 @@ func (b *bed) fixture(t *testing.T, name, fp, manifest string) { t.Helper() digest := b.manifest(t, name, manifest) mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/" + name, Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - b.putJSON(t, buckets.Modules, name, mod) + b.putModule(t, name, mod) built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest}) bkey, _ := builds.BuilderBuildKey(name, fp) b.putJSON(t, buckets.Builds, bkey, built) @@ -1305,9 +1320,9 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) { rec, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: node}}) b.putJSON(t, buckets.Have, node+".node", rec) } - b.putJSON(t, buckets.Modules, "web", []byte(`{"v":1}`)) - b.putJSON(t, buckets.Modules, "shell", []byte(`{"v":1}`)) - b.putJSON(t, buckets.Modules, "mesh-enrolment", []byte(`{"v":1}`)) + b.putModule(t, "web", []byte(`{"v":1}`)) + b.putModule(t, "shell", []byte(`{"v":1}`)) + b.putModule(t, "mesh-enrolment", []byte(`{"v":1}`)) // place writes a node's assignments as another writer would, in its node and module keys (research 072 row 37), and // answers its node key's revision. place := func(node string, modules ...string) uint64 { @@ -1611,9 +1626,9 @@ func assignmentsSubjects() []string { return out } -// indexerLate, when set before the bed is made, fails every ask of the indexer's first views until it is closed: the +// catalogLate, when set before the bed is made, fails every ask of the indexer's first views until it is closed: the // indexer down, its runtime coming up after the assigner asked (DLV-F5r, ADR 0631 §2). -var indexerLate chan struct{} +var catalogLate chan struct{} // serveIndexer is the indexer as the owner of its manifests, modules and module graph (ADR 0615, ADR 0621, ADR 0631): its // buckets, which a test fills as the indexer writes them; its views family said by the views emitter as ordinary events on EVENTS, @@ -1635,7 +1650,7 @@ func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) { if err := em.Start(b.ctx, state, kv, nil); err != nil { t.Fatal(err) } - late := indexerLate + late := catalogLate sub, err := nc.Subscribe("mesh.mod.mesh-delivery-indexer.call."+state+".>", func(m *nats.Msg) { if late != nil { // its runtime comes up after the assigner started (DLV-F5r): until then every ask fails, as on a down indexer select { @@ -1797,3 +1812,82 @@ func changeIn(body map[string]any) plc.Change { _ = json.Unmarshal(raw, &c) return c } + +// indexerFirstViews are the indexer's states the bed still serves (the verbs' check of a proposal asks the indexer's modules and +// manifests): the bucket each is kept in. +var indexerFirstViews = map[string]string{"manifests": buckets.Manifests, "modules": buckets.Modules} + +// putModule keeps a module's value as the indexer does for the verbs, and as an entry of the trunk's catalog half; a value that +// names no manifest the bed holds is kept for the verbs alone. +func (b *bed) putModule(t *testing.T, module string, value []byte) { + t.Helper() + b.putJSON(t, buckets.Modules, module, value) + v, err := meshcatalogclient.DecodeModule(value) + if err != nil { // a value the bed writes only for the module to exist + v = meshcatalog.ModulesByModuleValue{Repository: "novox/" + module} + } + b.publish(t, trunkBranch, module, v) +} + +// publish makes the trunk's catalog half of branch name module at v, a new revision of it, and says catalog-revision. +func (b *bed) publish(t *testing.T, branch, module string, v meshcatalog.ModulesByModuleValue) { + t.Helper() + e := meshcatalog.CatalogEntry{Module: module, Fp: v.Fp2, Commit: v.Commit, Repository: v.Repository, Path: v.Path, Needs: v.Needs, Verdict: "accepted"} + if body, ok := b.cat.bodies[v.Manifest]; ok { + _ = json.Unmarshal([]byte(body), &e.Manifest) + } + b.cat.mu.Lock() + if b.cat.entries[branch] == nil { + b.cat.entries[branch] = map[string]meshcatalog.CatalogEntry{} + } + b.cat.entries[branch][module] = e + b.cat.rev[branch]++ + rev, n := b.cat.rev[branch], int64(len(b.cat.entries[branch])) + b.cat.mu.Unlock() + raw, _ := json.Marshal(meshcatalog.CatalogRevisionEvent{Kind: "trunk", Branch: keys.EscapeName(branch), Revision: rev, Entries: n, Changed: []string{module}}) + if _, err := b.js.Publish(b.ctx, meshcatalog.SubjectCatalogRevision(branch), raw); err != nil { + t.Fatal(err) + } +} + +// serveCatalog answers the seat mesh-catalog's verb revision from the bed's catalog half, all in one page; until catalogLate is +// closed every ask fails, as on a seat whose holder is down. +func serveCatalog(t *testing.T, b *bed, nc *nats.Conn) { + t.Helper() + late := catalogLate + answer := func(m *nats.Msg) { + if late != nil { + select { + case <-late: + default: + _ = m.Respond([]byte(`{"error":"the seat mesh-catalog's holder is not up"}`)) + return + } + } + var in meshcatalog.RevisionInput + _ = json.Unmarshal(m.Data, &in) + branch, _ := keys.UnescapeName(in.Branch) + b.cat.mu.Lock() + defer b.cat.mu.Unlock() + if b.cat.rev[branch] == 0 { + _ = m.Respond([]byte(`{"error":"` + views.NoneYet + `"}`)) + return + } + rev := meshcatalog.CatalogRevision{Kind: "trunk", Branch: in.Branch, Revision: b.cat.rev[branch]} + var entries []meshcatalog.CatalogEntry + for _, name := range slices.Sorted(maps.Keys(b.cat.entries[branch])) { + entries = append(entries, b.cat.entries[branch][name]) + rev.Entries = append(rev.Entries, meshcatalog.EntryRef{Module: name, Digest: "sha256:" + strings.Repeat("0", 64)}) + } + raw, _ := json.Marshal(map[string]any{"result": meshcatalog.RevisionAnswer{Revision: rev, Entries: entries}, "node": "n0"}) + _ = m.Respond(raw) + } + for _, subject := range []string{"mesh.seat.mesh-catalog.tool.revision", "mesh.seat.mesh-catalog.call.revision.*", "mesh.seat.mesh-catalog.tool.revision.*"} { + sub, err := nc.Subscribe(subject, answer) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + } + subscribed(t, nc) +} 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 72177123..e99492ea 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -248,7 +248,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro conn.Close() } }() - iv := &indexerViews{js: js, account: account, call: said.Call, findings: findings, logf: logger.Printf, + iv := &catalogViews{js: js, account: account, call: said.Call, findings: findings, logf: logger.Printf, status: process.AwaitBucket(ctx, js, conn, firstViewsBucket, "assigner_first_views reads nothing and assigner_first_view_again is not heard", logger.Printf)} return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, logger) } @@ -306,7 +306,7 @@ var watched = []string{ // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after // each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends. -func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs kvSet, say func(oc.Condition) error, findings assign.Findings, bus caller, iv *indexerViews, logger *log.Logger) error { +func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs kvSet, say func(oc.Condition) error, findings assign.Findings, bus caller, iv *catalogViews, logger *log.Logger) error { w := newWorld() dirty := make(chan struct{}, 1) wake := func() { @@ -494,6 +494,13 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, // The runner's pipeline-ended and group-ended (design 47 G §29): kept in the state written, which the wants follow // once the first registrations are written (§95, §353). p.ends = newPipelineEnds(kvEnds{ctx, kv}, wake, logger.Printf) + if iv != nil && iv.h != nil { + p.ends.onBranches = func(branches []string) { + for _, b := range branches { + iv.h.follow(b) + } + } + } err := hearEnds(hub, p.ends) if err == nil && bus != nil { err = followRegistrations(ctx, hub, bus.Call, p.ends) 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 ef0ccc4f..33382b2d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -8,6 +8,7 @@ import ( "maps" "mesh-delivery-assigner/internal/buckets" "slices" + "sort" "strings" "time" @@ -23,6 +24,7 @@ import ( oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" + "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" @@ -219,19 +221,21 @@ func (p *passer) nodeWants(ctx context.Context) { // the hold of a fingerprint whose health check failed. The assigner compares its wants with nothing the controller // sends: the differences bucket is kept read-only with its history (ADR 0539 §8). func (p *passer) wants(ctx context.Context) { - get := p.w.getter(buckets.Manifests) // the indexer's manifests, as its first view and views family say them assigned := p.w.assigned() + assignedBranches := p.w.placedBranches() + // Every module of every branch's catalog half, for the parts that ask which exist; a node's entries come from the half of the + // branch its placement names (design 47 G §14, §184). trunk := p.w.trunk() - manifests := map[string]catalogue.Manifest{} - for name, mod := range trunk { - if mod.Manifest == "" { - continue - } - if got := p.w.manifest(ctx, get, mod.Manifest); got.err == "" { - manifests[name] = got.manifest + views := map[string]*branchView{} + viewOf := func(branch string) *branchView { + if v := views[branch]; v != nil { + return v } + v := p.branchView(branch, assigned) + views[branch] = v + return v } - nodes := p.w.nodes(manifests, assigned) + nodes := viewOf(trunkBranch).nodes p.engineReads(nodes) nodesOf := map[string][]string{} for node, modules := range assigned { @@ -268,111 +272,118 @@ func (p *passer) wants(ctx context.Context) { defer func() { p.sayNeither(neither, stale, left) }() // Where grants are read from, once for the pass: the grants part's (issue 774, ADR 0571 clause 9). grants := p.w.grantsBucket() - // The verb edges, once for the pass (ADR 0614 §2, §3). - moduleGraph := p.moduleGraph() - served := p.served(ctx, get, moduleGraph, nodes) - offering := p.offering(ctx, get, manifests, nodes) // bus state (ADR 0614 §5) - assignedBranches := p.w.placedBranches() for name := range names { - fp := trunk[name].Fp2 - name := name - // A want follows a passed pipeline: a module whose fingerprint no pipeline passed at on every branch its placements - // chose, and whose wants do not name it already, gets none (design 47 G §29, §35). - branches := p.ends.branchesOf(name, assignedBranches) - if fp != "" && !p.ends.mayWant(name, fp, branches, wanted[name]) { - continue - } - m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Needs: trunk[name].Needs, Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Offered: offered(manifests), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], FormNeeds: p.w.buildNeeds(name, fp), - Self: func(node string) bool { return placed[node] }, - Composed: func(node, fp string) (assign.ComposedDigest, bool) { - return p.w.composed(node, name, fp) // a placed node's own: the digest its own have names (ADR 0571 §6) - }, - GrantedAt: func(fp string) bool { - key, err := grantsstate.PartGrantKey(name, fp) - return err == nil && p.w.has(grants, key) - }, - Restored: func(node string) bool { - key, err := asks.RestoreAskKey(node, name, fp) - return err == nil && p.answeredWith(key, assign.OptionRestore) - }, - Allowed: func(node, digest string) bool { - key, err := asks.RecordAskKey(node, name, digest) - return err == nil && p.answeredWith(key, assign.OptionYes) - }, - Release: func(fp string) (string, bool) { return p.w.release(name, fp) }} - if fp != "" { - held, _ := mwc.HeldKey(name, fp) - grant, _ := grantsstate.PartGrantKey(name, fp) - hold, _ := asks.HoldAskKey(name, fp) - m.Released = p.answeredWith(hold, "release") - m.Kept = p.answeredWith(hold, "keep") - m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(buckets.Wants, held) && !m.Released, p.w.has(grants, grant) - if raw := p.w.value(grants, grant); raw != nil { - if g, err := grantsstate.DecodeGrant(raw); err == nil { - m.GrantedOn = func(node string) bool { return assign.GrantNamesNode(g, node, name) } + groups := p.groups(name, nodesOf[name], assignedBranches[name]) + for gi, g := range groups { + func() { + bv := viewOf(g.branch) + trunk, manifests, nodes, moduleGraph, served, offering := bv.trunk, bv.manifests, bv.nodes, bv.graph, bv.served, bv.offering + fp := trunk[name].Fp2 + name := name + nodesOf := map[string][]string{name: g.nodes} + wantedOf := map[string]map[string]mwc.Want{name: wantedHere(wanted[name], g.nodes, groups, gi == 0)} + // A want follows a passed pipeline: a module whose fingerprint no pipeline passed at on the branch its placements + // chose, and whose wants do not name it already, gets none (design 47 G §29, §35). + var branches []string + if p.ends.following() { + branches = []string{g.branch} } - } - if m.Released { - p.clearHeld(name, fp) - } - } - if served != nil { - m.Invokes, m.Served = keptInvokes(moduleGraph, name), served - } - m.Offering = offering - last := map[string]string{} - for _, node := range m.Nodes { - last[node] = p.w.lastHealthy(node, name, fp) - } - out := assign.Wants(m, nodes, wanted[name], last) - for _, node := range m.Nodes { - for _, prov := range m.MissingProvisions(node) { - missing[node+"."+name+"."+prov] = true - } - } - p.yeses(ctx, name, m, wanted[name], &out) - gatherNeither(neither, name, out) - leftNeither(left, m, wanted[name], out) - p.sayNoWant(name, m.Nodes, wanted[name], out) - p.sayUnreached(name, m.Nodes, out) - p.sayNodeFailed(name, fp, m.Nodes, nodes, out, first) - if first { - startClears(stale, m, out) - if fp != "" && !m.Held && out.Held == nil && !out.Lifted { - p.clearHeld(name, fp) - } - } - if out.Held != nil { - key, err := mwc.HeldKey(name, fp) - if err == nil { - held := *out.Held - p.put(ctx, buckets.Wants, key, func() ([]byte, error) { return mwc.EncodeHeld(held) }) - } - } - if out.Lifted { - p.lift(ctx, name, fp) - } - if out.Held != nil || (m.Held && !out.Lifted) { - p.holdAsk(ctx, name, fp, out.Held) - p.heldCondition(ctx, name, fp, out.Held, len(out.Restore) > 0) - } - for _, node := range out.Restore { - p.restoreAsk(ctx, name, node, fp) - } - for _, need := range out.Record { - p.recordAsk(ctx, name, fp, need) - } - for node, want := range out.Wants { - key, err := mwc.WantKey(node, name) - if err != nil { - p.logf("no want of %s on %s is written: %v", name, node, err) - continue - } - want := want - p.put(ctx, buckets.Wants, key, func() ([]byte, error) { return mwc.EncodeWant(want) }) - } - if len(out.Wants) > 0 { - p.ends.wrote(name, fp, branches) + if fp != "" && !p.ends.mayWant(name, fp, branches, wantedOf[name]) { + return + } + m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Needs: trunk[name].Needs, Chosen: chosenOf(chosen, name), Provisions: provisionsOf(name), Answering: answering(manifests, nodes), Offered: offered(manifests), Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], FormNeeds: p.w.buildNeeds(name, fp), + Self: func(node string) bool { return placed[node] }, + Composed: func(node, fp string) (assign.ComposedDigest, bool) { + return p.w.composed(node, name, fp) // a placed node's own: the digest its own have names (ADR 0571 §6) + }, + GrantedAt: func(fp string) bool { + key, err := grantsstate.PartGrantKey(name, fp) + return err == nil && p.w.has(grants, key) + }, + Restored: func(node string) bool { + key, err := asks.RestoreAskKey(node, name, fp) + return err == nil && p.answeredWith(key, assign.OptionRestore) + }, + Allowed: func(node, digest string) bool { + key, err := asks.RecordAskKey(node, name, digest) + return err == nil && p.answeredWith(key, assign.OptionYes) + }, + Release: func(fp string) (string, bool) { return p.w.release(name, fp) }} + if fp != "" { + held, _ := mwc.HeldKey(name, fp) + grant, _ := grantsstate.PartGrantKey(name, fp) + hold, _ := asks.HoldAskKey(name, fp) + m.Released = p.answeredWith(hold, "release") + m.Kept = p.answeredWith(hold, "keep") + m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(buckets.Wants, held) && !m.Released, p.w.has(grants, grant) + if raw := p.w.value(grants, grant); raw != nil { + if g, err := grantsstate.DecodeGrant(raw); err == nil { + m.GrantedOn = func(node string) bool { return assign.GrantNamesNode(g, node, name) } + } + } + if m.Released { + p.clearHeld(name, fp) + } + } + if served != nil { + m.Invokes, m.Served = keptInvokes(moduleGraph, name), served + } + m.Offering = offering + last := map[string]string{} + for _, node := range m.Nodes { + last[node] = p.w.lastHealthy(node, name, fp) + } + out := assign.Wants(m, nodes, wantedOf[name], last) + for _, node := range m.Nodes { + for _, prov := range m.MissingProvisions(node) { + missing[node+"."+name+"."+prov] = true + } + } + p.yeses(ctx, name, m, wantedOf[name], &out) + gatherNeither(neither, name, out) + leftNeither(left, m, wantedOf[name], out) + p.sayNoWant(name, m.Nodes, wantedOf[name], out) + p.sayUnreached(name, m.Nodes, out) + p.sayNodeFailed(name, fp, m.Nodes, nodes, out, first) + if first { + startClears(stale, m, out) + if fp != "" && !m.Held && out.Held == nil && !out.Lifted { + p.clearHeld(name, fp) + } + } + if out.Held != nil { + key, err := mwc.HeldKey(name, fp) + if err == nil { + held := *out.Held + p.put(ctx, buckets.Wants, key, func() ([]byte, error) { return mwc.EncodeHeld(held) }) + } + } + if out.Lifted { + p.lift(ctx, name, fp) + } + if out.Held != nil || (m.Held && !out.Lifted) { + p.holdAsk(ctx, name, fp, out.Held) + p.heldCondition(ctx, name, fp, out.Held, len(out.Restore) > 0) + } + for _, node := range out.Restore { + p.restoreAsk(ctx, name, node, fp) + } + for _, need := range out.Record { + p.recordAsk(ctx, name, fp, need) + } + for node, want := range out.Wants { + key, err := mwc.WantKey(node, name) + if err != nil { + p.logf("no want of %s on %s is written: %v", name, node, err) + continue + } + want := want + p.put(ctx, buckets.Wants, key, func() ([]byte, error) { return mwc.EncodeWant(want) }) + } + if len(out.Wants) > 0 { + p.ends.wrote(name, fp, branches) + } + }() } } } @@ -1313,3 +1324,82 @@ func answering(manifests map[string]catalogue.Manifest, n assign.Nodes) func(str return out } } + +// branchView is what one pass reads of the catalog half of one watched branch (design 47 G §184, §195): its modules, their +// manifests, the nodes the rules read through them, the module graph and the two readings of what runs. +type branchView struct { + trunk map[string]meshcatalog.ModulesByModuleValue + manifests map[string]catalogue.Manifest + nodes assign.Nodes + graph graph.Graph + served func(string) bool + offering func(string) bool +} + +// branchView reads the catalog half of branch for this pass. +func (p *passer) branchView(branch string, assigned map[string][]string) *branchView { + v := &branchView{trunk: p.w.trunkOf(branch), manifests: map[string]catalogue.Manifest{}} + for name, mod := range v.trunk { + if mod.Manifest == "" { + continue + } + if got := p.w.manifest(mod.Manifest); got.err == "" { + v.manifests[name] = got.manifest + } + } + v.nodes = p.w.nodes(v.manifests, assigned) + v.graph = p.moduleGraph(branch) + v.served = p.served(v.graph, v.nodes) + v.offering = p.offering(v.manifests, v.nodes) // bus state (ADR 0614 §5) + return v +} + +// group is the nodes of one module whose placements name one watched branch. +type group struct { + branch string + nodes []string +} + +// groups splits the nodes of module by the branch their placements name, the branches in name order (the trunk's first); a +// module placed nowhere is one group on its default branch. +func (p *passer) groups(module string, nodes []string, placed map[string]string) []group { + by := map[string][]string{} + for _, node := range nodes { + b := trunkBranch + if p.ends.following() { + b = p.ends.branchOf(module, placed[node]) + } + by[b] = append(by[b], node) + } + if len(by) == 0 { + b := trunkBranch + if p.ends.following() { + b = p.ends.branchOf(module, "") + } + by[b] = nil + } + var out []group + for b, ns := range by { + out = append(out, group{b, ns}) + } + sort.Slice(out, func(i, j int) bool { return out[i].branch < out[j].branch }) + return out +} + +// wantedHere is the wants of a module a group answers for: those of its own nodes, and (the first group only) those of nodes no +// group holds, which the rules take away. +func wantedHere(wanted map[string]mwc.Want, own []string, groups []group, first bool) map[string]mwc.Want { + held := map[string]bool{} + for _, g := range groups { + for _, n := range g.nodes { + held[n] = true + } + } + out := map[string]mwc.Want{} + for node, w := range wanted { + if slices.Contains(own, node) || (first && !held[node]) { + out[node] = w + } + } + return out +} 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 716ac50c..93654d0d 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 @@ -558,7 +558,7 @@ func putOfTheBus(t *testing.T) { placeOn(t, p.w, "novox", "nats") mdigest := strings.Repeat("e", 64) mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: mdigest, Repository: "novox/mesh-catalog", Commit: strings.Repeat("c", 40)}) - world(buckets.Modules, "nats", mod) + putModuleRaw(p.w, trunkBranch, "nats", mod) p.w.manifests[mdigest] = parsed{module: "nats", manifest: catalogue.Manifest{Module: "nats", Provides: []catalogue.Offer{{Name: "mesh-bus"}}}} bkey, _ := builds.BuilderBuildKey("nats", fp) built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest}) @@ -901,7 +901,7 @@ func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) { var said []oc.Condition p.say = func(c oc.Condition) error { said = append(said, c); return nil } mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - p.w.put(buckets.Modules, "web", mod) + putModuleRaw(p.w, trunkBranch, "web", mod) placeOn(t, p.w, "g14", "web") on, _ := presenceValue(true) p.w.put(buckets.Presence, "g14", on) @@ -939,7 +939,7 @@ func testAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) { said := &[]oc.Condition{} p.say = func(c oc.Condition) error { *said = append(*said, c); return nil } mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/resolved", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - p.w.put(buckets.Modules, "systemd-resolved", mod) + putModuleRaw(p.w, trunkBranch, "systemd-resolved", mod) placeOn(t, p.w, "shanks", "systemd-resolved") on, _ := presenceValue(true) p.w.put(buckets.Presence, "shanks", on) @@ -1198,7 +1198,7 @@ func aReleasesApproveAnswersTheMoveOfItsBuildAlone(t *testing.T) { p := newPasser(kvSet{buckets.Asks: asksKV}, time.Now()) p.logf = t.Logf mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: commit, Src: 7, FpSrc: c.fpSrc, Path: meshcatalogclient.RootPath}) - p.w.put(buckets.Modules, "web", mod) + putModuleRaw(p.w, trunkBranch, "web", mod) src, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/web", Heads: map[string]string{"novox/web": commit}}) p.w.putEntry(buckets.Source, mdc.SourceKey(trunkBranch), src, 7, time.Now()) later, _ := mdc.EncodeSource(mdseat.SourceByBranchValue{Group: "feat/other", Heads: map[string]string{"novox/web": commit, "novox/x": commit}}) @@ -1261,7 +1261,7 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) { p.logf = t.Logf put := func(bucket, key string, v []byte) { p.w.put(bucket, key, v) } mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - put(buckets.Modules, "web", mod) // the trunk's build is not done: rule 1 holds g14 at fpOld + putModuleRaw(p.w, trunkBranch, "web", mod) // the trunk's build is not done: rule 1 holds g14 at fpOld placeOn(t, p.w, "g14", "web") pl, _ := mmac.EncodeAssignments(mma.Assignments{Node: "g14", Modules: []mma.PlacedModule{{Module: "web"}}}) put(plc.AssignmentsBucket, "assignments.g14", pl) @@ -1380,7 +1380,7 @@ func TestAnApprovedAssignmentsCallsNoControllerVerb(t *testing.T) { func aWantWaitsOnTheVerbEdgesTheGraphKeeps(t *testing.T) { p := newPasser(kvSet{}, time.Now()) fp := func(c string) string { return meshcatalogclient.SourcePrefix2 + strings.Repeat(c, 64) } - if p.served(context.Background(), nil, graph.Graph{}, assign.Nodes{}) != nil { + if p.served(graph.Graph{}, assign.Nodes{}) != nil { t.Error("with no graph read, a want waits on a verb") } for module, n := range map[string]graph.Node{ @@ -1394,26 +1394,25 @@ func aWantWaitsOnTheVerbEdgesTheGraphKeeps(t *testing.T) { if err != nil { t.Fatal(err) } - p.w.put(buckets.Graph, module, raw) + _ = raw + p.w.putModule(trunkBranch, module, meshcatalog.ModulesByModuleValue{}, n, nil) } - g := p.moduleGraph() + g := p.moduleGraph(trunkBranch) if got := keptInvokes(g, "caller"); !slices.Equal(got, []string{"seat:x.v"}) || keptInvokes(g, "unknown") != nil { t.Errorf("the kept verb edges of caller = %v", got) } digest := strings.Repeat("d", 64) - man, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "server", Manifest: `{"module":"server","claims":[{"name":"x","scope":"mesh","serves":["w"]}]}`}) - get := func(_ context.Context, key string) ([]byte, error) { - if key == digest { - return man, nil - } - return nil, errNotThere + parsedServer, err := catalogue.Decode([]byte(`{"module":"server","claims":[{"name":"x","scope":"mesh","serves":["w"]}]}`)) + if err != nil { + t.Fatal(err) } + p.w.manifests[digest] = parsed{module: "server", manifest: parsedServer} build, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: digest}) key, _ := builds.BuilderBuildKey("server", fp("f")) p.w.put(buckets.Builds, key, build) running := nodeengine.HaveByNodeModuleValue{Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK} at := func(f string) nodeengine.HaveByNodeModuleValue { h := running; h.Fp = f; return h } - served := p.served(context.Background(), get, g, assign.Nodes{Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{ + served := p.served(g, assign.Nodes{Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{ "g14": {"server": at(fp("f")), "late": at(fp("2"))}}}) for v, want := range map[string]bool{"seat:x.w": true, "seat:x.v": false, "late.t": true} { if served(v) != want { @@ -1424,7 +1423,7 @@ func aWantWaitsOnTheVerbEdgesTheGraphKeeps(t *testing.T) { // offline node offers nothing, and one whose manifest is not read counts its trunk manifest's offer. bus := map[string]catalogue.Manifest{"bus": {Module: "bus", Provides: []catalogue.Offer{{Name: graph.BusState, Scope: catalogue.ScopeMesh}}}} for _, online := range []bool{false, true} { - offering := p.offering(context.Background(), get, bus, assign.Nodes{Online: map[string]bool{"n2": online}, + offering := p.offering(bus, assign.Nodes{Online: map[string]bool{"n2": online}, Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{"n2": {"bus": at(fp("2"))}}}) if offering(graph.BusState) != online { t.Errorf("with the bus module's node online %v, bus-state offered %v", online, offering(graph.BusState)) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go index 5e9ab67e..efddeb7a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines.go @@ -151,6 +151,8 @@ type pipelineEnds struct { // firsts is, by module, the first watched branch of its registration, the branch of a placement that names none (design // 47 G §14). firsts map[string]string + // onBranches is told the branches a registration names, for the catalog half of each to be followed (design 47 G §14, §195). + onBranches func([]string) } func newPipelineEnds(st endStore, wake func(), logf func(string, ...any)) *pipelineEnds { @@ -379,6 +381,9 @@ func (e *pipelineEnds) registered(reg mpcat.Registration, gone bool) { e.logf("the end %s is not deleted: %v", k, err) } } + if e.onBranches != nil && !gone { + e.onBranches(reg.Branches) + } if !e.on.Swap(true) { e.logf("the first registration is heard (%s): the wants follow the pipelines' ends from now", reg.Module) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go index acb0c207..db115598 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pipelines_test.go @@ -3,6 +3,11 @@ package main import ( "context" "encoding/json" + "fmt" + mmac "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" + mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" + plc "mesh-delivery-assigner/internal/assignment" + "slices" "strings" "testing" "time" @@ -147,7 +152,7 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) { p.say = func(oc.Condition) error { return nil } mdigest := strings.Repeat("e", 64) mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: mdigest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) - p.w.put(buckets.Modules, "web", mod) + putModuleRaw(p.w, trunkBranch, "web", mod) p.w.manifests[mdigest] = parsed{module: "web", manifest: catalogue.Manifest{Module: "web"}} bkey, _ := builds.BuilderBuildKey("web", fp) built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest}) @@ -211,3 +216,101 @@ func TestTheWantsSwitchOnTheFirstRegistration(t *testing.T) { t.Errorf("a registration gone left %d ends", len(st.vals)) } } + +// Design 47 G §14, §184 (test 68): a node's entry is computed from the catalog half of the branch its placement names, so a +// change on `beta` moves only the want of the node whose placement chose `beta`. +func TestACommitOnBetaMovesOnlyTheNodeThatChoseBeta(t *testing.T) { + fp := func(c string) string { return meshcatalogclient.SourcePrefix2 + strings.Repeat(c, 64) } + wantsKV := &memKV{values: map[string][]byte{}} + p := newPasser(kvSet{buckets.Wants: wantsKV, buckets.Asks: &memKV{values: map[string][]byte{}}}, time.Now()) + p.say = func(oc.Condition) error { return nil } + p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf) + p.ends.registered(mpcat.Registration{Module: "web", Branches: []string{"main", "beta"}}, false) + digest := "sha256:" + strings.Repeat("c", 64) + on, _ := presenceValue(true) + for _, node := range []string{"g14", "ace"} { + composes(t, p.w, node, digest) + p.w.put(buckets.Presence, node, on) + } + for node, branch := range map[string]string{"g14": "", "ace": "beta"} { + pl := mma.Assignments{Node: node, Modules: []mma.PlacedModule{{Module: "web", Branch: branch}}} + values, err := mmac.AssignmentsValues(pl) + if err != nil { + t.Fatal(err) + } + for key, raw := range values { + p.w.put(plc.AssignmentsBucket, key, raw) + } + } + // publish makes the half of branch name web at fingerprint f: built, granted and passed by its pipeline. + publish := func(branch, f string, source int64) { + mdigest := strings.Repeat("e", 62) + fmt.Sprintf("%02d", source) + if branch == "beta" { + mdigest = strings.Repeat("f", 62) + fmt.Sprintf("%02d", source) + } + mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: f, Manifest: mdigest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) + putModuleRaw(p.w, branch, "web", mod) + p.w.manifests[mdigest] = parsed{module: "web", manifest: catalogue.Manifest{Module: "web"}} + bkey, _ := builds.BuilderBuildKey("web", f) + built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest}) + p.w.put(buckets.Builds, bkey, built) + gkey, _ := grantsstate.PartGrantKey("web", f) + grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{}) + p.w.put(buckets.Grants, gkey, grant) + _ = p.ends.take(pipe("web", branch, source, 1, f, mpr.PipelineEndedOutcomePassed), true) + } + run := func(node, f string) { + have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: f, Composed: digest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) + p.w.put(buckets.Have, node+".web", have) + } + fpWant := func(node string) string { + key, _ := mwc.WantKey(node, "web") + w, err := mwc.DecodeWant(wantsKV.values[key]) + if err != nil { + t.Fatalf("the want of web on %s: %v", node, err) + } + return w.Fp + } + publish("main", fp("a"), 5) + publish("beta", fp("b"), 5) + run("g14", fp("a")) + run("ace", fp("b")) + p.wants(context.Background()) + if fpWant("g14") != fp("a") || fpWant("ace") != fp("b") { + t.Fatalf("each node is wanted at its branch's build: g14 %s, ace %s", fpWant("g14"), fpWant("ace")) + } + publish("beta", fp("d"), 6) + p.wants(context.Background()) + if fpWant("g14") != fp("a") || fpWant("ace") != fp("d") { + t.Errorf("a commit on beta moves only ace: g14 %s (want main's), ace %s (want beta's new)", fpWant("g14"), fpWant("ace")) + } +} + +// Design 47 G §14, §195: the catalog half of each branch is taken into the world per branch, an entry as the value and graph node +// the rules read, and a registration's branches are each followed. +func TestTheCatalogHalfIsTakenInPerBranch(t *testing.T) { + w := newWorld() + woken := 0 + h := &halves{w: w, wake: func() { woken++ }, logf: t.Logf} + entry := func(fp string) json.RawMessage { + raw, _ := json.Marshal(meshcatalog.CatalogEntry{Module: "web", Fp: fp, Repository: "novox/web", Commit: "c1", Manifest: map[string]any{"module": "web", "version": "1"}, + Invokes: []string{"seat:x.v"}}) + return raw + } + h.changed("main", "web", entry("src2:1"), false) + h.changed("beta", "web", entry("src2:2"), false) + if w.trunkOf("main")["web"].Fp2 != "src2:1" || w.trunkOf("beta")["web"].Fp2 != "src2:2" || woken != 2 { + t.Errorf("main %+v, beta %+v, woken %d: each branch keeps its own entry", w.trunkOf("main")["web"], w.trunkOf("beta")["web"], woken) + } + v := w.trunkOf("beta")["web"] + if got := w.manifest(v.Manifest); got.err != "" || got.manifest.Module != "web" { + t.Errorf("the entry's manifest is not read by its digest: %+v", got) + } + if n := w.graphOf("main")["web"]; n.FP != "src2:1" || !slices.Equal(n.Invokes, []string{"seat:x.v"}) { + t.Errorf("graph node %+v", n) + } + h.changed("beta", "web", nil, true) + if _, ok := w.trunkOf("beta")["web"]; ok || w.trunkOf("main")["web"].Fp2 == "" { + t.Error("a module taken out of beta's half leaves main's") + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index 0322e53c..75ffe180 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -112,7 +112,6 @@ func (p *passer) puts(ctx context.Context) { } assigned := p.w.assigned() trunk := p.w.trunk() - get := p.w.getter(buckets.Manifests) // the indexer's manifests, as its first view and views family say them wanted := p.w.wantsOf() modules := make([]string, 0, len(all)) for m := range all { @@ -122,7 +121,7 @@ func (p *passer) puts(ctx context.Context) { for _, module := range modules { var manifest catalogue.Manifest if mod, ok := trunk[module]; ok && mod.Manifest != "" { - if got := p.w.manifest(ctx, get, mod.Manifest); got.err == "" { + if got := p.w.manifest(mod.Manifest); got.err == "" { manifest = got.manifest } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go index 8efd714f..fb1a8330 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -16,7 +16,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" - "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" mdcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient" mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" @@ -465,10 +464,8 @@ func (p *passer) layAgain(ctx context.Context, prop mma.Proposal, rev uint64, cu // nodes with a record. func (p *passer) lays(ctx context.Context) (func(string) bool, func(string) bool) { held := map[string]bool{} - for key := range p.w.read(buckets.Modules) { - if name, err := meshcatalogclient.ParseModulesKey(key); err == nil { - held[name] = true - } + for name := range p.w.trunk() { + held[name] = true } records := p.w.records() return func(m string) bool { return held[m] }, func(n string) bool { _, ok := records[n]; return ok } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/verbs.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/verbs.go index bc3f55e6..9fa5ea2f 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/verbs.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/verbs.go @@ -1,7 +1,6 @@ package main import ( - "context" "mesh-delivery-assigner/internal/buckets" "slices" @@ -20,21 +19,8 @@ import ( // each kept edge until a build serving its verb runs healthy, read from the haves, the builder's builds and the indexer's // manifests, as a provider is counted (ADR 0603 §5). -// moduleGraph is the module graph as the indexer's first view and views family last said it (ADR 0631); a key that -// does not decode is left out, so its module waits on no verb. -func (p *passer) moduleGraph() graph.Graph { - g := graph.Graph{} - for key, raw := range p.w.read(buckets.Graph) { - module, err := graph.ParseKey(key) - if err != nil { - continue - } - if n, err := graph.Decode(raw); err == nil { - g[module] = n - } - } - return g -} +// moduleGraph is the module graph of the catalog half of branch, from the entries' invokes and serves (design 47 G §184, §195). +func (p *passer) moduleGraph(branch string) graph.Graph { return p.w.graphOf(branch) } // keptInvokes are the verbs of module the graph keeps as edges: each it invokes and does not leave out, that a module // other than itself serves. A verb no module of the graph serves is no edge, so it holds nothing; nil when the graph holds @@ -61,11 +47,11 @@ func keptInvokes(g graph.Graph, module string) []string { // serves are served. A healthy build whose manifest cannot be read counts what the graph says its module serves, said in // the log, so a build read late never holds its callers for good; nil when the graph holds nothing, and no want waits on // a verb. -func (p *passer) served(ctx context.Context, get getter, g graph.Graph, n assign.Nodes) func(string) bool { +func (p *passer) served(g graph.Graph, n assign.Nodes) func(string) bool { if len(g) == 0 { return nil } - served, unread := graph.Served(n.Haves, func(module, fp string) ([]string, bool) { return p.servesOf(ctx, get, module, fp) }, + served, unread := graph.Served(n.Haves, func(module, fp string) ([]string, bool) { return p.servesOf(module, fp) }, func(module string) []string { return g[module].Serves }) if len(unread) > 0 { p.logf("the manifest of the running build of %v cannot be read, so what the module graph says they serve is counted as served", unread) @@ -77,7 +63,7 @@ func (p *passer) served(ctx context.Context, get getter, g graph.Graph, n assign // §5, ADR 0614 §5): every running healthy have on an online node, its build's manifest read as servesOf reads it. A // healthy build whose manifest cannot be read counts what its module's trunk manifest offers, so a build read late never // holds a consumer for good. -func (p *passer) offering(ctx context.Context, get getter, trunk map[string]catalogue.Manifest, n assign.Nodes) func(string) bool { +func (p *passer) offering(trunk map[string]catalogue.Manifest, n assign.Nodes) func(string) bool { offered := map[string]bool{} for node, haves := range n.Haves { if !n.Online[node] { @@ -87,7 +73,7 @@ func (p *passer) offering(ctx context.Context, get getter, trunk map[string]cata if h.Phase != nodeengineclient.PhaseRunning || h.Health != nodeengineclient.HealthOK { continue } - m, ok := p.manifestOf(ctx, get, module, h.Fp) + m, ok := p.manifestOf(module, h.Fp) if !ok { m = trunk[module] } @@ -100,8 +86,8 @@ func (p *passer) offering(ctx context.Context, get getter, trunk map[string]cata } // servesOf are the verbs the manifest of module's build at fp serves (graph.ServesOf), and whether it could be read. -func (p *passer) servesOf(ctx context.Context, get getter, module, fp string) ([]string, bool) { - m, ok := p.manifestOf(ctx, get, module, fp) +func (p *passer) servesOf(module, fp string) ([]string, bool) { + m, ok := p.manifestOf(module, fp) if !ok { return nil, false } @@ -110,7 +96,7 @@ func (p *passer) servesOf(ctx context.Context, get getter, module, fp string) ([ // manifestOf is the manifest of module's build at fp (the builder's build names its digest in the indexer's manifests), // and whether it could be read. -func (p *passer) manifestOf(ctx context.Context, get getter, module, fp string) (catalogue.Manifest, bool) { +func (p *passer) manifestOf(module, fp string) (catalogue.Manifest, bool) { key, err := builds.BuilderBuildKey(module, fp) if err != nil { return catalogue.Manifest{}, false @@ -123,7 +109,7 @@ func (p *passer) manifestOf(ctx context.Context, get getter, module, fp string) if err != nil || b.Manifest == "" { return catalogue.Manifest{}, false } - got := p.w.manifest(ctx, get, b.Manifest) + got := p.w.manifest(b.Manifest) if got.err != "" { return catalogue.Manifest{}, false } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index e1b6659f..4186a832 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -6,6 +6,7 @@ import ( "mesh-delivery-assigner/internal/assignment" "mesh-delivery-assigner/internal/buckets" "slices" + "sync" "time" "github.com/nats-io/nats.go/jetstream" @@ -20,7 +21,8 @@ import ( sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard" - "git.novox.be/novox/mesh-delivery/sdk/graph" + mcat "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "mesh-delivery-assigner/internal/assign" plc "mesh-delivery-assigner/internal/assignment" @@ -126,10 +128,11 @@ func viewFindings(f assign.Findings) views.FindingWriter { // firstViewsBucket is the assigner's own state of its first views (views.StatusState). var firstViewsBucket = assignment.Module + "_" + views.StatusState -// indexerViews are the assigner's first views of the indexer's manifests, modules and module graph (ADR 0615 §1, -// ADR 0621 §2, ADR 0631; DLV-F2r): asked at the start through the indexer's tools, then followed on indexer-views, asking once more on a gap; -// past the bound its finding names the indexer and the tool, cleared by the answer. -type indexerViews struct { +// catalogViews is the assigner's first view of the catalog half of each watched branch (design 47 G §14, §184, §195): asked +// at the start through the seat mesh-catalog's verb `revision` (the trunk's, then each branch a registration names), then +// followed on its event `catalog-revision`, asking once more on a gap; past the bound its finding names the seat and the verb, +// cleared by the answer. Its status is read by assigner_first_views and asked again by assigner_first_view_again. +type catalogViews struct { js jetstream.JetStream account process.Account call func(string, any) (json.RawMessage, error) @@ -137,21 +140,17 @@ type indexerViews struct { status *process.AwaitedBucket logf func(string, ...any) hub *process.EventHub -} -// indexerFirstViews are the indexer's states the assigner takes by first view and the views family, each by the name of -// the indexer's bucket the world keeps it under: manifests and modules (ADR 0621 §2), and the module graph, whose verb -// edges a want waits on (ADR 0614 §2, §3; ADR 0631 §1). -var indexerFirstViews = map[string]string{"manifests": buckets.Manifests, "modules": buckets.Modules, graph.State: buckets.Graph} + h *halves +} // firstViewBound and firstViewWait are each first view's bound before its condition is raised, and its first wait // between two failed asks; zero is the views package's (views.Bound, two minutes, ADR 0615 §6; and a first wait of five // seconds). A test shortens them. var firstViewBound, firstViewWait time.Duration -// start asks the three first views, each answer taken into w under the indexer's bucket names and settled once (a pass waits -// for it), and hears the views family; it answers how many settles to wait for. -func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settled chan<- struct{}) int { +// start follows the trunk's catalog half and hears the seat's catalog-revision; it answers how many settles to wait for. +func (iv *catalogViews) start(ctx context.Context, w *world, wake func(), settled chan<- struct{}) int { finding := func(s views.Status, raised bool) { if iv.findings.Put == nil { return @@ -167,33 +166,28 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle iv.logf("views: the finding of the first view %s is not written: %v", s.Tool, err) } } - followers := map[string]*views.Follower{} - var all []*views.Follower - for state, bucket := range indexerFirstViews { - bucket := bucket - tool := buckets.IndexerModule + "." + state - f := &views.Follower{Owner: buckets.IndexerModule, Tool: tool, State: state, Ask: views.AskVia(iv.call, tool), Logf: iv.logf, - Bound: firstViewBound, Wait: firstViewWait, - Raise: func(s views.Status) { finding(s, true) }, Clear: func(s views.Status) { finding(s, false) }, - Changed: func(key string, value json.RawMessage, deleted bool, at time.Time) { - if deleted { - value = nil - } - w.putEntry(bucket, key, value, 0, at) - wake() - }} - followers[state] = f - all = append(all, f) - go func() { - if f.Start(ctx) == nil { - settled <- struct{}{} - } - }() + var statusKV jetstream.KeyValue + var mu sync.Mutex + iv.h = &halves{w: w, call: iv.call, wake: wake, logf: iv.logf, ctx: ctx, settled: settled} + iv.h.started = func(c *mcat.Catalog) { + f := c.Follower + f.Bound, f.Wait = firstViewBound, firstViewWait + f.Raise, f.Clear = func(s views.Status) { finding(s, true) }, func(s views.Status) { finding(s, false) } + mu.Lock() + defer mu.Unlock() + if statusKV != nil { + f.Said = views.KeepStatus(ctx, statusKV, iv.logf) + } } if iv.status != nil { iv.status.Ready(func(kv jetstream.KeyValue) { - for _, f := range all { - f.Said = views.KeepStatus(ctx, kv, iv.logf) + mu.Lock() + statusKV = kv + mu.Unlock() + var all []*views.Follower + for _, c := range iv.h.followers() { + c.Follower.Said = views.KeepStatus(ctx, kv, iv.logf) + all = append(all, c.Follower) } if err := views.HearAgain(ctx, kv, all, iv.logf); err != nil { iv.logf("views: assigner_first_view_again is not heard (%v) until the assigner starts again", err) @@ -202,16 +196,17 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle } // The hub says when the bus made its consumer: a picture asked before then is asked again (views.Follower.ConsumerMade). iv.hub.OnBound(func(created time.Time) { - for _, f := range all { - f.ConsumerMade(ctx, created) + for _, c := range iv.h.followers() { + c.Follower.ConsumerMade(ctx, created) } }) - // Not heard, the indexer's views family is the finding family-not-heard until it binds (ADR 0615 §5): the handler is - // registered on the module's hub, which the assigner starts once every handler is. - _ = notHeardWriter(iv.findings, iv.logf).Bind(process.EventsStream, "the indexer's manifests, modules and graph are the first views' alone until the assigner starts again", func() error { - return iv.hub.Handle("mesh.mod."+buckets.IndexerModule+".event."+views.EventPrefix+"*", views.Hear(ctx, followers, iv.logf)) + // Not heard, the catalog-revision is the finding family-not-heard until it binds (ADR 0615 §5): the handler is registered + // on the module's hub, which the assigner starts once every handler is. + _ = notHeardWriter(iv.findings, iv.logf).Bind(process.EventsStream, "the catalog half is the first views' alone until the assigner starts again", func() error { + return iv.hub.Handle(meshcatalog.SubjectCatalogRevisionAll, iv.h.hear) }) - return len(all) + iv.h.follow(trunkBranch) + return 1 } // notHeardWriter writes the assigner's finding of a family it does not hear into its findings (heard.Writer); with no 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 be881eb6..876e04e0 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -1,7 +1,6 @@ package main import ( - "context" "encoding/json" "errors" "maps" @@ -14,16 +13,14 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" wdc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" - "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient" mwc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshwantsclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-delivery/sdk/trunk" "git.novox.be/novox/mesh-sdk/go/keys" - "github.com/nats-io/nats.go/jetstream" - mmac "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -56,6 +53,10 @@ type world struct { // durable consumer (design 47 G §199, §309); runningHeard says that consumer is bound. running map[string]nodeengine.Running runningHeard bool + // modules and graphs are, by watched branch and module, the catalog half the assigner follows per branch (design 47 G §14, + // §184, §195), as the value and the graph node the rules read: written by putModule from the entries of meshcatalogclient.Catalog. + modules map[string]map[string]meshcatalog.ModulesByModuleValue + graphs map[string]map[string]graph.Node } // entryMeta is a value's revision in its bucket and when it was written. @@ -74,7 +75,8 @@ type parsed struct { func newWorld() *world { return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{}, meta: map[string]entryMeta{}, trunkBuilt: map[string]builds.BuildLogEntry{}, assignmentKeys: mmac.NewKeys(), - sourceGroups: map[uint64]string{}, running: map[string]nodeengine.Running{}} + sourceGroups: map[uint64]string{}, running: map[string]nodeengine.Running{}, + modules: map[string]map[string]meshcatalog.ModulesByModuleValue{}, graphs: map[string]map[string]graph.Node{}} } // putEntry is put, keeping the value's revision and the time it was written. @@ -194,71 +196,16 @@ func (w *world) read(bucket string) map[string][]byte { return out } -// getter reads one key's value; errNotThere when the key does not exist. -type getter func(ctx context.Context, key string) ([]byte, error) - var errNotThere = errors.New("no such key") -// kvGetter reads keys of a bucket. -func kvGetter(kv jetstream.KeyValue) getter { - return func(ctx context.Context, key string) ([]byte, error) { - e, err := kv.Get(ctx, key) - if errors.Is(err, jetstream.ErrKeyNotFound) { - return nil, errNotThere - } - if err != nil { - return nil, err - } - return e.Value(), nil - } -} - -// getter reads a key of a bucket the world holds: errNotThere where it holds none. -func (w *world) getter(bucket string) getter { - return func(_ context.Context, key string) ([]byte, error) { - w.mu.Lock() - defer w.mu.Unlock() - v, ok := w.values[bucket][key] - if !ok { - return nil, errNotThere - } - return v, nil - } -} - -// manifest reads the manifest of a digest from the indexer's bucket once and keeps it: a digest's manifest never changes. -func (w *world) manifest(ctx context.Context, get getter, digest string) parsed { +// manifest is the manifest of a digest the catalog half said (putModule keeps each once); one not said yet is not readable. +func (w *world) manifest(digest string) parsed { w.mu.Lock() - p, ok := w.manifests[digest] - w.mu.Unlock() - if ok { + defer w.mu.Unlock() + if p, ok := w.manifests[digest]; ok { return p } - key, err := meshcatalogclient.ManifestsKey(digest) - if err != nil { - return parsed{err: err.Error()} - } - value, err := get(ctx, key) - if errors.Is(err, errNotThere) { - return parsed{err: "the indexer holds no manifest of digest " + digest + " yet"} - } - if err != nil { - return parsed{err: err.Error()} - } - m, err := meshcatalogclient.DecodeManifest(value) - if err != nil { - p = parsed{err: err.Error()} - } else if man, err := catalogue.Decode([]byte(m.Manifest)); err != nil { - // Decoded, never judged (novox/hq ADR 0578 §3): the indexer's judge judged it, and its verdict is taken from the - // module's value where it is read (world.mesh). - p = parsed{module: m.Module, err: "it does not decode: " + err.Error()} - } else { - p = parsed{module: m.Module, manifest: man} - } - w.mu.Lock() - w.manifests[digest] = p - w.mu.Unlock() - return p + return parsed{err: "the catalog half holds no manifest of digest " + digest + " yet"} } // assigned is each node's modules the seat mesh-module-assigner's assignments names, in name order (ADR 0549 §5, ADR 0567 §8): the @@ -299,21 +246,74 @@ func (w *world) placedNodes() map[string]bool { return out } -// trunk is every module the indexer holds at the trunk, with its value. +// trunkOf is every module the catalog half of branch holds, with its value (design 47 G §184). +func (w *world) trunkOf(branch string) map[string]meshcatalog.ModulesByModuleValue { + w.mu.Lock() + defer w.mu.Unlock() + return maps.Clone(w.modules[branch]) +} + +// graphOf is the module graph of the catalog half of branch. +func (w *world) graphOf(branch string) graph.Graph { + w.mu.Lock() + defer w.mu.Unlock() + return graph.Graph(maps.Clone(w.graphs[branch])) +} + +// branches are the watched branches the catalog half is held of, the trunk's first. +func (w *world) branches() []string { + w.mu.Lock() + defer w.mu.Unlock() + out := slices.Sorted(maps.Keys(w.modules)) + if i := slices.Index(out, trunkBranch); i > 0 { + out = slices.Insert(slices.Delete(out, i, i+1), 0, trunkBranch) + } + return out +} + +// trunk is every module any branch's catalog half holds, each with its value on the first branch that holds it (the trunk's +// first): the modules that exist, for the parts that ask which, not at which fingerprint a node runs it. func (w *world) trunk() map[string]meshcatalog.ModulesByModuleValue { out := map[string]meshcatalog.ModulesByModuleValue{} - for key, value := range w.read(buckets.Modules) { - name, err := meshcatalogclient.ParseModulesKey(key) - if err != nil { - continue - } - if m, err := meshcatalogclient.DecodeModule(value); err == nil { - out[name] = m + for _, b := range w.branches() { + for name, v := range w.trunkOf(b) { + if _, ok := out[name]; !ok { + out[name] = v + } } } return out } +// putModule takes one module of branch's catalog half in: its value, its graph node and its parsed manifest by the digest the +// value names. +func (w *world) putModule(branch, module string, v meshcatalog.ModulesByModuleValue, n graph.Node, m *parsed) { + w.mu.Lock() + defer w.mu.Unlock() + if w.modules[branch] == nil { + w.modules[branch], w.graphs[branch] = map[string]meshcatalog.ModulesByModuleValue{}, map[string]graph.Node{} + } + w.modules[branch][module], w.graphs[branch][module] = v, n + if m != nil && v.Manifest != "" { + w.manifests[v.Manifest] = *m + } +} + +// dropModule takes a module out of branch's catalog half. +func (w *world) dropModule(branch, module string) { + w.mu.Lock() + defer w.mu.Unlock() + delete(w.modules[branch], module) + delete(w.graphs[branch], module) +} + +// holdsBranch says whether the catalog half of branch has been taken in (even empty). +func (w *world) holdsBranch(branch string) bool { + w.mu.Lock() + defer w.mu.Unlock() + return w.modules[branch] != nil +} + // records is each node's record in its own have key (`.node`): the nodes an assignments proposal is answered by, // with those assignments names (ADR 0593 §8). func (w *world) records() map[string]nodeengine.NodeRecord { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go index 73a1d5cf..7dcc316d 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go @@ -10,6 +10,7 @@ import ( "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/modules/mesh-grants/grantsstate" + "git.novox.be/novox/mesh-delivery/sdk/graph" ) // fakeManifests is the indexer's manifests bucket in memory. @@ -71,3 +72,12 @@ func theWorldCountsTheValuesAKeyTakes(t *testing.T) { } } } + +// putModuleRaw takes a module value encoded as the catalog's modules say it into the world as branch's catalog half holds it. +func putModuleRaw(w *world, branch, module string, raw []byte) { + v, err := meshcatalogclient.DecodeModule(raw) + if err != nil { + panic(err) + } + w.putModule(branch, module, v, graph.Node{FP: v.Fp2, Repository: v.Repository}, nil) +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index ba1b4796..0843516b 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -123,11 +123,11 @@ func TestManifest(t *testing.T) { } // ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) { - t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules, the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) + if !slices.Equal(m.Consumes, []string{"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) { + t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) } - if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", - "seat:mesh-catalog.registrations", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list + if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone "seat:node-backup.backed-up", "seat:node-backup.now", "seat:node-engine.manifest-reads", // ADR 0618 §3 diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index e69ae235..78ce919a 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -39,12 +39,10 @@ "operator-channel" ], "consumes": [ + "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", - "mesh-delivery-indexer.views.graph", - "mesh-delivery-indexer.views.manifests", - "mesh-delivery-indexer.views.modules", "mesh-delivery-teller.ask-ended", "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", @@ -52,10 +50,10 @@ "node-engine.running" ], "invokes": [ - "mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "seat:mesh-catalog.registrations", + "seat:mesh-catalog.revision", "seat:mesh-catalog.seat-definitions", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts",