feat(server-ng): enforce PAT limits and expose Prometheus metrics

core/metadata/src/stm/user.rs

@@ -205,6 +205,18 @@ impl UsersInner {
         tokens.sort_by(|(left, _), (right, _)| left.cmp(right));
         tokens
     }
+
+    /// Count of live personal access tokens for one user.
+    ///
+    /// Single-map read for the ingress-side create cap; the sibling
+    /// `personal_access_tokens_of` allocates and sorts, too heavy for a
+    /// per-request check.
+    #[must_use]
+    pub fn pat_count_of(&self, user_id: UserId) -> usize {
+        self.personal_access_tokens
+            .get(&user_id)
+            .map_or(0, |user_tokens| user_tokens.len())
+    }
 }
 
 impl Users {
@@ -950,6 +962,32 @@ mod tests {
         assert!(users.personal_access_token_index.is_empty());
     }
 
+    #[test]
+    fn pat_count_of_tracks_create_and_delete() {
+        let mut users = UsersInner::new();
+        assert_eq!(users.pat_count_of(9), 0);
+
+        for (name, hash_byte) in [("first", b'a'), ("second", b'b')] {
+            CreatePersonalAccessTokenRequest {
+                user_id: 9,
+                name: WireName::new(name).unwrap(),
+                expiry: 0,
+                token_hash: [hash_byte; PAT_TOKEN_HASH_BYTES],
+            }
+            .apply(&mut users, IggyTimestamp::now());
+        }
+        assert_eq!(users.pat_count_of(9), 2);
+        assert_eq!(users.pat_count_of(1), 0, "count is scoped per user");
+
+        DeletePersonalAccessTokenRequest {
+            user_id: 9,
+            name: WireName::new("first").unwrap(),
+            only_if_expired: false,
+        }
+        .apply(&mut users, IggyTimestamp::now());
+        assert_eq!(users.pat_count_of(9), 1);
+    }
+
     #[test]
     fn delete_keeps_expiry_index_in_sync() {
         let mut users = UsersInner::new();

core/server-ng/Cargo.toml

@@ -57,7 +57,6 @@ ignored = [
     "opentelemetry-semantic-conventions",
     "opentelemetry_sdk",
     "papaya",
-    "prometheus-client",
     "rand",
     "ringbuffer",
     "rmp-serde",

core/server-ng/src/bootstrap.rs

@@ -185,6 +185,7 @@ pub fn wire_shell_handlers<B, MJ, S>(
     bus: &B,
     shard_handle: &ShellShardHandle<B, MJ, S>,
     system_config: Arc<NgSystemConfig>,
+    max_tokens_per_user: u32,
 ) -> ShellHandlers
 where
     B: ShellBus,
@@ -200,6 +201,7 @@ where
             shard_handle,
             &sessions,
             system_config,
+            max_tokens_per_user,
         ),
         on_metadata_submit: make_metadata_submit_handler(shard_handle),
         on_list_clients: make_list_clients_handler(&sessions),
@@ -1256,8 +1258,12 @@ async fn shard_main(
             boot_view,
             shard.plane.metadata().client_table.borrow().client_ids(),
         );
-        let on_client_request =
-            make_client_request_handler(&shard, &sessions, Arc::clone(&config.system));
+        let on_client_request = make_client_request_handler(
+            &shard,
+            &sessions,
+            Arc::clone(&config.system),
+            config.personal_access_token.max_tokens_per_user,
+        );
         let (accepted_replica, dialed_replica) =
             make_replica_delegation_fns(Rc::clone(&coord), &bus);
         let accepted_client = make_shard_zero_client_accept_fns(coord, &bus, on_client_request);
@@ -1685,7 +1691,12 @@ async fn build_shard_for_thread(
         on_list_clients,
         on_partition_read,
         sessions,
-    } = wire_shell_handlers(&bus, &shard_handle, Arc::clone(&config.system));
+    } = wire_shell_handlers(
+        &bus,
+        &shard_handle,
+        Arc::clone(&config.system),
+        config.personal_access_token.max_tokens_per_user,
+    );
     sessions
         .borrow_mut()
         .set_cluster_roster(Rc::new(build_cluster_roster(
@@ -2464,6 +2475,7 @@ async fn start_tcp_runtime(
             http_addr,
             &config.http,
             config.metadata.clients_table_max,
+            config.personal_access_token.max_tokens_per_user,
             &config.cluster,
             Arc::clone(&config.system),
             self_ports,

core/server-ng/src/dispatch.rs

@@ -116,6 +116,7 @@ pub(crate) fn make_client_request_handler<B, MJ, S>(
     shard: &Rc<ShellShard<B, MJ, S>>,
     sessions: &Rc<RefCell<SessionManager>>,
     system_config: Arc<NgSystemConfig>,
+    max_tokens_per_user: u32,
 ) -> RequestHandler
 where
     B: ShellBus,
@@ -144,6 +145,7 @@ where
             Rc::clone(&shard),
             Rc::clone(&sessions),
             Arc::clone(&system_config),
+            max_tokens_per_user,
             Rc::clone(&queues),
             Rc::clone(&active),
             client_id,
@@ -444,6 +446,7 @@ pub(crate) fn make_deferred_client_request_handler<B, MJ, S>(
     shard_handle: &ShellShardHandle<B, MJ, S>,
     sessions: &Rc<RefCell<SessionManager>>,
     system_config: Arc<NgSystemConfig>,
+    max_tokens_per_user: u32,
 ) -> RequestHandler
 where
     B: ShellBus,
@@ -486,7 +489,16 @@ where
                 active.borrow_mut().remove(&client_id);
                 return;
             };
-            drain_client_requests(shard, sessions, system_config, queues, active, client_id).await;
+            drain_client_requests(
+                shard,
+                sessions,
+                system_config,
+                max_tokens_per_user,
+                queues,
+                active,
+                client_id,
+            )
+            .await;
         });
     })
 }
@@ -620,10 +632,12 @@ where
 // An unbound transport sending a replicated frame therefore gets the
 // empty-reply fail-fast below and must log in.
 
+#[allow(clippy::too_many_arguments)]
 fn enqueue_client_request<B, MJ, S>(
     shard: Rc<ShellShard<B, MJ, S>>,
     sessions: Rc<RefCell<SessionManager>>,
     system_config: Arc<NgSystemConfig>,
+    max_tokens_per_user: u32,
     queues: ClientRequestQueues,
     active: ActiveClientRequests,
     client_id: u128,
@@ -645,7 +659,16 @@ fn enqueue_client_request<B, MJ, S>(
 
     let bus = shard.bus.clone();
     bus.spawn(async move {
-        drain_client_requests(shard, sessions, system_config, queues, active, client_id).await;
+        drain_client_requests(
+            shard,
+            sessions,
+            system_config,
+            max_tokens_per_user,
+            queues,
+            active,
+            client_id,
+        )
+        .await;
     });
 }
 
@@ -654,6 +677,7 @@ async fn drain_client_requests<B, MJ, S>(
     shard: Rc<ShellShard<B, MJ, S>>,
     sessions: Rc<RefCell<SessionManager>>,
     system_config: Arc<NgSystemConfig>,
+    max_tokens_per_user: u32,
     queues: ClientRequestQueues,
     active: ActiveClientRequests,
     client_id: u128,
@@ -667,7 +691,15 @@ async fn drain_client_requests<B, MJ, S>(
         let Some(message) = pop_next_client_request(&queues, &active, client_id) else {
             return;
         };
-        handle_client_request(&shard, &sessions, &system_config, client_id, message).await;
+        handle_client_request(
+            &shard,
+            &sessions,
+            &system_config,
+            max_tokens_per_user,
+            client_id,
+            message,
+        )
+        .await;
     }
 }
 
@@ -696,6 +728,7 @@ async fn handle_client_request<B, MJ, S>(
     shard: &Rc<ShellShard<B, MJ, S>>,
     sessions: &Rc<RefCell<SessionManager>>,
     system_config: &Arc<NgSystemConfig>,
+    max_tokens_per_user: u32,
     transport_client_id: u128,
     message: Message<iggy_binary_protocol::GenericHeader>,
 ) where
@@ -856,19 +889,48 @@ async fn handle_client_request<B, MJ, S>(
             new_header.session = bound_session;
         }
     });
-    let (request, raw_pat_token) =
-        match maybe_rewrite_pat_request(sessions, transport_client_id, request) {
-            Ok(rewritten) => rewritten,
-            Err(error) => {
+    let (request, raw_pat_token) = match maybe_rewrite_pat_request(
+        sessions,
+        transport_client_id,
+        max_tokens_per_user,
+        |user_id| {
+            shard
+                .plane
+                .metadata()
+                .mux_stm
+                .users()
+                .read(|users| users.pat_count_of(user_id))
+        },
+        request,
+    ) {
+        Ok(rewritten) => rewritten,
+        Err(error) => {
+            // Pre-consensus rejection (token cap reached, malformed body, or a
+            // lost session binding): deny fast with the typed code. A silent
+            // drop would wedge every later request on the connection until the
+            // socket read timeout.
+            warn!(
+                transport_client_id,
+                error = %error,
+                operation = ?header.operation,
+                "denying personal-access-token request"
+            );
+            let commit = current_metadata_commit(shard);
+            let reply = build_deny_reply(&header, transport_client_id, 0, commit, error.as_code());
+            if let Err(send_error) = shard
+                .bus
+                .send_to_client(transport_client_id, reply.into_generic().into_frozen())
+                .await
+            {
                 warn!(
                     transport_client_id,
-                    error = %error,
-                    operation = ?header.operation,
-                    "dropping request with invalid PAT replication context"
+                    error = %send_error,
+                    "failed to send personal-access-token deny reply"
                 );
-                return;
             }
-        };
+            return;
+        }
+    };
     // Hash raw passwords and, for ChangePassword, verify the current password
     // on the primary before replication; see `crate::users`. Replicas store the
     // hash directly. A wrong current password is not denied here: it rides