diff --git a/cmd/mesh-controller/record_mirrors.go b/cmd/mesh-controller/record_mirrors.go index 6889d79e..b1af389f 100644 --- a/cmd/mesh-controller/record_mirrors.go +++ b/cmd/mesh-controller/record_mirrors.go @@ -2,12 +2,15 @@ package main import ( "context" + "encoding/json" "fmt" + "sort" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/recordmirror" ) @@ -49,9 +52,17 @@ func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats. failing := map[string]string{} tick := time.NewTicker(mirrorEvery) defer tick.Stop() + events := &assignmentEvents{} for { if holds() { + // The assignments, once they are keys of the bus, are written by the verbs that change them (novox/hq ADR + // 0408): this tick keeps the provisioner's plan in their values and says what changed; the mirror of the store + // writes them only until the move. + onTheBus := keepingAssignments(ctx, inv, announce, epoch, events, failing) for _, m := range recordmirror.Mirrors() { + if m.Name == recordmirror.Assignments.Name && onTheBus { + continue + } why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds) if why != failing[m.Name] { if why != "" { @@ -104,3 +115,81 @@ func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mir } return "" } + +// assignmentEvents is the assignments as the bucket held them at the last look, to say what changed since. +type assignmentEvents struct { + known map[string]inventory.KeptValue +} + +// keepingAssignments is the tick's work for the assignments once they are on the bus: the provisioner's plan written into +// the values it belongs to, and one `assignment-changed` event for each key that is new, changed or ended since the last +// look — whoever wrote it, a verb in this process or a command in another. False when they are not on the bus yet. +func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce recordmirror.Announcer, epoch func() uint64, + events *assignmentEvents, failing map[string]string) bool { + reading, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + const name = "assignments" + if _, on, err := inv.KeptAssignments(reading); err != nil || !on { + if err != nil { + // Not known to be off the bus: the store's mirror is not written over what the bus may hold. + if failing[name] == "" { + failing[name] = err.Error() + fmt.Printf("the assignments cannot be read from the bus: %v\n", err) + } + return true + } + return false + } + if _, _, err := inv.KeepAssignmentPlans(reading); err != nil { + if failing[name] != err.Error() { + failing[name] = err.Error() + fmt.Printf("the provisioner's plans are not kept in the assignments: %v\n", err) + } + } else if failing[name] != "" { + fmt.Printf("the provisioner's plans are kept in the assignments again\n") + failing[name] = "" + } + now, _, err := inv.KeptAssignments(reading) + if err != nil { + return true + } + if events.known != nil { + keys := make([]string, 0, len(now)+len(events.known)) + for key := range now { + keys = append(keys, key) + } + for key := range events.known { + if _, still := now[key]; !still { + keys = append(keys, key) + } + } + sort.Strings(keys) + for _, key := range keys { + was, had := events.known[key] + is, has := now[key] + e := recordmirror.Event{Mirror: recordmirror.Assignments.Name, Key: key, Revision: is.Revision, Epoch: epoch(), + At: time.Now().UTC()} + switch { + case has && !had: + e.Change = recordmirror.Added + case has && had && was.Revision != is.Revision: + e.Change, e.Fields = recordmirror.Changed, recordmirror.ChangedFields(was.Value, is.Value) + case !has && had: + e.Change, e.Revision = recordmirror.Removed, 0 + default: + continue + } + body, err := json.Marshal(e) + if err != nil { + continue + } + for try := 0; try < 2; try++ { + if err = announce.PublishSeatEvent(reading, broker.ControllerSeat, recordmirror.Assignments.Event, body); err == nil { + break + } + } + } + } + events.known = now + return true +} diff --git a/internal/broker/writers.go b/internal/broker/writers.go index 44d71a1a..2520359d 100644 --- a/internal/broker/writers.go +++ b/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. diff --git a/internal/chain/backend.go b/internal/chain/backend.go index 486095c7..a099ad3c 100644 --- a/internal/chain/backend.go +++ b/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 } diff --git a/internal/chain/bus.go b/internal/chain/bus.go index f34670b2..954813cf 100644 --- a/internal/chain/bus.go +++ b/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 diff --git a/internal/chain/chain.go b/internal/chain/chain.go index bb3f6862..f39a9b00 100644 --- a/internal/chain/chain.go +++ b/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, } } diff --git a/internal/chain/chain_test.go b/internal/chain/chain_test.go index 4b47260f..de0a26ab 100644 --- a/internal/chain/chain_test.go +++ b/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") + } +} diff --git a/internal/inventory/adoption.go b/internal/inventory/adoption.go index 02f3da4e..7d01b062 100644 --- a/internal/inventory/adoption.go +++ b/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 diff --git a/internal/inventory/catalogue.go b/internal/inventory/catalogue.go index 14c8edfd..56a11d17 100644 --- a/internal/inventory/catalogue.go +++ b/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. diff --git a/internal/inventory/catalogue_test.go b/internal/inventory/catalogue_test.go index f956ba31..24566a2b 100644 --- a/internal/inventory/catalogue_test.go +++ b/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()) diff --git a/internal/inventory/chain_assignments.go b/internal/inventory/chain_assignments.go new file mode 100644 index 00000000..4ebf8fbe --- /dev/null +++ b/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 `.` 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 +} diff --git a/internal/inventory/chain_assignments_key_test.go b/internal/inventory/chain_assignments_key_test.go new file mode 100644 index 00000000..47112e82 --- /dev/null +++ b/internal/inventory/chain_assignments_key_test.go @@ -0,0 +1,37 @@ +package inventory + +import ( + "encoding/json" + "testing" +) + +// **An assignment's key is `.`, 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) + } + } +} diff --git a/internal/inventory/chain_move.go b/internal/inventory/chain_move.go index 1f37dfd5..c97c36d3 100644 --- a/internal/inventory/chain_move.go +++ b/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 diff --git a/internal/inventory/chain_move_test.go b/internal/inventory/chain_move_test.go index febda177..aa8d757a 100644 --- a/internal/inventory/chain_move_test.go +++ b/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) + } +} diff --git a/internal/inventory/chain_moved_test.go b/internal/inventory/chain_moved_test.go index 5298aeef..3ea88b25 100644 --- a/internal/inventory/chain_moved_test.go +++ b/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 { diff --git a/internal/inventory/fortest.go b/internal/inventory/fortest.go index a063b1e5..5a2067d0 100644 --- a/internal/inventory/fortest.go +++ b/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 diff --git a/internal/inventory/given.go b/internal/inventory/given.go index fac8574b..20507be1 100644 --- a/internal/inventory/given.go +++ b/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() diff --git a/internal/inventory/given_test.go b/internal/inventory/given_test.go index a79a7877..ade0433a 100644 --- a/internal/inventory/given_test.go +++ b/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) } } diff --git a/internal/inventory/migrations/0104-the-chain-moves-to-the-bus.sql b/internal/inventory/migrations/0104-the-chain-moves-to-the-bus.sql index 92702cdb..5a865465 100644 --- a/internal/inventory/migrations/0104-the-chain-moves-to-the-bus.sql +++ b/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; diff --git a/internal/inventory/mirrors.go b/internal/inventory/mirrors.go index b4b09a68..3f35f767 100644 --- a/internal/inventory/mirrors.go +++ b/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. diff --git a/internal/inventory/nodes.go b/internal/inventory/nodes.go index 55cdfc27..be27fdda 100644 --- a/internal/inventory/nodes.go +++ b/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( diff --git a/internal/inventory/seats.go b/internal/inventory/seats.go index 2c7ceff0..33dc4706 100644 --- a/internal/inventory/seats.go +++ b/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`, diff --git a/internal/inventory/sends_test.go b/internal/inventory/sends_test.go index b08ab198..aba20d88 100644 --- a/internal/inventory/sends_test.go +++ b/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) diff --git a/internal/recordmirror/mirror.go b/internal/recordmirror/mirror.go index d23bee95..a751f73c 100644 --- a/internal/recordmirror/mirror.go +++ b/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 { diff --git a/testdata/store-tests b/testdata/store-tests index 49d5da6a..42ed2763 100644 --- a/testdata/store-tests +++ b/testdata/store-tests @@ -149,3 +149,6 @@ ./internal/inventory TestTheMoveCarriesTheBuildRequestsAndMarksTheNeverAskedToAskAgain ./internal/inventory TestTheMoveCarriesThePlansAndTheirMergesAndTheyReadTheSame ./cmd/mesh-controller TestAMergeWhosePlanCannotBeSavedIsHeldAndRecordedWhenHandedOverAgain +./internal/inventory TestAnAssignmentIsAKeyOfTheBucketAndAnUnassignmentEndsIt +./internal/inventory TestThePlansAreKeptInTheAssignmentsValuesAndNothingElseOfThemMoves +./internal/inventory TestTheMoveCarriesTheAssignmentsAndWhatEachMachineWasSent