soma_infra/
bollard_provider.rs

1use std::collections::BTreeMap;
2use std::path::{Component, Path, PathBuf};
3use std::sync::Arc;
4
5use async_trait::async_trait;
6use soma_fleet::{ConnectionPool, HostEndpoint, HostId, HostRecord, OpenSshConnector};
7use tokio_util::sync::CancellationToken;
8
9use crate::{
10    BollardReadClient, ContainerExecClientProvider, ContainerExecMutator, ContainerRecreateClient,
11    ContainerRecreateClientProvider, DockerArtifactClient, DockerArtifactClientProvider,
12    DockerCleanupClient, DockerCleanupClientProvider, DockerClientProvider, DockerMutationClient,
13    DockerMutationClientProvider, DockerReadClient, InfraError, InfraResult,
14};
15
16const DEFAULT_REMOTE_SOCKET: &str = "/var/run/docker.sock";
17
18/// Revision-aware Bollard provider for local and strict-SSH hosts.
19pub struct BollardClientProvider {
20    pool: Arc<ConnectionPool<OpenSshConnector>>,
21    remote_sockets: BTreeMap<HostId, PathBuf>,
22}
23
24impl BollardClientProvider {
25    /// Creates a provider backed by the shared strict-OpenSSH pool.
26    #[must_use]
27    pub fn new(pool: Arc<ConnectionPool<OpenSshConnector>>) -> Self {
28        Self {
29            pool,
30            remote_sockets: BTreeMap::new(),
31        }
32    }
33
34    /// Configures an explicit remote Docker socket for one SSH host.
35    pub fn with_remote_socket(
36        mut self,
37        host: HostId,
38        path: impl Into<PathBuf>,
39    ) -> InfraResult<Self> {
40        self.remote_sockets
41            .insert(host, validate_socket(path.into())?);
42        Ok(self)
43    }
44
45    fn plan<'a>(&'a self, host: &'a HostRecord) -> InfraResult<SocketPlan<'a>> {
46        match host.endpoint() {
47            HostEndpoint::Local => Ok(SocketPlan::Local),
48            HostEndpoint::Ssh(_) => Ok(SocketPlan::Remote(
49                self.remote_sockets
50                    .get(host.id())
51                    .map(PathBuf::as_path)
52                    .unwrap_or_else(|| Path::new(DEFAULT_REMOTE_SOCKET)),
53            )),
54            HostEndpoint::Http(_) => Err(InfraError::UnsupportedTarget {
55                domain: "docker",
56                host: host.id().clone(),
57            }),
58        }
59    }
60}
61
62#[async_trait]
63impl DockerClientProvider for BollardClientProvider {
64    async fn client(
65        &self,
66        host: &HostRecord,
67        cancellation: &CancellationToken,
68    ) -> InfraResult<Arc<dyn DockerReadClient>> {
69        match self.plan(host)? {
70            SocketPlan::Local => Ok(Arc::new(BollardReadClient::connect_local(host)?)),
71            SocketPlan::Remote(socket) => {
72                let connection = self.pool.get_or_connect(host, cancellation).await?;
73                Ok(Arc::new(
74                    BollardReadClient::connect_remote(connection, host, socket, cancellation)
75                        .await?,
76                ))
77            }
78        }
79    }
80}
81
82#[async_trait]
83impl DockerMutationClientProvider for BollardClientProvider {
84    async fn mutation_client(
85        &self,
86        host: &HostRecord,
87        cancellation: &CancellationToken,
88    ) -> InfraResult<Arc<dyn DockerMutationClient>> {
89        match self.plan(host)? {
90            SocketPlan::Local => Ok(Arc::new(BollardReadClient::connect_local(host)?)),
91            SocketPlan::Remote(socket) => {
92                let connection = self.pool.get_or_connect(host, cancellation).await?;
93                Ok(Arc::new(
94                    BollardReadClient::connect_remote(connection, host, socket, cancellation)
95                        .await?,
96                ))
97            }
98        }
99    }
100}
101
102#[async_trait]
103impl DockerCleanupClientProvider for BollardClientProvider {
104    async fn cleanup_client(
105        &self,
106        host: &HostRecord,
107        cancellation: &CancellationToken,
108    ) -> InfraResult<Arc<dyn DockerCleanupClient>> {
109        match self.plan(host)? {
110            SocketPlan::Local => Ok(Arc::new(BollardReadClient::connect_local(host)?)),
111            SocketPlan::Remote(socket) => {
112                let connection = self.pool.get_or_connect(host, cancellation).await?;
113                Ok(Arc::new(
114                    BollardReadClient::connect_remote(connection, host, socket, cancellation)
115                        .await?,
116                ))
117            }
118        }
119    }
120}
121
122#[async_trait]
123impl ContainerExecClientProvider for BollardClientProvider {
124    async fn exec_client(
125        &self,
126        host: &HostRecord,
127        cancellation: &CancellationToken,
128    ) -> InfraResult<Arc<dyn ContainerExecMutator>> {
129        match self.plan(host)? {
130            SocketPlan::Local => Ok(Arc::new(BollardReadClient::connect_local(host)?)),
131            SocketPlan::Remote(socket) => {
132                let connection = self.pool.get_or_connect(host, cancellation).await?;
133                Ok(Arc::new(
134                    BollardReadClient::connect_remote(connection, host, socket, cancellation)
135                        .await?,
136                ))
137            }
138        }
139    }
140}
141
142#[async_trait]
143impl ContainerRecreateClientProvider for BollardClientProvider {
144    async fn recreate_client(
145        &self,
146        host: &HostRecord,
147        cancellation: &CancellationToken,
148    ) -> InfraResult<Arc<dyn ContainerRecreateClient>> {
149        match self.plan(host)? {
150            SocketPlan::Local => Ok(Arc::new(BollardReadClient::connect_local(host)?)),
151            SocketPlan::Remote(socket) => {
152                let connection = self.pool.get_or_connect(host, cancellation).await?;
153                Ok(Arc::new(
154                    BollardReadClient::connect_remote(connection, host, socket, cancellation)
155                        .await?,
156                ))
157            }
158        }
159    }
160}
161
162#[async_trait]
163impl DockerArtifactClientProvider for BollardClientProvider {
164    async fn artifact_client(
165        &self,
166        host: &HostRecord,
167        cancellation: &CancellationToken,
168    ) -> InfraResult<Arc<dyn DockerArtifactClient>> {
169        match self.plan(host)? {
170            SocketPlan::Local => Ok(Arc::new(BollardReadClient::connect_local(host)?)),
171            SocketPlan::Remote(socket) => {
172                let connection = self.pool.get_or_connect(host, cancellation).await?;
173                Ok(Arc::new(
174                    BollardReadClient::connect_remote(connection, host, socket, cancellation)
175                        .await?,
176                ))
177            }
178        }
179    }
180}
181
182enum SocketPlan<'a> {
183    Local,
184    Remote(&'a Path),
185}
186
187fn validate_socket(path: PathBuf) -> InfraResult<PathBuf> {
188    if !path.is_absolute()
189        || path
190            .components()
191            .any(|part| matches!(part, Component::ParentDir | Component::CurDir))
192    {
193        Err(InfraError::InvalidRequest {
194            domain: "docker",
195            message: format!(
196                "Docker socket path must be absolute and normalized: {}",
197                path.display()
198            ),
199        })
200    } else {
201        Ok(path)
202    }
203}
204
205#[cfg(test)]
206#[path = "bollard_provider_tests.rs"]
207mod tests;
soma_infra/bollard_provider.rs

soma_infra/
bollard_provider.rs
