saluki_core/runtime/state/resources/
worker.rs

1use async_trait::async_trait;
2use saluki_api::{DynamicRoute, EndpointType};
3use saluki_common::sync::shutdown::ShutdownHandle;
4use saluki_error::generic_error;
5use tracing::warn;
6
7use super::ResourceRegistry;
8use crate::{
9    diagnostic::DiagnosticsEmitter,
10    runtime::{state::DataspaceRegistry, InitializationError, Supervisable, SupervisorFuture},
11    support::SubsystemIdentifier,
12};
13
14/// A worker that exposes the resource registry over the control plane.
15///
16/// The registry itself needs no event loop -- it only does work when a resource is acquired or returned -- so this
17/// worker exists purely to publish it: it asserts the resource API routes as a [`DynamicRoute`] on the unprivileged
18/// endpoint and registers a diagnostic artifact, then idles until shutdown. Both registrations are retracted when the
19/// worker's process exits.
20pub struct ResourceRegistryWorker {
21    resource_registry: ResourceRegistry,
22}
23
24impl ResourceRegistryWorker {
25    pub(super) fn new(resource_registry: ResourceRegistry) -> Self {
26        Self { resource_registry }
27    }
28}
29
30#[async_trait]
31impl Supervisable for ResourceRegistryWorker {
32    fn name(&self) -> &str {
33        "resource-registry"
34    }
35
36    async fn initialize(&self, process_shutdown: ShutdownHandle) -> Result<SupervisorFuture, InitializationError> {
37        let resource_routes = DynamicRoute::http(EndpointType::Unprivileged, self.resource_registry.api_handler());
38
39        let resource_registry = self.resource_registry.clone();
40
41        Ok(Box::pin(async move {
42            let dataspace =
43                DataspaceRegistry::try_current().ok_or_else(|| generic_error!("Dataspace not available."))?;
44
45            // Register our API routes before we actually start running.
46            dataspace.assert(resource_routes, "resource-registry-api");
47
48            // Expose our diagnostic artifact via the diagnostics control surface.
49            let diagnostics = DiagnosticsEmitter::from_dataspace(
50                SubsystemIdentifier::from_segments(["resource-registry"]),
51                dataspace,
52            );
53            diagnostics.register_collector("resources.json", move || {
54                let snapshot = resource_registry.snapshot();
55                let snapshot_pretty = match serde_json::to_string_pretty(&snapshot) {
56                    Ok(json) => json,
57                    Err(e) => {
58                        warn!(error = %e, "Failed to serialize resource registry snapshot during diagnostics collection.");
59                        String::from(r#"{"error": "failed to serialize"}"#)
60                    },
61                };
62
63                snapshot_pretty
64            });
65
66            process_shutdown.await;
67
68            Ok(())
69        }))
70    }
71}
saluki_core/runtime/state/resources/worker.rs

saluki_core/runtime/state/resources/
worker.rs
