soma_infra/
image_pull.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use serde::{Deserialize, Serialize};
5use soma_fleet::{HostId, HostRecord, TopologyRevision};
6use soma_ops::{MutationSendState, OperationId, OperationName, Timestamp, VerificationStatus};
7use tokio_util::sync::CancellationToken;
8
9use crate::{
10    ContainerReader, ImageReader, InfraError, MutationProgressReporter, MutationResult,
11    MutationVerification,
12};
13
14const MAX_IMAGE_REFERENCE_CHARS: usize = 512;
15#[cfg(feature = "bollard-driver")]
16const MAX_PROGRESS_FRAMES: usize = 256;
17#[cfg(feature = "bollard-driver")]
18const MAX_PROGRESS_DELIVERY_ERRORS: usize = 16;
19
20/// Deadline-bound request to pull one Docker/OCI image reference.
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct ImagePullRequest {
23    operation_id: OperationId,
24    operation: OperationName,
25    image: String,
26    deadline: Timestamp,
27}
28
29impl ImagePullRequest {
30    /// Creates a validated image pull request.
31    pub fn new(
32        operation_id: OperationId,
33        operation: OperationName,
34        image: impl Into<String>,
35        deadline: Timestamp,
36    ) -> Result<Self, InfraError> {
37        let image = image.into();
38        validate_image_reference(&image)?;
39        Ok(Self {
40            operation_id,
41            operation,
42            image,
43            deadline,
44        })
45    }
46
47    /// Returns the operation execution identity.
48    #[must_use]
49    pub fn operation_id(&self) -> &OperationId {
50        &self.operation_id
51    }
52
53    /// Returns the canonical operation name.
54    #[must_use]
55    pub fn operation(&self) -> &OperationName {
56        &self.operation
57    }
58
59    /// Returns the requested image reference.
60    #[must_use]
61    pub fn image(&self) -> &str {
62        &self.image
63    }
64
65    /// Returns the absolute deadline.
66    #[must_use]
67    pub const fn deadline(&self) -> Timestamp {
68        self.deadline
69    }
70}
71
72/// Stable image identity observed through the Docker read API.
73#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
74pub struct ImageIdentity {
75    /// Docker image content ID.
76    pub id: String,
77    /// Repository tags bound to the image.
78    pub repo_tags: Vec<String>,
79    /// Repository digests bound to the image.
80    pub repo_digests: Vec<String>,
81}
82
83/// One retained neutral image-pull progress frame.
84#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
85pub struct ImagePullProgressFrame {
86    /// One-based stream sequence.
87    pub sequence: u64,
88    /// Docker status text.
89    pub status: Option<String>,
90    /// Layer or object identity.
91    pub id: Option<String>,
92    /// Current byte count when reported.
93    pub current: Option<u64>,
94    /// Total byte count when reported.
95    pub total: Option<u64>,
96    /// Human-readable progress text.
97    pub message: Option<String>,
98    /// Engine-reported error text.
99    pub error: Option<String>,
100}
101
102/// Receipt returned after the image pull stream completes.
103#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
104pub struct ImagePullReceipt {
105    /// Target host.
106    pub host: HostId,
107    /// Exact topology revision.
108    pub topology_revision: TopologyRevision,
109    /// Requested image reference.
110    pub image: String,
111    /// Backend send state.
112    pub send_state: MutationSendState,
113    /// Total stream frames observed.
114    pub total_events: u64,
115    /// Bounded retained progress frames.
116    pub progress: Vec<ImagePullProgressFrame>,
117    /// Whether additional frames were omitted.
118    pub progress_truncated: bool,
119    /// Bounded progress sink failures that did not rewrite execution truth.
120    pub progress_delivery_errors: Vec<String>,
121}
122
123#[cfg(feature = "bollard-driver")]
124impl ImagePullReceipt {
125    pub(crate) fn retain_frame(&mut self, frame: ImagePullProgressFrame) {
126        self.total_events = self.total_events.saturating_add(1);
127        if self.progress.len() < MAX_PROGRESS_FRAMES {
128            self.progress.push(frame);
129        } else {
130            self.progress_truncated = true;
131        }
132    }
133
134    pub(crate) fn retain_delivery_error(&mut self, error: String) {
135        if self.progress_delivery_errors.len() < MAX_PROGRESS_DELIVERY_ERRORS {
136            self.progress_delivery_errors.push(error);
137        }
138    }
139}
140
141/// Verified image pull outcome.
142#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
143pub struct ImagePullOutcome {
144    /// Target host.
145    pub host: HostId,
146    /// Exact topology revision.
147    pub topology_revision: TopologyRevision,
148    /// Requested image reference.
149    pub image: String,
150    /// Whether the verified image content identity changed.
151    pub changed: bool,
152    /// Backend send state.
153    pub send_state: MutationSendState,
154    /// Image identity observed before the pull.
155    pub before: Option<ImageIdentity>,
156    /// Image identity observed after the pull.
157    pub after: Option<ImageIdentity>,
158    /// Total stream events observed.
159    pub total_events: u64,
160    /// Bounded retained progress.
161    pub progress: Vec<ImagePullProgressFrame>,
162    /// Whether retained progress was truncated.
163    pub progress_truncated: bool,
164    /// Progress delivery failures that did not alter mutation truth.
165    pub progress_delivery_errors: Vec<String>,
166    /// Independent verification status.
167    pub verification_status: VerificationStatus,
168    /// Verification explanation.
169    pub verification: MutationVerification,
170}
171
172/// Driver for one image-pull stream.
173#[async_trait]
174pub trait ImagePullMutator: Send + Sync {
175    /// Pulls one image while emitting canonical progress and preserving send uncertainty.
176    async fn pull_image(
177        &self,
178        host: &HostRecord,
179        request: &ImagePullRequest,
180        progress: &dyn MutationProgressReporter,
181        cancellation: &CancellationToken,
182    ) -> MutationResult<ImagePullReceipt>;
183}
184
185/// Complete Docker client required by artifact mutations.
186pub trait DockerArtifactClient: ContainerReader + ImageReader + ImagePullMutator {}
187
188impl<T> DockerArtifactClient for T where T: ContainerReader + ImageReader + ImagePullMutator {}
189
190/// Factory for host- and revision-bound artifact mutation clients.
191#[async_trait]
192pub trait DockerArtifactClientProvider: Send + Sync {
193    /// Returns a client bound to the exact host revision.
194    async fn artifact_client(
195        &self,
196        host: &HostRecord,
197        cancellation: &CancellationToken,
198    ) -> Result<Arc<dyn DockerArtifactClient>, InfraError>;
199}
200
201pub(crate) fn validate_image_reference(image: &str) -> Result<(), InfraError> {
202    let chars = image.chars().count();
203    if chars == 0
204        || chars > MAX_IMAGE_REFERENCE_CHARS
205        || image.chars().any(char::is_control)
206        || image.chars().any(char::is_whitespace)
207        || image.starts_with('-')
208    {
209        return Err(InfraError::InvalidRequest {
210            domain: "image-pull",
211            message: "invalid image reference".into(),
212        });
213    }
214    Ok(())
215}
216
217/// Returns the canonical tag used by Docker when no tag or digest is supplied.
218#[must_use]
219pub fn canonical_image_reference(image: &str) -> String {
220    if image.contains('@') {
221        return image.to_owned();
222    }
223    let last_slash = image.rfind('/');
224    let last_colon = image.rfind(':');
225    if last_colon.is_some_and(|colon| last_slash.is_none_or(|slash| colon > slash)) {
226        image.to_owned()
227    } else {
228        format!("{image}:latest")
229    }
230}
231
232#[cfg(test)]
233#[path = "image_pull_tests.rs"]
234mod tests;
soma_infra/image_pull.rs

soma_infra/
image_pull.rs
