fix: properly close buffers to prevent memory leaks

driver/src/main/java/eu/cloudnetservice/driver/network/buffer/DataBuf.java

@@ -231,8 +231,8 @@ public interface DataBuf extends AutoCloseable {
   <T> T readObject(@NonNull Type type);
 
   /**
-   * Reads the next requested data from the buffer. This method call is equivalent to {@code
-   * readNullable(readerWhenNonNull, null)}.
+   * Reads the next requested data from the buffer. This method call is equivalent to
+   * {@code readNullable(readerWhenNonNull, null)}.
    *
    * @param readerWhenNonNull the reader to read the data from the buffer when the next value is non-null.
    * @param <T>               the generic type of the data to read.
@@ -300,27 +300,32 @@ public interface DataBuf extends AutoCloseable {
   boolean accessible();
 
   /**
-   * Disables that the underlying buffer can be released when the reader index of this buffer reaches the tail or an
-   * explicit call to {@code release()} or {@code close()} is made.
+   * Get the amount of acquires that this data buf has. Initially a data buf is acquired once.
    *
-   * @return the same instance as used to call the method, for chaining.
+   * @return the amount of acquires. A value smaller or equal to zero means that the buffer was released.
    */
-  @NonNull DataBuf disableReleasing();
+  int acquires();
 
   /**
-   * Enables that the underlying buffer can be released when the reader index of this buffer reaches the tail or an
-   * explicit call to {@code release()} or {@code close()} is made.
+   * Acquires this buffer once. If a buffer gets acquired, further calls to {@link #release()} will decrease the count,
+   * but only release the buffer if there were more release than acquire calls.
    *
    * @return the same instance as used to call the method, for chaining.
    */
-  @NonNull DataBuf enableReleasing();
+  @NonNull DataBuf acquire();
 
   /**
-   * Explicitly releases all data associated with this buffer making it unavailable for further reads. This method does
-   * nothing if releasing was disables before calling this method.
+   * Explicitly releases all data associated with this buffer making it unavailable for further reads. This method only
+   * decreases the acquire count of the buffer in case it was acquired at least once.
    */
   void release();
 
+  /**
+   * Explicitly releases all data associated with this buffer making it unavailable for further reads. This method does
+   * not check if anyone acquired the buffer, it will be released in any case.
+   */
+  void forceRelease();
+
   /**
    * Explicitly releases all data associated with this buffer making it unavailable for further reads. This method does
    * nothing if releasing was disables before calling this method.

driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/DefaultFileChunkPacketSender.java

@@ -88,19 +88,24 @@ public DefaultFileChunkPacketSender(
       while (true) {
         var bytesRead = this.source.read(backingArray);
         if (bytesRead != -1 && bytesRead == backingArray.length) {
-          this.packetSplitter.accept(
-            ChunkedPacket.createChunk(this.chunkSessionInformation, readCalls++, backingArray));
+          // acquire the transfer information once before writing the data of the chunk
+          this.chunkSessionInformation.transferInformation().acquire();
+          this.packetSplitter.accept(ChunkedPacket.createChunk(
+            this.chunkSessionInformation,
+            readCalls++,
+            backingArray));
         } else {
           this.packetSplitter.accept(ChunkedPacket.createChunk(
             this.chunkSessionInformation,
             readCalls,
             readCalls,
             bytesRead == -1 ? 0 : bytesRead,
             bytesRead == -1 ? EMPTY_BYTE_ARRAY : backingArray));
-          // close the stream after reading the final chunk
+
+          // close the stream after reading the final chunk & release the extra content now
           this.source.close();
-          // release the extra content now
-          this.chunkSessionInformation.transferInformation().enableReleasing().release();
+          this.chunkSessionInformation.transferInformation().release();
+
           // successful transfer
           return TransferStatus.SUCCESS;
         }

driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/builder/DefaultChunkedPacketSenderBuilder.java

@@ -115,7 +115,9 @@ public abstract class DefaultChunkedPacketSenderBuilder implements ChunkedPacket
    */
   @Override
   public @NonNull ChunkedPacketSender.Builder withExtraData(@NonNull DataBuf extraData) {
-    this.transferInformation = extraData.disableReleasing();
+    // make sure that the old transfer information buffer is released
+    this.transferInformation.release();
+    this.transferInformation = extraData;
     return this;
   }
 

driver/src/main/java/eu/cloudnetservice/driver/network/chunk/defaults/splitter/NetworkChannelsPacketSplitter.java

@@ -37,18 +37,19 @@ public record NetworkChannelsPacketSplitter(@NonNull Collection<NetworkChannel>
    */
   @Override
   public void accept(@NonNull Packet packet) {
-    // disable releasing of the content as we need to content multiple times
-    packet.content().disableReleasing();
     // write to all channels
     for (var channel : this.channels) {
-      // mark the current indexes of the packet
-      packet.content().startTransaction();
-      // write the packet to the channel
-      channel.sendPacketSync(packet);
-      // redo the transaction to allow further writes
-      packet.content().redoTransaction();
+      try {
+        // acquire the packet content and start a transaction, then send the data
+        packet.content().acquire().startTransaction();
+        channel.sendPacketSync(packet);
+      } finally {
+        // redo the transaction to allow further writes
+        packet.content().redoTransaction();
+      }
     }
-    // enable the releasing and free the memory
-    packet.content().enableReleasing().release();
+
+    // release the packet content
+    packet.content().release();
   }
 }

driver/src/main/java/eu/cloudnetservice/driver/network/netty/NettyNetworkHandler.java

@@ -109,16 +109,23 @@ protected void doHandlePacket(@NonNull BasePacket packet) {
       if (uuid != null) {
         var task = this.channel.queryPacketManager().waitingHandler(uuid);
         if (task != null) {
+          // complete the waiting task
           task.complete(packet);
+
           // don't post a query response packet to another handler at all
+          // the packet might be inbound - we might be expected to respond
           return;
         }
       }
 
-      // check if we're allowed to handle the packet
-      if (this.channel.handler().handlePacketReceive(this.channel, packet)) {
-        this.channel.packetRegistry().handlePacket(this.channel, packet);
+      // check if any handler can handle the incoming packet
+      if (this.channel.handler().handlePacketReceive(this.channel, packet)
+        && this.channel.packetRegistry().handlePacket(this.channel, packet)) {
+        return;
       }
+
+      // release the packet content now, there are no handlers that are accepting the message
+      packet.content().forceRelease();
     } catch (Exception exception) {
       LOGGER.severe("Exception whilst handling packet %s", exception, packet);
     }

driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyImmutableDataBuf.java

@@ -36,11 +36,13 @@
 public class NettyImmutableDataBuf implements DataBuf {
 
   protected final Buffer buffer;
-  protected boolean releasable;
+
+  // the amount of times this buffer was acquired
+  protected int acquires = 1;
 
   // transaction offset data
-  protected volatile int readOffset;
-  protected volatile int writeOffset;
+  protected int readOffset;
+  protected int writeOffset;
 
   /**
    * Constructs a new netty immutable data buf instance.
@@ -50,7 +52,6 @@ public class NettyImmutableDataBuf implements DataBuf {
    */
   public NettyImmutableDataBuf(@NonNull Buffer buffer) {
     this.buffer = buffer;
-    this.enableReleasing();
   }
 
   /**
@@ -205,10 +206,8 @@ public <T> T readObject(@NonNull Type type) {
    */
   @Override
   public <T> T readNullable(@NonNull Function<DataBuf, T> readerWhenNonNull, T valueWhenNull) {
-    return this.hotRead(buf -> {
-      var isNonNull = buf.readBoolean();
-      return isNonNull ? readerWhenNonNull.apply(this) : valueWhenNull;
-    });
+    var isNonNull = this.readBoolean();
+    return isNonNull ? readerWhenNonNull.apply(this) : valueWhenNull;
   }
 
   /**
@@ -265,17 +264,16 @@ public boolean accessible() {
    * {@inheritDoc}
    */
   @Override
-  public @NonNull DataBuf disableReleasing() {
-    this.releasable = false;
-    return this;
+  public int acquires() {
+    return this.acquires;
   }
 
   /**
    * {@inheritDoc}
    */
   @Override
-  public @NonNull DataBuf enableReleasing() {
-    this.releasable = true;
+  public @NonNull DataBuf acquire() {
+    this.acquires++;
     return this;
   }
 
@@ -284,7 +282,25 @@ public boolean accessible() {
    */
   @Override
   public void release() {
-    if (this.releasable && this.buffer.isAccessible()) {
+    // release one acquire
+    this.acquires--;
+
+    // check if the buffer is no longer acquired somewhere
+    if (this.acquires <= 0 && this.buffer.isAccessible()) {
+      this.buffer.close();
+    }
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void forceRelease() {
+    // set acquires to 0 to indicate that the buffer was released
+    this.acquires = 0;
+
+    // actually release the buffer if needed
+    if (this.buffer.isAccessible()) {
       this.buffer.close();
     }
   }

driver/src/main/java/eu/cloudnetservice/driver/network/netty/buffer/NettyMutableDataBuf.java

@@ -158,11 +158,14 @@ public NettyMutableDataBuf(@NonNull Buffer buffer) {
   @Override
   public @NonNull DataBuf.Mutable writeDataBuf(@NonNull DataBuf buf) {
     buf.startTransaction();
-    // write the content
-    this.writeInt(buf.readableBytes());
-    this.buffer.writeBytes(((NettyImmutableDataBuf) buf).buffer);
-    // reset the data for later use
-    buf.redoTransaction();
+    try {
+      // write the content
+      this.writeInt(buf.readableBytes());
+      this.buffer.writeBytes(((NettyImmutableDataBuf) buf).buffer);
+    } finally {
+      // reset the data for later use & release the content if it isn't disabled
+      buf.redoTransaction().release();
+    }
 
     return this;
   }

driver/src/main/java/eu/cloudnetservice/driver/network/protocol/PacketListenerRegistry.java

@@ -138,7 +138,8 @@ public interface PacketListenerRegistry {
    *
    * @param channel the channel from which the packet came.
    * @param packet  the packet which was received.
+   * @return true if a listener handled the packet, false otherwise.
    * @throws NullPointerException if either the given channel or packet is null.
    */
-  void handlePacket(@NonNull NetworkChannel channel, @NonNull Packet packet);
+  boolean handlePacket(@NonNull NetworkChannel channel, @NonNull Packet packet);
 }

driver/src/main/java/eu/cloudnetservice/driver/network/protocol/defaults/DefaultPacketListenerRegistry.java

@@ -160,23 +160,29 @@ public void removeListeners() {
    * {@inheritDoc}
    */
   @Override
-  public void handlePacket(@NonNull NetworkChannel channel, @NonNull Packet packet) {
-    if (this.parent != null) {
-      this.parent.handlePacket(channel, packet);
-    }
+  public boolean handlePacket(@NonNull NetworkChannel channel, @NonNull Packet packet) {
+    // check if the parent handler registry is present and can handle the packet
+    var parentDidHandle = this.parent != null && this.parent.handlePacket(channel, packet);
 
+    // get the listeners that are registered in this packet registry for the packet channel
     var registeredListeners = this.listeners.get(packet.channel());
-    if (!registeredListeners.isEmpty()) {
-      for (var listener : registeredListeners) {
-        try {
-          listener.handle(channel, packet);
-        } catch (Exception exception) {
-          throw new IllegalStateException(String.format(
-            "Exception posting packet from channel %d to handler %s",
-            packet.channel(), listener.getClass().getCanonicalName()
-          ), exception);
-        }
+    if (registeredListeners.isEmpty()) {
+      return parentDidHandle;
+    }
+
+    // post the packet to each listener that handles the packet
+    for (var listener : registeredListeners) {
+      try {
+        listener.handle(channel, packet);
+      } catch (Exception exception) {
+        throw new IllegalStateException(String.format(
+          "Exception posting packet from channel %d to handler %s",
+          packet.channel(), listener.getClass().getCanonicalName()
+        ), exception);
       }
     }
+
+    // at least one listener handled the packet
+    return true;
   }
 }

node/src/main/java/eu/cloudnetservice/node/cluster/sync/DefaultDataSyncRegistry.java

@@ -188,8 +188,7 @@ public boolean hasHandler(@NonNull String handlerKey) {
       // no handler for the result
       LOGGER.fine("No handler for key %s to sync data", null, key);
     }
-    // try to release the input buf
-    input.release();
+
     // return the created result
     return result;
   }

node/src/main/java/eu/cloudnetservice/node/command/sub/ClusterCommand.java

@@ -352,16 +352,20 @@ private void pushTemplate(@NonNull CommandSource source, @NonNull ServiceTemplat
       // check if the template really exists in the given storage
       if (inputStream != null) {
         // deploy the template into the cluster
-        this.nodeServerProvider.deployTemplateToCluster(template, inputStream, true)
-          .thenAccept(transferStatus -> {
-            if (transferStatus == TransferStatus.FAILURE) {
-              // the transfer failed
-              source.sendMessage(I18n.trans("command-cluster-push-template-failed", templateName));
-            } else {
-              // the transfer was successful
-              source.sendMessage(I18n.trans("command-cluster-push-template-success", templateName));
+        this.nodeServerProvider.deployTemplateToCluster(template, inputStream, true).whenComplete((status, ex) -> {
+          if (ex != null || status == TransferStatus.FAILURE) {
+            // the transfer failed
+            source.sendMessage(I18n.trans("command-cluster-push-template-failed", templateName));
+
+            // print the detailed exception, if available
+            if (ex != null) {
+              LOGGER.severe("Unable to push template %s to cluster", ex, template);
             }
-          });
+          } else {
+            // the transfer was successful
+            source.sendMessage(I18n.trans("command-cluster-push-template-success", templateName));
+          }
+        });
       } else {
         source.sendMessage(I18n.trans("command-template-not-found", templateName));
       }