feat(api-keys): add synchronous v2 bulk operations

src/control-plane-services/api-keys/README.md

@@ -3,6 +3,39 @@
 NVIDIA API Keys service manages the lifecycle of API keys (issue,
 authenticate, introspect, rotate, revoke).
 
+## Synchronous bulk operations
+
+`SyncKeysV2OperationsService` provides key-owner, account-admin and platform-admin
+entry points. HTTP integration with the actor controllers is pending. The key-owner
+entry point fixes the user, and the account-admin entry point fixes the account.
+
+Set `operation` to `REVOKED` or `DELETE`. Each dimension not fixed by the actor
+requires an `accounts`, `users`, or `issuerServices` array of distinct IDs or
+the singleton `["*"]`. Platform requests selecting all three wildcards also
+require `confirm_platform_wide: true`. Configure the array limit with
+`apikeys.bulk-max-selector-values` (default 10).
+
+The service returns `KeyOperationResponse` after processing finishes. Selection
+uses the submission time as its creation cutoff. Processing stops on failure,
+and the exception's problem details include `bulkOperation` with `FAILED` status and counts
+for eligible keys processed so far. `partialApplication` indicates that a mutation was
+attempted. Earlier writes remain applied. Retries start a new selection with
+a new cutoff.
+
+A failure between v2 and legacy writes can partially update one migrated key.
+The operation reports failure until both representations are updated. Retrying
+repeats the per-key mutation to repair partial writes.
+
+Selection fails when a legacy key lacks the creation time or issuer mapping
+needed to process it. Correct the record or deployment mapping before retrying.
+
+Bulk mutations clear cached key rows and introspection responses on this instance
+before the response, including after partial failure. Other replicas expire cached key
+rows after 60 seconds and cached introspection responses after 60 seconds.
+These layers can retain a successful decision for up to 120 seconds after a
+mutation, plus in-flight request time. Existing caller-side HTTP caching can
+add another 360 seconds for introspection or 60 seconds for evaluation.
+
 ## Minimum Requirements
 
 - [Eclipse Temurin OpenJDK 25](https://adoptium.net/temurin/releases/)

src/control-plane-services/api-keys/src/main/java/com/nvidia/apikeys/caching/PrefetchingCache.java

@@ -52,8 +52,7 @@ public class PrefetchingCache<K, V> {
     private final LoadingCache<K, Instant> keyLastFetchTime;
 
     /**
-     * Methods invalidate all cache records. Not intended to be used in normal flows, only during
-     * testing.
+     * Clears cached values and prefetch tracking after management mutations.
      */
     public void invalidate() {
         records.clear();

src/control-plane-services/api-keys/src/main/java/com/nvidia/apikeys/config/NakProperties.java

@@ -59,6 +59,7 @@ public class NakProperties {
     private String dataDomainKey;
     @NotNull
     private Duration keepAfterExpiredDuration;
+    private int bulkMaxSelectorValues;
 
     @NotNull
     @Valid

src/control-plane-services/api-keys/src/main/java/com/nvidia/apikeys/persistance/dao/KeysDao.java

@@ -34,10 +34,14 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.stream.Stream;
 import lombok.RequiredArgsConstructor;
 import org.springframework.data.cassandra.core.CassandraBatchOperations;
 import org.springframework.data.cassandra.core.CassandraTemplate;
 import org.springframework.data.cassandra.core.WriteResult;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
+import org.springframework.data.cassandra.core.query.Criteria;
+import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.stereotype.Service;
 
 @Service
@@ -160,6 +164,24 @@ public List<KeyByOwnerAndServiceVo> list(KeyOwnerType ownerType, String ownerId)
                 .toList();
     }
 
+    /** Streams legacy rows in bounded driver pages, scanning when no owner is specified. */
+    public Stream<KeyByOwnerAndServiceVo> streamForBulk(
+            KeyOwnerType ownerType, String ownerId, String serviceId, int pageSize) {
+        var query = Query.empty();
+        if (ownerId != null) {
+            query = query.and(Criteria.where("owner_type").is(ownerType))
+                    .and(Criteria.where("owner_id").is(ownerId));
+            if (serviceId != null) {
+                query = query.and(Criteria.where("issuer_service_id").is(serviceId));
+            }
+        }
+        return cassandraTemplate.stream(query.queryOptions(QueryOptions.builder().pageSize(pageSize).build()),
+                        KeyByOwnerAndServiceModel.class)
+                .filter(row -> row.getKeyId() != null && row.getOwnerType() == ownerType)
+                .filter(row -> serviceId == null || serviceId.equals(row.getIssuerServiceId()))
+                .map(this::modelToValidatedVo);
+    }
+
 
     private KeyVo modelToValidatedVo(KeyModel keyModel) {
         return keyConverter.modelToVo(keyModel);

src/control-plane-services/api-keys/src/main/java/com/nvidia/apikeys/persistance/dao/KeysV2Dao.java

@@ -30,6 +30,7 @@
 import com.nvidia.apikeys.vo.KeysByAccountSliceVo;
 import com.nvidia.apikeys.vo.KeysByOwnerAndAccountAndServiceVo;
 import java.nio.ByteBuffer;
+import java.time.Instant;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -38,7 +39,10 @@
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.data.cassandra.core.CassandraTemplate;
+import org.springframework.data.cassandra.core.cql.QueryOptions;
 import org.springframework.data.cassandra.core.query.CassandraPageRequest;
+import org.springframework.data.cassandra.core.query.Criteria;
+import org.springframework.data.cassandra.core.query.Query;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.domain.Slice;
@@ -120,6 +124,26 @@ public KeysByAccountSliceVo list(String ncaId, int limit, String cursor) {
                 limit, cursor);
     }
 
+    /** Streams indexed bulk matches in driver-sized pages; null dimensions are unrestricted. */
+    public Stream<KeysByOwnerAndAccountAndServiceVo> streamForBulk(
+            String ncaId, KeyOwnerType ownerType, String ownerId, String issuerServiceId,
+            Instant cutoff, int pageSize) {
+        var query = Query.query(Criteria.where("owner_type").is(ownerType))
+                .and(Criteria.where("created_at").lte(cutoff));
+        if (ncaId != null) {
+            query = query.and(Criteria.where("nca_id").is(ncaId));
+        }
+        if (ownerId != null) {
+            query = query.and(Criteria.where("owner_id").is(ownerId));
+        }
+        if (issuerServiceId != null) {
+            query = query.and(Criteria.where("issuer_service_id").is(issuerServiceId));
+        }
+        return cassandraTemplate.stream(query.queryOptions(QueryOptions.builder().pageSize(pageSize).build()),
+                        KeysByOwnerAndAccountAndServiceModel.class)
+                .map(keyConverter::modelToVo);
+    }
+
     public KeysByAccountSliceVo list(
             String ncaId, String issuerServiceId, int limit, String cursor) {
         return slice(

src/control-plane-services/api-keys/src/main/java/com/nvidia/apikeys/services/KeysV2BulkSelectionService.java

@@ -0,0 +1,87 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nvidia.apikeys.services;
+
+import static com.nvidia.apikeys.vo.KeyOwnerType.USER;
+
+import com.nvidia.apikeys.persistance.dao.KeysDao;
+import com.nvidia.apikeys.persistance.dao.KeysV2Dao;
+import com.nvidia.apikeys.vo.KeyByOwnerAndServiceVo;
+import com.nvidia.apikeys.vo.KeysByOwnerAndAccountAndServiceVo;
+import java.time.Instant;
+import java.util.stream.Stream;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+/** Paged discovery for bulk selectors, including the configured legacy account. */
+@Service
+public class KeysV2BulkSelectionService {
+
+    private final KeysV2Dao keysV2Dao;
+    private final KeysDao keysDao;
+    private final LegacyAccountTranslator legacyAccount;
+    private final int pageSize;
+
+    public KeysV2BulkSelectionService(
+            KeysV2Dao keysV2Dao, KeysDao keysDao, LegacyAccountTranslator legacyAccount,
+            @Value("${spring.cassandra.request.page-size:5000}") int pageSize) {
+        this.keysV2Dao = keysV2Dao;
+        this.keysDao = keysDao;
+        this.legacyAccount = legacyAccount;
+        this.pageSize = pageSize;
+    }
+
+    /** Null dimensions select all values. The caller closes the returned stream. */
+    public Stream<KeysByOwnerAndAccountAndServiceVo> select(
+            String account, String user, String issuer, Instant cutoff) {
+        var current = keysV2Dao.streamForBulk(account, USER, user, issuer, cutoff, pageSize);
+        try {
+            // Exclude migrated legacy copies before mutations can delete their v2 rows.
+            return Stream.concat(selectLegacy(account, user, issuer, cutoff), current);
+        } catch (RuntimeException | Error failure) {
+            current.close();
+            throw failure;
+        }
+    }
+
+    private Stream<KeysByOwnerAndAccountAndServiceVo> selectLegacy(
+            String account, String user, String issuer, Instant cutoff) {
+        if (account != null && !legacyAccount.isLegacy(account)) {
+            return Stream.empty();
+        }
+        var legacyIssuer = issuer == null ? null
+                : legacyAccount.findLegacyIssuerServiceId(issuer).orElse(null);
+        if (issuer != null && legacyIssuer == null) {
+            return Stream.empty();
+        }
+        return keysDao.streamForBulk(USER, user, legacyIssuer, pageSize)
+                .filter(key -> key.getCreatedAt() == null || !key.getCreatedAt().isAfter(cutoff))
+                .map(this::validateAndTranslateLegacyKey)
+                // A migrated v2 row wins even when its creation time is after the cutoff.
+                .filter(key -> keysV2Dao.get(key.getNcaId(), USER, key.getOwnerId(),
+                        key.getIssuerServiceId(), key.getKeyId()).isEmpty());
+    }
+
+    private KeysByOwnerAndAccountAndServiceVo validateAndTranslateLegacyKey(KeyByOwnerAndServiceVo key) {
+        if (legacyAccount.findLegacyIssuerServiceId(legacyAccount.toServiceId(key.getIssuerServiceId()))
+                .filter(key.getIssuerServiceId()::equals).isEmpty()) {
+            throw new IllegalStateException("Legacy key issuer has no configured mapping");
+        }
+        return legacyAccount.fromLegacy(key);
+    }
+}

src/control-plane-services/api-keys/src/main/java/com/nvidia/apikeys/services/KeysV2Service.java

@@ -49,6 +49,7 @@
 import com.nvidia.boot.exceptions.NotFoundException;
 import com.nvidia.boot.exceptions.TooManyRequestsException;
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -153,22 +154,33 @@ public List<KeysByOwnerAndAccountAndServiceVo> listKeys(
     }
 
     /**
-     * Sets ACTIVE, SUSPENDED, or REVOKED. EXPIRED and REVOKED keys keep their status.
+     * Sets ACTIVE, SUSPENDED, or REVOKED. Revocation also denies retained expired keys and
+     * existing legacy copies, and rewrites both stored rows on retry.
      */
     public KeyV2Vo updateKeyStatus(
             String ncaId, KeyOwnerType ownerType, String ownerId, String issuerServiceId,
             String keyId, KeyStatus status) {
         var stored = load(ncaId, ownerType, ownerId, issuerServiceId, keyId);
         var key = stored.key();
-        if (key.getKeyStatus() == status) {
+        assertKeyScope(key, ncaId, ownerType, ownerId, issuerServiceId, keyId);
+        if (key.getKeyStatus() == status && status != KeyStatus.REVOKED) {
             return key;
         }
-        if (FINAL_KEY_STATUSES.contains(key.getKeyStatus())) {
+        if (FINAL_KEY_STATUSES.contains(key.getKeyStatus()) && status != KeyStatus.REVOKED) {
             throw new ConflictException(MESG_KEY_STATUS_FINAL.formatted(key.getKeyStatus()));
         }
 
+        var legacyCopy = status == KeyStatus.REVOKED && !stored.legacy()
+                ? findLegacyKeyRow(ncaId, ownerType, ownerId, issuerServiceId, keyId)
+                        .map(row -> keysDao.getKeyByHash(row.getKeyHash())
+                                .orElseThrow(() -> new NotFoundException(MESG_KEY_NOT_FOUND)))
+                        .map(legacyAccount::fromLegacy)
+                : Optional.<KeyV2Vo>empty();
+        legacyCopy.ifPresent(copy -> assertKeyScope(copy, ncaId, ownerType, ownerId, issuerServiceId, keyId));
+
         var updated = key.toBuilder().keyStatus(status).build();
         save(stored, updated);
+        legacyCopy.ifPresent(copy -> save(new StoredKey(copy, true), copy.toBuilder().keyStatus(status).build()));
         recordKeyAction(updated, switch (status) {
             case SUSPENDED -> SUSPEND;
             case REVOKED -> REVOKE;
@@ -208,6 +220,12 @@ public void deleteKeyById(
             throw new NotFoundException(MESG_KEY_NOT_FOUND);
         }
 
+        v2Key.flatMap(row -> keysV2Dao.getKeyByHash(row.getKeyHash()))
+                .ifPresent(key -> assertKeyScope(key, ncaId, ownerType, ownerId, issuerServiceId, keyId));
+        legacyKey.flatMap(row -> keysDao.getKeyByHash(row.getKeyHash()))
+                .map(legacyAccount::fromLegacy)
+                .ifPresent(key -> assertKeyScope(key, ncaId, ownerType, ownerId, issuerServiceId, keyId));
+
         v2Key.ifPresent(keysV2Dao::delete);
         legacyKey.ifPresent(keysDao::deleteKey);
         v2Key.ifPresentOrElse(
@@ -308,6 +326,17 @@ private void save(StoredKey stored, KeyV2Vo updated) {
         }
     }
 
+    private static void assertKeyScope(
+            KeyV2Vo key, String ncaId, KeyOwnerType ownerType, String ownerId,
+            String issuerServiceId, String keyId) {
+        if (!Objects.equals(key.getNcaId(), ncaId) || key.getOwnerType() != ownerType
+                || !Objects.equals(key.getOwnerId(), ownerId)
+                || !Objects.equals(key.getIssuerServiceId(), issuerServiceId)
+                || !Objects.equals(key.getKeyId(), keyId)) {
+            throw new BadRequestException("Key does not match the requested scope");
+        }
+    }
+
     /**
      * The legacy account keeps owner status with the owner's legacy keys.
      */