From f964176021c4fd8f99422d20248275d77a651ebe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robin=20Br=C3=A4mer?= <22003767+robinbraemer@users.noreply.github.com> Date: Sat, 8 Aug 2026 00:18:41 +0200 Subject: [PATCH 1/2] fix(core): isolate stale proposals from WatchService --- .../connect/register/WatcherRegister.java | 7 +++- .../connect/register/WatcherRegisterTest.java | 38 +++++++++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) 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..ef6ccb6d2 100644 --- a/core/src/main/java/com/minekube/connect/register/WatcherRegister.java +++ b/core/src/main/java/com/minekube/connect/register/WatcherRegister.java @@ -299,7 +299,12 @@ public void onProposal(SessionProposal proposal) { proposal, admissionCoordinator ).connect(); - } catch (RuntimeException | Error e) { + } catch (RuntimeException e) { + reject(proposal, StatusProto.fromThrowable(e)); + logger.warn("Rejected one Connect session proposal (category={}); " + + "keeping WatchService active", + e.getClass().getSimpleName()); + } catch (Error e) { reject(proposal, StatusProto.fromThrowable(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; } From 2b125cd41f11680f2165ad23bac9d8fa2d10e937 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Robin=20Br=C3=A4mer?= <22003767+robinbraemer@users.noreply.github.com> Date: Sat, 8 Aug 2026 09:45:26 +0200 Subject: [PATCH 2/2] fix(core): preserve proposal rejection status --- .../connect/register/WatcherRegister.java | 18 ++++++++++++++++-- 1 file changed, 16 insertions(+), 2 deletions(-) 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 ef6ccb6d2..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; @@ -300,12 +314,12 @@ public void onProposal(SessionProposal proposal) { admissionCoordinator ).connect(); } catch (RuntimeException e) { - reject(proposal, StatusProto.fromThrowable(e)); + reject(proposal, rejectionStatus(e)); logger.warn("Rejected one Connect session proposal (category={}); " + "keeping WatchService active", e.getClass().getSimpleName()); } catch (Error e) { - reject(proposal, StatusProto.fromThrowable(e)); + reject(proposal, rejectionStatus(e)); throw e; } }