mesh-controller/cmd/mesh-controller/composed_rejudge.go

package main

import (
	"context"
	"errors"
	"fmt"

	sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats.go/jetstream"
)

// The judgement beside each composition is rewritten when what it is judged against or with changes (novox/hq ADR
// 0539 clauses 2 and 14): when the node's have of the module changes (it runs another composition, or holds), and when
// a person sets the module's policy. Heard as the have bucket's writes and the `upgrade` verb, never on a timer.

// rejudge writes again the judgement and policy beside every composition of module on the node at, where they differ.
func rejudge(ctx context.Context, at composedPlace, module string) (int, error) {
	lister, err := at.bucket.ListKeys(ctx)
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return 0, nil
	}
	if err != nil {
		return 0, err
	}
	wrote := 0
	var errs []error
	for key := range lister.Keys() {
		node, m, _, isNode, err := sdkdelivery.ParseComposedKey(key)
		if err != nil || isNode || node != at.node || m != module {
			continue
		}
		c, revision, err := readComposed(ctx, at.bucket, key)
		if err != nil || revision == 0 {
			continue
		}
		c.Signature = nil
		digest, err := sdkdelivery.ComposedDigestOf(c)
		if err != nil {
			continue
		}
		judged, err := judgeMove(ctx, at, key, c, digest)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		w, err := keepDigest(ctx, at.digests, key, judged)
		if err != nil {
			errs = append(errs, err)
		}
		if w {
			wrote++
		}
	}
	return wrote, errors.Join(errs...)
}

// rejudgeOn opens the node's buckets and rejudges module there, under the lease.
func rejudgeOn(ctx context.Context, open *stores, nodes []string, module string) error {
	if _, err := theLease.epoch(ctx); err != nil {
		return err
	}
	bus, err := aBus()
	if err != nil {
		return err
	}
	defer bus.Close()
	js, err := jetstream.New(bus.Conn())
	if err != nil {
		return err
	}
	upgrades, err := open.inventory.Upgrades(ctx)
	if err != nil {
		return err
	}
	var errs []error
	for _, node := range nodes {
		at, err := openComposedPlace(ctx, js, bus.Conn(), node)
		if err != nil {
			errs = append(errs, err)
			continue
		}
		at.policy = func(m string) (string, string, string) {
			u, ok := upgrades[m]
			if !ok {
				return "", "", ""
			}
			return u.Policy(), u.From, u.Why
		}
		if _, err := rejudge(ctx, at, module); err != nil {
			errs = append(errs, err)
		}
	}
	return errors.Join(errs...)
}

// rejudgeModule rejudges a module on every node it is assigned to: what a person's policy changes.
func rejudgeModule(ctx context.Context, open *stores, module string) {
	entries, err := open.inventory.Catalogued(ctx)
	if err != nil {
		fmt.Printf("compositions: %s's policy is not written beside its compositions: %v\n", module, err)
		return
	}
	for _, e := range entries {
		if e.Manifest.Module == module && len(e.On) > 0 {
			if err := rejudgeOn(ctx, open, e.On, module); err != nil {
				fmt.Printf("compositions: %s's policy is not written beside its compositions: %v\n", module, err)
			}
		}
	}
}

// watchHavesForJudgement rejudges a node's module each time its have changes, until ctx ends: never on the watch's own
// goroutine.
func watchHavesForJudgement(ctx context.Context, open *stores) {
	bus, err := aBus()
	if err != nil {
		fmt.Printf("compositions: the haves are not watched: %v\n", err)
		return
	}
	js, err := jetstream.New(bus.Conn())
	if err != nil {
		bus.Close()
		return
	}
	kv, err := js.KeyValue(ctx, sdkdelivery.HaveBucket)
	if err != nil {
		bus.Close()
		fmt.Printf("compositions: the haves are not watched: %v\n", err)
		return
	}
	w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
	if err != nil {
		bus.Close()
		fmt.Printf("compositions: the haves are not watched: %v\n", err)
		return
	}
	go func() {
		defer bus.Close()
		defer w.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case e, ok := <-w.Updates():
				if !ok {
					return
				}
				if e == nil || e.Operation() != jetstream.KeyValuePut {
					continue
				}
				node, module, isNode, err := sdkdelivery.ParseHaveKey(e.Key())
				if err != nil || isNode {
					continue
				}
				inBackground(func() {
					if err := rejudgeOn(ctx, open, []string{node}, module); err != nil {
						fmt.Printf("compositions: %s on %s is not judged again: %v\n", module, node, err)
					}
					// A node's node-engine moved: the declarations' stream goes once every node runs the new one.
					if module == nodeEngineModule {
						if _, err := retireDeclarations(ctx, open); err != nil {
							fmt.Printf("compositions: whether the declarations' stream may go cannot be read: %v\n", err)
						}
					}
				})
			}
		}
	}()
}