jochen
e74c32ed50
A controller restart lost every call's outcome, `status` composed the mesh while its caller waited (18.6s live on 2026-10-06, past the 10s window), a repair by hand left no trace, and the core's bounds had nothing measured to be set from. - calls: kept in the controller's bucket mesh-controller_calls (last 1000 or 14 days, answers bounded to 64 KiB), read by id across a restart; a controller starting marks a stopped one's running calls abandoned; each call names its caller from the inbox its answer goes to. - status: the serving controller composes it at start, after news from a machine, a build or an acting verb, and every minute; the verb answers the last composition at once with when and how long it took. Composing resolves each machine once instead of twice. - hand-act log in mesh-controller_hand-acts: push (required through the seat), plans stop/close, broker consumer-reset and the new hand-act record take --why/--cause/--condition; `hand-acts` lists them and repeated causes; status counts the week's. - durations (migration 0066): apply (send to first report), heartbeat gap, plan tier and build, recorded as heard; `durations` summarises them. - the controller's seat row takes this binary's definition of its own verbs, so the console no longer judges calls against an older build's schema. - the controller is granted its two buckets' subjects.
313 lines
11 KiB
Go
313 lines
11 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
|
|
}
|
|
}
|
|
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()
|
|
}
|
|
|
|
// 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)
|
|
}
|