fix(gateways): survive Kafka accept errors and a late Iggy at startup

.config/nextest.toml

@@ -59,6 +59,7 @@ filter = '''
     + binary_id(iggy-gateway-kafka::metadata_real_bridge_tests)
     + binary_id(iggy-gateway-kafka::offsets_real_bridge_tests)
     + binary_id(iggy-gateway-kafka::phase1_e2e_tests)
+    + binary_id(iggy-gateway-kafka::sasl_real_server_tests)
 '''
 test-group = "kafka_bridge"
 

gateways/kafka/README.md

@@ -77,7 +77,7 @@ Default bind: a `PLAINTEXT` listener on `127.0.0.1:9093`. Environment variables:
 | `IGGY_KAFKA_WRITE_TIMEOUT_SECS` | `10` | Seconds allowed to write a response frame |
 | `IGGY_KAFKA_SHUTDOWN_DRAIN_TIMEOUT_SECS` | `25` | Seconds graceful shutdown waits for in-flight connections before abandoning them. The bridge then signs its Iggy clients out, 5 s at most |
 | `IGGY_KAFKA_INSTANCE_ID` | `0` | This gateway's number among the gateways fronting one Iggy cluster. It is the high half of every producer id `InitProducerId` hands out, and Kafka requires those to be cluster-unique, so give every gateway its own value. A single gateway can leave it at `0`. |
-| `IGGY_KAFKA_BRIDGE_ENABLED` | `false` | Connect the Iggy bridge at startup. While false every API answers with its stub, and the `IGGY_KAFKA_IGGY_*` variables below are read by nothing. A failed connection is fatal, not a downgrade to stubs. |
+| `IGGY_KAFKA_BRIDGE_ENABLED` | `false` | Connect the Iggy bridge at startup. While false every API answers with its stub, and the `IGGY_KAFKA_IGGY_*` variables below are read by nothing. If Iggy does not answer, startup tries again 1 s after each failed connect, for up to 60 s, and then the gateway stops. A wrong password or a missing permission stops it at once. It never falls back to stubs. |
 | `IGGY_KAFKA_SASL_ENABLED` | `false` | Single-listener form: require SASL/PLAIN before serving any other API (`true` or `false`, nothing else). With `IGGY_KAFKA_LISTENERS`, a `SASL_PLAINTEXT` or `SASL_SSL` listener says the same |
 | `IGGY_KAFKA_PRE_AUTH_TIMEOUT_SECS` | `15` | Seconds an unauthenticated connection may sit between frames. Waiting for an authentication slot and the verification itself each get this budget, the verification's starting once it holds a slot. Separate from the 10-minute idle timeout that applies once authenticated |
 | `IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS` | `4` | Credential verifications the gateway runs at once, across all connections. Each costs a password hash on an Iggy shard thread. This bounds the gateway's side only: a check that times out frees its slot while its hash keeps running inside Iggy. Size it below the Iggy node's shard count |
@@ -316,6 +316,10 @@ IGGY_KAFKA_LISTENERS=PLAINTEXT://127.0.0.1:9093,SSL://0.0.0.0:9094,SASL_SSL://0.
 
 Each security protocol can be bound once, since the protocol is also the listener's name.
 
+No `accept()` error stops a listener. An error of one pending connection, such as a reset, retries
+at once. Any other error, such as no free file descriptor, waits 10 ms before the next accept.
+The wait doubles up to 1 s while the error continues.
+
 TLS is rustls. One certificate serves every `SSL` and `SASL_SSL` listener, and the files are
 re-read every `IGGY_KAFKA_TLS_RELOAD_INTERVAL_SECS`: a rotated certificate takes effect for new
 connections without a restart, a key that does not match leaves the previous material serving and
@@ -422,8 +426,10 @@ Four things to know before switching it on:
 - **Enabling it breaks every existing client at once.** The two SASL keys appear in the
   `ApiVersions` advertisement only while it is on, and unauthenticated clients are refused.
 - **Every connection costs a login**, meaning one password hash on an Iggy shard thread and one
-  replicated registration. Verification is deliberately not cached, since caching it per username
-  would let a second connection present any password. Connection churn is therefore server load.
+  replicated registration. The check then signs out, so its Iggy session ends at once. Without
+  the sign out, Iggy keeps the session for 36 s, and a full session table refuses new logins.
+  Verification is deliberately not cached, since caching it per username would let a second
+  connection present any password. Connection churn is therefore server load.
   `IGGY_KAFKA_MAX_CONCURRENT_AUTHENTICATIONS` bounds the checks the gateway runs at once, and a peer
   whose login was rejected is refused for a delay that doubles per rejection, from 0.5s up to 30s.
 - **Authentication and an ACL view, not enforcement.** The gateway verifies the credentials and
@@ -596,7 +602,9 @@ The initial connect retries a fixed, bounded number of times (`RECONNECTION_RETR
 Iggy SDK client's own default of unlimited retries, one dial per second, forever), and the whole
 attempt - retries included - is capped at `REQUEST_TIMEOUT` (15s) wall-clock, so `IggyBridge::connect`
 fails in bounded time whether the address refuses the connection or silently drops it, instead of
-blocking the calling task indefinitely. Every other bridge call (`ensure_stream_and_topic`,
+blocking the calling task indefinitely. At startup, the gateway retries a connect that a retry can
+fix (6 or 7). It waits 1 s between tries, for up to 60 s. The cluster id load after it does the same.
+So a gateway that starts before its Iggy node waits for it. Every other bridge call (`ensure_stream_and_topic`,
 `high_watermark(s)`, `probe`, `close`) carries the same `REQUEST_TIMEOUT` for the same
 reason: the SDK reconnects internally, mid-call, on a transport error, through the same
 undead-lined dial path - a bridge call made well after the initial connect can still hit this if

gateways/kafka/docs/AUTHENTICATION.md

@@ -25,7 +25,7 @@ verified on every connection, against Iggy, and that verification is deliberatel
 
 **Authorization is meant to stay in Iggy.** The gateway will not implement its own ACL model. The intent is
 to carry the authenticated Iggy identity onto the data plane and let the server's existing permission checks
-decide what it can do. That is later work: today the verified client is shut down as soon as the credential
+decide what it can do. That is later work: today the verified client signs out and shuts down as soon as the credential
 clears, and nothing carries the principal past authentication.
 
 **A client certificate is an identity too, with the same reach.** On an `SSL` listener that verifies client
@@ -63,12 +63,12 @@ connection. The cost rules it out.
 | Argon2id verify, run inline on the shard thread | 19 MiB, t=2, p=1 |
 | Replicated VSR `Register` | one consensus commit |
 | Leader settlement round trip, plus a second login if it redirects | one or two round trips |
-| Session slots, cluster-wide across all transports | 8192, oldest committed evicted past it |
+| Session slots, cluster-wide across all transports | 8192, and a full table refuses new logins |
 
 The Argon2 verify runs on the shard's compio runtime, which has no blocking pool. A burst of Kafka
 connections therefore stalls that shard's entire request loop, including produce and fetch traffic that has
-nothing to do with authentication. The 8192 ceiling (`core/consensus/src/impls.rs:178`) is cluster-wide, so
-connection churn on the Kafka side turns directly into session-table pressure and eviction on the Iggy side.
+nothing to do with authentication. The 8192 ceiling (`[metadata] clients_table_max`, `core/consensus/src/impls.rs:211`) is cluster-wide.
+So connection churn on the Kafka side turns directly into session-table pressure on the Iggy side.
 
 ## Why not a shared service account
 
@@ -103,7 +103,7 @@ Two consequences worth naming rather than discovering:
   traffic on that shard. Measured on debug builds, a single login takes about 10 ms and 32 concurrent
   logins take about 75 ms each, with throughput flattening near 410 per second. The shape is the point,
   latency climbing while throughput plateaus, not the absolute numbers. `MANUAL_TESTING.md` has the table.
-- **Nothing holds the verified session.** The connection that proved a credential drops it at once.
+- **Nothing holds the verified session.** The connection that proved a credential signs out at once.
   Produce and Fetch act as the bridge's Iggy user for every client. A data client per principal needs
   pooling, a separate question from how credentials are verified. Pooling is gateway-side code over the
   public SDK and needs no SDK change.

gateways/kafka/docs/TEST_SUITE.md

@@ -20,7 +20,7 @@ suite goes through `tests/common/fixtures.rs::load_fixture_body_or_skip`, which
 regeneration hint when a fixture is missing, and panics instead when `KAFKA_FIXTURES_REQUIRED=1`
 is set (CI sets this) so a broken generation step can't leave a suite green with zero assertions.
 
-### `iggy-server` binary (required for `bridge_iggy_integration_tests` and every `*_real_bridge_tests` suite)
+### `iggy-server` binary (required for `bridge_iggy_integration_tests`, `sasl_real_server_tests` and every `*_real_bridge_tests` suite)
 
 No fixtures needed, but `iggy-server` has to be built *first* - these suites spawn it directly and
 do not build it for you:
@@ -80,6 +80,7 @@ file under `tests/` anymore.
 | [`sasl_tests.rs`](../tests/sasl_tests.rs) | SASL/PLAIN over a socket — full handshake, every refusal path, and the disabled default. Drives a stub verifier implementing `SaslAuthenticator`, so no Iggy server is needed | No |
 | [`tls_listener_tests.rs`](../tests/tls_listener_tests.rs) | The `SSL` and `SASL_SSL` listeners over a real socket with certificates minted per test: each listener advertising its own port, a plaintext client refused on the TLS port, SASL inside TLS, client certificates mapped to users through a stub `PrincipalResolver` (`required`, `requested`, unknown user, a subject no rule maps, a failing or hung lookup, `SASL_SSL` asking for no certificate), a multi-MiB response to a slow reader, the handshake budget, certificate reload without a restart, and the startup refusals | No |
 | [`kafka_client_e2e_tests.rs`](../tests/kafka_client_e2e_tests.rs) | **Real Kafka clients** against the whole stack: a spawned `iggy-server`, the gateway in-process with a real authenticator, and kcat / the Java tools from containers. With a bridge, kcat writes records, and kcat and the Java consumer read them back through Fetch. Java consumers in one group resume from their commits, across a gateway restart and a member leaving. `kafka-cluster cluster-id` prints the same id across a gateway restart. Lookups by time resolve through `kafka-get-offsets` (a time, 0, `-3` and a future time), `kafka-consumer-groups --reset-offsets --to-datetime`, and the consumer's `offsetsForTimes` in a probe compiled with the host `javac`. librdkafka and the Java client connect with `security.protocol=SSL` and `SASL_SSL`, and a client certificate admits each as the Iggy user it names. With auto topic creation on, the Java producer sends to a topic nobody created. The only suite that can catch a client-compatibility bug, since every other one hand-builds frames | No, but needs Docker, a built `iggy-server` and a host JDK (`javac`) |
+| [`sasl_real_server_tests.rs`](../tests/sasl_real_server_tests.rs) | `IggyAuthenticator` against a real, spawned `iggy-server`: twice as many logins as a small session table holds all succeed, because each check signs out | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`bridge_iggy_integration_tests.rs`](../tests/bridge_iggy_integration_tests.rs) | `IggyBridge` against a real, spawned `iggy-server` — provisioning idempotency, high watermark, credential/connection edge cases, and the bridge dialing a TLS-only server over TLS | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`list_offsets_real_bridge_tests.rs`](../tests/list_offsets_real_bridge_tests.rs) | ListOffsets (key 2) through the whole handler against a real, spawned `iggy-server`: LATEST, EARLIEST, lookups by time and `MAX_TIMESTAMP`, the `-1` answers, code 3, several partitions looked up by time in one request, and a stalled lookup leaving the later partitions their answers | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`fetch_real_bridge_tests.rs`](../tests/fetch_real_bridge_tests.rs) | Fetch (key 1) through the whole handler against a real, spawned `iggy-server` — records go in through the Iggy SDK or the Produce handler and come back through `kafka_protocol`'s client decoder, at every version, with codes 0, 1, 3, 6, -1 and 70, paged polls, and the wait | No (needs the `iggy-server` binary - see Prerequisites) |
@@ -88,7 +89,7 @@ file under `tests/` anymore.
 | [`offsets_real_bridge_tests.rs`](../tests/offsets_real_bridge_tests.rs) | OffsetCommit (key 8) and OffsetFetch (key 9) through the whole handler against a real, spawned `iggy-server`: offsets stored as sent, the Iggy external group key, a negative commit as a delete, -1 for nothing committed, a null topic list, and a gateway restart ([#3542](https://github.com/apache/iggy/issues/3542) acceptance) | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`offset_groups_real_bridge_tests.rs`](../tests/offset_groups_real_bridge_tests.rs) | ListGroups (key 16), DescribeGroups (key 15), DeleteGroups (key 42) and OffsetDelete (key 47) through the whole handler against a real, spawned `iggy-server`. A group with offsets and no members is `Empty`, and a live group is listed once. Also the filters, a commit after the last scan, a new gateway, a native Iggy group, and only the live groups once a scan failed with Iggy down. DeleteGroups refuses a group with members, and after they leave deletes it on every topic. Through a relay that holds an Iggy call past its deadline, a commit and a group deletion each keep the other out until the call ends, and past the SDK's 30 s deadline until a replay proves what the call did. A delete that an earlier delete in doubt kept from Iggy lets go of its group. A deletion that ends after its request answered 14 answers 0 to the retry. OffsetDelete keeps the offsets of a topic that a member subscribes to, deletes the others, and finds a group that another gateway committed for since the last scan | No (needs the `iggy-server` binary - see Prerequisites) |
 | [`cluster_id_real_bridge_tests.rs`](../tests/cluster_id_real_bridge_tests.rs) | The cluster id stored in a real, spawned `iggy-server`: one id for every bridge, the first write winning when several start at once, a persisted topic with no expiry or size cap, the id surviving an `iggy-server` SIGKILL, a stored value that is not an id refused, one id per default stream, and the compiled gateway binary reporting the same id in Metadata and DescribeCluster across a restart ([#4519](https://github.com/apache/iggy/issues/4519)) | No (needs both binaries - see Prerequisites) |
-| [`phase1_e2e_tests.rs`](../tests/phase1_e2e_tests.rs) | [#3539](https://github.com/apache/iggy/issues/3539) acceptance: CreateTopics → Metadata → Produce → ListOffsets over real TCP against the real compiled `iggy-gateway-kafka` binary bridged to a real spawned `iggy-server` - with `cluster_id_real_bridge_tests.rs`, one of the two suites that spawn the gateway as a process rather than calling `KafkaGateway::run` in-process | No (needs both binaries - see Prerequisites) |
+| [`phase1_e2e_tests.rs`](../tests/phase1_e2e_tests.rs) | [#3539](https://github.com/apache/iggy/issues/3539) acceptance: CreateTopics → Metadata → Produce → ListOffsets over real TCP against the real compiled `iggy-gateway-kafka` binary bridged to a real spawned `iggy-server` - with `cluster_id_real_bridge_tests.rs`, one of the two suites that spawn the gateway as a process rather than calling `KafkaGateway::run` in-process. A gateway started 16 s before its `iggy-server` waits for it and then serves | No (needs both binaries - see Prerequisites) |
 
 `tests/common/` holds shared helpers (`certs.rs`, `codec.rs`, `fixtures.rs`, `gateway_process.rs`,
 `scope.rs`, `server.rs`, `iggy_server.rs`, `tcp.rs`, `wire.rs`), compiled per test binary via `#[path]`, not a test binary itself. `certs.rs`

gateways/kafka/src/auth.rs

@@ -29,7 +29,7 @@ use std::time::{Duration, Instant};
 use async_trait::async_trait;
 use iggy::prelude::{
     AutoLogin, Client, Credentials, Identifier, IggyError, Permissions, TcpClient,
-    TcpClientConfigBuilder, UserStatus,
+    TcpClientConfigBuilder, UserClient, UserStatus,
 };
 use tokio::sync::SemaphorePermit;
 use tracing::{debug, warn};
@@ -64,12 +64,12 @@ const VERIFY_TIMEOUT: Duration = Duration::from_secs(10);
 /// would only hide the failure underneath one the client cannot see.
 const VERIFY_RECONNECTION_RETRIES: u32 = 0;
 
-/// Budget for tearing the verification client down again.
+/// Budget for signing the verification client out and tearing it down again.
 ///
 /// Deliberately far shorter than [`VERIFY_TIMEOUT`]. The authentication slot is already released by
 /// the time teardown runs, but every second here still comes out of the caller's own
-/// pre-authentication budget. Nothing is lost by cutting it short: dropping the client closes the
-/// socket regardless.
+/// pre-authentication budget. Little is lost by cutting it short: dropping the client closes the
+/// socket regardless, and Iggy ends a session that did not sign out when its lease runs out.
 const TEARDOWN_TIMEOUT: Duration = Duration::from_secs(1);
 
 /// Delay imposed on a peer after its first rejected login, doubled on every further rejection.
@@ -391,6 +391,7 @@ impl SaslAuthenticator for IggyAuthenticator {
         // The password verify the slot bounds is done. Holding it across the permission read and
         // teardown would keep a slot busy for two more round trips per login.
         drop(slot);
+        let signed_in = matches!(connected, Ok(Ok(())));
         let outcome = match connected {
             Err(_elapsed) => Err(AuthError::Unavailable),
             Ok(Err(error)) => Err(classify(&error)),
@@ -402,20 +403,28 @@ impl SaslAuthenticator for IggyAuthenticator {
             Ok(Ok(())) => fetch_permissions(&client, &credentials.username).await,
         };
 
-        // Shut down rather than drop, so the socket closes cleanly and Iggy releases the session
-        // now instead of noticing the dead peer later.
+        // Sign out first: a shutdown only closes the socket, and Iggy keeps the session in its
+        // clients table until the lease runs out, 36 s by default. A full table refuses every new
+        // login, so a burst of Kafka logins would lock out the cluster's other clients. Then shut
+        // down rather than drop, so the socket closes cleanly.
         //
         // Both failure shapes are reported, and separately: the outer `Elapsed` and the inner
         // `IggyError` mean different things, and folding them together would log "timed out" for a
         // shutdown that failed immediately. Neither leaks the client, since dropping it closes the
         // socket regardless, but an invisible failure here would hide a real one.
-        match tokio::time::timeout(TEARDOWN_TIMEOUT, client.shutdown()).await {
+        let teardown = async {
+            if signed_in && let Err(error) = client.logout_user().await {
+                debug!(%error, "a credential-verification client did not sign out");
+            }
+            client.shutdown().await
+        };
+        match tokio::time::timeout(TEARDOWN_TIMEOUT, teardown).await {
             Ok(Ok(())) => {}
             Ok(Err(error)) => {
                 warn!(%error, "failed to shut down a credential-verification client");
             }
             Err(_elapsed) => {
-                warn!("timed out shutting down a credential-verification client");
+                warn!("timed out signing out and shutting down a credential-verification client");
             }
         }