The controller no longer subscribes to the alive message: HostBeats, its handler, the heartbeat-gap duration and a fixed silent bound (issue 1183, AC3, AC4, AC5)
mesh/merge-gate head 19e3eac74658 merged into trunk 89cfa9431d8f — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 19e3eac74658 merged into trunk 89cfa9431d8f — pass: its merge-check.sh passed
mesh/delivery delivered
mesh/merge-gate head 19e3eac74658 merged into trunk 89cfa9431d8f — pass: builds build-agent, mesh-controller → ace, g14, novox, shanks; no bus step; every …
mesh/repo-check head 19e3eac74658 merged into trunk 89cfa9431d8f — pass: its merge-check.sh passed
mesh/delivery delivered
This commit is contained in:
@@ -67,17 +67,12 @@ const (
|
||||
// BuiltSubject is where a build's outcome lands, for results nobody was waiting for.
|
||||
BuiltSubject = "mesh.control.built"
|
||||
|
||||
// AliveSubjects is every node's heartbeat. Core NATS, never a stream: a lost heartbeat is the
|
||||
// next heartbeat, and a stream of them is the mesh's least valuable message competing for
|
||||
// retention with its most valuable (design 25 §3).
|
||||
AliveSubjects = "mesh.control.*.alive"
|
||||
|
||||
// ToolsAliveSubjects is every machine's node tools saying they are there (novox/hq to-be 45 §3,
|
||||
// S11): core NATS like the host's, for the same reason.
|
||||
ToolsAliveSubjects = "mesh.control.*.tools-alive"
|
||||
|
||||
// HealthSubjects is every machine's health statement between its reports (novox/hq ADR 0240): core
|
||||
// NATS like the heartbeat, because a statement lost is said again within a minute while anything is
|
||||
// NATS like the node tools' heartbeat, because a statement lost is said again within a minute while anything is
|
||||
// not healthy, and the next report carries it whatever happens.
|
||||
HealthSubjects = "mesh.control.*.health"
|
||||
)
|
||||
@@ -89,9 +84,6 @@ func HealthSubject(node string) string { return "mesh.control." + node + ".healt
|
||||
// message the store-window guarantee is about (ADR 0083).
|
||||
func ReportSubject(node string) string { return "mesh.control." + node + ".report" }
|
||||
|
||||
// AliveSubject is one node's heartbeat.
|
||||
func AliveSubject(node string) string { return "mesh.control." + node + ".alive" }
|
||||
|
||||
// ToolsAliveSubject is one machine's node tools' heartbeat.
|
||||
func ToolsAliveSubject(node string) string { return "mesh.control." + node + ".tools-alive" }
|
||||
|
||||
|
||||
@@ -32,8 +32,6 @@ var Contracts = map[string]Contract{
|
||||
KindSourceMoved: {Ordered: "by the merge's commit against what was built from it: a merge already acted on " +
|
||||
"or older than the last look is history, not a second plan (issues 250, 266)",
|
||||
Tests: []string{"TestAMergeOlderThanTheLastLookIsHistory", "TestAMergeMatchesTheSourcesBuiltFromIt"}},
|
||||
KindHeartbeat: {Unordered: "a word that the machine is there: the newest heard is the newest said, and one " +
|
||||
"lost is the next one"},
|
||||
KindToolsHeartbeat: {Unordered: "a word that the node tools are there, as a machine's heartbeat"},
|
||||
KindHealth: {Ordered: "by the time the node-engine looked (`at`, on the machine's clock): a statement older " +
|
||||
"than the one kept for that machine is refused, so an event arriving after a newer report does not undo it " +
|
||||
|
||||
@@ -15,7 +15,7 @@ import (
|
||||
// exists (novox/hq to-be 45 Phase 2, ADR 0227 rule 2).
|
||||
func TestEveryConsumedKindHasAContract(t *testing.T) {
|
||||
// What the controller can be handed, from the subjects it is granted and the ones it derives.
|
||||
subjects := []string{EnrolSubject, BuiltSubject, ReportSubject("anchor"), AliveSubject("anchor"),
|
||||
subjects := []string{EnrolSubject, BuiltSubject, ReportSubject("anchor"),
|
||||
ToolsAliveSubject("anchor"), HealthSubject("anchor"), "mesh.mod.postgres.event.provisioner.failing"}
|
||||
subjects = append(subjects, broker.ControllerFollows...)
|
||||
subjects = append(subjects, broker.ControllerHears...)
|
||||
@@ -83,3 +83,17 @@ func TestEveryConsumedKindHasAContract(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1183/AC3: the controller no longer takes the node-engine's alive message: its subject is no kind, so nothing is
|
||||
// subscribed to it and nothing hears it. The tool runner's tools-alive stays.
|
||||
func TestTheAliveMessageIsNoKindTheControllerHears(t *testing.T) { // 1183/AC3
|
||||
if kind, known := kindOfSubject("mesh.control.anchor.alive"); known {
|
||||
t.Fatalf("the controller still takes the alive message as %s", kind)
|
||||
}
|
||||
if kind, known := kindOfSubject(ToolsAliveSubject("anchor")); !known || kind != KindToolsHeartbeat {
|
||||
t.Fatalf("the tool runner's tools-alive is %q, %v", kind, known)
|
||||
}
|
||||
if _, ok := Contracts["heartbeat"]; ok {
|
||||
t.Fatal("a contract still names the alive message")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -384,9 +384,8 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err
|
||||
}
|
||||
|
||||
// A bare word that a node is there is not an account of what the machine did or holds: it
|
||||
// moves last_seen and touches nothing else. This arrives every minute (link.AliveEvery),
|
||||
// while a real report is rare, so recording it as one would overwrite the node's last real
|
||||
// apply with an empty one — wiping the declaration digest that decides whether the node is
|
||||
// moves last_seen and touches nothing else. Recording one as a report would overwrite the
|
||||
// node's last real apply with an empty one — wiping the declaration digest that decides whether the node is
|
||||
// current, the carried ports a push assigns around, and the clean-or-failed outcome — and a
|
||||
// node that had just caught up would read as behind within the minute. The alive path calls
|
||||
// this with only a node name; a real report always carries an account (something applied, or
|
||||
@@ -394,10 +393,6 @@ func (e Enrolment) Heard(ctx context.Context, report Report) (news bool, err err
|
||||
if report.Applied == nil && report.Refused == "" && len(report.Failed) == 0 {
|
||||
if report.Superseded != "" {
|
||||
log.Printf("%s set aside declaration %s for the newer %s", report.Node, report.Declared, report.Superseded)
|
||||
} else if err := e.Inventory.RecordHeartbeatGap(ctx, report.Node, node.LastSeen); err != nil {
|
||||
// The silence before this word, which the machine-silent bound will be set from (novox/hq
|
||||
// to-be 45 Phase 0). A measurement lost is said and costs the report nothing.
|
||||
log.Printf("the silence before %s's word could not be recorded: %v", report.Node, err)
|
||||
}
|
||||
return false, e.Inventory.Seen(ctx, node.ID)
|
||||
}
|
||||
|
||||
@@ -27,7 +27,6 @@ const ControlQueue = "control"
|
||||
const (
|
||||
KeyEnrol = "enrol"
|
||||
KeyReport = "report"
|
||||
KeyAlive = "alive"
|
||||
)
|
||||
|
||||
// QueueFor is the queue a node consumes from — the only one it may read.
|
||||
@@ -134,19 +133,6 @@ type Signed struct {
|
||||
Signature []byte `json:"signature"`
|
||||
}
|
||||
|
||||
// Alive is a node saying nothing except that it is there.
|
||||
//
|
||||
// How long a node has been out of touch is a fact only the mesh can hold — nobody else is
|
||||
// watching — and without it a node running last month's assignments looks exactly like one that
|
||||
// is current.
|
||||
type Alive struct {
|
||||
Node string `json:"node"`
|
||||
// IntervalSeconds is how often the node says it is there (novox/hq to-be 45 §3, S1): the
|
||||
// watchdog's bound is three of them. Zero from a host older than that, which is read as the
|
||||
// interval hosts have always used.
|
||||
IntervalSeconds int `json:"interval_seconds,omitempty"`
|
||||
}
|
||||
|
||||
// ToolsAlive is a machine's node tools saying they are there (novox/hq to-be 45 §3, S11): the
|
||||
// runtime every module's tools and every held seat's verbs are served by. Its own word, apart from the
|
||||
// host's, because a host heard and a runtime gone is a machine nobody can ask anything.
|
||||
|
||||
@@ -27,7 +27,6 @@ import (
|
||||
const (
|
||||
KindEnrolment = "enrolment"
|
||||
KindReport = "report"
|
||||
KindHeartbeat = "heartbeat"
|
||||
// KindToolsHeartbeat is a machine's node tools saying they are there (novox/hq to-be 45 S11).
|
||||
KindToolsHeartbeat = "tools-heartbeat"
|
||||
// KindHealth is a machine's statement of its long-running resources' health (novox/hq ADR 0240).
|
||||
|
||||
@@ -114,12 +114,8 @@ func (n *natsInbound) Receive(ctx context.Context, act func(context.Context, Con
|
||||
// Heartbeats, on core NATS and off any stream (design 25 §3). Their own subscription because
|
||||
// they are their own guarantee: a lost one is the next one.
|
||||
beats := make(chan *nats.Msg, Prefetch)
|
||||
alive, err := conn.ChanSubscribe(AliveSubjects, beats)
|
||||
if err != nil {
|
||||
return fmt.Errorf("subscribing to heartbeats: %w", err)
|
||||
}
|
||||
defer func() { _ = alive.Unsubscribe() }()
|
||||
// And each machine's node tools, on the same channel: as cheap, and lost the same way.
|
||||
// Each machine's node tools, as cheap and lost the same way. The node-engine's alive message is not subscribed to: a
|
||||
// machine's last word is the watchman's presence (novox/hq ADR 0433 §3, issue 1183).
|
||||
toolsAlive, err := conn.ChanSubscribe(ToolsAliveSubjects, beats)
|
||||
if err != nil {
|
||||
return fmt.Errorf("subscribing to the node tools' heartbeats: %w", err)
|
||||
@@ -283,8 +279,6 @@ func kindOfSubject(subject string) (string, bool) {
|
||||
switch rest {
|
||||
case "report":
|
||||
return KindReport, true
|
||||
case "alive":
|
||||
return KindHeartbeat, true
|
||||
case "tools-alive":
|
||||
return KindToolsHeartbeat, true
|
||||
case "health":
|
||||
|
||||
@@ -157,24 +157,31 @@ func TestNatsAnEnrolmentIsAnsweredOnTheAddressInItsPayload(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestNatsAHeartbeatIsHeardAndNothingIsKept(t *testing.T) {
|
||||
// 1183/AC3, AC5: a node-engine still sending its alive message is harmless: it errors nothing, nothing is recorded and
|
||||
// nothing is persisted; the tool runner's tools-alive is still heard.
|
||||
func TestNatsAnAliveMessageIsHarmlessAndToolsAliveIsStillHeard(t *testing.T) { // 1183/AC3
|
||||
js := aBus(t)
|
||||
store := &counted{}
|
||||
_, stop := servingOn(t, js, store)
|
||||
defer stop()
|
||||
|
||||
// Given time to subscribe: a core subscription that is not yet up misses what is published,
|
||||
// which is the guarantee a heartbeat has and not a fault.
|
||||
eventually(t, "the heartbeat subscription coming up", func() bool {
|
||||
body, _ := json.Marshal(Alive{Node: "anchor"})
|
||||
_ = js.Conn().Publish(AliveSubject("anchor"), body)
|
||||
// Given time to subscribe: a core subscription that is not yet up misses what is published.
|
||||
eventually(t, "the tools-alive subscription coming up", func() bool {
|
||||
if err := js.Conn().Publish("mesh.control.anchor.alive", []byte(`{"node":"anchor"}`)); err != nil {
|
||||
t.Fatalf("an alive message could not be published: %v", err)
|
||||
}
|
||||
body, _ := json.Marshal(ToolsAlive{Node: "anchor"})
|
||||
_ = js.Conn().Publish(ToolsAliveSubject("anchor"), body)
|
||||
_ = js.Conn().Flush()
|
||||
return store.count() >= 1
|
||||
_, heard := ToolsBeats.Of("anchor")
|
||||
return heard
|
||||
})
|
||||
if n := store.count(); n != 0 {
|
||||
t.Fatalf("an alive message was recorded %d time(s)", n)
|
||||
}
|
||||
info, err := js.Context().StreamInfo("CONTROL")
|
||||
if err != nil || info.State.Msgs != 0 {
|
||||
t.Fatalf("a heartbeat was persisted, and the mesh's least valuable message now competes "+
|
||||
"for retention with its most valuable: %+v, %v", info, err)
|
||||
t.Fatalf("an alive message was persisted: %+v, %v", info, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -125,9 +125,6 @@ type enrolsAs struct{ reply EnrolReply }
|
||||
|
||||
func (e enrolsAs) Enrol(context.Context, EnrolRequest) (EnrolReply, error) { return e.reply, nil }
|
||||
|
||||
// A heartbeat is core NATS: it reaches the controller and nothing is persisted, so the stream the
|
||||
// reports live in stays empty.
|
||||
|
||||
// The two events the controller follows arrive over one durable consumer with two filters, and it
|
||||
// can acknowledge them.
|
||||
//
|
||||
|
||||
+2
-26
@@ -205,8 +205,8 @@ func (s *Server) Close() {
|
||||
|
||||
// Serve acts on what arrives until the context ends.
|
||||
func (s *Server) Serve(ctx context.Context) error {
|
||||
s.log.Printf("consuming what nodes say: %s, %s, %s, %s",
|
||||
KindEnrolment, KindReport, KindHeartbeat, KindBuilt)
|
||||
s.log.Printf("consuming what nodes say: %s, %s, %s",
|
||||
KindEnrolment, KindReport, KindBuilt)
|
||||
if s.upgrader != nil {
|
||||
s.log.Printf("following %s", KindModuleMoved)
|
||||
}
|
||||
@@ -226,8 +226,6 @@ func (s *Server) act(ctx context.Context, m Control) {
|
||||
s.enrolling(ctx, m)
|
||||
case KindReport:
|
||||
s.reported(ctx, m)
|
||||
case KindHeartbeat:
|
||||
s.heartbeat(m)
|
||||
case KindToolsHeartbeat:
|
||||
s.toolsHeartbeat(m)
|
||||
case KindHealth:
|
||||
@@ -324,28 +322,6 @@ func (s *Server) outstanding(ctx context.Context, node string) string {
|
||||
return digest
|
||||
}
|
||||
|
||||
// heartbeat records that a node was heard from, and nothing else.
|
||||
//
|
||||
// Deliberately silent: a node saying it is there every minute would fill the log with the ordinary
|
||||
// case, and a log where the ordinary case is loud is a log nobody reads. Not held for the store
|
||||
// either — the next heartbeat is a minute away, and a heartbeat kept for two minutes to be written
|
||||
// late says nothing the one after it will not say better.
|
||||
func (s *Server) heartbeat(m Control) {
|
||||
var alive Alive
|
||||
if err := json.Unmarshal(m.Body(), &alive); err != nil || alive.Node == "" {
|
||||
_ = m.Drop()
|
||||
return
|
||||
}
|
||||
// The interval it says, for the watchdog's bound (novox/hq to-be 45 S1).
|
||||
HostBeats.Heard(alive.Node, time.Now(), time.Duration(alive.IntervalSeconds)*time.Second, "")
|
||||
if s.listener != nil {
|
||||
if _, err := s.listener.Heard(context.Background(), Report{Node: alive.Node}); err != nil {
|
||||
s.log.Printf("could not record that %s is here: %v", alive.Node, err)
|
||||
}
|
||||
}
|
||||
_ = m.Took()
|
||||
}
|
||||
|
||||
// toolsHeartbeat records that a machine's node tools were heard from (novox/hq to-be 45 S11), and
|
||||
// nothing else: in this process's memory, for the watchdog — the next one is a minute away.
|
||||
func (s *Server) toolsHeartbeat(m Control) {
|
||||
|
||||
@@ -121,15 +121,3 @@ func TestAReportCarryingWhatOnlyTheNodeKnowsIsActedOnHoweverOldItIs(t *testing.T
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A heartbeat is not held for the store: the next one is a minute away, and one kept for two
|
||||
// minutes to be written late says nothing the one after it will not say better.
|
||||
func TestAHeartbeatIsNotHeldForTheStore(t *testing.T) {
|
||||
s, in := serving()
|
||||
s.listener = heardWith{err: errors.Join(ErrTryAgain, errors.New("starting up"))}
|
||||
to := &settled{}
|
||||
s.act(context.Background(), in.sends(t, to, KindHeartbeat, Alive{Node: "anchor"}))
|
||||
if !to.acked || len(in.held) != 0 {
|
||||
t.Fatalf("a heartbeat was held for the store: %+v, %d held", *to, len(in.held))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -21,8 +21,7 @@ import (
|
||||
//
|
||||
// **In memory, on purpose.** A heartbeat is the least valuable message the mesh sends — the next one
|
||||
// is a minute away — and what a watchdog needs of it is the newest moment and the interval it said.
|
||||
// A machine's last word is kept in the store as well (`last_seen`), which is what S1 reads; these are
|
||||
// what the store does not keep: the interval, the node tools' word, and when the event loop last took
|
||||
// What the store does not keep: the node tools' word and its interval, and when the event loop last took
|
||||
// a message.
|
||||
|
||||
// Beat is one emitter's newest heartbeat.
|
||||
@@ -43,11 +42,9 @@ type Beats struct {
|
||||
// NewBeats is an empty record, started now.
|
||||
func NewBeats() *Beats { return &Beats{started: time.Now(), heard: map[string]Beat{}} }
|
||||
|
||||
// HostBeats are the node-engines' heartbeats (S1); ToolsBeats the node tools' (S11).
|
||||
var (
|
||||
HostBeats = NewBeats()
|
||||
ToolsBeats = NewBeats()
|
||||
)
|
||||
// ToolsBeats are the node tools' heartbeats (S11). The node-engine's alive message is not heard: a machine's last word is
|
||||
// the watchman's presence (novox/hq ADR 0433 §3).
|
||||
var ToolsBeats = NewBeats()
|
||||
|
||||
// Heard records one heartbeat.
|
||||
func (b *Beats) Heard(node string, at time.Time, every time.Duration, version string) {
|
||||
|
||||
Reference in New Issue
Block a user