Hi, when using exposed with r2dbc pool w/ postgres...

# exposed

[RJ Garcia]

Hi, when using exposed with r2dbc pool w/ postgres driver, i'm seeing weird exceptions related to IndexOutOfBoundsExceptions related to netty buffers. Here's a really simple reproduction:

R2dbcDatabase.connect(
    url = "r2dbc:pool:<postgresql://localhost:5432/test?schema=app&maxSize=1&initialSize=1>",
    databaseConfig = R2dbcDatabaseConfig.Builder().apply {
        connectionFactoryOptions {
            option(ConnectionFactoryOptions.USER, "test")
            option(ConnectionFactoryOptions.PASSWORD, "test")
        }
    }
)

repeat(100) {
    suspendTransaction {
        exec("SELECT 1") { it.get(0) }?.firstOrNull()
    }
},

and here's an example stacktrace:

index: 45, length: 1 (expected: range(0, 0))
java.lang.IndexOutOfBoundsException: index: 45, length: 1 (expected: range(0, 0))
	at io.netty.buffer.AbstractByteBuf.checkRangeBounds(AbstractByteBuf.java:1397)
	at io.netty.buffer.AbstractByteBuf.checkIndex0(AbstractByteBuf.java:1404)
	at io.netty.buffer.AbstractByteBuf.checkIndex(AbstractByteBuf.java:1391)
	at io.netty.buffer.AdaptivePoolingAllocator$AdaptiveByteBuf.getBytes(AdaptivePoolingAllocator.java:1233)
	at io.netty.buffer.ByteBufUtil.decodeString(ByteBufUtil.java:1365)
	at io.netty.buffer.AbstractByteBuf.toString(AbstractByteBuf.java:1253)
	at io.netty.buffer.AbstractByteBuf.getCharSequence(AbstractByteBuf.java:513)
	at io.netty.buffer.AbstractUnpooledSlicedByteBuf.getCharSequence(AbstractUnpooledSlicedByteBuf.java:269)
	at io.netty.buffer.AbstractByteBuf.readCharSequence(AbstractByteBuf.java:518)
	at io.r2dbc.postgresql.util.ByteBufUtils.decode(ByteBufUtils.java:42)
	at io.r2dbc.postgresql.codec.NumericDecodeUtils.decodeNumber(NumericDecodeUtils.java:69)
	at io.r2dbc.postgresql.codec.AbstractNumericCodec.decodeNumber(AbstractNumericCodec.java:139)
	at io.r2dbc.postgresql.codec.IntegerCodec.doDecode(IntegerCodec.java:44)
	at io.r2dbc.postgresql.codec.IntegerCodec.doDecode(IntegerCodec.java:28)
	at io.r2dbc.postgresql.codec.AbstractCodec.decode(AbstractCodec.java:81)
	at io.r2dbc.postgresql.codec.DefaultCodecs.decode(DefaultCodecs.java:221)
	at io.r2dbc.postgresql.PostgresqlRow.decode(PostgresqlRow.java:129)
	at io.r2dbc.postgresql.PostgresqlRow.get(PostgresqlRow.java:86)
	at io.r2dbc.spi.Readable.get(Readable.java:53)
	at com.example.DbTest$queries should not fail$1$2$1.invokeSuspend$lambda$0(DbTest.kt:33)
	at org.jetbrains.exposed.v1.r2dbc.R2dbcTransaction$exec$4.executeInternal$lambda$0(R2dbcTransaction.kt:186)
	at org.jetbrains.exposed.v1.r2dbc.statements.api.R2dbcResult$mapRows$1.invokeSuspend$lambda$3$lambda$1$lambda$0(R2dbcResult.kt:46)
	at org.jetbrains.exposed.v1.r2dbc.transactions.TransactionManagerKt.withThreadLocalTransaction(TransactionManager.kt:557)
	at org.jetbrains.exposed.v1.r2dbc.statements.api.R2dbcResult$mapRows$1.invokeSuspend$lambda$3$lambda$1(R2dbcResult.kt:45)
	at io.r2dbc.postgresql.PostgresqlResult.lambda$map$2(PostgresqlResult.java:129)
	at reactor.core.publisher.FluxHandleFuseable$HandleFuseableSubscriber.onNext(FluxHandleFuseable.java:179)
	at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drainRegular(FluxWindowPredicate.java:670)
	at reactor.core.publisher.FluxWindowPredicate$WindowFlux.drain(FluxWindowPredicate.java:748)
	at reactor.core.publisher.FluxWindowPredicate$WindowFlux.subscribe(FluxWindowPredicate.java:823)
	at reactor.core.publisher.Flux.subscribe(Flux.java:8773)
	at kotlinx.coroutines.reactive.ChannelKt.toChannel(Channel.kt:20)
	at kotlinx.coroutines.reactive.ChannelKt.toChannel$default(Channel.kt:17)
	at org.jetbrains.exposed.v1.r2dbc.statements.api.R2dbcResult$mapRows$1.invokeSuspend(R2dbcResult.kt:145)
	at _COROUTINE._BOUNDARY._(CoroutineDebugging.kt:42)
	at kotlinx.coroutines.channels.BufferedChannel$BufferedChannelIterator.onClosedHasNext(BufferedChannel.kt:1645)
	at kotlinx.coroutines.channels.BufferedChannel$BufferedChannelIterator.hasNext(BufferedChannel.kt:1634)
	at org.jetbrains.exposed.v1.r2dbc.statements.api.R2dbcResult$mapRows$1.invokeSuspend(R2dbcResult.kt:150)
	at org.jetbrains.exposed.v1.r2dbc.transactions.TransactionManagerKt.inTopLevelSuspendTransaction(TransactionManager.kt:437)
	at org.jetbrains.exposed.v1.r2dbc.transactions.TransactionManagerKt.suspendTransaction(TransactionManager.kt:398)
	at com.example.DbTest$queries should not fail$1.invokeSuspend(DbTest.kt:32)
Caused by: java.lang.IndexOutOfBoundsException: index: 45, length: 1 (expected: range(0, 0))
	at io.netty.buffer.AbstractByteBuf.checkRangeBounds(AbstractByteBuf.java:1397)
	at io.netty.buffer.AbstractByteBuf.checkIndex0(AbstractByteBuf.java:1404)
	at io.netty.buffer.AbstractByteBuf.checkIndex(AbstractByteBuf.java:1391)
	at io.netty.buffer.AdaptivePoolingAllocator$AdaptiveByteBuf.getBytes(AdaptivePoolingAllocator.java:1233)
	at io.netty.buffer.ByteBufUtil.decodeString(ByteBufUtil.java:1365)
	at io.netty.buffer.AbstractByteBuf.toString(AbstractByteBuf.java:1253)
	at io.netty.buffer.AbstractByteBuf.getCharSequence(AbstractByteBuf.java:513)
	at io.netty.buffer.AbstractUnpooledSlicedByteBuf.getCharSequence(AbstractUnpooledSlicedByteBuf.java:269)

This is what's included in my gradle deps

implementation("org.jetbrains.exposed:exposed-core:1.0.0-rc-2")
    implementation("org.jetbrains.exposed:exposed-r2dbc:1.0.0-rc-2")
    implementation("org.jetbrains.exposed:exposed-kotlin-datetime:1.0.0-rc-2")
    implementation("org.postgresql:r2dbc-postgresql:1.0.7.RELEASE")
    implementation("io.r2dbc:r2dbc-pool:1.0.2.RELEASE")

[RJ Garcia]

this problem is exacerbated the more times suspendTransaction is run or if you have multiple async coroutines running suspend transaction.

This works fine when not using the pool connection.

[RJ Garcia]

I also confirmed that this is an exposed issue by doing a similar setup directly using the r2dbc apis directly and never experienced this type of an issue.

My guess is something is getting polluted with the single connection across multiple runs?