Write each module's composition per node, signed, with its digest and the move's judgement beside it (issue 1945, B1)
Every act that sent a declaration now writes compositions instead, by compare-and-set under the lease, and a module's compositions are kept for its five newest fingerprints and whatever its want or have names (ADR 0539 clauses 1 to 5 and 14). A module the mesh provides itself, which no want can name, rides in the node's own.
This commit is contained in:
@@ -0,0 +1,198 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// composeModules composes each named node and writes its compositions that differ (ADR 0539 clause 5): what each act
|
||||
// that once sent a declaration calls instead. A node that cannot compose is that node's: the others are written, and
|
||||
// every refusal is named in the error. It answers the keys it wrote, by node. Under the lease only: a controller that
|
||||
// does not hold it writes nothing and says so.
|
||||
func composeModules(ctx context.Context, open *stores, nodes []string) (map[string][]string, error) {
|
||||
if _, err := theLease.epoch(ctx); err != nil {
|
||||
return nil, fmt.Errorf("nothing is composed: %w", err)
|
||||
}
|
||||
bus, err := aBus()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("nothing is composed: the bus: %w", err)
|
||||
}
|
||||
defer bus.Close()
|
||||
js, err := jetstream.New(bus.Conn())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
sign, err := controllerSigner(ctx, open)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("nothing is composed: the controller's signing key: %w", err)
|
||||
}
|
||||
fps, err := currentFingerprints(ctx, open.inventory)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entries, err := open.inventory.Catalogued(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
manifests := make(map[string]catalogue.Manifest, len(entries))
|
||||
for _, e := range entries {
|
||||
manifests[e.Manifest.Module] = e.Manifest
|
||||
}
|
||||
upgrades, err := open.inventory.Upgrades(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
policy := func(module string) (string, string, string) {
|
||||
u, ok := upgrades[module]
|
||||
if !ok {
|
||||
return "", "", ""
|
||||
}
|
||||
return u.Policy(), u.From, u.Why
|
||||
}
|
||||
sort.Strings(nodes)
|
||||
out := map[string][]string{}
|
||||
var errs []error
|
||||
for _, node := range nodes {
|
||||
at, err := openComposedPlace(ctx, js, bus.Conn(), node)
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s: %w", node, err))
|
||||
continue
|
||||
}
|
||||
at.sign, at.policy = sign, policy
|
||||
declared, err := composeSend(ctx, open, node)
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s: %w", node, err))
|
||||
continue
|
||||
}
|
||||
body, err := declared.Body()
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s: %w", node, err))
|
||||
continue
|
||||
}
|
||||
comps, moved, err := compositionsOf(node, body, fps, manifests)
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s: %w", node, err))
|
||||
continue
|
||||
}
|
||||
if len(moved) > 0 {
|
||||
fmt.Printf("compositions: %s's %s compose with the node's own: no fingerprint of the new form names them\n",
|
||||
node, strings.Join(moved, ", "))
|
||||
}
|
||||
wrote, err := writeNodeCompositions(ctx, at, comps)
|
||||
if len(wrote) > 0 {
|
||||
out[node] = wrote
|
||||
fmt.Printf("compositions: %s: wrote %s\n", node, strings.Join(wrote, ", "))
|
||||
}
|
||||
if err != nil {
|
||||
errs = append(errs, fmt.Errorf("%s: %w", node, err))
|
||||
}
|
||||
}
|
||||
return out, errors.Join(errs...)
|
||||
}
|
||||
|
||||
// composeAndSay is composeModules for an act that has nothing to do with its answer but say it: what each former send
|
||||
// of a verb or an event calls.
|
||||
func composeAndSay(ctx context.Context, open *stores, nodes []string) error {
|
||||
if len(nodes) == 0 {
|
||||
return nil
|
||||
}
|
||||
_, err := composeModules(ctx, open, nodes)
|
||||
if err != nil {
|
||||
fmt.Printf("compositions: %v\n", err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// openComposedPlace opens a node's buckets: its compositions, the digests, its files, and reads its want and have.
|
||||
func openComposedPlace(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, node string) (composedPlace, error) {
|
||||
if err := broker.EnsureComposed(ctx, js, node); err != nil {
|
||||
return composedPlace{}, err
|
||||
}
|
||||
name, err := sdkdelivery.ComposedBucketFor(node)
|
||||
if err != nil {
|
||||
return composedPlace{}, err
|
||||
}
|
||||
bucket, err := js.KeyValue(ctx, name)
|
||||
if err != nil {
|
||||
return composedPlace{}, fmt.Errorf("the bucket %s: %w", name, err)
|
||||
}
|
||||
digests, err := js.KeyValue(ctx, sdkdelivery.ControllerComposedBucket)
|
||||
if err != nil {
|
||||
return composedPlace{}, fmt.Errorf("the bucket %s: %w", sdkdelivery.ControllerComposedBucket, err)
|
||||
}
|
||||
objects, err := openBucket(ctx, conn, node)
|
||||
if err != nil {
|
||||
return composedPlace{}, err
|
||||
}
|
||||
return composedPlace{node: node, bucket: bucket, digests: digests, objects: objects,
|
||||
wanted: func(ctx context.Context, module string) (string, error) { return wantedOn(ctx, js, node, module) },
|
||||
ran: func(ctx context.Context, module string) (sdkdelivery.Have, bool, error) {
|
||||
return haveOn(ctx, js, node, module)
|
||||
},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// wantedOn is the fingerprint the assigner's want names for a module on a node; "" when there is none or the bucket is
|
||||
// not there yet.
|
||||
func wantedOn(ctx context.Context, js jetstream.JetStream, node, module string) (string, error) {
|
||||
kv, err := js.KeyValue(ctx, sdkdelivery.AssignerWantsBucket)
|
||||
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
||||
return "", nil
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
key, err := sdkdelivery.WantKey(node, module)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
e, err := kv.Get(ctx, key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return "", nil
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
w, err := sdkdelivery.DecodeWant(e.Value())
|
||||
if err != nil {
|
||||
return "", nil // a want of release 5, naming no composition: no fingerprint a composition must be kept for
|
||||
}
|
||||
return w.FP, nil
|
||||
}
|
||||
|
||||
// haveOn is what a node's node-engine says it runs of a module.
|
||||
func haveOn(ctx context.Context, js jetstream.JetStream, node, module string) (sdkdelivery.Have, bool, error) {
|
||||
kv, err := js.KeyValue(ctx, sdkdelivery.HaveBucket)
|
||||
if errors.Is(err, jetstream.ErrBucketNotFound) {
|
||||
return sdkdelivery.Have{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return sdkdelivery.Have{}, false, err
|
||||
}
|
||||
key, err := sdkdelivery.HaveKey(node, module)
|
||||
if err != nil {
|
||||
return sdkdelivery.Have{}, false, err
|
||||
}
|
||||
e, err := kv.Get(ctx, key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return sdkdelivery.Have{}, false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return sdkdelivery.Have{}, false, err
|
||||
}
|
||||
h, err := sdkdelivery.DecodeHave(e.Value())
|
||||
if err != nil {
|
||||
return sdkdelivery.Have{}, false, nil
|
||||
}
|
||||
return h, true, nil
|
||||
}
|
||||
Reference in New Issue
Block a user