Record whether a node is adopted and which modules were taken on it (hq ADR 0100)

internal/inventory/adoption.go

@@ -0,0 +1,149 @@
+package inventory
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+)
+
+// A node is adopted or converged (novox/hq ADR 0100).
+//
+// Adopted: what is found on the machine is kept until its module is taken, and the firewall found
+// there stays in force. Converged: the machine is what the mesh declares, as every node was before
+// adoption existed. The controller is authoritative, and every declaration it sends says which.
+
+// ErrNotAdopted is taking a module on a node that is converged. On a converged node every assigned
+// module converges already; there is nothing to take.
+var ErrNotAdopted = errors.New("the node is converged, so every module on it is taken already")
+
+// ErrNotAssigned is taking a module that is not on the node. Taking is the cutover of a module
+// the node runs; one it does not run has nothing to cut over.
+var ErrNotAssigned = errors.New("that module is not assigned to the node")
+
+// SetAdopted makes a node adopted or converged. Becoming adopted stamps when; converging stamps
+// when too. Neither touches what was taken: what was taken stays taken when a node returns to
+// adopted, and converging takes the rest by its own act.
+func (i *Inventory) SetAdopted(ctx context.Context, name string, adopted bool) error {
+	node, err := i.NodeByName(ctx, name)
+	if err != nil {
+		return err
+	}
+	if node.Adopted == adopted {
+		return nil
+	}
+	if adopted {
+		_, err = i.store.Pool().Exec(ctx,
+			`update node set adopted = true, adopted_since = now() where id = $1`, node.ID)
+		return err
+	}
+	_, err = i.store.Pool().Exec(ctx,
+		`update node set adopted = false, adopted_since = null, converged_at = now() where id = $1`,
+		node.ID)
+	return err
+}
+
+// Take records that a module has been taken on an adopted node: its cutover. From then on the
+// module's resources converge on that node like any other, replacing what was found.
+//
+// Refused on a converged node and for a module not assigned there. Taking again is not an error;
+// the first time it was taken is kept.
+func (i *Inventory) Take(ctx context.Context, nodeName, module string) error {
+	node, err := i.NodeByName(ctx, nodeName)
+	if err != nil {
+		return err
+	}
+	if !node.Adopted {
+		return fmt.Errorf("%w: %s", ErrNotAdopted, nodeName)
+	}
+	return i.take(ctx, node, module)
+}
+
+// take is Take without the adopted check, for converging, which takes every assigned module in
+// the same act that makes the node converged.
+func (i *Inventory) take(ctx context.Context, node Node, module string) error {
+	var assigned bool
+	if err := i.store.Pool().QueryRow(ctx,
+		`select exists (select 1 from assignment where node = $1 and module = $2)`,
+		node.ID, module).Scan(&assigned); err != nil {
+		return err
+	}
+	if !assigned {
+		return fmt.Errorf("%w: %s is not on %s; assign it first", ErrNotAssigned, module, node.Name)
+	}
+	_, err := i.store.Pool().Exec(ctx,
+		`insert into taken (node, module) values ($1, $2) on conflict do nothing`, node.ID, module)
+	return err
+}
+
+// Converge makes an adopted node converged in one act: every module assigned there is taken, and
+// the node is recorded converged. Returned is what this act took, in name order.
+func (i *Inventory) Converge(ctx context.Context, nodeName string) ([]string, error) {
+	node, err := i.NodeByName(ctx, nodeName)
+	if err != nil {
+		return nil, err
+	}
+	if !node.Adopted {
+		return nil, fmt.Errorf("%s is converged already", nodeName)
+	}
+	tx, err := i.store.Pool().Begin(ctx)
+	if err != nil {
+		return nil, err
+	}
+	defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
+	rows, err := tx.Query(ctx,
+		`insert into taken (node, module)
+		 select node, module from assignment where node = $1
+		 on conflict do nothing
+		 returning module`, node.ID)
+	if err != nil {
+		return nil, err
+	}
+	var took []string
+	for rows.Next() {
+		var m string
+		if err := rows.Scan(&m); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		took = append(took, m)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	if _, err := tx.Exec(ctx,
+		`update node set adopted = false, adopted_since = null, converged_at = now() where id = $1`,
+		node.ID); err != nil {
+		return nil, err
+	}
+	if err := tx.Commit(ctx); err != nil {
+		return nil, err
+	}
+	sort.Strings(took)
+	return took, nil
+}
+
+// Taken is every module taken on a node, in name order — including one no longer assigned there:
+// unassigning does not un-take.
+func (i *Inventory) Taken(ctx context.Context, nodeName string) ([]string, error) {
+	node, err := i.NodeByName(ctx, nodeName)
+	if err != nil {
+		return nil, err
+	}
+	rows, err := i.store.Pool().Query(ctx,
+		`select module from taken where node = $1 order by module`, node.ID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []string
+	for rows.Next() {
+		var m string
+		if err := rows.Scan(&m); err != nil {
+			return nil, err
+		}
+		out = append(out, m)
+	}
+	return out, rows.Err()
+}

internal/inventory/adoption_test.go

@@ -0,0 +1,134 @@
+package inventory
+
+import (
+	"errors"
+	"reflect"
+	"testing"
+)
+
+// novox/hq ADR 0100: a node is adopted or converged, and the controller records which.
+
+func TestANodeAddedWithoutSayingIsConverged(t *testing.T) {
+	inv := fresh(t)
+	if _, err := inv.AddNode(t.Context(), "anchor"); err != nil {
+		t.Fatal(err)
+	}
+	n, err := inv.NodeByName(t.Context(), "anchor")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if n.Adopted || !n.AdoptedSince.IsZero() {
+		t.Fatalf("a node nobody said anything about reads as adopted: %+v", n)
+	}
+}
+
+func TestAnAdoptedNodeRoundTripsThroughEveryReading(t *testing.T) {
+	inv := fresh(t)
+	made, err := inv.AddNodeAs(t.Context(), "anchor", true)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !made.Adopted || made.AdoptedSince.IsZero() {
+		t.Fatalf("added adopted, got %+v", made)
+	}
+	byName, err := inv.NodeByName(t.Context(), "anchor")
+	if err != nil {
+		t.Fatal(err)
+	}
+	all, err := inv.Nodes(t.Context())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !byName.Adopted || len(all) != 1 || !all[0].Adopted {
+		t.Fatalf("adoption did not survive reading back: %+v %+v", byName, all)
+	}
+
+	// And through a token: enrolment reads the node from the token it spends.
+	issued, err := inv.IssueToken(t.Context(), "anchor", 60e9)
+	if err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := inv.Claim(t.Context(), issued.Secret, "a-key", false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !claimed.Adopted {
+		t.Fatal("the node a token claims lost its mode")
+	}
+}
+
+func TestTakingIsRefusedOnAConvergedNodeAndForAnUnassignedModule(t *testing.T) {
+	inv := fresh(t)
+	if err := inv.RegisterModule(t.Context(), manifest("hello-web", nil, nil), Source{}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := inv.AddNode(t.Context(), "converged"); err != nil {
+		t.Fatal(err)
+	}
+	if err := inv.Assign(t.Context(), "converged", "hello-web"); err != nil {
+		t.Fatal(err)
+	}
+	if err := inv.Take(t.Context(), "converged", "hello-web"); !errors.Is(err, ErrNotAdopted) {
+		t.Fatalf("taking on a converged node gave %v", err)
+	}
+
+	if _, err := inv.AddNodeAs(t.Context(), "anchor", true); err != nil {
+		t.Fatal(err)
+	}
+	if err := inv.Take(t.Context(), "anchor", "hello-web"); !errors.Is(err, ErrNotAssigned) {
+		t.Fatalf("taking an unassigned module gave %v", err)
+	}
+}
+
+func TestATakenModuleOutlivesItsAssignmentAndReturningToAdopted(t *testing.T) {
+	inv := fresh(t)
+	for _, m := range []string{"hello-web", "postgres"} {
+		if err := inv.RegisterModule(t.Context(), manifest(m, nil, nil), Source{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := inv.AddNodeAs(t.Context(), "anchor", true); err != nil {
+		t.Fatal(err)
+	}
+	for _, m := range []string{"hello-web", "postgres"} {
+		if err := inv.Assign(t.Context(), "anchor", m); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := inv.Take(t.Context(), "anchor", "postgres"); err != nil {
+		t.Fatal(err)
+	}
+	if err := inv.Take(t.Context(), "anchor", "postgres"); err != nil {
+		t.Fatalf("taking twice is not an error: %v", err)
+	}
+	if err := inv.Unassign(t.Context(), "anchor", "postgres"); err != nil {
+		t.Fatal(err)
+	}
+	taken, err := inv.Taken(t.Context(), "anchor")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(taken, []string{"postgres"}) {
+		t.Fatalf("unassigning un-took it: %v", taken)
+	}
+
+	took, err := inv.Converge(t.Context(), "anchor")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(took, []string{"hello-web"}) {
+		t.Fatalf("converging took %v; it takes every assigned module not yet taken", took)
+	}
+	n, _ := inv.NodeByName(t.Context(), "anchor")
+	if n.Adopted {
+		t.Fatal("converged node still reads adopted")
+	}
+
+	if err := inv.SetAdopted(t.Context(), "anchor", true); err != nil {
+		t.Fatal(err)
+	}
+	taken, _ = inv.Taken(t.Context(), "anchor")
+	if !reflect.DeepEqual(taken, []string{"hello-web", "postgres"}) {
+		t.Fatalf("returning to adopted lost what was taken: %v", taken)
+	}
+}

internal/inventory/migrations/0029-a-node-is-adopted-or-converged.sql

@@ -0,0 +1,21 @@
+-- A node is adopted or converged, and the mesh records which (novox/hq ADR 0100).
+--
+-- A machine already serving a predecessor's services is adopted: what is found on it is kept
+-- until its module is taken, and the firewall found on it stays in force. It stays adopted until
+-- the operator converges it. False by default, so every node that exists is converged, as it was.
+
+alter table node add column adopted boolean not null default false;
+-- When it was last made adopted, and when it last converged. Both kept: a node returned to adopted
+-- after converging is a different history from one that never converged.
+alter table node add column adopted_since timestamptz;
+alter table node add column converged_at timestamptz;
+
+-- The modules taken on a node: its cutover, the operator's act, done when the module's data has
+-- moved. A row per node and module, and it outlives the assignment on purpose -- unassigning a
+-- module does not un-take it, and what was taken stays taken when a node returns to adopted.
+create table taken (
+    node      uuid        not null references node(id) on delete cascade,
+    module    text        not null references module(name) on delete cascade,
+    taken_at  timestamptz not null default now(),
+    primary key (node, module)
+);

internal/inventory/nodes.go

@@ -49,6 +49,12 @@ type Node struct {
 	// month's assignments, and until this existed those looked the same as a node that is
 	// current (novox/hq 09-the-node-lifecycle).
 	LastSeen time.Time
+
+	// Adopted is whether this node is adopted rather than converged (novox/hq ADR 0100): what is
+	// found on it is kept until its module is taken, and the firewall found on it stays in force.
+	// AdoptedSince is when it last became so; zero for a converged node.
+	Adopted      bool
+	AdoptedSince time.Time
 }
 
 // Silent is how long since this node was last heard from, and whether it ever was.
@@ -74,28 +80,59 @@ var ErrNameTaken = errors.New("a node of that name already exists")
 // (novox/hq 09-the-node-lifecycle), so the record is what a token binds to and must exist before
 // there is anything to join.
 func (i *Inventory) AddNode(ctx context.Context, name string) (Node, error) {
+	return i.AddNodeAs(ctx, name, false)
+}
+
+// AddNodeAs creates a node record, adopted or converged (novox/hq ADR 0100). The operator says
+// which; a node added without saying is converged, as every node was before adoption existed.
+func (i *Inventory) AddNodeAs(ctx context.Context, name string, adopted bool) (Node, error) {
 	name = strings.TrimSpace(name)
 	if name == "" {
 		return Node{}, errors.New("a node needs a name: it is how a token is issued for it")
 	}
 
 	var n Node
+	var since *time.Time
 	err := i.store.Pool().QueryRow(ctx,
-		`insert into node (name) values ($1) returning id, name, created`,
-		name).Scan(&n.ID, &n.Name, &n.Created)
+		`insert into node (name, adopted, adopted_since)
+		 values ($1, $2, case when $2 then now() end)
+		 returning id, name, created, adopted, adopted_since`,
+		name, adopted).Scan(&n.ID, &n.Name, &n.Created, &n.Adopted, &since)
 	if err != nil {
 		if strings.Contains(err.Error(), "node_name_key") {
 			return Node{}, fmt.Errorf("%w: %s", ErrNameTaken, name)
 		}
 		return Node{}, err
 	}
+	if since != nil {
+		n.AdoptedSince = *since
+	}
+	return n, nil
+}
+
+// nodeColumns and scanNode are the one reading of a node row, so every way of finding a node
+// says whether it is adopted.
+const nodeColumns = `id, name, created, last_seen, adopted, adopted_since`
+
+func scanNode(row pgx.Row) (Node, error) {
+	var n Node
+	var seen, since *time.Time
+	if err := row.Scan(&n.ID, &n.Name, &n.Created, &seen, &n.Adopted, &since); err != nil {
+		return Node{}, err
+	}
+	if seen != nil {
+		n.LastSeen = *seen
+	}
+	if since != nil {
+		n.AdoptedSince = *since
+	}
 	return n, nil
 }
 
 // Nodes are every node record, oldest first.
 func (i *Inventory) Nodes(ctx context.Context) ([]Node, error) {
 	rows, err := i.store.Pool().Query(ctx,
-		`select id, name, created, last_seen from node order by created, name`)
+		`select `+nodeColumns+` from node order by created, name`)
 	if err != nil {
 		return nil, err
 	}
@@ -103,14 +140,10 @@ func (i *Inventory) Nodes(ctx context.Context) ([]Node, error) {
 
 	var nodes []Node
 	for rows.Next() {
-		var n Node
-		var seen *time.Time
-		if err := rows.Scan(&n.ID, &n.Name, &n.Created, &seen); err != nil {
+		n, err := scanNode(rows)
+		if err != nil {
 			return nil, err
 		}
-		if seen != nil {
-			n.LastSeen = *seen
-		}
 		nodes = append(nodes, n)
 	}
 	return nodes, rows.Err()
@@ -118,9 +151,8 @@ func (i *Inventory) Nodes(ctx context.Context) ([]Node, error) {
 
 // NodeByName finds one node record.
 func (i *Inventory) NodeByName(ctx context.Context, name string) (Node, error) {
-	var n Node
-	err := i.store.Pool().QueryRow(ctx,
-		`select id, name, created from node where name = $1`, name).Scan(&n.ID, &n.Name, &n.Created)
+	n, err := scanNode(i.store.Pool().QueryRow(ctx,
+		`select `+nodeColumns+` from node where name = $1`, name))
 	if errors.Is(err, pgx.ErrNoRows) {
 		return Node{}, fmt.Errorf("%w: %s", ErrNoSuchNode, name)
 	}
@@ -248,10 +280,7 @@ func (i *Inventory) Claim(ctx context.Context, secret, by string, again bool) (N
 	if err != nil {
 		return Node{}, err
 	}
-	var n Node
-	err = i.store.Pool().QueryRow(ctx,
-		`select id, name, created from node where id = $1`, id).Scan(&n.ID, &n.Name, &n.Created)
-	return n, err
+	return scanNode(i.store.Pool().QueryRow(ctx, `select `+nodeColumns+` from node where id = $1`, id))
 }
 
 // Spend makes a claimed token used, only for the presenter holding the claim. The last write to the
@@ -293,10 +322,7 @@ func (i *Inventory) Redeem(ctx context.Context, secret string) (Node, error) {
 		return Node{}, err
 	}
 
-	var n Node
-	err = i.store.Pool().QueryRow(ctx,
-		`select id, name, created from node where id = $1`, id).Scan(&n.ID, &n.Name, &n.Created)
-	return n, err
+	return scanNode(i.store.Pool().QueryRow(ctx, `select `+nodeColumns+` from node where id = $1`, id))
 }
 
 // RecordProfile keeps the last thing a node said about what it can do.