Incorrect handling of allocator errors during "warmup" in SimpleDequePool

[k-tokarev]

Expected Behavior

Pool resources should be correctly restored according to size based allocation strategy after allocator problems disappear

Actual Behavior

If allocator can't allocate new resource (emits error in result of acquire) and pool of idle resources is empty then available permits decreasing by minSize-1 (permits for additional "warmup" resources don't return after main resource acquire failure!)

Steps to Reproduce

@Test
void reproCase() {
        val canAllocateResource = new AtomicBoolean(true);
        val allocator = Mono.defer(() ->
            canAllocateResource.get() ?
                Mono.just("value") :
                Mono.error(new IllegalStateException("Can't allocate"))
        );

        val pool = PoolBuilder
            .from(allocator)
            .maxPendingAcquireUnbounded()
            .sizeBetween(10, 10) // Spring Boot R2DBC connections pool default
            .buildPool();

        // New empty pool. No resources allocated yet, but has min-size (10) permits
        assertThat(pool.config().allocationStrategy().estimatePermitCount()).isEqualTo(10);
        assertThat(pool.metrics().idleSize()).isEqualTo(0);

        // Try to acquire one resource. This should trigger pool "warmup" to min-size of resources
        StepVerifier.create(pool.acquire().flatMap(PooledRef::release)).verifyComplete();
        assertThat(pool.config().allocationStrategy().estimatePermitCount()).isEqualTo(0);
        assertThat(pool.metrics().idleSize()).isEqualTo(10);

        // Now allocator will return errors (simulating inaccessible DB server for R2DBC connections pool)
        canAllocateResource.set(false);

        // We have 10 allocated resources in the pool, but they are not valid anymore, so invalidate them
        StepVerifier.create(Flux.range(0, 10).concatMap(ignore -> pool.acquire().flatMap(PooledRef::invalidate)))
            .verifyComplete();
        assertThat(pool.metrics().idleSize()).isEqualTo(0);
        assertThat(pool.config().allocationStrategy().estimatePermitCount()).isEqualTo(10);

        // Now we have empty pool, so it should be warmed up again but allocator still not working
        StepVerifier.create(pool.acquire()).verifyError();
        assertThat(pool.metrics().idleSize()).isEqualTo(0);
        assertThat(pool.config().allocationStrategy().estimatePermitCount()).isEqualTo(10); // Oops!
}

Possible Solution

Naive fix is add .onErrorComplete() or similar error handling to continue Flux processing here:

reactor-pool/reactor-pool/src/main/java/reactor/pool/SimpleDequePool.java

Line 454 in 6d98818

.startWith(primary.doOnSuccess(__ -> drain()).then())

but I didn't test this fix and don't know possible side effects.

Your Environment

io.projectreactor.addons:reactor-pool:1.0.1

• Reactor version(s) used: 3.5.9

• Other relevant libraries versions (eg. netty, ...): no

• JVM version (java -version):
  openjdk version "17.0.7" 2023-04-18 LTS
  OpenJDK Runtime Environment (build 17.0.7+7-LTS)
  OpenJDK 64-Bit Server VM (build 17.0.7+7-LTS, mixed mode, sharing)

• OS and version (eg uname -a): Windows 10 Pro 22H2 Build 19045.3324

[pderop]

Regression caused by #171 in 1.0.1 version.