The mesh keeps a copy of what each node owns

novox/hq 09-the-node-lifecycle asks for this and it was missing: the host reports what it owns and the mesh keeps the last report. A backup, never a source -- nothing decides anything from it, and a node that disagrees with it wins, because the node is the one that can see the machine. Its point is the orphans. A node that loses its state file currently strands whatever it applied: nothing on the machine knows those resources were the mesh's doing, so nothing removes them. With this, a rebuilt node receives both the declaration and the record of what it previously owned. Never reported and reported nothing are kept apart, and that is the whole care in it. A node that applied nothing holds nothing; a node that has never spoken is unknown -- and handing back an empty list for the second would tell a rebuilding node it owns nothing and have it remove whatever it found. The age comes back with the answer rather than being left for the caller to go and find. An answer about a machine is worth much less without one, and this repository has already been bitten by a cache with no age on it. A refusal or a partial failure moves last_seen and nothing else: neither is an account of what the machine holds, and recording one as though it were would tell a rebuilding node to remove what it still has.
2026-08-29 16:51:54 +02:00
parent 0e116d2d65
commit f563ababa1
6 changed files with 194 additions and 5 deletions
@@ -290,3 +290,87 @@ func TestAnUnknownSecretIsRefusedTheSameWayAsAnExpiredOne(t *testing.T) {
 		t.Errorf("the two are distinguishable:\n  unknown: %v\n  expired: %v", unknownErr, expiredErr)
 	}
 }
+
+func TestNeverReportedIsNotTheSameAsReportedNothing(t *testing.T) {
+	// The distinction that makes this safe to hand back. A node that applied nothing holds
+	// nothing; a node that has never spoken is unknown — and returning an empty list for the
+	// second would tell a rebuilding node it owns nothing, and have it remove whatever it found
+	// on the machine.
+	inv := fresh(t)
+	node, err := inv.AddNode(t.Context(), "laptop")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	owned, reported, err := inv.Owned(t.Context(), node.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if owned != nil {
+		t.Errorf("a node that never reported came back owning %v", owned)
+	}
+	if !reported.IsZero() {
+		t.Error("a node that never reported has a report time")
+	}
+
+	if err := inv.RecordOwned(t.Context(), node.ID, []string{}); err != nil {
+		t.Fatal(err)
+	}
+	owned, reported, err = inv.Owned(t.Context(), node.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if owned == nil {
+		t.Error("a node that reported holding nothing is indistinguishable from one that never spoke")
+	}
+	if reported.IsZero() {
+		t.Error("a report that happened has no time on it")
+	}
+}
+
+func TestWhatANodeOwnsIsReplacedNotAccumulated(t *testing.T) {
+	// The question this answers is what is on that machine now. A node that stopped owning
+	// something and had it remembered would be handed it back on a rebuild and put it there again.
+	inv := fresh(t)
+	node, err := inv.AddNode(t.Context(), "laptop")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := inv.RecordOwned(t.Context(), node.ID, []string{"a", "b", "c"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := inv.RecordOwned(t.Context(), node.ID, []string{"a"}); err != nil {
+		t.Fatal(err)
+	}
+
+	owned, _, err := inv.Owned(t.Context(), node.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(owned) != 1 || owned[0] != "a" {
+		t.Errorf("after reporting a, the mesh believes the node owns %v", owned)
+	}
+}
+
+func TestAnAnswerAboutAMachineCarriesItsAge(t *testing.T) {
+	// This repository has already been bitten by a cache with no age on it: a node running from
+	// one looked identical to a node running from the database. An answer about a machine is
+	// worth much less without knowing how old it is, so the age comes back with it.
+	inv := fresh(t)
+	node, err := inv.AddNode(t.Context(), "laptop")
+	if err != nil {
+		t.Fatal(err)
+	}
+	before := time.Now().Add(-time.Second)
+	if err := inv.RecordOwned(t.Context(), node.ID, []string{"a"}); err != nil {
+		t.Fatal(err)
+	}
+
+	_, reported, err := inv.Owned(t.Context(), node.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if reported.Before(before) {
+		t.Errorf("the report time is %s, which is before the report", reported)
+	}
+}
@@ -0,0 +1,18 @@
+-- The last thing a node said it owns.
+--
+-- novox/hq 09-the-node-lifecycle: the host reports what it owns and the mesh keeps the last
+-- report. **This is a backup, never a source.** The host never reads it to decide anything; it is
+-- handed back only when a node has lost its own store, and a node that disagrees with it wins,
+-- because the node is the one that can see the machine.
+--
+-- Its point is the orphans. A node that loses its state file currently strands whatever it had
+-- applied: nothing on the machine knows those resources were ever the mesh's doing, so nothing
+-- removes them. With this, a rebuilt node receives both the declaration and the record of what it
+-- previously owned, and can take away what is no longer declared.
+
+alter table node add column owned jsonb;
+
+-- When that report arrived. Separate from last_seen, which moves on any word from the node at
+-- all: a node can be plainly alive for weeks without applying anything, and reading one as the
+-- other would make a quiet node look like a stale one.
+alter table node add column owned_reported timestamptz;
@@ -233,3 +233,53 @@ func (i *Inventory) Seen(ctx context.Context, node string) error {
 	_, err := i.store.Pool().Exec(ctx, `update node set last_seen = now() where id = $1`, node)
 	return err
 }
+
+// RecordOwned keeps the last account a node gave of what it holds.
+//
+// A copy for recovery and never a source (novox/hq 09-the-node-lifecycle). Nothing here decides
+// anything from it; it is handed back to a node that has lost its own store, and if that node
+// then disagrees, the node wins — it is the one that can see the machine.
+//
+// Replaced rather than appended. A history of what a node used to own answers a question nobody
+// asks, and the one question this does answer — what is on that machine now — is only answered by
+// the latest.
+func (i *Inventory) RecordOwned(ctx context.Context, node string, owned []string) error {
+	raw, err := json.Marshal(owned)
+	if err != nil {
+		return err
+	}
+	_, err = i.store.Pool().Exec(ctx,
+		`update node set owned = $2, owned_reported = now(), last_seen = now() where id = $1`,
+		node, raw)
+	return err
+}
+
+// Owned is what a node last said it holds, and when it said so.
+//
+// The age is returned with it rather than left to the caller to look up, because an answer about
+// a machine is worth much less without one — and this repository has already been bitten by a
+// cache with no age on it.
+func (i *Inventory) Owned(ctx context.Context, node string) ([]string, time.Time, error) {
+	var raw []byte
+	var reported *time.Time
+	err := i.store.Pool().QueryRow(ctx,
+		`select owned, owned_reported from node where id = $1`, node).Scan(&raw, &reported)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return nil, time.Time{}, fmt.Errorf("%w: %s", ErrNoSuchNode, node)
+	}
+	if err != nil {
+		return nil, time.Time{}, err
+	}
+	if len(raw) == 0 || reported == nil {
+		// Never reported is not the same as reported nothing. A node that has applied nothing
+		// holds nothing; a node that has never spoken is unknown, and handing back an empty list
+		// as though it were a report would tell a rebuilding node it owns nothing and have it
+		// remove whatever it found.
+		return nil, time.Time{}, nil
+	}
+	var owned []string
+	if err := json.Unmarshal(raw, &owned); err != nil {
+		return nil, time.Time{}, err
+	}
+	return owned, *reported, nil
+}
@@ -106,3 +106,25 @@ var _ Enroller = Enrolment{}

 // ErrNoBrokerManagement is returned when an account cannot be made because nothing was configured.
 var ErrNoBrokerManagement = errors.New("no broker management configured")
+
+// Heard records what a node reported about itself.
+//
+// A node states; the owning context writes (novox/hq ADR 0006). What a node says it applied is
+// its own account of its own machine, kept as a copy for recovery — so this writes it down and
+// decides nothing from it.
+func (e Enrolment) Heard(ctx context.Context, report Report) error {
+	if report.Node == "" {
+		return errors.New("a report named no node")
+	}
+	node, err := e.Inventory.NodeByName(ctx, report.Node)
+	if err != nil {
+		return err
+	}
+	// A refusal or a failure is not an account of what the machine holds, so it moves last_seen
+	// and nothing else. Recording a partial list as though it were the whole would tell a
+	// rebuilding node to remove what it still has.
+	if report.Refused != "" || len(report.Failed) > 0 {
+		return e.Inventory.Seen(ctx, node.ID)
+	}
+	return e.Inventory.RecordOwned(ctx, node.ID, report.Applied)
+}
@@ -28,15 +28,22 @@ type Enroller interface {
 }

 // Server consumes what nodes say.
+// Listener is what the control plane 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
+}
+
 type Server struct {
 	conn     *amqp.Connection
 	channel  *amqp.Channel
 	enroller Enroller
+	listener Listener
 	log      *log.Logger
 }

 // Connect opens the control plane's own connection to the broker.
-func Connect(enroller Enroller) (*Server, error) {
+func Connect(enroller Enroller, listener Listener) (*Server, error) {
 	url := strings.TrimSpace(os.Getenv(AMQPVar))
 	if url == "" {
 		return nil, fmt.Errorf(
@@ -79,7 +86,7 @@ func Connect(enroller Enroller) (*Server, error) {
 		}
 	}

-	return &Server{conn: conn, channel: channel, enroller: enroller,
+	return &Server{conn: conn, channel: channel, enroller: enroller, listener: listener,
 		log: log.New(os.Stdout, "", log.LstdFlags)}, nil
 }

@@ -161,6 +168,14 @@ func (s *Server) handleReport(delivery amqp.Delivery) {
 		_ = delivery.Reject(false)
 		return
 	}
+	if s.listener != nil {
+		if err := s.listener.Heard(context.Background(), report); err != nil {
+			// Said rather than swallowed. A report the mesh heard and failed to write down is a
+			// node whose recovery copy is silently older than it looks.
+			s.log.Printf("could not record %s's report: %v", report.Node, err)
+		}
+	}
+
 	switch {
 	case report.Refused != "":
 		s.log.Printf("%s refused a declaration: %s", report.Node, report.Refused)