Manage unpooled buffer lifecycle

README.md

@@ -460,6 +460,7 @@ BinaryBlobAppendWriter, and an atomic BinaryBlobAtomicReplacementWriter.
 
 The append writer will continuously append to the same file, keeping only a
 small buffer in memory and resetting it with each successful invocation.
+Call `blob.close()` after the final write when the session can no longer reconnect.
 If users flush it every ~5kb
 (you can check it via `player.session.getBinaryBlobOrNull()?.readableBytes()`),
 there should never be more than ~10MB of heap memory allocated for this.

buffer/src/main/kotlin/net/rsprot/buffer/extensions/ByteBufExtensions.kt

@@ -11,3 +11,13 @@ public fun ByteBuf.toByteArray(): ByteArray {
     readBytes(array)
     return array
 }
+
+public inline fun <T> ByteBuf.releaseOnFailure(block: (() -> Unit) -> T): T {
+    var transferred = false
+    return try {
+        block { transferred = true }
+    } catch (throwable: Throwable) {
+        if (!transferred) release()
+        throw throwable
+    }
+}

haproxy/src/main/kotlin/org/jire/netty/haproxy/HAProxyPingHandler.kt

@@ -34,8 +34,7 @@ public class HAProxyPingHandler
         private val response: ByteBuf =
             Unpooled.unreleasableBuffer(
                 Unpooled
-                    .directBuffer(1, 1)
-                    .writeByte(responseOpcode),
+                    .wrappedBuffer(byteArrayOf(responseOpcode.toByte())),
             )
 
         override fun channelRead0(

protocol/osrs-221/osrs-221-api/src/main/kotlin/net/rsprot/protocol/api/login/GameLoginResponseHandler.kt

@@ -7,6 +7,7 @@ import io.netty.channel.ChannelFutureListener
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.ChannelPipeline
 import io.netty.handler.timeout.IdleStateHandler
+import net.rsprot.buffer.extensions.releaseOnFailure
 import net.rsprot.buffer.extensions.toJagByteBuf
 import net.rsprot.crypto.cipher.StreamCipher
 import net.rsprot.crypto.cipher.StreamCipherPair
@@ -91,80 +92,90 @@ public class GameLoginResponseHandler<R>(
         // game packets, due to the executor differing and race conditions taking place.
 
         val buffer = ctx.alloc().buffer(37).toJagByteBuf()
-        if (!networkService.betaWorld) {
-            buffer.p1(encoder.prot.opcode)
-        }
-        // Client expects a hardcoded 37 value for the size, even though it is not the exact size
-        // of the login packet
-        buffer.p1(37)
-        encoder.encode(cipher.encoderCipher, buffer, response)
+        buffer.buffer.releaseOnFailure { transfer ->
+            if (!networkService.betaWorld) {
+                buffer.p1(encoder.prot.opcode)
+            }
+            // Client expects a hardcoded 37 value for the size, even though it is not the exact size
+            // of the login packet
+            buffer.p1(37)
+            encoder.encode(cipher.encoderCipher, buffer, response)
 
-        val pipeline = ctx.channel().pipeline()
+            val pipeline = ctx.channel().pipeline()
 
-        val session =
-            createSession(loginBlock, pipeline, cipher.decodeCipher, oldSchoolClientType, cipher.encoderCipher)
-        networkService.js5Authorizer.authorize(ctx.hostAddress())
-        ctx.executor().submit {
-            ctx.write(buffer.buffer)
-            session.onLoginTransitionComplete()
-        }
-        networkLog(logger) {
-            "Successful game login from channel '${ctx.channel()}': $loginBlock"
+            val session =
+                createSession(loginBlock, pipeline, cipher.decodeCipher, oldSchoolClientType, cipher.encoderCipher)
+            networkService.js5Authorizer.authorize(ctx.hostAddress())
+            ctx.executor().submit {
+                ctx.write(buffer.buffer)
+                session.onLoginTransitionComplete()
+            }
+            transfer()
+            networkLog(logger) {
+                "Successful game login from channel '${ctx.channel()}': $loginBlock"
+            }
+            return session
         }
-        return session
     }
 
     public fun writeSuccessfulResponse(
         response: LoginResponse.ReconnectOk,
         loginBlock: LoginBlock<*>,
     ): Session<R> {
-        // Ensure it isn't null - our decoder pre-validates it long before hitting this function,
-        // so this exception should never be hit.
-        val oldSchoolClientType =
-            checkNotNull(loginBlock.clientType.toOldSchoolClientType()) {
-                "Login client type cannot be null"
-            }
-        val (encodingCipher, decodingCipher) = createStreamCipherPair(loginBlock)
+        try {
+            // Ensure it isn't null - our decoder pre-validates it long before hitting this function,
+            // so this exception should never be hit.
+            val oldSchoolClientType =
+                checkNotNull(loginBlock.clientType.toOldSchoolClientType()) {
+                    "Login client type cannot be null"
+                }
+            val (encodingCipher, decodingCipher) = createStreamCipherPair(loginBlock)
 
-        val encoder =
-            networkService
-                .encoderRepositories
-                .loginMessageEncoderRepository
-                .getEncoder(response::class.java)
+            val encoder =
+                networkService
+                    .encoderRepositories
+                    .loginMessageEncoderRepository
+                    .getEncoder(response::class.java)
 
-        // Allocate a perfectly-sized buffer for this packet
-        val bufLength = Byte.SIZE_BYTES + Short.SIZE_BYTES + response.content().readableBytes()
-        val buffer = ctx.alloc().buffer(bufLength).toJagByteBuf()
-        buffer.p1(encoder.prot.opcode)
+            // Allocate a perfectly-sized buffer for this packet
+            val bufLength = Byte.SIZE_BYTES + Short.SIZE_BYTES + response.content().readableBytes()
+            val buffer = ctx.alloc().buffer(bufLength).toJagByteBuf()
+            buffer.buffer.releaseOnFailure { transfer ->
+                buffer.p1(encoder.prot.opcode)
 
-        // Write a placeholder size of 0 bytes
-        val lengthPos = buffer.writerIndex()
-        buffer.p2(0)
+                // Write a placeholder size of 0 bytes
+                val lengthPos = buffer.writerIndex()
+                buffer.p2(0)
 
-        // Write the payload
-        val start = buffer.writerIndex()
-        encoder.encode(encodingCipher, buffer, response)
-        val end = buffer.writerIndex()
-        val written = end - start
+                // Write the payload
+                val start = buffer.writerIndex()
+                encoder.encode(encodingCipher, buffer, response)
+                val end = buffer.writerIndex()
+                val written = end - start
 
-        // Update the size with the actual number of bytes written
-        buffer.writerIndex(lengthPos)
-        buffer.p2(written)
-        buffer.writerIndex(end)
+                // Update the size with the actual number of bytes written
+                buffer.writerIndex(lengthPos)
+                buffer.p2(written)
+                buffer.writerIndex(end)
 
-        val pipeline = ctx.channel().pipeline()
+                val pipeline = ctx.channel().pipeline()
 
-        val session =
-            createSession(loginBlock, pipeline, decodingCipher, oldSchoolClientType, encodingCipher)
-        networkService.js5Authorizer.authorize(ctx.hostAddress())
-        ctx.executor().submit {
-            ctx.write(buffer.buffer)
-            session.onLoginTransitionComplete()
-        }
-        networkLog(logger) {
-            "Successful game login from channel '${ctx.channel()}': $loginBlock"
+                val session =
+                    createSession(loginBlock, pipeline, decodingCipher, oldSchoolClientType, encodingCipher)
+                networkService.js5Authorizer.authorize(ctx.hostAddress())
+                ctx.executor().submit {
+                    ctx.write(buffer.buffer)
+                    session.onLoginTransitionComplete()
+                }
+                transfer()
+                networkLog(logger) {
+                    "Successful game login from channel '${ctx.channel()}': $loginBlock"
+                }
+                return session
+            }
+        } finally {
+            response.release()
         }
-        return session
     }
 
     private fun createStreamCipherPair(loginBlock: LoginBlock<*>): StreamCipherPair {
@@ -256,6 +267,7 @@ public class GameLoginResponseHandler<R>(
             networkLog(logger) {
                 "Channel '${ctx.channel()}' has gone inactive, skipping failed response."
             }
+            response.safeRelease()
             return
         }
         networkLog(logger) {

protocol/osrs-221/osrs-221-api/src/main/kotlin/net/rsprot/protocol/api/login/LoginConnectionHandler.kt

@@ -108,6 +108,7 @@ public class LoginConnectionHandler<R>(
             }
             is GameLogin -> {
                 if (this.loginState != LoginState.UNINITIALIZED) {
+                    msg.buffer.buffer.release()
                     ctx.close()
                     return
                 }
@@ -116,6 +117,7 @@ public class LoginConnectionHandler<R>(
             }
 
             is GameReconnect -> {
+                releaseLoginBlock()
                 this.loginPacket = msg
                 continueLogin(ctx)
             }

protocol/osrs-221/osrs-221-model/src/main/kotlin/net/rsprot/protocol/loginprot/outgoing/LoginResponse.kt

@@ -104,8 +104,13 @@ public sealed interface LoginResponse : OutgoingLoginMessage {
                 playerInfo.ensureReconnectCalled()
                 val allocator = playerInfo.allocator
                 val buffer = allocator.buffer(PLAYER_INFO_BLOCK_SIZE)
-                playerInfo.handleAbsolutePlayerPositions(buffer)
-                return buffer
+                return try {
+                    playerInfo.handleAbsolutePlayerPositions(buffer)
+                    buffer
+                } catch (throwable: Throwable) {
+                    buffer.release()
+                    throw throwable
+                }
             }
         }
 

protocol/osrs-222/osrs-222-api/src/main/kotlin/net/rsprot/protocol/api/login/GameLoginResponseHandler.kt

@@ -7,6 +7,7 @@ import io.netty.channel.ChannelFutureListener
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.ChannelPipeline
 import io.netty.handler.timeout.IdleStateHandler
+import net.rsprot.buffer.extensions.releaseOnFailure
 import net.rsprot.buffer.extensions.toJagByteBuf
 import net.rsprot.crypto.cipher.StreamCipher
 import net.rsprot.crypto.cipher.StreamCipherPair
@@ -91,80 +92,90 @@ public class GameLoginResponseHandler<R>(
         // game packets, due to the executor differing and race conditions taking place.
 
         val buffer = ctx.alloc().buffer(37).toJagByteBuf()
-        if (!networkService.betaWorld) {
-            buffer.p1(encoder.prot.opcode)
-        }
-        // Client expects a hardcoded 37 value for the size, even though it is not the exact size
-        // of the login packet
-        buffer.p1(37)
-        encoder.encode(cipher.encoderCipher, buffer, response)
+        buffer.buffer.releaseOnFailure { transfer ->
+            if (!networkService.betaWorld) {
+                buffer.p1(encoder.prot.opcode)
+            }
+            // Client expects a hardcoded 37 value for the size, even though it is not the exact size
+            // of the login packet
+            buffer.p1(37)
+            encoder.encode(cipher.encoderCipher, buffer, response)
 
-        val pipeline = ctx.channel().pipeline()
+            val pipeline = ctx.channel().pipeline()
 
-        val session =
-            createSession(loginBlock, pipeline, cipher.decodeCipher, oldSchoolClientType, cipher.encoderCipher)
-        networkService.js5Authorizer.authorize(ctx.hostAddress())
-        ctx.executor().submit {
-            ctx.write(buffer.buffer)
-            session.onLoginTransitionComplete()
-        }
-        networkLog(logger) {
-            "Successful game login from channel '${ctx.channel()}': $loginBlock"
+            val session =
+                createSession(loginBlock, pipeline, cipher.decodeCipher, oldSchoolClientType, cipher.encoderCipher)
+            networkService.js5Authorizer.authorize(ctx.hostAddress())
+            ctx.executor().submit {
+                ctx.write(buffer.buffer)
+                session.onLoginTransitionComplete()
+            }
+            transfer()
+            networkLog(logger) {
+                "Successful game login from channel '${ctx.channel()}': $loginBlock"
+            }
+            return session
         }
-        return session
     }
 
     public fun writeSuccessfulResponse(
         response: LoginResponse.ReconnectOk,
         loginBlock: LoginBlock<*>,
     ): Session<R> {
-        // Ensure it isn't null - our decoder pre-validates it long before hitting this function,
-        // so this exception should never be hit.
-        val oldSchoolClientType =
-            checkNotNull(loginBlock.clientType.toOldSchoolClientType()) {
-                "Login client type cannot be null"
-            }
-        val (encodingCipher, decodingCipher) = createStreamCipherPair(loginBlock)
+        try {
+            // Ensure it isn't null - our decoder pre-validates it long before hitting this function,
+            // so this exception should never be hit.
+            val oldSchoolClientType =
+                checkNotNull(loginBlock.clientType.toOldSchoolClientType()) {
+                    "Login client type cannot be null"
+                }
+            val (encodingCipher, decodingCipher) = createStreamCipherPair(loginBlock)
 
-        val encoder =
-            networkService
-                .encoderRepositories
-                .loginMessageEncoderRepository
-                .getEncoder(response::class.java)
+            val encoder =
+                networkService
+                    .encoderRepositories
+                    .loginMessageEncoderRepository
+                    .getEncoder(response::class.java)
 
-        // Allocate a perfectly-sized buffer for this packet
-        val bufLength = Byte.SIZE_BYTES + Short.SIZE_BYTES + response.content().readableBytes()
-        val buffer = ctx.alloc().buffer(bufLength).toJagByteBuf()
-        buffer.p1(encoder.prot.opcode)
+            // Allocate a perfectly-sized buffer for this packet
+            val bufLength = Byte.SIZE_BYTES + Short.SIZE_BYTES + response.content().readableBytes()
+            val buffer = ctx.alloc().buffer(bufLength).toJagByteBuf()
+            buffer.buffer.releaseOnFailure { transfer ->
+                buffer.p1(encoder.prot.opcode)
 
-        // Write a placeholder size of 0 bytes
-        val lengthPos = buffer.writerIndex()
-        buffer.p2(0)
+                // Write a placeholder size of 0 bytes
+                val lengthPos = buffer.writerIndex()
+                buffer.p2(0)
 
-        // Write the payload
-        val start = buffer.writerIndex()
-        encoder.encode(encodingCipher, buffer, response)
-        val end = buffer.writerIndex()
-        val written = end - start
+                // Write the payload
+                val start = buffer.writerIndex()
+                encoder.encode(encodingCipher, buffer, response)
+                val end = buffer.writerIndex()
+                val written = end - start
 
-        // Update the size with the actual number of bytes written
-        buffer.writerIndex(lengthPos)
-        buffer.p2(written)
-        buffer.writerIndex(end)
+                // Update the size with the actual number of bytes written
+                buffer.writerIndex(lengthPos)
+                buffer.p2(written)
+                buffer.writerIndex(end)
 
-        val pipeline = ctx.channel().pipeline()
+                val pipeline = ctx.channel().pipeline()
 
-        val session =
-            createSession(loginBlock, pipeline, decodingCipher, oldSchoolClientType, encodingCipher)
-        networkService.js5Authorizer.authorize(ctx.hostAddress())
-        ctx.executor().submit {
-            ctx.write(buffer.buffer)
-            session.onLoginTransitionComplete()
-        }
-        networkLog(logger) {
-            "Successful game login from channel '${ctx.channel()}': $loginBlock"
+                val session =
+                    createSession(loginBlock, pipeline, decodingCipher, oldSchoolClientType, encodingCipher)
+                networkService.js5Authorizer.authorize(ctx.hostAddress())
+                ctx.executor().submit {
+                    ctx.write(buffer.buffer)
+                    session.onLoginTransitionComplete()
+                }
+                transfer()
+                networkLog(logger) {
+                    "Successful game login from channel '${ctx.channel()}': $loginBlock"
+                }
+                return session
+            }
+        } finally {
+            response.release()
         }
-        return session
     }
 
     private fun createStreamCipherPair(loginBlock: LoginBlock<*>): StreamCipherPair {
@@ -256,6 +267,7 @@ public class GameLoginResponseHandler<R>(
             networkLog(logger) {
                 "Channel '${ctx.channel()}' has gone inactive, skipping failed response."
             }
+            response.safeRelease()
             return
         }
         networkLog(logger) {