jochen
13b6fc6d97
mesh/delivery-group group feat/journal-window-on-the-seat delivering: 0 of 2 delivered
mesh/merge-gate pass: builds build-agent, mesh-controller, route-proxy → ace, g14, novox, shanks; no bus step; every machine composes with the change as it…
mesh/repo-check pass: its merge-check.sh passed
mesh/delivery delivered
failed was required, so the controller refused the systemd module running today and the module serving it was refused by the controller running today: neither could land first. It is now optional (Verb.Optional, #117). Two things kept either change from reaching a mesh whose seat rows already exist: re-seeding added a verb but never an argument to one, and the console refuses an argument the row does not name, so the journal window would stay unreachable; and the optional mark is never stored, so a verb seeded into a row came back required. A row's verb now gains the arguments the binary names, and the working set takes the optional mark from the compiled seat, which also keeps mesh-delivery's checks optional once seeded.
362 lines
13 KiB
Go
362 lines
13 KiB
Go
package inventory
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
|
|
"github.com/novox/mesh-controller/internal/catalogue"
|
|
)
|
|
|
|
// The seats the mesh has, as data (novox/hq ADR 0122).
|
|
//
|
|
// The set the control plane reads is a table here, not a slice compiled into it. It is seeded from
|
|
// the binary's defaults the first time the mesh comes up (SeedSeats), and thereafter it is the live
|
|
// copy: a rename or an added seat is a write here, and the control plane loads it at startup rather
|
|
// than being rebuilt for it.
|
|
|
|
// Seats is every seat the mesh defines, read from the store.
|
|
func (i *Inventory) Seats(ctx context.Context) ([]catalogue.Seat, error) {
|
|
rows, err := i.store.Pool().Query(ctx,
|
|
`select name, scope, delivers, decided, accepts, emits, serves from seat order by name`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var seats []catalogue.Seat
|
|
for rows.Next() {
|
|
var s catalogue.Seat
|
|
var accepts, emits, serves []byte
|
|
if err := rows.Scan(&s.Name, &s.Scope, &s.Delivers, &s.Decision, &accepts, &emits, &serves); err != nil {
|
|
return nil, err
|
|
}
|
|
// The protocol, from the row (novox/hq ADR 0132). A row that predates the columns has empty
|
|
// lists, and UseSeats keeps the compiled protocol for it until the next seeding fills them.
|
|
if err := json.Unmarshal(accepts, &s.Accepts); err != nil {
|
|
return nil, fmt.Errorf("seat %s: accepts: %w", s.Name, err)
|
|
}
|
|
if err := json.Unmarshal(emits, &s.Emits); err != nil {
|
|
return nil, fmt.Errorf("seat %s: emits: %w", s.Name, err)
|
|
}
|
|
if err := json.Unmarshal(serves, &s.Serves); err != nil {
|
|
return nil, fmt.Errorf("seat %s: serves: %w", s.Name, err)
|
|
}
|
|
seats = append(seats, s)
|
|
}
|
|
return seats, rows.Err()
|
|
}
|
|
|
|
// SeedSeats writes the mesh's default set into the table where it is not already present.
|
|
//
|
|
// **Idempotent, and never overwriting.** Run every time the control plane migrates, it fills an
|
|
// empty table on first boot and adds a seat a new release ships — but it leaves a row already there
|
|
// exactly as it is, so an operator's rename in the table is not undone by the next deploy putting
|
|
// the old name back. What a release removes from the defaults is not deleted here either; retiring a
|
|
// seat is its own decision, not a silent consequence of it dropping out of the binary.
|
|
//
|
|
// **The protocol is seeded additively** (novox/hq ADR 0132, design 33 §7). A row that has none takes
|
|
// the compiled protocol whole — that is the compiled fallback becoming data, once. A row that has one
|
|
// gains any verb the defaults name and it lacks, and loses nothing: a seat's tools are an interface,
|
|
// additive within a version, and a verb an operator added to the row is theirs to keep.
|
|
func (i *Inventory) SeedSeats(ctx context.Context, defaults []catalogue.Seat) (int, error) {
|
|
var added int
|
|
for _, s := range defaults {
|
|
accepts, emits, serves, err := protocolJSON(s)
|
|
if err != nil {
|
|
return added, err
|
|
}
|
|
tag, err := i.store.Pool().Exec(ctx,
|
|
`insert into seat (name, scope, delivers, decided, accepts, emits, serves)
|
|
values ($1, $2, $3, $4, $5, $6, $7)
|
|
on conflict (name) do nothing`,
|
|
s.Name, s.Scope, s.Delivers, s.Decision, accepts, emits, serves)
|
|
if err != nil {
|
|
return added, err
|
|
}
|
|
if n := int(tag.RowsAffected()); n > 0 {
|
|
added += n
|
|
continue
|
|
}
|
|
if err := i.widenProtocol(ctx, s); err != nil {
|
|
return added, err
|
|
}
|
|
}
|
|
return added, nil
|
|
}
|
|
|
|
// widenProtocol adds to a seat's row whatever the defaults name and the row lacks, by verb name.
|
|
func (i *Inventory) widenProtocol(ctx context.Context, s catalogue.Seat) error {
|
|
var accepts, emits, serves []byte
|
|
if err := i.store.Pool().QueryRow(ctx,
|
|
`select accepts, emits, serves from seat where name = $1`, s.Name).Scan(&accepts, &emits, &serves); err != nil {
|
|
return err
|
|
}
|
|
var row catalogue.Seat
|
|
if err := json.Unmarshal(accepts, &row.Accepts); err != nil {
|
|
return err
|
|
}
|
|
if err := json.Unmarshal(emits, &row.Emits); err != nil {
|
|
return err
|
|
}
|
|
if err := json.Unmarshal(serves, &row.Serves); err != nil {
|
|
return err
|
|
}
|
|
changed := false
|
|
row.Accepts, changed = union(row.Accepts, s.Accepts, changed)
|
|
row.Emits, changed = union(row.Emits, s.Emits, changed)
|
|
have := map[string]int{}
|
|
for n, v := range row.Serves {
|
|
have[v.Name] = n
|
|
}
|
|
for _, v := range s.Serves {
|
|
at, kept := have[v.Name]
|
|
if !kept {
|
|
row.Serves = append(row.Serves, v)
|
|
changed = true
|
|
continue
|
|
}
|
|
// **The controller's own verbs are described by the binary that runs them** (novox/hq
|
|
// issue 244, to-be 45 §7). Its seat's verbs are not an operator's to reshape: each is a
|
|
// command line this binary composes from the arguments its own table declares, and the
|
|
// console judges a call against the row. A row kept from an older build described `push`
|
|
// without the `behind` and `why` the binary takes, so the console refused an argument the verb
|
|
// needs. So a verb this binary defines takes this binary's definition; a verb only the row has
|
|
// — a newer build's, during a roll-out (ADR 0185) — is left as it is.
|
|
if s.Name == catalogue.ControllerSeatName && !sameVerb(row.Serves[at], v) {
|
|
row.Serves[at] = v
|
|
changed = true
|
|
continue
|
|
}
|
|
// **Any other seat's verb gains the arguments the binary names and the row lacks** — additive,
|
|
// as the rest of the protocol is. The console refuses an argument the row does not name (issue
|
|
// 244), so a holder taught a new argument (the service manager's journal window) would be
|
|
// unreachable through it while the row kept the older schema. Nothing the row has is removed or
|
|
// made required; the description is the binary's, since it describes the arguments added.
|
|
if widened, ok := widenInput(row.Serves[at], v); ok {
|
|
row.Serves[at] = widened
|
|
changed = true
|
|
}
|
|
}
|
|
if !changed {
|
|
return nil
|
|
}
|
|
a, e, sv, err := protocolJSON(row)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = i.store.Pool().Exec(ctx,
|
|
`update seat set accepts = $2, emits = $3, serves = $4 where name = $1`, s.Name, a, e, sv)
|
|
return err
|
|
}
|
|
|
|
func union(have, want []string, changed bool) ([]string, bool) {
|
|
seen := map[string]bool{}
|
|
for _, h := range have {
|
|
seen[h] = true
|
|
}
|
|
for _, w := range want {
|
|
if !seen[w] {
|
|
have = append(have, w)
|
|
seen[w] = true
|
|
changed = true
|
|
}
|
|
}
|
|
return have, changed
|
|
}
|
|
|
|
func protocolJSON(s catalogue.Seat) (accepts, emits, serves []byte, err error) {
|
|
if accepts, err = json.Marshal(orEmpty(s.Accepts)); err != nil {
|
|
return
|
|
}
|
|
if emits, err = json.Marshal(orEmpty(s.Emits)); err != nil {
|
|
return
|
|
}
|
|
verbs := s.Serves
|
|
if verbs == nil {
|
|
verbs = []catalogue.Verb{}
|
|
}
|
|
serves, err = json.Marshal(verbs)
|
|
return
|
|
}
|
|
|
|
func orEmpty(s []string) []string {
|
|
if s == nil {
|
|
return []string{}
|
|
}
|
|
return s
|
|
}
|
|
|
|
// Aliases is every former seat name and the seat it now resolves to (novox/hq ADR 0122).
|
|
func (i *Inventory) Aliases(ctx context.Context) (map[string]string, error) {
|
|
rows, err := i.store.Pool().Query(ctx, `select alias, seat from seat_alias`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
aliases := map[string]string{}
|
|
for rows.Next() {
|
|
var alias, seat string
|
|
if err := rows.Scan(&alias, &seat); err != nil {
|
|
return nil, err
|
|
}
|
|
aliases[alias] = seat
|
|
}
|
|
return aliases, rows.Err()
|
|
}
|
|
|
|
// RenameSeat gives a seat a new name and keeps the old one as an alias (novox/hq ADR 0122).
|
|
//
|
|
// **This is the whole of a rename.** The seat's canonical name becomes `to`; `from` is remembered as
|
|
// an alias so every reference to it — a manifest's claim, a held record, the build machine's
|
|
// embedded set — goes on resolving to the same seat, unchanged. Nothing is rebuilt and nothing
|
|
// freezes. Any alias that pointed to `from` is repointed to `to`, so a chain of renames does not
|
|
// leave an older name resolving to a name that no longer exists.
|
|
func (i *Inventory) RenameSeat(ctx context.Context, from, to string) error {
|
|
if from == to {
|
|
return fmt.Errorf("a seat is renamed to a different name; %q is already its name", to)
|
|
}
|
|
tag, err := i.store.Pool().Exec(ctx, `update seat set name = $1 where name = $2`, to, from)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return fmt.Errorf("no seat named %q to rename", from)
|
|
}
|
|
// The old name resolves to the new one; and any name that resolved to the old one now resolves
|
|
// to the new one, so no alias is left pointing at a name that is gone.
|
|
if _, err := i.store.Pool().Exec(ctx,
|
|
`insert into seat_alias (alias, seat) values ($1, $2)
|
|
on conflict (alias) do update set seat = excluded.seat`, from, to); err != nil {
|
|
return err
|
|
}
|
|
if _, err := i.store.Pool().Exec(ctx,
|
|
`update seat_alias set seat = $1 where seat = $2`, to, from); err != nil {
|
|
return err
|
|
}
|
|
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
|
|
// otherwise, and that refusal is the right one: a seat cannot be handed to something that is not
|
|
// running anywhere.
|
|
func (i *Inventory) HoldSeat(ctx context.Context, seat, scope, nodeName, module string) error {
|
|
node, err := i.NodeByName(ctx, nodeName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
tx, err := i.store.Pool().Begin(ctx)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
|
if _, err := tx.Exec(ctx,
|
|
`delete from seat_holding where seat = $1 and not (node = $2 and module = $3)`,
|
|
seat, node.ID, module); err != nil {
|
|
return fmt.Errorf("handing %s to %s on %s: %w", seat, module, nodeName, err)
|
|
}
|
|
if _, err := tx.Exec(ctx,
|
|
`insert into seat_holding (seat, scope, node, module) values ($1, $2, $3, $4)
|
|
on conflict (seat, node, module) do update set scope = excluded.scope, since = now()`,
|
|
seat, scope, node.ID, module); err != nil {
|
|
return fmt.Errorf("recording %s on %s as the holder of %s: %w", module, nodeName, seat, err)
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
|
|
// AddSeatHolder records one more assignment holding a replicated seat, beside those already on
|
|
// record (novox/hq ADR 0223). Whether the seat may have several holders is the caller's to judge —
|
|
// the seat's definition is compiled, and the store holds no column for it. Recording a holder
|
|
// already on record changes nothing.
|
|
func (i *Inventory) AddSeatHolder(ctx context.Context, seat, scope, nodeName, module string) error {
|
|
node, err := i.NodeByName(ctx, nodeName)
|
|
if 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`,
|
|
seat, scope, node.ID, module); err != nil {
|
|
return fmt.Errorf("adding %s on %s as a holder of %s: %w", module, nodeName, seat, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Holdings is every seat whose holder is on record, as the resolver reads it. A seat with no row here
|
|
// is held by derivation, exactly as before the table existed.
|
|
func (i *Inventory) Holdings(ctx context.Context) ([]catalogue.Held, error) {
|
|
rows, err := i.store.Pool().Query(ctx,
|
|
`select h.seat, h.scope, n.name, h.module, coalesce(n.site, '')
|
|
from seat_holding h join node n on n.id = h.node order by h.seat, n.name, h.module`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
var out []catalogue.Held
|
|
for rows.Next() {
|
|
var h catalogue.Held
|
|
if err := rows.Scan(&h.Claim, &h.Scope, &h.Node, &h.Module, &h.Site); err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, h)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// widenInput is the row's verb with every input property the binary's names and the row's lacks, and
|
|
// the binary's description; and whether anything was added.
|
|
func widenInput(row, binary catalogue.Verb) (catalogue.Verb, bool) {
|
|
want, _ := binary.Input["properties"].(map[string]any)
|
|
if len(want) == 0 {
|
|
return row, false
|
|
}
|
|
have, _ := row.Input["properties"].(map[string]any)
|
|
added := map[string]any{}
|
|
for name, p := range want {
|
|
if _, kept := have[name]; !kept {
|
|
added[name] = p
|
|
}
|
|
}
|
|
if len(added) == 0 {
|
|
return row, false
|
|
}
|
|
input := map[string]any{}
|
|
for k, v := range row.Input {
|
|
input[k] = v
|
|
}
|
|
if input["type"] == nil {
|
|
input["type"] = "object"
|
|
}
|
|
props := map[string]any{}
|
|
for k, v := range have {
|
|
props[k] = v
|
|
}
|
|
for k, v := range added {
|
|
props[k] = v
|
|
}
|
|
input["properties"] = props
|
|
row.Input = input
|
|
if binary.Description != "" {
|
|
row.Description = binary.Description
|
|
}
|
|
return row, true
|
|
}
|
|
|
|
// sameVerb is whether two definitions of a verb say the same, read as the row stores them.
|
|
func sameVerb(a, b catalogue.Verb) bool {
|
|
ja, errA := json.Marshal(a)
|
|
jb, errB := json.Marshal(b)
|
|
if errA != nil || errB != nil {
|
|
return false
|
|
}
|
|
var ra, rb any
|
|
_ = json.Unmarshal(ja, &ra)
|
|
_ = json.Unmarshal(jb, &rb)
|
|
ca, _ := json.Marshal(ra)
|
|
cb, _ := json.Marshal(rb)
|
|
return string(ca) == string(cb)
|
|
}
|