Struct MemoryTaskStore 

pub struct MemoryTaskStore { /* private fields */ }

In-memory TaskStore backed by a HashMap.

This is the default store. Suitable for single-instance deployments. For horizontal scaling, use an external store that shares state across instances. A lazy background worker signals task expiry at its exact TTL deadline and physically removes expired records at the configured cleanup interval. It uses a standard thread rather than assuming construction happens inside a Tokio runtime, and holds only weak task state while sleeping.

Completion wakeups for wait_for_completion use a per-task tokio::sync::Notify, which is an implementation detail of this store.

Implementations

impl MemoryTaskStore

pub fn new() -> Self

Create a task store with the default lifecycle and retention policy.

That is a five-minute TTL, one-minute cleanup interval, and the finite TaskRetentionLimits::default byte/count bounds.

pub fn with_config(config: MemoryTaskStoreConfig) -> Self

Create a task store with an explicit lifecycle policy and the default finite retention limits.

pub fn with_retention_limits(retention_limits: TaskRetentionLimits) -> Self

Create a task store with the default lifecycle policy and explicit record and encoded-payload limits.

pub fn with_config_and_retention( config: MemoryTaskStoreConfig, retention_limits: TaskRetentionLimits, ) -> Self

Create a task store with explicit lifecycle and retention policies.

pub fn cleanup_expired(&self) -> usize

Remove expired tasks immediately.

Returns the number removed. Not part of the TaskStore trait; external backends typically expire entries natively (e.g. Redis TTL).

The configured worker already calls this retirement path periodically; applications may call it to reclaim memory sooner. Calling it is an optimization, not a correctness requirement: expiry has already cancelled work, woken waiters, and made the task read as absent.

Example

use tower_mcp::async_task::{MemoryTaskStore, TaskStore};

let store = MemoryTaskStore::new();
// A one millisecond retention window, and no terminal state: the
// clock runs from creation, so the task retires while still working.
let (id, _cancel) = store
    .create_task("deploy", serde_json::json!({}), Some(1), None)
    .await
    .unwrap();
tokio::time::sleep(std::time::Duration::from_millis(10)).await;

// Already invisible, before anything has been reclaimed.
assert!(store.get_task(&id).await.unwrap().is_none());
assert!(store.list_tasks(None).await.unwrap().is_empty());

// Cleanup only frees the memory the entry was still holding.
assert_eq!(store.cleanup_expired(), 1);
assert_eq!(store.cleanup_expired(), 0);

pub fn usage(&self) -> TaskStoreUsage

Return content-free count and encoded-byte gauges.

The snapshot is taken under the same lock as task mutations, so its count and both byte totals always describe one committed store state.

Trait Implementations

impl Clone for MemoryTaskStore

fn clone(&self) -> MemoryTaskStore

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for MemoryTaskStore

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

Formats the value using the given formatter.

impl Default for MemoryTaskStore

fn default() -> Self

Returns the “default value” for a type.

impl TaskStore for MemoryTaskStore

fn task_presence<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<TaskPresence>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

This store keeps expired records until cleanup_expired runs, so it can tell an owner that a task expired rather than that it never existed (#1249). One read resolves both, so expiry cannot change between deciding presence and reading the owner.

fn create_task<'life0, 'life1, 'async_trait>( &'life0 self, tool_name: &'life1 str, arguments: Value, ttl: Option<u64>, owner: TaskOwner, ) -> Pin<Box<dyn Future<Output = Result<(String, CancellationToken)>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Create and store a new task owned by owner.

fn get_task<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskObject>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Get task object by ID. Returns None if unknown.

fn set_task_meta<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, meta: Value, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Persist protocol _meta for a task.

fn discard_task<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Remove a task that could not finish initialization.

fn task_owner<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskOwner>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read a task’s owner.

fn get_task_result<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskSnapshot>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Get a task’s full snapshot (task object, result, error) by ID.

fn wait_for_completion<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskSnapshot>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Wait for a task to reach a terminal state, then return its snapshot.

fn list_tasks<'life0, 'async_trait>( &'life0 self, status_filter: Option<TaskStatus>, ) -> Pin<Box<dyn Future<Output = Result<Vec<TaskObject>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

List all tasks, optionally filtered by status.

fn require_input<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, task_id: &'life1 str, requests: InputRequests, message: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Mark a task as requiring input, recording the requests to be answered.

fn outstanding_input_requests<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<InputRequests>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Read the requests a task is currently waiting on.

fn apply_input_responses<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, responses: InputResponses, ) -> Pin<Box<dyn Future<Output = Result<Option<AppliedInputResponses>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Apply tasks/update.inputResponses to a task.

fn set_status<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, task_id: &'life1 str, status: TaskStatus, message: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Set a task’s non-terminal status and message.

fn resume_context<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskResumeContext>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Everything needed to re-invoke a task’s handler after its input requests were answered.

fn set_ttl<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ttl_ms: u64, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Update a task’s time-to-live, measured from creation.

fn complete_task<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, result: CallToolResult, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Mark a task as completed with a result.

fn fail_task<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, error: JsonRpcError, ) -> Pin<Box<dyn Future<Output = Result<bool>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Mark a task as failed with a structured execution error.

fn cancel_task<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, task_id: &'life1 str, reason: Option<&'life2 str>, ) -> Pin<Box<dyn Future<Output = Result<Option<TaskObject>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Cancel a task.

fn input_responses<'life0, 'life1, 'async_trait>( &'life0 self, task_id: &'life1 str, ) -> Pin<Box<dyn Future<Output = Result<Option<InputResponses>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Every answer accumulated for a task so far, keyed as issued.