From 8597083493a32f4a1ee159c8ca6e7cf3449baa2d Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 09:26:49 +0100 Subject: [PATCH] AS6: the assigner says version-delivered and holds a dependent for its dependency's delivery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit At its start it reads whole, from the first message, version-released, its own version-delivered and the watchman's machine-away and machine-back (process.FromFirst), then hears them; it reads the watchman's presence state no more, a machine being present unless said away. A module's newest released version is said version-delivered once every present machine placing it on its release branch runs healthy the want naming it, the machines away named, none when every one is away. A dependent whose want changes fingerprint waits, by an edge of kind 2, 4 or 6, for its dependency's newest released version on its branch not yet said delivered: wait-held step delivery, bound 60 minutes from the release, kept in the state delivery-waits, ended ran on the delivery or released once assigner_release writes it. Builds: design 47 G §376, §389, §498, §374, A-189, A-233, A-246 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../cmd/mesh-delivery-assigner/buildended.go | 23 +- .../cmd/mesh-delivery-assigner/delivered.go | 380 ++++++++++++++++++ .../mesh-delivery-assigner/delivered_test.go | 166 ++++++++ .../cmd/mesh-delivery-assigner/halves.go | 1 + .../integration_test.go | 32 +- .../cmd/mesh-delivery-assigner/main.go | 23 +- .../cmd/mesh-delivery-assigner/pass.go | 5 + .../cmd/mesh-delivery-assigner/pass_test.go | 33 +- .../mesh-delivery-assigner/pipelines_test.go | 8 +- .../cmd/mesh-delivery-assigner/world.go | 89 ++-- .../internal/buckets/buckets.go | 8 +- .../internal/buckets/buckets_test.go | 6 +- .../mesh-delivery-assigner/manifest_test.go | 12 +- modules/mesh-delivery-assigner/module.json | 5 +- 14 files changed, 699 insertions(+), 92 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go index f670eb63..8d7cbb3b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/buildended.go @@ -78,6 +78,8 @@ func (b *builtEnds) partsOf(module, fp string) (assign.KeptBuild, bool) { type releasedVersions struct { mu sync.Mutex seen map[string]assign.KeptBuild + // newest is, by module, the last version-released heard of it (delivered.go). + newest map[string]releaseHeard } func (r *releasedVersions) key(module string, source int64) string { @@ -103,6 +105,14 @@ func (r *releasedVersions) put(data []byte) error { r.seen = map[string]assign.KeptBuild{} } r.seen[r.key(e.Module, e.Source)] = assign.KeptBuild{Version: e.Version, Parts: parts} + if r.newest == nil { + r.newest = map[string]releaseHeard{} + } + h := releaseHeard{Version: e.Version, Parts: parts, Source: e.Source} + if e.At != nil { + h.At = *e.At + } + r.newest[e.Module] = h // the stream's order is the order of release return nil } @@ -114,19 +124,6 @@ func (r *releasedVersions) of(module string, source int64) (assign.KeptBuild, bo return k, ok } -// hearReleases registers the hub's taker of version-released, waking the pass; before the hub starts. -func hearReleases(hub *process.EventHub, released *releasedVersions, wake func(), logf func(string, ...any)) error { - return hub.Handle(meshcatalog.SubjectVersionReleased, func(m jetstream.Msg) { - if err := released.put(m.Data()); err != nil { - logf("%s is not kept: %v", m.Subject(), err) - } - if err := m.Ack(); err != nil { - logf("acknowledging %s: %v", m.Subject(), err) - } - wake() - }) -} - // buildsStreamName is the builder's stream of build ends as the bus names it: kept last per subject (one end per module and // fingerprint, its manifest's `last-per-subject`, design 47 G A-181, §703). var buildsStreamName = catalogue.StreamBusName(builds.BuildsStream) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered.go new file mode 100644 index 00000000..97ccbbf1 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered.go @@ -0,0 +1,380 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "slices" + "sort" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + wd "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" + "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/meshwants" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-assigner/internal/assign" + "mesh-delivery-assigner/internal/assignment" + "mesh-delivery-assigner/internal/buckets" +) + +// A delivery is waited for mesh-wide (design 47 G §376, §389, A-189, A-246). At its start the assigner reads whole, from their first +// message, the streams of the catalog seat's version-released and of its own version-delivered (mesh-sdk process.FromFirst, A-233), +// and the watchman's machine-away and machine-back, never its state; then it hears them. A released version of a module is said +// version-delivered on the seat mesh-wants once, when every present machine placing it on the branch it is released from runs, healthy, +// the want the assigner wrote naming that version; a version no present machine places is not delivered. A dependent whose want +// changes fingerprint waits, by an edge of kind 2, 4 or 6, for its dependency's newest released version not yet said delivered, on +// the branch it is released from; on another branch no version is released and nothing waits. The wait is said wait-held, step +// delivery, bound 60 minutes from the version's release, kept in the state delivery-waits under the dependent's name, and said +// wait-ended when the version is delivered (ran) or released by hand (assigner_release, released). + +// deliveryWaitBound is a dependent's delivery wait's bound, from its dependency's version-released. +const deliveryWaitBound = 60 * time.Minute + +// Wait-ended's how of a delivery wait. +const ( + deliveryWaitRan = "ran" + deliveryWaitReleased = "released" +) + +// releaseHeard is one version-released heard: the version, each part's digest, the source revision and when it was said. +type releaseHeard struct { + Version string + Parts map[string]string + Source int64 + At time.Time +} + +// newestOf is the newest version of module released, the last heard of it in the stream's order. +func (r *releasedVersions) newestOf(module string) (releaseHeard, bool) { + r.mu.Lock() + defer r.mu.Unlock() + h, ok := r.newest[module] + return h, ok +} + +// deliveredVersions are the version-delivered said, the assigner's own, by module and version: each is said once. +type deliveredVersions struct { + mu sync.Mutex + said map[string]bool +} + +func (d *deliveredVersions) key(module, version string) string { return module + "@" + version } + +// put takes one version-delivered in. +func (d *deliveredVersions) put(data []byte) error { + var e meshwants.VersionDeliveredEvent + if err := json.Unmarshal(data, &e); err != nil { + return fmt.Errorf("a version-delivered that is none: %w", err) + } + if e.Module == "" || e.Version == "" { + return fmt.Errorf("a version-delivered naming no module or version: %s", data) + } + d.mark(e.Module, e.Version) + return nil +} + +func (d *deliveredVersions) mark(module, version string) { + d.mu.Lock() + defer d.mu.Unlock() + if d.said == nil { + d.said = map[string]bool{} + } + d.said[d.key(module, version)] = true +} + +// has says whether version-delivered of module's version was said. +func (d *deliveredVersions) has(module, version string) bool { + d.mu.Lock() + defer d.mu.Unlock() + return d.said[d.key(module, version)] +} + +// presenceOf reads one machine-away or machine-back: the machine and when it was said. +func presenceOf(data []byte) (string, time.Time, error) { + var e wd.MachineAwayEvent // machine-back has the same fields + if err := json.Unmarshal(data, &e); err != nil { + return "", time.Time{}, err + } + if e.Machine == "" { + return "", time.Time{}, fmt.Errorf("it names no machine: %s", data) + } + return e.Machine, e.At, nil +} + +// hearDelivery reads whole, then follows, the four streams a delivery is judged from: version-released, version-delivered, +// machine-away and machine-back. A stream not heard is said as the assigner's finding, and what it carries is not known until +// the assigner starts again. +func hearDelivery(ctx context.Context, js jetstream.JetStream, w *world, wake func(), findings assign.Findings, logf func(string, ...any)) { + acct := process.Account{Node: thisNode, Module: assignment.Module} + nh := notHeardWriter(findings, logf) + for _, s := range []struct { + stream, words string + take func([]byte) error + }{ + {meshcatalog.StreamVersionReleased, "no want names a released version and no delivery is judged", w.released.put}, + {meshwants.StreamVersionDelivered, "a version said delivered before is said again once it is judged", w.delivered.put}, + {wd.StreamMachineAway, "no machine is known away, so a delivery waits for a machine that is away", func(data []byte) error { + m, at, err := presenceOf(data) + if err == nil { + w.heardPresence(m, true, at) + } + return err + }}, + {wd.StreamMachineBack, "a machine once away is not known back", func(data []byte) error { + m, at, err := presenceOf(data) + if err == nil { + w.heardPresence(m, false, at) + } + return err + }}, + } { + take := s.take + cc, err := acct.HearSeatEvents(ctx, js, s.stream, func(m jetstream.Msg) { + if err := take(m.Data()); err != nil { + logf("%s is not kept: %v", m.Subject(), err) + } + if err := m.Ack(); err != nil { + logf("acknowledging %s: %v", m.Subject(), err) + } + wake() + }, logf, process.FromFirst()) + if err != nil { + nh.NotHeard(s.stream, s.words+" until the assigner starts again", err) + continue + } + nh.Heard(s.stream) + go func() { + <-ctx.Done() + cc.Stop() + }() + } +} + +// releaseBranch is the branch module's versions are released from: the first watched branch of its registration (§371). +func (p *passer) releaseBranch(module string) string { + if p.ends.following() { + return p.ends.branchOf(module, "") + } + return p.w.trunkName() +} + +// placing is every machine placing module on the branch its versions are released from, sorted. +func (p *passer) placing(module string) []string { + branch := p.releaseBranch(module) + var out []string + for node, placed := range p.w.placedBranches()[module] { + if p.ends.following() { + placed = p.ends.branchOf(module, placed) + } else { + placed = p.w.trunkName() // the trunk-only form: every placement runs the trunk's + } + if placed == branch { + out = append(out, node) + } + } + sort.Strings(out) + return out +} + +// deliveryStanding is where module's version stands: the present placing machines not yet running it healthy as its want names +// it, and the placing machines away. +func (p *passer) deliveryStanding(module, version string) (notRunning, away []string, placing int) { + gone := p.w.awayNow() + n := assign.Nodes{Running: p.w.runningNow()} + wanted := p.w.wantsOf()[module] + nodes := p.placing(module) + for _, node := range nodes { + if gone[node] { + away = append(away, node) + continue + } + want, ok := wanted[node] + if !ok || want.Removed || want.Version != version || !n.RunsHealthy(node, module, want.Fp) { + notRunning = append(notRunning, node) + } + } + return notRunning, away, len(nodes) +} + +// judgeDelivered says version-delivered of each module's newest released version not said yet, once every present machine placing +// it runs it healthy; judged again at each pass, which each running, machine-away, machine-back and placement change wakes. +func (p *passer) judgeDelivered() { + for _, module := range p.releasedModules() { + rel, _ := p.w.released.newestOf(module) + if p.w.delivered.has(module, rel.Version) { + continue + } + notRunning, away, placing := p.deliveryStanding(module, rel.Version) + if placing == 0 || len(away) == placing || len(notRunning) > 0 { + continue + } + if err := p.sayDelivered(module, rel, away); err != nil { + p.logf("version-delivered of %s %s is not said, and is judged again at the next pass: %v", module, rel.Version, err) + } + } +} + +// sayDelivered says version-delivered of module's release rel, with the machines away, and keeps it said. +func (p *passer) sayDelivered(module string, rel releaseHeard, away []string) error { + at := p.clock().UTC() + body := meshwants.VersionDeliveredEvent{V: keys.Version, Module: module, Version: rel.Version, Parts: rel.Parts, Away: away, At: &at} + if err := p.wantsSay(meshwants.EventVersionDelivered, body, fmt.Sprintf("%s.%s.%s.%s", meshwants.EventVersionDelivered, module, rel.Version, at.Format(time.RFC3339Nano))); err != nil { + return err + } + p.w.delivered.mark(module, rel.Version) + return nil +} + +// releasedModules is every module a version-released was heard of, sorted. +func (p *passer) releasedModules() []string { + r := &p.w.released + r.mu.Lock() + defer r.mu.Unlock() + out := make([]string, 0, len(r.newest)) + for m := range r.newest { + out = append(out, m) + } + sort.Strings(out) + return out +} + +// deliveryWait is what a dependent waits for: its dependency, the version and the machines not yet running it. +type deliveryWait struct { + dependency string + rel releaseHeard + notRunning []string + away []string +} + +// waitsForDelivery says whether module, whose want on its nodes changes to fp on branch, waits for a dependency's delivery: by +// each edge of kind 2, 4 or 6 into it in branch's catalog half, the dependency's newest released version on branch not yet said +// delivered. A want that changes no fingerprint waits for nothing; on a branch no version is released from, nothing waits. +func (p *passer) waitsForDelivery(module, fp, branch string, nodes []string) (deliveryWait, bool) { + wanted := p.w.wantsOf()[module] + changes := false + for _, node := range nodes { + if w, ok := wanted[node]; !ok || w.Removed || w.Fp != fp { + changes = true + } + } + if !changes || fp == "" { + return deliveryWait{}, false + } + var deps []string + for _, e := range p.w.edgesOf(branch, module) { + if e.Kind != 2 && e.Kind != 4 && e.Kind != 6 { + continue + } + dep, _, _ := strings.Cut(e.From, ":") + if dep != "" && dep != module && !slices.Contains(deps, dep) { + deps = append(deps, dep) + } + } + sort.Strings(deps) + for _, dep := range deps { + if p.releaseBranch(dep) != branch { + continue + } + rel, ok := p.w.released.newestOf(dep) + if !ok || p.w.delivered.has(dep, rel.Version) { + continue + } + notRunning, away, _ := p.deliveryStanding(dep, rel.Version) + return deliveryWait{dependency: dep, rel: rel, notRunning: notRunning, away: away}, true + } + return deliveryWait{}, false +} + +// words is a delivery wait in plain words, as assigner_why_not and the status view say it. +func (d deliveryWait) words() string { + if len(d.notRunning) == 0 && len(d.away) > 0 { + return fmt.Sprintf("waiting for %s to be back", strings.Join(d.away, ", ")) + } + return fmt.Sprintf("waiting for delivery of %s %s on %s", d.dependency, d.rel.Version, strings.Join(d.notRunning, ", ")) +} + +// deliveryWaitBody is the wait-held of module's delivery wait d. +func deliveryWaitBody(module, fp string, d deliveryWait, since time.Time) meshwants.Wait { + from := d.rel.At.UTC() + if from.IsZero() { + from = since.UTC() + } + return meshwants.Wait{Step: meshwants.WaitStepDelivery, Owner: meshwants.Seat, Module: module, Version: fp, Part: d.dependency, + On: d.dependency + "@" + d.rel.Version, Machines: d.notRunning, Since: since.UTC(), ClockFrom: &from, Bound: deliveryWaitBound.String(), + Words: d.words()} +} + +// deliveryWaiting is one delivery wait the state delivery-waits keeps, by its dependent's name: the wait-held said, and a release +// by hand written by assigner_release. +type deliveryWaiting struct { + meshwants.Wait + // Released is set by assigner_release: the wait ends released and the dependent's want is written. + Released bool `json:"released,omitempty"` +} + +// saidDeliveryWaits is every delivery wait the state delivery-waits holds, by module; a value that does not decode is said and left. +func (p *passer) saidDeliveryWaits() map[string]deliveryWaiting { + out := map[string]deliveryWaiting{} + for key, value := range p.w.read(buckets.DeliveryWaits) { + var w deliveryWaiting + if err := json.Unmarshal(value, &w); err != nil || w.Module == "" { + p.logf("%s %s holds no delivery wait, left: %v", buckets.DeliveryWaits, key, err) + continue + } + out[w.Module] = w + } + return out +} + +// holdForDelivery says whether module's want, changing to fp on branch, is held for a dependency's delivery this pass: it says +// wait-held once and keeps it; a wait released by hand holds nothing; a wait that no longer holds is ended ran. +func (p *passer) holdForDelivery(ctx context.Context, module, fp, branch string, nodes []string) bool { + if p.kvs[buckets.DeliveryWaits] == nil { + return false + } + key := keys.EscapeName(module) + said, was := p.saidDeliveryWaits()[module] + d, waits := p.waitsForDelivery(module, fp, branch, nodes) + switch { + case was && said.Released: + p.endDeliveryWait(ctx, key, said, deliveryWaitReleased) // the want is written now, so it changes nothing at the next pass + return false + case !waits: + if was { + p.endDeliveryWait(ctx, key, said, deliveryWaitRan) + } + return false + } + body := deliveryWaitBody(module, fp, d, p.clock()) + if was && said.On == body.On && said.Version == fp && slices.Equal(said.Machines, body.Machines) { + return true + } + if was { + body.Since = said.Since + } + id := fmt.Sprintf("%s.delivery.%s.%s.%s", key, fp, body.On, strings.Join(body.Machines, ",")) + if err := p.wantsSay(meshwants.EventWaitHeld, body, id); err != nil { + p.logf("the delivery wait of %s is not said, and is said at the next pass: %v", module, err) + return true + } + kept := deliveryWaiting{Wait: body} + p.put(ctx, buckets.DeliveryWaits, key, func() ([]byte, error) { return json.Marshal(kept) }) + return true +} + +// endDeliveryWait says wait-ended of a delivery wait with how and removes it from the state. +func (p *passer) endDeliveryWait(ctx context.Context, key string, w deliveryWaiting, how string) { + body := w.Wait + body.How = how + if err := p.wantsSay(meshwants.EventWaitEnded, body, key+".delivery."+body.Version+"."+how); err != nil { + p.logf("the end of the delivery wait of %s is not said, and is said at the next pass: %v", w.Module, err) + return + } + p.remove(ctx, buckets.DeliveryWaits, key) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered_test.go new file mode 100644 index 00000000..1a60b297 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/delivered_test.go @@ -0,0 +1,166 @@ +package main + +import ( + "context" + "encoding/json" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + mwc "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/meshwants" + + "mesh-delivery-assigner/internal/buckets" +) + +// Design 47 G §376, §389, §498, A-189, A-246: a released version is said version-delivered once, when every present machine placing +// it runs healthy the want naming it, a machine away named and not waited for, none when every placing machine is away; a dependent +// whose want changes fingerprint waits by an edge of kind 2, 4 or 6 for its dependency's newest released version not yet said +// delivered, said wait-held (step delivery, 60 minutes from the release) once and kept in delivery-waits, and wait-ended ran once +// the version is delivered; a kind-1 edge, or a dependency with no version released, waits for nothing; one released by hand ends +// released and its want is written. +func TestADependentWaitsForItsDependencysDelivery(t *testing.T) { + ctx := context.Background() + at := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC) + waits := &memKV{values: map[string][]byte{}} + p := newPasser(kvSet{buckets.Wants: &memKV{values: map[string][]byte{}}, buckets.DeliveryWaits: waits}, at) + p.logf = t.Logf + var events []said + p.publishWants = func(event string, body any, id string) error { + raw, _ := json.Marshal(body) + events = append(events, said{event, id, raw}) + return nil + } + p.w.runningHeard = true + placeOn(t, p.w, "ace", "db", "app") + placeOn(t, p.w, "g14", "db") + fp := func(c string) string { return meshcatalogclient.SourcePrefix2 + strings.Repeat(c, 64) } + want := func(node, module, fp, version string) { + key, _ := mwc.WantKey(node, module) + raw, err := mwc.EncodeWant(mwc.Want{Fp: fp, Version: version, Self: true}) + if err != nil { + t.Fatal(err) + } + p.w.put(buckets.Wants, key, raw) + } + runs := func(node, module, fp string) { + raw, _ := json.Marshal(map[string]any{"v": 1, "node": node, "module": module, "fingerprint": fp, "at": at}) + if err := p.w.hearRunning("running", raw); err != nil { + t.Fatal(err) + } + } + release := func(module, version string) { + raw, _ := json.Marshal(meshcatalog.VersionReleasedEvent{V: 1, Module: module, Version: version, Source: 4, At: &at, + Artifacts: []meshcatalog.VersionArtifact{{Artifact: "program", Digest: "sha256:d"}}}) + if err := p.w.released.put(raw); err != nil { + t.Fatal(err) + } + } + saidOf := func(event string) []said { + var out []said + for _, e := range events { + if e.event == event { + out = append(out, e) + } + } + return out + } + want("ace", "app", fp("1"), "") + + // No version of db is released: app's new fingerprint waits for nothing. + p.w.putEdges(trunkBranch, "app", []meshcatalog.Edge{{Kind: 4, From: "db:program", To: "app:program"}}) + if p.holdForDelivery(ctx, "app", fp("2"), trunkBranch, []string{"ace"}) { + t.Fatal("app waits with no version of db released") + } + release("db", "1.1.0") + want("ace", "db", fp("d"), "1.1.0") + want("g14", "db", fp("d"), "1.1.0") + runs("ace", "db", fp("d")) + // A kind-1 edge orders nothing. + p.w.putEdges(trunkBranch, "app", []meshcatalog.Edge{{Kind: 1, From: "db:program", To: "app:program"}}) + if p.holdForDelivery(ctx, "app", fp("2"), trunkBranch, []string{"ace"}) { + t.Fatal("a kind-1 edge holds app") + } + p.w.putEdges(trunkBranch, "app", []meshcatalog.Edge{{Kind: 4, From: "db:program", To: "app:program"}}) + // app's want unchanged waits for nothing. + if p.holdForDelivery(ctx, "app", fp("1"), trunkBranch, []string{"ace"}) { + t.Fatal("app's unchanged want waits") + } + // g14 places db and does not run 1.1.0 yet: not delivered, and app's new version waits, said once. + p.judgeDelivered() + if !p.holdForDelivery(ctx, "app", fp("2"), trunkBranch, []string{"ace"}) || !p.holdForDelivery(ctx, "app", fp("2"), trunkBranch, []string{"ace"}) { + t.Fatal("app does not wait for db 1.1.0") + } + if len(saidOf(meshwants.EventVersionDelivered)) != 0 { + t.Fatalf("db is said delivered while g14 does not run it: %+v", events) + } + held := saidOf(meshwants.EventWaitHeld) + if len(held) != 1 { + t.Fatalf("wait-held said %d times, want once", len(held)) + } + var w meshwants.Wait + _ = json.Unmarshal(held[0].body, &w) + if w.Step != meshwants.WaitStepDelivery || w.Part != "db" || w.Module != "app" || w.Version != fp("2") || w.Bound != "1h0m0s" || + w.ClockFrom == nil || !w.ClockFrom.Equal(at) || len(w.Machines) != 1 || w.Machines[0] != "g14" || w.Words != "waiting for delivery of db 1.1.0 on g14" { + t.Errorf("the wait said is %+v", w) + } + if _, ok := waits.values["app"]; !ok { + t.Errorf("the wait is not kept in delivery-waits: %v", waits.values) + } + // g14 away: every present machine runs it, so it is delivered, g14 named away; once only. + p.w.heardPresence("g14", true, at) + p.judgeDelivered() + p.judgeDelivered() + delivered := saidOf(meshwants.EventVersionDelivered) + var d meshwants.VersionDeliveredEvent + if len(delivered) != 1 || json.Unmarshal(delivered[0].body, &d) != nil || d.Module != "db" || d.Version != "1.1.0" || + len(d.Away) != 1 || d.Away[0] != "g14" || d.Parts["db:program"] != "sha256:d" { + t.Fatalf("version-delivered said %+v", delivered) + } + // The wait ends ran, and app's want goes on. + if p.holdForDelivery(ctx, "app", fp("2"), trunkBranch, []string{"ace"}) { + t.Fatal("app still waits once db 1.1.0 is delivered") + } + ended := saidOf(meshwants.EventWaitEnded) + if len(ended) != 1 || json.Unmarshal(ended[0].body, &w) != nil || w.How != deliveryWaitRan { + t.Errorf("wait-ended said %+v", ended) + } + if _, ok := waits.values["app"]; ok { + t.Error("the ended wait is kept") + } + // A newer release with every placing machine away is not delivered, and its wait says the machines it waits to be back. + release("db", "1.2.0") + p.w.heardPresence("ace", true, at) + p.judgeDelivered() + if len(saidOf(meshwants.EventVersionDelivered)) != 1 { + t.Error("a version is said delivered while every placing machine is away") + } + if !p.holdForDelivery(ctx, "app", fp("3"), trunkBranch, []string{"ace"}) { + t.Fatal("app does not wait for db 1.2.0") + } + held = saidOf(meshwants.EventWaitHeld) + if json.Unmarshal(held[len(held)-1].body, &w) != nil || w.Words != "waiting for ace, g14 to be back" { + t.Errorf("the wait while every machine is away says %q", w.Words) + } + // Released by hand (assigner_release writes it): the wait ends released and app's want is written. + var kept deliveryWaiting + _ = json.Unmarshal(waits.values["app"], &kept) + kept.Released = true + raw, _ := json.Marshal(kept) + p.w.put(buckets.DeliveryWaits, "app", raw) + if p.holdForDelivery(ctx, "app", fp("3"), trunkBranch, []string{"ace"}) { + t.Fatal("a released wait holds app") + } + ended = saidOf(meshwants.EventWaitEnded) + if json.Unmarshal(ended[len(ended)-1].body, &w) != nil || w.How != deliveryWaitReleased { + t.Errorf("the released wait ended %+v", w) + } + // Presence: the later of machine-away and machine-back holds, whichever stream is read first. + p.w.heardPresence("ace", false, at.Add(time.Minute)) + p.w.heardPresence("ace", true, at) + if p.w.awayNow()["ace"] || !p.w.online()["ace"] { + t.Error("an older machine-away read after a newer machine-back leaves ace away") + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go index 83ce5f4f..7d07b518 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go @@ -110,6 +110,7 @@ func (h *halves) changed(branch, module string, raw json.RawMessage, deleted boo } v, n, m := fromEntry(e) h.w.putModule(branch, module, v, n, m) + h.w.putEdges(branch, module, e.Edges) } // fromEntry is a catalog entry as the value and the graph node the rules read, and its manifest parsed once under the digest of 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 1606a79c..6cd46b78 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 @@ -8,6 +8,8 @@ import ( "errors" "fmt" mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + wd "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "git.novox.be/novox/mesh-sdk/go/keys" "io" "maps" @@ -187,6 +189,17 @@ func startBed(t *testing.T, leaveOut string) *bed { if _, err := js.CreateOrUpdateConsumer(ctx, nodeengine.StreamRunning, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { t.Fatal(err) } + // The watchman's machine-away and machine-back and the assigner's own version-delivered, kept by age, which it reads whole from their + // first message at its start and then follows through its durable (design 47 G §376, §389, A-189, A-233). + for _, k := range []struct{ stream, subject string }{{wd.StreamMachineAway, wd.SubjectMachineAwayAll}, {wd.StreamMachineBack, wd.SubjectMachineBackAll}, + {meshwants.StreamVersionDelivered, meshwants.SubjectVersionDelivered}} { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: k.stream, Subjects: []string{k.subject}, MaxAge: 90 * 24 * time.Hour}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, k.stream, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + } // The builder's stream of build ends, kept last per subject (its manifest's `last-per-subject`, design 47 G A-181), which the assigner // reads whole at its start and follows through its durable, as the grants part makes it (from the last of each subject). if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buildsStreamName, Subjects: []string{process.ModuleStreamSubject(builds.BuilderModule, builds.BuildEndedEvent+".>")}, MaxMsgsPerSubject: 1}); err != nil { @@ -376,7 +389,7 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { } for deadline := time.Now().Add(15 * time.Second); ; time.Sleep(200 * time.Millisecond) { // A change of a bucket it watches starts a pass, which opens the puts' bucket. - if _, err := b.buckets[buckets.Presence].Put(b.ctx, "n1", []byte(time.Now().String())); err != nil { + if _, err := b.buckets[buckets.Have].Put(b.ctx, "n1.pulse", []byte(time.Now().String())); err != nil { t.Fatal(err) } if e, err := putsKV.Get(b.ctx, pkey); err == nil { @@ -566,8 +579,6 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { runs := map[string]string{"ace": fOld, "g14": "build-9", "shanks": fOld} for _, node := range []string{"ace", "g14", "shanks"} { b.compose(t, node, 'c') - on, _ := presenceValue(true) - b.putJSON(t, buckets.Presence, node, on) have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: runs[node], Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) b.putJSON(t, buckets.Have, node+".web", have) b.place(t, node, "web") @@ -894,8 +905,6 @@ func TestAHeldFingerprintIsAskedAbout(t *testing.T) { 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.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"}) b.putJSON(t, buckets.Have, "ace.web", failed) b.place(t, "ace", "web") @@ -952,8 +961,6 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) { 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.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"}) b.putJSON(t, buckets.Have, "ace.mail", failed) wantKey, _ := mwc.WantKey("ace", "mail") @@ -992,8 +999,6 @@ func TestAFailedDataModuleIsAskedToRestore(t *testing.T) { func TestTheAssignerReplacedResumesWithNothingLostOrRepeated(t *testing.T) { b := startBed(t, "") const modules = 40 - on, _ := presenceValue(true) - b.putJSON(t, buckets.Presence, "ace", on) names := make([]string, modules) for i := range names { name := fmt.Sprintf("mod%02d", i) @@ -1054,8 +1059,6 @@ func (b *bed) fixture(t *testing.T, name, fp, manifest string) { gkey, _ := grantsstate.PartGrantKey(name, fp) grant, _ := grantsstate.EncodeGrant(grantsstate.Grant{}) b.putJSON(t, buckets.Grants, gkey, grant) - on, _ := presenceValue(true) - b.putJSON(t, buckets.Presence, "ace", on) b.place(t, "ace", name) } @@ -1373,6 +1376,13 @@ func (s bedStore) history(string, string) ([]stdio.StateRevision, error) { retur // and declined, writes nothing and is withdrawn. func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) { b := startBed(t, "") + // Both machines are away throughout (§389): a machine the watchman never said away is present. + for _, node := range []string{"anchor", "laptop"} { + raw, _ := json.Marshal(wd.MachineAwayEvent{V: 1, Machine: node, At: time.Now(), LastHeard: time.Now()}) + if _, err := b.js.Publish(b.ctx, wd.SubjectMachineAway(node), raw); err != nil { + t.Fatal(err) + } + } if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "MOD_" + assignment.Module + "_assignments", Subjects: assignmentsSubjects()}); err != nil { // the seat's four events: the views family has a stream of its own t.Fatal(err) 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 982be1ce..c3163794 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -63,7 +63,6 @@ func main() { // holds the two lists equal), so the controller's broker grants exactly these. var readBuckets = []string{ buckets.Grants, - buckets.Presence, buckets.Have, buckets.Provisions, } @@ -77,7 +76,7 @@ var readFamilies = []string{buckets.ComposedRead} var ownBuckets = []string{buckets.Asks, buckets.Differences, buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket, - engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.Inputs, pinsBucket} + engineReadsBucket, retiredBucket, copiesBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.DeliveryWaits, buckets.Inputs, pinsBucket} // engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key ``, as // the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again. @@ -118,9 +117,10 @@ var keptStates = []announce.State{ {Name: copiesState, Readable: true}, {Name: writtenState, Readable: true}, {Name: buckets.WaitsState, Readable: true}, - {Name: buckets.PutWaitsState, Readable: true}, // the put waits said (putwaits.go, design 47 G A-211) - {Name: buckets.InputsState, Readable: true}, // read by the tool process alone: the verb input and assigner_inputs (design 47 G §354) - {Name: pinsState, Readable: true}, // the records pinning each assignment graph revision (design 47 G §579) + {Name: buckets.PutWaitsState, Readable: true}, // the put waits said (putwaits.go, design 47 G A-211) + {Name: buckets.DeliveryWaitsState, Readable: true}, // the delivery waits said (delivered.go, design 47 G A-246) + {Name: buckets.InputsState, Readable: true}, // read by the tool process alone: the verb input and assigner_inputs (design 47 G §354) + {Name: pinsState, Readable: true}, // the records pinning each assignment graph revision (design 47 G §579) } // kvSet is every bucket the assigner opened, by name. @@ -189,6 +189,11 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any logf("the bucket %s cannot be opened (%v): no outside wait is kept until the assigner starts again", name, err) continue } + if name == buckets.DeliveryWaits && err != nil { + // New with A-246: without it no dependent waits for its dependency's delivery until the assigner starts again. + logf("the bucket %s cannot be opened (%v): no delivery wait is kept until the assigner starts again", name, err) + continue + } if name == buckets.PutWaits && err != nil { // New with A-211: without it no put's wait is said until the assigner starts again. logf("the bucket %s cannot be opened (%v): no put's wait is said until the assigner starts again", name, err) @@ -218,7 +223,7 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any func startsWithout(name string) bool { switch name { case plc.SeatHoldersBucket, buckets.Provisions, assign.PoliciesBucket, assign.FindingsBucket, - buckets.Puts, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.Inputs, pinsBucket: + buckets.Puts, engineReadsBucket, retiredBucket, writtenBucket, buckets.Waits, buckets.PutWaits, buckets.DeliveryWaits, buckets.Inputs, pinsBucket: return true } return slices.Contains(seatBuckets, name) || name == buckets.Grants @@ -336,7 +341,6 @@ var watched = []string{ plc.CandidatesBucket, plc.PulsesBucket, copiesBucket, - buckets.Presence, buckets.Have, buckets.Asks, buckets.Differences, @@ -348,6 +352,7 @@ var watched = []string{ writtenBucket, buckets.Waits, buckets.PutWaits, + buckets.DeliveryWaits, } // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after @@ -604,9 +609,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, logger.Printf("a pipeline's end or a registration is not heard, so the wants keep the trunk-only form until the assigner starts again: %v", err) } } - if err := hearReleases(hub, &w.released, wake, logger.Printf); err != nil { - logger.Printf("a version's release is not heard, so no want names its version until the assigner starts again: %v", err) - } + hearDelivery(ctx, js, w, wake, findings, logger.Printf) // version-released, version-delivered, machine-away and machine-back hearBuilds(ctx, js, &w.ends, wake, findings, logger.Printf) hearReadsChanged(ctx, js, hub, reads, findings, gate.EventsLive, logger.Printf) hearRunning(ctx, js, w, wake, findings, logger.Printf) 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 8bcfa64d..313737b9 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -205,6 +205,7 @@ func (p *passer) clock() time.Time { func (p *passer) pass(ctx context.Context) { p.assignmentsSeat(ctx) p.keepGraph(ctx) + p.judgeDelivered() p.wants(ctx) p.puts(ctx) p.nodeWants(ctx) @@ -303,6 +304,10 @@ func (p *passer) wants(ctx context.Context) { if fp != "" && !p.ends.mayWant(name, fp, branches, wantedOf[name]) { return } + // A dependent's new version waits for its dependency's newest released version to be delivered (§376, A-189). + if p.holdForDelivery(ctx, name, fp, g.branch, g.nodes) { + 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) { 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 67b4464f..59b992ba 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 @@ -16,7 +16,6 @@ import ( "time" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" - wdc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" mdc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" @@ -301,8 +300,8 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { return json.Marshal(nodeengine.ManifestReadsAnswer{V: 1, Node: node, ManifestReads: []string{"form:new"}, StatedAt: time.Now().Format(time.RFC3339Nano)}) } p := &passer{w: w, reads: &enginereads.Keeper{Call: call, Fallback: w.recordReads, Sync: true}} - on, _ := presenceValue(false) - w.put(buckets.Presence, "ace", on) + on := false + w.heardPresence("ace", !on, time.Now()) n = w.nodes(nil, nil) p.engineReads(n) if !slices.Equal(n.Reads["ace"], []string{"form:new"}) || len(n.Reads["g14"]) != 0 || asked["ace"] != 1 { @@ -312,8 +311,8 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { t.Errorf("the statements are %+v", s) } p.engineReads(w.nodes(nil, nil)) - on, _ = presenceValue(true) - w.put(buckets.Presence, "ace", on) + on = true + w.heardPresence("ace", !on, time.Now()) p.engineReads(w.nodes(nil, nil)) if asked["ace"] != 2 { t.Errorf("ace, back online, was asked %d times, want 2", asked["ace"]) @@ -337,16 +336,17 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { Keys: func() ([]string, error) { return slices.Collect(maps.Keys(held)), nil }}, reads: &enginereads.Keeper{Call: silent, Sync: true, Now: func() time.Time { return clock }}} key := mcc.Key(mcc.ScopeCall, enginereads.CallID("ace"), mcc.KindCallUnanswered) + w.heardPresence("g14", true, time.Now()) // g14 away throughout: a machine is present unless the watchman said it away (§389) before := asked["ace"] q.engineReads(w.nodes(nil, nil)) clock = clock.Add(6 * time.Minute) - off, _ := presenceValue(false) - w.put(buckets.Presence, "ace", off) + off := false + w.heardPresence("ace", !off, time.Now()) q.engineReads(w.nodes(nil, nil)) if len(held) != 0 { t.Errorf("a finding was written while ace is away: %v", slices.Collect(maps.Keys(held))) } - w.put(buckets.Presence, "ace", on) + w.heardPresence("ace", !on, time.Now()) q.engineReads(w.nodes(nil, nil)) if _, ok := held[key]; !ok || asked["ace"]-before < 3 { t.Errorf("ace unanswered six minutes, heard: findings %v, asked %d times", slices.Collect(maps.Keys(held)), asked["ace"]-before) @@ -899,8 +899,8 @@ func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) { mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) putModuleRaw(p.w, trunkBranch, "web", mod) placeOn(t, p.w, "g14", "web") - on, _ := presenceValue(true) - p.w.put(buckets.Presence, "g14", on) + on := true + p.w.heardPresence("g14", !on, time.Now()) have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpTest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) p.w.put(buckets.Have, "g14.web", have) key, _ := mwc.WantKey("g14", "web") @@ -937,8 +937,8 @@ func testAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) { mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fpTest, Repository: "novox/resolved", Commit: strings.Repeat("1", 40), Path: meshcatalogclient.RootPath}) putModuleRaw(p.w, trunkBranch, "systemd-resolved", mod) placeOn(t, p.w, "shanks", "systemd-resolved") - on, _ := presenceValue(true) - p.w.put(buckets.Presence, "shanks", on) + on := true + p.w.heardPresence("shanks", !on, time.Now()) have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpTest, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthFailed}) p.w.put(buckets.Have, "shanks.systemd-resolved", have) hkey, _ := mwc.HeldKey("systemd-resolved", fpTest) @@ -1220,8 +1220,8 @@ func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) { placeOn(t, p.w, "g14", "web") pl, _ := mmac.EncodeAssignments(mma.Assignments{Node: "g14", Modules: []mma.PlacedModule{{Module: "web"}}}) put(plc.AssignmentsBucket, "assignments.g14", pl) - on, _ := presenceValue(true) - put(buckets.Presence, "g14", on) + on := true + p.w.heardPresence("g14", !on, time.Now()) have, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Fp: fpOld, Composed: was, Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) put(buckets.Have, "g14.web", have) wkey, _ := mwc.WantKey("g14", "web") @@ -1403,11 +1403,6 @@ func mustAskID(a asks.Ask) string { return ask.ID } -// presenceValue is a node's presence as the watchman writes it. -func presenceValue(online bool) ([]byte, error) { - return json.Marshal(wdc.NodePresence{V: 1, Online: online}) -} - // Issue 1966 (nats on novox): a yes given for a record ask was lost because the same pass took the ask away as stale before the // want named the yes, and an ask taken away and written again was never asked again because it carried the id the router kept. // Given a record ask of a placed node, then (a) a have mid-move, or running with the operator's yes unwritten, leaves the ask, 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 801f4d10..1a1d594a 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 @@ -205,8 +205,8 @@ func TestAPassWritesNoWantWithoutAPassedPipeline(t *testing.T) { p.w.put(buckets.Have, "g14.web", have) composes(t, p.w, "g14", digest) placeOn(t, p.w, "g14", "web") - on, _ := presenceValue(true) - p.w.put(buckets.Presence, "g14", on) + on := true + p.w.heardPresence("g14", !on, time.Now()) if c.follow { p.ends = newPipelineEnds(newMemEnds(), func() {}, t.Logf) p.ends.trunk = p.w.trunkName @@ -270,10 +270,10 @@ func TestACommitOnBetaMovesOnlyTheNodeThatChoseBeta(t *testing.T) { p.ends.trunk = p.w.trunkName p.ends.registered(mpcat.Registration{Module: "web", Branches: []string{"main", "beta"}}, false) digest := "sha256:" + strings.Repeat("c", 64) - on, _ := presenceValue(true) + on := true for _, node := range []string{"g14", "ace"} { composes(t, p.w, node, digest) - p.w.put(buckets.Presence, node, on) + p.w.heardPresence(node, !on, time.Now()) } for node, branch := range map[string]string{"g14": "", "ace": "beta"} { pl := mma.Assignments{Node: node, Modules: []mma.PlacedModule{{Module: "web", Branch: branch}}} 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 ce13650a..4f65c6d4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -12,7 +12,6 @@ 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/modules/mesh-delivery-builder/builds" mdc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" mwc "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshwantsclient" @@ -44,6 +43,10 @@ type world struct { ends builtEnds // released are the catalog's version-released heard, by module and source revision (buildended.go). released releasedVersions + // delivered are the assigner's own version-delivered heard and said (delivered.go). + delivered deliveredVersions + // edges are, by watched branch and module, the edges into the module's artifacts its catalog entry names (§374's kinds). + edges map[string]map[string][]meshcatalog.Edge // assignments is the seat's state assignments assembled node by node from its keys in either form (mesh-sdk // mmac.Keys; research 072 row 37): the one reader of its keys in the assigner. assignmentKeys *mmac.Keys @@ -51,6 +54,8 @@ type world struct { // branches the registrations name, before the first registrations from those the gate's `source` answers (design 47 G §507, §532), // and never fixed in code (§7). first string + // presence is, by machine, the last machine-away or machine-back the watchman said of it (design 47 G §389). + presence map[string]presenceSaid // running is, by `.`, the last `running` the node-engine seat said of it, kept from its stream by the assigner's // durable consumer (design 47 G §199, §309); runningHeard says that consumer is bound. running map[string]nodeengine.Running @@ -269,6 +274,27 @@ func (w *world) dropModule(branch, module string) { defer w.mu.Unlock() delete(w.modules[branch], module) delete(w.graphs[branch], module) + delete(w.edges[branch], module) +} + +// putEdges keeps the edges into module's artifacts that branch's catalog entry names. +func (w *world) putEdges(branch, module string, edges []meshcatalog.Edge) { + w.mu.Lock() + defer w.mu.Unlock() + if w.edges == nil { + w.edges = map[string]map[string][]meshcatalog.Edge{} + } + if w.edges[branch] == nil { + w.edges[branch] = map[string][]meshcatalog.Edge{} + } + w.edges[branch][module] = edges +} + +// edgesOf is the edges into module's artifacts in branch's catalog half. +func (w *world) edgesOf(branch, module string) []meshcatalog.Edge { + w.mu.Lock() + defer w.mu.Unlock() + return w.edges[branch][module] } // holdsBranch says whether the catalog half of branch has been taken in (even empty). @@ -294,18 +320,37 @@ func (w *world) records() map[string]nodeengine.NodeRecord { return out } -// presenceOf reads one key of the watchman's presence: a node's key (``) with the node's online or not; the keys of -// the parts (`part..`) are not a node's and are none here. -func presenceOf(key string, value []byte) (node string, online, ok bool) { - node, part, _, err := wdc.ParsePresenceKey(key) - if err != nil || node == "" || part != "" { - return "", false, false +// presenceSaid is the last of the watchman's machine-away and machine-back heard of a machine (design 47 G §389). +type presenceSaid struct { + away bool + at time.Time +} + +// heardPresence takes one machine-away (away) or machine-back of machine said at at: the later of the two holds, so the two +// streams, read whole in either order, leave each machine as its last event says. +func (w *world) heardPresence(machine string, away bool, at time.Time) { + w.mu.Lock() + defer w.mu.Unlock() + if w.presence == nil { + w.presence = map[string]presenceSaid{} } - p, err := wdc.DecodeNodePresence(value) - if err != nil { - return "", false, false + if was, ok := w.presence[machine]; ok && at.Before(was.at) { + return } - return node, p.Online, true + w.presence[machine] = presenceSaid{away: away, at: at} +} + +// awayNow is every machine whose last presence event heard is machine-away. +func (w *world) awayNow() map[string]bool { + w.mu.Lock() + defer w.mu.Unlock() + out := map[string]bool{} + for m, p := range w.presence { + if p.away { + out[m] = true + } + } + return out } // seatHeard says a machine holding seat (a seat held once) is online as the watchman holds it: the callee of a call to @@ -326,13 +371,13 @@ func (w *world) seatHeard(seat string) bool { return false } -// online is, by node, whether the watchman holds the node online. +// online is, by node the assigner hears of, whether it is present: a machine is away from the watchman's machine-away until its +// machine-back, and present otherwise (design 47 G §389); the watchman's state is not read (ADR 0596 §2). func (w *world) online() map[string]bool { + away := w.awayNow() out := map[string]bool{} - for key, value := range w.read(buckets.Presence) { - if node, online, ok := presenceOf(key, value); ok { - out[node] = online - } + for _, node := range w.knownNodes() { + out[node] = !away[node] } return out } @@ -361,11 +406,7 @@ func (w *world) wantsOf() map[string]map[string]mwc.Want { func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[string][]string) assign.Nodes { n := assign.Nodes{Online: map[string]bool{}, Running: w.runningNow(), Haves: map[string]map[string]nodeengine.HaveByNodeModuleValue{}, Providers: map[string]map[string][]string{}, Reads: map[string][]string{}} - for key, value := range w.read(buckets.Presence) { - if node, online, ok := presenceOf(key, value); ok { - n.Online[node] = online - } - } + n.Online = w.online() for key, value := range w.read(buckets.Have) { node, module, isNode, err := nodeengineclient.ParseHaveKey(key) if err != nil { @@ -452,7 +493,7 @@ func (w *world) recordReads(node string) ([]string, bool) { return r.ManifestReads, ok } -// knownNodes is every node the assigner hears of: placed, with a record, or with a presence. +// knownNodes is every node the assigner hears of: placed, with a record, or named by machine-away or machine-back. func (w *world) knownNodes() []string { set := map[string]bool{} for node := range w.placedNodes() { @@ -461,9 +502,11 @@ func (w *world) knownNodes() []string { for node := range w.records() { set[node] = true } - for node := range w.online() { + w.mu.Lock() + for node := range w.presence { set[node] = true } + w.mu.Unlock() return slices.Sorted(maps.Keys(set)) } diff --git a/modules/mesh-delivery-assigner/internal/buckets/buckets.go b/modules/mesh-delivery-assigner/internal/buckets/buckets.go index 4711f9b1..4bccf1e1 100644 --- a/modules/mesh-delivery-assigner/internal/buckets/buckets.go +++ b/modules/mesh-delivery-assigner/internal/buckets/buckets.go @@ -6,7 +6,7 @@ package buckets import ( "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" - watchdogs "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" wants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "git.novox.be/novox/mesh-delivery/modules/mesh-grants/grantsstate" @@ -23,6 +23,10 @@ const ( // Waits holds the waits on an outside cause, one key `..` (design 47 G §706). WaitsState = "waits" Waits = plc.Module + "_" + WaitsState + // DeliveryWaits holds each dependent's delivery wait the assigner said, one key ``, its value the wait-held said + // (design 47 G §376, §498, A-246). + DeliveryWaitsState = "delivery-waits" + DeliveryWaits = plc.Module + "_" + DeliveryWaitsState // PutWaits holds each put wait the assigner said, one key `.`, its value the wait's id and the put as said // (design 47 G A-204, A-211), so a put that ends while the assigner is stopped gets its wait-ended at the next start. PutWaitsState = "put-waits" @@ -58,8 +62,6 @@ const ( // is its state provisions. ProvisionerModule = "mesh-provisioner" Provisions = ProvisionerModule + "_provisions" - // Presence holds each part's presence and the node's. - Presence = "mesh-delivery-watchman_" + watchdogs.StatePresence // TellerModule is the module whose hand the operator's answers pass: the one that holds each ask's answer, and the // name the hand-act log records a warrant under. Held equal to the manifest's reads by its test. TellerModule = "mesh-delivery-teller" diff --git a/modules/mesh-delivery-assigner/internal/buckets/buckets_test.go b/modules/mesh-delivery-assigner/internal/buckets/buckets_test.go index e5d8a881..9b2acc18 100644 --- a/modules/mesh-delivery-assigner/internal/buckets/buckets_test.go +++ b/modules/mesh-delivery-assigner/internal/buckets/buckets_test.go @@ -8,9 +8,9 @@ func TestBucketsAreTheEarlierNames(t *testing.T) { for name, c := range map[string][2]string{ "Have": {Have, "node-engine_have"}, "ComposedRead": {ComposedRead, "mesh-host.composed"}, "Modules": {Modules, "mesh-delivery-indexer_modules"}, "Manifests": {Manifests, "mesh-delivery-indexer_manifests"}, - "Presence": {Presence, "mesh-delivery-watchman_presence"}, "Answers": {Answers, "mesh-delivery-teller_answers"}, - "Grants": {Grants, "mesh-grants_grants"}, - "Wants": {Wants, "mesh-delivery-assigner_wants"}, "Differences": {Differences, "mesh-delivery-assigner_differences"}, + "Answers": {Answers, "mesh-delivery-teller_answers"}, + "Grants": {Grants, "mesh-grants_grants"}, + "Wants": {Wants, "mesh-delivery-assigner_wants"}, "Differences": {Differences, "mesh-delivery-assigner_differences"}, "Asks": {Asks, "mesh-delivery-assigner_asks"}, "Puts": {Puts, "mesh-delivery-assigner_puts"}, "Policies": {Policies, "mesh-delivery-assigner_policies"}, "Provisions": {Provisions, "mesh-provisioner_provisions"}, "TellerModule": {TellerModule, "mesh-delivery-teller"}, } { diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index ac4a2561..07ae1aea 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -49,13 +49,14 @@ func TestManifest(t *testing.T) { // A-181: what was built is heard from the builder's stream of ends, so the builder's builds bucket is no read. // §431, §507: the gate's source is read through its verb `source` and its event `source-moved`, and no `merges` is read. // A-175, A-178: the teller's answers are followed through its first view and its views family, never its bucket. - want := []string{"mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed", + // §389: the watchman's machine-away and machine-back are consumed, never its presence. + want := []string{"mesh-grants.grants", "mesh-host.composed", "mesh-provisioner.provisions", "node-engine.have"} if !reflect.DeepEqual(reads, want) { t.Errorf("reads = %v, want exactly %v", reads, want) } // the buckets the program names are the manifest's reads, `.` as `_` - named := []string{buckets.Presence, buckets.Grants, + named := []string{buckets.Grants, buckets.ComposedRead, "mesh-provisioner_provisions", buckets.Have} slices.Sort(named) for i, r := range want { @@ -69,7 +70,7 @@ func TestManifest(t *testing.T) { } if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20, "assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1, - "engine-reads": 1, "written": 5, "waits": 5, "put-waits": 1, "inputs": 1, "pins": 5}; !reflect.DeepEqual(states, want) { + "engine-reads": 1, "written": 5, "waits": 5, "put-waits": 1, "delivery-waits": 1, "inputs": 1, "pins": 5}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -131,8 +132,9 @@ 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-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-catalog.version-released", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildsStreamEvents, buckets.TellerModule + "." + tellerevents.AskEnded, buckets.TellerModule + ".views." + buckets.AnswersState, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) { + // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears; §376, §389: its own version-delivered and + // the watchman's machine-away and machine-back, read whole at its start. + if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-conditions.condition-raised", "mesh-conditions.condition-changed", "mesh-conditions.condition-cleared", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-catalog.version-released", "mesh-wants.version-delivered", "mesh-watchdogs.machine-away", "mesh-watchdogs.machine-back", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", builds.BuilderModule + "." + builds.BuildsStreamEvents, buckets.TellerModule + "." + tellerevents.AskEnded, buckets.TellerModule + ".views." + buckets.AnswersState, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running", "mesh-delivery.pinned", "mesh-delivery.freed", "mesh-delivery.heads-replaced", "mesh-delivery.pins-stated"}) { 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.manifests", "mesh-delivery-indexer.modules", diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 36762452..4eca5a31 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -30,7 +30,6 @@ } ], "reads": [ - "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed", "mesh-provisioner.provisions", @@ -48,6 +47,9 @@ "mesh-catalog.registration-changed", "mesh-delivery.source-moved", "mesh-catalog.version-released", + "mesh-wants.version-delivered", + "mesh-watchdogs.machine-away", + "mesh-watchdogs.machine-back", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-builder.build.ended.**", @@ -144,6 +146,7 @@ { "name": "written", "history": 5 }, { "name": "waits", "history": 5 }, { "name": "put-waits", "history": 1 }, + { "name": "delivery-waits", "history": 1 }, { "name": "inputs", "history": 1 }, { "name": "pins", "history": 5, "max-mib": 4 } ],