Fixed StackOverflow error for autoreleased sessions

[alex268]

No description provided.

[codecov]

Codecov Report

❌ Patch coverage is 88.23529% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 72.41%. Comparing base (bc6886d) to head (7f7218b).

Files with missing lines	Patch %	Lines
...ain/java/tech/ydb/core/impl/BaseGrpcTransport.java	33.33%	4 Missing ⚠️

Additional details and impacted files

@@                  Coverage Diff                  @@
##             release_v2.4.11     #714      +/-   ##
=====================================================
- Coverage              72.45%   72.41%   -0.04%     
  Complexity              3524     3524              
=====================================================
  Files                    391      391              
  Lines                  16341    16362      +21     
  Branches                1702     1704       +2     
=====================================================
+ Hits                   11840    11849       +9     
- Misses                  3864     3871       +7     
- Partials                 637      642       +5     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[robot-vibe-db]

Full analysis log

Analysis performed by claude, claude-opus-4-6.

[KirillKurdyukov]

Причина исходного дедлока, похоже, не в логгере и не в забытом finally, а в StackOverflowError внутри самого ReentrantReadWriteLock.lock().

WaitingQueue синхронно завершает CompletableFuture. Для autoreleased session continuation сразу вызывает session.close() → WaitingQueue.release(), который завершает следующий future. Возникает глубокая рекурсивная цепочка.

На почти исчерпанном стеке запрос может дойти до EndpointPool.getEndpoint() → readLock().lock(). В Java 8 реализация read lock сначала CAS-ом увеличивает shared count, а затем обновляет firstReader/hold counter. Если между этими действиями возникает StackOverflowError, lock() не возвращается и внешний try/finally ещё не начинается — следовательно, unlock() не выполняется.

Это объясняет debugger: reader-thread уже idle в thread pool, но shared count остался увеличенным. Writer discovery/pessimization ждёт reader, а новые readers оказываются за ожидающим writer. OpenJDK отдельно воспроизводит такой класс повреждений lock в ReservedStackTest:
https://cr.openjdk.org/~goetz/wr16/8156923-resStack/webrev.01/test/runtime/ReservedStack/ReservedStackTest.java.html

Перенос deadline-check перед getChannel() выглядит правильной дополнительной защитой.

Однако реализация trampoline через один ThreadLocal<T> содержит два blocker-дефекта.

1. Если последний waiter отменён, safeAcquireObject() возвращает false, цикл заканчивается, но localResource остаётся установленным. Следующий обычный release() на том же worker принимает stale marker за вложенный autorelease и теряет ресурс: его нет ни в used, ни в idle, а queueSize не уменьшается.

2. Если continuation, получив ресурс A, синхронно освобождает другой ресурс B, release(B) стирает marker A. Внешний цикл решает, что A тоже был освобождён, и кладёт его в idle, хотя waiter всё ещё владеет A. Один session/resource может быть одновременно выдан двум клиентам.

Я локально добавил следующие reproducer-тесты на head e491040 (в ветку PR они не входят):

@Test
public void nestedReleaseOfAnotherResourceMustNotReleaseCurrentResource() {
    ResourceHandler rs = new ResourceHandler();
    WaitingQueue<Resource> queue = new WaitingQueue<>(rs, 2, 3);

    CompletableFuture<Resource> first = pendingFuture(acquire(queue));
    CompletableFuture<Resource> second = pendingFuture(acquire(queue));
    rs.completeNext().completeNext();
    Resource a = pendingIsReady(first);
    Resource b = pendingIsReady(second);

    CompletableFuture<Resource> waitingForA = pendingFuture(acquire(queue));
    CompletableFuture<Resource> waitingForB = pendingFuture(acquire(queue));
    waitingForA.thenAccept(ignored -> queue.release(b));

    queue.release(a);

    Assert.assertSame(a, pendingIsReady(waitingForA));
    Assert.assertSame(b, pendingIsReady(waitingForB));
    Assert.assertEquals("resource A is still owned by waitingForA", 0, queue.getIdleCount());
}

@Test
public void canceledLastWaitingMustNotLeaveThreadLocalMarker() {
    ResourceHandler rs = new ResourceHandler();
    WaitingQueue<Resource> queue = new WaitingQueue<>(rs, 1, 1);

    CompletableFuture<Resource> first = pendingFuture(acquire(queue));
    rs.completeNext();
    Resource resource = pendingIsReady(first);

    CompletableFuture<Resource> canceled = pendingFuture(acquire(queue));
    canceled.cancel(true);
    queue.release(resource);
    Assert.assertEquals(1, queue.getIdleCount());

    Resource acquiredAgain = readyFuture(acquire(queue));
    queue.release(acquiredAgain);

    Assert.assertEquals("resource must return to idle after a normal release", 1, queue.getIdleCount());
    Assert.assertEquals(0, queue.getUsedCount());
}

Оба теста падают на текущей реализации:

Tests run: 2, Failures: 2
canceledLastWaitingMustNotLeaveThreadLocalMarker
nestedReleaseOfAnotherResourceMustNotReleaseCurrentResource

Предлагаю вместо marker-а текущего ресурса использовать настоящий per-thread drain loop: первый release() создаёт контекст с ArrayDeque<T>, вложенные release() только добавляют элементы в deque, внешний вызов итеративно обрабатывает очередь, а ThreadLocal.remove() выполняется в finally.

До исправления этих двух сценариев текущую реализацию ThreadLocal<T> лучше не мержить.

[KirillKurdyukov]

Проверил новый head 3d0a1b2 (Fixed waiting guard). Оба замечания из предыдущего review действительно исправлены:

• canceled last waiter больше не оставляет stale ThreadLocal;
• сценарий «waiter получил A и синхронно освободил другой ресурс B» теперь сохраняет A в used;
• цепочка из 10 000 autorelease одного и того же ресурса проходит.

Однако текущая комбинация одного ThreadLocal<T> и проверки used.containsKey(object) всё ещё содержит три blocker-сценария.

1. Вложенный release(B) уничтожает guard внешнего ресурса A

В tryToCompleteWaiting() вложенный вызов перезаписывает один ThreadLocal:

guard=A
complete waiter(A)
  release(B)
    guard=B
    finally: remove()  // удаляет guard внешнего A
  release(A)           // guard уже null

release(A) кладёт A в idle. После возвращения внешний frame видит used.containsKey(A) == false и кладёт A в idle повторно. Получается queueSize=2, но idleSize=3; один и тот же A может быть выдан двум acquire.

2. used.containsKey(object) создаёт race с release из другого потока

Если continuation освобождает A на другом потоке до возврата из CompletableFuture.complete():

T1: complete(waiter, A)
T2: release(A) → remove from used → put A into idle
T1: used.containsKey(A) == false → put A into idle ещё раз

Детерминированный тест получает queueSize=1, но idleSize=2. Это не только неправильная метрика: deque содержит один экземпляр дважды.

3. StackOverflow остаётся для цепочки разных ресурсов

Guard останавливает рекурсию только при waitingGuard.get() == object. Цепочка

complete waiter(A) → release(B)
complete waiter(B) → release(C)
complete waiter(C) → release(D)
...

по-прежнему рекурсивно входит в release() → tryToCompleteWaiting(). На 10 000 ресурсах CompletableFuture поймал StackOverflowError в continuation примерно на глубине 1300, а оставшиеся waiter-ы не завершились.

Ниже полный код локальных reproducer-тестов, добавленных поверх head 3d0a1b2 (в ветку PR они не входят):

@Test
public void nestedOtherReleaseMustNotLoseCurrentGuard() {
    ResourceHandler rs = new ResourceHandler();
    WaitingQueue<Resource> queue = new WaitingQueue<>(rs, 2, 3);

    CompletableFuture<Resource> first = pendingFuture(acquire(queue));
    CompletableFuture<Resource> second = pendingFuture(acquire(queue));
    rs.completeNext().completeNext();
    Resource a = pendingIsReady(first);
    Resource b = pendingIsReady(second);

    CompletableFuture<Resource> waiting = pendingFuture(acquire(queue));
    waiting.thenAccept(ignored -> {
        queue.release(b);
        queue.release(a);
    });

    queue.release(a);

    Assert.assertSame(a, pendingIsReady(waiting));
    check(queue).queueSize(2).idleSize(2).waitingsCount(0);
}

@Test
public void releaseFromAnotherThreadMustNotDuplicateResource() {
    ResourceHandler rs = new ResourceHandler();
    WaitingQueue<Resource> queue = new WaitingQueue<>(rs, 1, 1);

    CompletableFuture<Resource> first = pendingFuture(acquire(queue));
    rs.completeNext();
    Resource resource = pendingIsReady(first);

    CompletableFuture<Resource> waiting = pendingFuture(acquire(queue));
    waiting.thenAccept(acquired -> {
        Thread releaser = new Thread(() -> queue.release(acquired));
        releaser.start();
        try {
            releaser.join();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new AssertionError(e);
        }
    });

    queue.release(resource);

    Assert.assertSame(resource, pendingIsReady(waiting));
    check(queue).queueSize(1).idleSize(1).waitingsCount(0);
}

@Test
@SuppressWarnings("unchecked")
public void releasesOfDifferentResourcesMustNotRecurse() {
    int count = 10000;
    ResourceHandler rs = new ResourceHandler();
    WaitingQueue<Resource> queue = new WaitingQueue<>(rs, count, count);

    CompletableFuture<Resource>[] owners =
            (CompletableFuture<Resource>[]) new CompletableFuture<?>[count];
    Resource[] resources = new Resource[count];
    for (int idx = 0; idx < count; idx++) {
        owners[idx] = pendingFuture(acquire(queue));
    }
    for (int idx = 0; idx < count; idx++) {
        rs.completeNext();
        resources[idx] = pendingIsReady(owners[idx]);
    }

    CompletableFuture<Resource>[] waiters =
            (CompletableFuture<Resource>[]) new CompletableFuture<?>[count];
    CompletableFuture<Void>[] continuations =
            (CompletableFuture<Void>[]) new CompletableFuture<?>[count - 1];
    for (int idx = 0; idx < count; idx++) {
        waiters[idx] = pendingFuture(acquire(queue));
        if (idx + 1 < count) {
            Resource next = resources[idx + 1];
            continuations[idx] = waiters[idx].thenAccept(ignored -> queue.release(next));
        }
    }

    queue.release(resources[0]);

    for (int idx = 0; idx < continuations.length; idx++) {
        if (continuations[idx].isCompletedExceptionally()) {
            try {
                continuations[idx].join();
            } catch (CompletionException e) {
                Assert.assertFalse("continuation " + idx + " captured StackOverflowError",
                        e.getCause() instanceof StackOverflowError);
            }
        }
    }
    for (int idx = 0; idx < count; idx++) {
        Assert.assertTrue("waiting " + idx + " must be completed", waiters[idx].isDone());
    }
}

Результат свежего запуска:

Tests run: 3, Failures: 3
nestedOtherReleaseMustNotLoseCurrentGuard:
  expected idleSize=2, actual idleSize=3
releaseFromAnotherThreadMustNotDuplicateResource:
  expected idleSize=1, actual idleSize=2
releasesOfDifferentResourcesMustNotRecurse:
  continuation 1318 captured StackOverflowError

При этом штатный WaitingQueueTest на чистом head проходит: 21 tests, 0 failures.

Здесь нужен настоящий per-thread drain context с ArrayDeque<T>:

• первый release() создаёт deque и начинает итеративный drain;
• любой вложенный release() только добавляет освобождённый ресурс в deque;
• успешный safeAcquireObject() считается завершённой передачей без последующей проверки used.containsKey();
• release из другого потока обрабатывается его собственным drain;
• ThreadLocal.remove() выполняется в finally.

До устранения этих трёх сценариев новый guard всё ещё лучше не мержить.

[KirillKurdyukov]

Перепроверил новый head dae5068. Кейс с цепочкой из 10 000 разных ресурсов здесь больше не считаю блокером. Но остался один двухресурсный сценарий с потерей ресурса, а сам head сейчас не проходит checkstyle.

1. Вложенный guard теряет autorelease внешнего ресурса

Новый WaitingGuard исправляет предыдущие сценарии, но вложенный tryToCompleteWaiting(B) всё ещё перезаписывает guard внешнего A:

guard = A
complete waiter(A)
  release(A) → выставляет releaseIsInterrupted=true и возвращается
  release(B)
    waitingGuard.set(guardB)
    finally waitingGuard.remove()
outer frame:
  waitingGuard.get() → новый пустой guard
  isNotInterrupted(A) → true

Флаг, сообщавший внешнему циклу, что A был autoreleased, теряется. A уже удалён из used, но не попадает ни следующему waiter-у, ни в idle.

Детерминированный reproducer, локально добавленный поверх dae5068 (в ветку PR не входит):

@Test
public void currentReleaseBeforeOtherMustNotBeLost() {
    ResourceHandler rs = new ResourceHandler();
    WaitingQueue<Resource> queue = new WaitingQueue<>(rs, 2, 3);

    CompletableFuture<Resource> first = pendingFuture(acquire(queue));
    CompletableFuture<Resource> second = pendingFuture(acquire(queue));
    rs.completeNext().completeNext();
    Resource a = pendingIsReady(first);
    Resource b = pendingIsReady(second);

    CompletableFuture<Resource> waiting = pendingFuture(acquire(queue));
    waiting.thenAccept(ignored -> {
        queue.release(a);
        queue.release(b);
    });

    queue.release(a);

    Assert.assertSame(a, pendingIsReady(waiting));
    check(queue).queueSize(2).idleSize(2).waitingsCount(0);
}

Результат на текущей реализации:

expected idleSize=2
actual idleSize=1

То есть queueSize=2, но реально доступен только B — A потерян. Обратный порядок release(B); release(A) теперь проходит, поэтому добавленный в PR тест этого не замечает.

2. Head не проходит checkstyle

Локально воспроизводятся те же ошибки, из-за которых сейчас красные JDK 8/11 jobs:

WaitingQueue.java:326 Redundant 'public' modifier
WaitingQueue.java:330 Redundant 'public' modifier

Это public у конструкторов приватного WaitingGuard.

Минимальное исправление в текущем подходе

Для этих практических сценариев необязательно переходить на drain loop. Достаточно:

• сохранить созданный currentGuard в локальной переменной и после complete() проверять именно его;
• перед установкой нового guard сохранить previousGuard;
• в finally восстановить previousGuard, а не безусловно вызывать remove();
• заменить ThreadLocal.withInitial(...) на обычный new ThreadLocal<>() и делать null-check в release();
• убрать пустой конструктор guard и избыточные public.

Этот минимальный вариант проверил локально:

targeted practical scenarios: 6/6 passed
full WaitingQueueTest: 25/25 passed
checkstyle: 0 violations
BUILD SUCCESS

Предыдущие практические дефекты — canceled waiter, cross-thread release и порядок release(B); release(A) — на новом head действительно исправлены. После исправления зеркального порядка и checkstyle других блокеров в согласованном scope не вижу.

[KirillKurdyukov]

Финальный повторный review актуального HEAD 7f7218b.

Блокирующих замечаний больше не вижу — LGTM.

Новая реализация WaitingGuard закрывает проблемы предыдущей версии:

• guard теперь фактически образует стек: сохраняет предыдущий frame из ThreadLocal и восстанавливает его в close(), поэтому вложенная обработка другого resource больше не теряет состояние внешней передачи;
• release того же resource во время callback помечает текущую передачу как broken, и resource не теряется и не дублируется;
• release другого resource не ломает текущий guard;
• release с другого потока работает с независимым ThreadLocal-состоянием.

Дополнительно к тестам из ветки я локально добавил три регрессии: обратный порядок release двух resources, release переданного resource с другого потока и отмена последнего waiter. Результаты на этом HEAD:

• целевые сценарии: 6/6 passed;
• весь WaitingQueueTest, включая три локальные регрессии: 25/25 passed;
• Checkstyle: 0 violations.

Полный локальный reactor в моём sandbox упирается в ограничения окружения (self-attach Mockito/ByteBuddy и запрет открытия socket), а не в изменения PR. При этом GitHub CI полностью зелёный: build и Maven CI на JDK 8/11/17/21, coverage и Codecov.

Искусственную рекурсивную цепочку из 10 000 разных resources по договорённости считаю вне scope этого PR и блокером не считаю.

Итог: практический сценарий зависшего EndpointPool исправлен, найденные регрессии закрыты, PR можно мержить.