Fix double-release of buffers.

Guard spsc buffer in ReactorNettyClient against multiple concurrent release calls to avoid double releasing.
This has only became visible using Netty 4.2 AdaptivePoolingAllocator.

[#717]

Author: mp911de

src/main/java/io/r2dbc/postgresql/client/ReactorNettyClient.java

@@ -1032,9 +1032,19 @@ public void close(Supplier<? extends Throwable> supplier) {
                 receiver.onError(supplier.get());
             }
 
-            while (!this.buffer.isEmpty()) {
-                ReferenceCountUtil.release(this.buffer.poll());
-            }
+            // Drain the buffer under the single-consumer guard. The buffer is an Spsc queue to avoid double-releasing it (IllegalReferenceCountException).
+            do {
+                if (this.drain.compareAndSet(false, true)) {
+                    try {
+                        BackendMessage message;
+                        while ((message = this.buffer.poll()) != null) {
+                            ReferenceCountUtil.release(message);
+                        }
+                    } finally {
+                        this.drain.compareAndSet(true, false);
+                    }
+                }
+            } while (!this.buffer.isEmpty());
         }
 
     }

src/main/java/io/r2dbc/postgresql/codec/BlobCodec.java

@@ -29,6 +29,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.Collections;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import static io.r2dbc.postgresql.codec.PostgresqlObjectId.BYTEA;
 import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
@@ -87,22 +88,22 @@ public Iterable<PostgresTypeIdentifier> getDataTypes() {
 
     private static final class ByteABlob implements Blob {
 
+        private static final AtomicIntegerFieldUpdater<ByteABlob> RELEASED = AtomicIntegerFieldUpdater.newUpdater(ByteABlob.class, "released");
+
         private final ByteBuf byteBuf;
 
         private final Format format;
 
+        private volatile int released;
+
         private ByteABlob(ByteBuf byteBuf, Format format) {
             this.byteBuf = byteBuf.retain();
             this.format = format;
         }
 
         @Override
         public Mono<Void> discard() {
-            return Mono.fromRunnable(() -> {
-                if (this.byteBuf.refCnt() > 0) {
-                    this.byteBuf.release();
-                }
-            });
+            return Mono.fromRunnable(this::release);
         }
 
         @Override
@@ -113,7 +114,16 @@ public Mono<ByteBuffer> stream() {
                 }
 
                 return ByteBuffer.wrap(AbstractBinaryCodec.decodeFromHex(this.byteBuf));
-            }).doAfterTerminate(this.byteBuf::release);
+            }).doAfterTerminate(this::release);
+        }
+
+        /**
+         * Release the retained buffer exactly once.
+         */
+        private void release() {
+            if (RELEASED.compareAndSet(this, 0, 1)) {
+                this.byteBuf.release();
+            }
         }
 
     }

src/test/java/io/r2dbc/postgresql/codec/BlobCodecUnitTests.java

@@ -65,6 +65,33 @@ void decodeNoByteBuf() {
         assertThat(new BlobCodec(TEST).decode(null, dataType, FORMAT_TEXT, Blob.class)).isNull();
     }
 
+    @Test
+    void streamThenDiscardMustNotOverReleaseSharedBuffer() {
+
+        // The decoded column is a retained slice that shares its refCnt with the frame decoder's cumulation
+        // buffer. Consuming the Blob via stream() and then discard() (as Flux.usingWhen does) must not release
+        // the shared buffer twice, which would corrupt the still-alive cumulation (IllegalReferenceCountException).
+        io.netty.buffer.ByteBuf cumulation = TEST.buffer();
+        cumulation.writeBytes("\\x746573742d76616c7565".getBytes());
+        io.netty.buffer.ByteBuf column = cumulation.readRetainedSlice(cumulation.readableBytes());
+        assertThat(cumulation.refCnt()).isEqualTo(2);
+
+        Blob blob = new BlobCodec(TEST).doDecode(column, BYTEA, FORMAT_TEXT, Blob.class);
+        column.release(); // DataRow.deallocate() releases the column once
+
+        Flux.from(blob.stream()).reduce(TEST.compositeBuffer(), (a, b) -> a.addComponent(true, Unpooled.wrappedBuffer(b)))
+            .as(StepVerifier::create)
+            .expectNextCount(1)
+            .verifyComplete();
+
+        reactor.core.publisher.Mono.from(blob.discard()).block();
+
+        assertThat(cumulation.refCnt())
+            .describedAs("shared cumulation refCnt after stream()+discard()")
+            .isOne();
+        cumulation.release();
+    }
+
     @Test
     void doCanDecode() {
         BlobCodec codec = new BlobCodec(TEST);