From 69161590ee3d5d42649cd4c90f2f8956d7fcfff9 Mon Sep 17 00:00:00 2001 From: Frederick Baier Date: Mon, 14 Sep 2026 18:57:53 +0000 Subject: [PATCH] fix: Scope Cloud API reply inboxes to their network --- README.md | 4 + .../nats/NatsFailoverConnectionManager.java | 1 + .../NetworkScopedInboxIntegrationTest.java | 62 ++++++++++++ .../nats/SimpleCloudNatsListenerTest.java | 4 +- go/README.md | 4 + go/sdk/nats.go | 3 + go/sdk/nats_integration_test.go | 94 +++++++++++++++++++ go/sdk/options.go | 4 +- 8 files changed, 174 insertions(+), 2 deletions(-) create mode 100644 api/src/test/java/app/simplecloud/api/internal/nats/NetworkScopedInboxIntegrationTest.java create mode 100644 go/sdk/nats_integration_test.go diff --git a/README.md b/README.md index 8570aa1..2782dee 100644 --- a/README.md +++ b/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]. diff --git a/api/src/main/java/app/simplecloud/api/internal/nats/NatsFailoverConnectionManager.java b/api/src/main/java/app/simplecloud/api/internal/nats/NatsFailoverConnectionManager.java index 773df5e..e5b490f 100644 --- a/api/src/main/java/app/simplecloud/api/internal/nats/NatsFailoverConnectionManager.java +++ b/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) diff --git a/api/src/test/java/app/simplecloud/api/internal/nats/NetworkScopedInboxIntegrationTest.java b/api/src/test/java/app/simplecloud/api/internal/nats/NetworkScopedInboxIntegrationTest.java new file mode 100644 index 0000000..7a95785 --- /dev/null +++ b/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()); + } +} diff --git a/api/src/test/java/app/simplecloud/api/internal/nats/SimpleCloudNatsListenerTest.java b/api/src/test/java/app/simplecloud/api/internal/nats/SimpleCloudNatsListenerTest.java index 6004495..be2d343 100644 --- a/api/src/test/java/app/simplecloud/api/internal/nats/SimpleCloudNatsListenerTest.java +++ b/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 captureDefaultErrorLogger(Runnable action) { diff --git a/go/README.md b/go/README.md index a4c700d..4afcac5 100644 --- a/go/README.md +++ b/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 `._INBOX.`. 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: diff --git a/go/sdk/nats.go b/go/sdk/nats.go index 8031653..4eea321 100644 --- a/go/sdk/nats.go +++ b/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) diff --git a/go/sdk/nats_integration_test.go b/go/sdk/nats_integration_test.go new file mode 100644 index 0000000..a35f405 --- /dev/null +++ b/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) + }) + } +} diff --git a/go/sdk/options.go b/go/sdk/options.go index 5f89545..a557817 100644 --- a/go/sdk/options.go +++ b/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.