fix(bigtable): fix session creation leaks

[mutianf]

Fix budget leaks:

1. GO_AWAY before open handshake — session goes STARTING → WAIT_SERVER_CLOSE and closes with prevState == WAIT_SERVER_CLOSE, so the release never fired.
2. Synchronous failure in createSession — factory.createNew()/constructor/metadata-merge throws before any listener is wired, so no terminal callback ever runs to release the slot.
3. Session wedged in STARTING forever — a stream that connects but never sends OpenSession or a GO_AWAY leaves the session stuck with no terminal callback.

[gemini-code-assist]

Code Review

This pull request fixes a budget slot leak in SessionPoolImpl when a session receives a GO_AWAY before completing the open handshake. It introduces a set to track session handles holding a budget reservation, ensuring exactly-once release on success or failure. A test case and fake server support are added to verify this scenario. The feedback suggests using System.nanoTime() instead of System.currentTimeMillis() in the test's polling loop to avoid potential flakiness from system clock adjustments.

[gemini-code-assist]

Using System.currentTimeMillis() to measure elapsed time or timeouts can make tests flaky if the system clock is adjusted (e.g., via NTP synchronization) during execution. It is safer to use System.nanoTime() for monotonic time measurements.

Suggested change

	long deadlineMs = System.currentTimeMillis() + 5_000;
	while (System.currentTimeMillis() < deadlineMs) {
	long deadlineNs = System.nanoTime() + 5_000_000_000L;
	while (System.nanoTime() < deadlineNs) {

[snippet-bot]

No region tags are edited in this PR.

This comment is generated by snippet-bot.
If you find problems with this result, please file an issue at:
https://github.com/googleapis/repo-automation-bots/issues.
To update this comment, add snippet-bot:force-run label or use the checkbox below:

• Refresh this comment

[mutianf]

/gemini review

[gemini-code-assist]

Code Review

This pull request introduces an open-handshake timeout to prevent sessions from remaining in the STARTING state indefinitely, which would otherwise leak session-creation-budget slots. It also improves budget tracking by associating reservations directly with session handles in a new sessionsHoldingBudget set, ensuring that budget slots are released exactly once during both normal and abnormal session lifecycles (such as receiving a GO_AWAY before the handshake completes or encountering synchronous failures during creation). The review feedback identifies two critical concurrency issues in SessionPoolImpl.java where non-thread-safe collections and budget operations are accessed or modified without acquiring the poolLock, which could lead to data races.

[gemini-code-assist]

The sessions list and sessionsHoldingBudget set are non-thread-safe collections guarded by poolLock. Since createSession runs asynchronously and does not hold poolLock during stream creation, calling sessions.newHandle(session) and sessionsHoldingBudget.add(handle) directly here introduces a data race with other threads that access these collections under poolLock (e.g., in onSessionReady or onSessionClose).

Please wrap these registration steps in a poolLock critical section.

        poolLock.lock();
        try {
          handle = sessions.newHandle(session);
          // Bind the budget reservation made by tryReserveSession() above to this handle so it is
          // released exactly once when the session becomes READY or terminates.
          sessionsHoldingBudget.add(handle);
        } finally {
          poolLock.unlock();
        }

[mutianf]

wrong. createSession is guarded by pool lock.

[gemini-code-assist]

The operations in this catch block modify sessionsHoldingBudget, budget, and sessions (via removeUnstartedHandle), all of which are non-thread-safe resources guarded by poolLock. Since createSession runs asynchronously without holding poolLock, executing these modifications without synchronization introduces a critical data race with other threads (e.g., those executing onSessionClose or onSessionReady).

Please wrap these state updates in a poolLock critical section.

      poolLock.lock();
      try {
        if (handle == null || sessionsHoldingBudget.remove(handle)) {
          budget.onSessionCreationFailure();
        }
        if (handle != null) {
          sessions.removeUnstartedHandle(handle);
        }
      } finally {
        poolLock.unlock();
      }

[mutianf]

same as above. CreateSession is already guarded by poolLock.