fix(BucketWatcherManager): release mu before watcher.Start to avoid self-deadlock

cmd/opa-nats/docker-compose.test.override.yaml

@@ -0,0 +1,24 @@
+# Override for TestIntegration (cmd/opa-nats/main_test.go).
+#
+#  - Suppresses host-port bindings on services we don't talk to from
+#    the test runner so we don't collide with whatever else may be
+#    using those ports on the host (in CI, the workflow's
+#    `services: nats:` already binds 4222). OPA's 8181 is the only port
+#    the test reaches, so it stays published.
+#  - Mounts ./config-compose.yaml in place of the example config.yaml so
+#    OPA points at NATS via the docker-compose service name `nats:4222`
+#    rather than the example's host-friendly `localhost:4222` default.
+#  - Adds --pprof to OPA so the test can grab a goroutine dump from
+#    /debug/pprof/goroutine?debug=2 if a request hangs.
+services:
+  nats:
+    ports: !reset []
+  nats-ui:
+    ports: !reset []
+  opa:
+    command: ["run", "--server", "--config-file=/config/config.yaml", "/policies", "--addr", "0.0.0.0:8181", "--pprof"]
+    volumes:
+      # Override the default ./config.yaml mount so OPA uses the in-compose
+      # config (server_url=nats://nats:4222) instead of the host-friendly
+      # default. Path is relative to docker-compose.yaml's directory.
+      - ./config-compose.yaml:/config/config.yaml

cmd/opa-nats/main_test.go

@@ -32,6 +32,18 @@ func TestIntegration(t *testing.T) {
 		t.Fatalf("Examples directory does not exist: %s", examplesDir)
 	}
 
+	// Resolve the test-only compose override that drops host-port bindings on
+	// services we don't talk to (NATS, NATS UI). Only OPA's 8181 needs to be
+	// reachable from the test runner. This avoids host-port collisions with
+	// e.g. the GHA workflow's `services: nats:` (which already binds 4222).
+	pwd, err := os.Getwd()
+	require.NoError(t, err)
+	overrideFile, err := filepath.Abs(filepath.Join(pwd, "docker-compose.test.override.yaml"))
+	require.NoError(t, err)
+	if _, err := os.Stat(overrideFile); err != nil {
+		t.Fatalf("Test override compose file missing at %s: %v", overrideFile, err)
+	}
+
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 
@@ -43,23 +55,59 @@ func TestIntegration(t *testing.T) {
 	err = os.Chdir(examplesDir)
 	require.NoError(t, err)
 
-	// Clean up any existing containers
+	// Clean up any existing containers (also runs at end via t.Cleanup).
+	composeArgs := []string{"compose", "-f", "docker-compose.yaml", "-f", overrideFile}
+	dumpComposeLogs := func(reason string) {
+		t.Logf("Dumping diagnostics (%s):", reason)
+		// OPA goroutine dump first — most useful for diagnosing hung requests.
+		t.Logf("--- BEGIN OPA goroutine dump ---\n%s\n--- END OPA goroutine dump ---", fetchOPAGoroutineDump())
+		// Bound each docker invocation: if the daemon itself is wedged
+		// (rare, but seen in CI under OOM-kill aftermath) the cleanup
+		// would otherwise hang past the test framework's 10-min kill and
+		// no diagnostics would land. Match the 10s timeout already used
+		// by fetchOPAGoroutineDump.
+		runDocker := func(args []string) []byte {
+			ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+			defer cancel()
+			out, _ := exec.CommandContext(ctx, "docker", args...).CombinedOutput()
+			return out
+		}
+		logsArgs := append([]string{}, composeArgs...)
+		logsArgs = append(logsArgs, "logs", "--no-color", "--tail", "200")
+		t.Logf("--- BEGIN compose logs ---\n%s\n--- END compose logs ---", string(runDocker(logsArgs)))
+		psArgs := append([]string{}, composeArgs...)
+		psArgs = append(psArgs, "ps", "-a")
+		t.Logf("--- compose ps ---\n%s", string(runDocker(psArgs)))
+	}
 	t.Cleanup(func() {
-		os.Chdir(examplesDir)
-		exec.Command("docker", "compose", "down", "-v").Run()
+		_ = os.Chdir(examplesDir)
+		if t.Failed() {
+			dumpComposeLogs("test failed")
+		}
+		downArgs := append([]string{}, composeArgs...)
+		downArgs = append(downArgs, "down", "-v", "--remove-orphans")
+		exec.Command("docker", downArgs...).Run()
 	})
+	preDown := append([]string{}, composeArgs...)
+	preDown = append(preDown, "down", "-v", "--remove-orphans")
+	_ = exec.Command("docker", preDown...).Run()
 
-	// Start docker-compose services
+	// Start docker-compose services with the test override applied.
 	t.Log("Starting docker-compose services...")
-	cmd := exec.CommandContext(ctx, "docker", "compose", "up", "-d", "--build")
+	upArgs := append([]string{}, composeArgs...)
+	upArgs = append(upArgs, "up", "-d", "--build")
+	cmd := exec.CommandContext(ctx, "docker", upArgs...)
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
 	err = cmd.Run()
 	require.NoError(t, err)
 
 	// Wait for OPA to be ready
 	t.Log("Waiting for OPA to be ready...")
-	require.NoError(t, waitForOPA(ctx, "http://localhost:8181"))
+	if err := waitForOPA(ctx, "http://localhost:8181"); err != nil {
+		dumpComposeLogs("OPA never became ready")
+		t.Fatalf("OPA never became ready: %v", err)
+	}
 
 	// Test OPA health endpoint
 	t.Run("OPA Health Check", func(t *testing.T) {
@@ -83,7 +131,13 @@ func TestIntegration(t *testing.T) {
 		}
 	})
 
-	// Test bucket watching behavior - first call should show bucket_watched: false, second: true
+	// Test bucket watching behavior. The first call to nats.kv.watch_bucket
+	// for a fresh bucket spawns an async goroutine that loads bucket data
+	// and registers the watcher. We expect bucket_watched=false on the very
+	// first call, then bucket_watched=true on a subsequent call once the
+	// async registration has completed. Because that registration happens
+	// off the request-handling path, we poll instead of asserting that the
+	// second call alone is enough.
 	t.Run("Bucket Watching Behavior", func(t *testing.T) {
 		bucketID := "550e8400-e29b-41d4-a716-446655440000"
 		input := map[string]interface{}{
@@ -107,59 +161,105 @@ func TestIntegration(t *testing.T) {
 		require.True(t, ok, "x should be present")
 		t.Logf("First call - bucket_watched: %v, x: %+v", bucketWatched1, x1)
 
-		// Second call - should show bucket_watched: true (bucket is now cached)
-		result2 := evaluatePolicy(t, "data.test", input)
-		t.Logf("Second call result: %+v", result2)
-
-		require.Contains(t, result2, "result")
-		resultData2 := result2["result"].(map[string]interface{})
-
-		// Should have bucket_watched: true on second call
-		bucketWatched2, ok := resultData2["bucket_watched"]
-		require.True(t, ok, "bucket_watched should be present")
-		assert.True(t, bucketWatched2.(bool), "Second call should have bucket_watched: true")
-
-		// Should have x field
-		x2, ok := resultData2["x"]
-		require.True(t, ok, "x should be present")
-		t.Logf("Second call - bucket_watched: %v, x: %+v", bucketWatched2, x2)
-
-		// x should be the same in both calls
-		assert.Equal(t, x1, x2, "x should be the same in both calls")
-
-		t.Log("✅ Bucket watching behavior verified: first call cached bucket, second call used cached data")
+		// Poll until the async watcher registration completes (bucket_watched=true)
+		// or we exceed the deadline.
+		var resultData2 map[string]interface{}
+		require.Eventually(t, func() bool {
+			r := evaluatePolicy(t, "data.test", input)
+			rd, ok := r["result"].(map[string]interface{})
+			if !ok {
+				return false
+			}
+			bw, ok := rd["bucket_watched"].(bool)
+			if !ok {
+				return false
+			}
+			if bw {
+				resultData2 = rd
+			}
+			return bw
+		}, 30*time.Second, 200*time.Millisecond, "bucket_watched never became true within 30s")
+
+		t.Logf("Subsequent call - bucket_watched: true, x: %+v", resultData2["x"])
+
+		// The two calls intentionally return different shapes for `x`:
+		//   bucket_watched=false → x = nats.kv.get_data(bucket, "members")  (members map only)
+		//   bucket_watched=true  → x = data.nats.kv[bucket]                  (entire bucket)
+		// What we verify is that the members data the unwatched path returns
+		// matches the members data the watched path stores under the bucket.
+		x1Members, ok1 := x1.(map[string]interface{})
+		require.True(t, ok1, "x1 should be a map (members)")
+		x2Bucket, ok2 := resultData2["x"].(map[string]interface{})
+		require.True(t, ok2, "x2 should be a map (bucket)")
+		x2Members, ok2m := x2Bucket["members"].(map[string]interface{})
+		require.True(t, ok2m, "x2 should contain members map")
+		assert.Equal(t, x1Members, x2Members, "members data should match between unwatched and watched calls")
+
+		t.Log("Bucket watching behavior verified: bucket_watched flips to true once the async watcher registration completes")
 	})
 
-	// Test that x returns consistent data regardless of bucket watching state
+	// Test that x returns consistent data regardless of bucket watching state.
+	// Use a different bucket so we do not piggyback on the watcher registered
+	// by the previous subtest.
 	t.Run("Data Consistency Test", func(t *testing.T) {
 		bucketID := "660e8400-e29b-41d4-a716-446655440001" // Developers group
 		input := map[string]interface{}{
 			"bucket_id": bucketID,
 		}
 
-		// Call multiple times and verify x is consistent
-		results := make([]map[string]interface{}, 3)
-		for i := 0; i < 3; i++ {
-			result := evaluatePolicy(t, "data.test", input)
-			require.Contains(t, result, "result")
-			results[i] = result["result"].(map[string]interface{})
-			t.Logf("Call %d result: %+v", i+1, results[i])
-		}
-
-		// First call should be bucket_watched: false, subsequent calls: true
-		assert.False(t, results[0]["bucket_watched"].(bool), "First call should have bucket_watched: false")
-		assert.True(t, results[1]["bucket_watched"].(bool), "Second call should have bucket_watched: true")
-		assert.True(t, results[2]["bucket_watched"].(bool), "Third call should have bucket_watched: true")
+		// First call
+		first := evaluatePolicy(t, "data.test", input)
+		require.Contains(t, first, "result")
+		firstData := first["result"].(map[string]interface{})
+		t.Logf("First call result: %+v", firstData)
+		assert.False(t, firstData["bucket_watched"].(bool), "First call should have bucket_watched: false")
+
+		// Wait for the async registration before checking subsequent calls.
+		var watchedData map[string]interface{}
+		require.Eventually(t, func() bool {
+			r := evaluatePolicy(t, "data.test", input)
+			rd, ok := r["result"].(map[string]interface{})
+			if !ok {
+				return false
+			}
+			bw, _ := rd["bucket_watched"].(bool)
+			if bw {
+				watchedData = rd
+			}
+			return bw
+		}, 30*time.Second, 200*time.Millisecond, "bucket_watched never became true within 30s")
 
-		// All x values should be the same
-		x0 := results[0]["x"]
-		x1 := results[1]["x"]
-		x2 := results[2]["x"]
+		results := []map[string]interface{}{firstData, watchedData}
 
-		assert.Equal(t, x0, x1, "x should be same between first and second call")
-		assert.Equal(t, x1, x2, "x should be same between second and third call")
+		// One more call after the watcher is registered — should still see true.
+		extra := evaluatePolicy(t, "data.test", input)
+		require.Contains(t, extra, "result")
+		extraData := extra["result"].(map[string]interface{})
+		t.Logf("Post-watcher call result: %+v", extraData)
+		results = append(results, extraData)
 
-		t.Logf("✅ Data consistency verified: x remains %+v across all calls", x0)
+		assert.False(t, results[0]["bucket_watched"].(bool), "First call should have bucket_watched: false")
+		assert.True(t, results[1]["bucket_watched"].(bool), "Post-registration call should have bucket_watched: true")
+		assert.True(t, results[2]["bucket_watched"].(bool), "Subsequent call should have bucket_watched: true")
+
+		// As in the previous subtest, the bucket_watched=false branch returns
+		// only the `members` submap, while the bucket_watched=true branch
+		// returns the entire bucket. Compare the members slice from each
+		// representation rather than the raw `x` values.
+		x0Members, ok0 := results[0]["x"].(map[string]interface{})
+		require.True(t, ok0, "x[0] should be a members map")
+		x1Bucket, ok1 := results[1]["x"].(map[string]interface{})
+		require.True(t, ok1, "x[1] should be a bucket map")
+		x2Bucket, ok2 := results[2]["x"].(map[string]interface{})
+		require.True(t, ok2, "x[2] should be a bucket map")
+
+		x1Members, _ := x1Bucket["members"].(map[string]interface{})
+		x2Members, _ := x2Bucket["members"].(map[string]interface{})
+
+		assert.Equal(t, x0Members, x1Members, "members data should match between unwatched and post-watcher calls")
+		assert.Equal(t, x1Members, x2Members, "members data should be stable across consecutive watched calls")
+
+		t.Logf("Data consistency verified: members data is stable across watched/unwatched and consecutive calls")
 	})
 }
 
@@ -184,7 +284,13 @@ func waitForOPA(ctx context.Context, url string) error {
 	}
 }
 
-// evaluatePolicy evaluates a policy with input data
+// evaluatePolicy evaluates a policy with input data.
+//
+// Uses a per-request HTTP timeout of 30s. If OPA hangs (e.g. policy
+// evaluation deadlocks), the test fails with a clear timeout error
+// within 30s instead of hanging until the Go test framework's 10-minute
+// kill — which would prevent t.Cleanup from running and dumping
+// compose logs.
 func evaluatePolicy(t *testing.T, path string, input map[string]interface{}) map[string]interface{} {
 	payload := map[string]interface{}{
 		"input": input,
@@ -200,8 +306,9 @@ func evaluatePolicy(t *testing.T, path string, input map[string]interface{}) map
 	}
 
 	url := fmt.Sprintf("http://localhost:8181/v1/data/%s", apiPath)
-	resp, err := http.Post(url, "application/json", bytes.NewBuffer(jsonData))
-	require.NoError(t, err)
+	client := &http.Client{Timeout: 30 * time.Second}
+	resp, err := client.Post(url, "application/json", bytes.NewBuffer(jsonData))
+	require.NoError(t, err, "POST %s timed out or failed (OPA likely hung) — see compose logs in cleanup", url)
 	defer resp.Body.Close()
 
 	require.Equal(t, http.StatusOK, resp.StatusCode)
@@ -215,3 +322,23 @@ func evaluatePolicy(t *testing.T, path string, input map[string]interface{}) map
 
 	return result
 }
+
+// fetchOPAGoroutineDump returns OPA's full goroutine stack via pprof.
+// OPA must be started with --pprof for this to work; the test compose
+// override adds it. Returns the empty string if pprof is unavailable.
+func fetchOPAGoroutineDump() string {
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Get("http://localhost:8181/debug/pprof/goroutine?debug=2")
+	if err != nil {
+		return fmt.Sprintf("(failed to fetch goroutine dump: %v)", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusOK {
+		return fmt.Sprintf("(pprof returned HTTP %d)", resp.StatusCode)
+	}
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return fmt.Sprintf("(failed to read pprof body: %v)", err)
+	}
+	return string(body)
+}

examples/opa-nats/config-compose.yaml

@@ -1,6 +1,9 @@
+# OPA config for running INSIDE the docker-compose stack (where OPA
+# reaches NATS via the docker network service name `nats`). For running
+# OPA directly on the host, use config.yaml instead.
 plugins:
   nats:
-    server_url: "nats://localhost:4222"
+    server_url: "nats://nats:4222"
     ttl: "10m"
     refresh_interval: "30s"
     max_reconnect_attempts: 10

examples/opa-nats/config.yaml

@@ -1,5 +1,10 @@
 plugins:
   nats:
+    # Defaults to localhost so this example works out of the box for users
+    # running OPA directly on their host. The included docker-compose stack
+    # overrides this via cmd/opa-nats/docker-compose.test.override.yaml when
+    # OPA runs inside the compose network and needs to reach NATS via the
+    # docker service name `nats`.
     server_url: "nats://localhost:4222"
     ttl: "10m"
     refresh_interval: "30s"