Fix more possible leaks

src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionary.java

@@ -50,12 +50,12 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
 	private final Serializer<U> valueSerializer;
 
 	protected DatabaseMapDictionary(LLDictionary dictionary,
-			@Nullable Buffer prefixKey,
+			@Nullable BufSupplier prefixKeySupplier,
 			SerializerFixedBinaryLength<T> keySuffixSerializer,
 			Serializer<U> valueSerializer,
 			Runnable onClose) {
 		// Do not retain or release or use the prefixKey here
-		super(dictionary, prefixKey, keySuffixSerializer, new SubStageGetterSingle<>(valueSerializer), 0, onClose);
+		super(dictionary, prefixKeySupplier, keySuffixSerializer, new SubStageGetterSingle<>(valueSerializer), 0, onClose);
 		this.valueSerializer = valueSerializer;
 	}
 
@@ -67,11 +67,11 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
 	}
 
 	public static <T, U> DatabaseMapDictionary<T, U> tail(LLDictionary dictionary,
-			@Nullable Buffer prefixKey,
+			@Nullable BufSupplier prefixKeySupplier,
 			SerializerFixedBinaryLength<T> keySuffixSerializer,
 			Serializer<U> valueSerializer,
 			Runnable onClose) {
-		return new DatabaseMapDictionary<>(dictionary, prefixKey, keySuffixSerializer, valueSerializer, onClose);
+		return new DatabaseMapDictionary<>(dictionary, prefixKeySupplier, keySuffixSerializer, valueSerializer, onClose);
 	}
 
 	public static <K, V> Flux<Entry<K, V>> getLeavesFrom(DatabaseMapDictionary<K, V> databaseMapDictionary,
@@ -134,17 +134,19 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
 			if (exMessage != null && exMessage.contains("read 0 to 0, write 0 to ")) {
 				var totalZeroBytesErrors = this.totalZeroBytesErrors.incrementAndGet();
 				if (totalZeroBytesErrors < 512 || totalZeroBytesErrors % 10000 == 0) {
-					try (var keySuffixBytes = serializeKeySuffixToKey(keySuffix)) {
+					try (var keyPrefix = keyPrefixSupplier.get()) {
-						LOG.error(
+						try (var keySuffixBytes = serializeKeySuffixToKey(keySuffix)) {
-								"Unexpected zero-bytes value at "
+							LOG.error(
-										+ dictionary.getDatabaseName() + ":" + dictionary.getColumnName()
+									"Unexpected zero-bytes value at "
-										+ ":" + LLUtils.toStringSafe(this.keyPrefix) + ":" + keySuffix
+											+ dictionary.getDatabaseName() + ":" + dictionary.getColumnName()
-										+ "(" + LLUtils.toStringSafe(keySuffixBytes) + ") total=" + totalZeroBytesErrors);
+											+ ":" + LLUtils.toStringSafe(keyPrefix) + ":" + keySuffix
-					} catch (SerializationException e) {
+											+ "(" + LLUtils.toStringSafe(keySuffixBytes) + ") total=" + totalZeroBytesErrors);
-						LOG.error(
+						} catch (SerializationException e) {
-								"Unexpected zero-bytes value at " + dictionary.getDatabaseName() + ":" + dictionary.getColumnName()
+							LOG.error(
-										+ ":" + LLUtils.toStringSafe(this.keyPrefix) + ":" + keySuffix + "(?) total="
+									"Unexpected zero-bytes value at " + dictionary.getDatabaseName() + ":" + dictionary.getColumnName()
-										+ totalZeroBytesErrors);
+											+ ":" + LLUtils.toStringSafe(keyPrefix) + ":" + keySuffix + "(?) total="
+											+ totalZeroBytesErrors);
+						}
 					}
 				}
 				return null;
@@ -169,8 +171,8 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
 
 	private Buffer serializeKeySuffixToKey(T keySuffix) throws SerializationException {
 		Buffer keyBuf;
-		if (keyPrefix != null) {
+		if (keyPrefixSupplier != null) {
-			keyBuf = keyPrefix.copy();
+			keyBuf = keyPrefixSupplier.get();
 		} else {
 			keyBuf = this.dictionary.getAllocator().allocate(keyPrefixLength + keySuffixLength + keyExtLength);
 		}
@@ -188,11 +190,8 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
 
 	private Buffer toKey(Buffer suffixKey) {
 		assert suffixKeyLengthConsistency(suffixKey.readableBytes());
-		if (keyPrefix != null && keyPrefix.readableBytes() > 0) {
+		if (keyPrefixSupplier != null) {
-			var result = LLUtils.compositeBuffer(dictionary.getAllocator(),
+			var result = LLUtils.compositeBuffer(dictionary.getAllocator(), keyPrefixSupplier.get().send(), suffixKey.send());
-					LLUtils.copy(dictionary.getAllocator(), keyPrefix),
-					suffixKey.send()
-			);
 			try {
 				assert result.readableBytes() == keyPrefixLength + keySuffixLength + keyExtLength;
 				return result;
@@ -489,9 +488,8 @@ public class DatabaseMapDictionary<T, U> extends DatabaseMapDictionaryDeep<T, U,
 		if (key == null) {
 			return null;
 		}
-		var keyWithoutExtBuf = keyPrefix == null ? alloc.allocate(keySuffixLength + keyExtLength)
+		var keyWithoutExtBuf =
-				// todo: use a read-only copy
+				keyPrefixSupplier == null ? alloc.allocate(keySuffixLength + keyExtLength) : keyPrefixSupplier.get();
-				: keyPrefix.copy();
 		try {
 			keyWithoutExtBuf.ensureWritable(keySuffixLength + keyExtLength);
 			serializeSuffix(key, keyWithoutExtBuf);

src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryDeep.java

@@ -6,10 +6,10 @@ import io.netty5.buffer.api.DefaultBufferAllocators;
 import io.netty5.buffer.api.Drop;
 import io.netty5.buffer.api.Owned;
 import io.netty5.buffer.api.Resource;
-import io.netty5.buffer.api.Send;
 import io.netty5.buffer.api.internal.ResourceSupport;
 import it.cavallium.dbengine.client.BadBlock;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
 import it.cavallium.dbengine.database.LLDictionaryResultType;
 import it.cavallium.dbengine.database.LLRange;
@@ -24,7 +24,6 @@ import it.cavallium.dbengine.database.serialization.Serializer;
 import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
 import it.unimi.dsi.fastutil.objects.Object2ObjectSortedMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.Set;
@@ -56,19 +55,12 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 				LOG.error("Failed to close range", ex);
 			}
 			try {
-				if (obj.keyPrefix != null) {
+				if (obj.keyPrefixSupplier != null) {
-					obj.keyPrefix.close();
+					obj.keyPrefixSupplier.close();
 				}
 			} catch (Throwable ex) {
 				LOG.error("Failed to close keyPrefix", ex);
 			}
-			try {
-				if (obj.keySuffixAndExtZeroBuffer != null) {
-					obj.keySuffixAndExtZeroBuffer.close();
-				}
-			} catch (Throwable ex) {
-				LOG.error("Failed to close keySuffixAndExtZeroBuffer", ex);
-			}
 			try {
 				if (obj.onClose != null) {
 					obj.onClose.run();
@@ -100,8 +92,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 	protected final Mono<LLRange> rangeMono;
 
 	protected RangeSupplier rangeSupplier;
-	protected Buffer keyPrefix;
+	protected BufSupplier keyPrefixSupplier;
-	protected Buffer keySuffixAndExtZeroBuffer;
 	protected Runnable onClose;
 
 	private static void incrementPrefix(Buffer prefix, int prefixLength) {
@@ -199,73 +190,70 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 	}
 
 	public static <T, U, US extends DatabaseStage<U>> DatabaseMapDictionaryDeep<T, U, US> deepIntermediate(
-			LLDictionary dictionary, Buffer prefixKey, SerializerFixedBinaryLength<T> keySuffixSerializer,
+			LLDictionary dictionary, BufSupplier prefixKey, SerializerFixedBinaryLength<T> keySuffixSerializer,
 			SubStageGetter<U, US> subStageGetter, int keyExtLength, Runnable onClose) {
 		return new DatabaseMapDictionaryDeep<>(dictionary, prefixKey, keySuffixSerializer, subStageGetter,
 				keyExtLength, onClose);
 	}
 
 	@SuppressWarnings({"unchecked", "rawtypes"})
-	protected DatabaseMapDictionaryDeep(LLDictionary dictionary, @Nullable Buffer prefixKey,
+	protected DatabaseMapDictionaryDeep(LLDictionary dictionary, @Nullable BufSupplier prefixKeySupplier,
 			SerializerFixedBinaryLength<T> keySuffixSerializer, SubStageGetter<U, US> subStageGetter, int keyExtLength,
 			Runnable onClose) {
 		super((Drop<DatabaseMapDictionaryDeep<T, U, US>>) (Drop) DROP);
-		try {
+		try (var prefixKey = prefixKeySupplier != null ? prefixKeySupplier.get() : null) {
 			this.dictionary = dictionary;
 			this.alloc = dictionary.getAllocator();
 			this.subStageGetter = subStageGetter;
 			this.keySuffixSerializer = keySuffixSerializer;
-			assert prefixKey == null || prefixKey.isAccessible();
+			this.keyPrefixLength = prefixKey != null ? prefixKey.readableBytes() : 0;
-			this.keyPrefixLength = prefixKey == null ? 0 : prefixKey.readableBytes();
 			this.keySuffixLength = keySuffixSerializer.getSerializedBinaryLength();
 			this.keyExtLength = keyExtLength;
-			this.keySuffixAndExtZeroBuffer = alloc
+			try (var keySuffixAndExtZeroBuffer = alloc
 					.allocate(keySuffixLength + keyExtLength)
 					.fill((byte) 0)
 					.writerOffset(keySuffixLength + keyExtLength)
-					.makeReadOnly();
+					.makeReadOnly()) {
-			assert keySuffixAndExtZeroBuffer.readableBytes() == keySuffixLength + keyExtLength :
+				assert keySuffixAndExtZeroBuffer.readableBytes() == keySuffixLength + keyExtLength :
-					"Key suffix and ext zero buffer readable length is not equal"
+						"Key suffix and ext zero buffer readable length is not equal"
-							+ " to the key suffix length + key ext length. keySuffixAndExtZeroBuffer="
+								+ " to the key suffix length + key ext length. keySuffixAndExtZeroBuffer="
-							+ keySuffixAndExtZeroBuffer.readableBytes() + " keySuffixLength=" + keySuffixLength + " keyExtLength="
+								+ keySuffixAndExtZeroBuffer.readableBytes() + " keySuffixLength=" + keySuffixLength + " keyExtLength="
-							+ keyExtLength;
+								+ keyExtLength;
-			assert keySuffixAndExtZeroBuffer.readableBytes() > 0;
+				assert keySuffixAndExtZeroBuffer.readableBytes() > 0;
-			var firstKey = prefixKey == null ? alloc.allocate(keyPrefixLength + keySuffixLength + keyExtLength)
+				var firstKey = prefixKey != null ? prefixKeySupplier.get()
-					: prefixKey.copy();
+						: alloc.allocate(keyPrefixLength + keySuffixLength + keyExtLength);
-			try {
-				firstRangeKey(firstKey, keyPrefixLength, keySuffixAndExtZeroBuffer);
-				var nextRangeKey = prefixKey == null ? alloc.allocate(keyPrefixLength + keySuffixLength + keyExtLength)
-						: prefixKey.copy();
 				try {
-					nextRangeKey(nextRangeKey, keyPrefixLength, keySuffixAndExtZeroBuffer);
+					firstRangeKey(firstKey, keyPrefixLength, keySuffixAndExtZeroBuffer);
-					assert prefixKey == null || prefixKey.isAccessible();
+					var nextRangeKey = prefixKey != null ? prefixKeySupplier.get()
-					assert keyPrefixLength == 0 || !LLUtils.equals(firstKey, nextRangeKey);
+							: alloc.allocate(keyPrefixLength + keySuffixLength + keyExtLength);
-					if (keyPrefixLength == 0) {
+					try {
-						this.rangeSupplier = RangeSupplier.ofOwned(LLRange.all());
+						nextRangeKey(nextRangeKey, keyPrefixLength, keySuffixAndExtZeroBuffer);
-						firstKey.close();
+						assert prefixKey == null || prefixKey.isAccessible();
+						assert keyPrefixLength == 0 || !LLUtils.equals(firstKey, nextRangeKey);
+						if (keyPrefixLength == 0) {
+							this.rangeSupplier = RangeSupplier.ofOwned(LLRange.all());
+							firstKey.close();
+							nextRangeKey.close();
+						} else {
+							this.rangeSupplier = RangeSupplier.ofOwned(LLRange.ofUnsafe(firstKey, nextRangeKey));
+						}
+						this.rangeMono = Mono.fromSupplier(rangeSupplier);
+						assert subStageKeysConsistency(keyPrefixLength + keySuffixLength + keyExtLength);
+					} catch (Throwable t) {
 						nextRangeKey.close();
-					} else {
+						throw t;
-						this.rangeSupplier = RangeSupplier.ofOwned(LLRange.ofUnsafe(firstKey, nextRangeKey));
 					}
-					this.rangeMono = Mono.fromSupplier(rangeSupplier);
-					assert subStageKeysConsistency(keyPrefixLength + keySuffixLength + keyExtLength);
 				} catch (Throwable t) {
-					nextRangeKey.close();
+					firstKey.close();
 					throw t;
 				}
-			} catch (Throwable t) {
-				firstKey.close();
-				throw t;
-			}
 
-			this.keyPrefix = prefixKey;
+				this.keyPrefixSupplier = prefixKeySupplier;
-			this.onClose = onClose;
+				this.onClose = onClose;
-		} catch (Throwable t) {
-			if (this.keySuffixAndExtZeroBuffer != null && keySuffixAndExtZeroBuffer.isAccessible()) {
-				keySuffixAndExtZeroBuffer.close();
 			}
-			if (prefixKey != null && prefixKey.isAccessible()) {
+		} catch (Throwable t) {
-				prefixKey.close();
+			if (prefixKeySupplier != null) {
+				prefixKeySupplier.close();
 			}
 			throw t;
 		}
@@ -281,8 +269,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 			int keyExtLength,
 			Mono<LLRange> rangeMono,
 			RangeSupplier rangeSupplier,
-			Buffer keyPrefix,
+			BufSupplier keyPrefixSupplier,
-			Buffer keySuffixAndExtZeroBuffer,
 			Runnable onClose) {
 		super((Drop<DatabaseMapDictionaryDeep<T,U,US>>) (Drop) DROP);
 		this.dictionary = dictionary;
@@ -295,8 +282,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 		this.rangeMono = rangeMono;
 
 		this.rangeSupplier = rangeSupplier;
-		this.keyPrefix = keyPrefix;
+		this.keyPrefixSupplier = keyPrefixSupplier;
-		this.keySuffixAndExtZeroBuffer = keySuffixAndExtZeroBuffer;
 		this.onClose = onClose;
 	}
 
@@ -349,15 +335,18 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 	@Override
 	public Mono<US> at(@Nullable CompositeSnapshot snapshot, T keySuffix) {
 		var suffixKeyWithoutExt = Mono.fromCallable(() -> {
-			try (var keyWithoutExtBuf = keyPrefix == null
+			var keyWithoutExtBuf = keyPrefixSupplier == null
-					? alloc.allocate(keySuffixLength + keyExtLength) : keyPrefix.copy()) {
+				? alloc.allocate(keySuffixLength + keyExtLength) : keyPrefixSupplier.get();
+			try {
 				keyWithoutExtBuf.ensureWritable(keySuffixLength + keyExtLength);
 				serializeSuffix(keySuffix, keyWithoutExtBuf);
-				return keyWithoutExtBuf.send();
+			} catch (Throwable ex) {
+				keyWithoutExtBuf.close();
+				throw ex;
 			}
+			return keyWithoutExtBuf;
 		});
-		return this.subStageGetter
+		return this.subStageGetter.subStage(dictionary, snapshot, suffixKeyWithoutExt);
-				.subStage(dictionary, snapshot, suffixKeyWithoutExt);
 	}
 
 	@Override
@@ -374,21 +363,17 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 	public Flux<SubStageEntry<T, US>> getAllStages(@Nullable CompositeSnapshot snapshot, boolean smallRange) {
 		return dictionary
 				.getRangeKeyPrefixes(resolveSnapshot(snapshot), rangeMono, keyPrefixLength + keySuffixLength, smallRange)
-				.flatMapSequential(groupKeyWithoutExtSend_ -> Mono.using(
+				.flatMapSequential(groupKeyWithoutExt -> this.subStageGetter
-						() -> groupKeyWithoutExtSend_,
+						.subStage(dictionary, snapshot, Mono.fromCallable(groupKeyWithoutExt::copy))
-						groupKeyWithoutExtSend -> this.subStageGetter
+						.map(us -> {
-								.subStage(dictionary, snapshot, Mono.fromCallable(() -> groupKeyWithoutExtSend.copy().send()))
+							T deserializedSuffix;
-								.handle((us, sink) -> {
+							try (var splittedGroupSuffix = splitGroupSuffix(groupKeyWithoutExt)) {
-									T deserializedSuffix;
+								deserializedSuffix = this.deserializeSuffix(splittedGroupSuffix);
-									try (var splittedGroupSuffix = splitGroupSuffix(groupKeyWithoutExtSend)) {
+								return new SubStageEntry<>(deserializedSuffix, us);
-										deserializedSuffix = this.deserializeSuffix(splittedGroupSuffix);
+							}
-										sink.next(new SubStageEntry<>(deserializedSuffix, us));
+						})
-									} catch (SerializationException ex) {
+						.doFinally(s -> groupKeyWithoutExt.close())
-										sink.error(ex);
+				);
-									}
-								}),
-						Resource::close
-				));
 	}
 
 	/**
@@ -447,7 +432,6 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 	protected T deserializeSuffix(@NotNull Buffer keySuffix) throws SerializationException {
 		assert suffixKeyLengthConsistency(keySuffix.readableBytes());
 		var result = keySuffixSerializer.deserialize(keySuffix);
-		assert keyPrefix == null || keyPrefix.isAccessible();
 		return result;
 	}
 
@@ -457,7 +441,6 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 		keySuffixSerializer.serialize(keySuffix, output);
 		var afterWriterOffset = output.writerOffset();
 		assert suffixKeyLengthConsistency(afterWriterOffset - beforeWriterOffset);
-		assert keyPrefix == null || keyPrefix.isAccessible();
 	}
 
 	@Override
@@ -467,8 +450,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 
 	@Override
 	protected Owned<DatabaseMapDictionaryDeep<T, U, US>> prepareSend() {
-		var keyPrefix = this.keyPrefix.send();
+		var keyPrefixSupplier = this.keyPrefixSupplier;
-		var keySuffixAndExtZeroBuffer = this.keySuffixAndExtZeroBuffer.send();
 		var rangeSupplier = this.rangeSupplier;
 		var onClose = this.onClose;
 		return drop -> {
@@ -481,8 +463,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 					keyExtLength,
 					rangeMono,
 					rangeSupplier,
-					keyPrefix.receive(),
+					keyPrefixSupplier,
-					keySuffixAndExtZeroBuffer.receive(),
 					onClose
 			);
 			drop.attach(instance);
@@ -492,8 +473,7 @@ public class DatabaseMapDictionaryDeep<T, U, US extends DatabaseStage<U>> extend
 
 	@Override
 	protected void makeInaccessible() {
-		this.keyPrefix = null;
+		this.keyPrefixSupplier = null;
-		this.keySuffixAndExtZeroBuffer = null;
 		this.rangeSupplier = null;
 		this.onClose = null;
 	}

src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapDictionaryHashed.java

@@ -7,6 +7,7 @@ import io.netty5.buffer.api.Owned;
 import io.netty5.buffer.api.Send;
 import it.cavallium.dbengine.client.BadBlock;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
 import it.cavallium.dbengine.database.LLUtils;
 import io.netty5.buffer.api.internal.ResourceSupport;
@@ -67,7 +68,7 @@ public class DatabaseMapDictionaryHashed<T, U, TH> extends
 
 	@SuppressWarnings({"unchecked", "rawtypes"})
 	protected DatabaseMapDictionaryHashed(LLDictionary dictionary,
-			@Nullable Buffer prefixKey,
+			@Nullable BufSupplier prefixKeySupplier,
 			Serializer<T> keySuffixSerializer,
 			Serializer<U> valueSerializer,
 			Function<T, TH> keySuffixHashFunction,
@@ -82,7 +83,7 @@ public class DatabaseMapDictionaryHashed<T, U, TH> extends
 				= new ValueWithHashSerializer<>(keySuffixSerializer, valueSerializer);
 		ValuesSetSerializer<Entry<T, U>> valuesSetSerializer
 				= new ValuesSetSerializer<>(valueWithHashSerializer);
-		this.subDictionary = DatabaseMapDictionary.tail(dictionary, prefixKey, keySuffixHashSerializer,
+		this.subDictionary = DatabaseMapDictionary.tail(dictionary, prefixKeySupplier, keySuffixHashSerializer,
 				valuesSetSerializer, onClose);
 		this.keySuffixHashFunction = keySuffixHashFunction;
 	}
@@ -117,14 +118,14 @@ public class DatabaseMapDictionaryHashed<T, U, TH> extends
 	}
 
 	public static <T, U, UH> DatabaseMapDictionaryHashed<T, U, UH> tail(LLDictionary dictionary,
-			@Nullable Buffer prefixKey,
+			@Nullable BufSupplier prefixKeySupplier,
 			Serializer<T> keySuffixSerializer,
 			Serializer<U> valueSerializer,
 			Function<T, UH> keySuffixHashFunction,
 			SerializerFixedBinaryLength<UH> keySuffixHashSerializer,
 			Runnable onClose) {
 		return new DatabaseMapDictionaryHashed<>(dictionary,
-				prefixKey,
+				prefixKeySupplier,
 				keySuffixSerializer,
 				valueSerializer,
 				keySuffixHashFunction,

src/main/java/it/cavallium/dbengine/database/collections/DatabaseMapSingle.java

@@ -80,13 +80,9 @@ public class DatabaseMapSingle<U> extends ResourceSupport<DatabaseStage<U>, Data
 		}
 	}
 
-	private void deserializeValue(Buffer value, SynchronousSink<U> sink) {
+	private U deserializeValue(Buffer value) {
 		try {
-			U deserializedValue;
+			return serializer.deserialize(value);
-			try (value) {
-				deserializedValue = serializer.deserialize(value);
-			}
-			sink.next(deserializedValue);
 		} catch (IndexOutOfBoundsException ex) {
 			var exMessage = ex.getMessage();
 			if (exMessage != null && exMessage.contains("read 0 to 0, write 0 to ")) {
@@ -94,12 +90,12 @@ public class DatabaseMapSingle<U> extends ResourceSupport<DatabaseStage<U>, Data
 					LOG.error("Unexpected zero-bytes value at "
 							+ dictionary.getDatabaseName() + ":" + dictionary.getColumnName() + ":" + LLUtils.toStringSafe(key));
 				}
-				sink.complete();
+				return null;
 			} else {
-				sink.error(ex);
+				throw ex;
 			}
 		} catch (SerializationException ex) {
-			sink.error(ex);
+			throw ex;
 		}
 	}
 
@@ -118,22 +114,24 @@ public class DatabaseMapSingle<U> extends ResourceSupport<DatabaseStage<U>, Data
 
 	@Override
 	public Mono<U> get(@Nullable CompositeSnapshot snapshot, boolean existsAlmostCertainly) {
-		return dictionary
+		return Mono.usingWhen(dictionary.get(resolveSnapshot(snapshot), keyMono),
-				.get(resolveSnapshot(snapshot), keyMono)
+				buf -> Mono.fromSupplier(() -> deserializeValue(buf)),
-				.handle(this::deserializeValue);
+				buf -> Mono.fromRunnable(buf::close)
+		);
 	}
 
 	@Override
 	public Mono<U> setAndGetPrevious(U value) {
-		return dictionary
+		return Mono.usingWhen(dictionary
-				.put(keyMono, Mono.fromCallable(() -> serializeValue(value)), LLDictionaryResultType.PREVIOUS_VALUE)
+						.put(keyMono, Mono.fromCallable(() -> serializeValue(value)), LLDictionaryResultType.PREVIOUS_VALUE),
-				.handle(this::deserializeValue);
+				buf -> Mono.fromSupplier(() -> deserializeValue(buf)),
+				buf -> Mono.fromRunnable(buf::close));
 	}
 
 	@Override
 	public Mono<U> update(SerializationFunction<@Nullable U, @Nullable U> updater,
 			UpdateReturnMode updateReturnMode) {
-		return dictionary
+		var resultMono = dictionary
 				.update(keyMono, (oldValueSer) -> {
 					U result;
 					if (oldValueSer == null) {
@@ -147,8 +145,11 @@ public class DatabaseMapSingle<U> extends ResourceSupport<DatabaseStage<U>, Data
 					} else {
 						return serializeValue(result);
 					}
-				}, updateReturnMode)
+				}, updateReturnMode);
-				.handle(this::deserializeValue);
+		return Mono.usingWhen(resultMono,
+				result -> Mono.fromSupplier(() -> deserializeValue(result)),
+				result -> Mono.fromRunnable(result::close)
+		);
 	}
 
 	@Override
@@ -172,9 +173,10 @@ public class DatabaseMapSingle<U> extends ResourceSupport<DatabaseStage<U>, Data
 
 	@Override
 	public Mono<U> clearAndGetPrevious() {
-		return dictionary
+		return Mono.usingWhen(dictionary.remove(keyMono, LLDictionaryResultType.PREVIOUS_VALUE),
-				.remove(keyMono, LLDictionaryResultType.PREVIOUS_VALUE)
+				result -> Mono.fromSupplier(() -> deserializeValue(result)),
-				.handle(this::deserializeValue);
+				result -> Mono.fromRunnable(result::close)
+		);
 	}
 
 	@Override

src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionary.java

@@ -4,6 +4,7 @@ import io.netty5.buffer.api.Buffer;
 import io.netty5.buffer.api.Drop;
 import io.netty5.buffer.api.Send;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
 import it.cavallium.dbengine.database.LLUtils;
 import it.cavallium.dbengine.database.collections.DatabaseEmpty.Nothing;
@@ -19,10 +20,15 @@ import reactor.core.publisher.Mono;
 public class DatabaseSetDictionary<T> extends DatabaseMapDictionary<T, Nothing> {
 
 	protected DatabaseSetDictionary(LLDictionary dictionary,
-			Buffer prefixKey,
+			BufSupplier prefixKeySupplier,
 			SerializerFixedBinaryLength<T> keySuffixSerializer,
 			Runnable onClose) {
-		super(dictionary, prefixKey, keySuffixSerializer, DatabaseEmpty.nothingSerializer(dictionary.getAllocator()), onClose);
+		super(dictionary,
+				prefixKeySupplier,
+				keySuffixSerializer,
+				DatabaseEmpty.nothingSerializer(dictionary.getAllocator()),
+				onClose
+		);
 	}
 
 	public static <T> DatabaseSetDictionary<T> simple(LLDictionary dictionary,
@@ -32,10 +38,10 @@ public class DatabaseSetDictionary<T> extends DatabaseMapDictionary<T, Nothing>
 	}
 
 	public static <T> DatabaseSetDictionary<T> tail(LLDictionary dictionary,
-			Buffer prefixKey,
+			BufSupplier prefixKeySupplier,
 			SerializerFixedBinaryLength<T> keySuffixSerializer,
 			Runnable onClose) {
-		return new DatabaseSetDictionary<>(dictionary, prefixKey, keySuffixSerializer, onClose);
+		return new DatabaseSetDictionary<>(dictionary, prefixKeySupplier, keySuffixSerializer, onClose);
 	}
 
 	public Mono<Set<T>> getKeySet(@Nullable CompositeSnapshot snapshot) {

src/main/java/it/cavallium/dbengine/database/collections/DatabaseSetDictionaryHashed.java

@@ -4,6 +4,7 @@ import io.netty5.buffer.api.Buffer;
 import io.netty5.buffer.api.Drop;
 import io.netty5.buffer.api.Send;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
 import it.cavallium.dbengine.database.LLUtils;
 import it.cavallium.dbengine.database.collections.DatabaseEmpty.Nothing;
@@ -22,13 +23,13 @@ import reactor.core.publisher.Mono;
 public class DatabaseSetDictionaryHashed<T, TH> extends DatabaseMapDictionaryHashed<T, Nothing, TH> {
 
 	protected DatabaseSetDictionaryHashed(LLDictionary dictionary,
-			@Nullable Buffer prefixKey,
+			@Nullable BufSupplier prefixKeySupplier,
 			Serializer<T> keySuffixSerializer,
 			Function<T, TH> keySuffixHashFunction,
 			SerializerFixedBinaryLength<TH> keySuffixHashSerializer,
 			Runnable onClose) {
 		super(dictionary,
-				prefixKey,
+				prefixKeySupplier,
 				keySuffixSerializer,
 				DatabaseEmpty.nothingSerializer(dictionary.getAllocator()),
 				keySuffixHashFunction,
@@ -52,13 +53,13 @@ public class DatabaseSetDictionaryHashed<T, TH> extends DatabaseMapDictionaryHas
 	}
 
 	public static <T, TH> DatabaseSetDictionaryHashed<T, TH> tail(LLDictionary dictionary,
-			@Nullable Buffer prefixKey,
+			@Nullable BufSupplier prefixKeySupplier,
 			Serializer<T> keySuffixSerializer,
 			Function<T, TH> keyHashFunction,
 			SerializerFixedBinaryLength<TH> keyHashSerializer,
 			Runnable onClose) {
 		return new DatabaseSetDictionaryHashed<>(dictionary,
-				prefixKey,
+				prefixKeySupplier,
 				keySuffixSerializer,
 				keyHashFunction,
 				keyHashSerializer,

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetter.java

@@ -11,6 +11,6 @@ public interface SubStageGetter<U, US extends DatabaseStage<U>> {
 
 	Mono<US> subStage(LLDictionary dictionary,
 			@Nullable CompositeSnapshot snapshot,
-			Mono<Send<Buffer>> prefixKey);
+			Mono<Buffer> prefixKey);
 
 }

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashMap.java

@@ -4,6 +4,7 @@ import io.netty5.buffer.api.Buffer;
 import io.netty5.buffer.api.Resource;
 import io.netty5.buffer.api.Send;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
 import it.cavallium.dbengine.database.LLUtils;
 import it.cavallium.dbengine.database.serialization.Serializer;
@@ -36,18 +37,15 @@ public class SubStageGetterHashMap<T, U, TH> implements
 	@Override
 	public Mono<DatabaseMapDictionaryHashed<T, U, TH>> subStage(LLDictionary dictionary,
 			@Nullable CompositeSnapshot snapshot,
-			Mono<Send<Buffer>> prefixKeyMono) {
+			Mono<Buffer> prefixKeyMono) {
-		return prefixKeyMono.map(prefixKeyToReceive -> {
+		return prefixKeyMono.map(prefixKey -> DatabaseMapDictionaryHashed.tail(dictionary,
-			var prefixKey = prefixKeyToReceive.receive();
+				BufSupplier.ofOwned(prefixKey),
-			return DatabaseMapDictionaryHashed.tail(dictionary,
+				keySerializer,
-					prefixKey,
+				valueSerializer,
-					keySerializer,
+				keyHashFunction,
-					valueSerializer,
+				keyHashSerializer,
-					keyHashFunction,
+				null
-					keyHashSerializer,
+		));
-					null
-			);
-		});
 	}
 
 	public int getKeyHashBinaryLength() {

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterHashSet.java

@@ -4,6 +4,7 @@ import io.netty5.buffer.api.Buffer;
 import io.netty5.buffer.api.Resource;
 import io.netty5.buffer.api.Send;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
 import it.cavallium.dbengine.database.LLUtils;
 import it.cavallium.dbengine.database.collections.DatabaseEmpty.Nothing;
@@ -34,17 +35,14 @@ public class SubStageGetterHashSet<T, TH> implements
 	@Override
 	public Mono<DatabaseSetDictionaryHashed<T, TH>> subStage(LLDictionary dictionary,
 			@Nullable CompositeSnapshot snapshot,
-			Mono<Send<Buffer>> prefixKeyMono) {
+			Mono<Buffer> prefixKeyMono) {
-		return prefixKeyMono.map(prefixKeyToReceive -> {
+		return prefixKeyMono.map(prefixKey -> DatabaseSetDictionaryHashed.tail(dictionary,
-			var prefixKey = prefixKeyToReceive.receive();
+				BufSupplier.ofOwned(prefixKey),
-			return DatabaseSetDictionaryHashed.tail(dictionary,
+				keySerializer,
-					prefixKey,
+				keyHashFunction,
-					keySerializer,
+				keyHashSerializer,
-					keyHashFunction,
+				null
-					keyHashSerializer,
+		));
-					null
-			);
-		});
 	}
 
 	public int getKeyHashBinaryLength() {

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMap.java

@@ -1,15 +1,12 @@
 package it.cavallium.dbengine.database.collections;
 
 import io.netty5.buffer.api.Buffer;
-import io.netty5.buffer.api.Resource;
-import io.netty5.buffer.api.Send;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
-import it.cavallium.dbengine.database.LLUtils;
 import it.cavallium.dbengine.database.serialization.Serializer;
 import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
 import it.unimi.dsi.fastutil.objects.Object2ObjectSortedMap;
-import java.util.Map;
 import org.jetbrains.annotations.Nullable;
 import reactor.core.publisher.Mono;
 
@@ -28,11 +25,13 @@ public class SubStageGetterMap<T, U> implements
 	@Override
 	public Mono<DatabaseMapDictionary<T, U>> subStage(LLDictionary dictionary,
 			@Nullable CompositeSnapshot snapshot,
-			Mono<Send<Buffer>> prefixKeyMono) {
+			Mono<Buffer> prefixKeyMono) {
-		return prefixKeyMono.map(prefixKeyToReceive -> {
+		return prefixKeyMono.map(prefixKey -> DatabaseMapDictionary.tail(dictionary,
-			var prefixKey = prefixKeyToReceive.receive();
+				BufSupplier.ofOwned(prefixKey),
-			return DatabaseMapDictionary.tail(dictionary, prefixKey, keySerializer, valueSerializer, null);
+				keySerializer,
-		});
+				valueSerializer,
+				null
+		));
 	}
 
 	public int getKeyBinaryLength() {

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterMapDeep.java

@@ -4,6 +4,7 @@ import io.netty5.buffer.api.Buffer;
 import io.netty5.buffer.api.Resource;
 import io.netty5.buffer.api.Send;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
 import it.cavallium.dbengine.database.LLUtils;
 import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
@@ -42,17 +43,14 @@ public class SubStageGetterMapDeep<T, U, US extends DatabaseStage<U>> implements
 	@Override
 	public Mono<DatabaseMapDictionaryDeep<T, U, US>> subStage(LLDictionary dictionary,
 			@Nullable CompositeSnapshot snapshot,
-			Mono<Send<Buffer>> prefixKeyMono) {
+			Mono<Buffer> prefixKeyMono) {
-		return prefixKeyMono.map(prefixKeyToReceive -> {
+		return prefixKeyMono.map(prefixKey -> DatabaseMapDictionaryDeep.deepIntermediate(dictionary,
-			var prefixKey = prefixKeyToReceive.receive();
+				BufSupplier.ofOwned(prefixKey),
-			return DatabaseMapDictionaryDeep.deepIntermediate(dictionary,
+				keySerializer,
-					prefixKey,
+				subStageGetter,
-					keySerializer,
+				keyExtLength,
-					subStageGetter,
+				null
-					keyExtLength,
+		));
-					null
-			);
-		});
 	}
 
 	public int getKeyBinaryLength() {

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSet.java

@@ -4,6 +4,7 @@ import io.netty5.buffer.api.Buffer;
 import io.netty5.buffer.api.Resource;
 import io.netty5.buffer.api.Send;
 import it.cavallium.dbengine.client.CompositeSnapshot;
+import it.cavallium.dbengine.database.BufSupplier;
 import it.cavallium.dbengine.database.LLDictionary;
 import it.cavallium.dbengine.database.collections.DatabaseEmpty.Nothing;
 import it.cavallium.dbengine.database.serialization.SerializerFixedBinaryLength;
@@ -24,11 +25,12 @@ public class SubStageGetterSet<T> implements
 	@Override
 	public Mono<DatabaseSetDictionary<T>> subStage(LLDictionary dictionary,
 			@Nullable CompositeSnapshot snapshot,
-			Mono<Send<Buffer>> prefixKeyMono) {
+			Mono<Buffer> prefixKeyMono) {
-		return prefixKeyMono.map(prefixKeyToReceive -> {
+		return prefixKeyMono.map(prefixKey -> DatabaseSetDictionary.tail(dictionary,
-			var prefixKey = prefixKeyToReceive.receive();
+				BufSupplier.ofOwned(prefixKey),
-			return DatabaseSetDictionary.tail(dictionary, prefixKey, keySerializer, null);
+				keySerializer,
-		});
+				null
+		));
 	}
 
 	public int getKeyBinaryLength() {

src/main/java/it/cavallium/dbengine/database/collections/SubStageGetterSingle.java

@@ -20,8 +20,12 @@ public class SubStageGetterSingle<T> implements SubStageGetter<T, DatabaseStageE
 	@Override
 	public Mono<DatabaseStageEntry<T>> subStage(LLDictionary dictionary,
 			@Nullable CompositeSnapshot snapshot,
-			Mono<Send<Buffer>> keyPrefixMono) {
+			Mono<Buffer> keyPrefixMono) {
-		return keyPrefixMono.map(keyPrefix -> new DatabaseMapSingle<>(dictionary, BufSupplier.of(keyPrefix), serializer, null));
+		return keyPrefixMono.map(keyPrefix -> new DatabaseMapSingle<>(dictionary,
+				BufSupplier.ofOwned(keyPrefix),
+				serializer,
+				null
+		));
 	}
 
 }

src/main/java/it/cavallium/dbengine/database/disk/LLLocalDictionary.java

@@ -60,6 +60,7 @@ import org.rocksdb.Slice;
 import org.rocksdb.Snapshot;
 import org.rocksdb.WriteBatch;
 import org.rocksdb.WriteOptions;
+import reactor.core.Exceptions;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
@@ -772,13 +773,7 @@ public class LLLocalDictionary implements LLDictionary {
 	}
 
 	private Flux<LLEntry> getRangeSingle(LLSnapshot snapshot, Mono<Buffer> keyMono) {
-		return Mono
+		return Mono.zip(keyMono, this.get(snapshot, keyMono), LLEntry::of).flux();
-				.zip(keyMono, this.get(snapshot, keyMono))
-				.map(result -> LLEntry.of(
-						result.getT1().touch("get-range-single key"),
-						result.getT2().touch("get-range-single value")
-				))
-				.flux();
 	}
 
 	private Flux<LLEntry> getRangeMulti(LLSnapshot snapshot,