1use std::collections::HashMap;
9use std::future::Future;
10use std::pin::Pin;
11use std::sync::Arc;
12
13use a2a_protocol_types::error::A2aResult;
14use a2a_protocol_types::params::ListTasksParams;
15use a2a_protocol_types::responses::TaskListResponse;
16use a2a_protocol_types::task::{Task, TaskId};
17use tokio::sync::RwLock;
18
19use super::super::task_store::{InMemoryTaskStore, TaskStore, TaskStoreConfig};
20use super::context::TenantContext;
21
22#[derive(Debug, Clone)]
30#[non_exhaustive]
31pub struct TenantStoreConfig {
32 pub per_tenant: TaskStoreConfig,
35
36 pub max_tenants: usize,
61}
62
63impl Default for TenantStoreConfig {
64 fn default() -> Self {
65 Self {
66 per_tenant: TaskStoreConfig::default(),
67 max_tenants: 1000,
68 }
69 }
70}
71
72impl TenantStoreConfig {
73 #[must_use]
75 pub const fn with_per_tenant(mut self, config: TaskStoreConfig) -> Self {
76 self.per_tenant = config;
77 self
78 }
79
80 #[must_use]
83 pub const fn with_max_tenants(mut self, max: usize) -> Self {
84 self.max_tenants = max;
85 self
86 }
87}
88
89#[derive(Debug)]
126pub struct TenantAwareInMemoryTaskStore {
127 stores: RwLock<HashMap<String, Arc<InMemoryTaskStore>>>,
128 config: TenantStoreConfig,
129 overrides: HashMap<String, TaskStoreConfig>,
138}
139
140impl Default for TenantAwareInMemoryTaskStore {
141 fn default() -> Self {
142 Self::new()
143 }
144}
145
146impl TenantAwareInMemoryTaskStore {
147 #[must_use]
149 pub fn new() -> Self {
150 Self {
151 stores: RwLock::new(HashMap::new()),
152 config: TenantStoreConfig::default(),
153 overrides: HashMap::new(),
154 }
155 }
156
157 #[must_use]
159 pub fn with_config(config: TenantStoreConfig) -> Self {
160 Self {
161 stores: RwLock::new(HashMap::new()),
162 config,
163 overrides: HashMap::new(),
164 }
165 }
166
167 #[must_use]
193 pub fn with_tenant_override(
194 mut self,
195 tenant: impl Into<String>,
196 config: TaskStoreConfig,
197 ) -> Self {
198 self.overrides.insert(tenant.into(), config);
199 self
200 }
201
202 async fn get_store(&self) -> A2aResult<Arc<InMemoryTaskStore>> {
204 let tenant = TenantContext::current();
205
206 {
208 let stores = self.stores.read().await;
209 if let Some(store) = stores.get(&tenant) {
210 return Ok(Arc::clone(store));
211 }
212 }
213
214 let mut stores = self.stores.write().await;
216 if let Some(store) = stores.get(&tenant) {
218 return Ok(Arc::clone(store));
219 }
220
221 if stores.len() >= self.config.max_tenants {
222 return Err(a2a_protocol_types::error::A2aError::internal(format!(
223 "tenant limit exceeded: max {} tenants",
224 self.config.max_tenants
225 )));
226 }
227
228 let store = Arc::new(InMemoryTaskStore::with_config(
229 self.overrides
230 .get(&tenant)
231 .unwrap_or(&self.config.per_tenant)
232 .clone(),
233 ));
234 stores.insert(tenant, Arc::clone(&store));
235 drop(stores);
236 Ok(store)
237 }
238
239 async fn get_existing_store(&self) -> Option<Arc<InMemoryTaskStore>> {
245 let tenant = TenantContext::current();
246 let stores = self.stores.read().await;
247 stores.get(&tenant).map(Arc::clone)
248 }
249
250 pub async fn tenant_count(&self) -> usize {
252 self.stores.read().await.len()
253 }
254
255 pub async fn run_eviction_all(&self) {
259 let stores = self.stores.read().await;
260 for store in stores.values() {
261 store.run_eviction().await;
262 }
263 }
264
265 pub async fn prune_empty_tenants(&self) {
278 let mut stores = self.stores.write().await;
279 let mut empty_tenants = Vec::new();
280 for (tenant, store) in stores.iter() {
281 if store.count().await.unwrap_or(0) == 0 {
282 empty_tenants.push(tenant.clone());
283 }
284 }
285 for tenant in empty_tenants {
286 stores.remove(&tenant);
287 }
288 }
289}
290
291#[allow(clippy::manual_async_fn)]
292impl TaskStore for TenantAwareInMemoryTaskStore {
293 fn save<'a>(
294 &'a self,
295 task: &'a Task,
296 ) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> {
297 Box::pin(async move {
298 let store = self.get_store().await?;
299 store.save(task).await
300 })
301 }
302
303 fn get<'a>(
304 &'a self,
305 id: &'a TaskId,
306 ) -> Pin<Box<dyn Future<Output = A2aResult<Option<Task>>> + Send + 'a>> {
307 Box::pin(async move {
308 match self.get_existing_store().await {
309 Some(store) => store.get(id).await,
310 None => Ok(None),
311 }
312 })
313 }
314
315 fn list<'a>(
316 &'a self,
317 params: &'a ListTasksParams,
318 ) -> Pin<Box<dyn Future<Output = A2aResult<TaskListResponse>> + Send + 'a>> {
319 Box::pin(async move {
320 match self.get_existing_store().await {
321 Some(store) => store.list(params).await,
322 None => Ok(TaskListResponse::new(Vec::new())),
323 }
324 })
325 }
326
327 fn insert_if_absent<'a>(
328 &'a self,
329 task: &'a Task,
330 ) -> Pin<Box<dyn Future<Output = A2aResult<bool>> + Send + 'a>> {
331 Box::pin(async move {
332 let store = self.get_store().await?;
333 store.insert_if_absent(task).await
334 })
335 }
336
337 fn delete<'a>(
338 &'a self,
339 id: &'a TaskId,
340 ) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> {
341 Box::pin(async move {
342 match self.get_existing_store().await {
343 Some(store) => store.delete(id).await,
344 None => Ok(()),
345 }
346 })
347 }
348
349 fn count<'a>(&'a self) -> Pin<Box<dyn Future<Output = A2aResult<u64>> + Send + 'a>> {
350 Box::pin(async move {
351 match self.get_existing_store().await {
352 Some(store) => store.count().await,
353 None => Ok(0),
354 }
355 })
356 }
357}
358
359#[cfg(test)]
360mod tests {
361 use super::*;
362 use a2a_protocol_types::task::{ContextId, TaskState, TaskStatus};
363
364 fn make_task(id: &str, state: TaskState) -> Task {
366 Task {
367 id: TaskId::new(id),
368 context_id: ContextId::new("ctx-default"),
369 status: TaskStatus::new(state),
370 history: None,
371 artifacts: None,
372 metadata: None,
373 }
374 }
375
376 #[tokio::test]
388 async fn per_tenant_page_size_cap_reaches_the_delegate() {
389 let store = TenantAwareInMemoryTaskStore::with_config(TenantStoreConfig {
390 per_tenant: TaskStoreConfig {
391 max_page_size: 2,
392 ..TaskStoreConfig::default()
393 },
394 max_tenants: 10,
395 });
396
397 TenantContext::scope("capped", async {
398 for i in 0..5 {
399 store
400 .save(&make_task(&format!("t{i}"), TaskState::Submitted))
401 .await
402 .expect("save");
403 }
404
405 let listed = store
406 .list(&ListTasksParams {
407 page_size: Some(100),
408 ..Default::default()
409 })
410 .await
411 .expect("list");
412
413 assert_eq!(
414 listed.tasks.len(),
415 2,
416 "the caller asked for 100; per_tenant.max_page_size is 2. \
417 A delegate built with the default config would return 5"
418 );
419 })
420 .await;
421 }
422
423 #[tokio::test]
430 async fn an_override_gives_that_tenant_its_own_store_config() {
431 async fn saved_then_listed(
432 store: &TenantAwareInMemoryTaskStore,
433 tenant: &'static str,
434 ) -> usize {
435 TenantContext::scope(tenant, async {
436 for i in 0..5 {
437 store
438 .save(&make_task(&format!("{tenant}-{i}"), TaskState::Submitted))
439 .await
440 .expect("save");
441 }
442 store
443 .list(&ListTasksParams {
444 page_size: Some(100),
445 ..Default::default()
446 })
447 .await
448 .expect("list")
449 .tasks
450 .len()
451 })
452 .await
453 }
454
455 let store = TenantAwareInMemoryTaskStore::with_config(TenantStoreConfig {
456 per_tenant: TaskStoreConfig {
457 max_page_size: 50,
458 ..TaskStoreConfig::default()
459 },
460 max_tenants: 10,
461 })
462 .with_tenant_override(
463 "small",
464 TaskStoreConfig {
465 max_page_size: 1,
466 ..TaskStoreConfig::default()
467 },
468 );
469
470 assert_eq!(
471 saved_then_listed(&store, "small").await,
472 1,
473 "the override caps this tenant's page size at 1"
474 );
475 assert_eq!(
476 saved_then_listed(&store, "ordinary").await,
477 5,
478 "a tenant with no override keeps per_tenant's cap of 50"
479 );
480 }
481
482 #[tokio::test]
485 async fn tenant_context_default_is_empty_string() {
486 let tenant = TenantContext::current();
488 assert_eq!(tenant, "", "default tenant should be empty string");
489 }
490
491 #[tokio::test]
492 async fn tenant_context_scope_sets_and_restores() {
493 let before = TenantContext::current();
494 assert_eq!(before, "");
495
496 let inside = TenantContext::scope("acme", async { TenantContext::current() }).await;
497 assert_eq!(inside, "acme", "scope should set the tenant");
498
499 let after = TenantContext::current();
500 assert_eq!(after, "", "tenant should revert after scope exits");
501 }
502
503 #[tokio::test]
504 async fn tenant_context_nested_scopes() {
505 TenantContext::scope("outer", async {
506 assert_eq!(TenantContext::current(), "outer");
507 TenantContext::scope("inner", async {
508 assert_eq!(TenantContext::current(), "inner");
509 })
510 .await;
511 assert_eq!(
512 TenantContext::current(),
513 "outer",
514 "should restore outer tenant after inner scope"
515 );
516 })
517 .await;
518 }
519
520 #[tokio::test]
523 async fn tenant_isolation_save_and_get() {
524 let store = TenantAwareInMemoryTaskStore::new();
525
526 TenantContext::scope("tenant-a", async {
528 store
529 .save(&make_task("t1", TaskState::Submitted))
530 .await
531 .unwrap();
532 })
533 .await;
534
535 let found = TenantContext::scope("tenant-a", async {
537 store.get(&TaskId::new("t1")).await.unwrap()
538 })
539 .await;
540 assert!(found.is_some(), "tenant-a should see its own task");
541
542 let not_found = TenantContext::scope("tenant-b", async {
544 store.get(&TaskId::new("t1")).await.unwrap()
545 })
546 .await;
547 assert!(
548 not_found.is_none(),
549 "tenant-b should not see tenant-a's task"
550 );
551 }
552
553 #[tokio::test]
554 async fn tenant_isolation_list() {
555 let store = TenantAwareInMemoryTaskStore::new();
556
557 TenantContext::scope("alpha", async {
558 store
559 .save(&make_task("a1", TaskState::Submitted))
560 .await
561 .unwrap();
562 store
563 .save(&make_task("a2", TaskState::Working))
564 .await
565 .unwrap();
566 })
567 .await;
568
569 TenantContext::scope("beta", async {
570 store
571 .save(&make_task("b1", TaskState::Submitted))
572 .await
573 .unwrap();
574 })
575 .await;
576
577 let alpha_list = TenantContext::scope("alpha", async {
578 let params = ListTasksParams::default();
579 store.list(¶ms).await.unwrap()
580 })
581 .await;
582 assert_eq!(
583 alpha_list.tasks.len(),
584 2,
585 "alpha should see only its 2 tasks"
586 );
587
588 let beta_list = TenantContext::scope("beta", async {
589 let params = ListTasksParams::default();
590 store.list(¶ms).await.unwrap()
591 })
592 .await;
593 assert_eq!(beta_list.tasks.len(), 1, "beta should see only its 1 task");
594 }
595
596 #[tokio::test]
597 async fn tenant_isolation_delete() {
598 let store = TenantAwareInMemoryTaskStore::new();
599
600 TenantContext::scope("tenant-a", async {
601 store
602 .save(&make_task("t1", TaskState::Submitted))
603 .await
604 .unwrap();
605 })
606 .await;
607
608 TenantContext::scope("tenant-b", async {
610 store.delete(&TaskId::new("t1")).await.unwrap();
611 })
612 .await;
613
614 let still_exists = TenantContext::scope("tenant-a", async {
615 store.get(&TaskId::new("t1")).await.unwrap()
616 })
617 .await;
618 assert!(
619 still_exists.is_some(),
620 "tenant-a's task should survive tenant-b's delete"
621 );
622 }
623
624 #[tokio::test]
625 async fn tenant_isolation_insert_if_absent() {
626 let store = TenantAwareInMemoryTaskStore::new();
627
628 let inserted_a = TenantContext::scope("tenant-a", async {
630 store
631 .insert_if_absent(&make_task("shared-id", TaskState::Submitted))
632 .await
633 .unwrap()
634 })
635 .await;
636 assert!(inserted_a, "tenant-a insert should succeed");
637
638 let inserted_b = TenantContext::scope("tenant-b", async {
639 store
640 .insert_if_absent(&make_task("shared-id", TaskState::Working))
641 .await
642 .unwrap()
643 })
644 .await;
645 assert!(
646 inserted_b,
647 "tenant-b insert of same ID should also succeed (different partition)"
648 );
649 }
650
651 #[tokio::test]
652 async fn tenant_isolation_count() {
653 let store = TenantAwareInMemoryTaskStore::new();
654
655 TenantContext::scope("x", async {
656 store
657 .save(&make_task("t1", TaskState::Submitted))
658 .await
659 .unwrap();
660 store
661 .save(&make_task("t2", TaskState::Submitted))
662 .await
663 .unwrap();
664 })
665 .await;
666
667 TenantContext::scope("y", async {
668 store
669 .save(&make_task("t3", TaskState::Submitted))
670 .await
671 .unwrap();
672 })
673 .await;
674
675 let count_x = TenantContext::scope("x", async { store.count().await.unwrap() }).await;
676 assert_eq!(count_x, 2, "tenant x should have 2 tasks");
677
678 let count_y = TenantContext::scope("y", async { store.count().await.unwrap() }).await;
679 assert_eq!(count_y, 1, "tenant y should have 1 task");
680 }
681
682 #[tokio::test]
694 async fn a_full_tenant_table_refuses_new_tenants_and_pruning_does_not_help() {
695 let store = TenantAwareInMemoryTaskStore::with_config(TenantStoreConfig {
696 per_tenant: TaskStoreConfig::default(),
697 max_tenants: 2,
698 });
699
700 for junk in ["junk-1", "junk-2"] {
701 TenantContext::scope(junk, async {
702 store
703 .save(&make_task("t", TaskState::Working))
704 .await
705 .expect("a fresh tenant partition is created on demand");
706 })
707 .await;
708 }
709 assert_eq!(store.tenant_count().await, 2, "the table is full");
710
711 let refused = TenantContext::scope("legitimate", async {
712 store.save(&make_task("t", TaskState::Working)).await
713 })
714 .await;
715 assert!(
716 refused.is_err(),
717 "a new tenant must be refused once the cap is reached"
718 );
719
720 store.prune_empty_tenants().await;
722 assert_eq!(
723 store.tenant_count().await,
724 2,
725 "pruning reclaims nothing while the junk partitions hold live tasks"
726 );
727
728 let still_refused = TenantContext::scope("legitimate", async {
729 store.save(&make_task("t", TaskState::Working)).await
730 })
731 .await;
732 assert!(
733 still_refused.is_err(),
734 "so the lockout outlives the pruning that is supposed to end it"
735 );
736
737 for junk in ["junk-1", "junk-2"] {
739 TenantContext::scope(junk, async {
740 store.delete(&TaskId::new("t")).await.expect("delete");
741 })
742 .await;
743 }
744 store.prune_empty_tenants().await;
745 assert_eq!(store.tenant_count().await, 0, "now the slots come back");
746
747 let admitted = TenantContext::scope("legitimate", async {
748 store.save(&make_task("t", TaskState::Working)).await
749 })
750 .await;
751 assert!(admitted.is_ok(), "and the legitimate tenant gets in");
752 }
753
754 #[tokio::test]
755 async fn tenant_count_reflects_active_tenants() {
756 let store = TenantAwareInMemoryTaskStore::new();
757 assert_eq!(store.tenant_count().await, 0);
758
759 TenantContext::scope("a", async {
760 store
761 .save(&make_task("t1", TaskState::Submitted))
762 .await
763 .unwrap();
764 })
765 .await;
766 assert_eq!(store.tenant_count().await, 1);
767
768 TenantContext::scope("b", async {
769 store
770 .save(&make_task("t2", TaskState::Submitted))
771 .await
772 .unwrap();
773 })
774 .await;
775 assert_eq!(store.tenant_count().await, 2);
776 }
777
778 #[tokio::test]
779 async fn max_tenants_limit_enforced() {
780 let config = TenantStoreConfig {
781 per_tenant: TaskStoreConfig::default(),
782 max_tenants: 2,
783 };
784 let store = TenantAwareInMemoryTaskStore::with_config(config);
785
786 TenantContext::scope("t1", async {
788 store
789 .save(&make_task("task-a", TaskState::Submitted))
790 .await
791 .unwrap();
792 })
793 .await;
794 TenantContext::scope("t2", async {
795 store
796 .save(&make_task("task-b", TaskState::Submitted))
797 .await
798 .unwrap();
799 })
800 .await;
801
802 let result = TenantContext::scope("t3", async {
804 store.save(&make_task("task-c", TaskState::Submitted)).await
805 })
806 .await;
807 assert!(
808 result.is_err(),
809 "exceeding max_tenants should return an error"
810 );
811 }
812
813 #[tokio::test]
814 async fn existing_tenant_does_not_count_against_limit() {
815 let config = TenantStoreConfig {
816 per_tenant: TaskStoreConfig::default(),
817 max_tenants: 1,
818 };
819 let store = TenantAwareInMemoryTaskStore::with_config(config);
820
821 TenantContext::scope("only", async {
822 store
823 .save(&make_task("t1", TaskState::Submitted))
824 .await
825 .unwrap();
826 store
828 .save(&make_task("t2", TaskState::Working))
829 .await
830 .unwrap();
831 })
832 .await;
833
834 let count = TenantContext::scope("only", async { store.count().await.unwrap() }).await;
835 assert_eq!(count, 2, "existing tenant can add more tasks");
836 }
837
838 #[tokio::test]
841 async fn no_tenant_context_uses_default_partition() {
842 let store = TenantAwareInMemoryTaskStore::new();
843
844 store
846 .save(&make_task("default-task", TaskState::Submitted))
847 .await
848 .unwrap();
849
850 let fetched = store.get(&TaskId::new("default-task")).await.unwrap();
851 assert!(
852 fetched.is_some(),
853 "task saved without tenant context should be retrievable without context"
854 );
855
856 let not_found = TenantContext::scope("other", async {
858 store.get(&TaskId::new("default-task")).await.unwrap()
859 })
860 .await;
861 assert!(
862 not_found.is_none(),
863 "default partition task should not leak to named tenants"
864 );
865 }
866
867 #[tokio::test]
870 async fn prune_empty_tenants_removes_empty_partitions() {
871 let store = TenantAwareInMemoryTaskStore::new();
872
873 TenantContext::scope("keep", async {
874 store
875 .save(&make_task("t1", TaskState::Submitted))
876 .await
877 .unwrap();
878 })
879 .await;
880 TenantContext::scope("remove", async {
881 store
882 .save(&make_task("t2", TaskState::Submitted))
883 .await
884 .unwrap();
885 })
886 .await;
887 assert_eq!(store.tenant_count().await, 2);
888
889 TenantContext::scope("remove", async {
891 store.delete(&TaskId::new("t2")).await.unwrap();
892 })
893 .await;
894
895 store.prune_empty_tenants().await;
896 assert_eq!(
897 store.tenant_count().await,
898 1,
899 "empty tenant partition should be pruned"
900 );
901 }
902
903 #[test]
907 fn default_creates_new_tenant_store() {
908 let store = TenantAwareInMemoryTaskStore::default();
909 let rt = tokio::runtime::Builder::new_current_thread()
910 .enable_all()
911 .build()
912 .unwrap();
913 let count = rt.block_on(store.tenant_count());
914 assert_eq!(count, 0, "default store should have no tenants");
915 }
916
917 #[tokio::test]
919 async fn run_eviction_all_runs_without_error() {
920 let store = TenantAwareInMemoryTaskStore::new();
921
922 TenantContext::scope("t1", async {
924 store
925 .save(&make_task("task-a", TaskState::Completed))
926 .await
927 .unwrap();
928 })
929 .await;
930 TenantContext::scope("t2", async {
931 store
932 .save(&make_task("task-b", TaskState::Working))
933 .await
934 .unwrap();
935 })
936 .await;
937
938 store.run_eviction_all().await;
940 }
941
942 #[tokio::test]
946 async fn get_store_double_check_path() {
947 let store = TenantAwareInMemoryTaskStore::new();
948
949 TenantContext::scope("racer", async {
951 store
952 .save(&make_task("t1", TaskState::Submitted))
953 .await
954 .unwrap();
955 store
957 .save(&make_task("t2", TaskState::Working))
958 .await
959 .unwrap();
960
961 let count = store.count().await.unwrap();
962 assert_eq!(count, 2, "both tasks should be in same tenant store");
963 })
964 .await;
965
966 assert_eq!(
967 store.tenant_count().await,
968 1,
969 "should have exactly 1 tenant"
970 );
971 }
972
973 #[test]
974 fn default_tenant_store_config() {
975 let cfg = TenantStoreConfig::default();
976 assert_eq!(cfg.max_tenants, 1000);
977 }
978
979 #[tokio::test]
991 async fn run_eviction_all_evicts_in_every_tenant() {
992 let store = TenantAwareInMemoryTaskStore::with_config(TenantStoreConfig {
993 per_tenant: TaskStoreConfig {
994 task_ttl: Some(std::time::Duration::from_millis(1)),
995 ..TaskStoreConfig::default()
996 },
997 ..TenantStoreConfig::default()
998 });
999
1000 for tenant in ["tenant-a", "tenant-b"] {
1001 TenantContext::scope(tenant, async {
1002 store
1003 .save(&make_task("t1", TaskState::Completed))
1004 .await
1005 .expect("save");
1006 })
1007 .await;
1008 }
1009
1010 tokio::time::sleep(std::time::Duration::from_millis(20)).await;
1012 store.run_eviction_all().await;
1013
1014 for tenant in ["tenant-a", "tenant-b"] {
1015 let still_there = TenantContext::scope(tenant, async {
1016 store.get(&TaskId::new("t1")).await.expect("get")
1017 })
1018 .await;
1019 assert!(
1020 still_there.is_none(),
1021 "a terminal task past its TTL must be evicted in {tenant}; \
1022 surviving means run_eviction_all did not reach this partition"
1023 );
1024 }
1025 }
1026
1027 #[test]
1028 fn every_tenant_store_config_setter_sets_its_field() {
1029 let d = TenantStoreConfig::default();
1030 let per_tenant =
1031 TaskStoreConfig::default().with_max_page_size(d.per_tenant.max_page_size + 1);
1032 let cfg = TenantStoreConfig::default()
1033 .with_per_tenant(per_tenant)
1034 .with_max_tenants(d.max_tenants + 1);
1035 assert_eq!(cfg.per_tenant.max_page_size, d.per_tenant.max_page_size + 1);
1036 assert_eq!(cfg.max_tenants, d.max_tenants + 1);
1037 }
1038}