Struct TenantAwareInMemoryTaskStore 

pub struct TenantAwareInMemoryTaskStore { /* private fields */ }

Tenant-isolated in-memory TaskStore.

Maintains a separate InMemoryTaskStore per tenant, providing full data isolation between tenants. The current tenant is determined from TenantContext.

Usage

use a2a_protocol_server::store::tenant::{TenantAwareInMemoryTaskStore, TenantContext};
use a2a_protocol_server::store::TaskStore;

let store = TenantAwareInMemoryTaskStore::new();

// Tenant A saves a task
TenantContext::scope("tenant-a", async {
    let task = Task {
        id: TaskId::new("task-1"),
        context_id: ContextId::new("ctx-1"),
        status: TaskStatus::with_timestamp(TaskState::Submitted),
        history: None,
        artifacts: None,
        metadata: None,
    };
    store.save(&task).await.unwrap();
}).await;

// Tenant B cannot see tenant A's task
TenantContext::scope("tenant-b", async {
    let result = store.get(&TaskId::new("task-1")).await.unwrap();
    assert!(result.is_none());
}).await;

Implementations

impl TenantAwareInMemoryTaskStore

pub fn new() -> TenantAwareInMemoryTaskStore

Creates a new tenant-aware store with default configuration.

pub fn with_config(config: TenantStoreConfig) -> TenantAwareInMemoryTaskStore

Creates a new tenant-aware store with custom configuration.

pub fn with_metrics( self, metrics: MetricsHandle, ) -> TenantAwareInMemoryTaskStore

Sends every partition’s store metrics to metrics.

Each partition is an InMemoryTaskStore and reports its own dropped appends; without this they all report to the no-op sink, so the one deployment shape that most needs the signal is the one that has none.

A partition is created on a tenant’s first use, so this must be set before the store serves traffic — a partition already built keeps the handle it was given.

pub fn with_tenant_override( self, tenant: impl Into<String>, config: TaskStoreConfig, ) -> TenantAwareInMemoryTaskStore

Gives tenant its own TaskStoreConfig, overriding TenantStoreConfig::per_tenant for that tenant alone.

This is the per-tenant store bound that TenantLimits::max_stored_tasks claimed to be and could not: that field sits on PerTenantConfig, which the handler holds and a store never sees. Here the store owns the map and reads it as it creates the partition.

Every field of TaskStoreConfig is overridden, not just capacity — a tenant can be given its own TTL, eviction interval and page cap too. The override replaces the whole config rather than merging, so start from per_tenant (or the default) and set the one field you mean to change:

use a2a_protocol_server::{TaskStoreConfig, TenantAwareInMemoryTaskStore};

let store = TenantAwareInMemoryTaskStore::new().with_tenant_override(
    "small-fry",
    TaskStoreConfig::default().with_max_capacity(Some(100)),
);

A partition is created on a tenant’s first use, so an override for a tenant that already has one applies only to a store built afterwards.

pub async fn tenant_count(&self) -> usize

Returns the number of active tenant partitions.

pub async fn run_eviction_all(&self)

Runs eviction on all tenant stores.

Call periodically to clean up terminal tasks in idle tenants.

pub async fn prune_empty_tenants(&self)

Removes empty tenant partitions to reclaim memory.

A partition is empty when it holds neither a task nor an idempotency key, so this reclaims a slot only once every task in that partition is gone — evicted by TTL, by capacity, or deleted — and no key is still claimed. It is the only way a max_tenants slot is ever given back, and nothing calls it for you: schedule it alongside run_eviction_all, which is what makes partitions empty in the first place.

Why the key count is part of the test

It decided on count() alone, which counts tasks. A key deliberately outlives the task it names — dropping it would let a delayed retry of the same send execute a second time — so a partition whose tasks had all been swept still held a live idempotency index, and pruning it destroyed that index along with the partition. The next retry carrying one of those keys found no claim, and ran the send a second time: the one outcome an idempotency key exists to prevent, reached by a memory-reclamation path that looked unrelated to it.

Calling it will not relieve a store that is at its cap because of live tasks; see max_tenants for why that matters.

Trait Implementations

impl Debug for TenantAwareInMemoryTaskStore

fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), Error>

Formats the value using the given formatter.

impl Default for TenantAwareInMemoryTaskStore

fn default() -> TenantAwareInMemoryTaskStore

Returns the “default value” for a type.

impl TaskStore for TenantAwareInMemoryTaskStore

fn supports_idempotency(&self) -> bool

Every tenant gets its own InMemoryTaskStore, and that partition is exactly the scope an idempotency key needs: one tenant’s key can never name another tenant’s task, because it is never in the same index.

fn supports_event_log(&self) -> bool

Every tenant gets its own InMemoryTaskStore, and each of those keeps a log, so the partition is the scope: one tenant’s task id can never reach another tenant’s events because it is never in the same map.

fn claim_idempotency_key<'a>( &'a self, key: &'a str, message_id: &'a MessageId, task_id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<IdempotencyClaim, A2aError>> + Send + 'a>>

Atomically claims key for task_id on behalf of message_id.

fn release_idempotency_key<'a>( &'a self, key: &'a str, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Releases a key claimed by claim_idempotency_key, so it can be claimed again.

fn save<'a>( &'a self, task: &'a Task, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Saves (creates or updates) a task.

fn get<'a>( &'a self, id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<Option<Task>, A2aError>> + Send + 'a>>

Retrieves a task by its ID, returning None if not found.

fn list<'a>( &'a self, params: &'a ListTasksParams, ) -> Pin<Box<dyn Future<Output = Result<TaskListResponse, A2aError>> + Send + 'a>>

Lists tasks matching the given filter parameters.

fn insert_if_absent<'a>( &'a self, task: &'a Task, ) -> Pin<Box<dyn Future<Output = Result<bool, A2aError>> + Send + 'a>>

Atomically inserts a task only if no task with the same ID exists.

fn delete<'a>( &'a self, id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Deletes a task by its ID.

fn append_event<'a>( &'a self, task_id: &'a TaskId, seq: u64, event: &'a StreamResponse, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Appends one event at seq in the task’s log.

fn last_event_seq<'a>( &'a self, task_id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<u64, A2aError>> + Send + 'a>>

The highest seq this task’s log holds, or 0 when it is empty.

fn earliest_event_seq<'a>( &'a self, task_id: &'a TaskId, ) -> Pin<Box<dyn Future<Output = Result<Option<u64>, A2aError>> + Send + 'a>>

The lowest seq this task’s log still holds, or None when it holds nothing — or when the store cannot say, which is the default.

fn read_events<'a>( &'a self, task_id: &'a TaskId, after_seq: u64, limit: usize, ) -> Pin<Box<dyn Future<Output = Result<Vec<RecordedEvent>, A2aError>> + Send + 'a>>

Reads a task’s events in order, starting after after_seq.

fn count<'a>( &'a self, ) -> Pin<Box<dyn Future<Output = Result<u64, A2aError>> + Send + 'a>>

Returns the total number of tasks in the store.

fn save_artifact_delta<'a>( &'a self, task: &'a Task, delta: ArtifactDelta, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Persists an artifact change that has already been applied to task.

fn save_status_delta<'a>( &'a self, task: &'a Task, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Persists a task whose status alone changed, without rewriting the rest of the record.

fn save_appending_history<'a>( &'a self, task: &'a Task, messages: &'a [Message], max_history: usize, ) -> Pin<Box<dyn Future<Output = Result<(), A2aError>> + Send + 'a>>

Persists a task whose history grew by messages, without the caller shipping the history it has already stored.

fn event_log_covers<'a>( &'a self, task_id: &'a TaskId, after_seq: u64, ) -> Pin<Box<dyn Future<Output = Result<bool, A2aError>> + Send + 'a>>

Whether a replay starting after after_seq can be served in full.