updated Netty to 4.2.4

improved memory usage of the in-memory cache backend

build.gradle

@@ -46,6 +46,7 @@ allprojects { subproject ->
             testImplementation catalog.junit.jupiter.api
             testImplementation catalog.junit.jupiter.params
             testRuntimeOnly catalog.junit.jupiter.engine
+            testRuntimeOnly catalog.junit.platform.launcher
         }
 
         test {

gradle.properties

@@ -2,9 +2,9 @@ org.gradle.configuration-cache=false
 org.gradle.parallel=true
 org.gradle.caching=true
 
-rbcs.version = 0.3.1
+rbcs.version = 0.3.3
 
-lys.version = 2025.06.10
+lys.version = 2025.08.15
 
 gitea.maven.url = https://gitea.woggioni.net/api/packages/woggioni/maven
 docker.registry.url=gitea.woggioni.net

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.14.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.14.3-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

gradlew

@@ -114,7 +114,7 @@ case "$( uname )" in                #(
   NONSTOP* )        nonstop=true ;;
 esac
 
-CLASSPATH=$APP_HOME/gradle/wrapper/gradle-wrapper.jar
+CLASSPATH="\\\"\\\""
 
 
 # Determine the Java command to use to start the JVM.
@@ -205,7 +205,7 @@ fi
 DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 
 # Collect all arguments for the java command:
-#   * DEFAULT_JVM_OPTS, JAVA_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments,
+#   * DEFAULT_JVM_OPTS, JAVA_OPTS, and optsEnvironmentVar are not allowed to contain shell fragments,
 #     and any embedded shellness will be escaped.
 #   * For example: A user cannot expect ${Hostname} to be expanded, as it is an environment variable and will be
 #     treated as '${Hostname}' itself on the command line.
@@ -213,7 +213,7 @@ DEFAULT_JVM_OPTS='"-Xmx64m" "-Xms64m"'
 set -- \
         "-Dorg.gradle.appname=$APP_BASE_NAME" \
         -classpath "$CLASSPATH" \
-        org.gradle.wrapper.GradleWrapperMain \
+        -jar "$APP_HOME/gradle/wrapper/gradle-wrapper.jar" \
         "$@"
 
 # Stop when "xargs" is not available.

gradlew.bat

@@ -70,11 +70,11 @@ goto fail
 :execute
 @rem Setup the command line
 
-set CLASSPATH=%APP_HOME%\gradle\wrapper\gradle-wrapper.jar
+set CLASSPATH=
 
 
 @rem Execute Gradle
-"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" org.gradle.wrapper.GradleWrapperMain %*
+"%JAVA_EXE%" %DEFAULT_JVM_OPTS% %JAVA_OPTS% %GRADLE_OPTS% "-Dorg.gradle.appname=%APP_BASE_NAME%" -classpath "%CLASSPATH%" -jar "%APP_HOME%\gradle\wrapper\gradle-wrapper.jar" %*
 
 :end
 @rem End local scope for the variables with windows NT shell

rbcs-client/src/main/kotlin/net/woggioni/rbcs/client/RemoteBuildCacheClient.kt

@@ -254,19 +254,25 @@ class RemoteBuildCacheClient(private val profile: Configuration.Profile) : AutoC
     fun get(key: String): CompletableFuture<ByteArray?> {
         return executeWithRetry {
             sendRequest(profile.serverURI.resolve(key), HttpMethod.GET, null)
-        }.thenApply {
-            val status = it.status()
-            if (it.status() == HttpResponseStatus.NOT_FOUND) {
+        }.thenApply { response ->
+            val status = response.status()
+            if (response.status() == HttpResponseStatus.NOT_FOUND) {
+                response.release()
                 null
-            } else if (it.status() != HttpResponseStatus.OK) {
+            } else if (response.status() != HttpResponseStatus.OK) {
+                response.release()
                 throw HttpException(status)
             } else {
-                it.content()
+                response.content().also {
+                    it.retain()
+                    response.release()
+                }
             }
         }.thenApply { maybeByteBuf ->
-            maybeByteBuf?.let {
-                val result = ByteArray(it.readableBytes())
-                it.getBytes(0, result)
+            maybeByteBuf?.let { buf ->
+                val result = ByteArray(buf.readableBytes())
+                buf.getBytes(0, result)
+                buf.release()
                 result
             }
         }
@@ -318,7 +324,7 @@ class RemoteBuildCacheClient(private val profile: Configuration.Profile) : AutoC
                             response: FullHttpResponse
                         ) {
                             pipeline.remove(this)
-                            responseFuture.complete(response)
+                            responseFuture.complete(response.retainedDuplicate())
                             if (!profile.connection.requestPipelining) {
                                 pool.release(channel)
                             }

rbcs-client/src/test/kotlin/net/woggioni/rbcs/client/RetryTest.kt

@@ -12,6 +12,7 @@ import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.Arguments
 import org.junit.jupiter.params.provider.ArgumentsProvider
 import org.junit.jupiter.params.provider.ArgumentsSource
+import org.junit.jupiter.params.support.ParameterDeclarations
 
 class RetryTest {
 
@@ -23,7 +24,10 @@ class RetryTest {
     )
 
     class TestArguments : ArgumentsProvider {
-        override fun provideArguments(context: ExtensionContext): Stream<out Arguments> {
+        override fun provideArguments(
+            parameters: ParameterDeclarations,
+            context: ExtensionContext
+        ): Stream<out Arguments> {
             return Stream.of(
                 TestArgs(
                     seed = 101325,

rbcs-server-memcache/README.md

@@ -19,6 +19,8 @@ to `memcacheCacheType`.
 
 The plugins currently supports the following configuration attributes:
 - `max-age`: the amount of time cache entries will be retained on memcache
+- `key-prefix`: a string that will be prepended to all the keys inserted in memcache,
+  useful in case the caching backend is shared with other applications
 - `digest`: digest algorithm to use on the key before submission 
   to memcache (optional, no digest is applied if omitted)
 - `compression`: compression algorithm to apply to cache values before, 
@@ -35,6 +37,7 @@ The plugins currently supports the following configuration attributes:
     ...
     <cache xs:type="rbcs-memcache:memcacheCacheType" 
            max-age="P7D"
+           key-prefix="rbcs-"
            digest="SHA-256"
            compression-mode="deflate"
            compression-level="6">

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/RemoteBuildCacheServer.kt

@@ -345,7 +345,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
                 maxChunkSize = cfg.connection.chunkSize
             }
             pipeline.addLast(HttpServerCodec(httpDecoderConfig))
-            pipeline.addLast(ReadTriggerDuplexHandler.NAME, ReadTriggerDuplexHandler)
+            pipeline.addLast(ReadTriggerDuplexHandler.NAME, ReadTriggerDuplexHandler())
             pipeline.addLast(MaxRequestSizeHandler.NAME, MaxRequestSizeHandler(cfg.connection.maxRequestSize))
             pipeline.addLast(HttpChunkContentCompressor(1024))
             pipeline.addLast(ChunkedWriteHandler())

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCache.kt

@@ -1,6 +1,5 @@
 package net.woggioni.rbcs.server.cache
 
-import io.netty.buffer.ByteBuf
 import java.time.Duration
 import java.time.Instant
 import java.util.PriorityQueue
@@ -22,7 +21,7 @@ private class CacheKey(private val value: ByteArray) {
 
 class CacheEntry(
     val metadata: CacheValueMetadata,
-    val content: ByteBuf
+    val content: ByteArray
 )
 
 class InMemoryCache(
@@ -66,8 +65,6 @@ class InMemoryCache(
                                 val removed = map.remove(el.key, value)
                                 if (removed) {
                                     updateSizeAfterRemoval(value.content)
-                                    //Decrease the reference count for map
-                                    value.content.release()
                                 }
                             } else {
                                 removalQueue.offer(el)
@@ -75,9 +72,6 @@ class InMemoryCache(
                                 cond.await(interval.toMillis(), TimeUnit.MILLISECONDS)
                             }
                         }
-                        map.forEach {
-                            it.value.content.release()
-                        }
                         map.clear()
                     }
                     complete(null)
@@ -95,15 +89,13 @@ class InMemoryCache(
             val removed = map.remove(el.key, value)
             if (removed) {
                 val newSize = updateSizeAfterRemoval(value.content)
-                //Decrease the reference count for map
-                value.content.release()
                 return newSize
             }
         }
     }
 
-    private fun updateSizeAfterRemoval(removed: ByteBuf): Long {
-        mapSize -= removed.readableBytes()
+    private fun updateSizeAfterRemoval(removed: ByteArray): Long {
+        mapSize -= removed.size
         return mapSize
     }
 
@@ -117,7 +109,7 @@ class InMemoryCache(
 
     fun get(key: ByteArray) = lock.readLock().withLock {
         map[CacheKey(key)]?.run {
-            CacheEntry(metadata, content.retainedDuplicate())
+            CacheEntry(metadata, content)
         }
     }
 
@@ -127,12 +119,8 @@ class InMemoryCache(
     ) {
         val cacheKey = CacheKey(key)
         lock.writeLock().withLock {
-            val oldSize = map.put(cacheKey, value)?.let { old ->
-                val result = old.content.readableBytes()
-                old.content.release()
-                result
-            } ?: 0
-            val delta = value.content.readableBytes() - oldSize
+            val oldSize = map.put(cacheKey, value)?.content?.size ?: 0
+            val delta = value.content.size - oldSize
             mapSize += delta
             removalQueue.offer(RemovalQueueElement(cacheKey, value, Instant.now().plus(maxAge)))
             while (mapSize > maxSize) {
@@ -140,4 +128,4 @@ class InMemoryCache(
             }
         }
     }
-}
+}

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCacheHandler.kt

@@ -2,6 +2,8 @@ package net.woggioni.rbcs.server.cache
 
 import io.netty.buffer.ByteBuf
 import io.netty.channel.ChannelHandlerContext
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
 import java.util.zip.Deflater
 import java.util.zip.DeflaterOutputStream
 import java.util.zip.InflaterOutputStream
@@ -111,18 +113,23 @@ class InMemoryCacheHandler(
         handleCacheContent(ctx, msg)
         when (val req = inProgressRequest) {
             is InProgressGetRequest -> {
+//                this.inProgressRequest = null
                 cache.get(processCacheKey(req.request.key, null, digestAlgorithm))?.let { value ->
                     sendMessageAndFlush(ctx, CacheValueFoundResponse(req.request.key, value.metadata))
                     if (compressionEnabled) {
                         val buf = ctx.alloc().heapBuffer()
                         InflaterOutputStream(ByteBufOutputStream(buf)).use {
-                            value.content.readBytes(it, value.content.readableBytes())
-                            value.content.release()
+                            it.write(value.content)
                             buf.retain()
                         }
                         sendMessage(ctx, LastCacheContent(buf))
                     } else {
-                        sendMessage(ctx, LastCacheContent(value.content))
+                        val buf = ctx.alloc().heapBuffer()
+                        ByteBufOutputStream(buf).use {
+                            it.write(value.content)
+                            buf.retain()
+                        }
+                        sendMessage(ctx, LastCacheContent(buf))
                     }
                 } ?: sendMessage(ctx, CacheValueNotFoundResponse(req.request.key))
             }
@@ -132,8 +139,11 @@ class InMemoryCacheHandler(
                 val buf = req.buf
                 buf.retain()
                 req.close()
+
+                val bytes = ByteArray(buf.readableBytes()).also(buf::readBytes)
+                buf.release()
                 val cacheKey = processCacheKey(req.request.key, null, digestAlgorithm)
-                cache.put(cacheKey, CacheEntry(req.request.metadata, buf))
+                cache.put(cacheKey, CacheEntry(req.request.metadata, bytes))
                 sendMessageAndFlush(ctx, CachePutResponse(req.request.key))
             }
         }

rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/handler/ReadTriggerDuplexHandler.kt

@@ -1,23 +1,43 @@
 package net.woggioni.rbcs.server.handler
 
+import io.netty.buffer.ByteBufHolder
 import io.netty.channel.ChannelDuplexHandler
-import io.netty.channel.ChannelHandler.Sharable
 import io.netty.channel.ChannelHandlerContext
 import io.netty.channel.ChannelPromise
 import io.netty.handler.codec.http.LastHttpContent
+import net.woggioni.rbcs.common.createLogger
 
-@Sharable
-object ReadTriggerDuplexHandler : ChannelDuplexHandler() {
-    val NAME = ReadTriggerDuplexHandler::class.java.name
+class ReadTriggerDuplexHandler : ChannelDuplexHandler() {
+    companion object {
+        val NAME = ReadTriggerDuplexHandler::class.java.name
+        private val log = createLogger<ReadTriggerDuplexHandler>()
+    }
+
+    private var inFlight = 0
+    private val messageBuffer = ArrayDeque<Any>()
 
     override fun handlerAdded(ctx: ChannelHandlerContext) {
         ctx.read()
     }
 
     override fun channelRead(ctx: ChannelHandlerContext, msg: Any) {
-        super.channelRead(ctx, msg)
-        if(msg !is LastHttpContent) {
+        if(inFlight > 0) {
+            messageBuffer.addLast(msg)
+        } else {
+            super.channelRead(ctx, msg)
+            if(msg !is LastHttpContent) {
+                invokeRead(ctx)
+            } else {
+                inFlight += 1
+            }
+        }
+    }
+
+    private fun invokeRead(ctx : ChannelHandlerContext) {
+        if(messageBuffer.isEmpty()) {
             ctx.read()
+        } else {
+            this.channelRead(ctx, messageBuffer.removeFirst())
         }
     }
 
@@ -28,7 +48,18 @@ object ReadTriggerDuplexHandler : ChannelDuplexHandler() {
     ) {
         super.write(ctx, msg, promise)
         if(msg is LastHttpContent) {
-            ctx.read()
+            inFlight -= 1
+            invokeRead(ctx)
         }
     }
+
+    override fun channelInactive(ctx: ChannelHandlerContext) {
+        while(messageBuffer.isNotEmpty()) {
+            val msg = messageBuffer.removeFirst()
+            if(msg is ByteBufHolder) {
+                msg.release()
+            }
+        }
+        super.channelInactive(ctx)
+    }
 }