fix(protocol): defer CONNECTED state until auth completes on connect and failover

[ceache]

Addresses #594
Addresses #727

Summary & Motivation

In ZooKeeper, session state (session_id, watches, ephemeral nodes) is shared across an ensemble, but authentication is connection-bound. Every new TCP connection—whether on initial start, network reconnect, or cluster failover—must be authenticated with the server before the connection is authorized to perform operations for that identity.

Previously, client._session_callback(KeeperState.CONNECTED) (and CONNECTED_RO) was dispatched immediately after raw TCP connect response, before _authenticate_with_sasl completed and before client.auth_data credentials were sent over the wire.

This caused severe race conditions across connection establishment, network reconnect, and ensemble node failover:

1. NoAuthError on reconnect / failover (#594): Application listeners registered via add_listener or threads waiting on wait_for_connection() were notified of CONNECTED prematurely. Operations dispatched on reconnect (such as re-acquiring locks, leader election, or checking znodes) hit the server before the new socket was authenticated, failing with NoAuthError.
2. Server abort with code -124 on strict clusters (#727): On ZooKeeper 3.7+ clusters enforcing authentication (e.g., -Dzookeeper.sessionRequireClientSASLAuth=true or enforce.auth.enabled=true), any non-auth request received before SASL negotiation finishes causes ZooKeeper to immediately close the session with error code -124 (SessionClosedRequireSaslError / ZSESSIONCLOSEDREQUIRESASLAUTH).
3. Auth data mutation races (PR #509): User state listeners reacting to early CONNECTED during reconnect attempted to call add_auth(), mutating client.auth_data while _connect() was still iterating.

---

Comparison with Apache ZooKeeper Java Client

In the official ZooKeeper Java client, non-auth packet submission is strictly prevented while SASL negotiation is in progress, and auth failure is treated as a terminal state:

1. Outgoing Packet Gating During SASL Negotiation

In Java ClientCnxn, the client splits connection handling across SendThread and EventThread. When SASL authentication is active, the client prevents normal request packets from being dispatched over the socket until SASL completes:

• ClientCnxnSocketNIO.java lines 157–185 (findSendablePacket): Inspects the outgoing queue. If zooKeeperSaslClient.clientTunneledAuthenticationInProgress() is true, it only allows priming packets or SASL packets to be sent, skipping all regular operation packets.
• ClientCnxnSocketNIO.java lines 88–92: Explicitly enables socket write interest (SelectionKey.OP_WRITE) only after SASL authentication has completed.
• ClientCnxn.java lines 1451–1470 (tunnelAuthInProgress): Queries whether SASL negotiation is in progress before selecting packets for wire transmission.
• ZooKeeperSaslClient.java lines 415–448 (clientTunneledAuthenticationInProgress): Tracks SASL negotiation states (INITIAL, INTERMEDIATE, COMPLETE, FAILED).

In Kazoo, rather than managing a complex packet-filtering selector in the connection loop, ConnectionHandler._connect_attempt() dispatches SASL and all initial auth_data credentials synchronously during socket establishment. Deferring client._session_callback(KeeperState.CONNECTED) until after this stage guarantees that no user requests or state listeners resume until the socket is authorized.

2. Terminal AuthFailed State Handling

In Java ZooKeeper, an authentication failure immediately marks the client as failed, publishes AuthFailed to watchers, and permanently stops the connection thread:

• ClientCnxn.java lines 1194–1221: On SASL failure in SendThread, sets state = States.AUTH_FAILED, posts a WatchedEvent(None, AuthFailed, null) to EventThread, and sends an eventOfDeath to stop the thread loop.
• ClientCnxn.java lines 903–908: If the server returns an auth failure reply, the client immediately transitions to States.AUTH_FAILED.
• ZooKeeperSaslClient.java lines 381–395 (getKeeperState): Resolves the Keeper state to AuthFailed upon authentication failure, or SaslAuthenticated upon success.

---

Changes in this PR

1. Defer CONNECTED / CONNECTED_RO until Auth Finishes:

   • In kazoo.protocol.connection.ConnectionHandler, move client._session_callback(KeeperState.CONNECTED) to execute strictly after _authenticate_with_sasl() completes and all initial auth_data packets have been dispatched.

2. Unify start_async() Return Value with IAsyncResult:

   • Historically, start_async() returned self._live (threading.Event), which was only set on success. Purely async clients (res = zk.start_async()) were unable to receive terminal failure notifications or exceptions, hanging indefinitely on res.wait().
   • start_async() now returns an IAsyncResult instance—matching every other _async method in Kazoo.
   • The returned IAsyncResult completes with True when CONNECTED/CONNECTED_RO is reached, or is set with an exception (AuthFailedError, SessionExpiredError, or ConnectionClosedError) if a terminal failure occurs.
   • Backwards Compatibility: IAsyncResult.wait(timeout) has the identical signature as threading.Event.wait(timeout) (wait(timeout=None) -> bool). Existing code calling event = zk.start_async(); event.wait(30) continues to work unmodified, but now unblocks immediately on terminal failure instead of hanging.
   • Purely async callers can now inspect res.ready(), res.successful(), res.exception, call res.get(), or attach async callbacks with res.rawlink().

3. Immediate Unblock and Exception in KazooClient.start():

   • start(timeout=15.0) is now a clean consumer of self.start_async().get(timeout=timeout).
   • On authentication failure (KeeperState.AUTH_FAILED), start() unblocks immediately (<10ms) and raises AuthFailedError with the underlying cause instead of hanging for the full 15-second timeout and raising KazooTimeoutError.

4. Preserve Terminal States in zk_loop:

   • In ConnectionHandler.zk_loop, the finally: block now preserves terminal CLOSED_STATES (AUTH_FAILED, EXPIRED_SESSION) rather than clobbering client._state with CLOSED.

5. Documentation & Unit Tests:

   • Updated docs/async_usage.rst to document that all _async methods return IAsyncResult.
   • Added unit tests in kazoo.tests.test_client.TestSessionCallbacks verifying immediate failure on AUTH_FAILED, error propagation with custom causes, and start_async() IAsyncResult completion semantics.

[ceache]

This change is solves all sort of race conditions during failover/reconnection (where there is a lot of queued requests).

It is also a slight departure from what the Java ZK client does. It uses all sort of sending queue flags to filter what gets send so it does not transmit any other client request before auth is done. I did not want to bring this complexity in our codebase if I could avoid it.

That being said, there will be a behavior change, in the failure paths:

• Java: client connects, CONNECTED event triggers, concurrently auth immediately fails, state moves to AUTH_FAILED.
• Kazoo: client connects, auth fails, state moves to AUTH_FAILED (no CONNECTED event trigger).

In practice, I do not think it matters but I am flagging it for transparency.

[StephenSorriaux]

Thank you. To be honest, I would prefer we continue to follow the official developer guide https://zookeeper.apache.org/doc/r3.9.5/zookeeperProgrammers.html (see ZooKeeper Sessions since it is not possible to link directly to it...) which seems to indicate that a client library should be in CONNECTED state before an eventual AUTH_FAILED (at least from my understanding).

[ceache]

You are talking about the state machine diagram in https://zookeeper.apache.org/doc/r3.9.5/zookeeperProgrammers.html#ch_zkSessions, right?

There is a CONNECTING state => AUTH_FAILED transition to the left, without going through the CONNECTED state.

It does mean some auth exchanges are happening before the connection gets to the CONNECTED state, don't you agree?
I take it to say what I do is ok, albeit different from Java...

[codecov]

Codecov Report

❌ Patch coverage is 88.88889% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 94.81%. Comparing base (f54c4fd) to head (6e550e6).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
kazoo/client.py	85.71%	5 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master     #808      +/-   ##
==========================================
- Coverage   94.90%   94.81%   -0.10%     
==========================================
  Files          27       27              
  Lines        3811     3841      +30     
==========================================
+ Hits         3617     3642      +25     
- Misses        194      199       +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.