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 } 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: `.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, `:[:]`. 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 }