fix: Scope Cloud API reply inboxes to their network

README.md

@@ -114,6 +114,10 @@ compileOnly 'app.simplecloud.controller:controller-api:VERSION'
 Contributions to SimpleCloud are welcome and highly appreciated. However, before you jump right into it, we would like
 you to read our [Contribution Guide][docs-contribute].
 
+## NATS integration tests
+
+With an isolated NATS broker running, use `NATS_TEST_URL=nats://127.0.0.1:4222 ./gradlew :api:test --rerun-tasks` to test request inboxes before and after reconnecting. The broker test is skipped when the variable is unset. The [Go SDK README](go/README.md) describes the equivalent Go tests.
+
 ## License
 
 This repository is licensed under [Apache 2.0][license].

api/src/main/java/app/simplecloud/api/internal/nats/NatsFailoverConnectionManager.java

@@ -204,6 +204,7 @@ static Options createOptions(
         return Options.builder()
                 .server(natsUrl)
                 .userInfo(networkId, networkSecret)
+                .inboxPrefix(networkId + "._INBOX")
                 .maxReconnects(-1)
                 .errorListener(listener)
                 .connectionListener(listener)

api/src/test/java/app/simplecloud/api/internal/nats/NetworkScopedInboxIntegrationTest.java

@@ -0,0 +1,62 @@
+package app.simplecloud.api.internal.nats;
+
+import io.nats.client.Connection;
+import io.nats.client.Message;
+import io.nats.client.Nats;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+class NetworkScopedInboxIntegrationTest {
+
+    // Run against an isolated broker with NATS_TEST_URL=nats://127.0.0.1:4222.
+    @Test
+    void requestsUseNetworkInboxesAfterConnectingAndReconnecting() throws Exception {
+        String url = System.getenv("NATS_TEST_URL");
+        assumeTrue(url != null && !url.isBlank(), "set NATS_TEST_URL to run broker integration tests");
+
+        String subject = "network-1.test." + UUID.randomUUID();
+        try (Connection responder = Nats.connect(url)) {
+            responder.createDispatcher(message -> responder.publish(
+                    message.getReplyTo(), message.getReplyTo().getBytes(StandardCharsets.UTF_8)
+            )).subscribe(subject);
+            responder.flush(Duration.ofSeconds(5));
+
+            NatsFailoverConnectionManager manager = new NatsFailoverConnectionManager(
+                    url, "network-1", "secret", Duration.ofSeconds(30)
+            );
+            try {
+                Connection connection = manager.getConnection();
+                awaitConnected(connection);
+                assertScopedRequest(connection, subject);
+                connection.forceReconnect();
+                awaitConnected(connection);
+                assertScopedRequest(connection, subject);
+            } finally {
+                manager.shutdown();
+            }
+        }
+    }
+
+    private static void assertScopedRequest(Connection connection, String subject) throws Exception {
+        assertTrue(connection.createInbox().startsWith("network-1._INBOX."));
+        Message response = connection.request(subject, new byte[0], Duration.ofSeconds(5));
+        assertNotNull(response, "request timed out");
+        assertTrue(new String(response.getData(), StandardCharsets.UTF_8).startsWith("network-1._INBOX."));
+    }
+
+    private static void awaitConnected(Connection connection) throws InterruptedException {
+        long deadline = System.nanoTime() + Duration.ofSeconds(10).toNanos();
+        while (connection.getStatus() != Connection.Status.CONNECTED && System.nanoTime() < deadline) {
+            Thread.sleep(10);
+        }
+        assertEquals(Connection.Status.CONNECTED, connection.getStatus());
+    }
+}

api/src/test/java/app/simplecloud/api/internal/nats/SimpleCloudNatsListenerTest.java

@@ -13,6 +13,7 @@
 import java.util.logging.LogRecord;
 import java.util.logging.Logger;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -82,7 +83,7 @@ void retainsDefaultLoggingForUnexpectedExceptions() {
     }
 
     @Test
-    void failoverConnectionsInstallSimpleCloudListenerForErrorsAndConnectionEvents() {
+    void failoverConnectionsUseNetworkInboxesAndSimpleCloudListeners() {
         SimpleCloudNatsListener listener = new SimpleCloudNatsListener();
         Options options = NatsFailoverConnectionManager.createOptions(
                 "nats://localhost:4222",
@@ -93,6 +94,7 @@ void failoverConnectionsInstallSimpleCloudListenerForErrorsAndConnectionEvents()
 
         assertSame(listener, options.getErrorListener());
         assertSame(listener, options.getConnectionListener());
+        assertEquals("network._INBOX.", options.getInboxPrefix());
     }
 
     private static List<LogRecord> captureDefaultErrorLogger(Runnable action) {

go/README.md

@@ -59,6 +59,10 @@ func main() {
 
 You can also pass `Options` directly to `NewClient`, including a custom `http.Client` or additional NATS options.
 
+NATS request reply subjects use `<networkId>._INBOX.<random>`. The SDK applies this network prefix after `NATSOptions`, so a custom inbox prefix cannot override it. Existing responders continue replying to the request's supplied reply subject.
+
+To run the NATS connection and reconnect integration tests, start an isolated NATS broker and run `NATS_TEST_URL=nats://127.0.0.1:4222 go test ./...`. Without this variable, the broker integration tests are skipped.
+
 ## Example project
 
 A small runnable command is available in [`examples/basic`](examples/basic). It loads the standard environment variables, lists the network's groups, and prints their names and IDs:

go/sdk/nats.go

@@ -22,6 +22,9 @@ func (c *Client) NATS() (*nats.Conn, error) {
 		nats.Timeout(5 * time.Second),
 	}
 	options = append(options, c.options.NATSOptions...)
+	// Reply subjects must stay within this network, including when callers
+	// supply their own connection options.
+	options = append(options, nats.CustomInboxPrefix(c.options.NetworkID+"._INBOX"))
 	connection, err := nats.Connect(c.options.NATSURL, options...)
 	if err != nil {
 		return nil, fmt.Errorf("connect to SimpleCloud NATS: %w", err)

go/sdk/nats_integration_test.go

@@ -0,0 +1,94 @@
+package simplecloud
+
+import (
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// Run against an isolated broker with NATS_TEST_URL=nats://127.0.0.1:4222.
+func TestNetworkScopedRequestInboxes(t *testing.T) {
+	url := os.Getenv("NATS_TEST_URL")
+	if url == "" {
+		t.Skip("set NATS_TEST_URL to run broker integration tests")
+	}
+	responder, err := nats.Connect(url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer responder.Close()
+	subject := "network-1.test." + nats.NewInbox()
+	_, err = responder.Subscribe(subject, func(msg *nats.Msg) {
+		_ = msg.Respond([]byte(msg.Reply))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := responder.Flush(); err != nil {
+		t.Fatal(err)
+	}
+
+	for _, oldStyle := range []bool{false, true} {
+		name := "multiplexed"
+		if oldStyle {
+			name = "old-style"
+		}
+		t.Run(name, func(t *testing.T) {
+			options := []nats.Option{nats.CustomInboxPrefix("_INBOX"), nats.Name(name)}
+			if oldStyle {
+				options = append(options, nats.UseOldRequestStyle())
+			}
+			client, err := NewClient(Options{
+				NetworkID: "network-1", NetworkSecret: "secret", NATSURL: url, NATSOptions: options,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() { _ = client.Close() })
+			connection, err := client.NATS()
+			if err != nil {
+				t.Fatal(err)
+			}
+			assertScoped := func(connection *nats.Conn) {
+				t.Helper()
+				if connection.Opts.Name != name {
+					t.Fatal("custom connection options were not preserved")
+				}
+				if inbox := connection.NewInbox(); !strings.HasPrefix(inbox, "network-1._INBOX.") {
+					t.Fatalf("unexpected generated inbox %q", inbox)
+				}
+				response, err := connection.Request(subject, nil, 5*time.Second)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if !strings.HasPrefix(string(response.Data), "network-1._INBOX.") {
+					t.Fatalf("unexpected request reply subject %q", response.Data)
+				}
+			}
+			assertScoped(connection)
+			if err := connection.ForceReconnect(); err != nil {
+				t.Fatal(err)
+			}
+			deadline := time.Now().Add(10 * time.Second)
+			for !connection.IsConnected() && time.Now().Before(deadline) {
+				time.Sleep(10 * time.Millisecond)
+			}
+			if !connection.IsConnected() {
+				t.Fatal("connection did not reconnect")
+			}
+			assertScoped(connection)
+			connection.Close()
+			replacement, err := client.NATS()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if replacement == connection {
+				t.Fatal("closed connection was reused")
+			}
+			assertScoped(replacement)
+		})
+	}
+}

go/sdk/options.go

@@ -24,7 +24,9 @@ type Options struct {
 	NetworkSecret string
 	Component     string
 	HTTPClient    *http.Client
-	NATSOptions   []nats.Option
+	// NATSOptions customizes the connection. The SDK always sets the inbox
+	// prefix to NetworkID + "._INBOX" after applying these options.
+	NATSOptions []nats.Option
 }
 
 // DefaultOptions returns configuration populated from the environment.