Prove a missing or refused bucket is a watch error, so a node waits on the controller's compositions and tries again (issue 1126)

internal/link/history_integration_test.go

@@ -14,7 +14,10 @@ import (
 // The revisions a bucket keeps of a key, read over a real bus: oldest first, a delete and a purge said as deleted, and
 // none for a key that never was (issue 1945: the node agent takes the composition its want names from them).
 func TestNatsAKeysRevisionsAreReadBack(t *testing.T) {
-	srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	// A second account reads as a node's does: refused the one bucket "denied-it" (the seat's state before its grant arrives).
+	srv, err := natsserver.NewServer(&natsserver.Options{Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true,
+		Users: []*natsserver.User{{Username: "admin", Password: "a"}, {Username: "node", Password: "n", Permissions: &natsserver.Permissions{
+			Publish: &natsserver.SubjectPermission{Allow: []string{">"}, Deny: []string{"$JS.API.*.*.KV_denied-it", "$JS.API.*.*.*.KV_denied-it", "$JS.API.*.*.*.KV_denied-it.>", "$KV.denied-it.>"}}}}}})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -23,7 +26,7 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) {
 		t.Fatal("the bus did not come up")
 	}
 	t.Cleanup(srv.Shutdown)
-	conn, err := nats.Connect(srv.ClientURL())
+	conn, err := nats.Connect(srv.ClientURL(), nats.UserInfo("admin", "a"))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -65,4 +68,26 @@ func TestNatsAKeysRevisionsAreReadBack(t *testing.T) {
 	if got, err := bus.KVHistory(ctx, "composed-it", "it.never.x"); err != nil || len(got) != 0 {
 		t.Fatalf("a key that never was: %+v (%v)", got, err)
 	}
+
+	// A bucket that is not there, and one the account may not read, are errors at once, never a wait: the node agent
+	// follows its placement key by trying the watch again (endWhenPlaced) and stays on the controller's compositions.
+	if _, err := bus.KVWatch(ctx, "never-it", ">"); err == nil {
+		t.Error("a watch of a bucket that is not there gave no error")
+	}
+	if _, err := js.CreateKeyValue(&nats.KeyValueConfig{Bucket: "denied-it", History: 1}); err != nil {
+		t.Fatal(err)
+	}
+	nodeConn, err := nats.Connect(srv.ClientURL(), nats.UserInfo("node", "n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nodeConn.Close)
+	started := time.Now()
+	wctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+	if _, err := (OverNATS{Conn: nodeConn}).KVWatch(wctx, "denied-it", ">"); err == nil || wctx.Err() != nil {
+		t.Errorf("a watch the account may not make: error %v after %s, context %v", err, time.Since(started), wctx.Err())
+	} else {
+		t.Logf("a refused watch is an error after %s: %v", time.Since(started).Round(time.Millisecond), err)
+	}
 }