H-5: the seats a node passes on are decided from what every node said it runs, by SeatsToPassOn's new input
The node follows the node-engine seat's running events (an ordered read of the last message of each subject, then a durable consumer), keeps what each node runs by `at`, acts on none until the first read ended, and hands the composer each node's heard build with the manifest read by the event's digest; compose resolves the passes through catalogue.SeatsToPassOn (heard, entries) with the trunk's entry for a node that said nothing. Builds: design 47 G §199, §249, §459, §536, §556 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -669,6 +669,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
|
||||
machinesNow.Store(composer) // the machines a failing network part points at, by the assignments (issue 774)
|
||||
composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone
|
||||
session.SetManifestOf(composer.ManifestOf) // the event running names the manifest of the build that runs
|
||||
go followRunning(run, session, composer, say)
|
||||
env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday
|
||||
env.WhyUnmade, env.Waiting = composer.WhyUnmade, composer.Waiting
|
||||
env.HoldsData = composer.HoldsData // what runs from the kept declaration says Data as its manifest does
|
||||
@@ -1133,3 +1134,29 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com
|
||||
}
|
||||
return env
|
||||
}
|
||||
|
||||
// followRunning keeps the composer told what every node runs, from the seat node-engine's event running, for the run it is
|
||||
// in: the stream or the link not being there is said and tried again after a backoff, never polled.
|
||||
func followRunning(run context.Context, session *link.Session, composer *selfcompose.Composer, say func(string)) {
|
||||
said := ""
|
||||
for try := 1; run.Err() == nil; try++ {
|
||||
began := time.Now()
|
||||
err := session.FollowRunning(run, composer.NoteRunning, composer.RunningRead)
|
||||
composer.RunningLost()
|
||||
if run.Err() != nil {
|
||||
return
|
||||
}
|
||||
if time.Since(began) > time.Minute {
|
||||
try = 1
|
||||
}
|
||||
if why := fmt.Sprint(err); why != said {
|
||||
say("this node does not hear what the nodes run, and tries again: " + why)
|
||||
said = why
|
||||
}
|
||||
select {
|
||||
case <-run.Done():
|
||||
return
|
||||
case <-time.After(link.Backoff(try)):
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -156,6 +156,10 @@ type Scope struct {
|
||||
Assignments map[string]*meshmoduleassigner.Assignments
|
||||
InOwn []string
|
||||
Running []engcomp.Composed
|
||||
// Heard is, by node and module, what each node said it runs, from the node-engine seat's event `running` kept by the
|
||||
// composing node's own consumer (design 47 G §199): the passing of a seat reads it, and a module nothing was heard of
|
||||
// counts by the trunk's entry. Never read from a have key, a want or a call.
|
||||
Heard map[string]map[string]Heard
|
||||
// HeldToday is, in an answer, the modules the node keeps today as they run ("node" for its own composition), each with
|
||||
// how ("held", "failed", "runs a build of the old form"): a candidate that cannot compose one, or holds one only
|
||||
// behind one of them, keeps it as today, which is said under CandidateAnswer.Held and is no refusal (issue 1126).
|
||||
@@ -551,6 +555,7 @@ func (b base) finish(ctx context.Context, r Reader, build func(ctx context.Conte
|
||||
if err := fromBuilds(ctx, r, build, &in, s, b.assignments[s.Node], b.modules, read, parse, refuse, wanted); err != nil {
|
||||
return compose.Input{}, err
|
||||
}
|
||||
keepHeard(&in, s, read)
|
||||
// Its own modules' secrets, sealed to its own key by the vault: a composition carries them sealed (ADR 0550 §5).
|
||||
// Asked only when a module placed on it declares one: a node none of whose modules has a secret composes
|
||||
// whether or not the vault answers.
|
||||
@@ -581,6 +586,36 @@ func (b base) finish(ctx context.Context, r Reader, build func(ctx context.Conte
|
||||
return in, nil
|
||||
}
|
||||
|
||||
// Heard is one `running` event as the scope keeps it: the fingerprint of the build, the digest of its manifest, or Stopped.
|
||||
type Heard struct {
|
||||
Fingerprint string
|
||||
Manifest string
|
||||
Stopped bool
|
||||
}
|
||||
|
||||
// keepHeard puts, as each node's Heard, what it said it runs of each module it is assigned, with the manifest of the build
|
||||
// read by the event's digest; a digest that cannot be read leaves the entry unread, and the passing says so (§459).
|
||||
func keepHeard(in *compose.Input, s Scope, read func(module, digest string) (catalogue.Manifest, error)) {
|
||||
for i := range in.Nodes {
|
||||
n := &in.Nodes[i]
|
||||
heard := s.Heard[n.Name]
|
||||
if len(heard) == 0 {
|
||||
continue
|
||||
}
|
||||
n.Heard = map[string]compose.Heard{}
|
||||
for _, module := range sortedKeys(heard) {
|
||||
h := heard[module]
|
||||
got := compose.Heard{Fingerprint: h.Fingerprint, Stopped: h.Stopped}
|
||||
if !h.Stopped && h.Manifest != "" {
|
||||
if man, err := read(module, h.Manifest); err == nil {
|
||||
got.Manifest, got.Read = man, true
|
||||
}
|
||||
}
|
||||
n.Heard[module] = got
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Compose is the node's compositions over the bus (ADR 0547 §2): input for s, then composed at the fingerprints fps
|
||||
// gives each module in the Reading mode, as the controller composes them, by the one rule its answer to a candidate
|
||||
// follows too (composeEach, issue 1126): a module refused alone (its build unread or not taken in, its composition
|
||||
|
||||
@@ -575,6 +575,23 @@ func TestInput(t *testing.T) {
|
||||
t.Errorf("the want's build is kept on the composing node alone: %v, %v", in.Nodes, err)
|
||||
}
|
||||
|
||||
// Design 47 G §199 (test 56): what each node said it runs (the seat node-engine's event running) is that node's Heard,
|
||||
// with the manifest read by the event's digest; a digest that cannot be read leaves the entry unread, and a stopped
|
||||
// module has none.
|
||||
in, err = Input(ctx, kept, Mesh{}, Scope{Node: "laptop", Heard: map[string]map[string]Heard{
|
||||
"laptop": {"xterm": {Fingerprint: "f1", Manifest: digest}, "media": {Stopped: true}}, "anchor": {"xterm": {Fingerprint: "f2", Manifest: "gone"}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, n := range in.Nodes {
|
||||
h := n.Heard["xterm"]
|
||||
switch {
|
||||
case n.Name == "laptop" && (!h.Read || h.Fingerprint != "f1" || h.Manifest.Module != "xterm" || !n.Heard["media"].Stopped):
|
||||
t.Errorf("laptop's Heard holds %+v", n.Heard)
|
||||
case n.Name == "anchor" && (h.Read || h.Fingerprint != "f2"):
|
||||
t.Errorf("anchor's Heard holds %+v, want f2 unread", n.Heard)
|
||||
}
|
||||
}
|
||||
// Issue 1126, the seed's refusals. restic and the provisioner's client as the catalogue holds them (mesh-catalog
|
||||
// c05aa5e6), restic moved with an own secret the vault leaves out (it hands it to the client, mesh-vault ownnode.go
|
||||
// handed): composed from the indexer's manifest restic is not handed and stands in for that secret, as novox's,
|
||||
|
||||
@@ -240,6 +240,9 @@ type Node struct {
|
||||
// own resolution only, never to another node's, and a declaration carries the kept commit for the module.
|
||||
Kept map[string]catalogue.Manifest
|
||||
KeptBuilds map[string]string
|
||||
// Heard is, by module, what this node said it runs, from the node-engine seat's event `running` (design 47 G §199): the
|
||||
// passing of a seat resolves the build each assignment runs from it. Absent for a module nothing was heard of.
|
||||
Heard map[string]Heard
|
||||
// Contract is the contract of its node-engine's newest statement (inv.HealthOf); Stated is whether it ever stated one.
|
||||
Contract int
|
||||
Stated bool
|
||||
@@ -437,6 +440,15 @@ type Credential struct {
|
||||
Why string
|
||||
}
|
||||
|
||||
// Heard is what a node's `running` event for a module said: the build's fingerprint, or Stopped (no fingerprint, §365), with
|
||||
// the manifest of the build's entry read by the event's manifest digest (Read false when it could not be read).
|
||||
type Heard struct {
|
||||
Fingerprint string
|
||||
Stopped bool
|
||||
Manifest catalogue.Manifest
|
||||
Read bool
|
||||
}
|
||||
|
||||
// Run is the build an assignment names: the commit it is of and the build made from it (chain.Ref).
|
||||
type Run struct {
|
||||
Commit string
|
||||
|
||||
+53
-24
@@ -102,7 +102,7 @@ func (c *composer) theRestOfTheMeshNow(exclude string) (catalogue.World, Passes,
|
||||
}
|
||||
}
|
||||
var passes Passes
|
||||
holdings, passes = passSeatsOn(holdings, c.assignmentsOf(shelf, everyAssigned))
|
||||
holdings, passes = passSeatsOn(holdings, c.assignmentsOf(everyAssigned), c.heard, c.entries(shelf))
|
||||
for _, n := range nodes {
|
||||
if n.Name == exclude {
|
||||
continue
|
||||
@@ -177,10 +177,11 @@ func (c *composer) theRestOfTheMeshNow(exclude string) (catalogue.World, Passes,
|
||||
}
|
||||
|
||||
// passSeatsOn finds, in the composition that finds it, the new holder of each seat held once whose holder on record has left
|
||||
// it and which has exactly one claimant able to hold it (catalogue.SeatsToPassOn, novox/hq ADR 0660), and answers the
|
||||
// holdings as they are after it, with the passes. Nothing is recorded.
|
||||
func passSeatsOn(holdings []catalogue.Held, assignments []catalogue.Assignment) ([]catalogue.Held, Passes) {
|
||||
passes, said := catalogue.SeatsToPassOn(holdings, assignments)
|
||||
// it and which has exactly one claimant able to hold it (catalogue.SeatsToPassOn, novox/hq ADR 0660, design 47 G §199, §459), and
|
||||
// answers the holdings as they are after it, with the passes. Nothing is recorded.
|
||||
func passSeatsOn(holdings []catalogue.Held, assignments []catalogue.Assignment, heard func(node, module string) (catalogue.Heard, bool),
|
||||
entries catalogue.EntryReader) ([]catalogue.Held, Passes) {
|
||||
passes, said := catalogue.SeatsToPassOn(holdings, assignments, heard, entries)
|
||||
out := Passes{Said: said}
|
||||
for _, p := range passes {
|
||||
out.Passes = append(out.Passes, p)
|
||||
@@ -197,11 +198,10 @@ func passSeatsOn(holdings []catalogue.Held, assignments []catalogue.Assignment)
|
||||
return holdings, out
|
||||
}
|
||||
|
||||
// assignmentsOf are the assignments of every machine, as the passing of a seat reads them (novox/hq ADR 0660 clause 2): each
|
||||
// with the manifest of the build its node runs where the Input holds it (the node's Kept: what it runs of a module kept at
|
||||
// that build), else the trunk's on the shelf, which for the composing node's own modules is already the build they are
|
||||
// wanted at. A module assigned and missing from both is still assigned, marked unread, so its holder is not taken for gone.
|
||||
func (c *composer) assignmentsOf(shelf map[string]catalogue.Manifest, assigned map[string][]string) []catalogue.Assignment {
|
||||
// assignmentsOf are the assignments of every machine, as the passing of a seat reads them (§459): a node, a module and no
|
||||
// placement branch, this composer's view of the catalog half being the trunk's alone. The build each runs is resolved by
|
||||
// catalogue.SeatsToPassOn from c.heard and c.entries.
|
||||
func (c *composer) assignmentsOf(assigned map[string][]string) []catalogue.Assignment {
|
||||
var out []catalogue.Assignment
|
||||
names := make([]string, 0, len(assigned))
|
||||
for node := range assigned {
|
||||
@@ -209,26 +209,55 @@ func (c *composer) assignmentsOf(shelf map[string]catalogue.Manifest, assigned m
|
||||
}
|
||||
sort.Strings(names)
|
||||
for _, node := range names {
|
||||
var kept map[string]catalogue.Manifest
|
||||
for _, n := range c.in.Nodes {
|
||||
if n.Name == node {
|
||||
kept = n.Kept
|
||||
}
|
||||
}
|
||||
for _, module := range assigned[node] {
|
||||
m, ok := kept[module]
|
||||
if !ok {
|
||||
m, ok = shelf[module]
|
||||
}
|
||||
if !ok {
|
||||
m = catalogue.Manifest{Module: module}
|
||||
}
|
||||
out = append(out, catalogue.Assignment{Node: node, Manifest: m, Unread: !ok})
|
||||
out = append(out, catalogue.Assignment{Node: node, Module: module})
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// heard is what the node-engine seat's `running` said of a module on a node (Node.Heard): a fingerprint or stopped. Nothing
|
||||
// is heard of a module no event named, and the passing then takes the entry of the module on the trunk (§199).
|
||||
func (c *composer) heard(node, module string) (catalogue.Heard, bool) {
|
||||
for _, n := range c.in.Nodes {
|
||||
if n.Name == node {
|
||||
h, ok := n.Heard[module]
|
||||
return catalogue.Heard{Fingerprint: h.Fingerprint, Stopped: h.Stopped}, ok
|
||||
}
|
||||
}
|
||||
return catalogue.Heard{}, false
|
||||
}
|
||||
|
||||
// entries are the composer's reader of the catalog half for the passing of a seat: an entry by the fingerprint a node said it
|
||||
// runs, read through the digest its event named, and a module's entry on the trunk, the shelf.
|
||||
func (c *composer) entries(shelf map[string]catalogue.Manifest) catalogue.EntryReader {
|
||||
byFingerprint := map[string]catalogue.Manifest{}
|
||||
for _, n := range c.in.Nodes {
|
||||
for _, h := range n.Heard {
|
||||
if h.Read && h.Fingerprint != "" {
|
||||
byFingerprint[h.Fingerprint] = h.Manifest
|
||||
}
|
||||
}
|
||||
}
|
||||
return entryReader{byFingerprint: byFingerprint, shelf: shelf}
|
||||
}
|
||||
|
||||
type entryReader struct {
|
||||
byFingerprint map[string]catalogue.Manifest
|
||||
shelf map[string]catalogue.Manifest
|
||||
}
|
||||
|
||||
func (r entryReader) ByFingerprint(fingerprint string) (catalogue.Manifest, bool) {
|
||||
m, ok := r.byFingerprint[fingerprint]
|
||||
return m, ok
|
||||
}
|
||||
|
||||
// OnBranch answers the trunk's entry whatever the branch: this composer's shelf is the trunk's alone.
|
||||
func (r entryReader) OnBranch(module, _ string) (catalogue.Manifest, bool) {
|
||||
m, ok := r.shelf[module]
|
||||
return m, ok
|
||||
}
|
||||
|
||||
// Lines are what the passes say, one each, then what the finding itself says: a pass is composed so and recorded by nobody
|
||||
// (ADR 0660 clause 4).
|
||||
func (p Passes) Lines() []string {
|
||||
|
||||
@@ -162,19 +162,18 @@ func TestASeatWhoseHolderIsAssignedNowhereGoesToItsOnlyClaimant(t *testing.T) {
|
||||
func TestNoSeatIsPassedForTwoClaimantsAnAssignedHolderOrNoClaimant(t *testing.T) {
|
||||
withSeat(t)
|
||||
held := []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}
|
||||
_, passes := passSeatsOn(held, []catalogue.Assignment{
|
||||
{Node: "novox", Manifest: aBrokerClaiming("gate")}, {Node: "ace", Manifest: aBrokerClaiming("other")}})
|
||||
shelf := map[string]catalogue.Manifest{"gate": aBrokerClaiming("gate"), "other": aBrokerClaiming("other"), "old-broker": aBrokerClaiming("old-broker")}
|
||||
_, passes := passSeatsOn(held, []catalogue.Assignment{{Node: "novox", Module: "gate"}, {Node: "ace", Module: "other"}}, nil, entryReader{shelf: shelf})
|
||||
said := strings.Join(passes.Said, "\n")
|
||||
if len(passes.Passes) != 0 || !strings.Contains(said, "gate") || !strings.Contains(said, "other") ||
|
||||
!strings.Contains(said, "seat mesh-broker --to") {
|
||||
t.Fatalf("passes %+v said %q", passes.Passes, said)
|
||||
}
|
||||
got, passes := passSeatsOn(held, []catalogue.Assignment{
|
||||
{Node: "novox", Manifest: aBrokerClaiming("old-broker")}, {Node: "novox", Manifest: aBrokerClaiming("gate")}})
|
||||
got, passes := passSeatsOn(held, []catalogue.Assignment{{Node: "novox", Module: "old-broker"}, {Node: "novox", Module: "gate"}}, nil, entryReader{shelf: shelf})
|
||||
if len(passes.Passes) != 0 || len(got) != 1 || got[0].Module != "old-broker" {
|
||||
t.Fatalf("holder assigned: %+v %+v", got, passes)
|
||||
}
|
||||
got, passes = passSeatsOn(held, nil)
|
||||
got, passes = passSeatsOn(held, nil, nil, entryReader{shelf: shelf})
|
||||
if len(passes.Passes) != 0 || len(got) != 1 || got[0].Module != "old-broker" {
|
||||
t.Fatalf("no claimant: %+v %+v", got, passes)
|
||||
}
|
||||
@@ -183,13 +182,35 @@ func TestNoSeatIsPassedForTwoClaimantsAnAssignedHolderOrNoClaimant(t *testing.T)
|
||||
// seat_pass_test.go TestAHolderAssignedButNotOnTheShelfIsNotTakenForGone.
|
||||
func TestAHolderAssignedButNotOnTheShelfIsNotTakenForGone(t *testing.T) {
|
||||
withSeat(t)
|
||||
as := (&composer{}).assignmentsOf(map[string]catalogue.Manifest{"gate": aBrokerClaiming("gate")},
|
||||
map[string][]string{"novox": {"gate", "old-broker"}})
|
||||
as := (&composer{}).assignmentsOf(map[string][]string{"novox": {"gate", "old-broker"}})
|
||||
passes, _ := catalogue.SeatsToPassOn(
|
||||
[]catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}, as)
|
||||
[]catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "novox", Module: "old-broker"}}, as, nil,
|
||||
entryReader{shelf: map[string]catalogue.Manifest{"gate": aBrokerClaiming("gate")}})
|
||||
if len(passes) != 0 {
|
||||
t.Fatalf("a holder missing from the shelf lost its seat: %+v", passes)
|
||||
}
|
||||
// Design 47 G §199 (test 56): the build a node runs is the one it said in its running event, the trunk's entry counting
|
||||
// only for a node that said nothing. The trunk's old-broker still claims the seat; the one the away node runs does not.
|
||||
for name, c := range map[string]struct {
|
||||
heard Heard
|
||||
passTo string
|
||||
}{
|
||||
"the away node runs a build that no longer claims the seat": {Heard{Fingerprint: "f1", Manifest: catalogue.Manifest{Module: "old-broker"}, Read: true}, "gate"},
|
||||
"the away node says it stopped it": {Heard{Stopped: true}, "gate"},
|
||||
"the away node's build cannot be read": {Heard{Fingerprint: "f1"}, ""},
|
||||
"the away node runs a build that claims the seat": {Heard{Fingerprint: "f1", Manifest: aBrokerClaiming("old-broker"), Read: true}, ""},
|
||||
} {
|
||||
in := Input{
|
||||
Catalogue: map[string]catalogue.Manifest{"old-broker": aBrokerClaiming("old-broker"), "gate": aBrokerClaiming("gate")},
|
||||
Nodes: []Node{{Name: "away", Assigned: []string{"old-broker"}, Heard: map[string]Heard{"old-broker": c.heard}},
|
||||
{Name: "novox", Assigned: []string{"gate"}}},
|
||||
Holdings: []catalogue.Held{{Claim: "mesh-broker", Scope: catalogue.ScopeMesh, Node: "away", Module: "old-broker"}},
|
||||
}
|
||||
_, got, _, err := TheRestOfTheMesh(in, "")
|
||||
if err != nil || (c.passTo == "") != (len(got.Passes) == 0) || (c.passTo != "" && got.Passes[0].To.Module != c.passTo) {
|
||||
t.Errorf("%s: %v %+v", name, err, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The not-resolving node of a view with a holder on record: holdings reach the first pass (novox/hq ADR 0131), so a node running
|
||||
@@ -267,7 +288,7 @@ func TestAHolderWhoseManifestDroppedTheSeatGoesToItsOnlyClaimant(t *testing.T) {
|
||||
}
|
||||
// The build the holder's node runs still claims the seat, though the trunk dropped it: the holder keeps it (ADR 0660 §2).
|
||||
running := mk("gate")
|
||||
running.Nodes[0].Kept = map[string]catalogue.Manifest{"old-broker": aBrokerClaiming("old-broker")}
|
||||
running.Nodes[0].Heard = map[string]Heard{"old-broker": {Fingerprint: "f", Manifest: aBrokerClaiming("old-broker"), Read: true}}
|
||||
_, passes, _, err = TheRestOfTheMesh(running, "")
|
||||
if err != nil || len(passes.Passes) != 0 || len(passes.Said) != 0 {
|
||||
t.Fatalf("a holder whose running build claims the seat lost it: %v %+v", err, passes)
|
||||
|
||||
@@ -0,0 +1,95 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
// The node seat's event running, said by every node-engine for each module it saw running healthy, is kept on the stream
|
||||
// SEAT_node-engine_running, the last message of each subject with no age limit (design 47 G §199, §260). The node reads it
|
||||
// at each start through an ordered consumer it makes for that one read (§556), then follows a durable consumer of its own
|
||||
// from its acknowledged position: what the other nodes run is known from the last message of each subject however long ago it
|
||||
// was said, never from a have key or a call.
|
||||
|
||||
// RunningBus is a link that hears the seat's event running of every node.
|
||||
type RunningBus interface {
|
||||
// FirstRunning calls first with the last message of every subject, through an ordered consumer made for this read, and
|
||||
// answers when it has read them all.
|
||||
FirstRunning(ctx context.Context, first func([]byte)) error
|
||||
// HearRunning calls heard with each message after the durable consumer's acknowledged position (a new consumer begins at
|
||||
// the last message of each subject), until stop is called or the link ends.
|
||||
HearRunning(consumer string, heard func([]byte)) (stop func(), err error)
|
||||
}
|
||||
|
||||
// RunningConsumer is the name of this node's durable consumer of the stream.
|
||||
func RunningConsumer(node string) string { return "node-engine-" + node + "-running" }
|
||||
|
||||
func (l *natsLink) FirstRunning(ctx context.Context, first func([]byte)) error {
|
||||
sub, err := l.js.SubscribeSync(nodeengine.SubjectRunningAll, nats.OrderedConsumer(), nats.DeliverLastPerSubject(),
|
||||
nats.BindStream(nodeengine.StreamRunning))
|
||||
if err != nil {
|
||||
return fmt.Errorf("the stream %s cannot be read: %w", nodeengine.StreamRunning, err)
|
||||
}
|
||||
defer sub.Unsubscribe()
|
||||
info, err := sub.ConsumerInfo()
|
||||
if err != nil {
|
||||
return fmt.Errorf("the stream %s cannot be read: %w", nodeengine.StreamRunning, err)
|
||||
}
|
||||
for total, got := info.Delivered.Consumer+info.NumPending, uint64(0); got < total; got++ {
|
||||
msg, err := sub.NextMsgWithContext(ctx)
|
||||
if err != nil {
|
||||
return fmt.Errorf("reading the last running of each subject (%d of %d): %w", got, total, err)
|
||||
}
|
||||
first(msg.Data)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *natsLink) HearRunning(consumer string, heard func([]byte)) (func(), error) {
|
||||
sub, err := l.js.Subscribe(nodeengine.SubjectRunningAll, func(m *nats.Msg) {
|
||||
heard(m.Data)
|
||||
_ = m.Ack()
|
||||
}, nats.Durable(consumer), nats.BindStream(nodeengine.StreamRunning), nats.DeliverLastPerSubject(), nats.AckExplicit(),
|
||||
nats.ManualAck())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the stream %s cannot be followed: %w", nodeengine.StreamRunning, err)
|
||||
}
|
||||
return func() { _ = sub.Unsubscribe() }, nil
|
||||
}
|
||||
|
||||
var _ RunningBus = (*natsLink)(nil)
|
||||
|
||||
// FollowRunning hands every node's running event to heard: the last of each subject first, with readEnded called after them
|
||||
// (until then the caller acts on none, §556), then each after them, until ctx ends. An error is the stream or the link not
|
||||
// being there, which the caller tries again.
|
||||
func (q *Session) FollowRunning(ctx context.Context, heard func(nodeengine.Running), readEnded func()) error {
|
||||
q.init()
|
||||
q.mu.Lock()
|
||||
bus := q.bus
|
||||
q.mu.Unlock()
|
||||
h, ok := bus.(RunningBus)
|
||||
if bus == nil || !ok {
|
||||
return ErrNoLink
|
||||
}
|
||||
decode := func(raw []byte) {
|
||||
var ev nodeengine.Running
|
||||
if json.Unmarshal(raw, &ev) == nil && ev.Node != "" && ev.Module != "" {
|
||||
heard(ev)
|
||||
}
|
||||
}
|
||||
if err := h.FirstRunning(ctx, decode); err != nil {
|
||||
return err
|
||||
}
|
||||
stop, err := h.HearRunning(RunningConsumer(q.Membership.Node), decode)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer stop()
|
||||
readEnded()
|
||||
<-ctx.Done()
|
||||
return nil
|
||||
}
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
|
||||
"github.com/nats-io/nats.go"
|
||||
)
|
||||
|
||||
@@ -24,6 +25,26 @@ func newQuietLink() *quietLink { return &quietLink{lost: make(chan error, 1)} }
|
||||
func (l *quietLink) Lost() <-chan error { return l.lost }
|
||||
func (l *quietLink) Close() {}
|
||||
|
||||
// runningLink hears the node seat's running events: the last of each subject first, then the durable consumer's.
|
||||
type runningLink struct {
|
||||
*quietLink
|
||||
first, after [][]byte
|
||||
}
|
||||
|
||||
func (l *runningLink) FirstRunning(_ context.Context, first func([]byte)) error {
|
||||
for _, raw := range l.first {
|
||||
first(raw)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (l *runningLink) HearRunning(_ string, heard func([]byte)) (func(), error) {
|
||||
for _, raw := range l.after {
|
||||
heard(raw)
|
||||
}
|
||||
return func() {}, nil
|
||||
}
|
||||
|
||||
func aMember(t *testing.T) (Membership, ed25519.PrivateKey) {
|
||||
t.Helper()
|
||||
public, private, err := ed25519.GenerateKey(nil)
|
||||
@@ -46,6 +67,23 @@ func TestFactsWithNoLinkOpenAreNotSaid(t *testing.T) {
|
||||
if s.Linked() || s.SayFacts(context.Background(), Report{}) {
|
||||
t.Fatal("facts were said on a link that was let go")
|
||||
}
|
||||
|
||||
// Design 47 G §199, §556: with no link the running events are not followed; with one, the last message of each subject
|
||||
// comes first, the read ends, then the durable consumer's, and a body that is no running event is passed over.
|
||||
if err := s.FollowRunning(context.Background(), func(nodeengine.Running) {}, func() {}); !errors.Is(err, ErrNoLink) {
|
||||
t.Fatalf("followed with no link: %v", err)
|
||||
}
|
||||
rl := &runningLink{quietLink: newQuietLink(),
|
||||
first: [][]byte{[]byte(`{"node":"a","module":"x","fingerprint":"f1"}`), []byte(`not json`), []byte(`{"module":"x"}`)},
|
||||
after: [][]byte{[]byte(`{"node":"b","module":"y","stopped":true}`)}}
|
||||
s.Attach(context.Background(), rl)
|
||||
var order []string
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
err := s.FollowRunning(ctx, func(ev nodeengine.Running) { order = append(order, ev.Node+"/"+ev.Module) },
|
||||
func() { order = append(order, "read"); stop() })
|
||||
if err != nil || strings.Join(order, " ") != "a/x b/y read" {
|
||||
t.Fatalf("followed %v: %v, want a/x b/y read", order, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A link opening says the node's facts, off the link's own loop.
|
||||
|
||||
@@ -461,7 +461,7 @@ func (c *Composer) composeShadow(ctx context.Context, running []engcomp.Composed
|
||||
c.mu.Lock()
|
||||
c.wants = wants
|
||||
c.mu.Unlock()
|
||||
scope := frombus.Scope{Node: node, KeptAt: kept, InOwn: inOwn, Running: running, SameValue: c.sameValue}
|
||||
scope := frombus.Scope{Node: node, KeptAt: kept, InOwn: inOwn, Running: running, SameValue: c.sameValue, Heard: c.heardNow()}
|
||||
if input != nil {
|
||||
input(scope, fps)
|
||||
}
|
||||
|
||||
@@ -152,6 +152,23 @@ func TestTheNodePublishesItsCompositions(t *testing.T) {
|
||||
}
|
||||
saidMu.Unlock()
|
||||
}
|
||||
// Design 47 G §199, §556 (test 56): what the nodes say they run is acted on once the first read of the stream ended; the
|
||||
// newest by `at` stands, an older one and the same again change nothing.
|
||||
at := time.Now()
|
||||
c.NoteRunning(nodeengine.Running{Node: "n2", Module: "lemurs", Fingerprint: "a", Manifest: "sha256:a", At: at})
|
||||
if got := c.heardNow(); got != nil {
|
||||
t.Errorf("heard %v before the first read ended", got)
|
||||
}
|
||||
c.RunningRead()
|
||||
c.NoteRunning(nodeengine.Running{Node: "n2", Module: "lemurs", Fingerprint: "old", At: at.Add(-time.Hour)})
|
||||
c.NoteRunning(nodeengine.Running{Node: "n3", Module: "lemurs", Stopped: true, At: at})
|
||||
if got := c.heardNow(); len(got) != 2 || got["n2"]["lemurs"].Fingerprint != "a" || !got["n3"]["lemurs"].Stopped {
|
||||
t.Errorf("heard %v, want n2's lemurs at a (not the older) and n3's stopped", got)
|
||||
}
|
||||
c.RunningLost()
|
||||
if got := c.heardNow(); got != nil {
|
||||
t.Errorf("heard %v with the stream lost", got)
|
||||
}
|
||||
// A pass that changes nothing writes nothing.
|
||||
setFor(t, &askedWithin, 0)
|
||||
if _, err := c.ComposeNow(ctx); err != nil {
|
||||
|
||||
@@ -176,6 +176,10 @@ type Composer struct {
|
||||
unmadeSaid map[string]string
|
||||
// modulesRev is the revision of the last update of the indexer's modules taken: where the picture of them stands.
|
||||
modulesRev uint64
|
||||
// heard is, by node and module, what each node's event running last said it runs, kept from the seat's stream by this
|
||||
// node's own consumer (NoteRunning), under mu; heardRead is whether the first read of the stream ended (RunningRead).
|
||||
heard map[string]map[string]heardRun
|
||||
heardRead bool
|
||||
// rehand asks for a pass that hands again what the agent waits for (Waiting).
|
||||
rehand chan struct{}
|
||||
// sealedOwn is each own secret's sealed value as first handed, by module/name and generation: a value sealed again
|
||||
@@ -998,7 +1002,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update {
|
||||
var keptMu sync.Mutex
|
||||
var holds map[string]compose.Hold
|
||||
fromManifest := map[string]string{} // by module, why it is composed from the indexer's manifest and not a build
|
||||
scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running, Kept: func(module, words string) {
|
||||
scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running, Heard: c.heardNow(), Kept: func(module, words string) {
|
||||
keptMu.Lock()
|
||||
defer keptMu.Unlock()
|
||||
if !slices.Contains(keptRuns[module], words) {
|
||||
@@ -1861,7 +1865,7 @@ func (c *Composer) ComposedToday(ctx context.Context, module, fp string) (string
|
||||
if c.env.Running != nil {
|
||||
running = c.env.Running()
|
||||
}
|
||||
scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running}
|
||||
scope := frombus.Scope{Node: c.env.Node, KeptAt: kept, InOwn: inOwn, Running: running, Heard: c.heardNow()}
|
||||
comps, _, held, err := frombus.Compose(ctx, reader{c: c}, c.env.Mesh, scope, fps)
|
||||
if err != nil || held[module] != "" {
|
||||
return "", false
|
||||
@@ -2272,3 +2276,78 @@ func (c *Composer) WhyUnmade(module string) string {
|
||||
defer c.mu.Unlock()
|
||||
return c.last.Unmade[module]
|
||||
}
|
||||
|
||||
// heardRun is one running event as the composer keeps it, with when it was said.
|
||||
type heardRun struct {
|
||||
at time.Time
|
||||
fingerprint string
|
||||
manifest string
|
||||
stopped bool
|
||||
}
|
||||
|
||||
// NoteRunning takes the seat node-engine's event running of a node, this one's included (design 47 G §199): a module's
|
||||
// newest by `at` stands, an older one changes nothing (§556). A change of what a node runs asks a pass once the first read of
|
||||
// the stream has ended; the same value again asks none.
|
||||
func (c *Composer) NoteRunning(ev nodeengine.Running) {
|
||||
if ev.Node == "" || ev.Module == "" {
|
||||
return
|
||||
}
|
||||
run := heardRun{at: ev.At, fingerprint: ev.Fingerprint, manifest: ev.Manifest, stopped: ev.Stopped}
|
||||
c.mu.Lock()
|
||||
had, ok := c.heard[ev.Node][ev.Module]
|
||||
if ok && (ev.At.Before(had.at) || had == run) {
|
||||
c.mu.Unlock()
|
||||
return
|
||||
}
|
||||
if c.heard == nil {
|
||||
c.heard = map[string]map[string]heardRun{}
|
||||
}
|
||||
if c.heard[ev.Node] == nil {
|
||||
c.heard[ev.Node] = map[string]heardRun{}
|
||||
}
|
||||
c.heard[ev.Node][ev.Module] = run
|
||||
read := c.heardRead
|
||||
c.mu.Unlock()
|
||||
if read {
|
||||
c.askRehand()
|
||||
}
|
||||
}
|
||||
|
||||
// RunningRead says the first read of the stream ended: from now the composer acts on what it heard, and a pass is asked.
|
||||
func (c *Composer) RunningRead() {
|
||||
c.mu.Lock()
|
||||
c.heardRead = true
|
||||
c.mu.Unlock()
|
||||
c.askRehand()
|
||||
}
|
||||
|
||||
// RunningLost says the stream is not followed any more: nothing heard is acted on until a new read ends.
|
||||
func (c *Composer) RunningLost() {
|
||||
c.mu.Lock()
|
||||
c.heardRead = false
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
func (c *Composer) askRehand() {
|
||||
select {
|
||||
case c.rehand <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// heardNow is what the nodes said they run, as the passing of a seat reads it; nil until the first read of the stream ended.
|
||||
func (c *Composer) heardNow() map[string]map[string]frombus.Heard {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if !c.heardRead || len(c.heard) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make(map[string]map[string]frombus.Heard, len(c.heard))
|
||||
for node, modules := range c.heard {
|
||||
out[node] = make(map[string]frombus.Heard, len(modules))
|
||||
for module, h := range modules {
|
||||
out[node][module] = frombus.Heard{Fingerprint: h.fingerprint, Manifest: h.manifest, Stopped: h.stopped}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user