jochen
5efe999733
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 superseded: a newer delivery to the same trunk took over its walk
345 lines
12 KiB
Go
345 lines
12 KiB
Go
package inventory
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
|
|
"github.com/novox/mesh-controller/internal/catalogue"
|
|
)
|
|
|
|
// The gate's verdicts (novox/hq ADR 0236, to-be 45 §8): what a build did on its first machine, and,
|
|
// for one that failed there, how it was put back. One row per build, written by the plan that rolled it
|
|
// out, under the lease.
|
|
|
|
// The gate's verdicts and a failed build's rollback.
|
|
const (
|
|
GatePassed = "passed"
|
|
GateFailed = "failed"
|
|
|
|
RollingBack = "rolling-back"
|
|
RolledBack = "rolled-back"
|
|
NotRolledBack = "not-rolled-back"
|
|
)
|
|
|
|
// GateVerdict is one build's verdict at its gate.
|
|
type GateVerdict struct {
|
|
Build string
|
|
Module string
|
|
Commit string
|
|
Previous string
|
|
Plan string
|
|
Machines []string
|
|
Verdict string
|
|
Rollback string
|
|
Why string
|
|
Component string
|
|
// JudgingFrom is when the first machine reported the build applied and the judging began.
|
|
JudgingFrom *time.Time
|
|
JudgedAt time.Time
|
|
Epoch uint64
|
|
}
|
|
|
|
// RecordGate writes a build's verdict. A build that passed on one machine may still fail on the next one
|
|
// judged; **a failed build's row is written once**: any later verdict for it is refused with ErrGateKept, which is what keeps a rollback to one per build — the row
|
|
// is written before the rollback's send, and a controller replaced in between finds it.
|
|
func (i *Inventory) RecordGate(ctx context.Context, v GateVerdict) error {
|
|
epoch, err := i.actingEpoch(ctx)
|
|
if err != nil {
|
|
return fmt.Errorf("the gate's verdict on %s is not written: %w", v.Build, err)
|
|
}
|
|
if v.Machines == nil {
|
|
v.Machines = []string{}
|
|
}
|
|
tag, err := i.store.Pool().Exec(ctx,
|
|
`insert into build_gate (build, module, commit_hash, previous, plan, machines, verdict, rollback, why,
|
|
component, judging_from, judged_at, epoch)
|
|
values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, now(), $12)
|
|
on conflict (build) do update set verdict = excluded.verdict, rollback = excluded.rollback,
|
|
why = excluded.why, previous = excluded.previous, machines = excluded.machines,
|
|
judged_at = now(), epoch = excluded.epoch
|
|
where build_gate.verdict = 'passed'`,
|
|
v.Build, v.Module, v.Commit, v.Previous, v.Plan, v.Machines, v.Verdict, v.Rollback, v.Why, v.Component,
|
|
v.JudgingFrom, epoch)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return fmt.Errorf("%w: %s", ErrGateKept, v.Build)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ErrGateKept is a verdict already kept for the build, which is not written over.
|
|
var ErrGateKept = errors.New("this build's verdict at its gate is already kept")
|
|
|
|
// SetRollback records how a failed build's rollback went.
|
|
func (i *Inventory) SetRollback(ctx context.Context, build, rollback, why string) error {
|
|
if _, err := i.actingEpoch(ctx); err != nil {
|
|
return err
|
|
}
|
|
_, err := i.store.Pool().Exec(ctx,
|
|
`update build_gate set rollback = $2, why = $3, judged_at = now() where build = $1 and verdict = 'failed'`,
|
|
build, rollback, why)
|
|
return err
|
|
}
|
|
|
|
// GateOf is a build's verdict, and whether it has one.
|
|
func (i *Inventory) GateOf(ctx context.Context, build string) (GateVerdict, bool, error) {
|
|
rows, err := i.store.Pool().Query(ctx, gateSelect+` where build = $1`, build)
|
|
if err != nil {
|
|
return GateVerdict{}, false, err
|
|
}
|
|
list, err := scanGates(rows)
|
|
if err != nil || len(list) == 0 {
|
|
return GateVerdict{}, false, err
|
|
}
|
|
return list[0], true, nil
|
|
}
|
|
|
|
// GateFailed is whether a build failed its gate: one the mesh never registers or sends again on its own.
|
|
func (i *Inventory) GateFailed(ctx context.Context, build string) (bool, error) {
|
|
v, found, err := i.GateOf(ctx, build)
|
|
return found && v.Verdict == GateFailed, err
|
|
}
|
|
|
|
// LatestGates is the newest verdict of every module that has one: what the gate probe (DG) reads.
|
|
func (i *Inventory) LatestGates(ctx context.Context) ([]GateVerdict, error) {
|
|
rows, err := i.store.Pool().Query(ctx, `select distinct on (module) build, module, commit_hash, previous, plan,
|
|
machines, verdict, rollback, why, component, judging_from, judged_at, coalesce(epoch, 0)
|
|
from build_gate order by module, judged_at desc`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return scanGates(rows)
|
|
}
|
|
|
|
// Gates is the newest verdicts, newest first: what `plans gates` lists.
|
|
func (i *Inventory) Gates(ctx context.Context, limit int) ([]GateVerdict, error) {
|
|
rows, err := i.store.Pool().Query(ctx, gateSelect+` order by judged_at desc limit $1`, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return scanGates(rows)
|
|
}
|
|
|
|
const gateSelect = `select build, module, commit_hash, previous, plan, machines, verdict, rollback, why, component,
|
|
judging_from, judged_at, coalesce(epoch, 0) from build_gate`
|
|
|
|
func scanGates(rows pgx.Rows) ([]GateVerdict, error) {
|
|
defer rows.Close()
|
|
var out []GateVerdict
|
|
for rows.Next() {
|
|
var v GateVerdict
|
|
var epoch int64
|
|
if err := rows.Scan(&v.Build, &v.Module, &v.Commit, &v.Previous, &v.Plan, &v.Machines, &v.Verdict,
|
|
&v.Rollback, &v.Why, &v.Component, &v.JudgingFrom, &v.JudgedAt, &epoch); err != nil {
|
|
return nil, err
|
|
}
|
|
v.Epoch = uint64(epoch)
|
|
out = append(out, v)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// PreviousBuild is the build a module goes back to when a build of it fails its gate: the newest build
|
|
// that worked, made from the commit the first machine ran before, asked before the failed one, and not
|
|
// itself failed at a gate. Among the builds whose artifacts the mesh keeps (KeptBuilds): an older one
|
|
// may already be gone from the artifact store. False when there is none to go back to.
|
|
func (i *Inventory) PreviousBuild(ctx context.Context, module, commit string, failed Build) (Build, bool, error) {
|
|
builds, err := i.Builds(ctx, module, 50)
|
|
if err != nil {
|
|
return Build{}, false, err
|
|
}
|
|
kept := 0
|
|
for _, b := range builds {
|
|
if !b.Worked() {
|
|
continue
|
|
}
|
|
kept++
|
|
if kept > KeptBuilds {
|
|
break
|
|
}
|
|
if b.ID == failed.ID || (commit != "" && b.Commit != commit) {
|
|
continue
|
|
}
|
|
if !failed.AskedOrAt().IsZero() && !b.AskedOrAt().Before(failed.AskedOrAt()) {
|
|
continue
|
|
}
|
|
if bad, err := i.GateFailed(ctx, b.ID); err != nil {
|
|
return Build{}, false, err
|
|
} else if bad {
|
|
continue
|
|
}
|
|
var manifest []byte
|
|
if err := i.store.Pool().QueryRow(ctx, `select manifest from build where id = $1`, b.ID).Scan(&manifest); err != nil {
|
|
return Build{}, false, err
|
|
}
|
|
if len(manifest) == 0 || string(manifest) == "null" {
|
|
continue
|
|
}
|
|
b.Manifest = manifest
|
|
return b, true, nil
|
|
}
|
|
return Build{}, false, nil
|
|
}
|
|
|
|
// RestoreModule puts a module's registered build back to an earlier one: its manifest, the commit it
|
|
// was built from, and when it was asked — as now, so the build that failed its gate, asked before, can
|
|
// never register over it again (issue 219's order). The source's head is left where the merge moved it:
|
|
// the module IS behind its source, and `status` says so.
|
|
func (i *Inventory) RestoreModule(ctx context.Context, b Build) error {
|
|
if _, err := i.actingEpoch(ctx); err != nil {
|
|
return fmt.Errorf("%s is not put back: %w", b.Module, err)
|
|
}
|
|
m, err := catalogue.ParseManifest(b.Manifest)
|
|
if err != nil {
|
|
return fmt.Errorf("%s's build %s is not a manifest the mesh can register again: %w", b.Module, b.ID, err)
|
|
}
|
|
raw, err := json.Marshal(m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
tag, err := i.store.Pool().Exec(ctx,
|
|
`update module set manifest = $2, version = nullif($3, ''), built_from = nullif($4, ''),
|
|
built_asked = now(), registered = now()
|
|
where name = $1`, b.Module, raw, m.Version, b.Commit)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return fmt.Errorf("%w: %s", ErrNoSuchModule, b.Module)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// BuildFingerprints is, per commit, what the newest successful build of a module from it would put on a
|
|
// machine — its artifacts and its manifest, hashed — so a rebuild that changes nothing there is told
|
|
// from one that does (the bus's planned step, ADR 0236).
|
|
func (i *Inventory) BuildFingerprints(ctx context.Context, module string) (map[string]string, error) {
|
|
rows, err := i.store.Pool().Query(ctx,
|
|
`select commit_hash, made, coalesce(manifest::text, '') from build
|
|
where module = $1 and failed = '' and commit_hash <> '' order by at desc limit 50`, module)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := map[string]string{}
|
|
for rows.Next() {
|
|
var commit, manifest string
|
|
var made []byte
|
|
if err := rows.Scan(&commit, &made, &manifest); err != nil {
|
|
return nil, err
|
|
}
|
|
if _, seen := out[commit]; seen {
|
|
continue
|
|
}
|
|
sum := sha256.Sum256(append(append(made, 0), []byte(manifest)...))
|
|
out[commit] = hex.EncodeToString(sum[:])
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// Fingerprints is BuildFingerprints for every module at once: module → commit → fingerprint.
|
|
func (i *Inventory) Fingerprints(ctx context.Context) (map[string]map[string]string, error) {
|
|
rows, err := i.store.Pool().Query(ctx,
|
|
`select module, commit_hash, made, coalesce(manifest::text, '') from build
|
|
where module is not null and failed = '' and commit_hash <> '' order by at desc`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := map[string]map[string]string{}
|
|
for rows.Next() {
|
|
var module, commit, manifest string
|
|
var made []byte
|
|
if err := rows.Scan(&module, &commit, &made, &manifest); err != nil {
|
|
return nil, err
|
|
}
|
|
if out[module] == nil {
|
|
out[module] = map[string]string{}
|
|
}
|
|
if _, seen := out[module][commit]; seen {
|
|
continue
|
|
}
|
|
sum := sha256.Sum256(append(append(made, 0), []byte(manifest)...))
|
|
out[module][commit] = hex.EncodeToString(sum[:])
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// PassedCommits is, per module, the commits a build of which passed its gate on some machine.
|
|
func (i *Inventory) PassedCommits(ctx context.Context) (map[string]map[string]bool, error) {
|
|
rows, err := i.store.Pool().Query(ctx, `select module, commit_hash from build_gate where verdict = 'passed'`)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := map[string]map[string]bool{}
|
|
for rows.Next() {
|
|
var module, commit string
|
|
if err := rows.Scan(&module, &commit); err != nil {
|
|
return nil, err
|
|
}
|
|
if out[module] == nil {
|
|
out[module] = map[string]bool{}
|
|
}
|
|
out[module][commit] = true
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// BuildOf is the id of the newest successful build of a module from a commit; empty when none is
|
|
// recorded (a manifest handed over by hand).
|
|
func (i *Inventory) BuildOf(ctx context.Context, module, commit string) (string, error) {
|
|
var id string
|
|
err := i.store.Pool().QueryRow(ctx,
|
|
`select id from build where module = $1 and commit_hash = $2 and failed = '' order by at desc limit 1`,
|
|
module, commit).Scan(&id)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return "", nil
|
|
}
|
|
return id, err
|
|
}
|
|
|
|
// ManifestAt is the manifest a module was registered with at a commit (or a commit it abbreviates): the
|
|
// newest successful build from it, with the artifacts of the build standing for it when its source was
|
|
// unchanged (issue 280) — what a machine last sent that build runs. False when no such build, or none
|
|
// with a manifest, is kept (novox/hq issue 295: a recorded module is composed at the build its machine
|
|
// runs until a person's push moves it).
|
|
func (i *Inventory) ManifestAt(ctx context.Context, module, commit string) (catalogue.Manifest, bool, error) {
|
|
if commit == "" {
|
|
return catalogue.Manifest{}, false, nil
|
|
}
|
|
var id string
|
|
var raw []byte
|
|
err := i.store.Pool().QueryRow(ctx,
|
|
`select id, manifest from build
|
|
where module = $1 and failed = '' and manifest is not null and manifest::text <> 'null'
|
|
and (commit_hash = $2 or starts_with(commit_hash, $2))
|
|
order by at desc limit 1`, module, commit).Scan(&id, &raw)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return catalogue.Manifest{}, false, nil
|
|
}
|
|
if err != nil {
|
|
return catalogue.Manifest{}, false, err
|
|
}
|
|
if stands, same, err := i.StandingBuild(ctx, module, id); err != nil {
|
|
return catalogue.Manifest{}, false, err
|
|
} else if stands != "" && stands != id && len(same) > 0 {
|
|
raw = same
|
|
}
|
|
m, err := catalogue.ParseManifest(raw)
|
|
if err != nil {
|
|
return catalogue.Manifest{}, false, fmt.Errorf("%s's build from %s is not a manifest the mesh can compose: %w",
|
|
module, commit, err)
|
|
}
|
|
return m, true, nil
|
|
}
|