The engine names the filesystem a waiting folder is on; a warrant is recorded for that filesystem, and a folder on another is asked about again and not taken there meanwhile. Entries the engine could not take out while their disk was away, stated before a send, do not hold that delivery's gate. mesh-host pinned to d2d9fbbd.
128 lines
4.7 KiB
Go
128 lines
4.7 KiB
Go
package inventory
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"path/filepath"
|
|
"strings"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
|
|
"github.com/novox/mesh-controller/internal/catalogue"
|
|
)
|
|
|
|
// Module identities and taken folders (novox/hq ADR 0325 §3, §8, §11; migration 0096).
|
|
|
|
// AllocateIdentity is module's identity, allocated now when it has none: one number from the range of its own,
|
|
// mesh-wide, kept for ever. 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("an identity is allocated for a module, and none was named")
|
|
}
|
|
if id, had, err := i.IdentityOf(ctx, module); err != nil || had {
|
|
return id, err
|
|
}
|
|
// The number is 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.
|
|
_, err := i.store.Pool().Exec(ctx,
|
|
`insert into module_identity (module, uid, gid)
|
|
select $1, n, n from (select nextval('module_identity_number')::integer as n) drawn
|
|
on conflict (module) do nothing`, module)
|
|
if err != nil {
|
|
return catalogue.ModuleIdentity{}, fmt.Errorf("%s could not be given an identity: %w", module, err)
|
|
}
|
|
id, had, err := i.IdentityOf(ctx, module)
|
|
if err == nil && !had {
|
|
err = fmt.Errorf("%s was given an identity and it cannot be read back", module)
|
|
}
|
|
return id, err
|
|
}
|
|
|
|
// IdentityOf is module's identity, 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, gid from module_identity where module = $1`, module).
|
|
Scan(&id.UID, &id.GID)
|
|
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 module's identity, by module.
|
|
func (i *Inventory) Identities(ctx context.Context) (map[string]catalogue.ModuleIdentity, error) {
|
|
rows, err := i.store.Pool().Query(ctx, `select module, uid, gid 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, &id.GID); err != nil {
|
|
return nil, err
|
|
}
|
|
out[id.Module] = id
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// TakenFolder is one folder the operator let the mesh keep on a machine (ADR 0325 §8).
|
|
type TakenFolder struct {
|
|
Path string
|
|
By string
|
|
Ask string
|
|
Warrant string
|
|
Said string
|
|
// Placed is the filesystem the folder was on when taken, without its path (ADR 0325 §8).
|
|
Placed string
|
|
}
|
|
|
|
// KeptFolders is every folder taken on a machine, by its clean path.
|
|
func (i *Inventory) KeptFolders(ctx context.Context, nodeName string) (map[string]TakenFolder, error) {
|
|
rows, err := i.store.Pool().Query(ctx,
|
|
`select k.path, k.taken_by, k.ask, k.warrant, k.said, k.placed from kept_folder k join node n on n.id = k.node
|
|
where n.name = $1`, nodeName)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := map[string]TakenFolder{}
|
|
for rows.Next() {
|
|
var t TakenFolder
|
|
if err := rows.Scan(&t.Path, &t.By, &t.Ask, &t.Warrant, &t.Said, &t.Placed); err != nil {
|
|
return nil, err
|
|
}
|
|
out[t.Path] = t
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// RecordTaken keeps that a folder was taken on a machine, under the operator's warrant. A take is recorded once per
|
|
// filesystem: a second for the same folder on the same filesystem keeps the first, and one for the folder on another
|
|
// filesystem (it moved, and the operator was asked again) replaces it.
|
|
func (i *Inventory) RecordTaken(ctx context.Context, nodeName string, t TakenFolder) error {
|
|
switch {
|
|
case !strings.HasPrefix(t.Path, "/"):
|
|
return fmt.Errorf("a taken folder is an absolute path, not %q", t.Path)
|
|
case strings.TrimSpace(t.Warrant) == "" || strings.TrimSpace(t.Ask) == "":
|
|
return errors.New("a folder is taken only on the operator's warrant, and this names none")
|
|
}
|
|
node, err := i.NodeByName(ctx, nodeName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = i.store.Pool().Exec(ctx,
|
|
`insert into kept_folder (node, path, taken_by, ask, warrant, said, placed) values ($1, $2, $3, $4, $5, $6, $7)
|
|
on conflict (node, path) do update
|
|
set taken_at = now(), taken_by = excluded.taken_by, ask = excluded.ask, warrant = excluded.warrant,
|
|
said = excluded.said, placed = excluded.placed
|
|
where kept_folder.placed is distinct from excluded.placed`, node.ID, filepath.Clean(t.Path), t.By, t.Ask,
|
|
t.Warrant, t.Said, t.Placed)
|
|
return err
|
|
}
|