From 11d34df3fb171b4684b983c27b04bb60ff95e797 Mon Sep 17 00:00:00 2001 From: Robin Date: Sun, 14 Jun 2026 19:12:51 +0200 Subject: [PATCH 1/5] feat: add optional libp2p tunnel transport --- .../com/minekube/connect/BungeePlugin.java | 5 +- .../connect/tunnel/OptionalTunnelModules.java | 79 ++++++ .../tunnel/OptionalTunnelModulesTest.java | 18 ++ p2p/build.gradle.kts | 18 ++ .../connect/tunnel/p2p/Libp2pRuntime.java | 42 +++ .../tunnel/p2p/Libp2pTunnelModule.java | 40 +++ .../tunnel/p2p/Libp2pTunnelTransport.java | 261 ++++++++++++++++++ .../connect/tunnel/p2p/Libp2pRuntimeTest.java | 39 +++ .../tunnel/p2p/Libp2pTunnelTransportTest.java | 202 ++++++++++++++ settings.gradle.kts | 1 + .../com/minekube/connect/SpigotPlugin.java | 5 +- .../com/minekube/connect/VelocityPlugin.java | 5 +- 12 files changed, 709 insertions(+), 6 deletions(-) create mode 100644 core/src/main/java/com/minekube/connect/tunnel/OptionalTunnelModules.java create mode 100644 core/src/test/java/com/minekube/connect/tunnel/OptionalTunnelModulesTest.java create mode 100644 p2p/build.gradle.kts create mode 100644 p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pRuntime.java create mode 100644 p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java create mode 100644 p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransport.java create mode 100644 p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pRuntimeTest.java create mode 100644 p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransportTest.java diff --git a/bungee/src/main/java/com/minekube/connect/BungeePlugin.java b/bungee/src/main/java/com/minekube/connect/BungeePlugin.java index abab8018e..af585839f 100644 --- a/bungee/src/main/java/com/minekube/connect/BungeePlugin.java +++ b/bungee/src/main/java/com/minekube/connect/BungeePlugin.java @@ -34,6 +34,7 @@ import com.minekube.connect.module.Libp2pEndpointModule; import com.minekube.connect.module.ProxyCommonModule; import com.minekube.connect.module.WatcherModule; +import com.minekube.connect.tunnel.OptionalTunnelModules; import com.minekube.connect.util.ReflectionUtils; import net.md_5.bungee.api.plugin.Plugin; @@ -61,13 +62,13 @@ public void onLoad() { @Override public void onEnable() { - platform.enable( + platform.enable(OptionalTunnelModules.append( new CommandModule(), new BungeeListenerModule(), // new BungeeAddonModule(), - don't need proxy-side data injection new Libp2pEndpointModule(), new WatcherModule() - ); + )); } @Override diff --git a/core/src/main/java/com/minekube/connect/tunnel/OptionalTunnelModules.java b/core/src/main/java/com/minekube/connect/tunnel/OptionalTunnelModules.java new file mode 100644 index 000000000..7a304a03a --- /dev/null +++ b/core/src/main/java/com/minekube/connect/tunnel/OptionalTunnelModules.java @@ -0,0 +1,79 @@ +/* + * Copyright (c) 2021-2022 Minekube. https://minekube.com + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + * + * @author Minekube + * @link https://github.com/minekube/connect-java + */ + +package com.minekube.connect.tunnel; + +import com.google.inject.Module; +import java.lang.reflect.Constructor; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +public final class OptionalTunnelModules { + private static final String LIBP2P_MODULE = + "com.minekube.connect.tunnel.p2p.Libp2pTunnelModule"; + + private OptionalTunnelModules() { + } + + public static Module[] append(Module... modules) { + List all = new ArrayList<>(Arrays.asList(modules)); + Module libp2p = load(LIBP2P_MODULE); + if (libp2p != null) { + all.add(libp2p); + } + return all.toArray(new Module[0]); + } + + private static Module load(String className) { + if (javaFeatureVersion() < 11) { + return null; + } + try { + Class type = Class.forName(className); + if (!Module.class.isAssignableFrom(type)) { + return null; + } + Constructor constructor = type.getDeclaredConstructor(); + return (Module) constructor.newInstance(); + } catch (ClassNotFoundException e) { + return null; + } catch (ReflectiveOperationException | LinkageError e) { + throw new IllegalStateException("failed to load optional tunnel module " + className, e); + } + } + + private static int javaFeatureVersion() { + String spec = System.getProperty("java.specification.version", "8"); + if (spec.startsWith("1.")) { + return Integer.parseInt(spec.substring(2)); + } + int dot = spec.indexOf('.'); + if (dot >= 0) { + spec = spec.substring(0, dot); + } + return Integer.parseInt(spec); + } +} diff --git a/core/src/test/java/com/minekube/connect/tunnel/OptionalTunnelModulesTest.java b/core/src/test/java/com/minekube/connect/tunnel/OptionalTunnelModulesTest.java new file mode 100644 index 000000000..ccaf24fb6 --- /dev/null +++ b/core/src/test/java/com/minekube/connect/tunnel/OptionalTunnelModulesTest.java @@ -0,0 +1,18 @@ +package com.minekube.connect.tunnel; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; + +import com.google.inject.AbstractModule; +import com.google.inject.Module; +import org.junit.jupiter.api.Test; + +class OptionalTunnelModulesTest { + + @Test + void keepsModulesWhenLibp2pModuleIsNotOnClasspath() { + Module module = new AbstractModule() { + }; + + assertArrayEquals(new Module[] {module}, OptionalTunnelModules.append(module)); + } +} diff --git a/p2p/build.gradle.kts b/p2p/build.gradle.kts new file mode 100644 index 000000000..aaa182cc0 --- /dev/null +++ b/p2p/build.gradle.kts @@ -0,0 +1,18 @@ +java { + sourceCompatibility = JavaVersion.VERSION_11 + targetCompatibility = JavaVersion.VERSION_11 +} + +dependencies { + api(projects.core) + + implementation("io.libp2p:jvm-libp2p:${Versions.jvmLibp2pVersion}") + implementation("org.jetbrains.kotlin:kotlin-stdlib:${Versions.kotlinStdlibVersion}") + + testImplementation("org.junit.jupiter:junit-jupiter:5.10.5") + testRuntimeOnly("org.junit.platform:junit-platform-launcher") +} + +tasks.test { + useJUnitPlatform() +} diff --git a/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pRuntime.java b/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pRuntime.java new file mode 100644 index 000000000..b7dcbb2e4 --- /dev/null +++ b/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pRuntime.java @@ -0,0 +1,42 @@ +/* + * Copyright (c) 2021-2022 Minekube. https://minekube.com + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + * + * @author Minekube + * @link https://github.com/minekube/connect-java + */ + +package com.minekube.connect.tunnel.p2p; + +import io.libp2p.core.Host; + +public final class Libp2pRuntime { + + private Libp2pRuntime() { + } + + public static int minimumJavaFeatureVersion() { + return 11; + } + + public static String hostClassName() { + return Host.class.getName(); + } +} diff --git a/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java b/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java new file mode 100644 index 000000000..8f3c53ea5 --- /dev/null +++ b/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java @@ -0,0 +1,40 @@ +/* + * Copyright (c) 2021-2022 Minekube. https://minekube.com + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + * + * @author Minekube + * @link https://github.com/minekube/connect-java + */ + +package com.minekube.connect.tunnel.p2p; + +import com.google.inject.AbstractModule; +import com.google.inject.multibindings.Multibinder; +import com.minekube.connect.tunnel.TunnelClientTransport; + +public final class Libp2pTunnelModule extends AbstractModule { + + @Override + protected void configure() { + Multibinder.newSetBinder(binder(), TunnelClientTransport.class) + .addBinding() + .to(Libp2pTunnelTransport.class); + } +} diff --git a/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransport.java b/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransport.java new file mode 100644 index 000000000..7c1acde77 --- /dev/null +++ b/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransport.java @@ -0,0 +1,261 @@ +/* + * Copyright (c) 2021-2022 Minekube. https://minekube.com + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + * + * @author Minekube + * @link https://github.com/minekube/connect-java + */ + +package com.minekube.connect.tunnel.p2p; + +import com.google.inject.Inject; +import com.minekube.connect.tunnel.P2PTunnelHeader; +import com.minekube.connect.tunnel.TunnelClientTransport; +import com.minekube.connect.tunnel.TunnelConn; +import io.libp2p.core.Connection; +import io.libp2p.core.Host; +import io.libp2p.core.PeerId; +import io.libp2p.core.Stream; +import io.libp2p.core.StreamPromise; +import io.libp2p.core.crypto.KeyType; +import io.libp2p.core.dsl.HostBuilder; +import io.libp2p.core.mux.StreamMuxerProtocol; +import io.libp2p.core.multiformats.Multiaddr; +import io.libp2p.core.multistream.StrictProtocolBinding; +import io.libp2p.protocol.ProtocolHandler; +import io.libp2p.security.noise.NoiseXXSecureChannel; +import io.libp2p.transport.quic.QuicConfig; +import io.libp2p.transport.quic.QuicTransport; +import io.libp2p.transport.tcp.TcpTransport; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufUtil; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import java.util.Arrays; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.TimeUnit; +import minekube.connect.v1alpha1.WatchServiceOuterClass.TunnelTransport.Type; + +public final class Libp2pTunnelTransport implements TunnelClientTransport { + public static final String PROTOCOL_ID = "/minekube/connect/tunnel/1.0.0"; + + private static final long START_TIMEOUT_SECONDS = 10; + private static final long CONNECT_TIMEOUT_SECONDS = 15; + private static final long STREAM_TIMEOUT_SECONDS = 5; + + private final Host host; + private final ConcurrentMap warmConnections = new ConcurrentHashMap<>(); + + @Inject + public Libp2pTunnelTransport() { + this(createHost()); + } + + Libp2pTunnelTransport(Host host) { + this.host = Objects.requireNonNull(host, "host"); + this.host.addProtocolHandler(new TunnelProtocolBinding()); + await(host.start(), START_TIMEOUT_SECONDS, "start libp2p host"); + } + + static Host createHost() { + return new HostBuilder(HostBuilder.DefaultMode.None) + .keyType(KeyType.ED25519) + .transport(TcpTransport::new) + .secureChannel(NoiseXXSecureChannel::new) + .muxer(StreamMuxerProtocol::getYamux) + .secureTransport((priv, protocols) -> + QuicTransport.Ed25519(priv, protocols, new QuicConfig())) + .build(); + } + + @Override + public Type type() { + return Type.TYPE_LIBP2P; + } + + @Override + public void prepare(String address) { + if (address == null || address.isEmpty()) { + return; + } + + Multiaddr multiaddr = Multiaddr.fromString(address); + PeerId peerId = requirePeerId(multiaddr, address); + warmConnections.compute(peerId, (ignored, existing) -> { + if (isHealthy(existing)) { + return existing; + } + return connect(peerId, multiaddr); + }); + } + + boolean hasWarmConnection(String address) { + Multiaddr multiaddr = Multiaddr.fromString(address); + PeerId peerId = requirePeerId(multiaddr, address); + return isHealthy(warmConnections.get(peerId)); + } + + @Override + public TunnelConn tunnel(String address, String sessionId, TunnelConn.Handler handler) { + Objects.requireNonNull(handler, "handler"); + byte[] header = P2PTunnelHeader.encode(sessionId); + Multiaddr multiaddr = Multiaddr.fromString(address); + PeerId peerId = requirePeerId(multiaddr, address); + Connection connection = warmConnection(peerId, multiaddr); + Stream stream = openStream(connection); + try { + stream.pushHandler(new InboundBytesHandler(handler)); + stream.writeAndFlush(Unpooled.wrappedBuffer(header)); + return new Libp2pTunnelConn(stream); + } catch (RuntimeException e) { + stream.close(); + throw e; + } + } + + @Override + public void close() { + warmConnections.clear(); + await(host.stop(), START_TIMEOUT_SECONDS, "stop libp2p host"); + } + + private Connection warmConnection(PeerId peerId, Multiaddr multiaddr) { + prepare(multiaddr.toString()); + Connection connection = warmConnections.get(peerId); + if (!isHealthy(connection)) { + throw new IllegalStateException("libp2p connection is not warm for " + multiaddr); + } + return connection; + } + + private Connection connect(PeerId peerId, Multiaddr multiaddr) { + return await(host.getNetwork().connect(peerId, multiaddr), + CONNECT_TIMEOUT_SECONDS, "connect libp2p peer " + peerId); + } + + private Stream openStream(Connection connection) { + StreamPromise promise = host.newStream(Arrays.asList(PROTOCOL_ID), connection); + Stream stream = await(promise.getStream(), STREAM_TIMEOUT_SECONDS, "open libp2p tunnel stream"); + await(stream.getProtocol(), STREAM_TIMEOUT_SECONDS, "negotiate libp2p tunnel protocol"); + return stream; + } + + private static PeerId requirePeerId(Multiaddr multiaddr, String address) { + PeerId peerId = multiaddr.getPeerId(); + if (peerId == null) { + throw new IllegalArgumentException("p2p address must include /p2p/: " + address); + } + return peerId; + } + + private static boolean isHealthy(Connection connection) { + return connection != null && !connection.closeFuture().isDone(); + } + + private static T await(CompletableFuture future, long timeoutSeconds, String action) { + try { + return future.get(timeoutSeconds, TimeUnit.SECONDS); + } catch (InterruptedException e) { + future.cancel(true); + Thread.currentThread().interrupt(); + throw new IllegalStateException("failed to " + action, e); + } catch (TimeoutException e) { + future.cancel(true); + throw new IllegalStateException("failed to " + action, e); + } catch (Exception e) { + throw new IllegalStateException("failed to " + action, e); + } + } + + private static final class Libp2pTunnelConn extends TunnelConn { + private final Stream stream; + + private Libp2pTunnelConn(Stream stream) { + this.stream = stream; + } + + @Override + public void write(byte[] data) { + stream.writeAndFlush(Unpooled.wrappedBuffer(data)); + } + + @Override + public void close(Throwable t) { + stream.close(); + } + + @Override + public boolean opened() { + return true; + } + } + + private static final class InboundBytesHandler extends SimpleChannelInboundHandler { + private final TunnelConn.Handler handler; + + private InboundBytesHandler(TunnelConn.Handler handler) { + this.handler = handler; + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { + handler.onReceive(ByteBufUtil.getBytes(msg, msg.readerIndex(), msg.readableBytes(), false)); + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + handler.onError(cause); + ctx.close(); + } + + @Override + public void channelInactive(ChannelHandlerContext ctx) throws Exception { + handler.onClose(); + super.channelInactive(ctx); + } + } + + private static final class TunnelProtocolBinding extends StrictProtocolBinding { + private TunnelProtocolBinding() { + super(PROTOCOL_ID, new TunnelProtocolHandler()); + } + } + + private static final class TunnelProtocolHandler extends ProtocolHandler { + private TunnelProtocolHandler() { + super(Long.MAX_VALUE, Long.MAX_VALUE); + } + + @Override + protected CompletableFuture onStartInitiator(Stream stream) { + return CompletableFuture.completedFuture(null); + } + + @Override + protected CompletableFuture onStartResponder(Stream stream) { + return CompletableFuture.completedFuture(null); + } + } +} diff --git a/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pRuntimeTest.java b/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pRuntimeTest.java new file mode 100644 index 000000000..2dffbe767 --- /dev/null +++ b/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pRuntimeTest.java @@ -0,0 +1,39 @@ +/* + * Copyright (c) 2021-2022 Minekube. https://minekube.com + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + * + * @author Minekube + * @link https://github.com/minekube/connect-java + */ + +package com.minekube.connect.tunnel.p2p; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.junit.jupiter.api.Test; + +class Libp2pRuntimeTest { + + @Test + void resolvesJvmLibp2pHostClass() { + assertEquals(11, Libp2pRuntime.minimumJavaFeatureVersion()); + assertEquals("io.libp2p.core.Host", Libp2pRuntime.hostClassName()); + } +} diff --git a/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransportTest.java b/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransportTest.java new file mode 100644 index 000000000..6103824e7 --- /dev/null +++ b/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransportTest.java @@ -0,0 +1,202 @@ +/* + * Copyright (c) 2021-2022 Minekube. https://minekube.com + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN + * THE SOFTWARE. + * + * @author Minekube + * @link https://github.com/minekube/connect-java + */ + +package com.minekube.connect.tunnel.p2p; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.minekube.connect.tunnel.P2PTunnelHeader; +import com.minekube.connect.tunnel.TunnelConn; +import io.libp2p.core.Host; +import io.libp2p.core.Stream; +import io.libp2p.core.crypto.KeyType; +import io.libp2p.core.dsl.HostBuilder; +import io.libp2p.core.multiformats.Multiaddr; +import io.libp2p.core.multistream.StrictProtocolBinding; +import io.libp2p.protocol.ProtocolHandler; +import io.libp2p.security.noise.NoiseXXSecureChannel; +import io.libp2p.transport.tcp.TcpTransport; +import io.libp2p.core.mux.StreamMuxerProtocol; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufUtil; +import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import java.util.Arrays; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import minekube.connect.v1alpha1.WatchServiceOuterClass.TunnelTransport.Type; +import org.junit.jupiter.api.Test; + +class Libp2pTunnelTransportTest { + + @Test + void sendsMoxyCompatibleHeaderAndReceivesBytes() throws Exception { + Responder responder = startResponder(); + try { + Libp2pTunnelTransport transport = new Libp2pTunnelTransport(tcpOnlyHost()); + try { + RecordingHandler handler = new RecordingHandler(); + TunnelConn conn = transport.tunnel(responder.address(), "test-session", handler); + + assertEquals(Type.TYPE_LIBP2P, transport.type()); + assertTrue(conn.opened()); + assertArrayEquals(P2PTunnelHeader.encode("test-session"), + responder.awaitHeader()); + assertArrayEquals(new byte[] {1, 2, 3}, handler.awaitData()); + } finally { + transport.close(); + } + } finally { + responder.close(); + } + } + + @Test + void prepareWarmsConnectionBeforeOpeningSessionStream() throws Exception { + Responder responder = startResponder(); + try { + Libp2pTunnelTransport transport = new Libp2pTunnelTransport(tcpOnlyHost()); + try { + transport.prepare(responder.address()); + + assertTrue(transport.hasWarmConnection(responder.address())); + assertEquals(0, responder.streamCount()); + + transport.tunnel(responder.address(), "test-session", new RecordingHandler()); + + assertArrayEquals(P2PTunnelHeader.encode("test-session"), + responder.awaitHeader()); + assertEquals(1, responder.streamCount()); + } finally { + transport.close(); + } + } finally { + responder.close(); + } + } + + private static Responder startResponder() throws Exception { + Responder responder = new Responder(tcpOnlyHost()); + responder.start(); + return responder; + } + + private static Host tcpOnlyHost() { + return new HostBuilder(HostBuilder.DefaultMode.None) + .keyType(KeyType.ED25519) + .transport(TcpTransport::new) + .secureChannel(NoiseXXSecureChannel::new) + .muxer(StreamMuxerProtocol::getYamux) + .listen("/ip4/127.0.0.1/tcp/0") + .build(); + } + + private static final class Responder { + private final Host host; + private final CountDownLatch headerLatch = new CountDownLatch(1); + private final AtomicReference header = new AtomicReference<>(); + private final AtomicInteger streams = new AtomicInteger(); + + private Responder(Host host) { + this.host = host; + } + + private void start() throws Exception { + host.addProtocolHandler(new StrictProtocolBinding( + Libp2pTunnelTransport.PROTOCOL_ID, + new ProtocolHandler(Long.MAX_VALUE, Long.MAX_VALUE) { + @Override + protected CompletableFuture onStartInitiator(Stream stream) { + return CompletableFuture.completedFuture(null); + } + + @Override + protected CompletableFuture onStartResponder(Stream stream) { + streams.incrementAndGet(); + stream.pushHandler(new SimpleChannelInboundHandler() { + @Override + protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { + header.set(ByteBufUtil.getBytes( + msg, msg.readerIndex(), msg.readableBytes(), false)); + headerLatch.countDown(); + stream.writeAndFlush(Unpooled.wrappedBuffer(new byte[] {1, 2, 3})); + } + }); + return CompletableFuture.completedFuture(null); + } + }) { + }); + host.start().get(10, TimeUnit.SECONDS); + } + + private String address() { + return host.listenAddresses().stream() + .filter(addr -> addr.toString().contains("/tcp/")) + .findFirst() + .orElseThrow(() -> new AssertionError("no tcp listen address")) + .withP2P(host.getPeerId()) + .toString(); + } + + private byte[] awaitHeader() throws Exception { + assertTrue(headerLatch.await(5, TimeUnit.SECONDS), "timed out waiting for header"); + return header.get(); + } + + private int streamCount() { + return streams.get(); + } + + private void close() throws Exception { + host.stop().get(10, TimeUnit.SECONDS); + } + } + + private static final class RecordingHandler implements TunnelConn.Handler { + private final CountDownLatch dataLatch = new CountDownLatch(1); + private final AtomicReference data = new AtomicReference<>(); + + @Override + public void onReceive(byte[] data) { + this.data.set(Arrays.copyOf(data, data.length)); + dataLatch.countDown(); + } + + @Override + public void onError(Throwable t) { + } + + private byte[] awaitData() throws Exception { + assertTrue(dataLatch.await(5, TimeUnit.SECONDS), "timed out waiting for tunnel data"); + return data.get(); + } + } +} diff --git a/settings.gradle.kts b/settings.gradle.kts index 1f38f5684..4bd234966 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -70,6 +70,7 @@ rootProject.name = "connect-parent" include(":api") include(":core") +include(":p2p") include(":bungee") include(":spigot") include(":velocity") diff --git a/spigot/src/main/java/com/minekube/connect/SpigotPlugin.java b/spigot/src/main/java/com/minekube/connect/SpigotPlugin.java index 36bccc5e7..f41007e07 100644 --- a/spigot/src/main/java/com/minekube/connect/SpigotPlugin.java +++ b/spigot/src/main/java/com/minekube/connect/SpigotPlugin.java @@ -36,6 +36,7 @@ import com.minekube.connect.module.SpigotListenerModule; import com.minekube.connect.module.SpigotPlatformModule; import com.minekube.connect.module.WatcherModule; +import com.minekube.connect.tunnel.OptionalTunnelModules; import com.minekube.connect.util.ReflectionUtils; import com.minekube.connect.util.SpigotProtocolSupportHandler; import com.minekube.connect.util.SpigotProtocolSupportListener; @@ -65,13 +66,13 @@ public void onEnable() { boolean usePaperListener = ReflectionUtils.getClassSilently( "com.destroystokyo.paper.event.profile.PreFillProfileEvent") != null; - platform.enable( + platform.enable(OptionalTunnelModules.append( new SpigotCommandModule(this), new SpigotAddonModule(), (usePaperListener ? new PaperListenerModule() : new SpigotListenerModule()), new Libp2pEndpointModule(), new WatcherModule() - ); + )); //todo add proper support for disabling things on shutdown and enabling this on enable diff --git a/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java b/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java index 4174740ba..b231569f7 100644 --- a/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java +++ b/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java @@ -34,6 +34,7 @@ import com.minekube.connect.module.VelocityListenerModule; import com.minekube.connect.module.VelocityPlatformModule; import com.minekube.connect.module.WatcherModule; +import com.minekube.connect.tunnel.OptionalTunnelModules; import com.minekube.connect.util.ReflectionUtils; import com.velocitypowered.api.event.Subscribe; import com.velocitypowered.api.event.proxy.ProxyInitializeEvent; @@ -62,12 +63,12 @@ public VelocityPlugin(@DataDirectory Path dataDirectory, Injector guice) { @Subscribe public void onProxyInitialization(ProxyInitializeEvent event) { - platform.enable( + platform.enable(OptionalTunnelModules.append( new CommandModule(), new VelocityListenerModule(), // new VelocityAddonModule(), - don't need proxy-side data injection new Libp2pEndpointModule(), new WatcherModule() - ); + )); } } From a767ec92ee9cf03c8768b4d935b97e03557a66a5 Mon Sep 17 00:00:00 2001 From: Robin Date: Mon, 15 Jun 2026 01:30:06 +0200 Subject: [PATCH 2/5] build: ship libp2p tunnel on java 11 --- .../connect/tunnel/OptionalTunnelModules.java | 79 ------ .../tunnel/p2p/Libp2pTunnelModule.java | 0 .../tunnel/OptionalTunnelModulesTest.java | 18 -- p2p/build.gradle.kts | 18 -- .../connect/tunnel/p2p/Libp2pRuntime.java | 42 --- .../tunnel/p2p/Libp2pTunnelTransport.java | 261 ------------------ .../connect/tunnel/p2p/Libp2pRuntimeTest.java | 39 --- .../tunnel/p2p/Libp2pTunnelTransportTest.java | 202 -------------- settings.gradle.kts | 1 - 9 files changed, 660 deletions(-) delete mode 100644 core/src/main/java/com/minekube/connect/tunnel/OptionalTunnelModules.java rename {p2p => core}/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java (100%) delete mode 100644 core/src/test/java/com/minekube/connect/tunnel/OptionalTunnelModulesTest.java delete mode 100644 p2p/build.gradle.kts delete mode 100644 p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pRuntime.java delete mode 100644 p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransport.java delete mode 100644 p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pRuntimeTest.java delete mode 100644 p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransportTest.java diff --git a/core/src/main/java/com/minekube/connect/tunnel/OptionalTunnelModules.java b/core/src/main/java/com/minekube/connect/tunnel/OptionalTunnelModules.java deleted file mode 100644 index 7a304a03a..000000000 --- a/core/src/main/java/com/minekube/connect/tunnel/OptionalTunnelModules.java +++ /dev/null @@ -1,79 +0,0 @@ -/* - * Copyright (c) 2021-2022 Minekube. https://minekube.com - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN - * THE SOFTWARE. - * - * @author Minekube - * @link https://github.com/minekube/connect-java - */ - -package com.minekube.connect.tunnel; - -import com.google.inject.Module; -import java.lang.reflect.Constructor; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.List; - -public final class OptionalTunnelModules { - private static final String LIBP2P_MODULE = - "com.minekube.connect.tunnel.p2p.Libp2pTunnelModule"; - - private OptionalTunnelModules() { - } - - public static Module[] append(Module... modules) { - List all = new ArrayList<>(Arrays.asList(modules)); - Module libp2p = load(LIBP2P_MODULE); - if (libp2p != null) { - all.add(libp2p); - } - return all.toArray(new Module[0]); - } - - private static Module load(String className) { - if (javaFeatureVersion() < 11) { - return null; - } - try { - Class type = Class.forName(className); - if (!Module.class.isAssignableFrom(type)) { - return null; - } - Constructor constructor = type.getDeclaredConstructor(); - return (Module) constructor.newInstance(); - } catch (ClassNotFoundException e) { - return null; - } catch (ReflectiveOperationException | LinkageError e) { - throw new IllegalStateException("failed to load optional tunnel module " + className, e); - } - } - - private static int javaFeatureVersion() { - String spec = System.getProperty("java.specification.version", "8"); - if (spec.startsWith("1.")) { - return Integer.parseInt(spec.substring(2)); - } - int dot = spec.indexOf('.'); - if (dot >= 0) { - spec = spec.substring(0, dot); - } - return Integer.parseInt(spec); - } -} diff --git a/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java similarity index 100% rename from p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java rename to core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java diff --git a/core/src/test/java/com/minekube/connect/tunnel/OptionalTunnelModulesTest.java b/core/src/test/java/com/minekube/connect/tunnel/OptionalTunnelModulesTest.java deleted file mode 100644 index ccaf24fb6..000000000 --- a/core/src/test/java/com/minekube/connect/tunnel/OptionalTunnelModulesTest.java +++ /dev/null @@ -1,18 +0,0 @@ -package com.minekube.connect.tunnel; - -import static org.junit.jupiter.api.Assertions.assertArrayEquals; - -import com.google.inject.AbstractModule; -import com.google.inject.Module; -import org.junit.jupiter.api.Test; - -class OptionalTunnelModulesTest { - - @Test - void keepsModulesWhenLibp2pModuleIsNotOnClasspath() { - Module module = new AbstractModule() { - }; - - assertArrayEquals(new Module[] {module}, OptionalTunnelModules.append(module)); - } -} diff --git a/p2p/build.gradle.kts b/p2p/build.gradle.kts deleted file mode 100644 index aaa182cc0..000000000 --- a/p2p/build.gradle.kts +++ /dev/null @@ -1,18 +0,0 @@ -java { - sourceCompatibility = JavaVersion.VERSION_11 - targetCompatibility = JavaVersion.VERSION_11 -} - -dependencies { - api(projects.core) - - implementation("io.libp2p:jvm-libp2p:${Versions.jvmLibp2pVersion}") - implementation("org.jetbrains.kotlin:kotlin-stdlib:${Versions.kotlinStdlibVersion}") - - testImplementation("org.junit.jupiter:junit-jupiter:5.10.5") - testRuntimeOnly("org.junit.platform:junit-platform-launcher") -} - -tasks.test { - useJUnitPlatform() -} diff --git a/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pRuntime.java b/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pRuntime.java deleted file mode 100644 index b7dcbb2e4..000000000 --- a/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pRuntime.java +++ /dev/null @@ -1,42 +0,0 @@ -/* - * Copyright (c) 2021-2022 Minekube. https://minekube.com - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN - * THE SOFTWARE. - * - * @author Minekube - * @link https://github.com/minekube/connect-java - */ - -package com.minekube.connect.tunnel.p2p; - -import io.libp2p.core.Host; - -public final class Libp2pRuntime { - - private Libp2pRuntime() { - } - - public static int minimumJavaFeatureVersion() { - return 11; - } - - public static String hostClassName() { - return Host.class.getName(); - } -} diff --git a/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransport.java b/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransport.java deleted file mode 100644 index 7c1acde77..000000000 --- a/p2p/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransport.java +++ /dev/null @@ -1,261 +0,0 @@ -/* - * Copyright (c) 2021-2022 Minekube. https://minekube.com - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN - * THE SOFTWARE. - * - * @author Minekube - * @link https://github.com/minekube/connect-java - */ - -package com.minekube.connect.tunnel.p2p; - -import com.google.inject.Inject; -import com.minekube.connect.tunnel.P2PTunnelHeader; -import com.minekube.connect.tunnel.TunnelClientTransport; -import com.minekube.connect.tunnel.TunnelConn; -import io.libp2p.core.Connection; -import io.libp2p.core.Host; -import io.libp2p.core.PeerId; -import io.libp2p.core.Stream; -import io.libp2p.core.StreamPromise; -import io.libp2p.core.crypto.KeyType; -import io.libp2p.core.dsl.HostBuilder; -import io.libp2p.core.mux.StreamMuxerProtocol; -import io.libp2p.core.multiformats.Multiaddr; -import io.libp2p.core.multistream.StrictProtocolBinding; -import io.libp2p.protocol.ProtocolHandler; -import io.libp2p.security.noise.NoiseXXSecureChannel; -import io.libp2p.transport.quic.QuicConfig; -import io.libp2p.transport.quic.QuicTransport; -import io.libp2p.transport.tcp.TcpTransport; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.ByteBufUtil; -import io.netty.buffer.Unpooled; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.SimpleChannelInboundHandler; -import java.util.Arrays; -import java.util.Objects; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.TimeoutException; -import java.util.concurrent.TimeUnit; -import minekube.connect.v1alpha1.WatchServiceOuterClass.TunnelTransport.Type; - -public final class Libp2pTunnelTransport implements TunnelClientTransport { - public static final String PROTOCOL_ID = "/minekube/connect/tunnel/1.0.0"; - - private static final long START_TIMEOUT_SECONDS = 10; - private static final long CONNECT_TIMEOUT_SECONDS = 15; - private static final long STREAM_TIMEOUT_SECONDS = 5; - - private final Host host; - private final ConcurrentMap warmConnections = new ConcurrentHashMap<>(); - - @Inject - public Libp2pTunnelTransport() { - this(createHost()); - } - - Libp2pTunnelTransport(Host host) { - this.host = Objects.requireNonNull(host, "host"); - this.host.addProtocolHandler(new TunnelProtocolBinding()); - await(host.start(), START_TIMEOUT_SECONDS, "start libp2p host"); - } - - static Host createHost() { - return new HostBuilder(HostBuilder.DefaultMode.None) - .keyType(KeyType.ED25519) - .transport(TcpTransport::new) - .secureChannel(NoiseXXSecureChannel::new) - .muxer(StreamMuxerProtocol::getYamux) - .secureTransport((priv, protocols) -> - QuicTransport.Ed25519(priv, protocols, new QuicConfig())) - .build(); - } - - @Override - public Type type() { - return Type.TYPE_LIBP2P; - } - - @Override - public void prepare(String address) { - if (address == null || address.isEmpty()) { - return; - } - - Multiaddr multiaddr = Multiaddr.fromString(address); - PeerId peerId = requirePeerId(multiaddr, address); - warmConnections.compute(peerId, (ignored, existing) -> { - if (isHealthy(existing)) { - return existing; - } - return connect(peerId, multiaddr); - }); - } - - boolean hasWarmConnection(String address) { - Multiaddr multiaddr = Multiaddr.fromString(address); - PeerId peerId = requirePeerId(multiaddr, address); - return isHealthy(warmConnections.get(peerId)); - } - - @Override - public TunnelConn tunnel(String address, String sessionId, TunnelConn.Handler handler) { - Objects.requireNonNull(handler, "handler"); - byte[] header = P2PTunnelHeader.encode(sessionId); - Multiaddr multiaddr = Multiaddr.fromString(address); - PeerId peerId = requirePeerId(multiaddr, address); - Connection connection = warmConnection(peerId, multiaddr); - Stream stream = openStream(connection); - try { - stream.pushHandler(new InboundBytesHandler(handler)); - stream.writeAndFlush(Unpooled.wrappedBuffer(header)); - return new Libp2pTunnelConn(stream); - } catch (RuntimeException e) { - stream.close(); - throw e; - } - } - - @Override - public void close() { - warmConnections.clear(); - await(host.stop(), START_TIMEOUT_SECONDS, "stop libp2p host"); - } - - private Connection warmConnection(PeerId peerId, Multiaddr multiaddr) { - prepare(multiaddr.toString()); - Connection connection = warmConnections.get(peerId); - if (!isHealthy(connection)) { - throw new IllegalStateException("libp2p connection is not warm for " + multiaddr); - } - return connection; - } - - private Connection connect(PeerId peerId, Multiaddr multiaddr) { - return await(host.getNetwork().connect(peerId, multiaddr), - CONNECT_TIMEOUT_SECONDS, "connect libp2p peer " + peerId); - } - - private Stream openStream(Connection connection) { - StreamPromise promise = host.newStream(Arrays.asList(PROTOCOL_ID), connection); - Stream stream = await(promise.getStream(), STREAM_TIMEOUT_SECONDS, "open libp2p tunnel stream"); - await(stream.getProtocol(), STREAM_TIMEOUT_SECONDS, "negotiate libp2p tunnel protocol"); - return stream; - } - - private static PeerId requirePeerId(Multiaddr multiaddr, String address) { - PeerId peerId = multiaddr.getPeerId(); - if (peerId == null) { - throw new IllegalArgumentException("p2p address must include /p2p/: " + address); - } - return peerId; - } - - private static boolean isHealthy(Connection connection) { - return connection != null && !connection.closeFuture().isDone(); - } - - private static T await(CompletableFuture future, long timeoutSeconds, String action) { - try { - return future.get(timeoutSeconds, TimeUnit.SECONDS); - } catch (InterruptedException e) { - future.cancel(true); - Thread.currentThread().interrupt(); - throw new IllegalStateException("failed to " + action, e); - } catch (TimeoutException e) { - future.cancel(true); - throw new IllegalStateException("failed to " + action, e); - } catch (Exception e) { - throw new IllegalStateException("failed to " + action, e); - } - } - - private static final class Libp2pTunnelConn extends TunnelConn { - private final Stream stream; - - private Libp2pTunnelConn(Stream stream) { - this.stream = stream; - } - - @Override - public void write(byte[] data) { - stream.writeAndFlush(Unpooled.wrappedBuffer(data)); - } - - @Override - public void close(Throwable t) { - stream.close(); - } - - @Override - public boolean opened() { - return true; - } - } - - private static final class InboundBytesHandler extends SimpleChannelInboundHandler { - private final TunnelConn.Handler handler; - - private InboundBytesHandler(TunnelConn.Handler handler) { - this.handler = handler; - } - - @Override - protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { - handler.onReceive(ByteBufUtil.getBytes(msg, msg.readerIndex(), msg.readableBytes(), false)); - } - - @Override - public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { - handler.onError(cause); - ctx.close(); - } - - @Override - public void channelInactive(ChannelHandlerContext ctx) throws Exception { - handler.onClose(); - super.channelInactive(ctx); - } - } - - private static final class TunnelProtocolBinding extends StrictProtocolBinding { - private TunnelProtocolBinding() { - super(PROTOCOL_ID, new TunnelProtocolHandler()); - } - } - - private static final class TunnelProtocolHandler extends ProtocolHandler { - private TunnelProtocolHandler() { - super(Long.MAX_VALUE, Long.MAX_VALUE); - } - - @Override - protected CompletableFuture onStartInitiator(Stream stream) { - return CompletableFuture.completedFuture(null); - } - - @Override - protected CompletableFuture onStartResponder(Stream stream) { - return CompletableFuture.completedFuture(null); - } - } -} diff --git a/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pRuntimeTest.java b/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pRuntimeTest.java deleted file mode 100644 index 2dffbe767..000000000 --- a/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pRuntimeTest.java +++ /dev/null @@ -1,39 +0,0 @@ -/* - * Copyright (c) 2021-2022 Minekube. https://minekube.com - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN - * THE SOFTWARE. - * - * @author Minekube - * @link https://github.com/minekube/connect-java - */ - -package com.minekube.connect.tunnel.p2p; - -import static org.junit.jupiter.api.Assertions.assertEquals; - -import org.junit.jupiter.api.Test; - -class Libp2pRuntimeTest { - - @Test - void resolvesJvmLibp2pHostClass() { - assertEquals(11, Libp2pRuntime.minimumJavaFeatureVersion()); - assertEquals("io.libp2p.core.Host", Libp2pRuntime.hostClassName()); - } -} diff --git a/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransportTest.java b/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransportTest.java deleted file mode 100644 index 6103824e7..000000000 --- a/p2p/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelTransportTest.java +++ /dev/null @@ -1,202 +0,0 @@ -/* - * Copyright (c) 2021-2022 Minekube. https://minekube.com - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN - * THE SOFTWARE. - * - * @author Minekube - * @link https://github.com/minekube/connect-java - */ - -package com.minekube.connect.tunnel.p2p; - -import static org.junit.jupiter.api.Assertions.assertArrayEquals; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertTrue; - -import com.minekube.connect.tunnel.P2PTunnelHeader; -import com.minekube.connect.tunnel.TunnelConn; -import io.libp2p.core.Host; -import io.libp2p.core.Stream; -import io.libp2p.core.crypto.KeyType; -import io.libp2p.core.dsl.HostBuilder; -import io.libp2p.core.multiformats.Multiaddr; -import io.libp2p.core.multistream.StrictProtocolBinding; -import io.libp2p.protocol.ProtocolHandler; -import io.libp2p.security.noise.NoiseXXSecureChannel; -import io.libp2p.transport.tcp.TcpTransport; -import io.libp2p.core.mux.StreamMuxerProtocol; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.ByteBufUtil; -import io.netty.buffer.Unpooled; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.SimpleChannelInboundHandler; -import java.util.Arrays; -import java.util.concurrent.CompletableFuture; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicReference; -import minekube.connect.v1alpha1.WatchServiceOuterClass.TunnelTransport.Type; -import org.junit.jupiter.api.Test; - -class Libp2pTunnelTransportTest { - - @Test - void sendsMoxyCompatibleHeaderAndReceivesBytes() throws Exception { - Responder responder = startResponder(); - try { - Libp2pTunnelTransport transport = new Libp2pTunnelTransport(tcpOnlyHost()); - try { - RecordingHandler handler = new RecordingHandler(); - TunnelConn conn = transport.tunnel(responder.address(), "test-session", handler); - - assertEquals(Type.TYPE_LIBP2P, transport.type()); - assertTrue(conn.opened()); - assertArrayEquals(P2PTunnelHeader.encode("test-session"), - responder.awaitHeader()); - assertArrayEquals(new byte[] {1, 2, 3}, handler.awaitData()); - } finally { - transport.close(); - } - } finally { - responder.close(); - } - } - - @Test - void prepareWarmsConnectionBeforeOpeningSessionStream() throws Exception { - Responder responder = startResponder(); - try { - Libp2pTunnelTransport transport = new Libp2pTunnelTransport(tcpOnlyHost()); - try { - transport.prepare(responder.address()); - - assertTrue(transport.hasWarmConnection(responder.address())); - assertEquals(0, responder.streamCount()); - - transport.tunnel(responder.address(), "test-session", new RecordingHandler()); - - assertArrayEquals(P2PTunnelHeader.encode("test-session"), - responder.awaitHeader()); - assertEquals(1, responder.streamCount()); - } finally { - transport.close(); - } - } finally { - responder.close(); - } - } - - private static Responder startResponder() throws Exception { - Responder responder = new Responder(tcpOnlyHost()); - responder.start(); - return responder; - } - - private static Host tcpOnlyHost() { - return new HostBuilder(HostBuilder.DefaultMode.None) - .keyType(KeyType.ED25519) - .transport(TcpTransport::new) - .secureChannel(NoiseXXSecureChannel::new) - .muxer(StreamMuxerProtocol::getYamux) - .listen("/ip4/127.0.0.1/tcp/0") - .build(); - } - - private static final class Responder { - private final Host host; - private final CountDownLatch headerLatch = new CountDownLatch(1); - private final AtomicReference header = new AtomicReference<>(); - private final AtomicInteger streams = new AtomicInteger(); - - private Responder(Host host) { - this.host = host; - } - - private void start() throws Exception { - host.addProtocolHandler(new StrictProtocolBinding( - Libp2pTunnelTransport.PROTOCOL_ID, - new ProtocolHandler(Long.MAX_VALUE, Long.MAX_VALUE) { - @Override - protected CompletableFuture onStartInitiator(Stream stream) { - return CompletableFuture.completedFuture(null); - } - - @Override - protected CompletableFuture onStartResponder(Stream stream) { - streams.incrementAndGet(); - stream.pushHandler(new SimpleChannelInboundHandler() { - @Override - protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { - header.set(ByteBufUtil.getBytes( - msg, msg.readerIndex(), msg.readableBytes(), false)); - headerLatch.countDown(); - stream.writeAndFlush(Unpooled.wrappedBuffer(new byte[] {1, 2, 3})); - } - }); - return CompletableFuture.completedFuture(null); - } - }) { - }); - host.start().get(10, TimeUnit.SECONDS); - } - - private String address() { - return host.listenAddresses().stream() - .filter(addr -> addr.toString().contains("/tcp/")) - .findFirst() - .orElseThrow(() -> new AssertionError("no tcp listen address")) - .withP2P(host.getPeerId()) - .toString(); - } - - private byte[] awaitHeader() throws Exception { - assertTrue(headerLatch.await(5, TimeUnit.SECONDS), "timed out waiting for header"); - return header.get(); - } - - private int streamCount() { - return streams.get(); - } - - private void close() throws Exception { - host.stop().get(10, TimeUnit.SECONDS); - } - } - - private static final class RecordingHandler implements TunnelConn.Handler { - private final CountDownLatch dataLatch = new CountDownLatch(1); - private final AtomicReference data = new AtomicReference<>(); - - @Override - public void onReceive(byte[] data) { - this.data.set(Arrays.copyOf(data, data.length)); - dataLatch.countDown(); - } - - @Override - public void onError(Throwable t) { - } - - private byte[] awaitData() throws Exception { - assertTrue(dataLatch.await(5, TimeUnit.SECONDS), "timed out waiting for tunnel data"); - return data.get(); - } - } -} diff --git a/settings.gradle.kts b/settings.gradle.kts index 4bd234966..1f38f5684 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -70,7 +70,6 @@ rootProject.name = "connect-parent" include(":api") include(":core") -include(":p2p") include(":bungee") include(":spigot") include(":velocity") From a068baaf30c1befd40366d1370f7783a8fbc1fc0 Mon Sep 17 00:00:00 2001 From: Robin Date: Mon, 15 Jun 2026 07:56:41 +0200 Subject: [PATCH 3/5] fix: bind libp2p tunnel in common module --- .../com/minekube/connect/BungeePlugin.java | 5 +-- .../tunnel/p2p/Libp2pTunnelModule.java | 40 ------------------- .../com/minekube/connect/SpigotPlugin.java | 5 +-- .../com/minekube/connect/VelocityPlugin.java | 5 +-- 4 files changed, 6 insertions(+), 49 deletions(-) delete mode 100644 core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java diff --git a/bungee/src/main/java/com/minekube/connect/BungeePlugin.java b/bungee/src/main/java/com/minekube/connect/BungeePlugin.java index af585839f..abab8018e 100644 --- a/bungee/src/main/java/com/minekube/connect/BungeePlugin.java +++ b/bungee/src/main/java/com/minekube/connect/BungeePlugin.java @@ -34,7 +34,6 @@ import com.minekube.connect.module.Libp2pEndpointModule; import com.minekube.connect.module.ProxyCommonModule; import com.minekube.connect.module.WatcherModule; -import com.minekube.connect.tunnel.OptionalTunnelModules; import com.minekube.connect.util.ReflectionUtils; import net.md_5.bungee.api.plugin.Plugin; @@ -62,13 +61,13 @@ public void onLoad() { @Override public void onEnable() { - platform.enable(OptionalTunnelModules.append( + platform.enable( new CommandModule(), new BungeeListenerModule(), // new BungeeAddonModule(), - don't need proxy-side data injection new Libp2pEndpointModule(), new WatcherModule() - )); + ); } @Override diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java deleted file mode 100644 index 8f3c53ea5..000000000 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pTunnelModule.java +++ /dev/null @@ -1,40 +0,0 @@ -/* - * Copyright (c) 2021-2022 Minekube. https://minekube.com - * - * Permission is hereby granted, free of charge, to any person obtaining a copy - * of this software and associated documentation files (the "Software"), to deal - * in the Software without restriction, including without limitation the rights - * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell - * copies of the Software, and to permit persons to whom the Software is - * furnished to do so, subject to the following conditions: - * - * The above copyright notice and this permission notice shall be included in - * all copies or substantial portions of the Software. - * - * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR - * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, - * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE - * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN - * THE SOFTWARE. - * - * @author Minekube - * @link https://github.com/minekube/connect-java - */ - -package com.minekube.connect.tunnel.p2p; - -import com.google.inject.AbstractModule; -import com.google.inject.multibindings.Multibinder; -import com.minekube.connect.tunnel.TunnelClientTransport; - -public final class Libp2pTunnelModule extends AbstractModule { - - @Override - protected void configure() { - Multibinder.newSetBinder(binder(), TunnelClientTransport.class) - .addBinding() - .to(Libp2pTunnelTransport.class); - } -} diff --git a/spigot/src/main/java/com/minekube/connect/SpigotPlugin.java b/spigot/src/main/java/com/minekube/connect/SpigotPlugin.java index f41007e07..36bccc5e7 100644 --- a/spigot/src/main/java/com/minekube/connect/SpigotPlugin.java +++ b/spigot/src/main/java/com/minekube/connect/SpigotPlugin.java @@ -36,7 +36,6 @@ import com.minekube.connect.module.SpigotListenerModule; import com.minekube.connect.module.SpigotPlatformModule; import com.minekube.connect.module.WatcherModule; -import com.minekube.connect.tunnel.OptionalTunnelModules; import com.minekube.connect.util.ReflectionUtils; import com.minekube.connect.util.SpigotProtocolSupportHandler; import com.minekube.connect.util.SpigotProtocolSupportListener; @@ -66,13 +65,13 @@ public void onEnable() { boolean usePaperListener = ReflectionUtils.getClassSilently( "com.destroystokyo.paper.event.profile.PreFillProfileEvent") != null; - platform.enable(OptionalTunnelModules.append( + platform.enable( new SpigotCommandModule(this), new SpigotAddonModule(), (usePaperListener ? new PaperListenerModule() : new SpigotListenerModule()), new Libp2pEndpointModule(), new WatcherModule() - )); + ); //todo add proper support for disabling things on shutdown and enabling this on enable diff --git a/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java b/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java index b231569f7..4174740ba 100644 --- a/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java +++ b/velocity/src/main/java/com/minekube/connect/VelocityPlugin.java @@ -34,7 +34,6 @@ import com.minekube.connect.module.VelocityListenerModule; import com.minekube.connect.module.VelocityPlatformModule; import com.minekube.connect.module.WatcherModule; -import com.minekube.connect.tunnel.OptionalTunnelModules; import com.minekube.connect.util.ReflectionUtils; import com.velocitypowered.api.event.Subscribe; import com.velocitypowered.api.event.proxy.ProxyInitializeEvent; @@ -63,12 +62,12 @@ public VelocityPlugin(@DataDirectory Path dataDirectory, Injector guice) { @Subscribe public void onProxyInitialization(ProxyInitializeEvent event) { - platform.enable(OptionalTunnelModules.append( + platform.enable( new CommandModule(), new VelocityListenerModule(), // new VelocityAddonModule(), - don't need proxy-side data injection new Libp2pEndpointModule(), new WatcherModule() - )); + ); } } From 5a71ce86efe0838918c0bdaa783aa3e74536ede2 Mon Sep 17 00:00:00 2001 From: Robin Date: Wed, 1 Jul 2026 01:48:50 +0200 Subject: [PATCH 4/5] bootstrap connect libp2p from watch capabilities --- .../connect/register/WatcherRegister.java | 6 ++- .../connect/tunnel/p2p/Libp2pEndpoint.java | 13 +++---- .../tunnel/p2p/Libp2pEndpointConfig.java | 24 ++++++++++-- .../tunnel/p2p/Libp2pEndpointRuntime.java | 6 +-- .../connect/watch/WatchBootstrap.java | 37 ++++++++++++++++--- .../connect/register/WatcherRegisterTest.java | 31 +++++++++++++++- .../tunnel/p2p/Libp2pEndpointConfigTest.java | 25 +++++++++++++ .../connect/watch/WatchBootstrapTest.java | 14 +++++++ 8 files changed, 135 insertions(+), 21 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 6bca70100..35372b442 100644 --- a/core/src/main/java/com/minekube/connect/register/WatcherRegister.java +++ b/core/src/main/java/com/minekube/connect/register/WatcherRegister.java @@ -36,6 +36,7 @@ import com.minekube.connect.util.Utils; import com.minekube.connect.util.backoff.BackOff; import com.minekube.connect.util.backoff.ExponentialBackOff; +import com.minekube.connect.tunnel.p2p.Libp2pEndpoint; import com.minekube.connect.watch.SessionProposal; import com.minekube.connect.watch.SessionProposal.State; import com.minekube.connect.watch.WatchBootstrap; @@ -160,7 +161,10 @@ private class WatcherImpl implements Watcher { public void onOpen(WatchBootstrap bootstrap) { logger.translatedInfo("connect.watch.started"); if (bootstrap.hasLibp2p()) { - libp2pEndpoint.start(bootstrap.libp2pEdgeAddrs(), bootstrap.libp2pRelayAddrs()); + libp2pEndpoint.start( + bootstrap.libp2pEdgeAddrs(), + bootstrap.libp2pRelayAddrs(), + bootstrap.supportsWatchlessLibp2p()); } startResetBackOffTimer(); } diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java index 275830eee..3e752ec8d 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java @@ -15,9 +15,7 @@ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER - * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, - * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN - * THE SOFTWARE. + * DEALINGS IN THE SOFTWARE. * * @author Minekube * @link https://github.com/minekube/connect-java @@ -38,7 +36,7 @@ import java.nio.file.Path; import java.util.List; -public final class Libp2pEndpoint { +public class Libp2pEndpoint { static final String REGISTER_PROTOCOL_ID = "/minekube/connect/register/1.0.0"; private final ConnectLogger logger; @@ -80,7 +78,8 @@ public Libp2pEndpoint( platformInjector, api); this.startMethod = runtimeClass.getDeclaredMethod("start"); - this.startBootstrapMethod = runtimeClass.getDeclaredMethod("start", List.class, List.class); + this.startBootstrapMethod = runtimeClass.getDeclaredMethod( + "start", List.class, List.class, boolean.class); this.stopMethod = runtimeClass.getDeclaredMethod("stop"); this.startMethod.setAccessible(true); this.startBootstrapMethod.setAccessible(true); @@ -108,12 +107,12 @@ public synchronized void start() { } } - public synchronized void start(List registerAddrs, List relayAddrs) { + public synchronized void start(List registerAddrs, List relayAddrs, boolean watchless) { if (runtime == null) { return; } try { - startBootstrapMethod.invoke(runtime, registerAddrs, relayAddrs); + startBootstrapMethod.invoke(runtime, registerAddrs, relayAddrs, watchless); } catch (InvocationTargetException e) { stop(); Throwable cause = e.getCause() == null ? e : e.getCause(); diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java index 7d8dceb09..b3e864ea6 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java @@ -38,21 +38,26 @@ final class Libp2pEndpointConfig { static final String LISTEN_ADDR_ENV = "CONNECT_LIBP2P_LISTEN_ADDR"; static final String ADVERTISE_ADDRS_ENV = "CONNECT_LIBP2P_ADVERTISE_ADDRS"; static final String RELAY_ADDRS_ENV = "CONNECT_LIBP2P_RELAY_ADDRS"; + private static final List LEGACY_WATCH_CAPABILITIES = Collections.unmodifiableList(Arrays.asList("session", "status")); + private static final List WATCHLESS_CAPABILITIES = Collections.unmodifiableList(Arrays.asList("session", "status", "watchless")); private final List registerAddrs; private final List listenAddrs; private final List advertiseAddrs; private final List relayAddrs; + private final List capabilities; private Libp2pEndpointConfig( List registerAddrs, List listenAddrs, List advertiseAddrs, - List relayAddrs) { + List relayAddrs, + List capabilities) { this.registerAddrs = registerAddrs; this.listenAddrs = listenAddrs; this.advertiseAddrs = advertiseAddrs; this.relayAddrs = relayAddrs; + this.capabilities = capabilities; } static Libp2pEndpointConfig fromEnvironment(Map env) { @@ -65,7 +70,8 @@ static Libp2pEndpointConfig fromEnvironment(Map env) { registerAddrs, listenAddrs, split(env.get(ADVERTISE_ADDRS_ENV)), - split(env.get(RELAY_ADDRS_ENV))); + split(env.get(RELAY_ADDRS_ENV)), + LEGACY_WATCH_CAPABILITIES); } static Libp2pEndpointConfig fromSystemEnvironment() { @@ -73,11 +79,19 @@ static Libp2pEndpointConfig fromSystemEnvironment() { } static Libp2pEndpointConfig fromBootstrap(List registerAddrs, List relayAddrs) { + return fromBootstrap(registerAddrs, relayAddrs, false); + } + + static Libp2pEndpointConfig fromBootstrap( + List registerAddrs, + List relayAddrs, + boolean watchless) { return new Libp2pEndpointConfig( immutableCopy(registerAddrs), Collections.singletonList("/ip4/127.0.0.1/tcp/0"), Collections.emptyList(), - immutableCopy(relayAddrs)); + immutableCopy(relayAddrs), + watchless ? WATCHLESS_CAPABILITIES : LEGACY_WATCH_CAPABILITIES); } boolean enabled() { @@ -100,6 +114,10 @@ List relayAddrs() { return relayAddrs; } + List capabilities() { + return capabilities; + } + List edgePeerIds() { List peerIds = new ArrayList<>(); Set seen = new HashSet<>(); diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java index 6e6725d59..6a0b4c439 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java @@ -111,14 +111,14 @@ public synchronized void start() { startWithConfig(libp2pConfig); } - synchronized void start(List registerAddrs, List relayAddrs) { + synchronized void start(List registerAddrs, List relayAddrs, boolean watchless) { if (started) { return; } Libp2pEndpointConfig environmentConfig = Libp2pEndpointConfig.fromSystemEnvironment(); libp2pConfig = environmentConfig.enabled() ? environmentConfig - : Libp2pEndpointConfig.fromBootstrap(registerAddrs, relayAddrs); + : Libp2pEndpointConfig.fromBootstrap(registerAddrs, relayAddrs, watchless); startWithConfig(libp2pConfig); } @@ -244,7 +244,7 @@ private ActiveRegistration registerOnce(OfflineMode offlineMode, EndpointAuthTyp : connectConfig.getSuperEndpoints(), offlineMode, authType, - Arrays.asList("session", "status"), + libp2pConfig.capabilities(), this::currentCapacity); client = new PeerRegistrationClient(handshake); PeerRegisterResult result = await(client.install( diff --git a/core/src/main/java/com/minekube/connect/watch/WatchBootstrap.java b/core/src/main/java/com/minekube/connect/watch/WatchBootstrap.java index 708c1eff1..a6be64e6f 100644 --- a/core/src/main/java/com/minekube/connect/watch/WatchBootstrap.java +++ b/core/src/main/java/com/minekube/connect/watch/WatchBootstrap.java @@ -30,25 +30,45 @@ public final class WatchBootstrap { public static final String LIBP2P_EDGE_ADDRS_HEADER = "Connect-Libp2p-Edge-Addrs"; public static final String LIBP2P_RELAY_ADDRS_HEADER = "Connect-Libp2p-Relay-Addrs"; + public static final String LIBP2P_CAPABILITIES_HEADER = "Connect-Libp2p-Capabilities"; + + private static final String CAPABILITY_WATCHLESS = "watchless"; private final List libp2pEdgeAddrs; private final List libp2pRelayAddrs; + private final List libp2pCapabilities; - private WatchBootstrap(List libp2pEdgeAddrs, List libp2pRelayAddrs) { - this.libp2pEdgeAddrs = Collections.unmodifiableList(new ArrayList<>(libp2pEdgeAddrs)); - this.libp2pRelayAddrs = Collections.unmodifiableList(new ArrayList<>(libp2pRelayAddrs)); + private WatchBootstrap( + List libp2pEdgeAddrs, + List libp2pRelayAddrs, + List libp2pCapabilities) { + this.libp2pEdgeAddrs = immutableCopy(libp2pEdgeAddrs); + this.libp2pRelayAddrs = immutableCopy(libp2pRelayAddrs); + this.libp2pCapabilities = immutableCopy(libp2pCapabilities); } public static WatchBootstrap fromHeaders(Headers headers) { - return new WatchBootstrap( + return fromLists( parseHeaderValues(headers.values(LIBP2P_EDGE_ADDRS_HEADER)), - parseHeaderValues(headers.values(LIBP2P_RELAY_ADDRS_HEADER))); + parseHeaderValues(headers.values(LIBP2P_RELAY_ADDRS_HEADER)), + parseHeaderValues(headers.values(LIBP2P_CAPABILITIES_HEADER))); + } + + public static WatchBootstrap fromLists( + List libp2pEdgeAddrs, + List libp2pRelayAddrs, + List libp2pCapabilities) { + return new WatchBootstrap(libp2pEdgeAddrs, libp2pRelayAddrs, libp2pCapabilities); } public boolean hasLibp2p() { return !libp2pEdgeAddrs.isEmpty(); } + public boolean supportsWatchlessLibp2p() { + return libp2pCapabilities.contains(CAPABILITY_WATCHLESS); + } + public List libp2pEdgeAddrs() { return libp2pEdgeAddrs; } @@ -75,4 +95,11 @@ private static List parseHeaderValues(List values) { } return out; } + + private static List immutableCopy(List values) { + if (values == null || values.isEmpty()) { + return Collections.emptyList(); + } + return Collections.unmodifiableList(new ArrayList<>(values)); + } } 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 6fef80016..673be52d7 100644 --- a/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java +++ b/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java @@ -14,13 +14,16 @@ import com.minekube.connect.api.SimpleConnectApi; import com.minekube.connect.api.inject.PlatformInjector; import com.minekube.connect.api.logger.ConnectLogger; +import com.minekube.connect.tunnel.p2p.Libp2pEndpoint; import com.minekube.connect.tunnel.Tunneler; import com.minekube.connect.watch.SessionProposal; +import com.minekube.connect.watch.WatchBootstrap; import com.minekube.connect.watch.WatchClient; import com.minekube.connect.watch.Watcher; import java.lang.reflect.Field; import java.lang.reflect.Method; import java.net.InetSocketAddress; +import java.util.List; import java.util.Timer; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicBoolean; @@ -126,6 +129,25 @@ void resetBackOffTimerDoesNotThrowWhenSchedulerIsClearedAfterStop() throws Excep assertNull(scheduler(register)); } + @Test + void watchOpenStartsLibp2pEndpointFromBootstrap() throws Exception { + Fixture fixture = newFixture(); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + watcher.getValue().onOpen(WatchBootstrap.fromLists( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("session", "status", "watchless"))); + + verify(fixture.libp2pEndpoint).start( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + true); + } + @Test void watcherRegisterDoesNotDeclareTimerFields() { assertNoTimerFields(WatcherRegister.class); @@ -180,10 +202,12 @@ private static Fixture newFixture() throws Exception { inject(register, "platformInjector", mock(PlatformInjector.class)); inject(register, "logger", mock(ConnectLogger.class)); inject(register, "api", mock(SimpleConnectApi.class)); + inject(register, "libp2pEndpoint", mock(Libp2pEndpoint.class)); when(((PlatformInjector) getField(register, "platformInjector")).getServerSocketAddress()) .thenReturn(new InetSocketAddress("127.0.0.1", 25565)); return new Fixture(register, watchClient, (Tunneler) getField(register, "tunneler"), - (PlatformInjector) getField(register, "platformInjector")); + (PlatformInjector) getField(register, "platformInjector"), + (Libp2pEndpoint) getField(register, "libp2pEndpoint")); } private static void inject(WatcherRegister register, String fieldName, Object value) @@ -225,17 +249,20 @@ private static final class Fixture { private final WatchClient watchClient; private final Tunneler tunneler; private final PlatformInjector platformInjector; + private final Libp2pEndpoint libp2pEndpoint; private Fixture( WatcherRegister register, WatchClient watchClient, Tunneler tunneler, - PlatformInjector platformInjector + PlatformInjector platformInjector, + Libp2pEndpoint libp2pEndpoint ) { this.register = register; this.watchClient = watchClient; this.tunneler = tunneler; this.platformInjector = platformInjector; + this.libp2pEndpoint = libp2pEndpoint; } } } diff --git a/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfigTest.java b/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfigTest.java index 603650c2c..85e017c19 100644 --- a/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfigTest.java +++ b/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfigTest.java @@ -5,6 +5,7 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.List; import java.util.Map; import org.junit.jupiter.api.Test; @@ -17,6 +18,30 @@ void disabledWhenRegisterAddressIsMissing() { Libp2pEndpointConfig config = Libp2pEndpointConfig.fromEnvironment(Map.of()); assertFalse(config.enabled()); + assertEquals(List.of("session", "status"), config.capabilities()); + } + + @Test + void bootstrapConfigEnablesWatchlessWhenSupportedByEdge() { + Libp2pEndpointConfig config = Libp2pEndpointConfig.fromBootstrap( + List.of("/dns4/connect.example/tcp/4001/p2p/" + MOXY_PEER), + List.of("/dns4/connect.example/tcp/4001/p2p/" + MOXY_PEER), + true); + + assertTrue(config.enabled()); + assertEquals(List.of("/ip4/127.0.0.1/tcp/0"), config.listenAddrs()); + assertEquals(List.of("session", "status", "watchless"), config.capabilities()); + } + + @Test + void bootstrapConfigKeepsLegacyWatchCapabilityWhenWatchlessIsUnsupported() { + Libp2pEndpointConfig config = Libp2pEndpointConfig.fromBootstrap( + List.of("/dns4/connect.example/tcp/4001/p2p/" + MOXY_PEER), + List.of("/dns4/connect.example/tcp/4001/p2p/" + MOXY_PEER), + false); + + assertTrue(config.enabled()); + assertEquals(List.of("session", "status"), config.capabilities()); } @Test diff --git a/core/src/test/java/com/minekube/connect/watch/WatchBootstrapTest.java b/core/src/test/java/com/minekube/connect/watch/WatchBootstrapTest.java index 5a0012060..819246bd6 100644 --- a/core/src/test/java/com/minekube/connect/watch/WatchBootstrapTest.java +++ b/core/src/test/java/com/minekube/connect/watch/WatchBootstrapTest.java @@ -13,20 +13,34 @@ void parsesLibp2pBootstrapHeaders() { Headers headers = new Headers.Builder() .add(WatchBootstrap.LIBP2P_EDGE_ADDRS_HEADER, " /ip4/127.0.0.1/tcp/4001/p2p/a , /ip4/127.0.0.1/tcp/4001/p2p/b ") .add(WatchBootstrap.LIBP2P_RELAY_ADDRS_HEADER, "/ip4/127.0.0.1/tcp/4001/p2p/a") + .add(WatchBootstrap.LIBP2P_CAPABILITIES_HEADER, "session,status,watchless") .build(); WatchBootstrap bootstrap = WatchBootstrap.fromHeaders(headers); assertTrue(bootstrap.hasLibp2p()); + assertTrue(bootstrap.supportsWatchlessLibp2p()); assertEquals(2, bootstrap.libp2pEdgeAddrs().size()); assertEquals("/ip4/127.0.0.1/tcp/4001/p2p/a", bootstrap.libp2pEdgeAddrs().get(0)); assertEquals("/ip4/127.0.0.1/tcp/4001/p2p/a", bootstrap.libp2pRelayAddrs().get(0)); } + @Test + void missingWatchlessCapabilityKeepsLegacyWatchMode() { + WatchBootstrap bootstrap = WatchBootstrap.fromHeaders(new Headers.Builder() + .add(WatchBootstrap.LIBP2P_EDGE_ADDRS_HEADER, "/ip4/127.0.0.1/tcp/4001/p2p/a") + .add(WatchBootstrap.LIBP2P_CAPABILITIES_HEADER, "session,status") + .build()); + + assertTrue(bootstrap.hasLibp2p()); + assertFalse(bootstrap.supportsWatchlessLibp2p()); + } + @Test void emptyHeadersDisableLibp2pBootstrap() { WatchBootstrap bootstrap = WatchBootstrap.fromHeaders(Headers.of()); assertFalse(bootstrap.hasLibp2p()); + assertFalse(bootstrap.supportsWatchlessLibp2p()); } } From b703bfae3f95822b5cfa9efc6a228028529e7497 Mon Sep 17 00:00:00 2001 From: Robin Date: Fri, 3 Jul 2026 22:09:24 +0200 Subject: [PATCH 5/5] prefer connect libp2p watchless mode --- .../connect/register/WatcherRegister.java | 94 ++++++++++++++-- .../connect/tunnel/p2p/Libp2pEndpoint.java | 19 +++- .../tunnel/p2p/Libp2pEndpointConfig.java | 15 ++- .../tunnel/p2p/Libp2pEndpointRuntime.java | 70 ++++++++++-- .../tunnel/p2p/Libp2pStatusReporter.java | 4 +- .../tunnel/p2p/Libp2pWatchlessController.java | 91 ++++++++++++++++ .../connect/register/WatcherRegisterTest.java | 100 +++++++++++++++++- .../p2p/Libp2pWatchlessControllerTest.java | 89 ++++++++++++++++ 8 files changed, 456 insertions(+), 26 deletions(-) create mode 100644 core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessController.java create mode 100644 core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessControllerTest.java 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 35372b442..8c12fe916 100644 --- a/core/src/main/java/com/minekube/connect/register/WatcherRegister.java +++ b/core/src/main/java/com/minekube/connect/register/WatcherRegister.java @@ -33,16 +33,15 @@ import com.minekube.connect.api.logger.ConnectLogger; import com.minekube.connect.network.netty.LocalSession; import com.minekube.connect.tunnel.Tunneler; +import com.minekube.connect.tunnel.p2p.Libp2pEndpoint; import com.minekube.connect.util.Utils; import com.minekube.connect.util.backoff.BackOff; import com.minekube.connect.util.backoff.ExponentialBackOff; -import com.minekube.connect.tunnel.p2p.Libp2pEndpoint; import com.minekube.connect.watch.SessionProposal; import com.minekube.connect.watch.SessionProposal.State; import com.minekube.connect.watch.WatchBootstrap; import com.minekube.connect.watch.WatchClient; import com.minekube.connect.watch.Watcher; -import com.minekube.connect.tunnel.p2p.Libp2pEndpoint; import io.netty.util.concurrent.DefaultThreadFactory; import java.io.IOException; import java.time.Duration; @@ -67,6 +66,8 @@ public class WatcherRegister { // volatile: written from injection thread (start/stop) and read from the // scheduler thread (retry) and OkHttp dispatcher (WatcherImpl callbacks). private volatile WebSocket ws; + private volatile WatcherImpl watcher; + private volatile boolean legacyWatchEnabled; private ExponentialBackOff backOffPolicy; private final AtomicBoolean started = new AtomicBoolean(); @@ -79,6 +80,7 @@ public class WatcherRegister { @Inject public void start() { if (started.compareAndSet(false, true)) { + legacyWatchEnabled = true; scheduler = Executors.newSingleThreadScheduledExecutor( new DefaultThreadFactory("connect-watcher-scheduler", true)); backOffPolicy = new ExponentialBackOff.Builder() @@ -102,6 +104,7 @@ public void stop() { return; } logger.info("Stopped watching for sessions"); + legacyWatchEnabled = false; if (retryFuture != null) { retryFuture.cancel(false); retryFuture = null; @@ -111,13 +114,17 @@ public void stop() { scheduler = null; } if (ws != null) { + if (watcher != null) { + watcher.ignoreTerminalEvents(); + } ws.close(1000, "watcher stopped"); ws = null; } + watcher = null; } private void retry() { - if (!started.get()) { + if (!started.get() || !legacyWatchEnabled) { return; } if (retryFuture != null) { @@ -142,29 +149,81 @@ private void retry() { logger.info("Trying to reconnect in {}...", Utils.humanReadableFormat(Duration.ofMillis(millis))); retryFuture = s.schedule(() -> { - if (started.get()) { + if (started.get() && legacyWatchEnabled) { watch(); } }, millis, TimeUnit.MILLISECONDS); } private void watch() { + if (!started.get() || !legacyWatchEnabled) { + return; + } if (ws != null) { + if (watcher != null) { + watcher.ignoreTerminalEvents(); + } ws.close(1000, "watcher is reconnecting"); } - ws = watchClient.watch(new WatcherImpl()); + watcher = new WatcherImpl(); + ws = watchClient.watch(watcher); + } + + private void enterWatchlessMode() { + if (!started.get()) { + return; + } + legacyWatchEnabled = false; + if (retryFuture != null) { + retryFuture.cancel(false); + retryFuture = null; + } + WatcherImpl currentWatcher = watcher; + if (currentWatcher != null) { + currentWatcher.ignoreTerminalEvents(); + } + WebSocket currentWebSocket = ws; + ws = null; + watcher = null; + logger.info("Connect libp2p watchless endpoint mode enabled"); + if (currentWebSocket != null) { + currentWebSocket.close(1000, "watchless endpoint mode enabled"); + } + } + + private void enterLegacyWatchFallback() { + if (!started.get()) { + return; + } + if (legacyWatchEnabled && ws != null) { + return; + } + legacyWatchEnabled = true; + resetBackOff(); + logger.info("Connect libp2p watchless endpoint fallback enabled WatchService"); + watch(); } private class WatcherImpl implements Watcher { + private volatile boolean ignoreTerminalEvents; @Override public void onOpen(WatchBootstrap bootstrap) { logger.translatedInfo("connect.watch.started"); if (bootstrap.hasLibp2p()) { - libp2pEndpoint.start( - bootstrap.libp2pEdgeAddrs(), - bootstrap.libp2pRelayAddrs(), - bootstrap.supportsWatchlessLibp2p()); + if (bootstrap.supportsWatchlessLibp2p()) { + libp2pEndpoint.start( + bootstrap.libp2pEdgeAddrs(), + bootstrap.libp2pRelayAddrs(), + true, + WatcherRegister.this::enterWatchlessMode, + WatcherRegister.this::enterLegacyWatchFallback); + } else { + libp2pEndpoint.start( + bootstrap.libp2pEdgeAddrs(), + bootstrap.libp2pRelayAddrs(), + false); + } } startResetBackOffTimer(); } @@ -208,6 +267,9 @@ public void onProposal(SessionProposal proposal) { @Override public void onError(Throwable t) { + if (shouldIgnoreTerminalEvent()) { + return; + } logger.error("Connection error with WatchService: " + t + ( t.getCause() == null ? "" @@ -220,6 +282,9 @@ public void onError(Throwable t) { @Override public void onCompleted() { + if (shouldIgnoreTerminalEvent()) { + return; + } cancelResetBackOffTimer(); retry(); } @@ -230,7 +295,7 @@ void startResetBackOffTimer() { cancelResetBackOffTimer(); // Snapshot: a late onOpen after stop() can land here with scheduler == null. ScheduledExecutorService s = scheduler; - if (s == null || !started.get()) { + if (s == null || !started.get() || !legacyWatchEnabled || ignoreTerminalEvents) { return; } resetBackOffFuture = s.schedule(() -> { @@ -246,5 +311,14 @@ void cancelResetBackOffTimer() { resetBackOffFuture = null; } } + + void ignoreTerminalEvents() { + ignoreTerminalEvents = true; + cancelResetBackOffTimer(); + } + + private boolean shouldIgnoreTerminalEvent() { + return ignoreTerminalEvents || !legacyWatchEnabled || !started.get(); + } } } diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java index 3e752ec8d..4a8783138 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpoint.java @@ -43,6 +43,7 @@ public class Libp2pEndpoint { private Object runtime; private Method startMethod; private Method startBootstrapMethod; + private Method startWatchlessMethod; private Method stopMethod; @Inject @@ -80,9 +81,12 @@ public Libp2pEndpoint( this.startMethod = runtimeClass.getDeclaredMethod("start"); this.startBootstrapMethod = runtimeClass.getDeclaredMethod( "start", List.class, List.class, boolean.class); + this.startWatchlessMethod = runtimeClass.getDeclaredMethod( + "start", List.class, List.class, boolean.class, Runnable.class, Runnable.class); this.stopMethod = runtimeClass.getDeclaredMethod("stop"); this.startMethod.setAccessible(true); this.startBootstrapMethod.setAccessible(true); + this.startWatchlessMethod.setAccessible(true); this.stopMethod.setAccessible(true); } catch (Exception | LinkageError e) { this.runtime = null; @@ -108,11 +112,24 @@ public synchronized void start() { } public synchronized void start(List registerAddrs, List relayAddrs, boolean watchless) { + start(registerAddrs, relayAddrs, watchless, null, null); + } + + public synchronized void start( + List registerAddrs, + List relayAddrs, + boolean watchless, + Runnable watchlessReady, + Runnable watchFallback) { if (runtime == null) { return; } try { - startBootstrapMethod.invoke(runtime, registerAddrs, relayAddrs, watchless); + if (watchlessReady == null && watchFallback == null) { + startBootstrapMethod.invoke(runtime, registerAddrs, relayAddrs, watchless); + } else { + startWatchlessMethod.invoke(runtime, registerAddrs, relayAddrs, watchless, watchlessReady, watchFallback); + } } catch (InvocationTargetException e) { stop(); Throwable cause = e.getCause() == null ? e : e.getCause(); diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java index b3e864ea6..781260ae4 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointConfig.java @@ -46,18 +46,21 @@ final class Libp2pEndpointConfig { private final List advertiseAddrs; private final List relayAddrs; private final List capabilities; + private final boolean watchless; private Libp2pEndpointConfig( List registerAddrs, List listenAddrs, List advertiseAddrs, List relayAddrs, - List capabilities) { + List capabilities, + boolean watchless) { this.registerAddrs = registerAddrs; this.listenAddrs = listenAddrs; this.advertiseAddrs = advertiseAddrs; this.relayAddrs = relayAddrs; this.capabilities = capabilities; + this.watchless = watchless; } static Libp2pEndpointConfig fromEnvironment(Map env) { @@ -71,7 +74,8 @@ static Libp2pEndpointConfig fromEnvironment(Map env) { listenAddrs, split(env.get(ADVERTISE_ADDRS_ENV)), split(env.get(RELAY_ADDRS_ENV)), - LEGACY_WATCH_CAPABILITIES); + LEGACY_WATCH_CAPABILITIES, + false); } static Libp2pEndpointConfig fromSystemEnvironment() { @@ -91,7 +95,8 @@ static Libp2pEndpointConfig fromBootstrap( Collections.singletonList("/ip4/127.0.0.1/tcp/0"), Collections.emptyList(), immutableCopy(relayAddrs), - watchless ? WATCHLESS_CAPABILITIES : LEGACY_WATCH_CAPABILITIES); + watchless ? WATCHLESS_CAPABILITIES : LEGACY_WATCH_CAPABILITIES, + watchless); } boolean enabled() { @@ -118,6 +123,10 @@ List capabilities() { return capabilities; } + boolean watchless() { + return watchless; + } + List edgePeerIds() { List peerIds = new ArrayList<>(); Set seen = new HashSet<>(); diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java index 6a0b4c439..56653424f 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pEndpointRuntime.java @@ -67,6 +67,7 @@ final class Libp2pEndpointRuntime { private static final long STREAM_TIMEOUT_SECONDS = 5; private static final int REGISTER_ATTEMPTS_PER_ADDRESS = 4; private static final int RELAY_RESERVATION_ATTEMPTS_PER_ADDRESS = 4; + private static final int WATCHLESS_STATUS_FAILURE_THRESHOLD = 3; private final Path dataDirectory; private final ConnectConfig connectConfig; @@ -84,6 +85,7 @@ final class Libp2pEndpointRuntime { private ScheduledExecutorService registrationExecutor; private ScheduledExecutorService statusExecutor; private ActiveRegistration activeRegistration; + private Libp2pWatchlessController watchlessController; private List reservedRelayAddrs = Collections.emptyList(); private boolean started; @@ -112,6 +114,15 @@ public synchronized void start() { } synchronized void start(List registerAddrs, List relayAddrs, boolean watchless) { + start(registerAddrs, relayAddrs, watchless, null, null); + } + + synchronized void start( + List registerAddrs, + List relayAddrs, + boolean watchless, + Runnable watchlessReady, + Runnable watchFallback) { if (started) { return; } @@ -119,10 +130,14 @@ synchronized void start(List registerAddrs, List relayAddrs, boo libp2pConfig = environmentConfig.enabled() ? environmentConfig : Libp2pEndpointConfig.fromBootstrap(registerAddrs, relayAddrs, watchless); - startWithConfig(libp2pConfig); + startWithConfig(libp2pConfig, watchlessReady, watchFallback); } private void startWithConfig(Libp2pEndpointConfig config) { + startWithConfig(config, null, null); + } + + private void startWithConfig(Libp2pEndpointConfig config, Runnable watchlessReady, Runnable watchFallback) { if (started) { return; } @@ -130,6 +145,11 @@ private void startWithConfig(Libp2pEndpointConfig config) { if (!libp2pConfig.enabled()) { return; } + watchlessController = new Libp2pWatchlessController( + libp2pConfig.watchless(), + WATCHLESS_STATUS_FAILURE_THRESHOLD, + watchlessReady, + watchFallback); try { identity = EndpointPeerIdentity.loadOrCreate(dataDirectory.resolve("libp2p-identity.key")); host = Libp2pTunnelTransportRuntime.createHost( @@ -164,6 +184,7 @@ public synchronized void stop() { activeRegistration.close(); activeRegistration = null; } + watchlessController = null; reservedRelayAddrs = Collections.emptyList(); if (statusExecutor != null) { statusExecutor.shutdownNow(); @@ -344,6 +365,8 @@ private void scheduleRegisterRetry(OfflineMode offlineMode, EndpointAuthType aut private void registerAndWatch(OfflineMode offlineMode, EndpointAuthType authType) { ActiveRegistration registration = registerOnce(offlineMode, authType); ActiveRegistration previous; + Libp2pWatchlessController controller; + boolean statusHealthy; synchronized (this) { if (!started) { registration.close(); @@ -353,12 +376,27 @@ private void registerAndWatch(OfflineMode offlineMode, EndpointAuthType authType activeRegistration = registration; logger.info("Connect libp2p endpoint registered: " + registration.result().getEndpointId() + " (" + identity.peerId() + ")"); - startStatusReporter(registration.result()); + statusHealthy = startStatusReporter(registration.result()); + controller = watchlessController; } if (previous != null) { previous.close(); } + if (controller != null) { + controller.registrationCommitted(statusHealthy); + } registration.client().closedFuture().whenComplete((ignored, error) -> { + Libp2pWatchlessController currentController; + synchronized (this) { + if (activeRegistration != registration) { + return; + } + currentController = watchlessController; + stopStatusReporter(); + } + if (currentController != null) { + currentController.registrationClosed(); + } if (error == null) { logger.info("Connect libp2p registration stream closed; reconnecting"); } else if (Libp2pEndpointErrors.isTransientConnectError(error)) { @@ -518,11 +556,8 @@ private static T await(CompletableFuture future, long timeoutSeconds, Str } } - private void startStatusReporter(PeerRegisterResult result) { - if (statusExecutor != null) { - statusExecutor.shutdownNow(); - statusExecutor = null; - } + private boolean startStatusReporter(PeerRegisterResult result) { + stopStatusReporter(); Libp2pStatusReporter reporter = new Libp2pStatusReporter( host, endpointInstanceId, @@ -531,17 +566,34 @@ private void startStatusReporter(PeerRegisterResult result) { result, platformUtils, logger); - reporter.reportSafely(); + boolean initialStatusHealthy = reporter.reportSafely(); statusExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { Thread thread = new Thread(runnable, "connect-libp2p-status-report"); thread.setDaemon(true); return thread; }); statusExecutor.scheduleWithFixedDelay( - reporter::reportSafely, + () -> { + boolean healthy = reporter.reportSafely(); + Libp2pWatchlessController controller; + synchronized (this) { + controller = watchlessController; + } + if (controller != null) { + controller.statusReportResult(healthy); + } + }, Libp2pStatusReporter.REPORT_INTERVAL_SECONDS, Libp2pStatusReporter.REPORT_INTERVAL_SECONDS, TimeUnit.SECONDS); + return initialStatusHealthy; + } + + private void stopStatusReporter() { + if (statusExecutor != null) { + statusExecutor.shutdownNow(); + statusExecutor = null; + } } private static final class ActiveRegistration { diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pStatusReporter.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pStatusReporter.java index db96c913f..ce2769191 100644 --- a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pStatusReporter.java +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pStatusReporter.java @@ -98,9 +98,10 @@ final class Libp2pStatusReporter { this.streamOpener = streamOpener == null ? this::openStatusStream : streamOpener; } - void reportSafely() { + boolean reportSafely() { try { reportOnce(System.currentTimeMillis()); + return true; } catch (RuntimeException e) { if (Libp2pEndpointErrors.isTransientConnectError(e)) { logger.warn("Connect libp2p status report failed; retrying: " @@ -108,6 +109,7 @@ void reportSafely() { } else { logger.error("Failed to report Connect libp2p status", e); } + return false; } } diff --git a/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessController.java b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessController.java new file mode 100644 index 000000000..515059b1c --- /dev/null +++ b/core/src/main/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessController.java @@ -0,0 +1,91 @@ +/* + * Copyright (c) 2021-2022 Minekube. https://minekube.com + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in + * all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * DEALINGS IN THE SOFTWARE. + * + * @author Minekube + * @link https://github.com/minekube/connect-java + */ + +package com.minekube.connect.tunnel.p2p; + +import java.util.Objects; + +final class Libp2pWatchlessController { + private static final Runnable NOOP = () -> { + }; + + private final boolean supported; + private final int statusFailureThreshold; + private final Runnable readyCallback; + private final Runnable fallbackCallback; + + private boolean ready; + private boolean registrationActive; + private int statusFailures; + + Libp2pWatchlessController( + boolean supported, + int statusFailureThreshold, + Runnable readyCallback, + Runnable fallbackCallback) { + this.supported = supported; + this.statusFailureThreshold = Math.max(1, statusFailureThreshold); + this.readyCallback = readyCallback == null ? NOOP : Objects.requireNonNull(readyCallback, "readyCallback"); + this.fallbackCallback = fallbackCallback == null ? NOOP : Objects.requireNonNull(fallbackCallback, "fallbackCallback"); + } + + synchronized void registrationCommitted(boolean statusHealthy) { + if (!supported) { + return; + } + registrationActive = true; + statusReportResult(statusHealthy); + } + + synchronized void statusReportResult(boolean healthy) { + if (!supported || !registrationActive) { + return; + } + if (healthy) { + statusFailures = 0; + if (!ready) { + ready = true; + readyCallback.run(); + } + return; + } + statusFailures++; + if (ready && statusFailures >= statusFailureThreshold) { + ready = false; + fallbackCallback.run(); + } + } + + synchronized void registrationClosed() { + if (!supported) { + return; + } + registrationActive = false; + statusFailures = 0; + if (!ready) { + return; + } + ready = false; + fallbackCallback.run(); + } +} 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 673be52d7..1331ed989 100644 --- a/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java +++ b/core/src/test/java/com/minekube/connect/register/WatcherRegisterTest.java @@ -7,8 +7,13 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; import static org.mockito.Mockito.when; import com.minekube.connect.api.SimpleConnectApi; @@ -27,6 +32,7 @@ import java.util.Timer; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.atomic.AtomicBoolean; +import org.mockito.ArgumentCaptor; import minekube.connect.v1alpha1.WatchServiceOuterClass.GameProfile; import minekube.connect.v1alpha1.WatchServiceOuterClass.Player; import minekube.connect.v1alpha1.WatchServiceOuterClass.Session; @@ -35,7 +41,7 @@ import okhttp3.WebSocket; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; -import org.mockito.ArgumentCaptor; +import org.mockito.stubbing.Answer; class WatcherRegisterTest { private WatcherRegister register; @@ -142,10 +148,87 @@ void watchOpenStartsLibp2pEndpointFromBootstrap() throws Exception { List.of("/dns4/connect.example/tcp/4001/p2p/edge"), List.of("session", "status", "watchless"))); + verify(fixture.libp2pEndpoint).start( + eq(List.of("/dns4/connect.example/tcp/4001/p2p/edge")), + eq(List.of("/dns4/connect.example/tcp/4001/p2p/edge")), + eq(true), + any(Runnable.class), + any(Runnable.class)); + } + + @Test + void watchlessReadyClosesLegacyWatchSocketAndSuppressesReconnect() throws Exception { + Fixture fixture = newFixture(); + RunnableCaptor callbacks = new RunnableCaptor(); + doAnswer(callbacks.capture()).when(fixture.libp2pEndpoint).start( + any(), + any(), + anyBoolean(), + any(Runnable.class), + any(Runnable.class)); + WebSocket webSocket = mock(WebSocket.class); + when(fixture.watchClient.watch(any(Watcher.class))).thenReturn(webSocket); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + watcher.getValue().onOpen(WatchBootstrap.fromLists( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("session", "status", "watchless"))); + callbacks.ready.run(); + watcher.getValue().onCompleted(); + + verify(webSocket).close(1000, "watchless endpoint mode enabled"); + verifyNoMoreInteractions(fixture.watchClient); + } + + @Test + void watchlessFallbackReopensLegacyWatchSocket() throws Exception { + Fixture fixture = newFixture(); + RunnableCaptor callbacks = new RunnableCaptor(); + doAnswer(callbacks.capture()).when(fixture.libp2pEndpoint).start( + any(), + any(), + anyBoolean(), + any(Runnable.class), + any(Runnable.class)); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + watcher.getValue().onOpen(WatchBootstrap.fromLists( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("session", "status", "watchless"))); + callbacks.ready.run(); + callbacks.fallback.run(); + + verify(fixture.watchClient, times(2)).watch(any(Watcher.class)); + } + + @Test + void oldWatchBootstrapKeepsLegacyWatchSocketPrimary() throws Exception { + Fixture fixture = newFixture(); + WebSocket webSocket = mock(WebSocket.class); + when(fixture.watchClient.watch(any(Watcher.class))).thenReturn(webSocket); + register = fixture.register; + register.start(); + ArgumentCaptor watcher = ArgumentCaptor.forClass(Watcher.class); + verify(fixture.watchClient).watch(watcher.capture()); + + watcher.getValue().onOpen(WatchBootstrap.fromLists( + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("/dns4/connect.example/tcp/4001/p2p/edge"), + List.of("session", "status"))); + verify(fixture.libp2pEndpoint).start( List.of("/dns4/connect.example/tcp/4001/p2p/edge"), List.of("/dns4/connect.example/tcp/4001/p2p/edge"), - true); + false); + verifyNoMoreInteractions(webSocket); } @Test @@ -265,4 +348,17 @@ private Fixture( this.libp2pEndpoint = libp2pEndpoint; } } + + private static final class RunnableCaptor { + private Runnable ready; + private Runnable fallback; + + private Answer capture() { + return invocation -> { + ready = invocation.getArgument(3); + fallback = invocation.getArgument(4); + return null; + }; + } + } } diff --git a/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessControllerTest.java b/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessControllerTest.java new file mode 100644 index 000000000..ab75de545 --- /dev/null +++ b/core/src/test/java/com/minekube/connect/tunnel/p2p/Libp2pWatchlessControllerTest.java @@ -0,0 +1,89 @@ +package com.minekube.connect.tunnel.p2p; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.junit.jupiter.api.Test; + +class Libp2pWatchlessControllerTest { + @Test + void registrationAndHealthyStatusEnterWatchlessOnce() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + true, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.statusReportResult(true); + + assertEquals(1, ready.count); + assertEquals(0, fallback.count); + } + + @Test + void registrationCloseFallsBackAfterWatchlessReady() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + true, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.registrationClosed(); + + assertEquals(1, ready.count); + assertEquals(1, fallback.count); + } + + @Test + void repeatedStatusFailuresFallBackAfterThreshold() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + true, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.statusReportResult(false); + controller.statusReportResult(false); + controller.statusReportResult(false); + + assertEquals(1, ready.count); + assertEquals(1, fallback.count); + } + + @Test + void staleHealthyStatusAfterRegistrationCloseDoesNotReenterWatchless() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + true, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.registrationClosed(); + controller.statusReportResult(true); + + assertEquals(1, ready.count); + assertEquals(1, fallback.count); + } + + @Test + void unsupportedWatchlessDoesNotNotify() { + Counter ready = new Counter(); + Counter fallback = new Counter(); + Libp2pWatchlessController controller = new Libp2pWatchlessController( + false, 3, ready::increment, fallback::increment); + + controller.registrationCommitted(true); + controller.statusReportResult(false); + controller.registrationClosed(); + + assertEquals(0, ready.count); + assertEquals(0, fallback.count); + } + + private static final class Counter { + private int count; + + private void increment() { + count++; + } + } +}