jschoubben
53d9c4de19
The verdict keeps only a failure's first line (issue 1921), so the trace that names the cause rode on lines it dropped.
344 lines
12 KiB
Go
344 lines
12 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"path"
|
|
"sort"
|
|
"strings"
|
|
|
|
sdkdecl "git.novox.be/novox/mesh-sdk/go/declaration"
|
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// The judgement of a move (novox/hq ADR 0539 clauses 2 and 14, by ADR 0377 §3's tests): beside each composition's digest
|
|
// the controller writes whether moving the node from the composition it runs, as its have names it, to this one
|
|
// recreates a container, moves the module's data, or neither, or that it cannot be judged. The assigner holds a move of
|
|
// a module of policy record from it until a person allows it.
|
|
//
|
|
// ADR 0377 §3 over two compositions of one module: **recreates** when a container resource is changed, added or
|
|
// removed; **moves data** when, with no container changed, a directory or file resource is changed, added or removed
|
|
// that a container of the module mounts or that holds or lies within one of its directories (where a module keeps its
|
|
// data), when its `data` declaration changes, or, for a module that declares data, when one of its long-running
|
|
// processes changes; **neither** otherwise, and for a module the node has never run. **Cannot judge** when the
|
|
// composition the node runs is gone or cannot be read. A held composition never becomes what is judged against: with A
|
|
// running, B held and C composed, C is judged against A, the one the have names.
|
|
|
|
// judgeMove is the value written beside a composition whose digest is digest: its judgement against what the node's have
|
|
// names for the module, and the module's policy.
|
|
func judgeMove(ctx context.Context, at composedPlace, key string, to sdkdelivery.Composed, digest string) (sdkdelivery.ComposedDigest, error) {
|
|
out := sdkdelivery.ComposedDigest{Digest: digest}
|
|
if at.policy != nil && to.Module != "" {
|
|
out.Policy, out.PolicyFrom, out.PolicyWhy = at.policy(to.Module)
|
|
}
|
|
if to.Module == "" {
|
|
// The node's own composition: no module's containers or data are judged; its want is written as any.
|
|
out.Judgement = sdkdelivery.JudgedNeither
|
|
return out, nil
|
|
}
|
|
have, found, err := at.ran(ctx, to.Module)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
// A node whose have names no composition yet (the switch: it runs the declaration it last applied) is judged
|
|
// against that module's slice of that declaration (novox/hq ADR 0540 clause 9): equal is unchanged, a want that
|
|
// changes nothing on the node; otherwise by ADR 0377 §3's tests, as any move.
|
|
if found && have.Composed == "" && have.FP != "" && have.Phase != sdkdelivery.PhaseRemoved && at.lastApplied != nil {
|
|
slice, declared, applied, err := at.lastApplied(ctx, to.Module)
|
|
if err != nil {
|
|
// Unread is not judged: a person's yes holds it, and the words say why.
|
|
out.Judgement = sdkdelivery.JudgedCannot
|
|
out.Words = fmt.Sprintf("%s's have on %s names no composition, and the declaration it last applied cannot be read: %v",
|
|
to.Module, at.node, err)
|
|
return out, nil
|
|
}
|
|
if applied {
|
|
from := sdkdelivery.Composed{Node: at.node, Module: to.Module, FP: have.FP, Data: to.Data, Resources: slice}
|
|
if sameResources(from, to) {
|
|
out.Judgement, out.Words = sdkdelivery.JudgedUnchanged, "equal to the declaration "+at.node+" last applied"
|
|
return out, nil
|
|
}
|
|
// Against names the declaration, by its digest: the composition the node runs is that declaration's slice.
|
|
out.Against = declared
|
|
if onlyTheUserList(from, to) {
|
|
// Only the bus's user list, as against a composition (ADR 0539 clause 7): no yes waits on it at the switch.
|
|
out.Judgement, out.UserListOnly = sdkdelivery.JudgedNeither, true
|
|
out.Words = "only the bus's user list changes, against the declaration " + at.node + " last applied"
|
|
return out, nil
|
|
}
|
|
j, words := judged(from, to)
|
|
out.Judgement, out.Words = j, words+", against the declaration "+at.node+" last applied"
|
|
return out, nil
|
|
}
|
|
}
|
|
if !found || have.Phase == sdkdelivery.PhaseRemoved || (have.Composed == "" && have.FP == "") {
|
|
out.Judgement, out.Words = sdkdelivery.JudgedNeither, "the node runs no composition of "+to.Module
|
|
return out, nil
|
|
}
|
|
if have.Composed == "" {
|
|
out.Judgement = sdkdelivery.JudgedCannot
|
|
out.Words = fmt.Sprintf("%s's have on %s names no composition, only the fingerprint %s", to.Module, at.node, have.FP)
|
|
return out, nil
|
|
}
|
|
out.Against = have.Composed
|
|
if have.Composed == digest {
|
|
out.Judgement, out.Words = sdkdelivery.JudgedNeither, "the node runs this composition"
|
|
return out, nil
|
|
}
|
|
from, found, err := composedByDigest(ctx, at, to.Module, have.FP, have.Composed)
|
|
if err != nil {
|
|
return out, err
|
|
}
|
|
if !found {
|
|
out.Judgement = sdkdelivery.JudgedCannot
|
|
out.Words = fmt.Sprintf("the composition %s runs on %s (%s) is gone", to.Module, at.node, short(strings.TrimPrefix(have.Composed, "sha256:")))
|
|
return out, nil
|
|
}
|
|
if onlyTheUserList(from, to) {
|
|
// Only the bus's user list (ADR 0539 clause 7): a file change and a reload of the bus, no move of it, written
|
|
// into its want at once.
|
|
out.Judgement, out.UserListOnly, out.Words = sdkdelivery.JudgedNeither, true, "only the bus's user list changes"
|
|
return out, nil
|
|
}
|
|
out.Judgement, out.Words = judged(from, to)
|
|
return out, nil
|
|
}
|
|
|
|
// busUsersSuffix ends the id of the bus module's user list as the composer writes it: `<module>.bus-users`.
|
|
const busUsersSuffix = ".bus-users"
|
|
|
|
// onlyTheUserList says two compositions of a module differ, and only in its file of the bus's user list.
|
|
func onlyTheUserList(from, to sdkdelivery.Composed) bool {
|
|
if from.Data != to.Data {
|
|
return false
|
|
}
|
|
before, after := resourcesByID(from), resourcesByID(to)
|
|
if len(before) != len(after) {
|
|
return false
|
|
}
|
|
changed := false
|
|
for id, now := range after {
|
|
was, ok := before[id]
|
|
if !ok {
|
|
return false
|
|
}
|
|
if bytes.Equal(was.raw, now.raw) {
|
|
continue
|
|
}
|
|
if !strings.HasSuffix(id, busUsersSuffix) || now.kind != string(sdkdecl.TypeFile) {
|
|
return false
|
|
}
|
|
changed = true
|
|
}
|
|
return changed
|
|
}
|
|
|
|
// composedByDigest finds, in the node's bucket, the composition of module at fingerprint fp whose digest is digest: the
|
|
// key's current value or one of the revisions it keeps.
|
|
func composedByDigest(ctx context.Context, at composedPlace, module, fp, digest string) (sdkdelivery.Composed, bool, error) {
|
|
if fp == "" {
|
|
return sdkdelivery.Composed{}, false, nil
|
|
}
|
|
key, err := sdkdelivery.ComposedKey(at.node, module, fp)
|
|
if err != nil {
|
|
return sdkdelivery.Composed{}, false, nil
|
|
}
|
|
values, err := historyOf(ctx, at.bucket, key)
|
|
if err != nil {
|
|
return sdkdelivery.Composed{}, false, err
|
|
}
|
|
for _, v := range values {
|
|
c, err := sdkdelivery.DecodeComposed(v)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
if d, err := sdkdelivery.ComposedDigestOf(c); err == nil && d == digest {
|
|
return c, true, nil
|
|
}
|
|
}
|
|
return sdkdelivery.Composed{}, false, nil
|
|
}
|
|
|
|
// historyOf is every value a key keeps, the newest first; none when there is no such key.
|
|
func historyOf(ctx context.Context, kv composedBucketsKV, key string) ([][]byte, error) {
|
|
if h, ok := kv.(interface {
|
|
History(ctx context.Context, key string, opts ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error)
|
|
}); ok {
|
|
entries, err := h.History(ctx, key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out := make([][]byte, 0, len(entries))
|
|
for i := len(entries) - 1; i >= 0; i-- {
|
|
if entries[i].Operation() == jetstream.KeyValuePut {
|
|
out = append(out, entries[i].Value())
|
|
}
|
|
}
|
|
return out, nil
|
|
}
|
|
e, err := kv.Get(ctx, key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return [][]byte{e.Value()}, nil
|
|
}
|
|
|
|
// judged applies ADR 0377 §3 to two compositions of one module.
|
|
func judged(from, to sdkdelivery.Composed) (sdkdelivery.Judgement, string) {
|
|
before, after := resourcesByID(from), resourcesByID(to)
|
|
if ids := changedOf(before, after, func(kind string) bool { return kind == string(sdkdecl.TypeContainer) }); len(ids) > 0 {
|
|
return sdkdelivery.JudgedRecreates, "it recreates " + strings.Join(ids, ", ")
|
|
}
|
|
if from.Data != to.Data {
|
|
return sdkdelivery.JudgedMovesData, "its data declaration changes"
|
|
}
|
|
places := dataPlaces(before, after)
|
|
held := func(kind string) bool {
|
|
return kind == string(sdkdecl.TypeDirectory) || kind == string(sdkdecl.TypeFile)
|
|
}
|
|
var moved []string
|
|
for _, id := range changedOf(before, after, held) {
|
|
r := before[id]
|
|
if r == nil {
|
|
r = after[id]
|
|
}
|
|
if withinPlaces(r.path, places) {
|
|
moved = append(moved, id)
|
|
}
|
|
}
|
|
if to.Data || from.Data {
|
|
moved = append(moved, changedOf(before, after, func(kind string) bool { return kind == string(sdkdecl.TypeProcess) })...)
|
|
}
|
|
if len(moved) > 0 {
|
|
sort.Strings(moved)
|
|
return sdkdelivery.JudgedMovesData, "it changes where its data is kept: " + strings.Join(moved, ", ")
|
|
}
|
|
return sdkdelivery.JudgedNeither, "it recreates no container and moves no data"
|
|
}
|
|
|
|
// judgedResource is a resource as the judgement reads it: its kind, the path it acts on, what its containers mount,
|
|
// and its canonical bytes.
|
|
type judgedResource struct {
|
|
kind string
|
|
path string
|
|
mounts []string
|
|
raw []byte
|
|
}
|
|
|
|
func resourcesByID(c sdkdelivery.Composed) map[string]*judgedResource {
|
|
out := map[string]*judgedResource{}
|
|
for _, raw := range c.Resources {
|
|
var head struct {
|
|
ID string `json:"id"`
|
|
Type string `json:"type"`
|
|
Path string `json:"path"`
|
|
// A container's volumes, `<path on the machine>:<path inside>[:<options>]`.
|
|
Volumes []string `json:"volumes"`
|
|
}
|
|
if err := json.Unmarshal(raw, &head); err != nil {
|
|
continue
|
|
}
|
|
r := &judgedResource{kind: head.Type, path: head.Path, raw: compactComposed(raw)}
|
|
for _, v := range head.Volumes {
|
|
source, _, _ := strings.Cut(v, ":")
|
|
r.mounts = append(r.mounts, source)
|
|
}
|
|
out[head.ID] = r
|
|
}
|
|
return out
|
|
}
|
|
|
|
// compactComposed is a resource's bytes canonical: its keys sorted, its spacing dropped, its numbers as written, so
|
|
// two compositions that say the same compare equal however each was written.
|
|
func compactComposed(raw []byte) []byte {
|
|
var v any
|
|
d := json.NewDecoder(bytes.NewReader(raw))
|
|
d.UseNumber()
|
|
if err := d.Decode(&v); err != nil {
|
|
return raw
|
|
}
|
|
out, err := json.Marshal(v)
|
|
if err != nil {
|
|
return raw
|
|
}
|
|
return out
|
|
}
|
|
|
|
// changedOf are the ids of the resources of the kinds counted that are changed, added or removed, sorted.
|
|
func changedOf(before, after map[string]*judgedResource, counted func(kind string) bool) []string {
|
|
var out []string
|
|
for id, now := range after {
|
|
if !counted(now.kind) {
|
|
continue
|
|
}
|
|
if was, ok := before[id]; !ok || !bytes.Equal(was.raw, now.raw) {
|
|
out = append(out, id)
|
|
}
|
|
}
|
|
for id, was := range before {
|
|
if _, ok := after[id]; !ok && counted(was.kind) {
|
|
out = append(out, id)
|
|
}
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
// dataPlaces are the paths where the module keeps what a move must not lose: what its containers mount, and its
|
|
// directories, before and after.
|
|
func dataPlaces(before, after map[string]*judgedResource) []string {
|
|
var out []string
|
|
for _, rs := range []map[string]*judgedResource{before, after} {
|
|
for _, r := range rs {
|
|
if r.kind == string(sdkdecl.TypeContainer) {
|
|
out = append(out, r.mounts...)
|
|
}
|
|
if r.kind == string(sdkdecl.TypeDirectory) && r.path != "" {
|
|
out = append(out, r.path)
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// withinPlaces says p is one of places, or lies within one, or holds one.
|
|
func withinPlaces(p string, places []string) bool {
|
|
if p == "" {
|
|
return false
|
|
}
|
|
p = path.Clean(p)
|
|
for _, place := range places {
|
|
if place == "" || !strings.HasPrefix(place, "/") {
|
|
continue
|
|
}
|
|
place = path.Clean(place)
|
|
if p == place || strings.HasPrefix(p, place+"/") || strings.HasPrefix(place, p+"/") {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// sameResources says two compositions of a module hold the same resources in the same order, each compared canonical.
|
|
func sameResources(a, b sdkdelivery.Composed) bool {
|
|
if len(a.Resources) != len(b.Resources) {
|
|
return false
|
|
}
|
|
for i := range a.Resources {
|
|
if !bytes.Equal(compactComposed(a.Resources[i]), compactComposed(b.Resources[i])) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|