Keep assignments in the assignments bucket and move the rows there at start (issue 1093)

Once chain_moved records the kind, Assign, Unassign and Assigned write and read the keys <node>.<module>, and every join of the store that read assignments is made from the bucket: where a module runs, what is behind, what a machine awaits, a given value due for replacement, taking a module, and whether a module may be forgotten (refused while a standing assignment names it). An unassignment ends the key with a marker and keeps its earlier values, as it removed the row; no ended or pending mark is written yet. The generation rises with each change, and a seat held by an assignment goes with it: the foreign key from seat_holding is dropped by migration 0104 and the controller says both. The provisioner's plan is kept in the values by a pass of the tick that writes that one field by compare-and-set; the assignment-changed event is said from what the bucket holds, whoever wrote it. The start moves the rows with what each machine was last sent, keeping a plan the mirror put in a value.

internal/broker/writers.go

@@ -193,8 +193,9 @@ var WritersTable = []WriterRow{
 	// The record mirrors (novox/hq ADR 0289 §9, design 54 §2): rows of this table, so a grant that lets
 	// anything else write them is refused at composition. Built so far: the assignments and each registered
 	// module's declared config (design 57 F3), and the list of predefined keys kept (design 57 K3).
-	{State: "the assignments mirror", Writer: "controller (lease holder), from its store, only on a change",
-		KeptIn: "key-value " + AssignmentsBucket, Others: "read through `reads` as mesh-controller." + MirrorAssignments,
+	{State: "the assignments mirror", Writer: "controller (lease holder): the verbs that assign, and the sends that record what runs; " +
+		"a mirror of its store only until the assignments are moved to the bus (novox/hq ADR 0408)",
+		KeptIn: "key-value " + AssignmentsBucket + ", the record once moved", Others: "read through `reads` as mesh-controller." + MirrorAssignments,
 		Subjects: kvOf(AssignmentsBucket), Writes: isController},
 	// The chain (novox/hq ADR 0408): builds and rollouts beside the assignments above, which keep their bucket. Not
 	// mirrors of the store but the record, written by the lease holder by compare-and-set and removed by nobody.

internal/chain/backend.go

@@ -11,7 +11,9 @@ import (
 // Entry is one key's value at a revision. Caught is set on the entry a watch sends once, after the values the bucket
 // held when it began, and carries no key.
 type Entry struct {
-	Key      string
+	Key string
+	// Deleted marks the end of a key (Remove); it holds no value.
+	Deleted  bool
 	Value    []byte
 	Revision uint64
 	At       time.Time
@@ -33,6 +35,10 @@ type Backend interface {
 	Watch(ctx context.Context) (<-chan Entry, error)
 	// History is a key's earlier values and its current one, oldest first, at most the bucket keeps.
 	History(ctx context.Context, key string) ([]Entry, error)
+	// Remove ends a key at the revision it was read at, with a marker: **its earlier values stay**, read back by History, and
+	// nothing is purged. Used by an unassignment and by nothing else, in the delivery that moves the assignments; the
+	// delivery after it keeps an unassigned key with an end mark and removes this (issue 1096).
+	Remove(ctx context.Context, key string, revision uint64) error
 }
 
 var (
@@ -95,17 +101,46 @@ func (m *Memory) now() time.Time {
 	return time.Now()
 }
 
+// last is a key's newest entry, false when it holds none or was removed.
+func (m *Memory) last(key string) (Entry, bool) {
+	h := m.values[key]
+	if len(h) == 0 || h[len(h)-1].Deleted {
+		return Entry{}, false
+	}
+	return h[len(h)-1], true
+}
+
 func (m *Memory) Get(_ context.Context, key string) (Entry, bool, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	if m.fail != nil {
 		return Entry{}, false, m.fail
 	}
-	h := m.values[key]
-	if len(h) == 0 {
+	e, ok := m.last(key)
+	if !ok {
 		return Entry{}, false, nil
 	}
-	return clone(h[len(h)-1]), true, nil
+	return clone(e), true, nil
+}
+
+func (m *Memory) Remove(_ context.Context, key string, revision uint64) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.fail != nil {
+		return m.fail
+	}
+	e, ok := m.last(key)
+	if !ok || e.Revision != revision {
+		return ErrMoved
+	}
+	m.revision++
+	m.writes++
+	h := append(m.values[key], Entry{Key: key, Deleted: true, Revision: m.revision, At: m.now()})
+	if len(h) > m.keep() {
+		h = h[len(h)-m.keep():]
+	}
+	m.values[key] = h
+	return nil
 }
 
 func (m *Memory) Create(_ context.Context, key string, value []byte) (uint64, error) {
@@ -114,7 +149,7 @@ func (m *Memory) Create(_ context.Context, key string, value []byte) (uint64, er
 	if m.fail != nil {
 		return 0, m.fail
 	}
-	if len(m.values[key]) > 0 {
+	if _, held := m.last(key); held {
 		return 0, ErrExists
 	}
 	return m.write(key, value), nil
@@ -126,8 +161,8 @@ func (m *Memory) Update(_ context.Context, key string, value []byte, revision ui
 	if m.fail != nil {
 		return 0, m.fail
 	}
-	h := m.values[key]
-	if len(h) == 0 || h[len(h)-1].Revision != revision {
+	e, held := m.last(key)
+	if !held || e.Revision != revision {
 		return 0, ErrMoved
 	}
 	return m.write(key, value), nil
@@ -157,7 +192,9 @@ func (m *Memory) Keys(_ context.Context) ([]string, error) {
 	}
 	out := make([]string, 0, len(m.values))
 	for k := range m.values {
-		out = append(out, k)
+		if _, held := m.last(k); held {
+			out = append(out, k)
+		}
 	}
 	sort.Strings(out)
 	return out, nil
@@ -204,8 +241,9 @@ func (m *Memory) Watch(ctx context.Context) (<-chan Entry, error) {
 	}
 	sort.Strings(keys)
 	for _, k := range keys {
-		h := m.values[k]
-		w.q = append(w.q, clone(h[len(h)-1]))
+		if e, held := m.last(k); held {
+			w.q = append(w.q, clone(e))
+		}
 	}
 	w.q = append(w.q, Entry{Caught: true})
 	m.watchers = append(m.watchers, w)
@@ -250,7 +288,9 @@ func (m *Memory) History(_ context.Context, key string) ([]Entry, error) {
 	}
 	out := make([]Entry, 0, len(m.values[key]))
 	for _, e := range m.values[key] {
-		out = append(out, clone(e))
+		if !e.Deleted {
+			out = append(out, clone(e))
+		}
 	}
 	return out, nil
 }

internal/chain/bus.go

@@ -76,6 +76,16 @@ func (b busBucket) Update(ctx context.Context, key string, value []byte, revisio
 	return rev, err
 }
 
+func (b busBucket) Remove(ctx context.Context, key string, revision uint64) error {
+	// A delete marker, never a purge: the key's earlier values stay in the bucket's history.
+	err := b.kv.Delete(ctx, key, jetstream.LastRevision(revision))
+	var apiErr *jetstream.APIError
+	if errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence {
+		return ErrMoved
+	}
+	return err
+}
+
 func (b busBucket) Keys(ctx context.Context) ([]string, error) {
 	keys, err := b.kv.Keys(ctx)
 	if errors.Is(err, jetstream.ErrNoKeysFound) {
@@ -126,6 +136,9 @@ func (b busBucket) History(ctx context.Context, key string) ([]Entry, error) {
 	}
 	out := make([]Entry, 0, len(all))
 	for _, e := range all {
+		if e.Operation() != jetstream.KeyValuePut {
+			continue
+		}
 		out = append(out, entryOf(e))
 	}
 	return out, nil

internal/chain/chain.go

@@ -182,8 +182,20 @@ type Chain struct {
 	Rollouts    Bucket[Rollout]
 	Assignments Bucket[Assignment]
 
-	rollouts Backend
-	holds    Holds
+	rollouts    Backend
+	assignments Backend
+	holds       Holds
+}
+
+// RemoveAssignment ends an assignment's key with a marker, at the revision it was read at, only by a controller that holds
+// the lease. **The one removal the chain has**, for an unassignment in the delivery that moves the assignments: the
+// key's earlier values stay (history), and nothing is purged. The delivery after it keeps an unassigned key with an end
+// mark in its place and removes this (issue 1096). Builds and rollouts have none.
+func (c *Chain) RemoveAssignment(ctx context.Context, key string, revision uint64) error {
+	if err := c.Assignments.guard(ctx, key, nil); err != nil {
+		return err
+	}
+	return c.assignments.Remove(ctx, key, revision)
 }
 
 // RolloutsOf is the rollouts bucket typed as the controller's plan, which carries more than the chain's Rollout names:
@@ -197,6 +209,7 @@ func New(builds, rollouts, assignments Backend, holds Holds) *Chain {
 		Rollouts:    Bucket[Rollout]{rollouts, holds},
 		Assignments: Bucket[Assignment]{assignments, holds},
 		rollouts:    rollouts,
+		assignments: assignments,
 		holds:       holds,
 	}
 }

internal/chain/chain_test.go

@@ -334,3 +334,52 @@ func TestAnEmptyBucketIsReadAsEmpty(t *testing.T) {
 		t.Fatalf("%v %v", all, err)
 	}
 }
+
+// **The one removal the chain has ends an assignment's key with a marker and keeps its earlier values** (issue 1093,
+// ADR 0408 §5): the key reads as absent, can be made again, builds and rollouts have no removal, and a controller without
+// the lease removes nothing.
+func TestAnAssignmentsKeyEndsWithAMarkerAndKeepsItsHistory(t *testing.T) {
+	ctx := context.Background()
+	var away error
+	c := New(NewMemory(), NewMemory(), NewMemory(), func(context.Context) error { return away })
+	rev, err := c.Assignments.Create(ctx, "anchor.notes", Assignment{Node: "anchor", Module: "notes", Commit: "c1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	away = errors.New("another controller holds the lease")
+	if err := c.RemoveAssignment(ctx, "anchor.notes", rev); !errors.Is(err, ErrNotHeld) {
+		t.Fatalf("a removal without the lease answered %v", err)
+	}
+	away = nil
+	if err := c.RemoveAssignment(ctx, "anchor.notes", rev+1); !errors.Is(err, ErrMoved) {
+		t.Fatalf("a removal at a revision that moved answered %v", err)
+	}
+	if err := c.RemoveAssignment(ctx, "anchor.notes", rev); err != nil {
+		t.Fatal(err)
+	}
+	if _, ok, _ := c.Assignments.Get(ctx, "anchor.notes"); ok {
+		t.Fatal("a removed key reads as present")
+	}
+	if keys, _ := c.Assignments.Keys(ctx); len(keys) != 0 {
+		t.Fatalf("keys %v", keys)
+	}
+	if all, _ := c.Assignments.All(ctx); len(all) != 0 {
+		t.Fatalf("all %v", all)
+	}
+	if history, _ := c.Assignments.History(ctx, "anchor.notes"); len(history) != 1 || history[0].Value.Commit != "c1" {
+		t.Fatalf("the earlier values are %+v", history)
+	}
+	if _, err := c.Assignments.Create(ctx, "anchor.notes", Assignment{Node: "anchor", Module: "notes"}); err != nil {
+		t.Fatalf("a removed key could not be made again: %v", err)
+	}
+	if _, ok := any(c.Builds).(interface {
+		Remove(context.Context, string, uint64) error
+	}); ok {
+		t.Fatal("builds can be removed")
+	}
+	if _, ok := any(c.Rollouts).(interface {
+		Remove(context.Context, string, uint64) error
+	}); ok {
+		t.Fatal("rollouts can be removed")
+	}
+}

internal/inventory/adoption.go

@@ -66,16 +66,14 @@ func (i *Inventory) Take(ctx context.Context, nodeName, module string) error {
 // 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 {
+	assigned, err := i.isAssigned(ctx, node.Name, module)
+	if 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,
+	_, err = i.store.Pool().Exec(ctx,
 		`insert into taken (node, module) values ($1, $2) on conflict do nothing`, node.ID, module)
 	return err
 }
@@ -95,27 +93,24 @@ func (i *Inventory) Converge(ctx context.Context, nodeName string) ([]string, er
 		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)
+	assigned, err := i.assignedModules(ctx, node.Name)
 	if err != nil {
 		return nil, err
 	}
 	var took []string
-	for rows.Next() {
+	for _, module := range assigned {
 		var m string
-		if err := rows.Scan(&m); err != nil {
-			rows.Close()
+		err := tx.QueryRow(ctx,
+			`insert into taken (node, module) values ($1, $2) on conflict do nothing returning module`,
+			node.ID, module).Scan(&m)
+		if errors.Is(err, pgx.ErrNoRows) {
+			continue // taken already
+		}
+		if err != nil {
 			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(),
 		        held = null, reachable = null, firewall = null, adoption_reported = null

internal/inventory/catalogue.go

@@ -258,15 +258,31 @@ func (i *Inventory) SourceOf(ctx context.Context, module string) (Source, error)
 // machines are running last week's version" is the question somebody actually has, and it is the
 // one novox/hq ADR 0010 names as the thing that must not be lost.
 func (i *Inventory) Behind(ctx context.Context) (map[string][]string, error) {
-	rows, err := i.store.Pool().Query(ctx,
-		`select m.name, coalesce(n.name, '')
+	query := `select m.name, coalesce(n.name, '')
 		   from module m
 		   left join assignment a on a.module = m.name
 		   left join node n on n.id = a.node
 		  where m.source is not null
 		    and m.source_head is not null
 		    and coalesce(m.built_from, '') is distinct from m.source_head
-		  order by m.name, n.name`)
+		  order by m.name, n.name`
+	_, onTheBus, err := i.ChainReads(ctx, KindAssignments)
+	if err != nil {
+		return nil, err
+	}
+	var assigned []assignRow
+	if onTheBus {
+		query = `select m.name, ''
+		   from module m
+		  where m.source is not null
+		    and m.source_head is not null
+		    and coalesce(m.built_from, '') is distinct from m.source_head
+		  order by m.name`
+		if assigned, err = i.assignments(ctx); err != nil {
+			return nil, err
+		}
+	}
+	rows, err := i.store.Pool().Query(ctx, query)
 	if err != nil {
 		return nil, err
 	}
@@ -285,7 +301,18 @@ func (i *Inventory) Behind(ctx context.Context) (map[string][]string, error) {
 			out[module] = append(out[module], node)
 		}
 	}
-	return out, rows.Err()
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	for _, a := range assigned {
+		if _, behind := out[a.Module]; behind {
+			out[a.Module] = append(out[a.Module], a.Node)
+		}
+	}
+	for module := range out {
+		sort.Strings(out[module])
+	}
+	return out, nil
 }
 
 // Catalogue is every module the mesh knows about, which is what resolution needs: the question
@@ -401,22 +428,7 @@ func (i *Inventory) RetireUnshipped(ctx context.Context, shipped []string) (reti
 
 // assignedOn is the machines a module is assigned to, sorted.
 func (i *Inventory) assignedOn(ctx context.Context, name string) ([]string, error) {
-	rows, err := i.store.Pool().Query(ctx,
-		`select n.name from assignment a join node n on n.id = a.node where a.module = $1
-		 order by n.name`, name)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-	var on []string
-	for rows.Next() {
-		var node string
-		if err := rows.Scan(&node); err != nil {
-			return nil, err
-		}
-		on = append(on, node)
-	}
-	return on, rows.Err()
+	return i.assignedNodes(ctx, name)
 }
 
 // Provided reports whether a module came with the control plane rather than from a repository.
@@ -600,25 +612,10 @@ func (i *Inventory) mayForget(ctx context.Context, name string) error {
 			"%s comes with the control plane and cannot be forgotten; unassign it instead", name)
 	}
 
-	var on []string
-	rows, err := i.store.Pool().Query(ctx,
-		`select n.name from assignment a join node n on n.id = a.node where a.module = $1
-		 order by n.name`, name)
+	on, err := i.assignedNodes(ctx, name)
 	if err != nil {
 		return err
 	}
-	for rows.Next() {
-		var node string
-		if err := rows.Scan(&node); err != nil {
-			rows.Close()
-			return err
-		}
-		on = append(on, node)
-	}
-	rows.Close()
-	if err := rows.Err(); err != nil {
-		return err
-	}
 	if len(on) > 0 {
 		return fmt.Errorf("%w: %s. Unassign it first", ErrStillAssigned, strings.Join(on, ", "))
 	}
@@ -657,6 +654,11 @@ func (i *Inventory) Assign(ctx context.Context, nodeName, module string) (bool,
 	if err := i.runsSomewhere(ctx, module); err != nil {
 		return false, err
 	}
+	if c, on, err := i.ChainReads(ctx, KindAssignments); err != nil {
+		return false, err
+	} else if on {
+		return i.chainAssign(ctx, c, node, module)
+	}
 	tag, err := i.store.Pool().Exec(ctx,
 		`insert into assignment (node, module) values ($1, $2) on conflict do nothing`,
 		node.ID, module)
@@ -675,13 +677,25 @@ func (i *Inventory) Unassign(ctx context.Context, nodeName, module string) error
 	if err != nil {
 		return err
 	}
-	tag, err := i.store.Pool().Exec(ctx,
-		`delete from assignment where node = $1 and module = $2`, node.ID, module)
-	if err != nil {
+	if c, on, err := i.ChainReads(ctx, KindAssignments); err != nil {
 		return err
+	} else if on {
+		if err := i.chainUnassign(ctx, c, node, module); err != nil {
+			return err
+		}
+	} else {
+		tag, err := i.store.Pool().Exec(ctx,
+			`delete from assignment where node = $1 and module = $2`, node.ID, module)
+		if err != nil {
+			return err
+		}
+		if tag.RowsAffected() == 0 {
+			return fmt.Errorf("%s is not assigned to %s", module, nodeName)
+		}
 	}
-	if tag.RowsAffected() == 0 {
-		return fmt.Errorf("%s is not assigned to %s", module, nodeName)
+	// The seat it held falls back to derivation rather than pointing at nothing: what the store's foreign key did.
+	if _, err := i.store.Pool().Exec(ctx, `delete from seat_holding where node = $1 and module = $2`, node.ID, module); err != nil {
+		return err
 	}
 	// And its ports go back. Kept-once-chosen is a promise about a module that is still here —
 	// held past unassignment, a fixed port stays claimed in the name of something that is gone,
@@ -697,22 +711,7 @@ func (i *Inventory) Assigned(ctx context.Context, nodeName string) ([]string, er
 	if err != nil {
 		return nil, err
 	}
-	rows, err := i.store.Pool().Query(ctx,
-		`select module from assignment 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()
+	return i.assignedModules(ctx, node.Name)
 }
 
 // ProfileOf is what a node last said it can do, as resolution needs it: the capabilities that are
@@ -1363,8 +1362,7 @@ type Entry struct {
 // catalogue nobody lists, and the questions here — what is this, where did it come from, who is
 // running it — are asked together every time.
 func (i *Inventory) Catalogued(ctx context.Context) ([]Entry, error) {
-	rows, err := i.store.Pool().Query(ctx,
-		`select m.name, m.manifest,
+	query := `select m.name, m.manifest,
 		        coalesce(m.source, ''), m.source_path, m.source_seat, coalesce(m.ref, ''),
 		        coalesce(m.built_from, ''), coalesce(m.source_head, ''),
 		        coalesce(m.source_seen, to_timestamp(0)),
@@ -1374,7 +1372,30 @@ func (i *Inventory) Catalogued(ctx context.Context) ([]Entry, error) {
 		 left join node n on n.id = a.node
 		 group by m.name, m.manifest, m.source, m.source_path, m.source_seat, m.ref, m.built_from,
 		          m.source_head, m.source_seen
-		 order by m.name`)
+		 order by m.name`
+	// On the bus the assignments are read from the bucket, and the machines of a module put beside it here.
+	_, onTheBus, err := i.ChainReads(ctx, KindAssignments)
+	if err != nil {
+		return nil, err
+	}
+	var assignedBy map[string][]string
+	if onTheBus {
+		query = `select m.name, m.manifest,
+		        coalesce(m.source, ''), m.source_path, m.source_seat, coalesce(m.ref, ''),
+		        coalesce(m.built_from, ''), coalesce(m.source_head, ''),
+		        coalesce(m.source_seen, to_timestamp(0)), '{}'::text[]
+		 from module m
+		 order by m.name`
+		rows, err := i.assignments(ctx)
+		if err != nil {
+			return nil, err
+		}
+		assignedBy = map[string][]string{}
+		for _, a := range rows {
+			assignedBy[a.Module] = append(assignedBy[a.Module], a.Node)
+		}
+	}
+	rows, err := i.store.Pool().Query(ctx, query)
 	if err != nil {
 		return nil, err
 	}
@@ -1393,6 +1414,9 @@ func (i *Inventory) Catalogued(ctx context.Context) ([]Entry, error) {
 		if source.Seen.Unix() == 0 {
 			source.Seen = time.Time{}
 		}
+		if onTheBus {
+			on = assignedBy[name]
+		}
 		var m catalogue.Manifest
 		if err := json.Unmarshal(raw, &m); err != nil {
 			return nil, err
@@ -1612,23 +1636,7 @@ func (i *Inventory) CurrentBuilds(ctx context.Context) (map[string]CurrentBuild,
 // exactly the machine an upgrade most needs to reach; waiting for it to report the old version
 // first would mean the machines furthest behind are the last to be caught up.
 func (i *Inventory) Running(ctx context.Context, module string) ([]string, error) {
-	rows, err := i.store.Pool().Query(ctx,
-		`select n.name from assignment a join node n on n.id = a.node
-          where a.module = $1 order by n.name`, module)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	var out []string
-	for rows.Next() {
-		var name string
-		if err := rows.Scan(&name); err != nil {
-			return nil, err
-		}
-		out = append(out, name)
-	}
-	return out, rows.Err()
+	return i.assignedNodes(ctx, module)
 }
 
 // runsSomewhere refuses to put a module on a machine when it would put nothing there.

internal/inventory/catalogue_test.go

@@ -113,13 +113,15 @@ func TestRemovingANodeTakesItsAssignments(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	var left int
-	if err := inv.store.Pool().QueryRow(t.Context(),
-		`select count(*) from assignment`).Scan(&left); err != nil {
+	// On the bus the key outlives the node it names, and is read as no assignment: the machine is not one the store knows.
+	entries, err := inv.Catalogued(t.Context())
+	if err != nil {
 		t.Fatal(err)
 	}
-	if left != 0 {
-		t.Errorf("%d assignment(s) outlived the node they were on", left)
+	for _, e := range entries {
+		if e.Manifest.Module == "thing" && len(e.On) != 0 {
+			t.Errorf("thing is assigned to %v, a machine that is gone", e.On)
+		}
 	}
 	// And the module itself survives, because other nodes may be running it.
 	shelf, err := inv.Catalogue(t.Context())

internal/inventory/chain_assignments.go

@@ -0,0 +1,354 @@
+package inventory
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/jackc/pgx/v5"
+
+	"github.com/novox/mesh-controller/internal/chain"
+)
+
+// Assignments on the bus (novox/hq ADR 0408, issue 1093): once the kind `assignments` is moved (chain.go) an assignment is
+// a key `<node>.<module>` of the assignments bucket, written by compare-and-set by the controller that holds the lease,
+// and every question the joins of the store answered is answered from the bucket. **Live values only** in this delivery: an
+// unassignment ends the key with a marker, as it removed the row, and no `ended` or `pending` mark is written; the
+// delivery after it keeps an unassigned key (issue 1096).
+
+// KindAssignments is the kind of the chain that is each module assigned to a node, with the build it runs there.
+const KindAssignments = "assignments"
+
+func assignmentKey(node, module string) string { return node + "." + module }
+
+// assignRow is one assignment as the joins of the store read it.
+type assignRow struct {
+	Node, Module string
+	Assigned     time.Time
+}
+
+// assignments is every assignment, by node then module: from the bucket when the kind is moved, else from the store.
+// A key whose value names no time assigned is a module only pulled in by another (issue 765): no assignment. One on a
+// machine the store does not know is no assignment either, as the store took a removed machine's.
+func (i *Inventory) assignments(ctx context.Context) ([]assignRow, error) {
+	c, on, err := i.ChainReads(ctx, KindAssignments)
+	if err != nil {
+		return nil, err
+	}
+	if !on {
+		rows, err := i.store.Pool().Query(ctx,
+			`select n.name, a.module, a.assigned from assignment a join node n on n.id = a.node order by n.name, a.module`)
+		if err != nil {
+			return nil, err
+		}
+		defer rows.Close()
+		var out []assignRow
+		for rows.Next() {
+			var a assignRow
+			if err := rows.Scan(&a.Node, &a.Module, &a.Assigned); err != nil {
+				return nil, err
+			}
+			out = append(out, a)
+		}
+		return out, rows.Err()
+	}
+	all, err := c.Assignments.All(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("the assignments cannot be read from the bus: %w", err)
+	}
+	nodes, err := i.nodeNames(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var out []assignRow
+	for _, k := range all {
+		if k.Value.Assigned.IsZero() || !nodes[k.Value.Node] {
+			continue
+		}
+		out = append(out, assignRow{k.Value.Node, k.Value.Module, k.Value.Assigned})
+	}
+	sort.Slice(out, func(a, b int) bool {
+		if out[a].Node != out[b].Node {
+			return out[a].Node < out[b].Node
+		}
+		return out[a].Module < out[b].Module
+	})
+	return out, nil
+}
+
+func (i *Inventory) nodeNames(ctx context.Context) (map[string]bool, error) {
+	rows, err := i.store.Pool().Query(ctx, `select name from node`)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	out := map[string]bool{}
+	for rows.Next() {
+		var name string
+		if err := rows.Scan(&name); err != nil {
+			return nil, err
+		}
+		out[name] = true
+	}
+	return out, rows.Err()
+}
+
+// assignedNodes is the machines a module is assigned to, sorted.
+func (i *Inventory) assignedNodes(ctx context.Context, module string) ([]string, error) {
+	rows, err := i.assignments(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var out []string
+	for _, a := range rows {
+		if a.Module == module {
+			out = append(out, a.Node)
+		}
+	}
+	return out, nil
+}
+
+// assignedModules is the modules assigned to a machine, sorted.
+func (i *Inventory) assignedModules(ctx context.Context, node string) ([]string, error) {
+	rows, err := i.assignments(ctx)
+	if err != nil {
+		return nil, err
+	}
+	var out []string
+	for _, a := range rows {
+		if a.Node == node {
+			out = append(out, a.Module)
+		}
+	}
+	return out, nil
+}
+
+// isAssigned says a module is assigned to a machine.
+func (i *Inventory) isAssigned(ctx context.Context, node, module string) (bool, error) {
+	rows, err := i.assignments(ctx)
+	if err != nil {
+		return false, err
+	}
+	for _, a := range rows {
+		if a.Node == node && a.Module == module {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// raiseGeneration raises the assignment generation, as the store's triggers do for a change of its table (migration
+// 0094): in the chain nothing in the store changes with an assignment, so the writer says it.
+func (i *Inventory) raiseGeneration(ctx context.Context) error {
+	_, err := i.store.Pool().Exec(ctx, `update assignment_generation set generation = generation + 1`)
+	return err
+}
+
+// chainAssign puts a module on a machine, and says whether that is new.
+func (i *Inventory) chainAssign(ctx context.Context, c *chain.Chain, node Node, module string) (bool, error) {
+	var known bool
+	if err := i.store.Pool().QueryRow(ctx, `select exists (select 1 from module where name = $1)`, module).Scan(&known); err != nil {
+		return false, err
+	}
+	if !known {
+		return false, fmt.Errorf("%w: %s", ErrNoSuchModule, module)
+	}
+	key := assignmentKey(node.Name, module)
+	for try := 0; try < casTries; try++ {
+		cur, found, err := c.Assignments.Get(ctx, key)
+		if err != nil {
+			return false, err
+		}
+		if found && !cur.Value.Assigned.IsZero() {
+			return false, nil // one assignment of a module per machine: a repeat is absorbed (novox/hq ADR 0115)
+		}
+		now := time.Now().UTC()
+		if found {
+			// A module only pulled in by another becomes an assignment.
+			cur.Value.Assigned = now
+			_, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision)
+		} else {
+			_, err = c.Assignments.Create(ctx, key, chain.Assignment{Node: node.Name, Module: module, Assigned: now})
+		}
+		if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) {
+			continue
+		}
+		if err != nil {
+			return false, fmt.Errorf("%s is not assigned to %s on the bus: %w", module, node.Name, err)
+		}
+		if err := i.raiseGeneration(ctx); err != nil {
+			return true, err
+		}
+		return true, nil
+	}
+	return false, fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name)
+}
+
+// chainUnassign takes a module off a machine: the key ends with a marker, its earlier values stay.
+func (i *Inventory) chainUnassign(ctx context.Context, c *chain.Chain, node Node, module string) error {
+	key := assignmentKey(node.Name, module)
+	for try := 0; try < casTries; try++ {
+		cur, found, err := c.Assignments.Get(ctx, key)
+		if err != nil {
+			return err
+		}
+		if !found || cur.Value.Assigned.IsZero() {
+			return fmt.Errorf("%s is not assigned to %s", module, node.Name)
+		}
+		if err := c.RemoveAssignment(ctx, key, cur.Revision); errors.Is(err, chain.ErrMoved) {
+			continue
+		} else if err != nil {
+			return fmt.Errorf("%s is not taken off %s on the bus: %w", module, node.Name, err)
+		}
+		if err := i.raiseGeneration(ctx); err != nil {
+			return err
+		}
+		return nil
+	}
+	return fmt.Errorf("the assignment of %s to %s kept moving under the write", module, node.Name)
+}
+
+// chainAwaitingSince is AwaitingSince over the bucket: the oldest assignment on the machine made after it was last sent.
+func (i *Inventory) chainAwaitingSince(ctx context.Context, name string) (time.Time, error) {
+	var sentAt *time.Time
+	var created time.Time
+	err := i.store.Pool().QueryRow(ctx, `select sent_at, created from node where name = $1`, name).Scan(&sentAt, &created)
+	if errors.Is(err, pgx.ErrNoRows) {
+		return time.Time{}, fmt.Errorf("%w: %s", ErrNoSuchNode, name)
+	}
+	if err != nil {
+		return time.Time{}, err
+	}
+	rows, err := i.assignments(ctx)
+	if err != nil {
+		return time.Time{}, err
+	}
+	var oldest time.Time
+	for _, a := range rows {
+		if a.Node == name && (sentAt == nil || a.Assigned.After(*sentAt)) && (oldest.IsZero() || a.Assigned.Before(oldest)) {
+			oldest = a.Assigned
+		}
+	}
+	switch {
+	case !oldest.IsZero():
+		return oldest, nil
+	case sentAt != nil:
+		return *sentAt, nil
+	}
+	return created, nil
+}
+
+// KeepAssignmentPlans writes the provisioner's plan of each assignment into its value on the bus, and a value of its own for
+// each module only pulled in by another on a machine (issue 765), and ends the value of one that is no longer pulled in. The
+// plans are derived from what each machine was last sent, not part of the record of what is assigned, so they are read
+// again on a tick and written where they differ: **by compare-and-set on the one field**, never laying over the build, commit
+// and rollout another write put in the same value, and never ending an assignment. A plan the provisioner's rules refuse is
+// refused by name and nothing is written (AssignmentValues).
+func (i *Inventory) KeepAssignmentPlans(ctx context.Context) (written, removed int, err error) {
+	c, on, err := i.ChainReads(ctx, KindAssignments)
+	if err != nil || !on {
+		return 0, 0, err
+	}
+	var plans map[string]Provisioning
+	if i.provisioningPlans != nil {
+		if plans, err = i.provisioningPlans(ctx); err != nil {
+			return 0, 0, fmt.Errorf("composing the provisioner's plans: %w", err)
+		}
+	}
+	assigned, err := i.assignments(ctx)
+	if err != nil {
+		return 0, 0, err
+	}
+	rows := make([]AssignmentMirrored, 0, len(assigned))
+	for _, a := range assigned {
+		rows = append(rows, AssignmentMirrored{Node: a.Node, Module: a.Module, Assigned: a.Assigned.UTC()})
+	}
+	want, err := AssignmentValues(rows, plans)
+	if err != nil {
+		return 0, 0, err
+	}
+	have, err := c.Assignments.All(ctx)
+	if err != nil {
+		return 0, 0, fmt.Errorf("the assignments cannot be read from the bus: %w", err)
+	}
+	for key, body := range want {
+		var next chain.Assignment
+		if err := json.Unmarshal(body, &next); err != nil {
+			return written, removed, err
+		}
+		cur, has := have[key]
+		if has && sameJSON(cur.Value.Provisioning, next.Provisioning) && cur.Value.RequiredBy == next.RequiredBy {
+			continue
+		}
+		if !has {
+			next.Assigned = time.Time{}
+			_, err = c.Assignments.Create(ctx, key, next)
+		} else {
+			cur.Value.RequiredBy, cur.Value.Provisioning = next.RequiredBy, next.Provisioning
+			_, err = c.Assignments.Update(ctx, key, cur.Value, cur.Revision)
+		}
+		if errors.Is(err, chain.ErrMoved) || errors.Is(err, chain.ErrExists) {
+			continue // written meanwhile; the next pass reads it
+		}
+		if err != nil {
+			return written, removed, fmt.Errorf("the plan of %s is not written: %w", key, err)
+		}
+		written++
+	}
+	for key, cur := range have {
+		if _, wanted := want[key]; wanted || !cur.Value.Assigned.IsZero() {
+			continue
+		}
+		// A value of a module no longer pulled in, which no assignment names.
+		if err := c.RemoveAssignment(ctx, key, cur.Revision); errors.Is(err, chain.ErrMoved) {
+			continue
+		} else if err != nil {
+			return written, removed, fmt.Errorf("the plan of %s is not ended: %w", key, err)
+		}
+		removed++
+	}
+	return written, removed, nil
+}
+
+// sameJSON says two encoded values are the same bytes, an absent value being a null one.
+func sameJSON(a, b json.RawMessage) bool {
+	trim := func(x json.RawMessage) string {
+		if s := strings.TrimSpace(string(x)); s != "null" {
+			return s
+		}
+		return ""
+	}
+	return trim(a) == trim(b)
+}
+
+// KeptValue is an assignment's value as the bucket holds it, encoded, at its revision.
+type KeptValue struct {
+	Value    []byte
+	Revision uint64
+}
+
+// KeptAssignments is every key of the assignments bucket with its value, and false when the assignments are not on the bus.
+// What the serving controller compares from one look to the next to say a key changed (`assignment-changed`).
+func (i *Inventory) KeptAssignments(ctx context.Context) (map[string]KeptValue, bool, error) {
+	c, on, err := i.ChainReads(ctx, KindAssignments)
+	if err != nil || !on {
+		return nil, false, err
+	}
+	all, err := c.Assignments.All(ctx)
+	if err != nil {
+		return nil, true, err
+	}
+	out := make(map[string]KeptValue, len(all))
+	for key, k := range all {
+		body, err := json.Marshal(k.Value)
+		if err != nil {
+			return nil, true, err
+		}
+		out[key] = KeptValue{Value: body, Revision: k.Revision}
+	}
+	return out, true, nil
+}

internal/inventory/chain_assignments_key_test.go

@@ -0,0 +1,37 @@
+package inventory
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+// **An assignment's key is `<node>.<module>`, as the bucket has always had it** (ADR 0408 §3): its readers parse nothing
+// but it.
+func TestAnAssignmentsKeyIsTheNodeAndTheModule(t *testing.T) {
+	if got := assignmentKey("anchor", "notes"); got != "anchor.notes" {
+		t.Fatalf("the key is %q", got)
+	}
+	if got := assignmentKey("g14.example", "dunst"); got != "g14.example.dunst" {
+		t.Fatalf("the key of a machine whose name holds a dot is %q", got)
+	}
+}
+
+// **A plan that is absent and a null one are one** (the provisioner's plan in an assignment's value): the pass that keeps
+// the plans writes only where they differ.
+func TestAnAbsentPlanAndANullOneAreTheSame(t *testing.T) {
+	for name, tc := range map[string]struct {
+		a, b json.RawMessage
+		same bool
+	}{
+		"both absent":     {nil, nil, true},
+		"null and absent": {json.RawMessage(`null`), nil, true},
+		"the same plan":   {json.RawMessage(`{"files":[]}`), json.RawMessage(` {"files":[]} `), true},
+		"another plan":    {json.RawMessage(`{"files":[]}`), json.RawMessage(`{"files":[1]}`), false},
+		"a plan and none": {json.RawMessage(`{"files":[]}`), nil, false},
+		"none and a plan": {nil, json.RawMessage(`{"files":[]}`), false},
+	} {
+		if got := sameJSON(tc.a, tc.b); got != tc.same {
+			t.Errorf("%s: %v", name, got)
+		}
+	}
+}

internal/inventory/chain_move.go

@@ -160,9 +160,117 @@ func (i *Inventory) movers() []mover {
 			return n, err
 		},
 		lock: `lock table release_plan, batched_merge in share mode`,
+	}, {
+		kind: KindAssignments,
+		rows: func(ctx context.Context) ([]moveRow, int, error) {
+			assigned, err := i.assignedWithSends(ctx)
+			if err != nil {
+				return nil, 0, err
+			}
+			out := make([]moveRow, 0, len(assigned))
+			for _, a := range assigned {
+				body, err := json.Marshal(a)
+				if err != nil {
+					return nil, 0, err
+				}
+				out = append(out, moveRow{key: assignmentKey(a.Node, a.Module), value: a, size: len(body)})
+			}
+			return out, len(out), nil
+		},
+		create: func(ctx context.Context, c *chain.Chain, row moveRow) error {
+			want := row.value.(chain.Assignment)
+			// The mirror of the store kept a value under this key until now, with the provisioner's plan beside it: the
+			// plan stays, the record's links are laid in. A value that names its build already is a later word, and stays.
+			for try := 0; try < casTries; try++ {
+				cur, found, err := c.Assignments.Get(ctx, row.key)
+				if err != nil {
+					return err
+				}
+				if !found {
+					if _, err := c.Assignments.Create(ctx, row.key, want); !errors.Is(err, chain.ErrExists) {
+						return err
+					}
+					continue
+				}
+				if cur.Value.Build != "" || cur.Value.Commit != "" || !cur.Value.Since.IsZero() || (want.Build == "" && want.Commit == "") {
+					return chain.ErrExists
+				}
+				merged := want
+				merged.RequiredBy, merged.Provisioning = cur.Value.RequiredBy, cur.Value.Provisioning
+				if _, err := c.Assignments.Update(ctx, row.key, merged, cur.Revision); errors.Is(err, chain.ErrMoved) {
+					continue
+				} else {
+					return err
+				}
+			}
+			return fmt.Errorf("the assignment %s kept moving under the write", row.key)
+		},
+		keys: func(ctx context.Context, c *chain.Chain) ([]string, error) { return c.Assignments.Keys(ctx) },
+		count: func(ctx context.Context, tx pgx.Tx) (int, error) {
+			var n int
+			err := tx.QueryRow(ctx, `select count(*) from assignment`).Scan(&n)
+			return n, err
+		},
+		lock: `lock table assignment in share mode`,
 	}}
 }
 
+// assignedWithSends is every assignment of the store as the chain keeps it: with the commit its machine was last sent for
+// the module (node.sent_builds), the build that is of it and what the build published, and since when (when the machine was
+// last sent). An assignment of a module the machine was never sent has none of those.
+func (i *Inventory) assignedWithSends(ctx context.Context) ([]chain.Assignment, error) {
+	rows, err := i.store.Pool().Query(ctx,
+		`select n.name, a.module, a.assigned, n.sent_at, n.sent_builds
+		   from assignment a join node n on n.id = a.node order by n.name, a.module`)
+	if err != nil {
+		return nil, err
+	}
+	var out []chain.Assignment
+	for rows.Next() {
+		var a chain.Assignment
+		var sentAt *time.Time
+		var raw []byte
+		if err := rows.Scan(&a.Node, &a.Module, &a.Assigned, &sentAt, &raw); err != nil {
+			rows.Close()
+			return nil, err
+		}
+		a.Assigned = a.Assigned.UTC()
+		var sent map[string]string
+		if len(raw) > 0 {
+			_ = json.Unmarshal(raw, &sent)
+		}
+		if commit := sent[a.Module]; commit != "" && sentAt != nil {
+			a.Commit, a.Since = commit, sentAt.UTC()
+		}
+		out = append(out, a)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return nil, err
+	}
+	for k := range out {
+		if out[k].Commit == "" {
+			continue
+		}
+		id, err := i.BuildOf(ctx, out[k].Module, out[k].Commit)
+		if err != nil {
+			return nil, err
+		}
+		if id == "" {
+			continue
+		}
+		out[k].Build = id
+		if b, found, err := i.BuildByID(ctx, id); err != nil {
+			return nil, err
+		} else if found {
+			if published := publishedOf(b.Made); published != nil {
+				out[k].Artifact, out[k].Package = published.Artifact, published.Package
+			}
+		}
+	}
+	return out, nil
+}
+
 // holdMerges puts each batched merge into the record of the plan that answers it. A merge no plan answers — the plan_id
 // of a merge a failed walk carried is null, or names a plan the store never kept — is held, flagged waiting, by the
 // newest plan that lists it among its merges, else the record kept for the merges no plan answers: the record that held it

internal/inventory/chain_move_test.go

@@ -4,12 +4,15 @@ package inventory
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"reflect"
 	"slices"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/novox/mesh-controller/internal/catalogue"
 	"github.com/novox/mesh-controller/internal/chain"
 )
 
@@ -523,3 +526,234 @@ func TestTheMoveCarriesThePlansAndTheirMergesAndTheyReadTheSame(t *testing.T) {
 		t.Fatalf("the second move %+v: %v", again, err)
 	}
 }
+
+// **An assignment is a key of the assignments bucket; an unassignment ends the key and keeps its values; a repeat is
+// absorbed** (ADR 0408, issue 1093): the readers of the store — what a machine runs, where a module runs, whether a module
+// may be forgotten, what is behind — read the bucket, the generation rises with each change, and a seat held by the
+// assignment goes with it.
+func TestAnAssignmentIsAKeyOfTheBucketAndAnUnassignmentEndsIt(t *testing.T) {
+	inv, node := aNodeWithModules(t, "postgres", "web")
+	ctx := t.Context()
+	if _, err := inv.SeedSeats(ctx, catalogue.DefaultSeats()); err != nil {
+		t.Fatal(err)
+	}
+	start := generationNow(t, inv)
+	if added, err := inv.Assign(ctx, node, "postgres"); err != nil || !added {
+		t.Fatalf("assigned %v: %v", added, err)
+	}
+	if added, err := inv.Assign(ctx, node, "postgres"); err != nil || added {
+		t.Fatalf("assigned again %v: %v", added, err)
+	}
+	if n := rowsInTheTable(t, inv, "assignment"); n != "0" {
+		t.Fatalf("the table took %s assignments", n)
+	}
+	kept, found, err := inv.chain.Assignments.Get(ctx, "anchor.postgres")
+	if err != nil || !found || kept.Value.Node != "anchor" || kept.Value.Module != "postgres" || kept.Value.Assigned.IsZero() {
+		t.Fatalf("the assignment on the bus: %+v %v %v", kept.Value, found, err)
+	}
+	if generationNow(t, inv) <= start {
+		t.Fatal("an assignment left the generation where it was")
+	}
+	if _, err := inv.Assign(ctx, node, "no-such-module"); !errors.Is(err, ErrNoSuchModule) {
+		t.Fatalf("a module the mesh does not know: %v", err)
+	}
+	if on, err := inv.Assigned(ctx, node); err != nil || !reflect.DeepEqual(on, []string{"postgres"}) {
+		t.Fatalf("assigned on %s: %v %v", node, on, err)
+	}
+	if running, err := inv.Running(ctx, "postgres"); err != nil || !reflect.DeepEqual(running, []string{node}) {
+		t.Fatalf("running: %v %v", running, err)
+	}
+	entries, err := inv.Catalogued(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, e := range entries {
+		switch e.Manifest.Module {
+		case "postgres":
+			if !reflect.DeepEqual(e.On, []string{node}) {
+				t.Errorf("postgres is on %v", e.On)
+			}
+		case "web":
+			if len(e.On) != 0 {
+				t.Errorf("web is on %v", e.On)
+			}
+		}
+	}
+	if err := inv.ForgetModule(ctx, "postgres"); !errors.Is(err, ErrStillAssigned) {
+		t.Fatalf("a module still assigned was forgotten: %v", err)
+	}
+	// A seat is held by an assignment, or by nothing.
+	if err := inv.HoldSeat(ctx, "mesh-broker", catalogue.ScopeMesh, node, "web"); err == nil {
+		t.Fatal("a seat was handed to a module that is not assigned")
+	}
+	if err := inv.HoldSeat(ctx, "mesh-broker", catalogue.ScopeMesh, node, "postgres"); err != nil {
+		t.Fatalf("a seat handed to an assignment: %v", err)
+	}
+
+	before := generationNow(t, inv)
+	if err := inv.Unassign(ctx, node, "postgres"); err != nil {
+		t.Fatal(err)
+	}
+	if generationNow(t, inv) <= before {
+		t.Fatal("an unassignment left the generation where it was")
+	}
+	if _, found, _ := inv.chain.Assignments.Get(ctx, "anchor.postgres"); found {
+		t.Fatal("an unassigned key reads as present")
+	}
+	if history, _ := inv.chain.Assignments.History(ctx, "anchor.postgres"); len(history) != 1 || history[0].Value.Module != "postgres" {
+		t.Fatalf("the earlier values of the key: %+v", history)
+	}
+	if held, err := inv.Holdings(ctx); err != nil || len(held) != 0 {
+		t.Fatalf("the seat the assignment held is held by %+v: %v", held, err)
+	}
+	if err := inv.Unassign(ctx, node, "postgres"); err == nil || err.Error() != "postgres is not assigned to anchor" {
+		t.Fatalf("an unassignment of nothing: %v", err)
+	}
+	if err := inv.ForgetModule(ctx, "postgres"); err != nil {
+		t.Fatalf("a module nothing is assigned could not be forgotten: %v", err)
+	}
+	// Made again after it ended.
+	if added, err := inv.Assign(ctx, node, "web"); err != nil || !added {
+		t.Fatal(added, err)
+	}
+	if added, err := inv.Assign(ctx, node, "web"); err != nil || added {
+		t.Fatal(added, err)
+	}
+}
+
+// **The provisioner's plans are kept in the assignments' values, one field at a time** (issue 1093, design 57 P6): the
+// plan of a module the machine was sent is written into its assignment, the build and commit another write put there stay,
+// a module only pulled in by another has a value of its own that ends when it is no longer pulled in, and nothing here
+// ends an assignment.
+func TestThePlansAreKeptInTheAssignmentsValuesAndNothingElseOfThemMoves(t *testing.T) {
+	inv, node := aNodeWithModules(t, "dunst", "web")
+	ctx := t.Context()
+	for _, m := range []string{"dunst", "web"} {
+		if _, err := inv.Assign(ctx, node, m); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// A build and a commit put in the value by another write.
+	cur, _, _ := inv.chain.Assignments.Get(ctx, "anchor.dunst")
+	cur.Value.Build, cur.Value.Commit = "build-1", "c1"
+	if _, err := inv.chain.Assignments.Update(ctx, "anchor.dunst", cur.Value, cur.Revision); err != nil {
+		t.Fatal(err)
+	}
+	record, _ := inv.NodeByName(ctx, node)
+	digest := "sha256:" + strings.Repeat("d", 64)
+	plan := Provisioning{Templates: TemplatesArchive{Source: "http://store:5000/v2/dunst/files/blobs/" + digest,
+		Digest: digest}, Files: []ProvisionedFile{{ID: "configuration", Path: "/home/op/.config/dunst/dunstrc",
+		Template: "dunstrc", OwnerAtRuntime: "owned", Takes: "reload-on"}}}
+	raw := []byte(`{"dunst":` + mustJSON(t, plan) + `,"notify":` + strings.Replace(mustJSON(t, plan), `{`, `{"required-by":"desk",`, 1) + `}`)
+	if err := inv.RecordSentWith(ctx, record.ID, "first", nil, 0, 0, Send{Digest: "first", Provisioning: raw}); err != nil {
+		t.Fatal(err)
+	}
+	inv.SetProvisioningPlans(inv.SentProvisioningPlans)
+	written, removed, err := inv.KeepAssignmentPlans(ctx)
+	if err != nil || written != 2 || removed != 0 {
+		t.Fatalf("kept: %d written, %d ended: %v", written, removed, err)
+	}
+	dunst, _, _ := inv.chain.Assignments.Get(ctx, "anchor.dunst")
+	if dunst.Value.Provisioning == nil || dunst.Value.Build != "build-1" || dunst.Value.Commit != "c1" || dunst.Value.Assigned.IsZero() {
+		t.Fatalf("dunst's value is %+v", dunst.Value)
+	}
+	if web, _, _ := inv.chain.Assignments.Get(ctx, "anchor.web"); web.Value.Provisioning != nil {
+		t.Fatalf("web has a plan: %+v", web.Value)
+	}
+	notify, found, _ := inv.chain.Assignments.Get(ctx, "anchor.notify")
+	if !found || notify.Value.RequiredBy != "desk" || !notify.Value.Assigned.IsZero() || notify.Value.Provisioning == nil {
+		t.Fatalf("the pulled-in module's value is %+v %v", notify.Value, found)
+	}
+	if on, _ := inv.Assigned(ctx, node); !reflect.DeepEqual(on, []string{"dunst", "web"}) {
+		t.Fatalf("a pulled-in module is assigned: %v", on)
+	}
+	if written, removed, err := inv.KeepAssignmentPlans(ctx); err != nil || written != 0 || removed != 0 {
+		t.Fatalf("a second pass: %d written, %d ended: %v", written, removed, err)
+	}
+	// Sent again without the pulled-in module: its value ends, the assignments stay.
+	only := []byte(`{"dunst":` + mustJSON(t, plan) + `}`)
+	if err := inv.RecordSentWith(ctx, record.ID, "second", nil, 0, 0, Send{Digest: "second", Provisioning: only}); err != nil {
+		t.Fatal(err)
+	}
+	if _, removed, err := inv.KeepAssignmentPlans(ctx); err != nil || removed != 1 {
+		t.Fatalf("ended %d: %v", removed, err)
+	}
+	if _, found, _ := inv.chain.Assignments.Get(ctx, "anchor.notify"); found {
+		t.Fatal("a module no longer pulled in has a value")
+	}
+	if on, _ := inv.Assigned(ctx, node); !reflect.DeepEqual(on, []string{"dunst", "web"}) {
+		t.Fatalf("assigned: %v", on)
+	}
+}
+
+// **The move carries every assignment with what its machine was last sent, and keeps the plan the mirror put in the value**
+// (ADR 0408 §8, issue 1093): the commit, the build of it and what it published, since when; a value that names its build
+// already is left as it is.
+func TestTheMoveCarriesTheAssignmentsAndWhatEachMachineWasSent(t *testing.T) {
+	inv, finish := ForTestBefore(t, 105)
+	finish()
+	ctx := context.Background()
+	for _, name := range []string{"anchor", "laptop"} {
+		if _, err := inv.AddNode(ctx, name); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for _, m := range []string{"notes", "wiki"} {
+		if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: m, Version: "1"}, Source{}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	for _, a := range [][2]string{{"anchor", "notes"}, {"anchor", "wiki"}, {"laptop", "notes"}} {
+		if _, err := inv.Assign(ctx, a[0], a[1]); err != nil {
+			t.Fatal(err)
+		}
+	}
+	aBuildRow(t, inv, "build-notes", "notes", "c-notes", "")
+	anchor, _ := inv.NodeByName(ctx, "anchor")
+	if err := inv.RecordSent(ctx, anchor.ID, "sent-1", map[string]string{"notes": "c-notes"}); err != nil {
+		t.Fatal(err)
+	}
+	// The mirror of the store put a value under one key, with a plan.
+	if _, err := inv.chain.Assignments.Create(ctx, "anchor.wiki", chain.Assignment{Node: "anchor", Module: "wiki",
+		Assigned: time.Now().UTC(), Provisioning: json.RawMessage(`{"plan":1}`)}); err != nil {
+		t.Fatal(err)
+	}
+	reports, err := inv.MoveChain(ctx, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var moved MoveReport
+	for _, r := range reports {
+		if r.Kind == KindAssignments {
+			moved = r
+		}
+	}
+	if moved.Rows != 3 || moved.Written+moved.Present != 3 {
+		t.Fatalf("the move of the assignments reported %+v", moved)
+	}
+	notes, _, _ := inv.chain.Assignments.Get(ctx, "anchor.notes")
+	if notes.Value.Commit != "c-notes" || notes.Value.Build != "build-notes" || notes.Value.Since.IsZero() || notes.Value.Assigned.IsZero() {
+		t.Fatalf("anchor.notes moved as %+v", notes.Value)
+	}
+	if notes.Value.Artifact != "sha256:build-notes" {
+		t.Fatalf("what the build published: %+v", notes.Value)
+	}
+	laptop, _, _ := inv.chain.Assignments.Get(ctx, "laptop.notes")
+	if laptop.Value.Commit != "" || laptop.Value.Build != "" {
+		t.Fatalf("a machine never sent has a build: %+v", laptop.Value)
+	}
+	wiki, _, _ := inv.chain.Assignments.Get(ctx, "anchor.wiki")
+	if string(wiki.Value.Provisioning) != `{"plan":1}` {
+		t.Fatalf("the plan the mirror put in the value was lost: %+v", wiki.Value)
+	}
+	// From now on the bucket is what is written.
+	if _, err := inv.Assign(ctx, "laptop", "wiki"); err != nil {
+		t.Fatal(err)
+	}
+	if n := rowsInTheTable(t, inv, "assignment"); n != "3" {
+		t.Fatalf("the table took an assignment after the move: %s", n)
+	}
+	if on, _ := inv.Assigned(ctx, "laptop"); !reflect.DeepEqual(on, []string{"notes", "wiki"}) {
+		t.Fatalf("assigned on laptop: %v", on)
+	}
+}

internal/inventory/chain_moved_test.go

@@ -6,8 +6,8 @@ import (
 )
 
 // **The move of the chain has a switch, made by the last migration and written by no migration** (ADR 0408 §8, issue
-// 1089): migration 0104 makes the one-row table chain_moved, and does not touch a table of the chain, which a later
-// delivery drops once the move is read back.
+// 1089): migration 0104 makes the one-row table chain_moved, and does not drop a table of the chain, which a later
+// delivery does once the move is read back.
 func TestMigration0104MakesTheSwitchOfTheChainsMoveAndNothingElse(t *testing.T) {
 	migrations, err := Migrations()
 	if err != nil {
@@ -27,11 +27,16 @@ func TestMigration0104MakesTheSwitchOfTheChainsMoveAndNothingElse(t *testing.T)
 			if strings.HasPrefix(strings.TrimSpace(line), "--") {
 				continue
 			}
-			for _, verb := range []string{"drop ", "insert ", "delete ", "update ", "alter "} {
+			for _, verb := range []string{"drop table", "insert ", "delete ", "update "} {
 				if strings.Contains(line, verb) {
 					t.Errorf("migration 0104 does more than make the switch: %q", line)
 				}
 			}
+			// The one other change: a seat's holder is no longer a foreign key to the assignment (issue 1093); the tables of
+			// the chain are dropped by a later migration (issue 1096).
+			if strings.Contains(line, "alter ") && !strings.Contains(line, "alter table seat_holding drop constraint") {
+				t.Errorf("migration 0104 alters more than the seat holdings' foreign key: %q", line)
+			}
 		}
 	}
 	if !found {

internal/inventory/fortest.go

@@ -72,7 +72,7 @@ func ForTest(t *testing.T) *Inventory {
 	testChains.Store(os.Getenv(store.Variable(Name)), c)
 	t.Cleanup(func() { testChains.Delete(os.Getenv(store.Variable(Name))) })
 	if _, err := inv.store.Pool().Exec(t.Context(),
-		`insert into chain_moved (counts, largest) values ($1, $1)`, `{"builds": 0, "rollouts": 0}`); err != nil {
+		`insert into chain_moved (counts, largest) values ($1, $1)`, `{"builds": 0, "rollouts": 0, "assignments": 0}`); err != nil {
 		t.Fatal(err)
 	}
 	return inv

internal/inventory/given.go

@@ -93,11 +93,28 @@ func (i *Inventory) ReplaceGivenAfterStart(ctx context.Context, node, declared s
 	if declared == "" {
 		return nil, nil
 	}
+	// Only a module assigned there: from the bucket when the assignments are on the bus, else the store's join.
+	_, onTheBus, err := i.ChainReads(ctx, KindAssignments)
+	if err != nil {
+		return nil, err
+	}
+	assignedHere := map[string]bool{}
+	join := `join assignment a on a.node = s.node and a.module = s.module`
+	if onTheBus {
+		join = ``
+		modules, err := i.assignedModules(ctx, node)
+		if err != nil {
+			return nil, err
+		}
+		for _, m := range modules {
+			assignedHere[m] = true
+		}
+	}
 	rows, err := i.store.Pool().Query(ctx,
 		`select s.node, s.module, s.name, s.replace_after_start
 		   from module_secret s
 		   join node n on n.id = s.node
-		   join assignment a on a.node = s.node and a.module = s.module
+		   `+join+`
 		  where n.name = $1 and n.sent = $2 and n.sent_at > s.replace_after_start
 		    and s.origin = 'accepted' and s.replace_after_start is not null
 		    and (not n.adopted or exists (select 1 from taken t where t.node = s.node and t.module = s.module))
@@ -117,6 +134,9 @@ func (i *Inventory) ReplaceGivenAfterStart(ctx context.Context, node, declared s
 			rows.Close()
 			return nil, err
 		}
+		if onTheBus && !assignedHere[d.module] {
+			continue
+		}
 		dues = append(dues, d)
 	}
 	rows.Close()

internal/inventory/given_test.go

@@ -7,8 +7,10 @@ import (
 	"errors"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/novox/mesh-controller/internal/catalogue"
+	"github.com/novox/mesh-controller/internal/chain"
 )
 
 // A module with one secret of each kind the rule tells apart (novox/hq ADR 0228), assigned to the
@@ -36,8 +38,9 @@ func assign(t *testing.T, inv *Inventory, ctx context.Context, node, module stri
 	if err != nil {
 		t.Fatal(err)
 	}
-	if _, err := inv.store.Pool().Exec(ctx,
-		`insert into assignment (node, module) values ($1, $2) on conflict do nothing`, n.ID, module); err != nil {
+	// Written as the fixture it is, past the checks Assign makes: the assignment on the bus.
+	if _, err := inv.chain.Assignments.Create(ctx, assignmentKey(n.Name, module),
+		chain.Assignment{Node: n.Name, Module: module, Assigned: time.Now().UTC()}); err != nil && !errors.Is(err, chain.ErrExists) {
 		t.Fatal(err)
 	}
 }

internal/inventory/migrations/0104-the-chain-moves-to-the-bus.sql

@@ -14,3 +14,8 @@ create table chain_moved (
     counts   jsonb       not null,
     largest  jsonb       not null
 );
+
+-- A seat's holder was an assignment by a foreign key; the assignments are keys of the bus once moved, so the controller says
+-- it (HoldSeat refuses a holder that is not assigned) and an unassignment takes the holding with it (Unassign), as the
+-- key's `on delete cascade` did.
+alter table seat_holding drop constraint seat_holding_node_module_fkey;

internal/inventory/mirrors.go

@@ -88,22 +88,15 @@ func (i *Inventory) AssignmentsMirror(ctx context.Context) (map[string][]byte, e
 }
 
 func (i *Inventory) assignmentsMirrored(ctx context.Context) ([]AssignmentMirrored, error) {
-	rows, err := i.store.Pool().Query(ctx,
-		`select n.name, a.module, a.assigned from assignment a join node n on n.id = a.node`)
+	rows, err := i.assignments(ctx)
 	if err != nil {
 		return nil, err
 	}
-	defer rows.Close()
-	var out []AssignmentMirrored
-	for rows.Next() {
-		var a AssignmentMirrored
-		if err := rows.Scan(&a.Node, &a.Module, &a.Assigned); err != nil {
-			return nil, err
-		}
-		a.Assigned = a.Assigned.UTC()
-		out = append(out, a)
+	out := make([]AssignmentMirrored, 0, len(rows))
+	for _, a := range rows {
+		out = append(out, AssignmentMirrored{Node: a.Node, Module: a.Module, Assigned: a.Assigned.UTC()})
 	}
-	return out, rows.Err()
+	return out, nil
 }
 
 // DeclaredConfigMirror is every registered module's mirrored config declarations, keyed by the module.

internal/inventory/nodes.go

@@ -1285,6 +1285,11 @@ func (i *Inventory) RecordSentUnder(ctx context.Context, node, digest string, bu
 // read from: every change a push carries happened after the last push, and an assignment is the one
 // that says when.
 func (i *Inventory) AwaitingSince(ctx context.Context, name string) (time.Time, error) {
+	if _, on, err := i.ChainReads(ctx, KindAssignments); err != nil {
+		return time.Time{}, err
+	} else if on {
+		return i.chainAwaitingSince(ctx, name)
+	}
 	var since time.Time
 	err := i.store.Pool().QueryRow(ctx,
 		`select coalesce(

internal/inventory/seats.go

@@ -238,6 +238,20 @@ func (i *Inventory) RenameSeat(ctx context.Context, from, to string) error {
 	return nil
 }
 
+// mustBeAssigned refuses to record a holder that is not an assignment: a seat cannot be handed to something that is not
+// running anywhere. The store's foreign key said it while the assignments were rows of the store; they are keys of the bus
+// now, so the controller says it.
+func (i *Inventory) mustBeAssigned(ctx context.Context, seat, nodeName, module string) error {
+	assigned, err := i.isAssigned(ctx, nodeName, module)
+	if err != nil {
+		return err
+	}
+	if !assigned {
+		return fmt.Errorf("recording %s on %s as the holder of %s: %s is not assigned to %s", module, nodeName, seat, module, nodeName)
+	}
+	return nil
+}
+
 // HoldSeat records that one assignment holds a seat, replacing whoever held it — every holder, for a
 // replicated seat (novox/hq ADR 0223) — as one transaction, so the seat is never without a holder in
 // between (novox/hq ADR 0131, design 28 task 5.3). The assignment must exist; the store refuses
@@ -248,6 +262,9 @@ func (i *Inventory) HoldSeat(ctx context.Context, seat, scope, nodeName, module
 	if err != nil {
 		return err
 	}
+	if err := i.mustBeAssigned(ctx, seat, nodeName, module); err != nil {
+		return err
+	}
 	tx, err := i.store.Pool().Begin(ctx)
 	if err != nil {
 		return err
@@ -276,6 +293,9 @@ func (i *Inventory) AddSeatHolder(ctx context.Context, seat, scope, nodeName, mo
 	if err != nil {
 		return err
 	}
+	if err := i.mustBeAssigned(ctx, seat, nodeName, module); err != nil {
+		return err
+	}
 	if _, err := i.store.Pool().Exec(ctx,
 		`insert into seat_holding (seat, scope, node, module) values ($1, $2, $3, $4)
 		 on conflict (seat, node, module) do nothing`,

internal/inventory/sends_test.go

@@ -22,9 +22,8 @@ func generationNow(t *testing.T, inv *Inventory) int64 {
 	return g
 }
 
-// **Every change to what is assigned where raises the generation, in its own transaction** — an
-// assignment, an unassignment, and one taken by the machine's removal, which no writer in Go makes — and
-// a repeated assignment, which changes nothing, does not.
+// **Every change to what is assigned where raises the generation** — an assignment and an unassignment — and a repeated
+// assignment, which changes nothing, does not.
 func TestEveryAssignmentChangeRaisesTheGeneration(t *testing.T) {
 	inv, node := aNodeWithModules(t, "postgres", "web")
 	ctx := t.Context()
@@ -52,16 +51,8 @@ func TestEveryAssignmentChangeRaisesTheGeneration(t *testing.T) {
 	if unassigned <= assigned {
 		t.Fatalf("an unassignment left the generation at %d", unassigned)
 	}
-	if _, err := inv.Assign(ctx, node, "web"); err != nil {
-		t.Fatal(err)
-	}
-	before := generationNow(t, inv)
-	if _, err := inv.RemoveNodeForTest(ctx, node); err != nil {
-		t.Fatal(err)
-	}
-	if removed := generationNow(t, inv); removed <= before {
-		t.Errorf("a machine's removal took its assignments and left the generation at %d", removed)
-	}
+	// A machine's removal, which no verb makes, is read from the machines the store knows: an assignment on a machine it
+	// does not know is none (assignments in the bucket have no key of the store to cascade through).
 }
 
 // **The generation never goes down, and is raised past what a machine applied when the mesh's own counter
@@ -175,22 +166,6 @@ func TestASendRecordThatCannotBeWrittenLeavesTheMachinesRecord(t *testing.T) {
 	}
 }
 
-// **An update that leaves an assignment as it was raises nothing** (the trigger's WHEN).
-func TestAnUpdateThatChangesNoAssignmentRaisesNothing(t *testing.T) {
-	inv, node := aNodeWithModules(t, "web")
-	ctx := t.Context()
-	if _, err := inv.Assign(ctx, node, "web"); err != nil {
-		t.Fatal(err)
-	}
-	before := generationNow(t, inv)
-	if _, err := inv.store.Pool().Exec(ctx, `update assignment set module = module`); err != nil {
-		t.Fatal(err)
-	}
-	if after := generationNow(t, inv); after != before {
-		t.Errorf("an update that changed nothing moved the generation from %d to %d", before, after)
-	}
-}
-
 // **A machine that reads a generation is recorded from its reports**, and one rolled back says so no longer.
 func TestWhetherAMachineReadsAGenerationIsItsLatestWord(t *testing.T) {
 	inv, node := aNodeWithModules(t)

internal/recordmirror/mirror.go

@@ -325,6 +325,10 @@ func validKey(key string) error {
 	return nil
 }
 
+// ChangedFields names the top-level fields whose values differ between two encoded objects, sorted; nil when either is
+// not an object: what an event of a changed key says of it.
+func ChangedFields(before, after []byte) []string { return changedFields(before, after) }
+
 // changedFields names the top-level fields whose values differ between two encoded objects, sorted; nil
 // when either is not an object.
 func changedFields(before, after []byte) []string {