mesh-sdk/go/delivery/delivery.go

// Package delivery is the one home of the keys and values the parts of delivery read and write on the bus
// (novox/hq ADR 0429 §2: the key layouts another repository reads live in mesh-sdk; research 063 section 01). The
// node-engine writes the have keys, the watchman of the repository mesh-delivery writes presence and pulse, the
// view reads all of them, and the controller grants them; every one imports this package, and nothing else defines
// these names.
//
// **Buckets** are named `<owner module>_<name>` as the broker names a module's state (provider.BucketOf): the
// node-engine's `have`, and the watchman's `presence` and `pulse`.
//
// **Keys.** A have key is `<node>.<module>`, and a node's own facts are `<node>.node`, so a module may not be named
// `node`. A presence key is `<node>` for a node and `part.<part>.<instance>` for a part's instance, so a node may
// not be named `part`. A pulse key is `<bucket>.<key>`: the full name of the bucket watched, then that bucket's key,
// which may hold dots, so it splits at the first dot. A name is refused when it is empty or holds a dot, a wildcard,
// whitespace or a character a bus key refuses, and the error names it.
//
// **Values** are JSON of under about 500 bytes, each carrying its version in `v`. A reader refuses a value whose
// version is missing or one it does not know, with an error that is ErrUnknownVersion (research 063 section 05
// question 9 item 7): a writer writes a new version only once every reader's have shows a version that reads it. A
// writer writes only when the new value differs from the one before (the Same functions).
//
// Standard library only.
package delivery

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"regexp"
	"strings"
	"time"
)

// The buckets of delivery's parts. Each is written by one owner: every node's node-engine writes the have keys of
// its own node only, and the watchman writes presence and pulse.
const (
	HaveBucket     = "node-engine_have"
	PresenceBucket = WatchmanModule + "_presence"
	PulseBucket    = WatchmanModule + "_pulse"
)

// HaveHistory and PresenceHistory are the revisions the have and presence buckets keep per key (research 063
// section 01).
const (
	HaveHistory     = 20
	PresenceHistory = 20
)

// PulseMoveBound is how long a move may take: a pulse key lives for this time minus the move's age, and the bucket
// expires it on its own (the server writes a delete marker), so a move that overruns it is seen to stall.
const PulseMoveBound = 120 * time.Second

// Version is the version of every value this package writes and the only one it reads.
const Version = 1

// ErrUnknownVersion is what a decode returns for a value whose version is missing or not Version. Test it with
// errors.Is; the message names the version seen and the one read.
var ErrUnknownVersion = errors.New("delivery: unknown value version")

// Phase is how far a module's change has come on a node.
type Phase string

// The phases of a module's change on a node.
const (
	PhaseBackingUp Phase = "backing-up"
	PhaseApplying  Phase = "applying"
	PhaseChecking  Phase = "checking"
	PhaseRunning   Phase = "running"
	PhaseFailed    Phase = "failed"
)

// Health is what the module's own check says.
type Health string

// The healths a have carries.
const (
	HealthOK      Health = "ok"
	HealthFailed  Health = "failed"
	HealthWaits   Health = "waits"
	HealthUnknown Health = "unknown"
)

var nameOK = regexp.MustCompile(`^[-/_=a-zA-Z0-9]+$`)

// checkName refuses a name that is empty or holds a dot, a wildcard, whitespace or a character a bus key refuses.
func checkName(kind, name string) error {
	if name == "" {
		return fmt.Errorf("delivery: the %s name is empty", kind)
	}
	if !nameOK.MatchString(name) {
		return fmt.Errorf("delivery: the %s name %q holds a dot, a wildcard, whitespace or a character a key refuses", kind, name)
	}
	return nil
}

// HaveKey is the key, in HaveBucket, of a module's have on a node: `<node>.<module>`. A module named `node` is
// refused, since `<node>.node` is the node's own facts.
func HaveKey(node, module string) (string, error) {
	if err := checkName("node", node); err != nil {
		return "", err
	}
	if err := checkName("module", module); err != nil {
		return "", err
	}
	if module == "node" {
		return "", fmt.Errorf(`delivery: the module name "node" is refused: %s.node is the node's own key`, node)
	}
	return node + "." + module, nil
}

// HaveNodeKey is the key, in HaveBucket, of a node's own facts: `<node>.node`.
func HaveNodeKey(node string) (string, error) {
	if err := checkName("node", node); err != nil {
		return "", err
	}
	return node + ".node", nil
}

// ParseHaveKey is the inverse of HaveKey and HaveNodeKey: isNode is true for `<node>.node`, with module empty.
func ParseHaveKey(key string) (node, module string, isNode bool, err error) {
	node, module, ok := strings.Cut(key, ".")
	if !ok {
		return "", "", false, fmt.Errorf("delivery: %q is not a have key, <node>.<module>", key)
	}
	if err := checkName("node", node); err != nil {
		return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
	}
	if err := checkName("module", module); err != nil {
		return "", "", false, fmt.Errorf("delivery: %q is not a have key: %w", key, err)
	}
	if module == "node" {
		return node, "", true, nil
	}
	return node, module, false, nil
}

// PresenceNodeKey is the key, in PresenceBucket, of a node: `<node>`. A node named `part` is refused.
func PresenceNodeKey(node string) (string, error) {
	if err := checkName("node", node); err != nil {
		return "", err
	}
	if node == "part" {
		return "", errors.New(`delivery: the node name "part" is refused: part.<part>.<instance> is a part's key`)
	}
	return node, nil
}

// PresencePartKey is the key, in PresenceBucket, of a part's instance: `part.<part>.<instance>`.
func PresencePartKey(part, instance string) (string, error) {
	if err := checkName("part", part); err != nil {
		return "", err
	}
	if err := checkName("instance", instance); err != nil {
		return "", err
	}
	return "part." + part + "." + instance, nil
}

// ParsePresenceKey is the inverse of PresenceNodeKey and PresencePartKey: for a node's key part and instance are
// empty, for a part's key node is.
func ParsePresenceKey(key string) (node, part, instance string, err error) {
	if !strings.Contains(key, ".") {
		node, err = PresenceNodeKey(key)
		if err != nil {
			return "", "", "", fmt.Errorf("delivery: %q is not a presence key: %w", key, err)
		}
		return node, "", "", nil
	}
	fields := strings.Split(key, ".")
	if len(fields) != 3 || fields[0] != "part" {
		return "", "", "", fmt.Errorf("delivery: %q is not a presence key, <node> or part.<part>.<instance>", key)
	}
	if _, err := PresencePartKey(fields[1], fields[2]); err != nil {
		return "", "", "", fmt.Errorf("delivery: %q is not a presence key: %w", key, err)
	}
	return "", fields[1], fields[2], nil
}

// PulseKey is the key, in PulseBucket, of the move watched at key of bucket: `<bucket>.<key>`. The bucket is the
// full bucket name and holds no dot; the key may.
func PulseKey(bucket, key string) (string, error) {
	if err := checkName("bucket", bucket); err != nil {
		return "", err
	}
	if key == "" || strings.HasPrefix(key, ".") || strings.HasSuffix(key, ".") || strings.Contains(key, "..") {
		return "", fmt.Errorf("delivery: the pulse key %q is empty or has an empty part", key)
	}
	for _, part := range strings.Split(key, ".") {
		if err := checkName("key", part); err != nil {
			return "", err
		}
	}
	return bucket + "." + key, nil
}

// ParsePulseKey is the inverse of PulseKey, splitting at the first dot.
func ParsePulseKey(pulseKey string) (bucket, key string, err error) {
	bucket, key, ok := strings.Cut(pulseKey, ".")
	if !ok {
		return "", "", fmt.Errorf("delivery: %q is not a pulse key, <bucket>.<key>", pulseKey)
	}
	if _, err := PulseKey(bucket, key); err != nil {
		return "", "", fmt.Errorf("delivery: %q is not a pulse key: %w", pulseKey, err)
	}
	return bucket, key, nil
}

// PulseTTL is the time to live of a pulse key for a move of the given age: PulseMoveBound minus the age. At or past
// the bound the move is stalled, with no time left and nothing to write: the watchman raises it at once.
func PulseTTL(moveAge time.Duration) (ttl time.Duration, stalled bool) {
	if moveAge >= PulseMoveBound {
		return 0, true
	}
	return PulseMoveBound - moveAge, false
}

// Have is a module's state on a node, written by that node's node-engine under HaveKey. FP is the module's
// fingerprint once the indexer exists (design 47 release 3); until then Build is the build the node was sent, and
// the view compares by build.
type Have struct {
	V      int    `json:"v"`
	FP     string `json:"fp,omitempty"`
	Build  string `json:"build,omitempty"`
	Phase  Phase  `json:"phase"`
	Health Health `json:"health"`
	Detail string `json:"detail,omitempty"`
}

// HaveNode is a node's own facts, written under HaveNodeKey. Each field is the node-engine's own JSON, carried whole:
// this package names the facts and does not type them.
type HaveNode struct {
	V       int             `json:"v"`
	Windows json.RawMessage `json:"windows,omitempty"`
	Busy    json.RawMessage `json:"busy,omitempty"`
	Strays  json.RawMessage `json:"strays,omitempty"`
	Filters json.RawMessage `json:"filters,omitempty"`
	Profile json.RawMessage `json:"profile,omitempty"`
}

// Presence is whether a node or a part's instance is online; the entry's own time says since when. The watchman
// writes it.
type Presence struct {
	V      int  `json:"v"`
	Online bool `json:"online"`
}

// Pulse is the value of a pulse key: the watched key mirrored, its revision, and who owes the next move (a node or
// a part instance). The watchman writes it with the time to live PulseTTL gives.
type Pulse struct {
	V        int    `json:"v"`
	Bucket   string `json:"bucket"`
	Key      string `json:"key"`
	Revision uint64 `json:"revision"`
	Owes     string `json:"owes,omitempty"`
}

func checkVersion(data []byte) error {
	var probe struct {
		V *int `json:"v"`
	}
	if err := json.Unmarshal(data, &probe); err != nil {
		return fmt.Errorf("delivery: value is not JSON: %w", err)
	}
	if probe.V == nil {
		return fmt.Errorf("%w: no version in the value, this reader reads %d", ErrUnknownVersion, Version)
	}
	if *probe.V != Version {
		return fmt.Errorf("%w: the value is version %d, this reader reads %d", ErrUnknownVersion, *probe.V, Version)
	}
	return nil
}

func decode(data []byte, into any) error {
	if err := checkVersion(data); err != nil {
		return err
	}
	if err := json.Unmarshal(data, into); err != nil {
		return fmt.Errorf("delivery: value does not fit its shape: %w", err)
	}
	return nil
}

func (h Have) check() error {
	switch h.Phase {
	case PhaseBackingUp, PhaseApplying, PhaseChecking, PhaseRunning, PhaseFailed:
	default:
		return fmt.Errorf("delivery: phase %q is not backing-up, applying, checking, running or failed", h.Phase)
	}
	switch h.Health {
	case HealthOK, HealthFailed, HealthWaits, HealthUnknown:
	default:
		return fmt.Errorf("delivery: health %q is not ok, failed, waits or unknown", h.Health)
	}
	return nil
}

// EncodeHave is h as JSON at Version; it refuses a phase or health outside the lists.
func EncodeHave(h Have) ([]byte, error) {
	if err := h.check(); err != nil {
		return nil, err
	}
	h.V = Version
	return json.Marshal(h)
}

// DecodeHave reads a have; it refuses an unknown version (ErrUnknownVersion) and a phase or health outside the lists.
func DecodeHave(data []byte) (Have, error) {
	var h Have
	if err := decode(data, &h); err != nil {
		return Have{}, err
	}
	if err := h.check(); err != nil {
		return Have{}, err
	}
	return h, nil
}

// EncodeHaveNode is n as JSON at Version.
func EncodeHaveNode(n HaveNode) ([]byte, error) {
	n.V = Version
	return json.Marshal(n)
}

// DecodeHaveNode reads a node's facts; it refuses an unknown version (ErrUnknownVersion).
func DecodeHaveNode(data []byte) (HaveNode, error) {
	var n HaveNode
	err := decode(data, &n)
	return n, err
}

// EncodePresence is p as JSON at Version.
func EncodePresence(p Presence) ([]byte, error) {
	p.V = Version
	return json.Marshal(p)
}

// DecodePresence reads a presence; it refuses an unknown version (ErrUnknownVersion).
func DecodePresence(data []byte) (Presence, error) {
	var p Presence
	err := decode(data, &p)
	return p, err
}

// EncodePulse is p as JSON at Version.
func EncodePulse(p Pulse) ([]byte, error) {
	p.V = Version
	return json.Marshal(p)
}

// DecodePulse reads a pulse; it refuses an unknown version (ErrUnknownVersion).
func DecodePulse(data []byte) (Pulse, error) {
	var p Pulse
	err := decode(data, &p)
	return p, err
}

// SameHave says whether two haves are the same value, so a writer writes only when they differ.
func SameHave(a, b Have) bool { return a == b }

// SameHaveNode says whether two node facts are the same value, comparing each carried JSON without its spacing.
func SameHaveNode(a, b HaveNode) bool {
	return a.V == b.V && sameJSON(a.Windows, b.Windows) && sameJSON(a.Busy, b.Busy) &&
		sameJSON(a.Strays, b.Strays) && sameJSON(a.Filters, b.Filters) && sameJSON(a.Profile, b.Profile)
}

// SamePresence says whether two presences are the same value.
func SamePresence(a, b Presence) bool { return a == b }

// SamePulse says whether two pulses are the same value.
func SamePulse(a, b Pulse) bool { return a == b }

func sameJSON(a, b json.RawMessage) bool {
	var ca, cb bytes.Buffer
	if json.Compact(&ca, a) != nil || json.Compact(&cb, b) != nil {
		return bytes.Equal(a, b)
	}
	return bytes.Equal(ca.Bytes(), cb.Bytes())
}