mesh-controller/internal/broker/composed.go

package broker

import (
	"context"
	"fmt"

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

// The compositions' buckets (novox/hq ADR 0539 clauses 2 and 4): one per node, `composed-<node>`, holding that node's
// compositions, which only that node's node-engine reads, since a composition can carry a sealed value (ADR 0488 §2's
// reason); and `mesh-controller_composed`, holding each composition's digest and the judgement of its move, which the
// assigner reads. The controller writes both, by compare-and-set while it holds the lease; each keeps five revisions
// of a key. The names and layouts are mesh-sdk's (delivery.ComposedBucketFor, ControllerComposedBucket, ComposedKey).

// ControllerComposed is the local name of the compositions' digests, as a reader names it in `reads` after
// `mesh-controller.`: delivery.ControllerComposedBucket.
const ControllerComposed = "composed"

// ComposedValueBytes is the most a composition may hold: a file over 64 KiB travels by digest, so a module's
// composition is far smaller; just under the server's largest message, which also carries the subject and headers.
const ComposedValueBytes = 1<<20 - 8<<10

// ComposedBucketBytes is a node's composition bucket's size: five revisions of every module's compositions of the node.
const ComposedBucketBytes = 256 << 20

// ComposedDigestsBytes is the size of the bucket of digests: a value of a few hundred bytes per node, module and
// fingerprint kept.
const ComposedDigestsBytes = 64 << 20

// ComposedConfig is a node's composition bucket as the controller asserts it.
func ComposedConfig(node string) (jetstream.KeyValueConfig, error) {
	bucket, err := delivery.ComposedBucketFor(node)
	if err != nil {
		return jetstream.KeyValueConfig{}, err
	}
	return jetstream.KeyValueConfig{
		Bucket: bucket,
		Description: "the compositions of " + node + " (novox/hq ADR 0539): each module's resources for this node at one " +
			"fingerprint, and the node's own, signed; written by the controller alone, read by this node's node-engine " +
			"alone; deleted when outside the module's five newest fingerprints and named by no want",
		History:      delivery.ComposedHistory,
		MaxValueSize: ComposedValueBytes,
		MaxBytes:     ComposedBucketBytes,
		Storage:      jetstream.FileStorage,
	}, nil
}

// ComposedDigestsConfig is the bucket of the compositions' digests as the controller asserts it.
func ComposedDigestsConfig() jetstream.KeyValueConfig {
	return jetstream.KeyValueConfig{
		Bucket: delivery.ControllerComposedBucket,
		Description: "each composition's digest and the judgement of its move from what the node runs (novox/hq ADR 0539 " +
			"clauses 2 and 14): written by the controller alone, read by the assigner",
		History:      delivery.ControllerComposedHistory,
		MaxValueSize: 4 << 10,
		MaxBytes:     ComposedDigestsBytes,
		Storage:      jetstream.FileStorage,
	}
}

// EnsureComposed creates the bucket of digests and each named node's composition bucket if absent and brings their
// options to match. An update, never a delete.
func EnsureComposed(ctx context.Context, js jetstream.JetStream, nodes ...string) error {
	if _, err := js.CreateOrUpdateKeyValue(ctx, ComposedDigestsConfig()); err != nil {
		return fmt.Errorf("asserting bucket %s: %w", delivery.ControllerComposedBucket, err)
	}
	for _, node := range nodes {
		cfg, err := ComposedConfig(node)
		if err != nil {
			return err
		}
		if _, err := js.CreateOrUpdateKeyValue(ctx, cfg); err != nil {
			return fmt.Errorf("asserting bucket %s: %w", cfg.Bucket, err)
		}
	}
	return nil
}

// ComposedWrites is what the controller publishes to write a node's compositions: the bucket's own subjects.
func ComposedWrites(node string) (string, error) {
	bucket, err := delivery.ComposedBucketFor(node)
	if err != nil {
		return "", err
	}
	return "$KV." + bucket + ".>", nil
}

// ComposedReads is what a node's node-engine publishes to follow its wants and compositions (ADR 0539 clauses 1, 2 and
// 9): every key of its own composition bucket, and its own wants (`want.<node>.<module>` and `want.<node>.node`) in the
// assigner's bucket; nothing of another node's, nor the bucket of digests.
func ComposedReads(node string) ([]string, error) {
	bucket, err := delivery.ComposedBucketFor(node)
	if err != nil {
		return nil, err
	}
	if !safeSubject.MatchString(node) {
		return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node)
	}
	return append(readGrants(bucket, ""), readGrants(delivery.AssignerWantsBucket, "want."+node+".>")...), nil
}

// VerbsTheNodeAgentAsks are the seat verbs a node's node-engine calls of its own machine while it applies a want (ADR
// 0539 clause 16, ADR 0429 §14): its backup holder backs a data module up before it moves, and says when it was. Asked
// of its own machine only, `mesh.seat.<seat>.tool.<verb>.<node>`.
var VerbsTheNodeAgentAsks = []SeatVerb{{Seat: "node-backup", Verb: "now"}, {Seat: "node-backup", Verb: "backed-up"}}

// NodeAgentAsks are the subjects of VerbsTheNodeAgentAsks for one node.
func NodeAgentAsks(node string) ([]string, error) {
	if !safeSubject.MatchString(node) {
		return nil, fmt.Errorf("%q cannot be part of a subject: a permission is a subject pattern, and this would widen it", node)
	}
	var out []string
	for _, v := range VerbsTheNodeAgentAsks {
		out = append(out, "mesh.seat."+v.Seat+".tool."+v.Verb+"."+node)
	}
	return out, nil
}

// NodeAgentHears are the events a node's node-engine subscribes while it applies a want: the backup holder's word that a
// backup is taken (`backed-up`), from whichever module holds node-backup on its machine (restic today).
var NodeAgentHears = []string{"mesh.mod." + BackupHolderModule + ".event.backed-up"}

// BackupHolderModule is the module that holds node-backup on every node, restic: the one whose word a node-engine hears.
// A second holder of the seat is a change of this grant.
const BackupHolderModule = "restic"

// administers says a person may invoke every tool (`*`): the operator.
func administers(invokes []string) bool {
	for _, i := range invokes {
		if i == "*" {
			return true
		}
	}
	return false
}

// HandPathWrites is what the operator's account publishes for release 6's hand path (ADR 0539 clauses 2 and 10): a want
// of any node, written and read back, in the assigner's bucket; each named node's compositions; the digests; and each
// node's object bucket for a file over 64 KiB, whose put reads an object's metadata and purges a replaced one's chunks.
func HandPathWrites(nodes []string) ([]string, error) {
	out := append([]string{"$KV." + delivery.AssignerWantsBucket + ".want.>"}, readGrants(delivery.AssignerWantsBucket, "want.>")...)
	out = append(out, "$KV."+delivery.ControllerComposedBucket+".>")
	out = append(out, readGrants(delivery.ControllerComposedBucket, "")...)
	for _, node := range nodes {
		w, err := ComposedWrites(node)
		if err != nil {
			return nil, err
		}
		bucket, _ := delivery.ComposedBucketFor(node)
		out = append(out, w)
		out = append(out, readGrants(bucket, "")...)
		files, err := DeclarationFilesWrites(node)
		if err != nil {
			return nil, err
		}
		stream := "OBJ_" + DeclarationFilesBucket + node
		out = append(out, files, "$JS.API.STREAM.INFO."+stream, "$JS.API.DIRECT.GET."+stream+".>",
			"$JS.API.STREAM.MSG.GET."+stream, "$JS.API.STREAM.PURGE."+stream)
	}
	return out, nil
}

// DeclarationsStream is the stream of each node's last declaration, `mesh.node.<node>.declare` (issue 107). Release 6
// sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the
// switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have
// names the new node-engine, and from then on neither asserts it nor any node's consumer on it.
const DeclarationsStream = "NODES"

// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
// declarations' stream, and no node's consumer of a declaration.
func RaiseAfterDeclarations(r Raiser) error {
	for _, s := range MeshStreams() {
		if s.Name == DeclarationsStream {
			continue
		}
		if err := r.EnsureStream(s); err != nil {
			return fmt.Errorf("asserting stream %s: %w", s.Name, err)
		}
	}
	return AssertMeshConsumers(r)
}