diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 2036a129..76873b20 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -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)): + } + } +} diff --git a/internal/compose/frombus/frombus.go b/internal/compose/frombus/frombus.go index 89e6e931..7af21b84 100644 --- a/internal/compose/frombus/frombus.go +++ b/internal/compose/frombus/frombus.go @@ -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 diff --git a/internal/compose/frombus/frombus_test.go b/internal/compose/frombus/frombus_test.go index 6ae853e8..c34bbd50 100644 --- a/internal/compose/frombus/frombus_test.go +++ b/internal/compose/frombus/frombus_test.go @@ -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, diff --git a/internal/compose/input.go b/internal/compose/input.go index 7c1de99a..d4dd4a07 100644 --- a/internal/compose/input.go +++ b/internal/compose/input.go @@ -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 diff --git a/internal/compose/rest.go b/internal/compose/rest.go index f97ef72a..f921f1e1 100644 --- a/internal/compose/rest.go +++ b/internal/compose/rest.go @@ -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 { diff --git a/internal/compose/rest_test.go b/internal/compose/rest_test.go index 8cb1e61d..8ba82ad0 100644 --- a/internal/compose/rest_test.go +++ b/internal/compose/rest_test.go @@ -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) diff --git a/internal/link/running.go b/internal/link/running.go new file mode 100644 index 00000000..722ce19a --- /dev/null +++ b/internal/link/running.go @@ -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 +} diff --git a/internal/link/session_test.go b/internal/link/session_test.go index 7fc1fdd6..64783c1d 100644 --- a/internal/link/session_test.go +++ b/internal/link/session_test.go @@ -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. diff --git a/internal/selfcompose/candidate.go b/internal/selfcompose/candidate.go index 3791f1e2..a070e45b 100644 --- a/internal/selfcompose/candidate.go +++ b/internal/selfcompose/candidate.go @@ -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) } diff --git a/internal/selfcompose/composedstate_test.go b/internal/selfcompose/composedstate_test.go index 38c02b51..fb5b10cd 100644 --- a/internal/selfcompose/composedstate_test.go +++ b/internal/selfcompose/composedstate_test.go @@ -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 { diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 1bb6fa1c..cba3a1f2 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -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 +}