photon_backend/
executor_services.rs

1//! Shared runtime services for handler delivery (constructed at Photon build time).
2
3use std::sync::Arc;
4
5use crate::checkpoint::CheckpointCoalescer;
6use crate::delivery::{DlqSink, WorkerPool};
7use crate::retention::{PartitionReclaim, RetentionDeps, RetentionHook, RetentionPolicy, RetentionReclaimer};
8use crate::storage::StoragePort;
9
10/// Shared runtime services for handler delivery.
11pub struct ExecutorServices {
12    /// Bounded handler concurrency pool.
13    pub worker_pool: Arc<WorkerPool>,
14    /// Coalesced checkpoint writer.
15    pub checkpoint_coalescer: Arc<CheckpointCoalescer>,
16    /// Dead-letter sink for failed deliveries.
17    pub dlq: Arc<DlqSink>,
18    /// Storage retention reclaim coordinator.
19    pub retention_reclaimer: Arc<RetentionReclaimer>,
20}
21
22impl ExecutorServices {
23    /// Construct delivery services bound to the storage port.
24    #[allow(clippy::needless_pass_by_value)] // Arc-by-value is the public ownership API
25    pub fn new(
26        port: Arc<dyn StoragePort>,
27        policy: RetentionPolicy,
28        hook: Option<Arc<dyn RetentionHook>>,
29    ) -> Self {
30        let coalescer = Arc::new(CheckpointCoalescer::new(Arc::clone(&port)));
31        let dlq = Arc::new(DlqSink::new());
32        let reclaimer = RetentionReclaimer::new(RetentionDeps {
33            port: Arc::clone(&port),
34            coalescer: Arc::clone(&coalescer),
35            dlq: Arc::clone(&dlq),
36            policy,
37            hook,
38        });
39        coalescer.attach_reclaimer(Arc::clone(&reclaimer) as Arc<dyn PartitionReclaim>);
40        Self {
41            worker_pool: Arc::new(WorkerPool::from_env()),
42            checkpoint_coalescer: coalescer,
43            dlq,
44            retention_reclaimer: reclaimer,
45        }
46    }
47}
photon_backend/executor_services.rs

photon_backend/
executor_services.rs
