Watchdogs client: the keys, check and codec of presence, pulse and stuck over the seat mesh-watchdogs, goldens from go/delivery's output (B7n2; ADR 0634 §4; issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
module git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshwatchdogsclient
|
||||
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-20261105000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.0.0-20261105000000-000000000000
|
||||
)
|
||||
@@ -0,0 +1,205 @@
|
||||
// Package meshwatchdogsclient is the client of the seat mesh-watchdogs: the keys, the check and the codec of the seat's
|
||||
// three states presence, pulse and stuck, which every module that reads or writes them shares (the watchman that writes
|
||||
// them, the assigner and the self-check that read them). The layouts are the seat's contract; this file writes the
|
||||
// version and refuses what the earlier codec refused (novox/hq ADR 0433, ADR 0634 §4).
|
||||
package meshwatchdogsclient
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
)
|
||||
|
||||
// The three values of the seat's states, in the seat's types.
|
||||
type (
|
||||
// NodePresence is whether a node is online (state presence, key `<node>`).
|
||||
NodePresence = seat.PresenceByNodeValue
|
||||
// PartPresence is whether a part's instance is online (state presence, key `part.<part>.<instance>`).
|
||||
PartPresence = seat.PresenceByPartPartInstanceValue
|
||||
// Pulse is the watched key mirrored, its revision, and who owes the next move (state pulse).
|
||||
Pulse = seat.PulseByBucketKeyValue
|
||||
// Stuck is the key that stopped moving, what it waited for, since when, and whether it moved again (state stuck).
|
||||
Stuck = seat.StuckByBucketKeyValue
|
||||
)
|
||||
|
||||
// PartWord is the first part of a part's presence key; a node of that name is refused.
|
||||
const PartWord = "part"
|
||||
|
||||
// PulseMoveBound is how long a move may take: a pulse key lives for this time minus the move's age.
|
||||
const PulseMoveBound = 120 * time.Second
|
||||
|
||||
// PresenceHistory is the number of revisions the presence state keeps per key.
|
||||
const PresenceHistory = 20
|
||||
|
||||
// PresenceNodeKey is the key of a node's presence: `<node>`. A node named `part` is refused.
|
||||
func PresenceNodeKey(node string) (string, error) {
|
||||
if err := keys.CheckName("node", node); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if node == PartWord {
|
||||
return "", errors.New(`the node name "part" is refused: part.<part>.<instance> is a part's key`)
|
||||
}
|
||||
return seat.PresenceByNodeKey(node), nil
|
||||
}
|
||||
|
||||
// PresencePartKey is the key of a part's instance: `part.<part>.<instance>`.
|
||||
func PresencePartKey(part, instance string) (string, error) {
|
||||
if err := keys.CheckName("part", part); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if err := keys.CheckName("instance", instance); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return seat.PresenceByPartPartInstanceKey(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("%q is not a presence key: %w", key, err)
|
||||
}
|
||||
return node, "", "", nil
|
||||
}
|
||||
fields := strings.Split(key, ".")
|
||||
if len(fields) != 3 || fields[0] != PartWord {
|
||||
return "", "", "", fmt.Errorf("%q is not a presence key, <node> or part.<part>.<instance>", key)
|
||||
}
|
||||
if _, err := PresencePartKey(fields[1], fields[2]); err != nil {
|
||||
return "", "", "", fmt.Errorf("%q is not a presence key: %w", key, err)
|
||||
}
|
||||
return "", fields[1], fields[2], nil
|
||||
}
|
||||
|
||||
// PulseKey is the key 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 := keys.CheckName("bucket", bucket); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if key == "" || strings.HasPrefix(key, ".") || strings.HasSuffix(key, ".") || strings.Contains(key, "..") {
|
||||
return "", fmt.Errorf("the pulse key %q is empty or has an empty part", key)
|
||||
}
|
||||
for _, part := range strings.Split(key, ".") {
|
||||
if err := keys.CheckName("key", part); err != nil {
|
||||
return "", err
|
||||
}
|
||||
}
|
||||
return seat.PulseByBucketKeyKey(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 := seat.ParsePulseByBucketKeyKey(pulseKey)
|
||||
if !ok {
|
||||
return "", "", fmt.Errorf("%q is not a pulse key, <bucket>.<key>", pulseKey)
|
||||
}
|
||||
if _, err := PulseKey(bucket, key); err != nil {
|
||||
return "", "", fmt.Errorf("%q is not a pulse key: %w", pulseKey, err)
|
||||
}
|
||||
return bucket, key, nil
|
||||
}
|
||||
|
||||
// StuckKey is the key of the key that stopped moving: `<bucket>.<key>`, formed as PulseKey forms a pulse's.
|
||||
func StuckKey(bucket, key string) (string, error) {
|
||||
k, err := PulseKey(bucket, key)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("no stuck key: %w", err)
|
||||
}
|
||||
return k, nil
|
||||
}
|
||||
|
||||
// ParseStuckKey is the inverse of StuckKey, splitting at the first dot.
|
||||
func ParseStuckKey(stuckKey string) (bucket, key string, err error) {
|
||||
bucket, key, err = ParsePulseKey(stuckKey)
|
||||
if err != nil {
|
||||
return "", "", fmt.Errorf("%q is not a stuck key: %w", stuckKey, 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.
|
||||
func PulseTTL(moveAge time.Duration) (ttl time.Duration, stalled bool) {
|
||||
if moveAge >= PulseMoveBound {
|
||||
return 0, true
|
||||
}
|
||||
return PulseMoveBound - moveAge, false
|
||||
}
|
||||
|
||||
// EncodeNodePresence is p as JSON at the current version.
|
||||
func EncodeNodePresence(p NodePresence) ([]byte, error) {
|
||||
p.V = keys.Version
|
||||
return json.Marshal(p)
|
||||
}
|
||||
|
||||
// DecodeNodePresence reads a node's presence; it refuses an unknown version (keys.ErrUnknownVersion).
|
||||
func DecodeNodePresence(data []byte) (NodePresence, error) {
|
||||
if err := keys.CheckVersion(data); err != nil {
|
||||
return NodePresence{}, err
|
||||
}
|
||||
return decodeStrict[NodePresence](data, "presence")
|
||||
}
|
||||
|
||||
// EncodePartPresence is p as JSON at the current version.
|
||||
func EncodePartPresence(p PartPresence) ([]byte, error) {
|
||||
p.V = keys.Version
|
||||
return json.Marshal(p)
|
||||
}
|
||||
|
||||
// DecodePartPresence reads a part's presence; it refuses an unknown version (keys.ErrUnknownVersion).
|
||||
func DecodePartPresence(data []byte) (PartPresence, error) {
|
||||
if err := keys.CheckVersion(data); err != nil {
|
||||
return PartPresence{}, err
|
||||
}
|
||||
return decodeStrict[PartPresence](data, "presence")
|
||||
}
|
||||
|
||||
// EncodePulse is p as JSON at the current version.
|
||||
func EncodePulse(p Pulse) ([]byte, error) {
|
||||
p.V = keys.Version
|
||||
return json.Marshal(p)
|
||||
}
|
||||
|
||||
// DecodePulse reads a pulse; it refuses an unknown version (keys.ErrUnknownVersion).
|
||||
func DecodePulse(data []byte) (Pulse, error) {
|
||||
if err := keys.CheckVersion(data); err != nil {
|
||||
return Pulse{}, err
|
||||
}
|
||||
return decodeStrict[Pulse](data, "pulse")
|
||||
}
|
||||
|
||||
// EncodeStuck is s as JSON at the current version.
|
||||
func EncodeStuck(s Stuck) ([]byte, error) {
|
||||
s.V = keys.Version
|
||||
return json.Marshal(s)
|
||||
}
|
||||
|
||||
// DecodeStuck reads a stuck value; it refuses an unknown version (keys.ErrUnknownVersion).
|
||||
func DecodeStuck(data []byte) (Stuck, error) {
|
||||
if err := keys.CheckVersion(data); err != nil {
|
||||
return Stuck{}, err
|
||||
}
|
||||
return decodeStrict[Stuck](data, "stuck")
|
||||
}
|
||||
|
||||
func decodeStrict[T any](data []byte, what string) (T, error) {
|
||||
var v T
|
||||
if err := json.Unmarshal(data, &v); err != nil {
|
||||
return v, fmt.Errorf("the %s value does not fit its shape: %w", what, err)
|
||||
}
|
||||
return v, nil
|
||||
}
|
||||
|
||||
// SameStuck says whether two stuck values are the same, so a writer writes only when they differ.
|
||||
func SameStuck(a, b Stuck) bool {
|
||||
return a.V == b.V && a.Bucket == b.Bucket && a.Key == b.Key && a.Waited == b.Waited && a.Since.Equal(b.Since) &&
|
||||
a.Moved == b.Moved
|
||||
}
|
||||
@@ -0,0 +1,140 @@
|
||||
package meshwatchdogsclient
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
)
|
||||
|
||||
func sameJSON(t *testing.T, got []byte, want string) {
|
||||
t.Helper()
|
||||
var g, w any
|
||||
if err := json.Unmarshal(got, &g); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := json.Unmarshal([]byte(want), &w); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(g, w) {
|
||||
t.Fatalf("wrote %s, the earlier codec wrote %s", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// TestTheValuesAreWhatTheEarlierCodecWrote holds the values equal to what go/delivery's codec wrote for the same values
|
||||
// (the golden texts are its output; the field order is the contract's now, so values are compared decoded), reads them
|
||||
// back, and refuses an unknown version.
|
||||
func TestTheValuesAreWhatTheEarlierCodecWrote(t *testing.T) {
|
||||
since := time.Date(2026, 10, 25, 11, 52, 58, 51397201, time.UTC)
|
||||
for name, row := range map[string]struct {
|
||||
encode func() ([]byte, error)
|
||||
decode func([]byte) (any, error)
|
||||
golden string
|
||||
want any
|
||||
}{
|
||||
"node online": {
|
||||
func() ([]byte, error) { return EncodeNodePresence(NodePresence{Online: true}) },
|
||||
func(b []byte) (any, error) { return DecodeNodePresence(b) },
|
||||
`{"v":1,"online":true}`, NodePresence{V: 1, Online: true},
|
||||
},
|
||||
"part offline": {
|
||||
func() ([]byte, error) { return EncodePartPresence(PartPresence{}) },
|
||||
func(b []byte) (any, error) { return DecodePartPresence(b) },
|
||||
`{"v":1,"online":false}`, PartPresence{V: 1},
|
||||
},
|
||||
"pulse": {
|
||||
func() ([]byte, error) {
|
||||
return EncodePulse(Pulse{Bucket: "mesh-delivery-assigner_wants", Key: "want.a.m", Revision: 7, Owes: "a"})
|
||||
},
|
||||
func(b []byte) (any, error) { return DecodePulse(b) },
|
||||
`{"v":1,"bucket":"mesh-delivery-assigner_wants","key":"want.a.m","revision":7,"owes":"a"}`,
|
||||
Pulse{V: 1, Bucket: "mesh-delivery-assigner_wants", Key: "want.a.m", Revision: 7, Owes: "a"},
|
||||
},
|
||||
"stuck": {
|
||||
func() ([]byte, error) {
|
||||
return EncodeStuck(Stuck{Bucket: "mesh-delivery-watchman_presence", Key: "part.b.n", Waited: "b on n to come back", Since: since, Moved: true})
|
||||
},
|
||||
func(b []byte) (any, error) { return DecodeStuck(b) },
|
||||
`{"v":1,"bucket":"mesh-delivery-watchman_presence","key":"part.b.n","waited":"b on n to come back","since":"2026-10-25T11:52:58.051397201Z","moved":true}`,
|
||||
Stuck{V: 1, Bucket: "mesh-delivery-watchman_presence", Key: "part.b.n", Waited: "b on n to come back", Since: since, Moved: true},
|
||||
},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
raw, err := row.encode()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
sameJSON(t, raw, row.golden)
|
||||
back, err := row.decode([]byte(row.golden))
|
||||
if err != nil || !reflect.DeepEqual(back, row.want) {
|
||||
t.Fatalf("read %+v, %v; want %+v", back, err, row.want)
|
||||
}
|
||||
if _, err := row.decode([]byte(`{"v":2}`)); !errors.Is(err, keys.ErrUnknownVersion) {
|
||||
t.Fatalf("version 2: %v", err)
|
||||
}
|
||||
if _, err := row.decode([]byte(`{}`)); !errors.Is(err, keys.ErrUnknownVersion) {
|
||||
t.Fatalf("no version: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
if !SameStuck(Stuck{V: 1, Since: since}, Stuck{V: 1, Since: since.In(time.FixedZone("x", 3600))}) || SameStuck(Stuck{Moved: true}, Stuck{}) {
|
||||
t.Fatal("SameStuck")
|
||||
}
|
||||
}
|
||||
|
||||
// TestTheKeysAreWhatTheEarlierCodecFormed holds the key forms and their refusals.
|
||||
func TestTheKeysAreWhatTheEarlierCodecFormed(t *testing.T) {
|
||||
if k, err := PresenceNodeKey("ace"); err != nil || k != "ace" {
|
||||
t.Fatal(k, err)
|
||||
}
|
||||
if k, err := PresencePartKey("mesh-delivery-builder", "novox"); err != nil || k != "part.mesh-delivery-builder.novox" {
|
||||
t.Fatal(k, err)
|
||||
}
|
||||
for _, bad := range []string{"part", "", "a.b"} {
|
||||
if _, err := PresenceNodeKey(bad); err == nil {
|
||||
t.Fatalf("node %q accepted", bad)
|
||||
}
|
||||
}
|
||||
for key, want := range map[string][3]string{"ace": {"ace", "", ""}, "part.p.i": {"", "p", "i"}} {
|
||||
n, p, i, err := ParsePresenceKey(key)
|
||||
if err != nil || [3]string{n, p, i} != want {
|
||||
t.Fatal(key, n, p, i, err)
|
||||
}
|
||||
}
|
||||
for _, bad := range []string{"part.p", "x.p.i", "part..i", "a.b.c.d", "part"} {
|
||||
if _, _, _, err := ParsePresenceKey(bad); err == nil {
|
||||
t.Fatalf("presence key %q accepted", bad)
|
||||
}
|
||||
}
|
||||
k, err := PulseKey("b_w", "want.a.m")
|
||||
if err != nil || k != "b_w.want.a.m" {
|
||||
t.Fatal(k, err)
|
||||
}
|
||||
if s, err := StuckKey("b_w", "want.a.m"); err != nil || s != k {
|
||||
t.Fatal(s, err)
|
||||
}
|
||||
for _, f := range []func(string) (string, string, error){ParsePulseKey, ParseStuckKey} {
|
||||
if b, key, err := f(k); err != nil || b != "b_w" || key != "want.a.m" {
|
||||
t.Fatal(b, key, err)
|
||||
}
|
||||
for _, bad := range []string{"nodot", ".k", "b.", "b..k", "b.k."} {
|
||||
if _, _, err := f(bad); err == nil {
|
||||
t.Fatalf("pulse key %q accepted", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, bad := range [][2]string{{"", "k"}, {"b.c", "k"}, {"b", ""}, {"b", "k..j"}, {"b", "k j"}} {
|
||||
if _, err := PulseKey(bad[0], bad[1]); err == nil {
|
||||
t.Fatalf("pulse key %q accepted", bad)
|
||||
}
|
||||
}
|
||||
if ttl, stalled := PulseTTL(30 * time.Second); stalled || ttl != 90*time.Second {
|
||||
t.Fatal(ttl, stalled)
|
||||
}
|
||||
if ttl, stalled := PulseTTL(PulseMoveBound); !stalled || ttl != 0 {
|
||||
t.Fatal(ttl, stalled)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user