diff --git a/go/delivery/delivery.go b/go/delivery/delivery.go index a9abf83..fd109ec 100644 --- a/go/delivery/delivery.go +++ b/go/delivery/delivery.go @@ -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 diff --git a/go/delivery/stuck.go b/go/delivery/stuck.go new file mode 100644 index 0000000..2de3c46 --- /dev/null +++ b/go/delivery/stuck.go @@ -0,0 +1,65 @@ +package delivery + +import ( + "encoding/json" + "fmt" + "time" +) + +// WatchmanModule is the watchman's module: its buckets are its own state, `_`, 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: `.`, formed as PulseKey forms a +// pulse's, so a node or part gone offline is `mesh-delivery-watchman_presence.`. +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 +} diff --git a/go/delivery/stuck_test.go b/go/delivery/stuck_test.go new file mode 100644 index 0000000..fcca151 --- /dev/null +++ b/go/delivery/stuck_test.go @@ -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") + } +}