novox/mesh-host
13
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Merge pull request 'The node-engine sends no alive message: presence replaces it (design 47 R2d, issue 1158)' (#120) from feat/alive-gone into main

Browse Source
This commit was merged in pull request #120.
This commit is contained in:
mesh-tools
2026-10-19 07:07:56 +00:00
parent 89025ad004 671f284004
commit 2474bf2833
7 changed files with 61 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/link/bus.go
+1 -20
View File
@@ -26,11 +26,6 @@ type Bus interface {
// disagreeing with nothing anywhere saying so — the shape of fault this project keeps
// finding. Returns false when it could not be delivered, so the caller can say so.
Report(ctx context.Context, node string, body []byte) error
// Alive says this node is here, and nothing else. **Losing one is nothing**: the next is a
// minute away and the mesh reads a gap rather than counting arrivals. Insisting on delivery
// would turn a harmless miss into a logged failure every minute.
Alive(ctx context.Context, node string, body []byte) error
}
// --- The bus the mesh runs on today -----------------------------------------------------------
@@ -42,11 +37,10 @@ type OverNATS struct {
JS nats.JetStreamContext
}
// ReportSubject and AliveSubject are this node's own, and no other node's: a host's account may
// ReportSubject is this node's own, and no other node's: a host's account may
// publish `mesh.control.<its own node>.>` and nothing wider, so the subject is the authority on
// which node a report is about.
func ReportSubject(node string) string { return "mesh.control." + node + ".report" }
func AliveSubject(node string) string { return "mesh.control." + node + ".alive" }
// HealthSubject is where this node says its long-running resources' health between reports (novox/hq
// ADR 0240): its own, inside the grant every host already has, and on core NATS like the heartbeat — a
@@ -105,19 +99,6 @@ func (b OverNATS) Report(ctx context.Context, node string, body []byte) error {
return nil
}
func (b OverNATS) Alive(ctx context.Context, node string, body []byte) error {
// Core, deliberately: a heartbeat in a stream is the mesh's least valuable message competing
// for retention with its most valuable, and a lost one is the next one.
if err := b.Conn.Publish(AliveSubject(node), body); err != nil {
return err
}
// Flushed rather than fired and forgotten, so "could not tell the mesh" means the write
// failed rather than that nobody has looked yet.
flush, cancel := context.WithTimeout(ctx, 2*time.Second)
defer cancel()
return b.Conn.FlushWithContext(flush)
}
// HaveBind binds the have bucket the controller made (novox/hq ADR 0429 §4, issue 1132), once for a pass of writes.
// **Only binds**: a node-engine's account reaches no part of the JetStream API beyond what it is granted, and the bucket
// is the controller's to create and to configure, so a bucket that is not there is an error said as what it is. Every
internal/link/hearing_nats.go
-4
View File
@@ -334,10 +334,6 @@ func (l *natsLink) Report(ctx context.Context, node string, body []byte) error {
return OverNATS{Conn: l.conn, JS: l.js}.Report(ctx, node, body)
}
func (l *natsLink) Alive(ctx context.Context, node string, body []byte) error {
return OverNATS{Conn: l.conn, JS: l.js}.Alive(ctx, node, body)
}
func (l *natsLink) Health(ctx context.Context, node string, body []byte) error {
return OverNATS{Conn: l.conn, JS: l.js}.Health(ctx, node, body)
}
internal/link/hearing_nats_test.go
+3 -8
View File
@@ -203,9 +203,8 @@ func TestNatsANodeThatWasAwayGetsOnlyTheNewest(t *testing.T) {
}
}
// A report goes through the stream and a heartbeat does not: the one that must survive the
// controller's store restarting is kept, and the one that must not is not.
func TestNatsAReportIsKeptAndAHeartbeatIsNot(t *testing.T) {
// A report goes through the stream: it must survive the controller's store restarting.
func TestNatsAReportIsKept(t *testing.T) {
conn, js := aBus(t)
bus := OverNATS{Conn: conn, JS: js}
ctx := context.Background()
@@ -214,17 +213,13 @@ func TestNatsAReportIsKeptAndAHeartbeatIsNot(t *testing.T) {
if err := bus.Report(ctx, "anchor", body); err != nil {
t.Fatal(err)
}
beat, _ := json.Marshal(Alive{Node: "anchor"})
if err := bus.Alive(ctx, "anchor", beat); err != nil {
t.Fatal(err)
}
info, err := js.StreamInfo("CONTROL")
if err != nil {
t.Fatal(err)
}
if info.State.Msgs != 1 {
t.Fatalf("%d messages were kept; a report must be and a heartbeat must not", info.State.Msgs)
t.Fatalf("%d messages were kept; a report must be, once", info.State.Msgs)
}
}
internal/link/messages.go
-18
View File
@@ -16,26 +16,8 @@ import (
// queue, so it can say these things and nothing else.
const (
KeyReport = "report"
KeyAlive = "alive"
)
// Alive is a node saying nothing except that it is there.
//
// novox/hq 09-the-node-lifecycle: *how long it has been disconnected is a fact the mesh must
// hold, and nothing holds it today. Without it, a node running last month's assignments looks
// exactly like one that is current.*
//
// Separate from a report because the two happen at completely different rates: a node is alive
// constantly and applies something rarely, and reading one as the other would make a quiet node
// look like a stale one.
type Alive struct {
Node string `json:"node"`
// IntervalSeconds is how often this node says it is there (novox/hq to-be 45 §3, S1): the
// controller's watchdog is bound to three of them, so the bound moves with the interval rather
// than with a number the controller guessed. A controller older than that ignores it.
IntervalSeconds int `json:"interval_seconds"`
}
// Signed is a declaration and the signature over it.
//
// novox/hq ADR 0004: the transport is verified once at connect, and **each declaration is
internal/link/nothing_alive_test.go
+53
View File
@@ -0,0 +1,53 @@
package link
import (
"context"
"sync"
"testing"
"time"
)
// listeningForAlive is a quiet link that also records anything said on an alive subject. The method
// is deliberately kept here: once the bus has no such method, this link can never be asked for it,
// and the test still holds the loop to saying nothing.
type listeningForAlive struct {
*quietLink
mu sync.Mutex
heard int
}
func (l *listeningForAlive) Alive(context.Context, string, []byte) error {
l.mu.Lock()
defer l.mu.Unlock()
l.heard++
return nil
}
func (l *listeningForAlive) count() int {
l.mu.Lock()
defer l.mu.Unlock()
return l.heard
}
// 1158/AC1: the node-engine publishes nothing on its alive subject (novox/hq ADR 0429 §7, ADR 0433
// §3: presence replaces the alive message). The loop runs over a fake bus and no alive is published,
// neither at the start nor on a tick.
func TestTheLinkLoopPublishesNoAlive(t *testing.T) {
m, _ := aMember(t)
l := &listeningForAlive{quietLink: newQuietLink()}
q := aQueue(m, nil, &keptInMemory{})
ctx, stop := context.WithCancel(context.Background())
done := make(chan error, 1)
go func() { done <- serve(ctx, l, m, q, nil, time.Second) }()
time.Sleep(300 * time.Millisecond)
stop()
select {
case <-done:
case <-time.After(5 * time.Second):
t.Fatal("the loop did not end")
}
if n := l.count(); n != 0 {
t.Fatalf("the node published %d alive messages (1158/AC1)", n)
}
}
internal/link/run.go
+1 -34
View File
@@ -18,13 +18,6 @@ import (
// the first means somebody is trying, the second means something is broken.
var ErrForged = errors.New("this declaration was not signed by the mesh this node joined")
// AliveEvery is how often a node says it is there.
//
// Often enough that "no word for five minutes" means something, rarely enough that a hundred
// nodes are not a hundred messages a second. The mesh reads absence rather than presence, so what
// matters is the interval being known and steady.
const AliveEvery = 60 * time.Second
// Membership is what a node needs to reach its mesh again, held by the caller.
type Membership struct {
Node string
@@ -218,13 +211,6 @@ func serve(ctx context.Context, link Link, m Membership, queue *Queue, say Annou
// still broken when it means the opposite.
say("in the mesh, hearing what this node should be")
// A word every so often, so the mesh can tell a node that is quiet from one that is gone.
// Cheap on purpose: it carries a name and nothing else, because anything more would be a
// report, and reports are rare where this is constant.
beat := time.NewTicker(AliveEvery)
defer beat.Stop()
publishAlive(ctx, link, m, say, timeout)
// **What the last apply did, if the mesh never heard it**, is said before anything newly
// delivered can be: the queue says it as the link is handed to it (novox/hq issue 264). And the
// link is not let go while the queue's worker is mid-act: an apply that ends the link — the host
@@ -269,8 +255,6 @@ func serve(ctx context.Context, link Link, m Membership, queue *Queue, say Annou
select {
case <-ctx.Done():
return nil
case <-beat.C:
publishAlive(ctx, link, m, say, timeout)
case reason := <-link.Lost():
return reason
case <-asks:
@@ -288,7 +272,7 @@ func serve(ctx context.Context, link Link, m Membership, queue *Queue, say Annou
}
// Whatever else is already waiting goes with it. **Enqueued, not applied**: the queue
// applies the newest of everything delivered and not yet taken, and reports the rest as
// set aside (novox/hq to-be 45 §6). The link goes on beating meanwhile.
// set aside (novox/hq to-be 45 §6).
queue.Deliver(gather(declarations, declaration, drainWindow)...)
}
}
@@ -396,23 +380,6 @@ func publishReport(ctx context.Context, bus Bus, m Membership, report Report,
return true
}
// publishAlive says this node is here, and nothing else.
func publishAlive(ctx context.Context, bus Bus, m Membership, say Announce,
timeout time.Duration) {
body, err := json.Marshal(Alive{Node: m.Node, IntervalSeconds: int(AliveEvery / time.Second)})
if err != nil {
return
}
publish, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
// Not mandatory, unlike a report. Losing one is nothing: the next is a minute away, and the
// mesh is reading a gap rather than counting arrivals. Insisting on delivery would turn a
// harmless miss into a logged failure every minute.
if err := bus.Alive(publish, m.Node, body); err != nil {
say("could not tell the mesh this node is here: " + err.Error())
}
}
// heldNote is what this apply did NOT do, for the line that says what it did.
//
// **A count that does not add up is the only symptom a held resource had** (novox/hq 04-ISSUES/125).
internal/link/unsaid_test.go
+3 -4
View File
@@ -48,10 +48,9 @@ func (l *quietLink) Report(ctx context.Context, _ string, body []byte) error {
l.reports = append(l.reports, r)
return nil
}
func (l *quietLink) Alive(context.Context, string, []byte) error { return nil }
func (l *quietLink) Declarations() <-chan Declaration { return l.delivered }
func (l *quietLink) Lost() <-chan error { return l.lost }
func (l *quietLink) Close() {}
func (l *quietLink) Declarations() <-chan Declaration { return l.delivered }
func (l *quietLink) Lost() <-chan error { return l.lost }
func (l *quietLink) Close() {}
func (l *quietLink) said() []Report {
l.mu.Lock()