//go:build integration package process import ( "slices" "strings" "sync" "testing" "time" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" ) // ADR 0593 ยง6: a state its owner keeps one bucket per node (composed-) is read as one state: a key is read from // the bucket its node part names, the keys of every node's bucket are listed together, and a watch hears every bucket, // one created after it began included, while a bucket outside the family is never read. func TestAPerNodeStateIsReadAcrossEveryNodesBucket(t *testing.T) { url, _ := grantedSendProcess(t) _, admin := adminJS(t, url) put := func(bucket, key, value string) { t.Helper() kv, err := admin.CreateOrUpdateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: bucket}) if err != nil { t.Fatal(err) } if _, err := kv.Put(ctxFor(t), key, []byte(value)); err != nil { t.Fatal(err) } } put("composed-ace", "ace.netcheck.aa", `{"n":1}`) put("composed-g14", "g14.netcheck.bb", `{"n":2}`) put("composedx-ace", "ace.other.cc", `{"n":3}`) conn, err := Connect(Account{URL: url, Node: "novox", Module: "mesh-provisioner", User: adminUser, Password: adminPass}) if err != nil { t.Fatal(err) } defer conn.Close() family := conn.ReadPerNode("mesh-host.composed", "composed-") if nodes, err := family.Nodes(); err != nil || !slices.Equal(nodes, []string{"ace", "g14"}) { t.Fatalf("nodes = %v, %v", nodes, err) } if e, err := family.Get("g14.netcheck.bb"); err != nil || e == nil || string(e.Value) != `{"n":2}` { t.Fatalf("get from g14's bucket: %+v, %v", e, err) } if e, err := family.Get("laptop.netcheck.bb"); err != nil || e != nil { t.Fatalf("a node with no bucket: %+v, %v", e, err) } if keys, err := family.Keys(); err != nil || !slices.Equal(keys, []string{"ace.netcheck.aa", "g14.netcheck.bb"}) { t.Fatalf("keys = %v, %v", keys, err) } var mu sync.Mutex heard := map[string]bool{} err = family.Watch("", func(c stdio.StateChange) error { mu.Lock() defer mu.Unlock() heard[c.Key] = true return nil }) if err != nil { t.Fatal(err) } put("composed-shanks", "shanks.netcheck.dd", `{"n":4}`) put("composed-ace", "ace.netcheck.ee", `{"n":5}`) deadline := time.Now().Add(5 * time.Second) want := []string{"ace.netcheck.aa", "ace.netcheck.ee", "g14.netcheck.bb", "shanks.netcheck.dd"} for { mu.Lock() var got []string for k := range heard { got = append(got, k) } mu.Unlock() slices.Sort(got) if slices.Equal(got, want) { break } if time.Now().After(deadline) || slices.ContainsFunc(got, func(k string) bool { return strings.HasPrefix(k, "ace.other") }) { t.Fatalf("heard %v, want %v", got, want) } time.Sleep(50 * time.Millisecond) } }