Netty transports

.github/workflows/compatibility.yaml

@@ -0,0 +1,41 @@
+# Owned by node; management-terraform supplies the separate classpath test workflow.
+name: Compatibility
+
+on:
+  pull_request:
+  workflow_call:
+
+permissions:
+  contents: read
+
+jobs:
+  compatibility:
+    name: Released node compatibility
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v7
+
+      - uses: gradle/actions/wrapper-validation@v6
+
+      - uses: graalvm/setup-graalvm@v1
+        with:
+          distribution: 'graalvm'
+          version: '25i-latest'
+          native-image-musl: true
+          cache: 'gradle'
+
+      - uses: gradle/actions/setup-gradle@v6
+
+      - name: Test against the released node
+        run: ./gradlew test --tests cash.atto.node.CucumberTest --no-daemon --info --fail-fast -Datto.test.neighbour-mode=live
+
+      - uses: actions/upload-artifact@v7
+        if: failure()
+        with:
+          name: released-node-compatibility-reports
+          path: |
+            build/reports/tests/test
+            build/test-results/test
+            build/reports/Cucumber.html
+            build/reports/released-neighbours

.github/workflows/pipeline.yaml

@@ -11,12 +11,15 @@
     uses: ./.github/workflows/test.yaml
     with:
       agent_enabled: true
+  compatibility:
+    uses: ./.github/workflows/compatibility.yaml
   build:
     permissions:
       contents: write
       packages: write
     needs:
       - test
+      - compatibility
     uses: ./.github/workflows/build.yaml
   release-tag:
     permissions:

.github/workflows/pull-request.yaml

@@ -7,12 +7,4 @@ on:
 
 jobs:
   test:
-    strategy:
-      fail-fast: false
-      matrix:
-        neighbour-mode:
-          - classpath
-          - live
     uses: ./.github/workflows/test.yaml
-    with:
-      neighbour_mode: ${{ matrix['neighbour-mode'] }}

.github/workflows/test.yaml

@@ -8,11 +8,6 @@ on:
         required: false
         default: false
         type: boolean
-      neighbour_mode:
-        description: "Neighbour node implementation used by integration tests"
-        required: false
-        default: "classpath"
-        type: string
 
 jobs:
   test:
@@ -31,7 +26,7 @@ jobs:
 
       - uses: gradle/actions/setup-gradle@v6
 
-      - run: ./gradlew ${{ inputs.agent_enabled && '-Pagent' || '' }} test --no-daemon --info --fail-fast -Datto.test.neighbour-mode=${{ inputs.neighbour_mode }}
+      - run: ./gradlew ${{ inputs.agent_enabled && '-Pagent' || '' }} test --no-daemon --info --fail-fast
 
       - uses: actions/upload-artifact@v7
         if: inputs.agent_enabled

NETWORK.md

@@ -4,21 +4,61 @@ The networking subsystem authenticates peers in [NetworkProcessor][processor], m
 [NodeConnectionManager][connections], and counts traffic in [Guardian][guardian]. These owners communicate through direct
 queries and asynchronous network-message and lifecycle events.
 
+## Transport Ownership
+
+The [peer server][peer-server] uses Reactor Netty directly on `websocket.port` (default `8082`). It owns its I/O event loops
+and virtual-thread application dispatcher. Spring WebFlux controllers remain on `server.port` (default `8080`), with management
+on `8081` by default.
+
+[PeerNettyClient][peer-client] owns the outgoing Reactor Netty WebSocket and HTTP callback clients. They use private event
+loops, a virtual-thread application dispatcher, separate connection providers and an asynchronous bridge to the existing JVM
+DNS resolver. Neither peer transport uses Spring's shared Reactor resources. Request hostnames remain available for HTTP Host,
+TLS SNI and certificate hostname verification.
+Callback POSTs do not follow redirects or retry automatically; WebSocket redirects retain the existing secure-downgrade
+restriction and request limit.
+
+[PeerSession][peer-session] exposes complete binary messages, ordered enqueue, graceful close and cancellation to the
+connection manager. The [Reactor session][peer-reactor-session] owns transport framing and cleanup for inbound and outbound
+connections, including enabling frame aggregation synchronously in the upgrade callback, before application dispatch.
+Both directions retain a 300-byte receive limit,
+including cumulative fragmented messages, and bounded incoming/outgoing queues with close-on-overflow. These transport limits
+belong to the session contract.
+Text messages are ignored after admission to the incoming queue; ping/pong and close frames remain transport concerns.
+
 ## Peer Authentication
 
-`NetworkProcessor` starts an embedded Ktor CIO server with two endpoints:
+`NetworkProcessor` starts its private peer server with two endpoints:
 
 - **`POST /handshakes`** accepts a `CounterChallengeResponse`. Guardian admits the remote address, the processor consumes
   a one-time challenge, checks the expected public URI, genesis hash, signature and counter-challenge prefix, then signs
   a `ChallengeResponse`. Successful authentication completes the waiting outbound attempt before the HTTP response is sent.
 - **`WS /`** accepts a WebSocket with `Atto-Public-Uri` and `Atto-Http-Challenge` headers. Guardian admits the remote address,
   the processor validates the headers and challenge prefix, then posts a counter-challenge to the peer's `/handshakes`
   endpoint. It checks the returned URI and signature before handing the session to `NodeConnectionManager.manage()`.
+  Authentication owns its attempt reservation: failure or cancellation removes that attempt conditionally by identity;
+  success retains the existing five-second expiry window.
 
 An inbound HTTP handshake has a five-second preparation deadline covering admission, body reading, verification and signing.
 A separate semaphore limits concurrent HTTP handshakes to 64. The [callback service][callback] applies a five-second deadline
 covering URI validation, request construction/signing and the callback exchange. Both handshake payload directions have bounded
 body readers in [HandshakePayloadLimit][payload-limit]. Cancellation propagates to ongoing handshake work.
+Early HTTP rejection closes the connection after writing its status, so unread request content cannot leave a reusable
+connection stalled. WebSocket authentication runs after upgrade; a rejected attempt terminates without registering a peer.
+
+### Remote Signing
+
+When `atto.signer.backend=REMOTE`, [NettyRemoteSigner][remote-signer] calls the existing signer's `public-keys`, `blocks`,
+`votes` and `challenges` endpoints. It preserves the configured authorization token, lazy public-key cache, signature checks,
+one-second retry delay and socket timeouts (five seconds for the public key, one second for signing). HTTP 408 and 429 are
+retried, as are transport and decoding failures; other HTTP errors terminate the operation. Cancellation stops retrying.
+
+The signer owns JSON, retries and signature validation. Its [HTTP transport][signer-transport] owns connection and operation
+admission, shutdown, and separate `atto-signer-io` event loops, connection provider and `atto-signer-work` virtual-thread dispatcher.
+The whole signing operation runs on that dispatcher, including codecs, retry waits and signature validation.
+Peer server/client I/O uses separate `atto-peer-server-io` and `atto-peer-client-io` groups. Each transport closes only its owned
+resources. Virtual-thread executors have separate lifecycles but share the JVM's carrier scheduler. Local signing and signer
+metrics retain their existing interfaces. The node runtime does not depend on Ktor. Cucumber's `live` neighbour mode checks
+compatibility with the actual released node image, including connections initiated in either direction and reconnection.
 
 ### Outbound Connections
 
@@ -64,7 +104,7 @@ Consumers query this owner instead of reconstructing membership from connection
 
 ### Session Lifecycle
 
-`manage()` registers the connection and consumes binary frames from its incoming channel. It deserializes each binary
+`manage()` registers the connection and consumes complete binary messages from its peer session. It deserializes each binary
 message through [NetworkSerializer][serializer], which also checks protocol validity. Invalid messages remove the connection. Valid messages become `InboundNetworkMessage` events with the peer URI,
 socket address and `WEBSOCKET` source.
 
@@ -80,7 +120,10 @@ side effect; its connectivity check prevents a delayed disconnect from releasing
 Gap sessions identify peers by URI and retain their existing expiry; this is not connection-generation isolation.
 
 `clear()` removes connections and waits for already scheduled cleanup jobs. `stop()` clears the map and cancels the manager's
-scope. `NetworkProcessor.stop()` separately closes its client, cancels outbound work and stops its embedded server.
+scope. `NetworkProcessor.stop()` stops admission, cancels outbound work, closes the server and client transports, then clears
+managed connections. Closing the transports first releases stalled writes before manager cleanup waits for disconnect jobs.
+`PeerNettyClient` cancels active operations and closes sockets before disposing its connection providers, resolver work and
+event loops. Transport closure is idempotent. Neither shutdown path disposes Spring's HTTP resources.
 
 ### Sending and Keepalives
 
@@ -130,6 +173,12 @@ transport, feature, traffic-accounting or metrics listener to finish.
 | `NodeUnbanned` | `PeerBans` expiry review | No current consumers. |
 
 [processor]: src/main/kotlin/cash/atto/node/network/NetworkProcessor.kt
+[peer-server]: src/main/kotlin/cash/atto/node/network/PeerNettyServer.kt
+[peer-client]: src/main/kotlin/cash/atto/node/network/PeerNettyClient.kt
+[peer-session]: src/main/kotlin/cash/atto/node/network/PeerSession.kt
+[peer-reactor-session]: src/main/kotlin/cash/atto/node/network/ReactorPeerSession.kt
+[remote-signer]: src/main/kotlin/cash/atto/node/signature/NettyRemoteSigner.kt
+[signer-transport]: src/main/kotlin/cash/atto/node/signature/RemoteSignerHttpTransport.kt
 [connections]: src/main/kotlin/cash/atto/node/network/NodeConnectionManager.kt
 [connected-peers]: src/main/kotlin/cash/atto/node/network/ConnectedPeers.kt
 [guardian]: src/main/kotlin/cash/atto/node/network/guardian/Guardian.kt

README.md

@@ -4,8 +4,9 @@ Atto Node is the network-facing service for the **Atto** cryptocurrency. It expo
 submitting/querying blocks (transactions), maintains account state in a MySQL database, connects to other nodes over
 WebSockets, and participates in consensus by producing and relaying votes.
 
-This repository is a **Kotlin + Spring Boot (WebFlux)** application with a small embedded **Ktor WebSocket/HTTP server**
-for node-to-node communication.
+This repository is a **Kotlin + Spring Boot (WebFlux)** application with dedicated **Reactor Netty WebSocket/HTTP transports**
+for node-to-node communication and remote signing. Peer server, peer client and signer I/O resources are isolated from each
+other and from the public HTTP API. See [Network Layer](NETWORK.md) for transport ownership and authentication.
 
 [Website](https://atto.cash/) | [Docs](https://atto.cash/docs/integration) | [Commons](https://github.com/attocash/commons)
 
@@ -59,6 +60,16 @@ On startup, the node will:
 - start the REST API
 - start the node-to-node WebSocket server
 
+Run `./gradlew test` to test with neighbours built from the current checkout. For compatibility with the released node, run
+`./gradlew test -Datto.test.neighbour-mode=live`. This mode pulls `ghcr.io/attocash/node:live` once, reuses the resolved image
+throughout the run, and logs its digest and release metadata. It requires a local Linux container engine with host networking.
+Released neighbours use their production APIs and scheduled bootstrap processing; test-only controls remain on classpath nodes.
+
+CI runs the full suite through the Terraform-managed classpath workflow. The node-owned
+[compatibility workflow](.github/workflows/compatibility.yaml) runs Cucumber against `:live` for pull requests and is also
+called by the [pipeline](.github/workflows/pipeline.yaml). Builds require both checks; native-image metadata comes from the
+classpath run. Keep released-node checks in the node-owned workflow so Terraform can regenerate the shared workflows.
+
 ## Configuration
 
 Configuration is split across `application.yaml` and profile overrides such as:
@@ -115,7 +126,7 @@ From `application.yaml`:
 
 - **8080**: public REST API (Spring WebFlux)
 - **8081**: management/metrics (Actuator + Prometheus)
-- **8082**: node-to-node WebSocket server (Ktor)
+- **8082**: node-to-node WebSocket server (Reactor Netty)
 
 Swagger UI is configured at:
 

build.gradle.kts

@@ -56,12 +56,10 @@ dependencies {
     val commonsVersion = "8.0.0"
     val cucumberVersion = "7.34.8"
     val springdocVersion = "3.1.1"
-    val ktorVersion = "3.6.0"
 
     implementation("cash.atto:commons-core:$commonsVersion")
     implementation("cash.atto:commons-node:$commonsVersion")
     implementation("cash.atto:commons-worker:$commonsVersion")
-    implementation("cash.atto:commons-signer-remote:$commonsVersion")
     implementation("cash.atto:commons-spring-boot-starter:$commonsVersion")
 
     implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES))
@@ -76,16 +74,6 @@ dependencies {
     implementation("org.springframework.boot:spring-boot-starter-webflux")
     annotationProcessor("org.springframework.boot:spring-boot-configuration-processor")
 
-    implementation("io.ktor:ktor-client-cio:$ktorVersion")
-    implementation("io.ktor:ktor-server-cio:$ktorVersion")
-    implementation("io.ktor:ktor-serialization-kotlinx-json:$ktorVersion")
-
-    implementation("io.ktor:ktor-server-websockets:$ktorVersion")
-    implementation("io.ktor:ktor-server-content-negotiation:$ktorVersion")
-
-    implementation("io.ktor:ktor-client-websockets:$ktorVersion")
-    implementation("io.ktor:ktor-client-content-negotiation:$ktorVersion")
-
     implementation("org.springdoc:springdoc-openapi-starter-webflux-ui:$springdocVersion")
 
     implementation("com.github.ben-manes.caffeine:caffeine")
@@ -137,6 +125,15 @@ dependencies {
 tasks.withType<Test> {
     useJUnitPlatform()
     maxHeapSize = "4g"
+    val neighbourMode = providers.systemProperty("atto.test.neighbour-mode").orElse("classpath").get()
+    require(neighbourMode in setOf("classpath", "live")) { "Unsupported neighbour mode: $neighbourMode" }
+    systemProperty("atto.test.neighbour-mode", neighbourMode)
+    systemProperty(
+        "atto.test.runtime-classpath",
+        sourceSets.test
+            .get()
+            .runtimeClasspath.asPath,
+    )
 }
 
 tasks.named("check") {

src/main/kotlin/cash/atto/node/JvmAddressResolverGroup.kt

@@ -0,0 +1,61 @@
+package cash.atto.node
+
+import io.netty.resolver.AbstractAddressResolver
+import io.netty.resolver.AddressResolver
+import io.netty.resolver.AddressResolverGroup
+import io.netty.util.concurrent.EventExecutor
+import io.netty.util.concurrent.Promise
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.launch
+import java.net.InetAddress
+import java.net.InetSocketAddress
+
+/** Preserve JVM/system DNS semantics on the transport owner's workers, outside Netty I/O event loops. */
+internal class JvmAddressResolverGroup(
+    private val dispatcher: CoroutineDispatcher,
+) : AddressResolverGroup<InetSocketAddress>() {
+    override fun newResolver(executor: EventExecutor): AddressResolver<InetSocketAddress> =
+        object : AbstractAddressResolver<InetSocketAddress>(executor, InetSocketAddress::class.java) {
+            private val scope = CoroutineScope(SupervisorJob() + dispatcher)
+
+            override fun doIsResolved(address: InetSocketAddress): Boolean = !address.isUnresolved
+
+            override fun doResolve(
+                address: InetSocketAddress,
+                promise: Promise<InetSocketAddress>,
+            ) {
+                resolve(promise) { InetSocketAddress(InetAddress.getByName(address.hostString), address.port) }
+            }
+
+            override fun doResolveAll(
+                address: InetSocketAddress,
+                promise: Promise<MutableList<InetSocketAddress>>,
+            ) {
+                resolve(promise) {
+                    InetAddress.getAllByName(address.hostString).map { InetSocketAddress(it, address.port) }.toMutableList()
+                }
+            }
+
+            private fun <T> resolve(
+                promise: Promise<T>,
+                lookup: () -> T,
+            ) {
+                val job =
+                    scope.launch {
+                        try {
+                            promise.trySuccess(lookup())
+                        } catch (exception: Exception) {
+                            promise.tryFailure(exception)
+                        }
+                    }
+                promise.addListener { if (it.isCancelled) job.cancel() }
+            }
+
+            override fun close() {
+                scope.cancel()
+            }
+        }
+}

src/main/kotlin/cash/atto/node/NodeInfoContributor.kt

@@ -2,11 +2,14 @@ package cash.atto.node
 
 import cash.atto.protocol.AttoNode
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import org.springframework.aot.hint.annotation.RegisterReflectionForBinding
 import org.springframework.boot.actuate.info.Info
 import org.springframework.boot.actuate.info.InfoContributor
 import org.springframework.stereotype.Component
 
 @ExperimentalCoroutinesApi
+// Info's untyped details map hides this DTO from automatic response-type hints.
+@RegisterReflectionForBinding(AttoNode::class)
 @Component
 class NodeInfoContributor(
     val thisNode: AttoNode,