Skip to content

Commit 85e2a43

Browse files
mschuwalowmschuwalow
committed
factor out reverse index
1 parent bfaf191 commit 85e2a43

12 files changed

Lines changed: 220 additions & 180 deletions

File tree

‎golem-debugging-service/src/debug_context.rs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@ use golem_worker_executor::services::agent_types::AgentTypesService;
5050
use golem_worker_executor::services::agent_webhooks::AgentWebhooksService;
5151
use golem_worker_executor::services::blob_store::BlobStoreService;
5252
use golem_worker_executor::services::card::CardService;
53+
use golem_worker_executor::services::card_interest::CardInterestIndex;
5354
use golem_worker_executor::services::component::ComponentService;
5455
use golem_worker_executor::services::environment_state::EnvironmentStateService;
5556
use golem_worker_executor::services::file_loader::FileLoader;
@@ -567,6 +568,7 @@ impl WorkerCtx for DebugContext {
567568
rpc: Arc<dyn Rpc>,
568569
worker_proxy: Arc<dyn WorkerProxy>,
569570
card_service: Arc<dyn CardService>,
571+
card_interest_index: Arc<CardInterestIndex>,
570572
component_service: Arc<dyn ComponentService>,
571573
_extra_deps: Self::ExtraDeps,
572574
config: Arc<GolemConfig>,
@@ -610,6 +612,7 @@ impl WorkerCtx for DebugContext {
610612
rpc,
611613
worker_proxy,
612614
card_service,
615+
card_interest_index,
613616
component_service,
614617
account_resource_limits,
615618
config,

‎golem-worker-executor-test-utils/src/lib.rs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,7 @@ use golem_worker_executor::services::blob_store::{
8888
BlobStoreError, BlobStoreService, DefaultBlobStoreService,
8989
};
9090
use golem_worker_executor::services::card::{CardService, NoopCardService};
91+
use golem_worker_executor::services::card_interest::CardInterestIndex;
9192
use golem_worker_executor::services::component::ComponentService;
9293
use golem_worker_executor::services::direct_invocation_auth::{
9394
DirectInvocationAuthService, NoOpDirectInvocationAuthService,
@@ -1482,6 +1483,7 @@ impl WorkerCtx for TestWorkerCtx {
14821483
rpc: Arc<dyn Rpc>,
14831484
worker_proxy: Arc<dyn WorkerProxy>,
14841485
card_service: Arc<dyn CardService>,
1486+
card_interest_index: Arc<CardInterestIndex>,
14851487
component_service: Arc<dyn ComponentService>,
14861488
extra_deps: Self::ExtraDeps,
14871489
config: Arc<GolemConfig>,
@@ -1535,6 +1537,7 @@ impl WorkerCtx for TestWorkerCtx {
15351537
rpc,
15361538
worker_proxy,
15371539
card_service,
1540+
card_interest_index,
15381541
component_service,
15391542
account_resource_limits,
15401543
config,

‎golem-worker-executor/src/durable_host/mod.rs‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,7 @@ use crate::services::agent_types::AgentTypesService;
5252
use crate::services::agent_webhooks::AgentWebhooksService;
5353
use crate::services::blob_store::BlobStoreService;
5454
use crate::services::card::{CardService, CardState};
55+
use crate::services::card_interest::CardInterestIndex;
5556
use crate::services::component::ComponentService;
5657
use crate::services::environment_state::EnvironmentStateService;
5758
use crate::services::file_loader::{FileLoader, FileUseToken};
@@ -355,6 +356,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
355356
rpc: Arc<dyn Rpc>,
356357
worker_proxy: Arc<dyn WorkerProxy>,
357358
card_service: Arc<dyn CardService>,
359+
card_interest_index: Arc<CardInterestIndex>,
358360
component_service: Arc<dyn ComponentService>,
359361
resource_limits: Arc<AtomicResourceEntry>,
360362
config: Arc<GolemConfig>,
@@ -521,6 +523,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
521523
rdbms_service,
522524
quota_service,
523525
card_service,
526+
card_interest_index,
524527
component_service,
525528
agent_types_service,
526529
environment_state_service,
@@ -732,7 +735,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
732735
candidate_wallet_card_ids.push(card_id);
733736
}
734737
self.state
735-
.card_service
738+
.card_interest_index
736739
.set_card_interest(self.owned_agent_id.clone(), &candidate_wallet_card_ids)
737740
.await;
738741

@@ -751,7 +754,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
751754
.copied()
752755
.collect::<Vec<_>>();
753756
self.state
754-
.card_service
757+
.card_interest_index
755758
.set_card_interest(self.owned_agent_id.clone(), &wallet_card_ids)
756759
.await;
757760

@@ -781,7 +784,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
781784
.copied()
782785
.collect::<Vec<_>>();
783786
self.state
784-
.card_service
787+
.card_interest_index
785788
.set_card_interest(self.owned_agent_id.clone(), &wallet_card_ids)
786789
.await;
787790

@@ -813,7 +816,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
813816
.copied()
814817
.collect::<Vec<_>>();
815818
self.state
816-
.card_service
819+
.card_interest_index
817820
.set_card_interest(self.owned_agent_id.clone(), &wallet_card_ids)
818821
.await;
819822

@@ -2634,7 +2637,7 @@ impl<Ctx: WorkerCtx> DurableWorkerCtx<Ctx> {
26342637
.copied()
26352638
.collect::<Vec<_>>();
26362639
self.state
2637-
.card_service
2640+
.card_interest_index
26382641
.set_card_interest(self.owned_agent_id.clone(), &wallet_card_ids)
26392642
.await;
26402643

@@ -4607,6 +4610,7 @@ struct PrivateDurableWorkerState {
46074610
rdbms_service: Arc<dyn RdbmsService>,
46084611
quota_service: Arc<dyn QuotaService>,
46094612
card_service: Arc<dyn CardService>,
4613+
card_interest_index: Arc<CardInterestIndex>,
46104614
component_service: Arc<dyn ComponentService>,
46114615
agent_types_service: Arc<dyn AgentTypesService>,
46124616
agent_webhooks_service: Arc<AgentWebhooksService>,
@@ -4790,6 +4794,7 @@ impl PrivateDurableWorkerState {
47904794
rdbms_service: Arc<dyn RdbmsService>,
47914795
quota_service: Arc<dyn QuotaService>,
47924796
card_service: Arc<dyn CardService>,
4797+
card_interest_index: Arc<CardInterestIndex>,
47934798
component_service: Arc<dyn ComponentService>,
47944799
agent_types_service: Arc<dyn AgentTypesService>,
47954800
environment_state_service: Arc<dyn EnvironmentStateService>,
@@ -4860,6 +4865,7 @@ impl PrivateDurableWorkerState {
48604865
rdbms_service,
48614866
quota_service,
48624867
card_service,
4868+
card_interest_index,
48634869
component_service,
48644870
agent_types_service,
48654871
environment_state_service,

‎golem-worker-executor/src/lib.rs‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ use crate::services::worker_enumeration::{
7777
};
7878
use crate::services::worker_proxy::{RemoteWorkerProxy, WorkerProxy};
7979
use crate::services::{
80-
All, HasActiveWorkers, HasAgentTypesService, HasComponentService, HasConfig,
80+
All, HasActiveWorkers, HasAgentTypesService, HasCardService, HasComponentService, HasConfig,
8181
HasEnvironmentStateService, HasOplogService, HasWorkerActivator, HasWorkerService, rdbms,
8282
};
8383
use crate::storage::indexed::IndexedStorage;
@@ -786,7 +786,6 @@ pub async fn create_worker_executor_impl<
786786
&golem_config.memory,
787787
&golem_config.filesystem_storage,
788788
&golem_config.agent_status_flush,
789-
card_service.clone(),
790789
shutdown_token.clone(),
791790
));
792791

@@ -1044,6 +1043,7 @@ pub async fn bootstrap_and_run_worker_executor<
10441043
if start_registry_invalidation_handler {
10451044
let registry_service = registry_service.clone();
10461045
let active_workers = worker_executor_impl.active_workers();
1046+
let card_service = worker_executor_impl.card_service();
10471047
let component_service = worker_executor_impl.component_service();
10481048
let environment_state_service = worker_executor_impl.environment_state_service();
10491049
let agent_types_service = worker_executor_impl.agent_types();
@@ -1052,6 +1052,7 @@ pub async fn bootstrap_and_run_worker_executor<
10521052
WorkerExecutorRegistryInvalidationHandler::run(
10531053
registry_service,
10541054
active_workers,
1055+
card_service,
10551056
component_service,
10561057
environment_state_service,
10571058
agent_types_service,

‎golem-worker-executor/src/services/active_workers/mod.rs‎

Lines changed: 9 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -33,12 +33,12 @@ use tokio_util::sync::CancellationToken;
3333

3434
use tracing::{Instrument, debug};
3535

36-
use crate::services::card::CardService;
36+
use crate::services::HasAll;
37+
use crate::services::card_interest::CardInterestIndex;
3738
use crate::services::golem_config::{
3839
AgentStatusFlushConfig, FilesystemStorageConfig, MemoryConfig,
3940
};
4041
use crate::services::resource_limits::AtomicResourceEntry;
41-
use crate::services::{HasAll, HasCardService};
4242
use crate::worker::Worker;
4343
use crate::worker::status_flusher::AgentStatusFlushQueue;
4444
use crate::workerctx::WorkerCtx;
@@ -71,7 +71,7 @@ impl RegisteredConcurrentAccount {
7171
/// Holds the metadata and wasmtime structures of currently active Golem workers
7272
pub struct ActiveWorkers<Ctx: WorkerCtx> {
7373
workers: Cache<AgentId, (), Arc<Worker<Ctx>>, WorkerExecutorError>,
74-
card_service: Arc<dyn CardService>,
74+
card_interest_index: Arc<CardInterestIndex>,
7575
worker_memory: Arc<Semaphore>,
7676
worker_filesystem_storage: Arc<FilesystemStorageSemaphore>,
7777
concurrent_agents: Arc<ConcurrentAgentsScheduler>,
@@ -120,7 +120,6 @@ impl<Ctx: WorkerCtx> ActiveWorkers<Ctx> {
120120
memory_config: &MemoryConfig,
121121
storage_config: &FilesystemStorageConfig,
122122
agent_status_flush_config: &AgentStatusFlushConfig,
123-
card_service: Arc<dyn CardService>,
124123
shutdown_token: CancellationToken,
125124
) -> Self {
126125
let worker_memory_size = memory_config.worker_memory();
@@ -131,7 +130,7 @@ impl<Ctx: WorkerCtx> ActiveWorkers<Ctx> {
131130
BackgroundEvictionMode::None,
132131
"active_workers",
133132
),
134-
card_service,
133+
card_interest_index: Arc::new(CardInterestIndex::new()),
135134
worker_memory: Arc::new(Semaphore::new(worker_memory_size)),
136135
worker_filesystem_storage: Arc::new(FilesystemStorageSemaphore::new(
137136
storage_config.worker_filesystem_storage(),
@@ -167,7 +166,7 @@ impl<Ctx: WorkerCtx> ActiveWorkers<Ctx> {
167166
principal: Principal,
168167
) -> Result<Arc<Worker<Ctx>>, WorkerExecutorError>
169168
where
170-
T: HasAll<Ctx> + HasCardService + Clone + Send + Sync + 'static,
169+
T: HasAll<Ctx> + Clone + Send + Sync + 'static,
171170
{
172171
let agent_id = owned_agent_id.agent_id();
173172

@@ -179,6 +178,7 @@ impl<Ctx: WorkerCtx> ActiveWorkers<Ctx> {
179178
Box::pin(async move {
180179
let worker = Worker::new(
181180
&deps,
181+
self.card_interest_index.clone(),
182182
owned_agent_id.clone(),
183183
worker_env,
184184
worker_agent_config,
@@ -203,15 +203,15 @@ impl<Ctx: WorkerCtx> ActiveWorkers<Ctx> {
203203

204204
pub async fn remove(&self, agent_id: &AgentId) {
205205
if let Some(worker) = self.workers.get(agent_id).await {
206-
self.card_service
206+
self.card_interest_index
207207
.set_card_interest(worker.owned_agent_id().clone(), &[])
208208
.await;
209209
}
210210
self.workers.remove(agent_id).await
211211
}
212212

213-
pub async fn record_revoked_cards(&self, card_ids: &[CardId]) {
214-
let affected_agent_cards = self.card_service.record_revoked_cards(card_ids).await;
213+
pub async fn notify_revoked_cards(&self, card_ids: &[CardId]) {
214+
let affected_agent_cards = self.card_interest_index.interested_agents(card_ids).await;
215215

216216
for (owned_agent_id, affected_card_ids) in affected_agent_cards {
217217
let Some(worker) = self.try_get(&owned_agent_id).await else {

0 commit comments

Comments
 (0)