Files
mesh-controller/internal/inventory/libraries.go
T
jschoubben 5b9908bed2 Hold a joining build for a person without superseding its delivery, join only on an engine that keeps own groups, and keep a held module's settings (hq ADR 0328 §8, ADR 0334; re-review of 96e15f86)
A delivery does not send a joining build held for its library or its engine: the module waits for a person,
the plan stays open saying so and judges its other modules, and sends it once nothing holds it. A held move is
no move for the gate. A module joins only on an engine at contract 7, and a held module is composed with the
settings it last ran with. The runner's verdicts are said for their modules and judged by whether a move
changes them. mesh-host pinned at #97 (7a41e120), its bundle now examples/foundation-first-node.lock.
2026-10-13 00:55:54 +02:00

332 lines
12 KiB
Go

package inventory
import (
"context"
"encoding/json"
"errors"
"fmt"
"path/filepath"
"strings"
"time"
"github.com/jackc/pgx/v5"
"github.com/novox/mesh-controller/internal/catalogue"
)
// Joining modules' numbers, library groups, taken libraries and the libraries themselves (novox/hq ADR 0328 §1, §3,
// §4, §7, §15; migration 0096).
// AllocateIdentity is a joining module's number, allocated now when it has none: one number from the mesh's range,
// mesh-wide, kept for ever, its user's and its own group's. Two controllers allocating at once agree: the second's
// insert finds the first's row.
func (i *Inventory) AllocateIdentity(ctx context.Context, module string) (catalogue.ModuleIdentity, error) {
if strings.TrimSpace(module) == "" {
return catalogue.ModuleIdentity{}, errors.New("a number is allocated for a module, and none was named")
}
if id, had, err := i.IdentityOf(ctx, module); err != nil || had {
return id, err
}
// Drawn only when the row is new; a draw lost to a concurrent insert is a gap in the range, never a number given
// twice.
if _, err := i.store.Pool().Exec(ctx,
`insert into module_identity (module, uid) select $1, nextval('mesh_number')::integer
on conflict (module) do nothing`, module); err != nil {
return catalogue.ModuleIdentity{}, fmt.Errorf("%s could not be given its number: %w", module, err)
}
id, had, err := i.IdentityOf(ctx, module)
if err == nil && !had {
err = fmt.Errorf("%s was given its number and it cannot be read back", module)
}
return id, err
}
// IdentityOf is a module's number, and false when it was never allocated one.
func (i *Inventory) IdentityOf(ctx context.Context, module string) (catalogue.ModuleIdentity, bool, error) {
id := catalogue.ModuleIdentity{Module: module}
err := i.store.Pool().QueryRow(ctx, `select uid from module_identity where module = $1`, module).Scan(&id.UID)
if errors.Is(err, pgx.ErrNoRows) {
return catalogue.ModuleIdentity{}, false, nil
}
if err != nil {
return catalogue.ModuleIdentity{}, false, err
}
return id, true, nil
}
// Identities is every joining module's number, by module.
func (i *Inventory) Identities(ctx context.Context) (map[string]catalogue.ModuleIdentity, error) {
rows, err := i.store.Pool().Query(ctx, `select module, uid from module_identity`)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[string]catalogue.ModuleIdentity{}
for rows.Next() {
var id catalogue.ModuleIdentity
if err := rows.Scan(&id.Module, &id.UID); err != nil {
return nil, err
}
out[id.Module] = id
}
return out, rows.Err()
}
// AllocateLibraryGroup is a library's group number, allocated now when it has none, from the same sequence as the
// modules' numbers (§3).
func (i *Inventory) AllocateLibraryGroup(ctx context.Context, library string) (int, error) {
if strings.TrimSpace(library) == "" {
return 0, errors.New("a group number is allocated for a library, and none was named")
}
groups, err := i.LibraryGroups(ctx)
if err != nil {
return 0, err
}
if gid, had := groups[library]; had {
return gid, nil
}
if _, err := i.store.Pool().Exec(ctx,
`insert into library_group (library, gid) select $1, nextval('mesh_number')::integer
on conflict (library) do nothing`, library); err != nil {
return 0, fmt.Errorf("the library %s could not be given its group's number: %w", library, err)
}
if groups, err = i.LibraryGroups(ctx); err != nil {
return 0, err
}
gid, has := groups[library]
if !has {
return 0, fmt.Errorf("the library %s was given its group's number and it cannot be read back", library)
}
return gid, nil
}
// LibraryGroups is every library's group number, by library.
func (i *Inventory) LibraryGroups(ctx context.Context) (map[string]int, error) {
rows, err := i.store.Pool().Query(ctx, `select library, gid from library_group`)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[string]int{}
for rows.Next() {
var name string
var gid int
if err := rows.Scan(&name, &gid); err != nil {
return nil, err
}
out[name] = gid
}
return out, rows.Err()
}
// Libraries is every library the operator defined (§1): the mesh-wide `libraries` layer of the controller's module.
// None when it was never set.
func (i *Inventory) Libraries(ctx context.Context) (map[string]catalogue.Library, error) {
var raw []byte
err := i.store.Pool().QueryRow(ctx,
`select values from settings where module = $1 and node is null`, catalogue.LibrariesModule).Scan(&raw)
if errors.Is(err, pgx.ErrNoRows) {
return nil, nil
}
if err != nil {
return nil, err
}
var layer map[string]any
if err := json.Unmarshal(raw, &layer); err != nil {
return nil, err
}
return catalogue.ParseLibraries(layer[catalogue.LibrariesSetting])
}
// judgeLibraries is what a layer of the controller's module is held to when it defines libraries (§1, §4, §13): only
// mesh-wide, well formed, no library named as a registered module, none overlapping another on a machine.
func (i *Inventory) judgeLibraries(ctx context.Context, nodeName, module string, values map[string]any) error {
v, defines := values[catalogue.LibrariesSetting]
if module != catalogue.LibrariesModule || !defines {
return nil
}
if nodeName != "" {
return fmt.Errorf("refused: %s is the mesh's, set on the whole mesh and never on one machine (%s §1)",
catalogue.LibrariesSetting, catalogue.ADR0328)
}
libs, err := catalogue.ParseLibraries(v)
if err != nil {
return fmt.Errorf("refused: %w", err)
}
shelf, err := i.Catalogue(ctx)
if err != nil {
return err
}
modules := make([]string, 0, len(shelf))
for name := range shelf {
modules = append(modules, name)
}
if problems := catalogue.LibrariesProblems(libs, modules); len(problems) > 0 {
return fmt.Errorf("refused: %s", strings.Join(problems, "; "))
}
return nil
}
// LibraryNamed refuses a module whose name is a defined library's (§4): mesh-<module> and mesh-<library> never collide.
func (i *Inventory) LibraryNamed(ctx context.Context, module string) error {
libs, err := i.Libraries(ctx)
if err != nil {
return err
}
if _, clash := libs[module]; clash {
return fmt.Errorf("the module %s has the name of the library %s: %s would be both (%s §4)", module, module,
catalogue.MeshName(module), catalogue.ADR0328)
}
return nil
}
// LibraryPaths is every sub-path a module on any machine asks of each library (§6): what its home keeps. A module on a
// machine is one assigned there or one a requirement pulled in — what the machine was last sent.
func (i *Inventory) LibraryPaths(ctx context.Context) (map[string][]string, error) {
shelf, err := i.Catalogue(ctx)
if err != nil {
return nil, err
}
// Every module assigned anywhere, and every module a machine was last sent — those a requirement pulled in too.
rows, err := i.store.Pool().Query(ctx,
`select module from assignment
union select jsonb_object_keys(sent_builds) from node where sent_builds is not null and jsonb_typeof(sent_builds) = 'object'`)
if err != nil {
return nil, err
}
defer rows.Close()
seen := map[string]map[string]bool{}
for rows.Next() {
var module string
if err := rows.Scan(&module); err != nil {
return nil, err
}
for _, u := range shelf[module].Libraries {
if seen[u.Library] == nil {
seen[u.Library] = map[string]bool{}
}
seen[u.Library][filepath.Clean(u.Path)] = true
}
}
if err := rows.Err(); err != nil {
return nil, err
}
out := map[string][]string{}
for lib, paths := range seen {
for p := range paths {
out[lib] = append(out[lib], p)
}
}
return out, nil
}
// TakenLibrary is one library the operator let the mesh take, on its home (§7).
type TakenLibrary struct {
Library string
Home string
Root string
Placed string
By string
Ask string
Warrant string
Said string
// Done is whether the engine stated the take's act done: only then is the library taken.
Done bool
}
// TakenLibraries is every library whose take was started, by name; Done says it is taken.
func (i *Inventory) TakenLibraries(ctx context.Context) (map[string]TakenLibrary, error) {
rows, err := i.store.Pool().Query(ctx,
`select library, home, root, placed, taken_by, ask, warrant, said, done from kept_library`)
if err != nil {
return nil, err
}
defer rows.Close()
out := map[string]TakenLibrary{}
for rows.Next() {
var t TakenLibrary
if err := rows.Scan(&t.Library, &t.Home, &t.Root, &t.Placed, &t.By, &t.Ask, &t.Warrant, &t.Said, &t.Done); err != nil {
return nil, err
}
out[t.Library] = t
}
return out, rows.Err()
}
// RecordTakeStarted keeps that a take of a library was started on its home, under the operator's warrant: the
// engine said it began. The library is taken only once the engine states the act done (TakeDone). Once per home, root
// and filesystem: a second start of the same keeps the first, and one on another (it moved, and the operator was asked
// again) replaces it, not done.
func (i *Inventory) RecordTakeStarted(ctx context.Context, t TakenLibrary) error {
switch {
case t.Library == "" || t.Home == "" || !strings.HasPrefix(t.Root, "/"):
return errors.New("a taken library names itself, its home and its root")
case strings.TrimSpace(t.Warrant) == "" || strings.TrimSpace(t.Ask) == "":
return errors.New("a library is taken only on the operator's warrant, and this names none")
}
_, err := i.store.Pool().Exec(ctx,
`insert into kept_library (library, home, root, placed, taken_by, ask, warrant, said, done)
values ($1, $2, $3, $4, $5, $6, $7, $8, false)
on conflict (library) do update
set home = excluded.home, root = excluded.root, placed = excluded.placed, taken_at = now(),
taken_by = excluded.taken_by, ask = excluded.ask, warrant = excluded.warrant, said = excluded.said,
done = false
where kept_library.home <> excluded.home or kept_library.root <> excluded.root
or kept_library.placed <> excluded.placed`,
t.Library, t.Home, t.Root, t.Placed, t.By, t.Ask, t.Warrant, t.Said)
return err
}
// TakeDone keeps that the engine stated a library's take done under a warrant: the library is taken. A warrant the
// recorded take was not started under changes nothing.
func (i *Inventory) TakeDone(ctx context.Context, library, warrant string) error {
_, err := i.store.Pool().Exec(ctx,
`update kept_library set done = true where library = $1 and warrant = $2 and not done`, library, warrant)
return err
}
// TakeFailed forgets a take the engine stated failed under a warrant, never one done: the library is asked about again.
func (i *Inventory) TakeFailed(ctx context.Context, library, warrant string) error {
_, err := i.store.Pool().Exec(ctx,
`delete from kept_library where library = $1 and warrant = $2 and not done`, library, warrant)
return err
}
// SettingsAsOf is a module's settings layers on a machine as they stood at a moment: each layer — the whole mesh's,
// then the machine's — as it was set then, read from the layer standing now or from its history. What a module held at
// the build it ran is composed with: the settings it last ran with (novox/hq ADR 0328 §8).
func (i *Inventory) SettingsAsOf(ctx context.Context, nodeName, module string, at time.Time) ([]catalogue.Layer, error) {
node, err := i.NodeByName(ctx, nodeName)
if err != nil {
return nil, err
}
var layers []catalogue.Layer
for _, scope := range []struct {
id *string
from string
}{{nil, catalogue.MeshWideLayer}, {&node.ID, nodeName}} {
var raw []byte
err := i.store.Pool().QueryRow(ctx,
`select values from settings
where module = $1 and node is not distinct from $2::uuid and set_at <= $3
union all
select values from (select values from settings_history
where module = $1 and node is not distinct from $2::uuid
and (set_at is null or set_at <= $3) and replaced_at > $3
order by replaced_at asc limit 1) h
limit 1`, module, scope.id, at).Scan(&raw)
if errors.Is(err, pgx.ErrNoRows) {
continue
}
if err != nil {
return nil, err
}
values := map[string]any{}
if err := json.Unmarshal(raw, &values); err != nil {
return nil, err
}
layers = append(layers, catalogue.Layer{From: scope.from, Values: values})
}
return layers, nil
}