fix(core): isolate stale proposals from WatchService

core/src/main/java/com/minekube/connect/register/WatcherRegister.java

@@ -235,6 +235,20 @@ private void reject(SessionProposal proposal, Status reason) {
         }
     }
 
+    private static Status rejectionStatus(Throwable failure) {
+        Status status = StatusProto.fromThrowable(failure);
+        if (status != null) {
+            return status;
+        }
+        String message = failure.getMessage();
+        return Status.newBuilder()
+                .setCode(Code.INTERNAL_VALUE)
+                .setMessage(message == null || message.isBlank()
+                        ? failure.getClass().getSimpleName()
+                        : message)
+                .build();
+    }
+
     private class WatcherImpl implements Watcher {
         private volatile boolean ignoreTerminalEvents;
 
@@ -299,8 +313,13 @@ public void onProposal(SessionProposal proposal) {
                         proposal,
                         admissionCoordinator
                 ).connect();
-            } catch (RuntimeException | Error e) {
-                reject(proposal, StatusProto.fromThrowable(e));
+            } catch (RuntimeException e) {
+                reject(proposal, rejectionStatus(e));
+                logger.warn("Rejected one Connect session proposal (category={}); " +
+                                "keeping WatchService active",
+                        e.getClass().getSimpleName());
+            } catch (Error e) {
+                reject(proposal, rejectionStatus(e));
                 throw e;
             }
         }

core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java

@@ -40,6 +40,7 @@
 import java.util.Timer;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import org.mockito.ArgumentCaptor;
 import minekube.connect.v1alpha1.WatchServiceOuterClass.GameProfile;
 import minekube.connect.v1alpha1.WatchServiceOuterClass.GameProfileProperty;
@@ -438,6 +439,43 @@ void invalidWatchProposalCancelsPrivateAdmissionImmediately() throws Exception {
         }
     }
 
+    @Test
+    void supersededProposalIsRejectedWithoutClosingTheWatchStream() throws Exception {
+        BedrockAdmissionCoordinator coordinator = new BedrockAdmissionCoordinator(
+                new VerifiedBedrockIdentityRegistry());
+        try {
+            Fixture fixture = newFixture(coordinator);
+            register = fixture.register;
+            register.start();
+            ArgumentCaptor<Watcher> watcher = ArgumentCaptor.forClass(Watcher.class);
+            verify(fixture.watchClient).watch(watcher.capture());
+            Session session = Session.newBuilder()
+                    .setId("duplicate-session")
+                    .setTunnelServiceAddr("wss://tunnel.example")
+                    .setPlayer(Player.newBuilder()
+                            .setAddr("127.0.0.1")
+                            .setProfile(GameProfile.newBuilder()
+                                    .setId("00000000-0000-0000-0000-000000000001")
+                                    .setName("Player")))
+                    .build();
+            AtomicReference<com.google.rpc.Status> rejection = new AtomicReference<>();
+            SessionProposal superseded = coordinator.proposal(
+                    session, rejection::set, "endpoint-1", "org-1");
+            SessionProposal current = coordinator.proposal(
+                    session, reason -> {}, "endpoint-1", "org-1");
+
+            assertDoesNotThrow(() -> watcher.getValue().onProposal(superseded));
+
+            assertNotNull(rejection.get());
+            assertTrue(rejection.get().getMessage()
+                    .contains("expired or been superseded"));
+            verify(fixture.watchClient, times(1)).watch(any(Watcher.class));
+            coordinator.discard(current);
+        } finally {
+            coordinator.close();
+        }
+    }
+
     private static WatcherRegister newRegister() throws Exception {
         return newFixture().register;
     }