diff --git a/core/src/main/java/com/minekube/connect/register/WatcherRegister.java b/core/src/main/java/com/minekube/connect/register/WatcherRegister.java index 88029157d..4b8e7be28 100644 --- a/core/src/main/java/com/minekube/connect/register/WatcherRegister.java +++ b/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; } } diff --git a/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java b/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java index 463000b4e..597913cc2 100644 --- a/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java +++ b/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 = 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 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; }