Give the watchman's stuck keys a layout the controller reads (issue 1153) #34
@@ -35,8 +35,8 @@ import (
|
||||
// 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"
|
||||
PresenceBucket = WatchmanModule + "_presence"
|
||||
PulseBucket = WatchmanModule + "_pulse"
|
||||
)
|
||||
|
||||
// HaveHistory and PresenceHistory are the revisions the have and presence buckets keep per key (research 063
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
package delivery
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// WatchmanModule is the watchman's module: its buckets are its own state, `<module>_<name>`, and the controller's
|
||||
// broker grants it alone the bus's connect and disconnect events (novox/hq ADR 0431).
|
||||
const WatchmanModule = "mesh-delivery-watchman"
|
||||
|
||||
// StuckBucket holds what stopped moving (novox/hq ADR 0433 §2): one key per key whose pulse expired, written by the
|
||||
// watchman, read by the controller's self-check, which raises each as a condition until it says moved. No key of it
|
||||
// is removed: a key that moves again is written as moved.
|
||||
const StuckBucket = WatchmanModule + "_stuck"
|
||||
|
||||
// StuckKey is the key, in StuckBucket, of the key that stopped moving: `<bucket>.<key>`, formed as PulseKey forms a
|
||||
// pulse's, so a node or part gone offline is `mesh-delivery-watchman_presence.<presence key>`.
|
||||
func StuckKey(bucket, key string) (string, error) {
|
||||
stuck, err := PulseKey(bucket, key)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("delivery: no stuck key: %w", err)
|
||||
}
|
||||
return stuck, 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("delivery: %q is not a stuck key: %w", stuckKey, err)
|
||||
}
|
||||
return bucket, key, nil
|
||||
}
|
||||
|
||||
// Stuck is the value of a stuck key: the key that stopped moving, what it waited for in words, since when, and
|
||||
// whether it has moved again since. The watchman writes it; the controller reads it.
|
||||
type Stuck struct {
|
||||
V int `json:"v"`
|
||||
Bucket string `json:"bucket"`
|
||||
Key string `json:"key"`
|
||||
Waited string `json:"waited,omitempty"`
|
||||
Since time.Time `json:"since"`
|
||||
Moved bool `json:"moved,omitempty"`
|
||||
}
|
||||
|
||||
// EncodeStuck is s as JSON at Version.
|
||||
func EncodeStuck(s Stuck) ([]byte, error) {
|
||||
s.V = Version
|
||||
return json.Marshal(s)
|
||||
}
|
||||
|
||||
// DecodeStuck reads a stuck value; it refuses an unknown version (ErrUnknownVersion).
|
||||
func DecodeStuck(data []byte) (Stuck, error) {
|
||||
var s Stuck
|
||||
err := decode(data, &s)
|
||||
return s, err
|
||||
}
|
||||
|
||||
// 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,88 @@
|
||||
package delivery
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 1153/AC4: the watchman's module and its three buckets, each that module's state.
|
||||
func TestWatchmanBuckets(t *testing.T) {
|
||||
if WatchmanModule != "mesh-delivery-watchman" {
|
||||
t.Fatalf("WatchmanModule = %q", WatchmanModule)
|
||||
}
|
||||
for name, bucket := range map[string]string{"presence": PresenceBucket, "pulse": PulseBucket, "stuck": StuckBucket} {
|
||||
if bucket != WatchmanModule+"_"+name {
|
||||
t.Errorf("the %s bucket is %q, not %s_%s", name, bucket, WatchmanModule, name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1153/AC1: the stuck key is the stopped key's bucket and key, split at the first dot, refused as a pulse key is.
|
||||
func TestStuckKey(t *testing.T) {
|
||||
key, err := StuckKey(PresenceBucket, "ace")
|
||||
if err != nil || key != "mesh-delivery-watchman_presence.ace" {
|
||||
t.Fatalf("StuckKey = %q, %v", key, err)
|
||||
}
|
||||
key, err = StuckKey("mesh-controller_builds", "zsh.build.b-1")
|
||||
if err != nil || key != "mesh-controller_builds.zsh.build.b-1" {
|
||||
t.Fatalf("StuckKey = %q, %v", key, err)
|
||||
}
|
||||
bucket, rest, err := ParseStuckKey(key)
|
||||
if err != nil || bucket != "mesh-controller_builds" || rest != "zsh.build.b-1" {
|
||||
t.Fatalf("ParseStuckKey = %q %q %v", bucket, rest, err)
|
||||
}
|
||||
for _, c := range [][2]string{{"", "k"}, {"a.b", "k"}, {"b", ""}, {"b", "k x"}, {"b", ".k"}, {"b", "k."}, {"b", "k>"}} {
|
||||
if _, err := StuckKey(c[0], c[1]); err == nil {
|
||||
t.Errorf("StuckKey(%q, %q) accepted", c[0], c[1])
|
||||
}
|
||||
}
|
||||
for _, bad := range []string{"", "nodot", ".k", "b."} {
|
||||
if _, _, err := ParseStuckKey(bad); err == nil {
|
||||
t.Errorf("ParseStuckKey(%q) accepted", bad)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1153/AC2: a stuck value round-trips with "v":1, and an unknown or missing version is refused loudly.
|
||||
func TestStuckValue(t *testing.T) {
|
||||
s := Stuck{Bucket: PresenceBucket, Key: "ace", Waited: "ace to come back online",
|
||||
Since: time.Date(2026, 10, 18, 20, 0, 0, 0, time.UTC)}
|
||||
raw, err := EncodeStuck(s)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var probe map[string]any
|
||||
if err := json.Unmarshal(raw, &probe); err != nil || probe["v"] != float64(1) {
|
||||
t.Fatalf("encoded %s: v = %v (%v)", raw, probe["v"], err)
|
||||
}
|
||||
back, err := DecodeStuck(raw)
|
||||
s.V = Version
|
||||
if err != nil || !SameStuck(back, s) {
|
||||
t.Fatalf("decoded %+v, %v; want %+v", back, err, s)
|
||||
}
|
||||
for _, bad := range []string{`{"v":2,"bucket":"b","key":"k"}`, `{"bucket":"b","key":"k"}`} {
|
||||
_, err := DecodeStuck([]byte(bad))
|
||||
if !errors.Is(err, ErrUnknownVersion) {
|
||||
t.Errorf("DecodeStuck(%s) = %v, want ErrUnknownVersion", bad, err)
|
||||
}
|
||||
if strings.Contains(bad, `"v":2`) && (err == nil || !strings.Contains(err.Error(), "2") || !strings.Contains(err.Error(), "1")) {
|
||||
t.Errorf("DecodeStuck(%s) = %v: does not name the version seen and the one read", bad, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1153/AC3: equal stuck values are the same; one moved is not.
|
||||
func TestSameStuck(t *testing.T) {
|
||||
a := Stuck{V: Version, Bucket: "b", Key: "k", Waited: "w", Since: time.Unix(100, 0).UTC()}
|
||||
b := a
|
||||
if !SameStuck(a, b) {
|
||||
t.Fatal("equal values are not the same")
|
||||
}
|
||||
b.Moved = true
|
||||
if SameStuck(a, b) {
|
||||
t.Fatal("a moved value is the same as a stuck one")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user