// 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 `_` 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 `.`, and a node's own facts are `.node`, so a module may not be named // `node`. A presence key is `` for a node and `part..` for a part's instance, so a node may // not be named `part`. A pulse key is `.`: 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 = "mesh-delivery-watchman_presence" PulseBucket = "mesh-delivery-watchman_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: `.`. A module named `node` is // refused, since `.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`. 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`, 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, .", 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: ``. 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.. is a part's key`) } return node, nil } // PresencePartKey is the key, in PresenceBucket, of a part's instance: `part..`. 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, or part..", 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: `.`. 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, .", 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()) }