diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..8331aaa --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,105 @@ +name: Tests + +on: + push: + pull_request: + +permissions: + contents: read + +jobs: + boxlang: + name: BoxLang bleeding edge / Java 21 + runs-on: ubuntu-latest + timeout-minutes: 15 + + steps: + - name: Checkout + uses: actions/checkout@v7 + + - name: Set up Java 21 + uses: actions/setup-java@v5 + with: + distribution: temurin + java-version: "21" + + - name: Set up CommandBox + uses: Ortus-Solutions/setup-commandbox@v2.0.1 + with: + version: 6.3.1 + + - name: Install dependencies + run: box install + + - name: Show runtime versions + run: | + java -version + box version + + - name: Start BoxLang + run: box server start server-boxlang-be.json + + - name: Run tests with BoxLang + run: >- + curl --fail-with-body --silent --show-error + "http://127.0.0.1:8600/tests/runner.cfm?reporter=json&directory=tests.specs" + + - name: Verify test result + shell: bash + run: | + if ! grep -qx 'test.passed=true' tests/results/TEST.properties; then + echo '::error::TestBox reported failures or errors.' + exit 1 + fi + + - name: Stop BoxLang + if: always() + run: box server stop server-boxlang-be.json + + lucee: + name: Lucee 5.4 / Java 11 + runs-on: ubuntu-latest + timeout-minutes: 15 + + steps: + - name: Checkout + uses: actions/checkout@v7 + + - name: Set up Java 11 + uses: actions/setup-java@v5 + with: + distribution: temurin + java-version: "11" + + - name: Set up CommandBox + uses: Ortus-Solutions/setup-commandbox@v2.0.1 + with: + version: 6.3.1 + + - name: Install dependencies + run: box install + + - name: Show runtime versions + run: | + java -version + box version + + - name: Start Lucee + run: box server start server-lucee5.json + + - name: Run tests with Lucee + run: >- + curl --fail-with-body --silent --show-error + "http://127.0.0.1:8599/tests/runner.cfm?reporter=json&directory=tests.specs" + + - name: Verify test result + shell: bash + run: | + if ! grep -qx 'test.passed=true' tests/results/TEST.properties; then + echo '::error::TestBox reported failures or errors.' + exit 1 + fi + + - name: Stop Lucee + if: always() + run: box server stop server-lucee5.json diff --git a/.gitignore b/.gitignore index 111819d..bd6004c 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,3 @@ /models/cluster/cbproxies +/testbox/ +/tests/results/ diff --git a/README.md b/README.md index 749cf55..8371594 100644 --- a/README.md +++ b/README.md @@ -219,6 +219,14 @@ If `debugMode` is set to true, the entire config will be reloaded every request The `heartBeatMS` setting controls the number of ms between heartbeat "pings" from every client websocket connection. Set to `0` to disable this. Heartbeats are useful so clients can reconnect right away if the connection is interrupted, but can create a lot of network chatter if you have many connections. +#### Cluster lifecycle + +When clustering is enabled, SocketBox uses one shared JDK HTTP client per cluster manager for all outgoing peer WebSockets. The `cluster.peerConnectionTimeoutSeconds` setting applies to the actual WebSocket opening handshake. Failed handshakes are cancelled and retried using the cluster manager's adaptive delay; repeated failures do not reset the delay to its fastest interval. + +Call the inherited `shutdown()` method from your application's shutdown callback. This removes the server from the shared peer cache, cancels pending handshakes, and closes its peer WebSockets. Java 21 and newer explicitly shut down the shared HTTP client; on Java 11 through 20, reusing one client keeps the selector-thread count bounded and releasing it after the WebSockets close lets its selector be reclaimed. + +Cache-backed discovery stores a shared list of peer names plus a last-check-in entry for each peer. Expired peers are removed from the shared list as one verified batch so concurrent cleanup cannot cause in-process removal updates to overwrite one another. Cache providers with atomic collection and expiry operations may still provide stronger guarantees than SocketBox's portable `get()`, `set()`, and `clear()` contract. + #### Exchanges You can configure exchanges as shown above. The direct exchange will always be configured by default and you only need to specify it if you want to pass custom bindings to it. Exchanges receive the incoming messages and decide what destinations, if any, to route them to. An exchange could route a message to 0 destinations or 1000 destinations. And each destination could have 0 subscribers or 1000 subscribers. The exchange abstracts all that away from you. Publishers simply send messages to an exchange wihtout caring who is subscribed, and subscribers simply receive messages from their subscriptions without caring who sent it. @@ -297,4 +305,4 @@ You can create a simple STOMP connection like so: client.activate(); -``` \ No newline at end of file +``` diff --git a/box.json b/box.json index 6b9f337..a159ef2 100644 --- a/box.json +++ b/box.json @@ -3,6 +3,9 @@ "dependencies":{ "cbproxies":"^1.7.0+8" }, + "devDependencies":{ + "testbox":"^7.0.0+19" + }, "description":"WebSocket Listener library to be used with CommandBox Websocket and BoxLang WebSocket server.", "documentation":"https://forgebox.io/view/socketbox", "homepage":"https://forgebox.io/view/socketbox", @@ -12,7 +15,8 @@ "/tests/" ], "installPaths":{ - "cbproxies":"models/cluster/cbproxies/" + "cbproxies":"models/cluster/cbproxies/", + "testbox":"testbox/" }, "license":[ { @@ -29,7 +33,8 @@ }, "scripts":{ "onRelease":"publish", - "postPublish":"!git push --follow-tags" + "postPublish":"!git push --follow-tags", + "test":"!testbox/run" }, "shortDescription":"WebSocket Listener library", "slug":"socketbox", diff --git a/models/WebSocketCore.cfc b/models/WebSocketCore.cfc index 4c7fdf7..2f8a648 100644 --- a/models/WebSocketCore.cfc +++ b/models/WebSocketCore.cfc @@ -19,7 +19,7 @@ component { // should use the HTTP port that the server is listening on. Note, if this server is behind a proxy or load balancer, you need to provide // an INTERNAL address and/or port that the other servers in the cluster can connect to directly which doesn't flow through the proxy. // Defaults to the server's hostname and the HTTP port in use. - "name" : "ws://#createObject("java", "java.net.InetAddress").getLocalHost().getHostName()#:#cgi.server_port#/ws", + "name" : getDefaultClusterName(), // Hard-coded list of cluster peers to connect to. These are always used regardless of external cache. "peers" : [], // A class or object with MINIMUM get(), set(), and clear() methods to use as a cache provider. @@ -42,6 +42,21 @@ component { "defaultRPCTimeoutSeconds" : 15 } }; + + /** + * Build the default cluster address from the current web request. + * Non-web contexts use port 80 until an explicit configured name replaces + * this construction-time default. + */ + private function getDefaultClusterName() { + var serverPort = 80; + try { + serverPort = cgi.server_port; + } catch( any ignored ) { + // Non-web contexts have no CGI scope. + } + return "ws://#createObject("java", "java.net.InetAddress").getLocalHost().getHostName()#:#serverPort#/ws"; + } /** * Front controller for all WebSocket incoming messages @@ -155,9 +170,26 @@ component { } application.SocketBoxConfig = local.config; + // A reconfiguration must explicitly stop the previous manager before + // replacing it. Its peer connections belong to its shared HTTP client + // and cannot be transferred safely to a new manager. + var oldManager = ''; + if( + application.keyExists( 'socketBoxClusterManagement' ) && + application.socketBoxClusterManagement.keyExists( 'clusterManager' ) + ) { + oldManager = application.socketBoxClusterManagement.clusterManager; + } + if( !isSimpleValue( oldManager ) ) { + try { + oldManager.shutdown(); + } finally { + application.delete( 'socketBoxClusterManagement' ); + } + } + // Setup the cluster if enabled if( local.config.cluster.enable ) { - var oldManager = application.socketBoxClusterManagement.clusterManager ?: ''; application.socketBoxClusterManagement = { "clusterManager" : new cluster.ClusterManager( this, config ), // Incoming connections from regular clients @@ -168,12 +200,6 @@ component { // cluster name is configured correctly. "selfChannels" : {} }; - - // Don't let existing connections leak - if( !isSimpleValue( oldManager ) ) { - application.socketBoxClusterManagement.clusterManager.setPeerConnections( oldManager.getPeerConnections() ); - } - // Start the cluster manager once the config is fully set up application.socketBoxClusterManagement.clusterManager.start() } @@ -741,4 +767,4 @@ component { return target; } -} \ No newline at end of file +} diff --git a/models/cluster/ClusterManager.cfc b/models/cluster/ClusterManager.cfc index 57df4c6..4fb5251 100644 --- a/models/cluster/ClusterManager.cfc +++ b/models/cluster/ClusterManager.cfc @@ -76,13 +76,21 @@ component accessors="true" { */ property name="config" type="struct"; + /** + * Shared HTTP client for all outgoing cluster WebSocket connections. + * Creating a client per connection creates a SelectorManager thread per attempt. + */ + property name="httpClient" type="any"; + /** * Constructor * @socketBox The SocketBox instance to use for cluster management * @config The configuration for the SocketBox instance. This is mostly to avoid circular references. during startup. + * @httpClient Optional HTTP client override, primarily for testing + * @peerListenerFactory Optional WebSocket listener factory, primarily for testing * @return The ClusterManager instance */ - function init( required any socketBox, required struct config ) { + function init( required any socketBox, required struct config, any httpClient, any peerListenerFactory ) { variables.startTick = getTickCount(); variables.config = config; variables.clusterManagerKey = createUUID(); @@ -93,6 +101,17 @@ component accessors="true" { variables.jThread = createObject( "java", "java.lang.Thread" ); variables.jSystem = createObject( "java", "java.lang.System" ); variables.jFuture = createObject( "java", "java.util.concurrent.CompletableFuture" ); + variables.httpClient = !isNull( arguments.httpClient ) + ? arguments.httpClient + : createObject( "java", "java.net.http.HttpClient" ).newHttpClient(); + variables.httpClientShutdown = false; + variables.shutdownRequested = createObject( "java", "java.util.concurrent.atomic.AtomicBoolean" ).init( false ); + variables.resourceShutdownStarted = createObject( "java", "java.util.concurrent.atomic.AtomicBoolean" ).init( false ); + variables.pendingConnectionFutures = createObject( "java", "java.util.concurrent.ConcurrentHashMap" ).init(); + variables.peerListenerFactoryExists = arguments.keyExists( "peerListenerFactory" ) && !isNull( arguments.peerListenerFactory ); + if( variables.peerListenerFactoryExists ) { + variables.peerListenerFactory = arguments.peerListenerFactory; + } if( !isSimpleValue( config.cluster.cacheProvider ) ) { variables.cacheProviderExists = true; variables.cacheProvider = config.cluster.cacheProvider; @@ -147,7 +166,7 @@ component accessors="true" { } } socketBox.logMessage( "SocketBox cluster manager thread stopped." ); - shutdownPeerConnections(); + shutdownResources(); } } @@ -177,13 +196,20 @@ component accessors="true" { if( !hasCacheProvider() ) { return; } - getCachePeers().each( (cachePeer)=>{ + var expiredPeers = getCachePeers().filter( (cachePeer)=>{ if( isPeerExpired( cachePeer ) ) { - // This peer has timed out socketBox.logMessage( "SocketBox Removing expired peer from cache: " & cachePeer ); - removePeerFromCache( cachePeer, 3 ); + return true; } - }, true ); + return false; + } ); + + // Remove all expired peers with a single read/modify/write cycle. Running one + // non-atomic update per peer in parallel allows those updates to overwrite + // one another and reintroduce stale peers. + if( expiredPeers.len() ) { + removePeersFromCache( expiredPeers, 3 ); + } } /** @@ -323,7 +349,10 @@ component accessors="true" { var i = 1; while ( i <= arguments.attempts ) { cachedPeers = getCachePeersRaw(); - if ( cachedPeers contains myPeerName ) { + var cachedPeersArray = cachedPeers + .listToArray( chr(13) & chr(10) ) + .map( ( peer )=>trim( peer ) ); + if ( cachedPeersArray.contains( myPeerName ) ) { if( i > 1 ) socketBox.logMessage("SocketBox success adding self to cache!"); return; } @@ -349,29 +378,67 @@ component accessors="true" { * */ function removePeerFromCache( required string peerName, numeric attempts=3 ) { - var cacheKey = cacheKeyPrefix & "-" & peerName; - variables.cacheProvider.clear( cacheKey ); + return removePeersFromCache( [ peerName ], arguments.attempts ); + } + + /** + * Remove one or more peers from the cache provider in a single update. + * The generic cache API has no compare-and-set operation, so verify each write + * and retry if another cluster node changed the shared peer list concurrently. + * + * @peerNames The peer names to remove + * @attempts The number of attempts before giving up + * @return true when all requested peers are absent from the shared list + */ + function removePeersFromCache( required array peerNames, numeric attempts=3 ) { + var peersToRemove = {}; + arguments.peerNames.each( ( peerName )=>{ + if( len( trim( peerName ) ) ) { + peersToRemove[ trim( peerName ) ] = true; + variables.cacheProvider.clear( cacheKeyPrefix & "-" & trim( peerName ) ); + } + } ); + + if( !peersToRemove.len() ) { + return true; + } - var cachedPeers = ""; var i = 1; while ( i <= arguments.attempts ) { - cachedPeers = getCachePeersRaw(); - if ( !(cachedPeers contains peerName) ) { - if( i > 1 ) socketBox.logMessage("SocketBox Success removing peer [#peerName#] from cache!"); - return; + var cachedPeers = getCachePeers(); + var remainingPeers = cachedPeers.filter( ( peerName )=>!peersToRemove.keyExists( peerName ) ); + + if( remainingPeers.len() == cachedPeers.len() ) { + if( i > 1 ) socketBox.logMessage( "SocketBox successfully removed expired peers from cache." ); + return true; } - // Remove the peer - var peersArray = cachedPeers.listToArray( chr(13) & chr(10) ).filter( (peer) => trim(peer) != peerName ); - cachedPeers = peersArray.toList( chr(13) & chr(10) ); - socketBox.logMessage("SocketBox removed peer [#peerName#] from cache attempt #i#." ); - variables.cacheProvider.set( cacheKeyPrefix, cachedPeers ); - // Pause 1-3 seconds to allow for other in-process writes to complete - sleep( randRange( 1000, 3000 ) ); + socketBox.logMessage( "SocketBox removed [#cachedPeers.len() - remainingPeers.len()#] peer(s) from cache attempt #i#." ); + variables.cacheProvider.set( cacheKeyPrefix, remainingPeers.toList( chr(13) & chr(10) ) ); + + if( !cacheContainsAnyPeers( peersToRemove ) ) { + return true; + } + + // Briefly stagger retries made by multiple cluster nodes. + sleep( randRange( 25, 100 ) ); - // Double check our work in the next loop iteration i++; } + + return !cacheContainsAnyPeers( peersToRemove ); + } + + /** + * Check whether the shared cache list contains any key in the supplied struct. + */ + private boolean function cacheContainsAnyPeers( required struct peers ) { + for( var peerName in getCachePeers() ) { + if( arguments.peers.keyExists( peerName ) ) { + return true; + } + } + return false; } /** @@ -380,7 +447,7 @@ component accessors="true" { * @param peerName The name of the peer to connect to */ function ensurePeer( required string peerName ) { - if( peerName == myPeerName ) { + if( peerName == myPeerName || variables.shutdownRequested.get() ) { return; // Don't connect to ourselves } if( !variables.peerConnections.keyExists( peerName ) ) { @@ -400,15 +467,24 @@ component accessors="true" { * @param peerName The name of the peer to connect to */ function addPeer( required string peerName ) { + if( variables.shutdownRequested.get() ) { + return; + } socketBox.logMessage("SocketBox Connecting to cluster peer: " & peerName); + var connectionFuture; + var connectionFutureCreated = false; + var connectionAttemptKey = createUUID(); try { - var httpClient = createObject("java", "java.net.http.HttpClient").newHttpClient(); var javaURI = createObject("java", "java.net.URI").create( peerName ); - var peer = new ClusterPeer( socketbox, this, peerName ); - - var timeUnit = createObject("java", "java.util.concurrent.TimeUnit"); - httpClient.newWebSocketBuilder() + var connectionTimeoutMS = max( 1, int( config.cluster.peerConnectionTimeoutSeconds * 1000 ) ); + var duration = createObject( "java", "java.time.Duration" ); + + connectionFuture = variables.httpClient.newWebSocketBuilder() + // Apply the timeout to the actual WebSocket handshake. A timeout on + // CompletableFuture.get() alone only stops waiting and leaves the + // connection attempt running in the background. + .connectTimeout( duration.ofMillis( javaCast( "long", connectionTimeoutMS ) ) ) // Authorization header .header( "socketbox-management", @@ -421,31 +497,42 @@ component accessors="true" { ) .buildAsync( javaURI, - createDynamicProxy( - peer, - [ "java.net.http.WebSocket$Listener" ] - ) - ) - .get(config.cluster.peerConnectionTimeoutSeconds, timeUnit.SECONDS) + createPeerListener( peer ) + ); + connectionFutureCreated = true; + variables.pendingConnectionFutures.put( connectionAttemptKey, connectionFuture ); + + if( variables.shutdownRequested.get() ) { + connectionFuture.cancel( true ); + return; + } + + // The builder timeout completes the future exceptionally, so this wait + // cannot outlive the configured handshake timeout. + connectionFuture.get(); } catch( any e ) { - // If there are issues connecting to peers, then log this as the cluster updating to keep our manager thread running quickly until things settle down - clusterUpdated(); + // Ensure a caller-side timeout or interruption cannot leave a pending + // handshake retaining the HTTP client and its SelectorManager thread. + if( connectionFutureCreated && !connectionFuture.isDone() ) { + connectionFuture.cancel( true ); + } // Special message for a few specific cases + var errorDetail = lCase( "#e.type# #e.message# #e.stackTrace#" ); // generic timeout - if( e.type contains "timeout") { + if( errorDetail contains "timeout" ) { socketBox.logMessage("SocketBox Timeout connecting to cluster peer [#peerName#]"); return; } // unresovled hostname - if( e.stacktrace contains "UnresolvedAddressException" ) { + if( errorDetail contains "unresolvedaddressexception" ) { socketBox.logMessage("SocketBox Cannot resolve host for cluster peer [#peerName#]"); return; } - if( e.stacktrace contains "ConnectException" ) { + if( errorDetail contains "connectexception" ) { socketBox.logMessage("SocketBox Connection error connecting to cluster peer [#peerName#]"); return; } @@ -454,9 +541,17 @@ component accessors="true" { socketBox.logMessage("SocketBox Error connecting to cluster peer [#peerName#]: " & e.message); //println( e.stackTrace ) return; + } finally { + if( connectionFutureCreated ) { + variables.pendingConnectionFutures.remove( connectionAttemptKey ); + } } peerConnections[peerName] = peer; + if( variables.shutdownRequested.get() ) { + removePeerConnection( peerName, true, true ); + return; + } // println("SocketBox connected to cluster node: " & peerName); clusterUpdated(); @@ -466,15 +561,16 @@ component accessors="true" { * Remove a dead peer connection * @param peerName The name of the peer to remove * @param close Whether to close the connection or not + * @param force Whether to abort the WebSocket after initiating its close */ - function removePeerConnection( required string peerName, boolean close=true ) { + function removePeerConnection( required string peerName, boolean close=true, boolean force=false ) { var existed = variables.peerConnections.keyExists( peerName ); if( close ) { // May not exist. Handle without locking var peer = peerConnections[peerName] ?: ""; if( !isSimpleValue( peer ) ) { // If already closed, should not error per spec - peer.close(); + peer.close( arguments.force ); } } // Won't fail if not exists @@ -485,6 +581,19 @@ component accessors="true" { } } + /** + * Create the JDK WebSocket listener for a cluster peer. + */ + private function createPeerListener( required any peer ) { + if( variables.peerListenerFactoryExists ) { + return variables.peerListenerFactory( arguments.peer ); + } + return createDynamicProxy( + arguments.peer, + [ "java.net.http.WebSocket$Listener" ] + ); + } + /** * Do we have a cache provider configured? */ @@ -497,34 +606,93 @@ component accessors="true" { */ function shutdown() { // This will signal to the manager thread to stop + variables.shutdownRequested.set( true ); variables.clusterManagerKey = ""; - // If we're in cluster mode, remove ourselves from the cache - if( hasCacheProvider() ) { - socketBox.logMessage("SocketBox - Removing myself from cache..."); - // Don't try too hard here. Another node will eventually flush this. - // I've found that while testing I often times have a node start right up after shutting down and it's - // already tryingn to add itself while another node is still trying to remove it. - // No use fighting over it. - removePeerFromCache( myPeerName, 2 ); + try { + // If we're in cluster mode, remove ourselves from the cache + if( hasCacheProvider() ) { + socketBox.logMessage("SocketBox - Removing myself from cache..."); + // Don't try too hard here. Another node will eventually flush this. + // A node can start right after shutdown and add itself while this node + // is still trying to remove its old entry, so avoid fighting indefinitely. + removePeerFromCache( myPeerName, 2 ); + } + } finally { + // Cache failures must never prevent WebSocket and HTTP client cleanup. + shutdownResources(); } - - shutdownPeerConnections(); - } /** * Shutdown all peer connections + * @force Whether to abort each WebSocket after initiating its close */ - function shutdownPeerConnections() { + function shutdownPeerConnections( boolean force=false ) { socketBox.logMessage("SocketBox Shutting down [#peerConnections.count()#] management connections"); + var forceClose = arguments.force; peerConnections.each( (peerName,clusterPeer)=>{ try { - clusterPeer.close() + clusterPeer.close( forceClose ) } catch( any e ) { socketBox.logMessage("SocketBox Error closing management connection [#peerName#]: " & e.message); } } ); + peerConnections.clear(); + } + + /** + * Close cluster connections and release the shared HTTP client. + */ + private function shutdownResources() { + if( !variables.resourceShutdownStarted.compareAndSet( false, true ) ) { + return; + } + variables.shutdownRequested.set( true ); + cancelPendingConnections(); + shutdownPeerConnections( true ); + shutdownHttpClient(); + } + + /** + * Cancel handshakes that may still retain the shared HTTP client. + */ + private function cancelPendingConnections() { + for( var connectionFuture in variables.pendingConnectionFutures.values() ) { + try { + if( !connectionFuture.isDone() ) { + connectionFuture.cancel( true ); + } + } catch( any e ) { + socketBox.logMessage( "SocketBox Error cancelling pending peer connection: " & e.message ); + } + } + variables.pendingConnectionFutures.clear(); + } + + /** + * Java 21 added explicit HttpClient shutdown methods. On older supported JVMs, + * sharing a single client still bounds the SelectorManager count and closing + * all WebSockets allows the client to be reclaimed with this manager. + */ + private function shutdownHttpClient() { + if( variables.httpClientShutdown || isNull( variables.httpClient ) ) { + return; + } + variables.httpClientShutdown = true; + + if( val( variables.jSystem.getProperty( "java.specification.version" ) ) >= 21 ) { + try { + variables.httpClient.shutdownNow(); + } catch( any e ) { + socketBox.logMessage( "SocketBox Error shutting down HTTP client: " & e.message ); + } + } + + // Java 11-20 have no public HttpClient shutdown API. Once all WebSockets + // are closed, dropping this manager's last client reference allows the + // JDK's weak-reference lifecycle to terminate its SelectorManager. + variables.delete( "httpClient" ); } /** @@ -689,4 +857,4 @@ component accessors="true" { writedump( var=message.toString(), output="console" ); } -} \ No newline at end of file +} diff --git a/models/cluster/ClusterPeer.cfc b/models/cluster/ClusterPeer.cfc index ac21353..51ba1e6 100644 --- a/models/cluster/ClusterPeer.cfc +++ b/models/cluster/ClusterPeer.cfc @@ -148,18 +148,32 @@ component extends="cbproxies.models.BaseProxy" accessors="true" { /** * Close the peer connection + * @force Whether to abort the WebSocket after initiating its close * @return A CompletableFuture indicating the completion of the close operation */ - public function close() { + public function close( boolean force=false ) { if( isNull( variables.webSocket ) ) { return; } + var forceClose = arguments.force; lock name="websocket_#getWebsocketHash()#" timeout=60 type="exclusive" { - var future = variables.webSocket.sendClose(1000, "SocketBox Peer [#clusterManager.getMyPeerName()#] shutting down"); - future.get(); - variables.delete( "webSocket" ); + var webSocket = variables.webSocket; + try { + var future = webSocket.sendClose(1000, "SocketBox Peer [#clusterManager.getMyPeerName()#] shutting down"); + if( !forceClose ) { + future.get(); + } + } finally { + try { + if( forceClose ) { + webSocket.abort(); + } + } finally { + variables.delete( "webSocket" ); + } + } } - } + } /** * Check if the peer connection is open @@ -176,4 +190,4 @@ component extends="cbproxies.models.BaseProxy" accessors="true" { } } -} \ No newline at end of file +} diff --git a/server-boxlang-be.json b/server-boxlang-be.json new file mode 100644 index 0000000..7cf8ab4 --- /dev/null +++ b/server-boxlang-be.json @@ -0,0 +1,20 @@ +{ + "name": "socketbox-boxlang-be", + "openBrowser": false, + "profile": "development", + "trayEnable": false, + "app": { + "cfengine": "boxlang@be" + }, + "jvm": { + "javaVersion": "openjdk21_jdk" + }, + "web": { + "host": "127.0.0.1", + "bindings": { + "HTTP": { + "listen": "8600" + } + } + } +} diff --git a/server-lucee5.json b/server-lucee5.json new file mode 100644 index 0000000..a06032b --- /dev/null +++ b/server-lucee5.json @@ -0,0 +1,20 @@ +{ + "name": "socketbox-lucee5", + "openBrowser": false, + "profile": "development", + "trayEnable": false, + "app": { + "cfengine": "lucee@5.4.8+2" + }, + "jvm": { + "javaVersion": "openjdk11" + }, + "web": { + "host": "127.0.0.1", + "bindings": { + "HTTP": { + "listen": "8599" + } + } + } +} diff --git a/tests/Application.cfc b/tests/Application.cfc new file mode 100644 index 0000000..3a0d99c --- /dev/null +++ b/tests/Application.cfc @@ -0,0 +1,12 @@ +component { + this.name = "SocketBoxRegressionTests"; + this.applicationTimeout = createTimeSpan( 0, 0, 5, 0 ); + + variables.testsPath = getDirectoryFromPath( getCurrentTemplatePath() ); + variables.rootPath = getDirectoryFromPath( variables.testsPath ); + + this.mappings[ "/models" ] = variables.rootPath & "models"; + this.mappings[ "/socketbox" ] = variables.rootPath; + this.mappings[ "/testbox" ] = variables.rootPath & "testbox"; + this.mappings[ "/tests" ] = variables.testsPath; +} diff --git a/tests/resources/ApplicationScopedBase.cfc b/tests/resources/ApplicationScopedBase.cfc new file mode 100644 index 0000000..20c419f --- /dev/null +++ b/tests/resources/ApplicationScopedBase.cfc @@ -0,0 +1,7 @@ + + + + diff --git a/tests/resources/ConfigurableWebSocketCore.cfc b/tests/resources/ConfigurableWebSocketCore.cfc new file mode 100644 index 0000000..d6c1418 --- /dev/null +++ b/tests/resources/ConfigurableWebSocketCore.cfc @@ -0,0 +1,23 @@ +/** + * Test fixture that makes WebSocketCore configuration deterministic and quiet. + */ +component extends="models.WebSocketCore" { + + variables.testConfig = {}; + + function setTestConfig( required struct config ) { + variables.testConfig = arguments.config; + return this; + } + + function configure() { + return variables.testConfig; + } + + function logMessage( required any message ) { + } + + function println( required message ) { + } + +} diff --git a/tests/runner.cfm b/tests/runner.cfm new file mode 100644 index 0000000..e0e3d61 --- /dev/null +++ b/tests/runner.cfm @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + + + + + + + diff --git a/tests/specs/ClusterManagerTest.cfc b/tests/specs/ClusterManagerTest.cfc new file mode 100644 index 0000000..d774688 --- /dev/null +++ b/tests/specs/ClusterManagerTest.cfc @@ -0,0 +1,579 @@ +/** + * Regression tests for SocketBox cluster resource management. + * Run with either checked-in CommandBox server config and request /tests/runner.cfm. + */ +component extends="testbox.system.BaseSpec" { + +function selectorThreadCount() { + var count = 0; + var threadClass = createObject( "java", "java.lang.Thread" ); + for( var threadEntry in threadClass.getAllStackTraces().entrySet() ) { + var thread = threadEntry.getKey(); + if( thread.isAlive() && thread.getName().matches( "HttpClient-\d+-SelectorManager" ) ) { + count++; + } + } + return count; +} + +function waitForSelectorCount( required numeric expected, numeric timeoutMS=2000 ) { + var endTick = getTickCount() + arguments.timeoutMS; + do { + if( selectorThreadCount() == arguments.expected ) { + return true; + } + sleep( 25 ); + } while( getTickCount() < endTick ); + return selectorThreadCount() == arguments.expected; +} + +function waitForSelectorCountAfterGC( required numeric expected, numeric timeoutMS=5000 ) { + var jSystem = createObject( "java", "java.lang.System" ); + var endTick = getTickCount() + arguments.timeoutMS; + do { + jSystem.gc(); + jSystem.runFinalization(); + if( selectorThreadCount() == arguments.expected ) { + return true; + } + sleep( 50 ); + } while( getTickCount() < endTick ); + return selectorThreadCount() == arguments.expected; +} + +function newClusterManager( required any socketBox, required struct config, any httpClient, any peerListenerFactory ) { + var manager = createObject( "component", "models.cluster.ClusterManager" ); + + if( arguments.keyExists( "peerListenerFactory" ) && !isNull( arguments.peerListenerFactory ) ) { + return manager.init( + arguments.socketBox, + arguments.config, + arguments.httpClient, + arguments.peerListenerFactory + ); + } + if( arguments.keyExists( "httpClient" ) && !isNull( arguments.httpClient ) ) { + return manager.init( arguments.socketBox, arguments.config, arguments.httpClient ); + } + return manager.init( arguments.socketBox, arguments.config ); +} + +function newSocketBoxMock() { + var mock = { messages : [] }; + mock.logMessage = ( message )=>mock.messages.append( message ); + mock.getSocketBoxKey = ()=>"socketbox-tests"; + return mock; +} + +function newConfig( any cacheProvider="", string name="ws://current:8080/ws" ) { + return { + cluster : { + cachePrefix : "test-", + cacheProvider : arguments.cacheProvider, + defaultRPCTimeoutSeconds : 1, + name : arguments.name, + peerConnectionTimeoutSeconds : 0.01, + peerIdleTimeoutSeconds : 60, + peers : [], + secretKey : "test-secret" + } + }; +} + +function testSharedHttpClientLifecycle() { + var javaVersion = val( createObject( "java", "java.lang.System" ).getProperty( "java.specification.version" ) ); + var initialSelectors = selectorThreadCount(); + var manager = newClusterManager( newSocketBoxMock(), newConfig() ); + $assert.isTrue( + waitForSelectorCount( initialSelectors + 1 ), + "A cluster manager should create exactly one shared HttpClient selector thread." + ); + + manager.shutdown(); + $assert.isTrue( + javaVersion >= 21 + ? waitForSelectorCount( initialSelectors ) + : waitForSelectorCountAfterGC( initialSelectors ), + "Cluster manager shutdown should terminate its shared HttpClient selector thread." + ); +} + +function testConnectionTimeoutCancellation() { + var pendingFuture = { cancelled : false }; + pendingFuture.get = ()=>{ + throw( type="java.util.concurrent.TimeoutException", message="test timeout" ); + }; + pendingFuture.isDone = ()=>false; + pendingFuture.cancel = ( mayInterruptIfRunning )=>{ + pendingFuture.cancelled = true; + return true; + }; + + var webSocketBuilder = { connectTimeoutMS : 0 }; + webSocketBuilder.connectTimeout = ( duration )=>{ + webSocketBuilder.connectTimeoutMS = duration.toMillis(); + return webSocketBuilder; + }; + webSocketBuilder.header = ( name, value )=>webSocketBuilder; + webSocketBuilder.buildAsync = ( uri, listener )=>pendingFuture; + + var httpClient = { shutdownCalled : false }; + httpClient.newWebSocketBuilder = ()=>webSocketBuilder; + httpClient.shutdownNow = ()=>{ + httpClient.shutdownCalled = true; + }; + + var socketBox = newSocketBoxMock(); + var manager = newClusterManager( + socketBox, + newConfig(), + httpClient, + ( peer )=>peer + ); + manager.addPeer( "ws://offline:8080/ws" ); + + $assert.isEqual( 10, webSocketBuilder.connectTimeoutMS, "The timeout must be applied to the WebSocket handshake." ); + $assert.isTrue( + pendingFuture.cancelled, + "An unfinished connection future must be cancelled after an error. Logs: #serializeJSON( socketBox.messages )#" + ); + $assert.isEqual( 0, manager.getPeerConnections().len(), "A failed peer must not be retained." ); + + var connectedPeer = { forceClosed : false }; + connectedPeer.close = ( force )=>{ + connectedPeer.forceClosed = force; + }; + manager.setPeerConnections( { "connected-peer" : connectedPeer } ); + manager.shutdown(); + $assert.isTrue( connectedPeer.forceClosed, "Manager shutdown must force-close established peer WebSockets." ); + $assert.isEqual( 0, manager.getPeerConnections().len(), "Manager shutdown must clear established peers." ); + var javaVersion = val( createObject( "java", "java.lang.System" ).getProperty( "java.specification.version" ) ); + $assert.isEqual( + javaVersion >= 21, + httpClient.shutdownCalled, + "Explicit HTTP client shutdown should only be used when the JVM provides it." + ); +} + +function testShutdownCancelsPendingHandshake() { + var manager = ""; + var pendingFuture = { cancelled : false }; + pendingFuture.isDone = ()=>pendingFuture.cancelled; + pendingFuture.cancel = ( mayInterruptIfRunning )=>{ + pendingFuture.cancelled = true; + return true; + }; + pendingFuture.get = ()=>{ + manager.shutdown(); + throw( type="java.util.concurrent.CancellationException", message="cancelled by shutdown" ); + }; + + var webSocketBuilder = {}; + webSocketBuilder.connectTimeout = ( duration )=>webSocketBuilder; + webSocketBuilder.header = ( name, value )=>webSocketBuilder; + webSocketBuilder.buildAsync = ( uri, listener )=>pendingFuture; + + var httpClient = {}; + httpClient.newWebSocketBuilder = ()=>webSocketBuilder; + httpClient.shutdownNow = ()=>{}; + + manager = newClusterManager( + newSocketBoxMock(), + newConfig(), + httpClient, + ( peer )=>peer + ); + manager.addPeer( "ws://pending:8080/ws" ); + + $assert.isTrue( pendingFuture.cancelled, "Manager shutdown must cancel an in-flight peer handshake." ); + $assert.isEqual( 0, manager.getPeerConnections().len(), "A handshake cancelled by shutdown must not add a peer." ); +} + +function testCacheFailureCannotSkipResourceCleanup() { + var cacheProvider = {}; + cacheProvider.get = ( key )=>""; + cacheProvider.set = ( key, value )=>{}; + cacheProvider.clear = ( key )=>{ + throw( type="CacheFailure", message="simulated cache outage" ); + }; + + var httpClient = {}; + httpClient.newWebSocketBuilder = ()=>{}; + httpClient.shutdownNow = ()=>{}; + var connectedPeer = { forceClosed : false }; + connectedPeer.close = ( force )=>{ + connectedPeer.forceClosed = force; + }; + + var manager = newClusterManager( + newSocketBoxMock(), + newConfig( cacheProvider ), + httpClient + ); + manager.setPeerConnections( { "connected-peer" : connectedPeer } ); + var cacheFailureRaised = false; + try { + manager.shutdown(); + } catch( CacheFailure e ) { + cacheFailureRaised = true; + } + + $assert.isTrue( cacheFailureRaised, "The simulated cache failure should reach the caller." ); + $assert.isTrue( connectedPeer.forceClosed, "A cache failure must not skip peer WebSocket cleanup." ); + $assert.isEqual( 0, manager.getPeerConnections().len(), "A cache failure must not leave retained peers." ); +} + +function testForceCloseAbortsWebSocket() { + var clusterManager = {}; + clusterManager.getMyPeerName = ()=>"ws://current:8080/ws"; + var clusterPeer = createObject( "component", "models.cluster.ClusterPeer" ) + .init( newSocketBoxMock(), clusterManager, "ws://peer:8080/ws" ); + var closeFuture = {}; + closeFuture.get = ()=>{ + throw( type="AssertionFailed", message="A forced close must not wait for the graceful close future." ); + }; + var webSocket = { closeSent : false, aborted : false }; + webSocket.sendClose = ( statusCode, reason )=>{ + webSocket.closeSent = true; + return closeFuture; + }; + webSocket.abort = ()=>{ + webSocket.aborted = true; + }; + clusterPeer.setWebSocket( webSocket ); + + clusterPeer.close( true ); + + $assert.isTrue( webSocket.closeSent, "A forced peer close should still initiate a close frame." ); + $assert.isTrue( webSocket.aborted, "A forced peer close must abort the WebSocket." ); +} + +function testExpiredPeersAreRemovedInOneBatch() { + var currentPeer = "ws://current:8080/ws"; + var expiredPeerOne = "ws://expired-one:8080/ws"; + var expiredPeerTwo = "ws://expired-two:8080/ws"; + var cacheKeyPrefix = "test-socketbox-cluster-peers"; + var nowEpoch = int( createObject( "java", "java.lang.System" ).currentTimeMillis() / 1000 ); + var cacheData = { + "#cacheKeyPrefix#" : [ expiredPeerOne, expiredPeerTwo, currentPeer ].toList( chr(13) & chr(10) ), + "#cacheKeyPrefix#-#expiredPeerOne#" : nowEpoch - 120, + "#cacheKeyPrefix#-#expiredPeerTwo#" : nowEpoch - 120, + "#cacheKeyPrefix#-#currentPeer#" : nowEpoch + }; + var peerListWrites = 0; + var cacheProvider = {}; + cacheProvider.get = ( key )=>cacheData.keyExists( key ) ? cacheData[ key ] : nullValue(); + cacheProvider.set = ( key, value )=>{ + cacheData[ key ] = value; + if( key == cacheKeyPrefix ) { + peerListWrites++; + } + }; + cacheProvider.clear = ( key )=>{ + var existed = cacheData.keyExists( key ); + cacheData.delete( key ); + return existed; + }; + + var httpClient = { shutdownNow : ()=>{} }; + httpClient.newWebSocketBuilder = ()=>{}; + var manager = newClusterManager( + newSocketBoxMock(), + newConfig( cacheProvider ), + httpClient + ); + + manager.reapExpiredCachePeers(); + var remainingPeers = manager.getCachePeers(); + + $assert.isEqual( 1, peerListWrites, "Multiple expired peers should be removed with one peer-list write." ); + $assert.isEqual( 1, remainingPeers.len(), "Only the current peer should remain in the cache list." ); + $assert.isEqual( currentPeer, remainingPeers[ 1 ], "The active peer must be preserved." ); + $assert.isTrue( !cacheData.keyExists( "#cacheKeyPrefix#-#expiredPeerOne#" ), "The first expired heartbeat should be cleared." ); + $assert.isTrue( !cacheData.keyExists( "#cacheKeyPrefix#-#expiredPeerTwo#" ), "The second expired heartbeat should be cleared." ); + + manager.shutdown(); +} + +function testSuccessfulConnectionsReuseSharedHttpClient() { + var connectionAttempts = []; + var httpClient = { builderCount : 0, shutdownCount : 0 }; + httpClient.newWebSocketBuilder = ()=>{ + httpClient.builderCount++; + var attempt = { + headers : {}, + timeoutMS : 0, + cancelCount : 0, + webSocket : { + closeCount : 0, + abortCount : 0 + } + }; + attempt.webSocket.sendClose = ( statusCode, reason )=>{ + attempt.webSocket.closeCount++; + return { get : ()=>{} }; + }; + attempt.webSocket.abort = ()=>attempt.webSocket.abortCount++; + + var builder = {}; + builder.connectTimeout = ( duration )=>{ + attempt.timeoutMS = duration.toMillis(); + return builder; + }; + builder.header = ( name, value )=>{ + attempt.headers[ name ] = value; + return builder; + }; + builder.buildAsync = ( uri, listener )=>{ + attempt.uri = uri.toString(); + attempt.listener = listener; + listener.setWebSocket( attempt.webSocket ); + + var future = {}; + future.get = ()=>attempt.webSocket; + future.isDone = ()=>true; + future.cancel = ( mayInterruptIfRunning )=>{ + attempt.cancelCount++; + return true; + }; + connectionAttempts.append( attempt ); + return future; + }; + return builder; + }; + httpClient.shutdownNow = ()=>httpClient.shutdownCount++; + + var manager = newClusterManager( + newSocketBoxMock(), + newConfig(), + httpClient, + ( peer )=>peer + ); + manager.setDelaySeconds( 60 ); + + manager.addPeer( "ws://peer-one:8080/ws" ); + manager.addPeer( "ws://peer-two:8080/ws" ); + + $assert.isEqual( 2, httpClient.builderCount, "Both connections must use the manager's shared HTTP client." ); + $assert.isEqual( 2, manager.getPeerConnections().len(), "Successful peer connections must be retained." ); + $assert.isEqual( 2, manager.getDelaySeconds(), "A successful connection should reset the adaptive delay." ); + $assert.isEqual( 2, connectionAttempts.len(), "Both successful handshakes should be recorded." ); + for( var attempt in connectionAttempts ) { + $assert.isEqual( 10, attempt.timeoutMS, "Each handshake must receive the configured timeout." ); + $assert.isEqual( "test-secret", attempt.headers[ "socketbox-management" ], "Each handshake must include the cluster secret." ); + $assert.isEqual( "ws://current:8080/ws", attempt.headers[ "socketbox-management-name" ], "Each handshake must identify the current peer." ); + $assert.isEqual( 0, attempt.cancelCount, "A completed handshake must not be cancelled." ); + } + + manager.shutdown(); + for( var attempt in connectionAttempts ) { + $assert.isEqual( 1, attempt.webSocket.closeCount, "Shutdown should initiate one close per connected peer." ); + $assert.isEqual( 1, attempt.webSocket.abortCount, "Shutdown should force-abort each connected peer." ); + $assert.isEqual( 0, attempt.cancelCount, "Completed futures must not be retained for shutdown cancellation." ); + } +} + +function testShutdownIsIdempotentAndContinuesAfterPeerCloseFailure() { + var failingPeer = { closeCount : 0 }; + failingPeer.close = ( force )=>{ + failingPeer.closeCount++; + throw( type="CloseFailure", message="simulated close failure" ); + }; + var healthyPeer = { closeCount : 0, forceClosed : false }; + healthyPeer.close = ( force )=>{ + healthyPeer.closeCount++; + healthyPeer.forceClosed = force; + }; + var httpClient = { shutdownCount : 0 }; + httpClient.newWebSocketBuilder = ()=>{}; + httpClient.shutdownNow = ()=>httpClient.shutdownCount++; + + var manager = newClusterManager( newSocketBoxMock(), newConfig(), httpClient ); + manager.setPeerConnections( { + "failing-peer" : failingPeer, + "healthy-peer" : healthyPeer + } ); + + manager.shutdown(); + manager.shutdown(); + + $assert.isEqual( 1, failingPeer.closeCount, "A failing peer should only be closed once." ); + $assert.isEqual( 1, healthyPeer.closeCount, "A peer failure must not prevent other peers from closing." ); + $assert.isTrue( healthyPeer.forceClosed, "Manager shutdown must force-close healthy peers." ); + $assert.isEqual( 0, manager.getPeerConnections().len(), "Shutdown must clear all peers even when one close fails." ); + var javaVersion = val( createObject( "java", "java.lang.System" ).getProperty( "java.specification.version" ) ); + $assert.isEqual( + javaVersion >= 21 ? 1 : 0, + httpClient.shutdownCount, + "The shared HTTP client should be shut down at most once." + ); +} + +function testSuccessfulHandshakeCompletingDuringShutdownIsNotRetained() { + var manager = ""; + var socket = { closeCount : 0, abortCount : 0 }; + socket.sendClose = ( statusCode, reason )=>{ + socket.closeCount++; + return { get : ()=>{} }; + }; + socket.abort = ()=>socket.abortCount++; + var connectionFuture = { cancelCount : 0 }; + connectionFuture.isDone = ()=>true; + connectionFuture.cancel = ( mayInterruptIfRunning )=>{ + connectionFuture.cancelCount++; + return true; + }; + connectionFuture.get = ()=>{ + manager.shutdown(); + return socket; + }; + + var builder = {}; + builder.connectTimeout = ( duration )=>builder; + builder.header = ( name, value )=>builder; + builder.buildAsync = ( uri, listener )=>{ + listener.setWebSocket( socket ); + return connectionFuture; + }; + var httpClient = { shutdownCount : 0 }; + httpClient.newWebSocketBuilder = ()=>builder; + httpClient.shutdownNow = ()=>httpClient.shutdownCount++; + + manager = newClusterManager( + newSocketBoxMock(), + newConfig(), + httpClient, + ( peer )=>peer + ); + manager.addPeer( "ws://late-peer:8080/ws" ); + + $assert.isEqual( 0, connectionFuture.cancelCount, "A completed future should not be cancelled during shutdown." ); + $assert.isEqual( 1, socket.closeCount, "A peer that completes during shutdown must be closed." ); + $assert.isEqual( 1, socket.abortCount, "A peer that completes during shutdown must be force-aborted." ); + $assert.isEqual( 0, manager.getPeerConnections().len(), "A peer that completes during shutdown must not be retained." ); +} + +function testCacheRemovalRetriesAfterConflictingWrite() { + var currentPeer = "ws://current:8080/ws"; + var expiredPeer = "ws://expired:8080/ws"; + var concurrentPeer = "ws://concurrent:8080/ws"; + var cacheKeyPrefix = "test-socketbox-cluster-peers"; + var cacheData = { + "#cacheKeyPrefix#" : [ expiredPeer, currentPeer ].toList( chr(13) & chr(10) ), + "#cacheKeyPrefix#-#expiredPeer#" : 1 + }; + var peerListWrites = 0; + var cacheProvider = {}; + cacheProvider.get = ( key )=>cacheData.keyExists( key ) ? cacheData[ key ] : nullValue(); + cacheProvider.set = ( key, value )=>{ + cacheData[ key ] = value; + if( key == cacheKeyPrefix ) { + peerListWrites++; + if( peerListWrites == 1 ) { + cacheData[ key ] = [ expiredPeer, currentPeer, concurrentPeer ].toList( chr(13) & chr(10) ); + } + } + }; + cacheProvider.clear = ( key )=>{ + var existed = cacheData.keyExists( key ); + cacheData.delete( key ); + return existed; + }; + var httpClient = { shutdownNow : ()=>{} }; + httpClient.newWebSocketBuilder = ()=>{}; + var manager = newClusterManager( newSocketBoxMock(), newConfig( cacheProvider ), httpClient ); + + var removed = manager.removePeersFromCache( [ expiredPeer ], 2 ); + var remainingPeers = manager.getCachePeers(); + + $assert.isTrue( removed, "Cache removal should succeed after retrying a conflicting write." ); + $assert.isEqual( 2, peerListWrites, "A conflicting write should force one verified retry." ); + $assert.isFalse( remainingPeers.contains( expiredPeer ), "The expired peer must be absent after the retry." ); + $assert.isTrue( remainingPeers.contains( currentPeer ), "The current peer must be preserved during the retry." ); + $assert.isTrue( remainingPeers.contains( concurrentPeer ), "A concurrently observed active peer must be preserved." ); + + manager.shutdown(); +} + +function testCacheRegistrationUsesExactPeerIdentity() { + var currentPeer = "ws://node-1:8080/ws"; + var similarPeer = "ws://node-10:8080/ws"; + var cacheKeyPrefix = "test-socketbox-cluster-peers"; + var cacheData = { + "#cacheKeyPrefix#" : similarPeer + }; + var cacheProvider = {}; + cacheProvider.get = ( key )=>cacheData.keyExists( key ) ? cacheData[ key ] : nullValue(); + cacheProvider.set = ( key, value )=>cacheData[ key ] = value; + cacheProvider.clear = ( key )=>{ + var existed = cacheData.keyExists( key ); + cacheData.delete( key ); + return existed; + }; + var httpClient = { shutdownNow : ()=>{} }; + httpClient.newWebSocketBuilder = ()=>{}; + var manager = newClusterManager( + newSocketBoxMock(), + newConfig( cacheProvider, currentPeer ), + httpClient + ); + + manager.ensureMyselfInCache( 1 ); + var cachedPeers = manager.getCachePeers(); + + $assert.isTrue( cachedPeers.contains( currentPeer ), "The current peer must be registered when only a similar name exists." ); + $assert.isTrue( cachedPeers.contains( similarPeer ), "Registering the current peer must preserve the similarly named peer." ); + $assert.isEqual( 2, cachedPeers.len(), "Similar peer names must remain distinct cache entries." ); + + manager.shutdown(); +} + +function testConnectionFailurePreservesAdaptiveDelayUntilSuccess() { + var buildCount = 0; + var successfulSocket = { closeCount : 0, abortCount : 0 }; + successfulSocket.sendClose = ( statusCode, reason )=>{ + successfulSocket.closeCount++; + return { get : ()=>{} }; + }; + successfulSocket.abort = ()=>successfulSocket.abortCount++; + var httpClient = { shutdownNow : ()=>{} }; + httpClient.newWebSocketBuilder = ()=>{ + buildCount++; + var builder = {}; + builder.connectTimeout = ( duration )=>builder; + builder.header = ( name, value )=>builder; + builder.buildAsync = ( uri, listener )=>{ + var future = {}; + future.isDone = ()=>true; + future.cancel = ( mayInterruptIfRunning )=>true; + if( buildCount == 1 ) { + future.get = ()=>{ + throw( type="java.util.concurrent.TimeoutException", message="test timeout" ); + }; + } else { + listener.setWebSocket( successfulSocket ); + future.get = ()=>successfulSocket; + } + return future; + }; + return builder; + }; + var manager = newClusterManager( + newSocketBoxMock(), + newConfig(), + httpClient, + ( peer )=>peer + ); + manager.setDelaySeconds( 60 ); + + manager.addPeer( "ws://retry-peer:8080/ws" ); + $assert.isEqual( 60, manager.getDelaySeconds(), "A failed connection must not reset the adaptive delay." ); + + manager.addPeer( "ws://retry-peer:8080/ws" ); + $assert.isEqual( 2, manager.getDelaySeconds(), "A successful connection should reset the adaptive delay." ); + $assert.isEqual( 1, manager.getPeerConnections().len(), "The successful retry should retain the peer." ); + + manager.shutdown(); +} + +} diff --git a/tests/specs/WebSocketCoreTest.cfc b/tests/specs/WebSocketCoreTest.cfc new file mode 100644 index 0000000..fa021ea --- /dev/null +++ b/tests/specs/WebSocketCoreTest.cfc @@ -0,0 +1,169 @@ +/** + * Regression tests for SocketBox hot-reconfiguration lifecycle management. + * Run with either checked-in CommandBox server config and request /tests/runner.cfm. + */ +component extends="tests.resources.ApplicationScopedBase" { + +function setup() { + variables.applicationState = {}; + for( var key in [ "socketBox", "SocketBoxConfig", "socketBoxClusterManagement" ] ) { + variables.applicationState[ key ] = { + exists : application.keyExists( key ) + }; + if( variables.applicationState[ key ].exists ) { + variables.applicationState[ key ].value = application[ key ]; + } + } + + variables.serverManagersExisted = server.keyExists( "socketBoxManagers" ); + if( variables.serverManagersExisted ) { + variables.originalServerManagers = server.socketBoxManagers; + } + + // Bypass environment auto-detection: these specs exercise configuration only. + application.socketBox = { serverType : "test" }; + application.delete( "SocketBoxConfig" ); + application.delete( "socketBoxClusterManagement" ); + server.socketBoxManagers = {}; +} + +function teardown() { + // Stop a real replacement manager if a test reached the enabled state. + if( + application.keyExists( "socketBoxClusterManagement" ) && + application.socketBoxClusterManagement.keyExists( "clusterManager" ) + ) { + var manager = application.socketBoxClusterManagement.clusterManager; + if( !isSimpleValue( manager ) ) { + try { + manager.shutdown(); + } catch( any ignored ) { + } + } + } + + // Signal any manager thread started by a failed assertion to exit. + server.socketBoxManagers = {}; + + for( var key in variables.applicationState ) { + if( variables.applicationState[ key ].exists ) { + application[ key ] = variables.applicationState[ key ].value; + } else { + application.delete( key ); + } + } + + if( variables.serverManagersExisted ) { + server.socketBoxManagers = variables.originalServerManagers; + } else { + server.delete( "socketBoxManagers" ); + } +} + +function newCore( required struct config ) { + return createObject( "component", "tests.resources.ConfigurableWebSocketCore" ) + .setTestConfig( arguments.config ); +} + +function newClusterConfig( boolean enabled=true ) { + return { + cluster : { + enable : arguments.enabled, + name : "ws://current:8080/ws", + peers : [], + secretKey : "test-secret" + } + }; +} + +function newOldManager( boolean failOnShutdown=false ) { + var shouldFailOnShutdown = arguments.failOnShutdown; + var manager = { + marker : createUUID(), + shutdownCount : 0, + peerStateReadCount : 0, + futureStateReadCount : 0, + wasCurrentAtShutdown : false + }; + manager.getPeerConnections = ()=>{ + manager.peerStateReadCount++; + return { "old-peer" : { marker : "must-not-transfer" } }; + }; + manager.getPendingConnectionFutures = ()=>{ + manager.futureStateReadCount++; + return [ { marker : "must-not-transfer" } ]; + }; + manager.shutdown = ()=>{ + manager.shutdownCount++; + manager.wasCurrentAtShutdown = + application.socketBoxClusterManagement.clusterManager.marker == manager.marker; + if( shouldFailOnShutdown ) { + throw( type="ReconfigurationFailure", message="simulated old-manager shutdown failure" ); + } + }; + return manager; +} + +function installOldManager( required any manager ) { + application.socketBoxClusterManagement = { + clusterManager : arguments.manager, + channels : {}, + managementChannels : {}, + selfChannels : {} + }; +} + +function testEnabledReconfigurationShutsDownBeforeReplacementWithoutTransferringState() { + var oldManager = newOldManager(); + installOldManager( oldManager ); + var core = newCore( newClusterConfig( true ) ); + var managerKey = core.getSocketBoxKey(); + server.socketBoxManagers[ managerKey ] = "old-manager-key"; + + var config = core._configure(); + var newManager = application.socketBoxClusterManagement.clusterManager; + + $assert.isTrue( config.cluster.enable, "The replacement configuration should remain cluster-enabled." ); + $assert.isEqual( 1, oldManager.shutdownCount, "The old manager must be shut down exactly once." ); + $assert.isTrue( oldManager.wasCurrentAtShutdown, "Shutdown must happen while the old manager is still installed." ); + $assert.isEqual( 0, oldManager.peerStateReadCount, "Peer connections must not be copied to the replacement manager." ); + $assert.isEqual( 0, oldManager.futureStateReadCount, "Pending futures must not be copied to the replacement manager." ); + $assert.isEqual( 0, newManager.getPeerConnections().len(), "The replacement manager must begin without old peer state." ); + $assert.isTrue( len( server.socketBoxManagers[ managerKey ] ), "The replacement manager must register its own lifecycle key." ); +} + +function testEnabledReconfigurationCanDisableClustering() { + var oldManager = newOldManager(); + installOldManager( oldManager ); + var core = newCore( newClusterConfig( false ) ); + + var config = core._configure(); + + $assert.isFalse( config.cluster.enable, "The new configuration should disable clustering." ); + $assert.isEqual( 1, oldManager.shutdownCount, "Disabling clustering must shut down the old manager once." ); + $assert.isTrue( oldManager.wasCurrentAtShutdown, "The old manager must remain installed until shutdown starts." ); + $assert.isFalse( application.keyExists( "socketBoxClusterManagement" ), "Disabling clustering must remove management state." ); +} + +function testReplacementFailureClearsConfigurationAndManagerRegistration() { + var oldManager = newOldManager( true ); + installOldManager( oldManager ); + var core = newCore( newClusterConfig( true ) ); + var managerKey = core.getSocketBoxKey(); + server.socketBoxManagers[ managerKey ] = "old-manager-key"; + var failureRaised = false; + + try { + core._configure(); + } catch( ReconfigurationFailure e ) { + failureRaised = true; + } + + $assert.isTrue( failureRaised, "A manager replacement failure should reach the caller." ); + $assert.isEqual( 1, oldManager.shutdownCount, "A failed shutdown must not be retried during the same replacement." ); + $assert.isFalse( application.keyExists( "SocketBoxConfig" ), "A failed replacement must remove its partial configuration." ); + $assert.isFalse( application.keyExists( "socketBoxClusterManagement" ), "A failed replacement must remove management state." ); + $assert.isEqual( "", server.socketBoxManagers[ managerKey ], "A failed replacement must invalidate the manager thread key." ); +} + +}