novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

VW5: the view takes each watched branch's source from the gate's verb source, per branch

Browse Source 
The view read the gate's state source and its history, fixed to the branch main. Each watched branch's source is now
the seat mesh-delivery's verb source given that branch, asked again on its source-moved; the source row and the event
delivered are per branch, a repository's reach is from the source of its modules' registration's first branch, laid
over the reach the view's own state kept at its start, and view_tiers given no revision answers per branch. The
manifest drops its read of mesh-delivery-gate.source and names the source answer's fields it relies on.

Builds: design 47 G §507, §512, §532, §558, A-173, A-225

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 06:21:49 +01:00
parent 2940df27b1
commit da8384d10e
16 changed files with 619 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go
+2 -2
View File
@@ -17,8 +17,8 @@ import (
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
@@ -437,7 +437,7 @@ func main() {
r.SignersState = read(stdio.State("mesh-vault.signers"))
r.ConditionsState, r.AttemptsState = read(stdio.State(view.StatusConditionsState)), read(stdio.State(view.StatusAttemptsState))
r.PresenceState, r.StuckState = read(stdio.State(view.StatusPresenceState)), read(stdio.State(view.StatusStuckState))
r.SourceState = read(stdio.State(view.GateModule + ".source"))
r.Sources = view.SourcesOf(stdio.Ask, branches.Followed)
r.ProvisionsState = read(stdio.State(view.ProvisionsRead))
r.BuildLog = stdio.Log(builds.BuildLogRead).Since
r.Pending = pendingOf
modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/tiers_rows_test.go
+17 -4
View File
@@ -4,6 +4,7 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
"mesh-delivery-view/internal/view"
"reflect"
"strings"
@@ -137,10 +138,22 @@ func tiersToolSaysEachModulesWait(t *testing.T) {
t.Errorf("view_tiers with no answer of indexer_pending = %+v, %v", out, err)
}
r.Pending = nil
// No revision named: the gate's newest, its source key's revision.
r.SourceState = view.StateRead{GetEntry: func(string) (json.RawMessage, uint64, error) { return json.RawMessage(`{}`), 9, nil }}
if out, err := run(t, r, view.TiersTool, map[string]any{}); err != nil || out.(view.Tiers).Revision != 9 || out.(view.Tiers).Modules != 3 {
t.Errorf("view_tiers of the gate's newest = %+v, %v", out, err)
// No revision named: per watched branch, the newest source revision the gate's verb source answers for it (design 47 G
// A-173); a branch with no source revision yet is left out.
asked := map[string]bool{}
call := func(key string, body any) (json.RawMessage, error) {
in, _ := body.(meshdelivery.SourceInput)
asked[key+" "+in.Branch] = true
rev := map[string]int64{"release": 9}[in.Branch]
return json.Marshal(meshdelivery.SourceAnswer{Branch: in.Branch, Revision: rev})
}
r.Sources = view.SourcesOf(call, func() []string { return []string{"release", "beta"} })
if out, err := run(t, r, view.TiersTool, map[string]any{}); err != nil || len(out.(view.BranchTiers).Branches) != 1 ||
out.(view.BranchTiers).Branches[0].Branch != "release" || out.(view.BranchTiers).Branches[0].Revision != 9 || out.(view.BranchTiers).Branches[0].Modules != 3 {
t.Errorf("view_tiers of each branch's newest = %+v, %v", out, err)
}
if !asked["seat:mesh-delivery.source release"] || !asked["seat:mesh-delivery.source beta"] {
t.Errorf("the gate's verb source was asked %v, want once per watched branch", asked)
}
// ADR 0614 §2-§5: a want held on a verb, a reader capability or bus state is named by its edge; bus state is no
// provisioner key (none is ever made for it), so a module asking it shows no provision wait of it; and each verb edge
modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go
+34 -10
View File
@@ -59,7 +59,9 @@ type catalogHolder struct {
// rev is the revision's number; entries are its entries by module.
rev int64
entries map[string]meshcatalog.CatalogEntry
ctx context.Context
// src is the gate's source revision of catalogBranch, as its verb source answers it.
src meshdelivery.SourceAnswer
ctx context.Context
}
func serveCatalog(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) *catalogHolder {
@@ -98,8 +100,17 @@ func serveCatalog(t testing.TB, ctx context.Context, js jetstream.JetStream, nc
raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"})
_ = m.Respond(raw)
}
// The gate's verb source: given no branch the branches it keeps, given catalogBranch its source revision (A-173).
source := func(m *nats.Msg) {
raw, _ := json.Marshal(map[string]any{"result": map[string]any{"branches": []string{catalogBranch}}, "node": "n0"})
var in meshdelivery.SourceInput
_ = json.Unmarshal(m.Data, &in)
var a any = map[string]any{"branches": []string{catalogBranch}}
if in.Branch != "" {
h.mu.Lock()
a = meshdelivery.SourceAnswer{Branch: in.Branch, Revision: h.src.Revision, Group: h.src.Group, Repositories: h.src.Repositories}
h.mu.Unlock()
}
raw, _ := json.Marshal(map[string]any{"result": a, "node": "n0"})
_ = m.Respond(raw)
}
for subject, handle := range map[string]nats.MsgHandler{"mesh.seat.mesh-catalog.call.revision": answer,
@@ -118,6 +129,21 @@ func serveCatalog(t testing.TB, ctx context.Context, js jetstream.JetStream, nc
// catalogBranch is the one watched branch of these tests: no branch is named in the view's code (design 47 G §532).
const catalogBranch = "release"
// moveSource is the gate's next source revision of catalogBranch, said as source-moved (design 47 G §507, A-173).
func (h *catalogHolder) moveSource(rev int64, group string, heads map[string]string) {
h.t.Helper()
h.mu.Lock()
h.src = meshdelivery.SourceAnswer{Revision: rev, Group: group}
for repository, head := range heads {
h.src.Repositories = append(h.src.Repositories, meshdelivery.SourceHead{Repository: repository, Head: head, Trunk: catalogBranch})
}
h.mu.Unlock()
body, _ := json.Marshal(meshdelivery.SourceMovedEvent{Branch: catalogBranch, Revision: rev, Group: group})
if _, err := h.js.Publish(h.ctx, meshdelivery.SubjectSourceMoved(catalogBranch), body); err != nil {
h.t.Fatal(err)
}
}
// put is the holder's next trunk revision with the entry of one module changed, said as catalog-revision.
func (h *catalogHolder) put(e meshcatalog.CatalogEntry) {
h.t.Helper()
@@ -762,7 +788,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
t.Fatal(err)
}
kvs := map[string]jetstream.KeyValue{}
for _, name := range []string{view.SourceBucket, view.AssignerWantsBucket, view.HaveBucket, viewstate.ReachBucket} {
for _, name := range []string{view.AssignerWantsBucket, view.HaveBucket, viewstate.ReachBucket} {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
if err != nil {
t.Fatal(err)
@@ -791,7 +817,6 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
want, _ := meshwantsclient.EncodeWant(meshwantsclient.Want{Fp: fp, Composed: "sha256:" + strings.Repeat("c", 64)})
failed, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed})
good, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fp, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
src, _ := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}})
catalog.put(meshcatalog.CatalogEntry{Module: "web", Fp: fp, Manifest: map[string]any{"module": "a"}, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath})
if _, err := kvs[view.AssignerWantsBucket].Put(ctx, "want.ace.web", want); err != nil {
t.Fatal(err)
@@ -799,10 +824,9 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
if _, err := kvs[view.HaveBucket].Put(ctx, "ace.web", failed); err != nil {
t.Fatal(err)
}
rev, err := kvs[view.SourceBucket].Put(ctx, meshdeliveryclient.SourceKey("main"), src)
if err != nil {
t.Fatal(err)
}
// The gate's source of the watched branch moves: the view asks the verb source for it and reads no gate state (A-173).
const rev = 41
catalog.moveSource(rev, "feat/x", map[string]string{"novox/web": strings.Repeat("1", 40)})
time.Sleep(time.Second)
if info, err := events.Info(ctx, jetstream.WithSubjectFilter("mesh.mod.mesh-delivery-view.event."+DeliveredEventType)); err != nil || len(info.State.Subjects) != 0 {
t.Fatalf("an event was said before the revision was delivered: %v %+v", err, info.State)
@@ -818,8 +842,8 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
t.Fatal("the delivery of the source revision was never said")
}
var e view.DeliveredEvent
if err := json.Unmarshal(msg.Data, &e); err != nil || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) {
t.Errorf("the event = %+v, %v; want revision %d, group feat/x and the heads", e, err, rev)
if err := json.Unmarshal(msg.Data, &e); err != nil || e.Branch != catalogBranch || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) {
t.Errorf("the event = %+v, %v; want the branch %s, revision %d, group feat/x and the heads", e, err, catalogBranch, rev)
}
if msg.Header.Get("x-source") != "mesh-delivery-view" || msg.Header.Get("x-event-id") == "" {
t.Errorf("the headers = %v", msg.Header)
modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go
+35 -5
View File
@@ -142,12 +142,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
}
open := buckets{js: js, reader: reader}
for _, bucket := range []string{view.HaveBucket, view.AssignmentsBucket, view.PresenceBucket, view.StuckBucket, view.CandidatesBucket, view.AnswersBucket, builds.BuilderBuildsBucket, view.AssignerWantsBucket, view.SourceBucket, view.AssignerAsksBucket, viewstate.ViewAnswersBucket, view.TakenBucket} {
for _, bucket := range []string{view.HaveBucket, view.AssignmentsBucket, view.PresenceBucket, view.StuckBucket, view.CandidatesBucket, view.AnswersBucket, builds.BuilderBuildsBucket, view.AssignerWantsBucket, view.AssignerAsksBucket, viewstate.ViewAnswersBucket, view.TakenBucket, viewstate.ReachBucket} {
go follow(ctx, open, bucket, false, model, logger)
}
// The source's history gives each head's source revision (ADR 0556); the modules' history is each revision of the
// catalog half the view takes, in its order (below).
go follow(ctx, open, view.SourceBucket, true, model, logger)
// The view's own state reach, as it stood at the start, is what the derivation lays its history over (design 47 G A-225);
// the modules' history is each revision of the catalog half the view takes, in its order (below), and each watched
// branch's source is the seat mesh-delivery's verb source per branch, followed by its source-moved (A-173).
statusBucket := process.AwaitBucket(ctx, js, conn, firstViewsBucket, "view_first_views reads nothing and view_first_view_again is not heard", logger.Printf)
// The modules are the catalog half's, per watched branch, through meshcatalogclient's branch follower adopted on the
// reader (design 47 G §532, §614): the registrations' branches, before the first registrations the gate's source's;
@@ -158,7 +158,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err
}
branches.Logf = logger.Printf
modules := &catalogModules{ctx: ctx, branches: branches, model: model, logger: logger, applied: map[string]bool{}}
sources := &branchSources{ctx: ctx, reader: reader, call: calls.Call, model: model, logger: logger}
modules := &catalogModules{ctx: ctx, branches: branches, sources: sources, model: model, logger: logger, applied: map[string]bool{}}
branches.Changed = modules.changed
reader.Start(ctx, js, cred, statusBucket)
go func() {
@@ -341,6 +342,7 @@ func emitEvent(ctx context.Context, js jetstream.JetStream, cred process.Account
type catalogModules struct {
ctx context.Context
branches *meshcatalogclient.Branches
sources *branchSources // the source of each branch followed (A-173)
model *view.Model
logger *log.Logger
@@ -352,6 +354,7 @@ type catalogModules struct {
// changed is the branch follower's Changed: a module of one branch changed.
func (c *catalogModules) changed(_, module string, _ json.RawMessage, _ bool, at time.Time) {
c.follow()
c.mu.Lock()
held := c.held
c.mu.Unlock()
@@ -411,6 +414,7 @@ func (c *catalogModules) settle() {
// whole hands every module the followed branches list, removes from the model each it holds that none lists any more,
// and tells the model the modules are read.
func (c *catalogModules) whole() {
c.follow()
at := time.Now()
listed := map[string]bool{}
for _, branch := range c.branches.Followed() {
@@ -443,9 +447,35 @@ func (c *catalogModules) whole() {
c.model.Initial(view.ReachHistory(view.ModulesBucket))
}
// follow has the source of every followed branch followed, and no other (A-173).
func (c *catalogModules) follow() {
if c.sources != nil {
c.sources.ensure(c.branches.Followed())
}
}
// branchOf is the watched branch a module's entry is read from: the first branch of its registration, before the first
// registrations the first followed branch that lists it (design 47 G A-103, A-173); "" when none.
func (c *catalogModules) branchOf(module string) string {
if reg, ok := c.branches.Registrations.Registration(module); ok && len(reg.Branches) > 0 {
return reg.Branches[0]
}
for _, branch := range c.branches.Followed() {
if cat, ok := c.branches.Catalog(branch); ok {
if _, ok := cat.Entry(module); ok {
return branch
}
}
}
return ""
}
// apply hands the model one module's entry, or its removal when no branch lists it.
func (c *catalogModules) apply(module string, entry meshcatalog.CatalogEntry, listed bool, at time.Time) {
e := view.Entry{Bucket: view.ModulesBucket, Key: module, At: at, Deleted: !listed}
if listed {
e.Branch = c.branchOf(module)
}
if listed {
raw, _ := json.Marshal(entry)
v, err := view.ModuleValue(raw)
modules/mesh-delivery-view/cmd/mesh-delivery-view/sources.go
+153
View File
@@ -0,0 +1,153 @@
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"mesh-delivery-view/internal/view"
"sync"
"time"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/sourcebranches"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/views"
)
// sourceState is the state of the view of one watched branch's source in the view's reader: `mesh-delivery.source.<branch>`.
const sourceState = "source"
// adopter is how branchSources puts a branch's view on the view's one reader and takes it off (views.ProcessReader).
type adopter interface {
Adopt(f *views.Follower, seat, event string, hear func(context.Context, []byte) error) (*views.Picture, error)
Drop(owner, state string) error
}
// branchSources follows the source of each watched branch on its own (design 47 G §507, §512, A-173): per branch the
// first view of the seat mesh-delivery's verb source given that branch, then each source-moved of that branch, asked
// again until the revision heard is held; each answer goes to the model as the branch's SourceBucket entry. The branches
// are the ones the branch follower follows (ensure); none is named here.
type branchSources struct {
ctx context.Context
reader adopter
call func(key string, body any) (json.RawMessage, error)
model *view.Model
logger *log.Logger
mu sync.Mutex
branches map[string]*views.Follower // by escaped branch
}
// ensure follows the source of every branch named and drops the view of each followed branch no longer named.
func (s *branchSources) ensure(followed []string) {
s.mu.Lock()
defer s.mu.Unlock()
if s.branches == nil {
s.branches = map[string]*views.Follower{}
}
named := map[string]bool{}
for _, branch := range followed {
esc := keys.EscapeName(branch)
named[esc] = true
if s.branches[esc] != nil {
continue
}
f := s.follower(branch)
if _, err := s.reader.Adopt(f, meshdelivery.Seat, meshdelivery.EventSourceMoved, s.hear(branch, f)); err != nil {
s.logger.Printf("the source of %s is not followed, and is adopted again at the next change of the catalog half: %v", branch, err)
continue
}
s.branches[esc] = f
}
for esc := range s.branches {
if named[esc] {
continue
}
if err := s.reader.Drop(meshdelivery.Seat, views.BranchState(sourceState, mustUnescape(esc))); err != nil {
s.logger.Printf("the source of %s is not dropped: %v", esc, err)
continue
}
delete(s.branches, esc)
s.model.Apply(view.Entry{Bucket: view.SourceBucket, Key: esc, Deleted: true, At: time.Now()})
}
}
func mustUnescape(esc string) string {
name, err := keys.UnescapeName(esc)
if err != nil {
return esc
}
return name
}
// follower is the view of one branch's source: its one value, keyed by the escaped branch, the verb's answer.
func (s *branchSources) follower(branch string) *views.Follower {
esc := keys.EscapeName(branch)
f := &views.Follower{Owner: meshdelivery.Seat, Tool: view.SourceVerb, State: views.BranchState(sourceState, branch)}
f.Ask = func(context.Context, string) (views.Answer, error) {
a, err := view.SourceOf(s.call, branch)
if err != nil {
return views.Answer{}, err
}
raw, err := json.Marshal(a)
if err != nil {
return views.Answer{}, err
}
return views.Answer{V: 1, Owner: meshdelivery.Seat, State: f.State, Revision: uint64(a.Revision), Created: sourcebranches.Created,
Values: map[string]json.RawMessage{esc: raw}, At: time.Now()}, nil
}
f.Changed = func(_ string, value json.RawMessage, deleted bool, at time.Time) {
e := view.Entry{Bucket: view.SourceBucket, Key: esc, Deleted: deleted, At: at}
if !deleted {
var a meshdelivery.SourceAnswer
if err := json.Unmarshal(value, &a); err != nil {
s.logger.Printf("the source of %s is not taken: %v", branch, err)
return
}
se, err := view.SourceEntry(branch, a)
if err != nil {
s.logger.Printf("the source of %s is not taken: %v", branch, err)
return
}
se.At = at
e = se
}
s.model.Apply(e)
}
return f
}
// revisionOf is the source revision f holds of the branch, 0 when none.
func revisionOf(f *views.Follower, esc string) int64 {
raw, ok := f.Get(esc)
if !ok {
return 0
}
var a meshdelivery.SourceAnswer
if json.Unmarshal(raw, &a) != nil {
return 0
}
return a.Revision
}
// hear is the handler of source-moved for one branch's view: another branch's event is not its own; a revision it does not
// hold yet asks the first view again until it does, on its own goroutine, so the consumer is never held.
func (s *branchSources) hear(branch string, f *views.Follower) func(context.Context, []byte) error {
esc := keys.EscapeName(branch)
return func(ctx context.Context, data []byte) error {
var ev meshdelivery.SourceMovedEvent
if err := json.Unmarshal(data, &ev); err != nil || ev.Branch == "" {
return fmt.Errorf("not a %s event: %s", meshdelivery.EventSourceMoved, sourcebranches.ClipBytes(data))
}
if ev.Branch != branch && ev.Branch != esc {
return nil
}
held := func() bool { return revisionOf(f, esc) >= ev.Revision }
if held() {
return nil
}
go sourcebranches.CatchUp(s.ctx, f, held, fmt.Sprintf("source of %s moved to %d", branch, ev.Revision))
return nil
}
}
modules/mesh-delivery-view/internal/view/names.go
+3 -7
View File
@@ -3,7 +3,6 @@ package view
import (
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk"
"git.novox.be/novox/mesh-sdk/go/keys"
@@ -40,8 +39,9 @@ const (
AssignerWantsBucket = AssignerModule + "_wants"
AssignerAsksBucket = AssignerModule + "_asks"
// The gate's source, and the teller's state answers as the view follows it by its views family (design 47 G §431, A-175):
// its picture `<owner>_<state>`.
// SourceBucket names, in the model's entries, each watched branch's source as the seat mesh-delivery's verb source answers
// it, keyed by the escaped branch (design 47 G §507, A-173): never a bucket the view reads. The teller's state answers as
// the view follows it by its views family (design 47 G §431, A-175): its picture `<owner>_<state>`.
SourceBucket = GateModule + "_source"
AnswersState = "answers"
AnswersBucket = TellerModule + "_" + AnswersState
@@ -88,7 +88,3 @@ func PackagesKey(module string) (string, error) {
}
return meshcatalog.PackagesByModuleKey(keys.EscapeName(module)), nil
}
// trunkSourceKey is the key of the gate's state source for the trunk, the watched branch `main` (design 47 G §305: the source
// is kept per watched branch, meshdeliveryclient.SourceKey; the view's reads are of the trunk's).
var trunkSourceKey = meshdeliveryclient.SourceKey("main")
modules/mesh-delivery-view/internal/view/reach.go
+60 -30
View File
@@ -9,7 +9,6 @@ import (
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
@@ -22,14 +21,15 @@ import (
// head as its event delivered, per repository. Nothing here records a merge as delivered (ADR 0429 §1): a merge at or before
// the reached head is on the machines, derived by whoever reads it.
// ReachHistory names the watch of a bucket with its history, as Initial is told it: the modules' and the source's history
// give each fingerprint's first commit and each head's source revision.
// ReachHistory names the watch of a bucket with its history, as Initial is told it: the modules' history gives each
// fingerprint's first commit. Each head's source revision is from the branch's sources heard since the start (design 47 G
// A-173, A-225), laid over the reach the view's own state kept.
func ReachHistory(bucket string) string { return bucket + "#history" }
// reachBuckets are the watches the derivation waits for before it writes or says anything, so a start does not write a reach
// derived from half its inputs.
var reachBuckets = []string{ModulesBucket, AssignerWantsBucket, HaveBucket, SourceBucket,
TakenBucket, ReachHistory(ModulesBucket), ReachHistory(SourceBucket)}
// derived from half its inputs; the view's own reach first, which a restart lays the derivation over (A-225).
var reachBuckets = []string{ModulesBucket, AssignerWantsBucket, HaveBucket, TakenBucket, ReachHistory(ModulesBucket),
viewstate.ReachBucket}
// ReachWrite is a reach value to write under its key in the view's state reach.
type ReachWrite struct {
@@ -45,8 +45,7 @@ func (m *Model) SetReachWriter(put func(ReachWrite) error) {
m.reachPut = put
}
// applyHistory takes one revision of a watch with history: of the indexer's modules (each module's values in order) and of
// the gate's source (each revision's heads).
// applyHistory takes one revision of a watch with history: of the catalog half's modules (each module's values in order).
func (m *Model) applyHistory(e Entry) {
switch e.Bucket {
case ModulesBucket:
@@ -70,15 +69,6 @@ func (m *Model) applyHistory(e Entry) {
h = h[len(h)-viewstate.MostChanges:]
}
m.modHist[e.Key] = h
case SourceBucket:
if e.Key != trunkSourceKey || e.Deleted || e.Revision == 0 {
return
}
src, err := meshdeliveryclient.DecodeSource(e.Value)
if err != nil {
return
}
m.srcHist[e.Revision] = src.Heads
}
}
@@ -88,11 +78,13 @@ type headAt struct {
rev uint64
}
// headsOf are the heads of a repository in the source's history, oldest first, each with the first revision naming it.
func (m *Model) headsOf(repository string) []headAt {
// headsOf are the heads of a repository in the history of one branch's source (escaped), oldest first, each with the first
// revision naming it.
func (m *Model) headsOf(branch, repository string) []headAt {
var out []headAt
for _, rev := range slices.Sorted(maps.Keys(m.srcHist)) {
for r, head := range m.srcHist[rev] {
hist := m.srcHist[branch]
for _, rev := range slices.Sorted(maps.Keys(hist)) {
for r, head := range hist[rev] {
if !trunk.SameRepository(r, repository) || head == "" {
continue
}
@@ -136,10 +128,11 @@ func (m *Model) wantKeys() map[string][]string {
return out
}
// deriveReach is the reach of one repository from the model as it stands; wants are the wants' keys by module (wantKeys).
func (m *Model) deriveReach(repository string, wants map[string][]string) viewstate.Reach {
// deriveReach is the reach of one repository from the source of its branch (escaped) as the model holds it; wants are the
// wants' keys by module (wantKeys).
func (m *Model) deriveReach(branch, repository string, wants map[string][]string) viewstate.Reach {
r := viewstate.Reach{Repository: trunk.RepositoryName(repository), Modules: map[string]viewstate.ModuleReach{}}
heads := m.headsOf(repository)
heads := m.headsOf(branch, repository)
key, _ := trunk.RepositoryKey(repository)
if t, ok := m.taken[key]; ok {
r.Indexed = t.Merge
@@ -178,6 +171,11 @@ func (m *Model) deriveReach(repository string, wants map[string][]string) viewst
r.Reached, r.ReachedSrc = h.commit, h.rev
}
}
// The history is the revisions heard since the start: the reach the view's own state kept at the start stands until one
// derived from them is newer (A-225).
if k, ok := m.kept[key]; ok && k.Reached != "" && (r.Reached == "" || r.ReachedSrc < k.ReachedSrc) {
r.Reached, r.ReachedSrc = k.Reached, k.ReachedSrc
}
if n := len(heads); n > 0 && !sameCommit(heads[n-1].commit, r.Reached) {
if indexed != 0 && indexed < heads[n-1].rev {
holds = append(holds, fmt.Sprintf("the indexer has not yet taken in %.12s", heads[n-1].commit))
@@ -387,11 +385,30 @@ func (m *Model) reachReady() bool {
return false
}
}
return m.source != nil && m.source.err == ""
return true
}
// reachNow derives the reach of every repository the source names, and answers the values that changed and the reached
// heads not yet said. It is called with the lock held.
// branchOf is the escaped branch a repository's reach is derived from: the first branch of its modules' registration (by module
// name, the first that has one), before the first registrations the first branch whose source names the repository (design 47
// G A-173). "" when none.
func (m *Model) branchOf(repository string) string {
for _, name := range slices.Sorted(maps.Keys(m.modules)) {
if me := m.modules[name]; me.err == "" && trunk.SameRepository(me.value.Repository, repository) && m.modBranch[name] != "" {
return m.modBranch[name]
}
}
for _, b := range slices.Sorted(maps.Keys(m.sources)) {
for r := range m.sources[b].heads {
if trunk.SameRepository(r, repository) {
return b
}
}
}
return ""
}
// reachNow derives the reach of every repository a watched branch's source names, from the source of the repository's own
// branch (branchOf), and answers the values that changed and the reached heads not yet said. It is called with the lock held.
func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) {
if (m.reachPut == nil && m.deliveredEvent == nil) || !m.reachReady() {
return nil, nil
@@ -399,12 +416,25 @@ func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) {
var writes []ReachWrite
var events []DeliveredEvent
wants := m.wantKeys()
for _, repository := range slices.Sorted(maps.Keys(m.source.heads)) {
type repoAt struct{ branch, repository string }
var repos []repoAt
for _, b := range slices.Sorted(maps.Keys(m.sources)) {
if m.sources[b].err != "" {
continue
}
for _, repository := range slices.Sorted(maps.Keys(m.sources[b].heads)) {
if m.branchOf(repository) == b {
repos = append(repos, repoAt{b, repository})
}
}
}
for _, ra := range repos {
repository := ra.repository
key, err := trunk.RepositoryKey(repository)
if err != nil {
continue
}
r := m.deriveReach(repository, wants)
r := m.deriveReach(ra.branch, repository, wants)
if prev, ok := m.reach[key]; !ok || !viewstate.SameReach(prev, r) {
if data, err := viewstate.EncodeReach(r); err == nil {
m.reach[key] = r
@@ -415,7 +445,7 @@ func (m *Model) reachNow() ([]ReachWrite, []DeliveredEvent) {
said := strings.Join(notYet, ",")
if r.Reached != "" && (r.ReachedSrc > m.reachSaid[key] || said != m.reachNotYet[key]) {
m.reachSaid[key], m.reachNotYet[key] = r.ReachedSrc, said
events = append(events, DeliveredEvent{Revision: r.ReachedSrc, Heads: map[string]string{r.Repository: r.Reached},
events = append(events, DeliveredEvent{Branch: m.sources[ra.branch].branch, Revision: r.ReachedSrc, Heads: map[string]string{r.Repository: r.Reached},
Repository: r.Repository, NotYet: notYet})
}
}
modules/mesh-delivery-view/internal/view/reach_test.go
+34 -14
View File
@@ -16,6 +16,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// 1280 (ADR 0556 clauses 2 and 3): the view derives, per repository, how far each module reached and the reached head, writes
@@ -30,14 +31,14 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: ModulesBucket, Key: module, Value: v, At: t0, History: history}
return Entry{Bucket: ModulesBucket, Key: module, Value: v, At: t0, History: history, Branch: testBranch}
}
modIn := func(fp, commit string, src uint64, in *meshcatalog.ModulesByModuleValueInputs2, history bool) Entry {
v, err := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: strings.Repeat("a", 64), Repository: web, Commit: commit, Src: int64(src), Path: "web", Inputs2: in})
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: ModulesBucket, Key: "web", Value: v, At: t0, History: history}
return Entry{Bucket: ModulesBucket, Key: "web", Value: v, At: t0, History: history, Branch: testBranch}
}
in := func(tree string, pkgs ...string) *meshcatalog.ModulesByModuleValueInputs2 {
return &meshcatalog.ModulesByModuleValueInputs2{Tree: tree, Packages: pkgs}
@@ -48,12 +49,23 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
later := func(commit string, to *meshcatalog.ModulesByModuleValueInputs2, extra ...Entry) []Entry {
return append([]Entry{modIn(fpLater, commit, 12, to, true), modIn(fpLater, commit, 12, to, false)}, extra...)
}
src := func(rev uint64, webHead string) Entry {
// Each watched branch's source as its verb source answers it, one revision heard after another (design 47 G A-173).
srcOn := func(branch string, rev uint64, webHead string) Entry {
v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "g", Heads: map[string]string{web: webHead, sdk: commitS}})
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: SourceBucket, Key: trunkSourceKey, Value: v, At: t0, Revision: rev, History: true}
return Entry{Bucket: SourceBucket, Key: keys.EscapeName(branch), Value: v, At: t0, Revision: rev}
}
src := func(rev uint64, webHead string) Entry { return srcOn(testBranch, rev, webHead) }
// The reach the view's own state held at its start (A-225).
kept := func(reached string, rev uint64) Entry {
key, _ := trunk.RepositoryKey(web)
v, err := viewstate.EncodeReach(viewstate.Reach{Repository: web, Reached: reached, ReachedSrc: rev})
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: viewstate.ReachBucket, Key: key, Value: v, At: t0}
}
taken := func(repository, merge string) Entry {
key, _ := trunk.RepositoryKey(repository)
@@ -70,17 +82,18 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
return e
}
// web changed at A (fpOld) and at C (fpNew); mail first appeared at A. The source named A, B, then C as web's head.
world := func(t *testing.T) []Entry {
cur, _ := meshdeliveryclient.DecodeSource(src(12, commitC).Value)
current := Entry{Bucket: SourceBucket, Key: trunkSourceKey, Revision: 12, At: t0}
current.Value, _ = meshdeliveryclient.EncodeSource(cur)
return []Entry{src(10, commitA), src(11, commitB), src(12, commitC), current,
world := func(t *testing.T, restarted bool) []Entry {
heard := []Entry{src(10, commitA), src(11, commitB), src(12, commitC)}
if restarted { // a start after which only the newest revision is heard, over the reach the state kept (A-225)
heard = []Entry{kept(commitB, 11), src(12, commitC)}
}
return append(heard,
mod("web", fpOld, commitA, 10, true), modIn(fpNew, commitC, 12, in("t1", "mesh-sdk/go"), true), mod("mail", fpMail, commitA, 0, true),
modIn(fpNew, commitC, 12, in("t1", "mesh-sdk/go"), false), mod("mail", fpMail, commitA, 0, false),
taken(web, commitC), taken(sdk, commitS),
wantE(t, "ace", "web", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "ace", "web", fpNew, nodeengineclient.HealthOK),
wantE(t, "g14", "web", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "g14", "web", fpNew, nodeengineclient.HealthOK),
wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpMail}), haveFP(t, "ace", "mail", fpMail, nodeengineclient.HealthOK)}
wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpMail}), haveFP(t, "ace", "mail", fpMail, nodeengineclient.HealthOK))
}
for _, tc := range []struct {
name string
@@ -90,8 +103,17 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
waiting string // words web's module web waits with; "" none
notYet []string // the merges the reached head's event says are not yet on the machines
refused bool // the bus refuses every write and event until the watches handed over (a start, issue 1126)
// restarted: the view started again, heard only the newest source revision, and its state reach kept commitB (A-225)
restarted bool
}{
{name: "every node runs the newest build", reached: commitC},
// design 47 G A-173: a repository's reach is from the source of its modules' branch; another branch's source moves it not.
{name: "another branch's source naming the repository moves not its reach", reached: commitC,
then: []Entry{srcOn("beta", 13, commitX)}},
// A-225: after a restart the history is the revisions heard since; the kept reach stands until a newer one is derived.
{name: "after a restart the reach kept stands while the newest head lags", reached: commitB, holds: "web on g14", restarted: true,
then: []Entry{haveFP(t, "g14", "web", fpOld, nodeengineclient.HealthOK)}, waiting: "not yet the"},
{name: "after a restart a newer head reached replaces the kept one", reached: commitC, restarted: true},
{name: "a node lagging stops the head before the change it lacks", reached: commitB, holds: "web on g14",
then: []Entry{haveFP(t, "g14", "web", fpOld, nodeengineclient.HealthOK)}, waiting: "not yet the"},
// ADR 0603 §7: a node away holds no head back, is named, and holds it again once it returns still behind.
@@ -134,12 +156,10 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
{name: "a merge that removed the last module is reached at the merge the indexer took in", reached: commitD,
then: func() []Entry {
head := src(13, commitD)
current := head
current.History = false
gone := func(module string, history bool) Entry {
return Entry{Bucket: ModulesBucket, Key: module, Deleted: true, At: t0, History: history}
}
return []Entry{head, current, gone("web", true), gone("mail", true), gone("web", false), gone("mail", false), taken(web, commitD)}
return []Entry{head, gone("web", true), gone("mail", true), gone("web", false), gone("mail", false), taken(web, commitD)}
}()},
// 1126: the view started while the bus did not answer its writes; its reach and its event are made at the next change it
// takes in, though the derivation is the same, so the state reach never keeps an older have than node_haves says.
@@ -168,7 +188,7 @@ func TestTheViewDerivesHowFarEachRepositoryReached(t *testing.T) {
}
return nil
})
fill(m, world(t)...)
fill(m, world(t, tc.restarted)...)
if len(written) != 0 {
t.Fatalf("written before every watch handed over: %v", written)
}
modules/mesh-delivery-view/internal/view/read.go
+6 -3
View File
@@ -66,9 +66,12 @@ type Reader struct {
PredefinedState, ConfigCandidatesState StateRead
// PackagesState is the indexer's published packages (PackagesBucket, ADR 0593 §2); unset reads none.
PackagesState StateRead
// GraphState is the indexer's module graph (graph.Bucket, ADR 0603 §2), SourceState the gate's source and
// ProvisionsState the provisioner's keys: what view_tiers reads (tiers.go). Unset reads none, said in the answer.
GraphState, SourceState, ProvisionsState StateRead
// GraphState is the indexer's module graph (graph.Bucket, ADR 0603 §2) and ProvisionsState the provisioner's keys: what
// view_tiers reads (tiers.go). Unset reads none, said in the answer.
GraphState, ProvisionsState StateRead
// Sources is the newest source revision of each watched branch, by branch name, as the seat mesh-delivery's verb source
// answers it (design 47 G §507, A-173): what view_tiers given no revision answers per branch. Unset, it must be named.
Sources func() (map[string]uint64, error)
// Pending reads the indexer's waits through its tool indexer_pending (ADR 0633 §5); unset names no module held back.
Pending func() (takein.PendingAnswer, error)
// BuildLog reads the builder's log from a time (stdio.Log(builds.BuildLogRead).Since); unset reads none.
modules/mesh-delivery-view/internal/view/serve.go
+7 -5
View File
@@ -251,13 +251,15 @@ func wantText(s Snapshot) [][]string {
return out
}
// sourceText is the source revision's derived state in one line, then one line for each module on a node it waits for.
// sourceText is each watched branch's source revision's derived state in one line, then one line for each module on a node it
// waits for.
func sourceText(s Snapshot) []string {
if s.Source == nil {
return nil
var out []string
for _, src := range s.Sources {
out = append(out, fmt.Sprintf("%s: source revision %d: %s", src.Branch, src.Revision, src.State))
out = append(out, src.Waiting...)
}
out := []string{fmt.Sprintf("source revision %d: %s", s.Source.Revision, s.Source.State)}
return append(out, s.Source.Waiting...)
return out
}
// backlogText is one line per language and kind: its waiting asks, the asks held by a node, the oldest waiting ask's age.
modules/mesh-delivery-view/internal/view/source.go
+71
View File
@@ -0,0 +1,71 @@
package view
import (
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"git.novox.be/novox/mesh-sdk/go/keys"
)
// SourceVerb is the seat mesh-delivery's verb source as the view's invokes names it: each watched branch's source revision,
// asked of the seat and never read from the gate's state (design 47 G §507, §512, A-173).
const SourceVerb = "seat:" + meshdelivery.Seat + "." + meshdelivery.VerbSource
// SourceOf asks the seat, through call, for one watched branch's source revision (branch as named) and reads its declared
// answer; an answer that carries an error is that error.
func SourceOf(call func(key string, body any) (json.RawMessage, error), branch string) (meshdelivery.SourceAnswer, error) {
raw, err := call(SourceVerb, meshdelivery.SourceInput{Branch: keys.EscapeName(branch)})
if err != nil {
return meshdelivery.SourceAnswer{}, fmt.Errorf("%s of %s: %w", SourceVerb, branch, err)
}
answer, said, ok := buscalls.Unwrap(raw)
if !ok {
return meshdelivery.SourceAnswer{}, fmt.Errorf("%s of %s answered: %s", SourceVerb, branch, said)
}
var a meshdelivery.SourceAnswer
if err := json.Unmarshal(answer, &a); err != nil {
return a, fmt.Errorf("the answer of %s of %s is not a source revision: %w", SourceVerb, branch, err)
}
if a.Error != "" {
return a, fmt.Errorf("%s of %s: %s", SourceVerb, branch, a.Error)
}
return a, nil
}
// SourceEntry is a branch's source answer as the model's entry of SourceBucket: keyed by the escaped branch, its revision the
// source revision, its value the heads by repository; a branch the gate holds no source revision of yet is a removal.
func SourceEntry(branch string, a meshdelivery.SourceAnswer) (Entry, error) {
heads := map[string]string{}
for _, h := range a.Repositories {
heads[h.Repository] = h.Head
}
e := Entry{Bucket: SourceBucket, Key: keys.EscapeName(branch), Revision: uint64(a.Revision)}
if a.Revision == 0 && len(heads) == 0 { // the gate holds no source revision of the branch yet
e.Deleted = true
return e, nil
}
v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: a.Group, Heads: heads})
if err != nil {
return Entry{}, err
}
e.Value = v
return e, nil
}
// SourcesOf is Reader.Sources over the watched branches followed now: each one's newest source revision through call.
func SourcesOf(call func(key string, body any) (json.RawMessage, error), followed func() []string) func() (map[string]uint64, error) {
return func() (map[string]uint64, error) {
out := map[string]uint64{}
for _, branch := range followed() {
a, err := SourceOf(call, branch)
if err != nil {
return nil, err
}
out[branch] = uint64(a.Revision)
}
return out, nil
}
}
modules/mesh-delivery-view/internal/view/tiers.go
+40 -12
View File
@@ -41,6 +41,8 @@ const TiersMost = 500
// could not be read (then every module is in the first tier), Cycle a cycle among them (its modules form the last
// tier), and Away the nodes left out.
type Tiers struct {
// Branch is the watched branch whose newest source revision it is, when view_tiers was given no revision (A-173).
Branch string `json:"branch,omitempty"`
Revision uint64 `json:"revision"`
Modules int `json:"modules"`
Tiers [][]TierModule `json:"tiers"`
@@ -76,20 +78,43 @@ type TierWait struct {
Since time.Time `json:"since,omitzero"`
}
// Tiers works out the tiers of a source revision, the gate's newest when revision is 0.
// BranchTiers is view_tiers' answer given no revision: the tiers of each watched branch's newest source revision, in branch
// order (design 47 G A-173).
type BranchTiers struct {
Branches []Tiers `json:"branches"`
}
// Newest works out the tiers of each watched branch's newest source revision, as the gate's verb source answers it.
func (r Reader) Newest() (BranchTiers, error) {
if r.Sources == nil {
return BranchTiers{}, fmt.Errorf("the gate's source cannot be read here: name the source revision")
}
revs, err := r.Sources()
if err != nil {
return BranchTiers{}, fmt.Errorf("the gate's source: %w", err)
}
out := BranchTiers{Branches: []Tiers{}}
for _, branch := range slices.Sorted(maps.Keys(revs)) {
if revs[branch] == 0 {
continue
}
t, err := r.Tiers(revs[branch])
if err != nil {
return BranchTiers{}, fmt.Errorf("%s: %w", branch, err)
}
t.Branch = branch
out.Branches = append(out.Branches, t)
}
if len(out.Branches) == 0 {
return BranchTiers{}, fmt.Errorf("the gate holds no source revision of any watched branch yet")
}
return out, nil
}
// Tiers works out the tiers of one source revision (not 0).
func (r Reader) Tiers(revision uint64) (Tiers, error) {
if revision == 0 {
if r.SourceState.GetEntry == nil {
return Tiers{}, fmt.Errorf("the gate's source cannot be read here: name the source revision")
}
_, rev, err := r.SourceState.GetEntry(trunkSourceKey)
if err != nil {
return Tiers{}, fmt.Errorf("the gate's source: %w", err)
}
if rev == 0 {
return Tiers{}, fmt.Errorf("the gate holds no source revision yet")
}
revision = rev
return Tiers{}, fmt.Errorf("name the source revision")
}
modules, err := values(r.State)
if err != nil {
@@ -453,5 +478,8 @@ func (r Reader) TiersOf(args map[string]any) (any, error) {
if err != nil {
return nil, err
}
if revision == 0 {
return r.Newest()
}
return r.Tiers(revision)
}
modules/mesh-delivery-view/internal/view/view.go
+98 -41
View File
@@ -26,6 +26,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildrecord"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/trunk"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate"
"git.novox.be/novox/mesh-sdk/go/keys"
"git.novox.be/novox/mesh-sdk/go/motion"
)
@@ -36,8 +37,12 @@ type Entry struct {
Value []byte
At time.Time
Deleted bool
// Revision is the bucket revision of the change, when the watch gives one (the source revision is the gate's source key's).
// Revision is the bucket revision of the change, when the watch gives one; for a branch's source (SourceBucket, keyed by the
// escaped branch) it is the source revision the gate's verb source answers.
Revision uint64
// Branch is, for a module's entry (ModulesBucket), the watched branch it is read from: the first branch of the module's
// registration, before the first registrations the first branch the gate's source answers that lists it (design 47 G A-173).
Branch string
// History says the change comes from a watch with history (ReachHistory): it feeds only the history the reach reads.
History bool
}
@@ -87,9 +92,9 @@ type Snapshot struct {
Asks []AskRow `json:"asks"`
// Wants is every want of the assigner beside the have of its node and where it stands (novox/hq ADR 0539 §8, issue 1906 D7).
Wants []WantRow `json:"wants"`
// Source is the state of the mesh's source revision, derived from the wants and the haves (ADR 0429 §1): recorded by no part.
// Nil when the gate holds no source yet.
Source *SourceRow `json:"source,omitempty"`
// Sources is the state of each watched branch's source revision, derived from the wants and the haves (ADR 0429 §1;
// design 47 G A-173): recorded by no part, by branch name. Empty while the gate's verb source answers no branch's.
Sources []SourceRow `json:"sources,omitempty"`
// Backlog is, for each language and kind of build ask with one waiting or held, the asks waiting for a node, the asks a node
// holds and the oldest waiting ask's age (issue 1887 A8a).
Backlog []BacklogRow `json:"backlog"`
@@ -129,10 +134,12 @@ const (
SourceNotDelivered = "not delivered"
)
// SourceRow is the mesh's source revision (the revision of the gate's source key) and whether it is delivered: every want equals its
// module's fingerprint at the indexer and every have equals its want, running and healthy, and every node a module is assigned to
// has a want. Waiting names each module on a node that is not there yet, and why.
// SourceRow is one watched branch's source revision (as the gate's verb source answers it) and whether it is delivered: every
// want of a module whose branch it is equals its module's fingerprint in the catalog half and every have equals its want, running
// and healthy, and every node such a module is assigned to has a want. Waiting names each module on a node that is not there
// yet, and why.
type SourceRow struct {
Branch string `json:"branch"`
Revision uint64 `json:"revision"`
State string `json:"state"`
Waiting []string `json:"waiting,omitempty"`
@@ -235,12 +242,13 @@ type Model struct {
modules map[string]moduleEntry // the indexer's modules, by module
keys map[string]keyEntry // the builder's build keys, by <module>.<fingerprint>
wants map[string]wantEntry // the assigner's wants, by <node>.<module>
source *sourceEntry // the gate's source key
sources map[string]*sourceEntry // each watched branch's source, by escaped branch (design 47 G A-173)
modBranch map[string]string // the escaped branch each module's entry is read from, by module
deliveredEvent func(DeliveredEvent) error
motionEvent func(MotionEvent) error
told map[string]bool // the motion events said, by their ids
initial map[string]bool // the buckets whose watch handed over what stood at its start
said uint64 // the source revision whose delivery was said
said map[string]uint64 // the source revision whose delivery was said, by escaped branch
aasks map[string]assignerAskRecord // the assigner's asks (hold, restore), by key
chosen map[string]viewstate.ViewAnswer // the answers the view wrote, by ask key
writer Writer
@@ -256,7 +264,8 @@ type Model struct {
stuck map[string]stuckEntry
// how far each repository reached (reach.go, ADR 0556)
modHist map[string][]meshcatalog.ModulesByModuleValue // the indexer's values of each module, oldest first
srcHist map[uint64]map[string]string // the source's heads, by revision
srcHist map[string]map[uint64]map[string]string // each branch's source heads heard since the start, by escaped branch and revision
kept map[string]viewstate.Reach // the reach the view's own state held at its start, by repository key (A-225)
taken map[string]takein.Taken // the indexer's taken, by repository key
reach map[string]viewstate.Reach // the reach derived last, by repository key
reachSaid map[string]uint64 // the reached head's source revision said last, by repository key
@@ -326,6 +335,7 @@ type wantEntry struct {
}
type sourceEntry struct {
branch string // the branch's name
revision uint64
group string
heads map[string]string
@@ -359,7 +369,8 @@ func New() *Model {
return &Model{placed: meshmoduleassignerclient.NewPlaced(), builds: map[string]buildEntry{}, groups: map[string]groupEntry{}, modules: map[string]moduleEntry{}, keys: map[string]keyEntry{}, wants: map[string]wantEntry{}, told: map[string]bool{}, initial: map[string]bool{}, aasks: map[string]assignerAskRecord{}, chosen: map[string]viewstate.ViewAnswer{}, answers: map[string]string{}, answerOf: map[string]meshdeliveryclient.Answer{}, holds: map[string]heldEntry{}, newest: map[string]string{},
haves: map[string]haveEntry{}, nodes: map[string]nodeEntry{},
presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{},
modHist: map[string][]meshcatalog.ModulesByModuleValue{}, srcHist: map[uint64]map[string]string{}, taken: map[string]takein.Taken{},
modHist: map[string][]meshcatalog.ModulesByModuleValue{}, srcHist: map[string]map[uint64]map[string]string{}, kept: map[string]viewstate.Reach{},
sources: map[string]*sourceEntry{}, modBranch: map[string]string{}, said: map[string]uint64{}, taken: map[string]takein.Taken{},
reach: map[string]viewstate.Reach{}, reachSaid: map[string]uint64{}, reachNotYet: map[string]string{}}
}
@@ -367,21 +378,23 @@ func New() *Model {
// the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released.
func (m *Model) Apply(e Entry) {
m.mu.Lock()
var event *DeliveredEvent
var events []DeliveredEvent
var motion []MotionEvent
if e.History {
m.applyHistory(e)
} else {
m.apply(e)
event = m.deliveredNow()
events = m.deliveredNow()
motion = m.motionNow(e)
}
writes, reached := m.reachNow()
say, put := m.deliveredEvent, m.reachPut
sayMotion := m.motionEvent
m.mu.Unlock()
if event != nil && say != nil && say(*event) != nil {
m.unsay(*event)
for _, event := range events {
if say != nil && say(event) != nil {
m.unsay(event)
}
}
m.sendReach(put, say, writes, reached)
m.sayMotion(sayMotion, motion)
@@ -449,8 +462,8 @@ const unsaidMark = "\x00not said"
func (m *Model) unsay(ev DeliveredEvent) {
m.mu.Lock()
defer m.mu.Unlock()
if m.said == ev.Revision {
m.said = 0
if b := keys.EscapeName(ev.Branch); m.said[b] == ev.Revision {
delete(m.said, b)
}
}
@@ -462,6 +475,8 @@ func (m *Model) unsay(ev DeliveredEvent) {
// is the source revision that named that head and Group is empty; a consumer takes every earlier merge of that repository as on
// the machines.
type DeliveredEvent struct {
// Branch is the watched branch whose source revision it is (design 47 G A-173).
Branch string `json:"branch"`
Revision uint64 `json:"revision"`
Group string `json:"group"`
Heads map[string]string `json:"heads"`
@@ -478,18 +493,26 @@ func (m *Model) SetDeliveredEvent(say func(DeliveredEvent) error) {
m.deliveredEvent = say
}
// deliveredNow is the event of the source revision when it is delivered now and has not been said, or nil. A revision that went
// bad after it was delivered and recovered is not said again.
func (m *Model) deliveredNow() *DeliveredEvent {
if m.deliveredEvent == nil || m.source == nil || m.source.err != "" || m.said == m.source.revision {
// deliveredNow is the event of each watched branch's source revision that is delivered now and has not been said, in branch
// order. A revision that went bad after it was delivered and recovered is not said again.
func (m *Model) deliveredNow() []DeliveredEvent {
if m.deliveredEvent == nil {
return nil
}
row := m.sourceRow()
if row == nil || row.State != SourceDelivered || row.Wants == 0 {
return nil
var out []DeliveredEvent
for _, b := range slices.Sorted(maps.Keys(m.sources)) {
src := m.sources[b]
if src.err != "" || m.said[b] == src.revision {
continue
}
row := m.sourceRow(b)
if row == nil || row.State != SourceDelivered || row.Wants == 0 {
continue
}
m.said[b] = src.revision
out = append(out, DeliveredEvent{Branch: src.branch, Revision: src.revision, Group: src.group, Heads: src.heads})
}
m.said = m.source.revision
return &DeliveredEvent{Revision: m.source.revision, Group: m.source.group, Heads: m.source.heads}
return out
}
func (m *Model) apply(e Entry) {
@@ -542,10 +565,16 @@ func (m *Model) apply(e Entry) {
case ModulesBucket:
if e.Deleted {
delete(m.modules, e.Key)
delete(m.modBranch, e.Key)
return
}
mod, err := meshcatalogclient.DecodeModule(e.Value)
m.modules[e.Key] = moduleEntry{value: mod, err: errText(err)}
if e.Branch != "" {
m.modBranch[e.Key] = keys.EscapeName(e.Branch)
} else {
delete(m.modBranch, e.Key)
}
case AssignerAsksBucket:
if e.Deleted {
delete(m.aasks, e.Key)
@@ -593,16 +622,31 @@ func (m *Model) apply(e Entry) {
} else {
delete(m.taken, e.Key)
}
case SourceBucket:
if e.Key != trunkSourceKey {
case SourceBucket: // one watched branch's source, keyed by the escaped branch (design 47 G A-173)
if e.Deleted {
delete(m.sources, e.Key)
delete(m.said, e.Key)
return
}
if e.Deleted {
m.source = nil
name, err := keys.UnescapeName(e.Key)
if err != nil {
return
}
src, err := meshdeliveryclient.DecodeSource(e.Value)
m.source = &sourceEntry{revision: e.Revision, group: src.Group, heads: src.Heads, err: errText(err)}
m.sources[e.Key] = &sourceEntry{branch: name, revision: e.Revision, group: src.Group, heads: src.Heads, err: errText(err)}
if err == nil && e.Revision != 0 { // the history the reach reads: each revision heard since the start (A-225)
if m.srcHist[e.Key] == nil {
m.srcHist[e.Key] = map[uint64]map[string]string{}
}
m.srcHist[e.Key][e.Revision] = src.Heads
}
case viewstate.ReachBucket: // the view's own state reach, as it stood at the start (A-225)
if m.initial[viewstate.ReachBucket] {
return
}
if r, err := viewstate.DecodeReach(e.Value); err == nil && !e.Deleted {
m.kept[e.Key] = r
}
case AnswersBucket:
if e.Deleted {
delete(m.answers, e.Key)
@@ -791,7 +835,7 @@ func (m *Model) snapshot() Snapshot {
return unreadable[i].Bucket+"."+unreadable[i].Key < unreadable[j].Bucket+"."+unreadable[j].Key
})
return Snapshot{Rows: rows, Nodes: nodes, Offline: offline, Stuck: stuck, Unreadable: unreadable,
Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows(), Wants: m.wantRows(), Source: m.sourceRow()}
Groups: m.groupRows(), Fingerprints: m.fingerprintRows(), Asks: m.askRows(), Wants: m.wantRows(), Sources: m.sourceRows()}
}
// groupRows are the readable candidate groups, ordered by group, their pull requests by repository and number.
@@ -1033,12 +1077,25 @@ func (m *Model) wantRows() []WantRow {
return out
}
// sourceRow derives the state of the source revision the gate holds (ADR 0429 §1; recorded by no part): delivered when every want
// equals its module's fingerprint at the indexer and every have equals its want, running and healthy, and every node a module is
// assigned to has a want, compared only on the nodes present: a node the watchman says is offline is named away and left out
// (ADR 0603 §7, changing ADR 0429 §1). It is nil when the gate holds no readable source.
func (m *Model) sourceRow() *SourceRow {
if m.source == nil || m.source.err != "" {
// sourceRows are the rows of every watched branch's readable source, in branch order (design 47 G A-173).
func (m *Model) sourceRows() []SourceRow {
var out []SourceRow
for _, b := range slices.Sorted(maps.Keys(m.sources)) {
if row := m.sourceRow(b); row != nil {
out = append(out, *row)
}
}
return out
}
// sourceRow derives the state of one watched branch's source revision (ADR 0429 §1; recorded by no part), branch escaped:
// delivered when every want of a module whose branch it is equals its module's fingerprint and every have equals its want,
// running and healthy, and every node such a module is assigned to has a want, compared only on the nodes present: a node the
// watchman says is offline is named away and left out (ADR 0603 §7, changing ADR 0429 §1). It is nil when the gate's verb
// answers no readable source of the branch.
func (m *Model) sourceRow(branch string) *SourceRow {
src := m.sources[branch]
if src == nil || src.err != "" {
return nil
}
waiting := []string{}
@@ -1057,7 +1114,7 @@ func (m *Model) sourceRow() *SourceRow {
continue
}
mod, ok := m.modules[module]
if w.err != "" || w.value.Removed || !ok || mod.err != "" || mod.value.Fp2 == "" {
if w.err != "" || w.value.Removed || !ok || mod.err != "" || mod.value.Fp2 == "" || m.modBranch[module] != branch {
continue
}
looked++
@@ -1081,12 +1138,12 @@ func (m *Model) sourceRow() *SourceRow {
if _, ok := m.wants[wk]; ok {
continue
}
if mod, ok := m.modules[a.Module]; ok && mod.err == "" && mod.value.Fp2 != "" {
if mod, ok := m.modules[a.Module]; ok && mod.err == "" && mod.value.Fp2 != "" && m.modBranch[a.Module] == branch {
waiting = append(waiting, fmt.Sprintf("%s on %s has no want yet", a.Module, a.Node))
}
}
sort.Strings(waiting)
row := &SourceRow{Revision: m.source.revision, State: SourceDelivered, Wants: looked, Away: slices.Sorted(maps.Keys(away))}
row := &SourceRow{Branch: src.branch, Revision: src.revision, State: SourceDelivered, Wants: looked, Away: slices.Sorted(maps.Keys(away))}
if len(waiting) > 0 {
row.State, row.Waiting = SourceNotDelivered, waiting
}
modules/mesh-delivery-view/internal/view/wants_test.go
+49 -10
View File
@@ -14,6 +14,7 @@ import (
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery"
"git.novox.be/novox/mesh-sdk/go/keys"
)
const (
@@ -56,16 +57,32 @@ func moduleE(t *testing.T, module, fp string) Entry {
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: ModulesBucket, Key: module, Value: v, At: t0}
return Entry{Bucket: ModulesBucket, Key: module, Value: v, At: t0, Branch: testBranch}
}
func sourceE(t *testing.T, revision uint64) Entry {
// testBranch is the watched branch of these tests: no branch is named in the view's code (design 47 G §532, A-173).
const testBranch = "release"
func sourceE(t *testing.T, revision uint64) Entry { return sourceOn(t, testBranch, revision) }
// sourceOn is one watched branch's source as the gate's verb source answers it (A-173).
func sourceOn(t *testing.T, branch string, revision uint64) Entry {
t.Helper()
v, err := meshdeliveryclient.EncodeSource(meshdelivery.SourceByBranchValue{Group: "g", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}})
if err != nil {
t.Fatal(err)
}
return Entry{Bucket: SourceBucket, Key: trunkSourceKey, Value: v, At: t0, Revision: revision}
return Entry{Bucket: SourceBucket, Key: keys.EscapeName(branch), Value: v, At: t0, Revision: revision}
}
// sourceRowOf is the snapshot's source row of one branch, nil when it has none.
func sourceRowOf(m *Model, branch string) *SourceRow {
for _, r := range m.Snapshot().Sources {
if r.Branch == branch {
return &r
}
}
return nil
}
// 1906/D7 AC7: every want is a row beside the have of its node, with where it stands.
@@ -107,44 +124,66 @@ func TestEveryWantIsARowAgainstItsHave(t *testing.T) {
// want, healthy; one failed have shows it not delivered and names the module.
func TestASourceRevisionIsDerivedFromWantsAndHaves(t *testing.T) {
m := New()
var said []DeliveredEvent
m.SetDeliveredEvent(func(e DeliveredEvent) error {
if e.Repository == "" {
said = append(said, e)
}
return nil
})
fill(m, sourceE(t, 14355), moduleE(t, "web", fpNew), moduleE(t, "mail", fpNew),
place(t, "ace", "web"), place(t, "g14", "web"),
wantE(t, "ace", "web", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "ace", "web", fpNew, nodeengineclient.HealthOK),
wantE(t, "g14", "web", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "g14", "web", fpNew, nodeengineclient.HealthOK),
wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpNew}), haveFP(t, "ace", "mail", fpNew, nodeengineclient.HealthOK))
s := m.Snapshot().Source
s := sourceRowOf(m, testBranch)
if s == nil || s.Revision != 14355 || s.State != SourceDelivered || len(s.Waiting) != 0 {
t.Fatalf("source = %+v, want revision 14355 delivered", s)
}
// design 47 G A-173: each watched branch's source has its own row and its own delivered, over the modules whose branch it
// is: a module of another branch not yet run holds that branch's revision alone.
cms := moduleE(t, "cms", fpNew)
cms.Branch = "beta"
fill(m, sourceOn(t, "beta", 14360), cms, wantE(t, "ace", "cms", meshwantsclient.Want{Fp: fpNew}))
if b := sourceRowOf(m, "beta"); b == nil || b.Revision != 14360 || b.State != SourceNotDelivered || b.Wants != 1 {
t.Errorf("the branch beta's source = %+v, want revision 14360 not delivered over its one want", b)
}
if s := sourceRowOf(m, testBranch); s == nil || s.State != SourceDelivered || s.Wants != 3 {
t.Errorf("the branch %s's source = %+v, want delivered over its own three wants", testBranch, s)
}
fill(m, haveFP(t, "ace", "cms", fpNew, nodeengineclient.HealthOK))
if len(said) != 2 || said[0].Branch != testBranch || said[0].Revision != 14355 || said[1].Branch != "beta" || said[1].Revision != 14360 {
t.Errorf("delivered said = %+v, want %s at 14355, then beta at 14360", said, testBranch)
}
fill(m, haveFP(t, "g14", "web", fpNew, nodeengineclient.HealthFailed))
s = m.Snapshot().Source
s = sourceRowOf(m, testBranch)
if s.State != SourceNotDelivered || len(s.Waiting) != 1 || !strings.Contains(s.Waiting[0], "web") || !strings.Contains(s.Waiting[0], "g14") {
t.Errorf("a failed have: %+v", s)
}
fill(m, haveFP(t, "g14", "web", fpNew, nodeengineclient.HealthOK), wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpOld}))
s = m.Snapshot().Source
s = sourceRowOf(m, testBranch)
if s.State != SourceNotDelivered || !strings.Contains(strings.Join(s.Waiting, ";"), "mail") {
t.Errorf("a want that is not the module's fingerprint: %+v", s)
}
fill(m, wantE(t, "ace", "mail", meshwantsclient.Want{Fp: fpNew}), place(t, "novox", "web"))
s = m.Snapshot().Source
s = sourceRowOf(m, testBranch)
if s.State != SourceNotDelivered || !strings.Contains(strings.Join(s.Waiting, ";"), "novox") {
t.Errorf("a node assigned the module with no want: %+v", s)
}
// ADR 0603 §7: a node away holds nothing back and is named; it counts once it returns and runs its wants.
fill(m, presenceE(t, "novox", false, t0))
s = m.Snapshot().Source
s = sourceRowOf(m, testBranch)
if s.State != SourceDelivered || len(s.Away) != 1 || s.Away[0] != "novox" {
t.Errorf("a node away: %+v", s)
}
fill(m, presenceE(t, "novox", true, t0))
if s = m.Snapshot().Source; s.State != SourceNotDelivered || len(s.Away) != 0 {
if s = sourceRowOf(m, testBranch); s.State != SourceNotDelivered || len(s.Away) != 0 {
t.Errorf("the node returned and runs nothing yet: %+v", s)
}
}
func TestNoSourceRevisionHeldMeansNoStateIsDerived(t *testing.T) {
if s := New().Snapshot().Source; s != nil {
if s := New().Snapshot().Sources; len(s) != 0 {
t.Errorf("source = %+v", s)
}
}
modules/mesh-delivery-view/manifest_test.go
+9 -10
View File
@@ -132,7 +132,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
if m.Module != "mesh-delivery-view" {
t.Errorf("module = %q", m.Module)
}
want := []string{"mesh-conditions.conditions", "mesh-config.resolved", "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.source", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-healers.attempts", "mesh-provisioner.provisions", "mesh-vault.signers", "node-build-go.pauses", "node-build-go.steps", "node-build-image.pauses", "node-build-image.steps", "node-build-node.pauses", "node-build-node.steps", "node-engine.have"} // the build seats' states by seat, never a build module (design 47 G A-82); // no reads of the indexer's or the assigner's states (ADR 0615 §1, DLV-M4); view_tiers reads the module graph and the provisioner's keys (ADR 0603 §6); view_status reads the keeper's conditions and the healers' attempts (1952/V1); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
want := []string{"mesh-conditions.conditions", "mesh-config.resolved", "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-healers.attempts", "mesh-provisioner.provisions", "mesh-vault.signers", "node-build-go.pauses", "node-build-go.steps", "node-build-image.pauses", "node-build-image.steps", "node-build-node.pauses", "node-build-node.steps", "node-engine.have"} // the build seats' states by seat, never a build module (design 47 G A-82); no read of the gate's source: each watched branch's is its verb source's (§512, A-173); // no reads of the indexer's or the assigner's states (ADR 0615 §1, DLV-M4); view_tiers reads the module graph and the provisioner's keys (ADR 0603 §6); view_status reads the keeper's conditions and the healers' attempts (1952/V1); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
got := slices.Clone(m.Reads)
slices.Sort(got)
if !reflect.DeepEqual(got, want) {
@@ -151,13 +151,13 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
{Verb: "mesh-delivery-indexer.indexer_read_packages"}, {Verb: "mesh-delivery-indexer.indexer_read_predefined"}, {Verb: "mesh-delivery-indexer.indexer_read_taken"},
{Verb: "mesh-delivery-teller.teller_read_answers"}, // A-175: the teller's answers by their views family
{Verb: "seat:mesh-catalog.revision", Fields: []string{"none-yet"}},
{Verb: "seat:mesh-catalog.registrations"}, // §614: the watched branches of the registrations, then registration-changed
{Verb: "seat:mesh-catalog.entry", Fields: []string{"manifest-digest"}}, // §558: the module's value is the entry's manifest-digest (§506)
{Verb: "seat:mesh-delivery.source", Fields: []string{"branches"}}, // §507, §614: before the first registrations, the gate's source's branches // ADR 0663: the modules are the catalog half's, read through the seat's verb revision, never the indexer's
{Verb: "seat:mesh-catalog.seat-definitions"}, // design 47 G A-82, §204: the build seats are read from the seat list
{Verb: "seat:mesh-hand-acts.hand-act"}, // the join: its first view asked again by hand is a hand act (views.StatusTools)
{Verb: "seat:mesh-module-assigner.assignments"}, // A-82: the holder of each build seat
{Verb: "seat:node-engine.manifest-reads"}}; !reflect.DeepEqual(m.Invokes, want) { // ADR 0618 §3: node_manifest_reads asks the node-engine seat's verb of every node; 1585/T24: view_tiers asks indexer_pending, a verb edge (ADR 0614 §2)
{Verb: "seat:mesh-catalog.registrations"}, // §614: the watched branches of the registrations, then registration-changed
{Verb: "seat:mesh-catalog.entry", Fields: []string{"manifest-digest"}}, // §558: the module's value is the entry's manifest-digest (§506)
{Verb: "seat:mesh-delivery.source", Fields: []string{"branches", "group", "repositories", "revision"}}, // §507, §614, A-173: the branches before the first registrations, and each watched branch's source (§558: fields G adds) // ADR 0663: the modules are the catalog half's, read through the seat's verb revision, never the indexer's
{Verb: "seat:mesh-catalog.seat-definitions"}, // design 47 G A-82, §204: the build seats are read from the seat list
{Verb: "seat:mesh-hand-acts.hand-act"}, // the join: its first view asked again by hand is a hand act (views.StatusTools)
{Verb: "seat:mesh-module-assigner.assignments"}, // A-82: the holder of each build seat
{Verb: "seat:node-engine.manifest-reads"}}; !reflect.DeepEqual(m.Invokes, want) { // ADR 0618 §3: node_manifest_reads asks the node-engine seat's verb of every node; 1585/T24: view_tiers asks indexer_pending, a verb edge (ADR 0614 §2)
t.Errorf("invokes = %v, want %v", m.Invokes, want)
}
// design 47 G §387: the keeper's ask for the findings is heard, the finding stream declared (§383), and §368: version is MAJOR.MINOR.
@@ -202,8 +202,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
// D-11: the holders and buckets the view names are its manifest's (no contract names a holder, novox/hq ADR 0634 §1): a
// state it reads directly is in `reads`, one it follows is a family it consumes, and the owners' tools it asks are invoked.
t.Run("the names are the manifest's", func(t *testing.T) {
direct := map[string]string{view.SourceBucket: view.GateModule + ".source",
view.PresenceBucket: view.WatchmanModule + ".presence", view.StuckBucket: view.WatchmanModule + ".stuck", view.HaveBucket: view.EngineModule + ".have"}
direct := map[string]string{view.PresenceBucket: view.WatchmanModule + ".presence", view.StuckBucket: view.WatchmanModule + ".stuck", view.HaveBucket: view.EngineModule + ".have"}
for bucket, read := range direct {
if strings.ReplaceAll(read, ".", "_") != bucket {
t.Errorf("the bucket %s is not the state %s", bucket, read)
modules/mesh-delivery-view/module.json
+1 -2
View File
@@ -89,7 +89,7 @@
{"verb": "seat:mesh-catalog.revision", "fields": ["none-yet"]},
"seat:mesh-catalog.registrations",
{"verb": "seat:mesh-catalog.entry", "fields": ["manifest-digest"]},
{"verb": "seat:mesh-delivery.source", "fields": ["branches"]},
{"verb": "seat:mesh-delivery.source", "fields": ["branches", "group", "repositories", "revision"]},
"seat:mesh-catalog.seat-definitions",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-module-assigner.assignments",
@@ -106,7 +106,6 @@
"mesh-delivery-builder.build-log",
"mesh-delivery-builder.builds",
"mesh-delivery-checker.tries",
"mesh-delivery-gate.source",
"mesh-conditions.conditions",
"mesh-healers.attempts",
"mesh-provisioner.provisions",