diff --git a/internal/link/bus.go b/internal/link/bus.go index c07d8fd..62bd227 100644 --- a/internal/link/bus.go +++ b/internal/link/bus.go @@ -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..>` 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 diff --git a/internal/link/hearing_nats.go b/internal/link/hearing_nats.go index dbe3d79..ba8d84b 100644 --- a/internal/link/hearing_nats.go +++ b/internal/link/hearing_nats.go @@ -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) } diff --git a/internal/link/hearing_nats_test.go b/internal/link/hearing_nats_test.go index 0226bc7..f32bc88 100644 --- a/internal/link/hearing_nats_test.go +++ b/internal/link/hearing_nats_test.go @@ -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) } } diff --git a/internal/link/messages.go b/internal/link/messages.go index b480f75..b8ee481 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -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 diff --git a/internal/link/nothing_alive_test.go b/internal/link/nothing_alive_test.go new file mode 100644 index 0000000..300bea5 --- /dev/null +++ b/internal/link/nothing_alive_test.go @@ -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) + } +} diff --git a/internal/link/run.go b/internal/link/run.go index 72028b9..4628640 100644 --- a/internal/link/run.go +++ b/internal/link/run.go @@ -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). diff --git a/internal/link/unsaid_test.go b/internal/link/unsaid_test.go index 85e3361..13257fd 100644 --- a/internal/link/unsaid_test.go +++ b/internal/link/unsaid_test.go @@ -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()