diff --git a/go/compose/network.go b/go/compose/network.go index 1e66c51..afc8363 100644 --- a/go/compose/network.go +++ b/go/compose/network.go @@ -74,7 +74,10 @@ func (c *composer) overlayRange() string { } // network builds the private network over the machines that resolved the module for it. A machine is on it because it -// was given the module, or because it is joining with a key and an address (novox/hq ADR 0169). +// was given the module, or because it is joining with a key and an address (novox/hq ADR 0169), or because its own set +// could not be resolved and it holds a place (a key and an address): that place is its last good one on the network, and +// it does not change because its modules failed, so every other machine keeps its path to it -- the hub above all -- and +// only its own declaration is refused (issue 1833). A machine refused with no place is left off. func (c *composer) network(on map[string]bool, refused map[string]string) (*overlay.Generator, error) { tunnels := c.in.Tunnels carried := c.in.CarriedPeers @@ -85,9 +88,15 @@ func (c *composer) network(on map[string]bool, refused map[string]string) (*over var nodes []overlay.Node for _, node := range c.in.sorted() { p := node.Place - if !on[node.Name] && !(isJoining[node.Name] && p.Key != "" && p.Address != "") { + placed := p.Key != "" && p.Address != "" + _, wasRefused := refused[node.Name] + if !on[node.Name] && !((isJoining[node.Name] || wasRefused) && placed) { continue } + if !on[node.Name] && wasRefused && !isJoining[node.Name] { + c.say("%s could not be resolved and keeps its place on the private network at %s, so the others keep their path to it", + node.Name, p.Address) + } n := overlay.Node{ Name: node.Name, Key: p.Key, Endpoint: p.Endpoint, Site: p.Site, Hub: p.Hub, Address: p.Address, @@ -233,7 +242,9 @@ func (c *composer) whereEveryoneIs() (map[string]string, error) { } // onTheNetwork is every placed machine that resolves the private network: has an address AND runs what puts it there -// (novox/hq issue 079). A machine that does not resolve is not counted, and said. +// (novox/hq issue 079). A machine that does not resolve keeps its place when it holds one (a key and an address), by the +// same rule as network(), so its name and its address stay where the tunnel to it does (issue 1833); one without is not +// counted, and said. func (c *composer) onTheNetwork() []Node { var out []Node for _, n := range c.in.sorted() { @@ -247,6 +258,12 @@ func (c *composer) onTheNetwork() []Node { catalogue.Node{Name: n.Name, Site: n.Place.Site, Capabilities: n.Capabilities}, catalogue.World{Unchecked: true, Holdings: c.in.Holdings}) if err != nil { + if n.Place.Key != "" { + c.say("%s could not be resolved and keeps its place on the private network at %s, so the others keep their path to it", + n.Name, n.Place.Address) + out = append(out, n) + continue + } c.say("%s is not counted as on the network: it does not resolve: %v", n.Name, err) continue } diff --git a/go/compose/network_test.go b/go/compose/network_test.go index 9f5717f..ff73e08 100644 --- a/go/compose/network_test.go +++ b/go/compose/network_test.go @@ -159,9 +159,11 @@ func TestANodeNotOnThePrivateNetworkGetsNoAddressAndNoNetwork(t *testing.T) { } // 1360/AC3: a node whose resolution is refused is said with today's words: the network missing its hub names the -// machine that could not be resolved, which is almost always why. +// machine that could not be resolved, which is almost always why. The hub here holds no complete place (no key): a refused +// machine with one keeps it on the network (issue 1833), so only one without is missing from it. func TestARefusedNodeIsSaidWhenTheNetworkCannotBeComposed(t *testing.T) { in := aMesh(t) + nodeOf(t, &in, "anchor").Place.Key = "" one := catalogue.Manifest{Module: "rival-one", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} two := catalogue.Manifest{Module: "rival-two", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} in.Catalogue["rival-one"], in.Catalogue["rival-two"] = one, two @@ -210,9 +212,10 @@ func TestEachFieldOfTheInputReachesTheGenerator(t *testing.T) { if len(base) != 2 { t.Fatalf("the base mesh has %v", base) } - t.Run("Catalogue: without the module nobody is on the network", func(t *testing.T) { + t.Run("Catalogue: without the module offering it nobody is on the network", func(t *testing.T) { in := aMesh(t) - in.Catalogue = map[string]catalogue.Manifest{} + // The module stays known, so every machine still resolves: a machine that does not resolve keeps its place (issue 1833). + in.Catalogue = map[string]catalogue.Manifest{overlay.Name: {Module: overlay.Name, Version: "1"}} if got := addressesOf(in); len(got) != 0 { t.Errorf("got %v", got) } @@ -290,6 +293,8 @@ func TestEachFieldOfTheInputReachesTheGenerator(t *testing.T) { } laptop := nodeOf(t, &in, "laptop") laptop.Assigned = append(laptop.Assigned, "holder-one", "holder-two") + // Without a key: a machine that does not resolve and holds a complete place keeps it on the network (issue 1833). + laptop.Place.Key = "" if got := addressesOf(in); got["laptop"] != "" { t.Fatalf("a machine whose set claims one seat twice is on the network: %v", got) } diff --git a/go/compose/node_test.go b/go/compose/node_test.go index b7be07a..c2485dc 100644 --- a/go/compose/node_test.go +++ b/go/compose/node_test.go @@ -387,9 +387,11 @@ func TestAnotherNodesKeptBuildDoesNotChangeTheRestOfTheMeshOrWhoIsOnTheNetwork(t func overlayName() string { return overlay.Name } -// Round 1, finding 6: the nodes the composition did not count as on the network are returned, said, not lost. +// Round 1, finding 6: the nodes the composition did not count as on the network are returned, said, not lost. laptop holds no +// complete place (no key): one that does keeps it on the network (issue 1833) and is counted. func TestANodeNotCountedAsOnTheNetworkIsSaidInTheResult(t *testing.T) { in := aMesh(t) + nodeOf(t, &in, "laptop").Place.Key = "" in.Catalogue["rival-one"] = catalogue.Manifest{Module: "rival-one", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} in.Catalogue["rival-two"] = catalogue.Manifest{Module: "rival-two", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} laptop := nodeOf(t, &in, "laptop") diff --git a/go/compose/one_machine_alone_test.go b/go/compose/one_machine_alone_test.go new file mode 100644 index 0000000..aa6b4c9 --- /dev/null +++ b/go/compose/one_machine_alone_test.go @@ -0,0 +1,167 @@ +package compose + +import ( + "slices" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/overlay" +) + +// fourOnTheNetwork is aMesh and two more machines on the private network at the site `there`: `ace`, dialled at +// ace.example:51820, at 10.77.0.3, and `box`, behind no endpoint, at 10.77.0.4. A module `office` requires +// `document-server`, which nothing in the catalogue provides: the shape of issue 1833 (nextcloud on the hub wanting a +// document server nobody offered). +func fourOnTheNetwork(t *testing.T) Input { + t.Helper() + in := aMesh(t) + in.Catalogue["office"] = catalogue.Manifest{Module: "office", Version: "1", Requires: []string{"document-server"}} + caps := map[string]bool{"container-runtime": true, "wireguard": true, "systemd": true} + in.Nodes = append(in.Nodes, + Node{Name: "ace", Place: Place{Key: keyOf(t, "ace"), Endpoint: "ace.example:51820", Site: "there", Address: "10.77.0.3"}, + Assigned: []string{overlay.Name}, Capabilities: caps}, + Node{Name: "box", Place: Place{Key: keyOf(t, "box"), Site: "there", Address: "10.77.0.4"}, + Assigned: []string{overlay.Name}, Capabilities: caps}) + return in +} + +// unresolvableBy makes one machine's own set fail to resolve: it is also given `office`, whose requirement nothing provides. +func unresolvableBy(t *testing.T, in *Input, name string) { + t.Helper() + n := nodeOf(t, in, name) + n.Assigned = append(n.Assigned, "office") +} + +// networkOf is the generator of an Input's private network, failing the test when it is not composed. +func networkOf(t *testing.T, in Input) *overlay.Generator { + t.Helper() + gens, err := Generators(in) + if err != nil { + t.Fatalf("the private network was not composed: %v", err) + } + return gens[overlay.Name].(*overlay.Generator) +} + +func placesOf(g *overlay.Generator) map[string]overlay.Node { + out := map[string]overlay.Node{} + for _, n := range g.Nodes() { + out[n.Name] = n + } + return out +} + +// 1833/AC1: four machines, the hub's own set does not resolve; the other three compose their declarations, and the hub alone is +// refused, with an error that is its own set failing and names what could not be resolved. +func TestOneMachineThatDoesNotResolveIsRefusedAlone(t *testing.T) { + in := fourOnTheNetwork(t) + unresolvableBy(t, &in, "anchor") + for _, node := range []string{"laptop", "ace", "box"} { + d, problems, err := Compose(in, node) + if err != nil || len(problems) != 0 { + t.Errorf("%s does not compose because another machine does not resolve: %v %v", node, err, problems) + continue + } + if !has(resourceIDs(d), "file:mesh-wireguard.overlay-config") { + t.Errorf("%s is given no private network: %v", node, resourceIDs(d)) + } + } + _, _, err := Compose(in, "anchor") + if err == nil || !IsUnresolvable(err) { + t.Fatalf("the machine that does not resolve is not refused as its own set failing: %v", err) + } + if !strings.Contains(err.Error(), `nothing provides "document-server", wanted by office`) { + t.Errorf("the refusal does not name what could not be resolved: %v", err) + } +} + +// 1833/AC2: the private network is composed with every machine on it: the hub that does not resolve keeps its place (key, +// address, endpoint, site, hub role), so every other machine keeps its path through it. +func TestAMachineThatDoesNotResolveKeepsItsPlaceOnTheNetwork(t *testing.T) { + in := fourOnTheNetwork(t) + unresolvableBy(t, &in, "anchor") + got := placesOf(networkOf(t, in)) + if len(got) != 4 { + t.Fatalf("the network is over %v, want all four machines", got) + } + hub := got["anchor"] + if !hub.Hub || hub.Key != keyOf(t, "anchor") || hub.Address != "10.77.0.1" || hub.Endpoint != "anchor.example:51820" || hub.Site != "here" { + t.Errorf("the hub did not keep its place: %+v", hub) + } + nodes, graph, err := Graph(in) + if err != nil || len(nodes) != 4 { + t.Fatalf("the graph is not shown with every machine: %v %v", nodes, err) + } + for _, node := range []string{"ace", "box"} { + through := false + for _, p := range graph[node] { + through = through || (p.Name == "anchor" && p.Endpoint == "anchor.example:51820") + } + if !through { + t.Errorf("%s has no path through the hub: %+v", node, graph[node]) + } + } +} + +// 1833/AC3: a machine that does not resolve and was never placed (a new machine: no key, no address) is left off the network, +// which is composed from the rest -- here with the hub that does not resolve on its last place -- and that machine alone is refused. +func TestANewMachineThatDoesNotResolveIsLeftOffAndRefusedAlone(t *testing.T) { + in := fourOnTheNetwork(t) + unresolvableBy(t, &in, "anchor") + in.Nodes = append(in.Nodes, Node{Name: "fresh", Place: Place{Site: "there"}, + Assigned: []string{overlay.Name, "office"}, Capabilities: map[string]bool{"wireguard": true, "systemd": true}}) + got := placesOf(networkOf(t, in)) + if _, on := got["fresh"]; on || len(got) != 4 { + t.Fatalf("the network is over %v, want the four placed machines without fresh", got) + } + if _, _, err := Compose(in, "fresh"); err == nil || !IsUnresolvable(err) { + t.Errorf("the new machine is not refused as its own set failing: %v", err) + } + if _, problems, err := Compose(in, "box"); err != nil || len(problems) != 0 { + t.Errorf("another machine does not compose: %v %v", err, problems) + } +} + +// 1833/AC4: a network that truly cannot be composed from what remains still fails: the hub does not resolve and has no +// complete place (no key), so nothing is the hub; the refusal is today's, naming the machine that could not be resolved. +func TestANetworkThatCannotBeComposedFromWhatRemainsStillFails(t *testing.T) { + in := fourOnTheNetwork(t) + unresolvableBy(t, &in, "anchor") + nodeOf(t, &in, "anchor").Place.Key = "" + _, err := Generators(in) + if err == nil { + t.Fatal("a network without a hub was composed") + } + for _, want := range []string{overlay.ErrNoHub.Error(), "and 1 node(s) could not be resolved at all, which is probably why:", " anchor: "} { + if !strings.Contains(err.Error(), want) { + t.Errorf("the refusal does not say %q:\n%v", want, err) + } + } + if _, _, err := Compose(in, "laptop"); err == nil { + t.Error("a declaration was composed over a network that cannot be") + } +} + +// 1833/AC2, a failure the composition's unchecked pass refuses too (two modules claiming one seat on the hub, the 1360/AC3 +// shape): the hub keeps its place in what every other machine is sent besides the tunnel -- its internal name and its address +// among the sources "from the mesh" -- and another machine still composes. +func TestAMachineThatDoesNotResolveKeepsItsNameAndItsAddressInTheMesh(t *testing.T) { + in := fourOnTheNetwork(t) + in.Catalogue["rival-one"] = catalogue.Manifest{Module: "rival-one", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} + in.Catalogue["rival-two"] = catalogue.Manifest{Module: "rival-two", Version: "1", Claims: []catalogue.Claim{{Name: "the-seat", Scope: catalogue.ScopeNode}}} + hub := nodeOf(t, &in, "anchor") + hub.Assigned = append(hub.Assigned, "rival-one", "rival-two") + if _, err := Resolve(in, "anchor"); !IsUnresolvable(err) { + t.Fatalf("the hub is meant not to resolve: %v", err) + } + c := newComposer(in) + if got := c.namesInTheMesh()["anchor."+c.suffix()]; got != "10.77.0.1" { + t.Errorf("the hub's name is not in the mesh's names: %v", c.namesInTheMesh()) + } + if got := c.onThePrivateNetwork(); !slices.Contains(got, "10.77.0.1") || len(got) != 4 { + t.Errorf("the sources from the mesh are %v, want all four machines with the hub", got) + } + if _, problems, err := Compose(in, "ace"); err != nil || len(problems) != 0 { + t.Errorf("another machine does not compose: %v %v", err, problems) + } +} diff --git a/go/process/state.go b/go/process/state.go index 8e294e3..777a9fd 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -5,6 +5,8 @@ import ( "encoding/json" "errors" "fmt" + "sort" + "strings" "time" "github.com/nats-io/nats.go" @@ -27,6 +29,8 @@ type State struct { // kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a // unit test, a fake that never reaches a bus. kv func(ctx context.Context) (jetstream.KeyValue, error) + // closed says the connection is gone, so a watch ends; nil in a unit test. + closed func() bool } // bucket is this state's name on the bus. @@ -38,7 +42,7 @@ func (s State) bucket() string { return s.Module + "_" + s.Name } // not granted is refused there, promptly, as nats.ErrPermissionViolation. func (c *Conn) State(name string) State { bucket := c.account.Module + "_" + name - return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) { + return State{Module: c.account.Module, Name: name, closed: func() bool { return c.nc.IsClosed() }, kv: func(ctx context.Context) (jetstream.KeyValue, error) { if err := c.refusedUnannounced(name); err != nil { return nil, err } @@ -131,3 +135,148 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { } return rev, nil } + +// Put writes a key whole, whatever it held, and returns the revision written. +func (s State) Put(key string, value any) (uint64, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return 0, err + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Put(ctx, key, data) + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} + +// Keys is every key with a current value, sorted; none is an empty list. +func (s State) Keys() ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + if err != nil { + return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err) + } + sort.Strings(keys) + return keys, nil +} + +// watchPattern is a key narrowing as the bus names it: `*` stays one name, `**` is the rest, empty is every key. +func watchPattern(key string) string { + if key == "" { + return jetstream.AllKeys + } + tokens := strings.Split(key, ".") + for i, t := range tokens { + if t == "**" { + tokens[i] = ">" + } + } + return strings.Join(tokens, ".") +} + +// Watch hands handle every current value — none that is deleted, each with Current set — and then every +// change in order, and returns once the current values have all been handled. key narrows it: a key, with +// `*` for one name and `**` for the rest; empty is every key. The changes go on for the life of the +// connection; an error from handle, or the connection closing, ends them. Before the current values are +// handled, either is returned as the error. +func (s State) Watch(key string, handle func(stdio.StateChange) error) error { + // the watch outlives this call, so its context is not the call's timeout: the connection ends it. + ctx, cancel := context.WithCancel(context.Background()) + bind, stopBind := context.WithTimeout(ctx, stateTimeout) + kv, err := s.store(bind) + stopBind() + if err != nil { + cancel() + return err + } + w, err := kv.Watch(ctx, watchPattern(key)) + if err != nil { + cancel() + return fmt.Errorf("%s's state %s, watching %q: %w", s.Module, s.Name, key, err) + } + end := func() { w.Stop(); cancel() } + next := func() (jetstream.KeyValueEntry, bool, error) { + tick := time.NewTicker(250 * time.Millisecond) + defer tick.Stop() + for { + select { + case e, open := <-w.Updates(): + if !open { + return nil, false, fmt.Errorf("%s's state %s: the watch ended", s.Module, s.Name) + } + return e, true, nil + case <-tick.C: + if s.closed != nil && s.closed() { + return nil, false, fmt.Errorf("%s's state %s: the connection closed", s.Module, s.Name) + } + } + } + } + deliver := func(e jetstream.KeyValueEntry, current bool) error { + c := stdio.StateChange{State: s.Name, Key: e.Key(), Op: "put", Revision: e.Revision(), Current: current} + if e.Operation() == jetstream.KeyValuePut { + c.Value = json.RawMessage(e.Value()) + } else { + c.Op = "delete" + } + return handle(c) + } + for { + e, _, err := next() + if err != nil { + end() + return err + } + if e == nil { + break + } + if e.Operation() != jetstream.KeyValuePut { + continue + } + if err := deliver(e, true); err != nil { + end() + return err + } + } + go func() { + defer end() + for { + e, _, err := next() + if err != nil || e == nil { + return + } + if deliver(e, false) != nil { + return + } + } + }() + return nil +} + +// Delete removes a key. A key that was not there is not an error. +func (s State) Delete(key string) error { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return err + } + if err := kv.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return nil +} diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index cc01c15..32c1e72 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -6,6 +6,7 @@ import ( "errors" "strings" "testing" + "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -94,3 +95,182 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) { t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error()) } } + +// 1840/S1: given a process connection granted its bucket, when it puts keys whole and lists them, then Put +// returns a new revision each time and Keys lists them sorted (none for a bucket with none). +func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + s := conn.State("sends") + + if keys, err := s.Keys(); err != nil || len(keys) != 0 { + t.Fatalf("Keys of an empty bucket: %v, %v", keys, err) + } + rev, err := s.Put("b", map[string]any{"v": 1}) + if err != nil || rev == 0 { + t.Fatalf("Put on a granted bucket: %d, %v", rev, err) + } + rev2, err := s.Put("b", map[string]any{"v": 2}) + if err != nil || rev2 <= rev { + t.Fatalf("Put over a key: %d, %v", rev2, err) + } + if _, err := s.Put("a", 1); err != nil { + t.Fatal(err) + } + keys, err := s.Keys() + if err != nil || strings.Join(keys, ",") != "a,b" { + t.Fatalf("Keys: %v, %v, want a,b", keys, err) + } +} + +// 1840/S2: given a process connection granted its bucket, which already holds two keys and a deleted one, +// when it watches, then it gets the two current values marked Current, and then a later put and a delete in +// order; a narrowed watch gets only the keys it names. +func TestStateWatchOnAGrantedBucket(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + kv, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}) + if err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } + ctx := ctxFor(t) + for k, v := range map[string]string{"a.x": `1`, "a.y": `2`, "b.z": `3`, "gone": `4`} { + if _, err := kv.Put(ctx, k, []byte(v)); err != nil { + t.Fatal(err) + } + } + if err := kv.Delete(ctx, "gone"); err != nil { + t.Fatal(err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + s := conn.State("sends") + + all := make(chan stdio.StateChange, 16) + if err := s.Watch("", func(c stdio.StateChange) error { all <- c; return nil }); err != nil { + t.Fatalf("Watch on a granted bucket: %v", err) + } + current := map[string]bool{} + for len(all) > 0 { + c := <-all + if !c.Current || c.Op != "put" { + t.Fatalf("a current value arrived as %+v", c) + } + current[c.Key] = true + } + if len(current) != 3 || current["gone"] { + t.Fatalf("current values %v, want a.x, a.y, b.z and not gone", current) + } + + narrow := make(chan stdio.StateChange, 16) + if err := s.Watch("a.*", func(c stdio.StateChange) error { narrow <- c; return nil }); err != nil { + t.Fatal(err) + } + if len(narrow) != 2 { + t.Fatalf("a narrowed watch handed %d current values, want 2", len(narrow)) + } + // the current values come in the order of their last revision, not of their keys: taken as a set. + narrowCurrent := map[string]bool{} + for len(narrow) > 0 { + c := <-narrow + if !c.Current || c.Op != "put" { + t.Fatalf("a narrowed watch's current value arrived as %+v", c) + } + narrowCurrent[c.Key] = true + } + if !narrowCurrent["a.x"] || !narrowCurrent["a.y"] { + t.Fatalf("a narrowed watch's current values %v, want a.x and a.y", narrowCurrent) + } + + if _, err := s.Put("a.x", 5); err != nil { + t.Fatal(err) + } + if err := kv.Delete(ctx, "a.y"); err != nil { + t.Fatal(err) + } + for _, want := range []struct{ key, op string }{{"a.x", "put"}, {"a.y", "delete"}} { + select { + case c := <-all: + if c.Key != want.key || c.Op != want.op || c.Current { + t.Fatalf("got %+v, want %s %s, not current", c, want.op, want.key) + } + case <-time.After(10 * time.Second): + t.Fatalf("no %s of %s", want.op, want.key) + } + } + if _, err := s.Put("b.q", 6); err != nil { + t.Fatal(err) + } + for _, want := range []string{"a.x", "a.y"} { + select { + case c := <-narrow: + if c.Key != want || c.Current { + t.Fatalf("narrowed watch got %+v, want the change of %s", c, want) + } + case <-time.After(10 * time.Second): + t.Fatalf("narrowed watch missed %s", want) + } + } + select { + case c := <-narrow: + t.Fatalf("narrowed watch got %+v for a key it did not name", c) + case <-time.After(500 * time.Millisecond): + } +} + +// 1840/AC8: given an account not granted a bucket, when it watches it, then the refusal names the bucket. +func TestStateWatchRefusesAnUngrantedBucketNamingIt(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_unrelated"}); err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + err = conn.State("unrelated").Watch("", func(stdio.StateChange) error { return nil }) + if err == nil || !errors.Is(err, nats.ErrPermissionViolation) || !strings.Contains(err.Error(), "mesh-mail_unrelated") { + t.Fatalf("Watch on an ungranted bucket: %v, want a refusal naming mesh-mail_unrelated", err) + } +} + +// 1840/S1: given a process connection granted its bucket, when it deletes a key, then the key is gone and +// deleting it again is no error. +func TestStateDeleteOnAGrantedBucket(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + s := conn.State("sends") + if _, err := s.Put("k", 1); err != nil { + t.Fatal(err) + } + if err := s.Delete("k"); err != nil { + t.Fatalf("Delete on a granted bucket: %v", err) + } + if e, err := s.Get("k"); err != nil || e != nil { + t.Fatalf("Get after Delete: %+v, %v", e, err) + } + if err := s.Delete("k"); err != nil { + t.Fatalf("Delete of an absent key: %v", err) + } +} diff --git a/go/process/state_test.go b/go/process/state_test.go index 542153b..02b084a 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -3,8 +3,11 @@ package process import ( "context" "errors" + "sort" + "strings" "sync" "testing" + "time" "github.com/nats-io/nats.go/jetstream" @@ -32,6 +35,9 @@ type fakeKV struct { values map[string][]byte revs map[string]uint64 next uint64 + // initial is what a Watch hands first, before its nil marker; watcher is the last one made. + initial []jetstream.KeyValueEntry + watcher *fakeWatcher } func newFakeKV() *fakeKV { return &fakeKV{values: map[string][]byte{}, revs: map[string]uint64{}} } @@ -68,6 +74,58 @@ func (f *fakeKV) Update(_ context.Context, key string, value []byte, revision ui return f.next, nil } +func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.next++ + f.values[key], f.revs[key] = value, f.next + return f.next, nil +} + +func (f *fakeKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { + f.mu.Lock() + defer f.mu.Unlock() + delete(f.values, key) + delete(f.revs, key) + return nil +} + +func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.values) == 0 { + return nil, jetstream.ErrNoKeysFound + } + keys := make([]string, 0, len(f.values)) + for k := range f.values { + keys = append(keys, k) + } + sort.Sort(sort.Reverse(sort.StringSlice(keys))) // as a bucket lists them: in no order the caller may rely on + return keys, nil +} + +// fakeWatcher is a jetstream.KeyWatcher fed by a test. +type fakeWatcher struct { + jetstream.KeyWatcher + updates chan jetstream.KeyValueEntry + pattern string +} + +func (w *fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.updates } +func (w *fakeWatcher) Stop() error { return nil } + +func (f *fakeKV) Watch(_ context.Context, pattern string, _ ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + f.mu.Lock() + defer f.mu.Unlock() + w := &fakeWatcher{updates: make(chan jetstream.KeyValueEntry, 16), pattern: pattern} + f.watcher = w + for _, e := range f.initial { + w.updates <- e + } + w.updates <- nil + return w, nil +} + func stateOn(kv jetstream.KeyValue) State { return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }} } @@ -103,3 +161,144 @@ func TestStateSatisfiesAskerStore(t *testing.T) { t.Fatal("Update did not advance the revision") } } + +// 1840/S1: given a state, when a key is put twice, then each Put returns a new revision and the key reads +// back the last value, whether or not it had one. +func TestStatePutWritesAKeyWhole(t *testing.T) { + s := stateOn(newFakeKV()) + rev, err := s.Put("k", map[string]any{"a": 1}) + if err != nil || rev == 0 { + t.Fatalf("Put of a new key: %d, %v", rev, err) + } + rev2, err := s.Put("k", map[string]any{"a": 2}) + if err != nil || rev2 <= rev { + t.Fatalf("Put over a key: %d, %v, want a revision after %d", rev2, err, rev) + } + e, err := s.Get("k") + if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` { + t.Fatalf("Get after Put: %+v, %v", e, err) + } +} + +// 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error. +func TestStateKeysAreSortedAndEmptyWhenNone(t *testing.T) { + s := stateOn(newFakeKV()) + if keys, err := s.Keys(); err != nil || len(keys) != 0 { + t.Fatalf("Keys of an empty state: %v, %v", keys, err) + } + for _, k := range []string{"b", "c", "a"} { + if _, err := s.Put(k, 1); err != nil { + t.Fatal(err) + } + } + keys, err := s.Keys() + if err != nil || strings.Join(keys, ",") != "a,b,c" { + t.Fatalf("Keys: %v, %v, want a,b,c", keys, err) + } +} + +type watchEntry struct { + fakeEntry + op jetstream.KeyValueOp +} + +func (e watchEntry) Operation() jetstream.KeyValueOp { return e.op } + +// 1840/S2: given a state with current values (one of them deleted), when it is watched, then the handler gets +// each current value marked Current and not the deleted one, Watch returns, and a later put and delete arrive +// in order unmarked. +func TestStateWatchHandsCurrentValuesThenChanges(t *testing.T) { + kv := newFakeKV() + kv.initial = []jetstream.KeyValueEntry{ + watchEntry{fakeEntry{key: "a", value: []byte(`1`), rev: 1}, jetstream.KeyValuePut}, + watchEntry{fakeEntry{key: "gone", rev: 2}, jetstream.KeyValueDelete}, + watchEntry{fakeEntry{key: "b", value: []byte(`2`), rev: 3}, jetstream.KeyValuePut}, + } + got := make(chan stdio.StateChange, 8) + s := stateOn(kv) + if err := s.Watch("", func(c stdio.StateChange) error { got <- c; return nil }); err != nil { + t.Fatal(err) + } + for _, want := range []stdio.StateChange{ + {State: "sends", Key: "a", Op: "put", Value: []byte(`1`), Revision: 1, Current: true}, + {State: "sends", Key: "b", Op: "put", Value: []byte(`2`), Revision: 3, Current: true}, + } { + select { + case c := <-got: + if c.Key != want.Key || c.Op != want.Op || string(c.Value) != string(want.Value) || c.Revision != want.Revision || c.Current != want.Current || c.State != want.State { + t.Fatalf("got %+v, want %+v", c, want) + } + default: + t.Fatalf("Watch returned before handling %+v", want) + } + } + if len(got) != 0 { + t.Fatalf("a deleted key was handed: %+v", <-got) + } + kv.watcher.updates <- watchEntry{fakeEntry{key: "a", value: []byte(`3`), rev: 4}, jetstream.KeyValuePut} + kv.watcher.updates <- watchEntry{fakeEntry{key: "b", rev: 5}, jetstream.KeyValueDelete} + for _, want := range []stdio.StateChange{ + {Key: "a", Op: "put", Revision: 4}, {Key: "b", Op: "delete", Revision: 5}, + } { + select { + case c := <-got: + if c.Key != want.Key || c.Op != want.Op || c.Revision != want.Revision || c.Current { + t.Fatalf("got %+v, want %+v", c, want) + } + case <-time.After(5 * time.Second): + t.Fatalf("no change %+v", want) + } + } +} + +// 1840/S2: a key narrowing reaches the bus as a subject pattern: `*` one name, `**` the rest, empty every key. +func TestStateWatchNarrowing(t *testing.T) { + for key, want := range map[string]string{"": ">", "a.b": "a.b", "a.*": "a.*", "a.**": "a.>", "**": ">"} { + if got := watchPattern(key); got != want { + t.Errorf("watchPattern(%q) = %q, want %q", key, got, want) + } + } +} + +// 1840/S2: given a handler that refuses a current value, then Watch returns its error. +func TestStateWatchReturnsAHandlersErrorOnCurrentValues(t *testing.T) { + kv := newFakeKV() + kv.initial = []jetstream.KeyValueEntry{watchEntry{fakeEntry{key: "a", value: []byte(`1`), rev: 1}, jetstream.KeyValuePut}} + boom := errors.New("boom") + if err := stateOn(kv).Watch("", func(stdio.StateChange) error { return boom }); !errors.Is(err, boom) { + t.Fatalf("Watch: %v, want the handler's error", err) + } +} + +// 1840/S2: given a connection that closes before the current values arrive, then Watch returns an error. +func TestStateWatchEndsWithAnErrorWhenTheConnectionCloses(t *testing.T) { + s := State{Module: "mesh-mail", Name: "sends", closed: func() bool { return true }, + kv: func(context.Context) (jetstream.KeyValue, error) { return silentKV{}, nil }} + if err := s.Watch("", func(stdio.StateChange) error { return nil }); err == nil { + t.Fatal("Watch on a closed connection returned no error") + } +} + +// silentKV watches but never delivers. +type silentKV struct{ jetstream.KeyValue } + +func (silentKV) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return &fakeWatcher{updates: make(chan jetstream.KeyValueEntry)}, nil +} + +// 1840/S1: given a key, when it is deleted, then it reads as absent; deleting a key that was not there is no error. +func TestStateDeleteRemovesAKeyAndAnAbsentOneIsNoError(t *testing.T) { + s := stateOn(newFakeKV()) + if _, err := s.Put("k", 1); err != nil { + t.Fatal(err) + } + if err := s.Delete("k"); err != nil { + t.Fatalf("Delete of a key: %v", err) + } + if e, err := s.Get("k"); err != nil || e != nil { + t.Fatalf("Get after Delete: %+v, %v, want absent", e, err) + } + if err := s.Delete("never"); err != nil { + t.Fatalf("Delete of an absent key: %v", err) + } +}