The mesh says what it applied, and the replay has an address it may use
The pipeline was observable from a merge to an artifact and went dark where it touched a machine: a node's report is control traffic only the control plane reads, so nothing said which version a machine runs, or that it refused to (novox/hq ADR 0134). The control plane now states both under the seat it holds — a role's events belong to the role and keep their address when the holder is replaced — and only when the report is news, because a machine reconciles every minute and a fact per report would be a fact per minute per machine. Whether a report is news is the store's answer: it holds the previous one, so the listener returns it and the server states the fact. That also gives the catch-up replay a subject the controller may publish: it was published as a module's event from a module called "control-plane", which does not exist, so the controller's own account refused it and every catalogue that asked what it missed was answered with nothing.
This commit is contained in:
+66
-4
@@ -29,7 +29,12 @@ type Enroller interface {
|
||||
// Listener is what the controller does with a report. Separate from Enroller so the two can be
|
||||
// given independently, and so a server that only sends declarations needs neither.
|
||||
type Listener interface {
|
||||
Heard(ctx context.Context, report Report) error
|
||||
// Heard records what a node said, and says whether it was **news** — a machine that now runs
|
||||
// something else, or refuses something it did not refuse before. A machine reconciles
|
||||
// continuously and reports each time, so what is news is the store's answer rather than the
|
||||
// bus's: only this side has the previous report to compare with. What the mesh states about it
|
||||
// is the server's (novox/hq ADR 0134).
|
||||
Heard(ctx context.Context, report Report) (news bool, err error)
|
||||
}
|
||||
|
||||
// Recorder keeps what builders say.
|
||||
@@ -257,7 +262,7 @@ func (s *Server) heartbeat(m Control) {
|
||||
return
|
||||
}
|
||||
if s.listener != nil {
|
||||
if err := s.listener.Heard(context.Background(), Report{Node: alive.Node}); err != 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)
|
||||
}
|
||||
}
|
||||
@@ -291,7 +296,7 @@ func (s *Server) reported(ctx context.Context, m Control) {
|
||||
return
|
||||
}
|
||||
|
||||
err := s.listener.Heard(context.Background(), report)
|
||||
news, err := s.listener.Heard(context.Background(), report)
|
||||
switch s.decide(ctx, m, what, declaredIn, outstanding, err) {
|
||||
case Hold:
|
||||
// Held, not settled, while the store cannot take it: the node reports an apply once,
|
||||
@@ -307,6 +312,13 @@ func (s *Server) reported(ctx context.Context, m Control) {
|
||||
// node whose recovery copy is silently older than it looks.
|
||||
s.log.Printf("could not record %s's report: %v", report.Node, err)
|
||||
}
|
||||
// **And the mesh says what it did** (novox/hq ADR 0134). Only when the report was news: a
|
||||
// machine reports every convergence, and a fact per report would be a fact per minute per
|
||||
// machine saying nothing. Stated after it is recorded, so nothing is announced that the
|
||||
// mesh does not hold.
|
||||
if err == nil && news {
|
||||
s.saysWhatItDid(ctx, report)
|
||||
}
|
||||
}
|
||||
|
||||
switch {
|
||||
@@ -460,7 +472,11 @@ func (s *Server) catchingUp(ctx context.Context, m Control) {
|
||||
sent := 0
|
||||
for _, a := range announcements {
|
||||
a.Replay = true
|
||||
if err := EmitEvent(ctx, s.bus, KeyModuleBuilt, "control-plane", "", a); err != nil {
|
||||
// Under the control plane's own seat (novox/hq ADR 0134). It used to be published as a
|
||||
// module's event from a module called "control-plane", which does not exist — so the
|
||||
// controller's own account refused it, every catalogue that asked what it missed was
|
||||
// answered with nothing, and its graph kept the gap (found 2026-09-28).
|
||||
if err := s.bus.PublishSeatEvent(ctx, MeshControllerSeat, KeyBuiltBefore, replayed(a)); err != nil {
|
||||
// Said and abandoned rather than retried: the catalogue asks again every time it
|
||||
// starts, and half a graph delivered twice is no better than half delivered once.
|
||||
s.log.Printf("replaying %s at %s failed, and the rest is abandoned: %v",
|
||||
@@ -551,3 +567,49 @@ func (s *Server) sourceMoved(ctx context.Context, m Control) {
|
||||
}
|
||||
_ = m.Took()
|
||||
}
|
||||
|
||||
// saysWhatItDid states what a machine now runs, or what it would not take, as a fact on the bus
|
||||
// (novox/hq ADR 0134).
|
||||
//
|
||||
// **The control plane speaks, as the holder of its seat.** A node's report is control traffic only
|
||||
// this process may read, so the chain from a merge to a machine went dark exactly where it touched
|
||||
// one: nothing said which version a machine runs, or that it refused to. The facts are second-hand
|
||||
// on purpose — one emitter, one ordering — and a machine that cannot reach the bus produces none, so
|
||||
// absence is not health.
|
||||
//
|
||||
// A failure to state a fact is logged and nothing else: the report is recorded, which is the part
|
||||
// that must not be lost, and the next change says the same thing again.
|
||||
func (s *Server) saysWhatItDid(ctx context.Context, report Report) {
|
||||
if s.bus == nil {
|
||||
return
|
||||
}
|
||||
event, body := KeyApplied, any(Applied{
|
||||
Node: report.Node, Declared: report.Declared, Resources: len(report.Applied),
|
||||
})
|
||||
if report.Refused != "" || len(report.Failed) > 0 {
|
||||
event, body = KeyRefused, Refused{
|
||||
Node: report.Node, Declared: report.Declared,
|
||||
Refused: report.Refused, Failed: report.Failed,
|
||||
}
|
||||
}
|
||||
raw, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
s.log.Printf("could not say what %s did: %v", report.Node, err)
|
||||
return
|
||||
}
|
||||
if err := s.bus.PublishSeatEvent(ctx, MeshControllerSeat, event, raw); err != nil {
|
||||
s.log.Printf("could not say that %s %s: %v", report.Node, event, err)
|
||||
}
|
||||
}
|
||||
|
||||
// replayed is one announcement as the control plane states it. The same body the build machine's
|
||||
// outcome carries, because what the catalogue does with it is the same.
|
||||
func replayed(a Announcement) []byte {
|
||||
raw, err := json.Marshal(a)
|
||||
if err != nil {
|
||||
// A body that cannot be marshalled is a programming error, not a bus failure, and an empty
|
||||
// one is refused by the reader rather than silently taken as an announcement of nothing.
|
||||
return nil
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user