diff --git a/docs/content/dev/configuration.md b/docs/content/dev/configuration.md index 2458b9c09..8b3fea787 100644 --- a/docs/content/dev/configuration.md +++ b/docs/content/dev/configuration.md @@ -65,6 +65,15 @@ a2a.request-context.populate-referred-tasks=true When enabled, task IDs referenced in incoming messages are looked up in the `TaskStore` and made available to the `AgentExecutor` via `RequestContext.getRelatedTasks()`. This is useful for multi-task conversations where the agent needs access to state from related tasks. Enabled by default; set to `false` to avoid extra `TaskStore` lookups when not needed. +### Push Notification Config Store + +```properties +# Maximum push notification configs per task (default: 100) +a2a.push-notification-config.max-per-task=100 +``` + +Limits the number of distinct push notification configurations a single task may register. Each config consumes memory and can trigger an outbound HTTP request on every task event. Re-registering an existing config ID (updating it) does not count against the limit. Both `InMemoryPushNotificationConfigStore` and `JpaDatabasePushNotificationConfigStore` enforce this limit, throwing `InvalidParamsError` when exceeded. + ### Tuning Guidelines - **Streaming Performance**: The executor handles streaming subscriptions. Too few threads can cause timeouts under concurrent load. diff --git a/extras/push-notification-config-store-database-jpa/src/main/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStore.java b/extras/push-notification-config-store-database-jpa/src/main/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStore.java index a651d7634..dee737d4b 100644 --- a/extras/push-notification-config-store-database-jpa/src/main/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStore.java +++ b/extras/push-notification-config-store-database-jpa/src/main/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStore.java @@ -7,12 +7,15 @@ import jakarta.annotation.Priority; import jakarta.enterprise.context.ApplicationScoped; import jakarta.enterprise.inject.Alternative; +import jakarta.inject.Inject; import jakarta.persistence.EntityManager; import jakarta.persistence.PersistenceContext; import jakarta.transaction.Transactional; import org.a2aproject.sdk.jsonrpc.common.json.JsonProcessingException; +import org.a2aproject.sdk.server.config.A2AConfigProvider; import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore; +import org.a2aproject.sdk.spec.InvalidParamsError; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult; import org.a2aproject.sdk.spec.TaskPushNotificationConfig; @@ -34,6 +37,9 @@ public class JpaDatabasePushNotificationConfigStore implements PushNotificationC @PersistenceContext(unitName = "a2a-java") EntityManager em; + @Inject + A2AConfigProvider config; + @Transactional @Override public TaskPushNotificationConfig setInfo(TaskPushNotificationConfig notificationConfig) { @@ -66,6 +72,17 @@ public TaskPushNotificationConfig setInfo(TaskPushNotificationConfig notificatio LOGGER.debug("Updated existing PushNotificationConfig for Task '{}' with ID: {}", taskId, notificationConfig.id()); } else { + // Enforce the per-task limit (configurable via + // a2a.push-notification-config.max-per-task); only genuinely new + // configs count against it. + int maxPerTask = PushNotificationConfigStore.maxPushConfigsPerTask(config); + Long existingCount = em.createQuery( + "SELECT COUNT(c) FROM JpaPushNotificationConfig c WHERE c.id.taskId = :taskId", + Long.class).setParameter("taskId", taskId).getSingleResult(); + if (existingCount >= maxPerTask) { + throw new InvalidParamsError("Too many push notification configs for task " + taskId + + " (max " + maxPerTask + ")"); + } // Create new entity JpaPushNotificationConfig jpaConfig = JpaPushNotificationConfig.createFromConfig(taskId, notificationConfig, resolvedVersion); em.persist(jpaConfig); diff --git a/extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java b/extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java index 4ff600268..0d51bd039 100644 --- a/extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java +++ b/extras/push-notification-config-store-database-jpa/src/test/java/org/a2aproject/sdk/extras/pushnotificationconfigstore/database/jpa/JpaDatabasePushNotificationConfigStoreIntegrationTest.java @@ -28,6 +28,7 @@ import org.a2aproject.sdk.spec.AgentCard; import org.a2aproject.sdk.spec.DeleteTaskPushNotificationConfigParams; import org.a2aproject.sdk.spec.GetTaskPushNotificationConfigParams; +import org.a2aproject.sdk.spec.InvalidParamsError; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult; import org.a2aproject.sdk.spec.Message; @@ -581,4 +582,29 @@ private void createSamples(String taskId, int size) { } } } + + @Test + @Transactional + public void testPushConfigLimitEnforced() { + String taskId = "task_limit_" + System.currentTimeMillis(); + int limit = org.a2aproject.sdk.server.tasks.PushNotificationConfigStore.DEFAULT_MAX_PUSH_CONFIGS_PER_TASK; + + // Register configs up to the limit. + for (int i = 0; i < limit; i++) { + TaskPushNotificationConfig config = createSamplePushConfig( + "http://limit" + i + ".com/callback", "limit-cfg" + i, "token" + i); + pushNotificationConfigStore.setInfo(TaskPushNotificationConfig.builder(config).taskId(taskId).build()); + } + + // The (limit+1)-th distinct config must be rejected. + TaskPushNotificationConfig extra = createSamplePushConfig( + "http://extra.com/callback", "limit-extra", "token-extra"); + assertThrows(InvalidParamsError.class, () -> + pushNotificationConfigStore.setInfo(TaskPushNotificationConfig.builder(extra).taskId(taskId).build())); + + // Re-registering an existing config ID at the limit is still allowed. + TaskPushNotificationConfig existing = createSamplePushConfig( + "http://limit0.com/callback-updated", "limit-cfg0", "token-updated"); + pushNotificationConfigStore.setInfo(TaskPushNotificationConfig.builder(existing).taskId(taskId).build()); + } } diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStore.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStore.java index 0d239758f..5bc1c7ac8 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStore.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStore.java @@ -10,6 +10,8 @@ import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; +import org.a2aproject.sdk.server.config.A2AConfigProvider; +import org.a2aproject.sdk.spec.InvalidParamsError; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult; import org.a2aproject.sdk.spec.TaskPushNotificationConfig; @@ -24,9 +26,20 @@ @ApplicationScoped public class InMemoryPushNotificationConfigStore implements PushNotificationConfigStore { + /** + * Default maximum number of push notification configs allowed per task. + * Prevents a single task from accumulating an unbounded list of configs + * (each config consumes memory and can trigger outbound HTTP requests). + * Overridable via {@code a2a.push-notification-config.max-per-task}. + */ + public static final int MAX_PUSH_CONFIGS_PER_TASK = PushNotificationConfigStore.DEFAULT_MAX_PUSH_CONFIGS_PER_TASK; + private final Map> pushNotificationInfos = Collections.synchronizedMap(new HashMap<>()); private final Map protocolVersions = Collections.synchronizedMap(new HashMap<>()); + @Inject + @Nullable A2AConfigProvider configProvider; + @Inject public InMemoryPushNotificationConfigStore() { } @@ -41,14 +54,18 @@ public TaskPushNotificationConfig setInfo(TaskPushNotificationConfig notificatio } notificationConfig = builder.build(); - Iterator notificationConfigIterator = notificationConfigList.iterator(); - while (notificationConfigIterator.hasNext()) { - TaskPushNotificationConfig config = notificationConfigIterator.next(); - if (config.id() != null && config.id().equals(notificationConfig.id())) { - notificationConfigIterator.remove(); - break; - } + // Enforce the per-task limit and remove any existing config with the same + // ID in a single pass. Re-registering/updating an already-registered config + // ID is allowed; only genuinely new configs count against the limit. + String configId = notificationConfig.id(); + boolean isExistingConfig = notificationConfigList.removeIf( + existing -> existing.id() != null && existing.id().equals(configId)); + int maxPerTask = PushNotificationConfigStore.maxPushConfigsPerTask(configProvider); + if (!isExistingConfig && notificationConfigList.size() >= maxPerTask) { + throw new InvalidParamsError("Too many push notification configs for task " + taskId + + " (max " + maxPerTask + ")"); } + notificationConfigList.add(notificationConfig); pushNotificationInfos.put(taskId, notificationConfigList); return notificationConfig; diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationConfigStore.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationConfigStore.java index 5d2bf3392..64138f3aa 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationConfigStore.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/PushNotificationConfigStore.java @@ -1,5 +1,6 @@ package org.a2aproject.sdk.server.tasks; +import org.a2aproject.sdk.server.config.A2AConfigProvider; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult; import org.a2aproject.sdk.spec.TaskPushNotificationConfig; @@ -21,6 +22,13 @@ *
  • Used for retrieval and deletion of specific configurations
    • * * + *

    Per-task Limit

    + * Implementations MUST reject registering more than + * {@code a2a.push-notification-config.max-per-task} (default 100) distinct configs for + * a single task, throwing {@code InvalidParamsError}. Re-registering an already-registered + * config ID does not count against the limit. See + * {@link #maxPushConfigsPerTask(A2AConfigProvider)}. + * *

    Pagination Support

    * {@link #getInfo(ListTaskPushNotificationConfigsParams)} supports pagination for tasks * with many push notification configurations: @@ -70,6 +78,13 @@ */ public interface PushNotificationConfigStore { + /** + * Default maximum number of push notification configs allowed per task, + * used when {@code a2a.push-notification-config.max-per-task} is not + * configured. See {@code META-INF/a2a-defaults.properties}. + */ + int DEFAULT_MAX_PUSH_CONFIGS_PER_TASK = 100; + /** * Sets or updates the push notification configuration for a task. *

    @@ -105,6 +120,33 @@ static String resolveProtocolVersion(@Nullable String protocolVersion) { return protocolVersion != null ? protocolVersion : org.a2aproject.sdk.spec.AgentInterface.CURRENT_PROTOCOL_VERSION; } + /** + * Maximum number of push notification configs allowed per task. + * + *

    Reads the {@code a2a.push-notification-config.max-per-task} configuration value; + * a {@code null} provider, or a missing, non-numeric, or non-positive value, falls + * back to 100. See {@code META-INF/a2a-defaults.properties}.

    + * + * @param config the configuration provider, or {@code null} when the store is used + * without CDI injection (e.g., in tests) + * @return the per-task limit (always positive) + */ + static int maxPushConfigsPerTask(@Nullable A2AConfigProvider config) { + if (config == null) { + return DEFAULT_MAX_PUSH_CONFIGS_PER_TASK; + } + return config.getOptionalValue("a2a.push-notification-config.max-per-task") + .map(value -> { + try { + int parsed = Integer.parseInt(value.trim()); + return parsed > 0 ? parsed : DEFAULT_MAX_PUSH_CONFIGS_PER_TASK; + } catch (NumberFormatException e) { + return DEFAULT_MAX_PUSH_CONFIGS_PER_TASK; + } + }) + .orElse(DEFAULT_MAX_PUSH_CONFIGS_PER_TASK); + } + /** * Retrieves push notification configurations for a task with pagination support. *

    diff --git a/server-common/src/main/resources/META-INF/a2a-defaults.properties b/server-common/src/main/resources/META-INF/a2a-defaults.properties index e1a71fe24..0ab5b660d 100644 --- a/server-common/src/main/resources/META-INF/a2a-defaults.properties +++ b/server-common/src/main/resources/META-INF/a2a-defaults.properties @@ -32,3 +32,9 @@ a2a.executor.queue-capacity=100 # When true, referenced task IDs in messages are resolved from the TaskStore # and made available via RequestContext.getRelatedTasks() a2a.request-context.populate-referred-tasks=true + +# PushNotificationConfigStore - per-task configuration limit +# Maximum number of push notification configs a single task may register. +# Prevents a task from accumulating an unbounded list of configs (each config +# consumes memory and can trigger outbound HTTP requests). +a2a.push-notification-config.max-per-task=100 diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStoreTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStoreTest.java index 110168603..7c27efb9b 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStoreTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/InMemoryPushNotificationConfigStoreTest.java @@ -5,6 +5,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.never; @@ -17,6 +18,7 @@ import org.a2aproject.sdk.client.http.A2AHttpResponse; import org.a2aproject.sdk.common.A2AHeaders; import org.a2aproject.sdk.spec.AgentInterface; +import org.a2aproject.sdk.spec.InvalidParamsError; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult; import org.a2aproject.sdk.spec.Task; @@ -659,4 +661,55 @@ public void testPaginationFullIteration() { assertEquals(3, pageCount, "Should have exactly 3 pages (3+3+1)"); } + @Test + public void testSetInfoAtLimitExactlyAllowed() { + String taskId = "task_limit_exact"; + for (int i = 0; i < InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK; i++) { + configStore.setInfo(createSamplePushConfig(taskId, + "http://url" + i + ".com/callback", "cfg" + i, null)); + } + + ListTaskPushNotificationConfigsResult result = configStore.getInfo(new ListTaskPushNotificationConfigsParams(taskId)); + assertEquals(InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK, result.configs().size()); + } + + @Test + public void testSetInfoRejectsExceedingPerTaskLimit() { + String taskId = "task_limit_exceed"; + for (int i = 0; i < InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK; i++) { + configStore.setInfo(createSamplePushConfig(taskId, + "http://url" + i + ".com/callback", "cfg" + i, null)); + } + + // The (MAX+1)-th distinct config for the same task must be rejected + TaskPushNotificationConfig overflow = createSamplePushConfig(taskId, + "http://url-overflow.com/callback", "cfg-overflow", null); + assertThrows(InvalidParamsError.class, () -> configStore.setInfo(overflow)); + + // The store is unchanged + ListTaskPushNotificationConfigsResult result = configStore.getInfo(new ListTaskPushNotificationConfigsParams(taskId)); + assertEquals(InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK, result.configs().size()); + assertTrue(result.configs().stream().noneMatch(c -> "cfg-overflow".equals(c.id()))); + } + + @Test + public void testSetInfoUpdateExistingConfigAtLimitAllowed() { + String taskId = "task_limit_update"; + for (int i = 0; i < InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK; i++) { + configStore.setInfo(createSamplePushConfig(taskId, + "http://url" + i + ".com/callback", "cfg" + i, null)); + } + + // Updating an existing config at the limit must still be allowed + TaskPushNotificationConfig updated = createSamplePushConfig(taskId, + "http://url-updated.com/callback", "cfg0", "new-token"); + TaskPushNotificationConfig result = configStore.setInfo(updated); + + assertEquals("cfg0", result.id()); + ListTaskPushNotificationConfigsResult configs = configStore.getInfo(new ListTaskPushNotificationConfigsParams(taskId)); + assertEquals(InMemoryPushNotificationConfigStore.MAX_PUSH_CONFIGS_PER_TASK, configs.configs().size()); + assertEquals("http://url-updated.com/callback", + configs.configs().stream().filter(c -> "cfg0".equals(c.id())).findFirst().orElseThrow().url()); + } + }