diff --git a/health-and-repair/go/clients/meshwatchdogsclient/go.mod b/health-and-repair/go/clients/meshwatchdogsclient/go.mod new file mode 100644 index 0000000..5d59e6d --- /dev/null +++ b/health-and-repair/go/clients/meshwatchdogsclient/go.mod @@ -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 +) diff --git a/health-and-repair/go/clients/meshwatchdogsclient/watchdogs.go b/health-and-repair/go/clients/meshwatchdogsclient/watchdogs.go new file mode 100644 index 0000000..2855912 --- /dev/null +++ b/health-and-repair/go/clients/meshwatchdogsclient/watchdogs.go @@ -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 ``). + NodePresence = seat.PresenceByNodeValue + // PartPresence is whether a part's instance is online (state presence, key `part..`). + 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: ``. 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.. is a part's key`) + } + return seat.PresenceByNodeKey(node), nil +} + +// PresencePartKey is the key of a part's instance: `part..`. +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, or part..", 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: `.`. 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, .", 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: `.`, 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 +} diff --git a/health-and-repair/go/clients/meshwatchdogsclient/watchdogs_test.go b/health-and-repair/go/clients/meshwatchdogsclient/watchdogs_test.go new file mode 100644 index 0000000..3f5c296 --- /dev/null +++ b/health-and-repair/go/clients/meshwatchdogsclient/watchdogs_test.go @@ -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) + } +}