AS6: the assigner says version-delivered and holds a dependent for its dependency's delivery
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
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 `<node>`, 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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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}}}
|
||||
|
||||
@@ -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 `<node>.<module>`, 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 (`<node>`) with the node's online or not; the keys of
|
||||
// the parts (`part.<part>.<instance>`) 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))
|
||||
}
|
||||
|
||||
|
||||
@@ -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 `<node>.<module>.<fingerprint hex>` (design 47 G §706).
|
||||
WaitsState = "waits"
|
||||
Waits = plc.Module + "_" + WaitsState
|
||||
// DeliveryWaits holds each dependent's delivery wait the assigner said, one key `<module>`, 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 `<node>.<module>`, 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"
|
||||
|
||||
@@ -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"},
|
||||
} {
|
||||
|
||||
@@ -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, `<module>.<state>` as `<module>_<state>`
|
||||
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",
|
||||
|
||||
@@ -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 }
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user