soma_provider_adapters/python/
materializer_update.rs

1//! Explicit dependency update into an immutable candidate environment.
2
3use std::{
4    ffi::OsString,
5    fs, io,
6    path::{Path, PathBuf},
7    sync::atomic::{AtomicU64, Ordering},
8};
9
10use serde::Serialize;
11use thiserror::Error;
12
13use super::{
14    PreparedPythonEnvironment, PythonEnvironmentMaterializer, PythonMaterializationError,
15    PythonMaterializationRequest, READY_FILE, READY_SCHEMA_VERSION, ReadyMarker, UvRunner,
16    environment_python, read_verified_sdk, render_project, sha256_hex,
17};
18use crate::python::environment::{Pep723Metadata, PythonEnvironmentPlan};
19
20const UPDATE_PLAN_VERSION: u32 = 3;
21static UPDATE_SEQUENCE: AtomicU64 = AtomicU64::new(0);
22
23#[derive(Debug, Clone, Copy)]
24pub struct PythonEnvironmentUpdateRequest<'a> {
25    pub materialization: PythonMaterializationRequest<'a>,
26    pub provider_source_sha256: &'a str,
27}
28
29#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
30#[serde(rename_all = "snake_case")]
31pub enum PythonEnvironmentUpdateOutcome {
32    Prepared,
33    Reused,
34}
35
36#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
37pub struct PythonEnvironmentUpdateReport {
38    pub outcome: PythonEnvironmentUpdateOutcome,
39    pub current: Option<PreparedPythonEnvironment>,
40    pub candidate: PreparedPythonEnvironment,
41}
42
43#[derive(Debug, Error)]
44pub enum PythonEnvironmentUpdateError {
45    #[error("provider source SHA-256 must contain exactly 64 hexadecimal characters")]
46    InvalidSourceDigest,
47    #[error("current Python environment must be repaired before update: {0}")]
48    CurrentInvalid(String),
49    #[error("resolved Python update candidate is invalid: {0}")]
50    CandidateInvalid(String),
51    #[error("Python update cache plan has no managed cache root")]
52    MissingCacheRoot,
53    #[error("Python update cache path is not a real directory: {}", path.display())]
54    UnsafeCachePath { path: PathBuf },
55    #[error("uv command failed during update {operation}: {message}")]
56    Uv {
57        operation: &'static str,
58        message: String,
59    },
60    #[error(transparent)]
61    Materialization(#[from] PythonMaterializationError),
62    #[error("Python update I/O failed: {0}")]
63    Io(#[from] io::Error),
64}
65
66impl<R: UvRunner> PythonEnvironmentMaterializer<R> {
67    pub fn update(
68        &self,
69        plan: &PythonEnvironmentPlan,
70        request: PythonEnvironmentUpdateRequest<'_>,
71    ) -> Result<PythonEnvironmentUpdateReport, PythonEnvironmentUpdateError> {
72        let source_sha256 = normalize_digest(request.provider_source_sha256)?;
73        let current = match self.open_verified(plan) {
74            Ok(environment) => environment,
75            Err(PythonMaterializationError::IncompleteCache(message))
76            | Err(PythonMaterializationError::InvalidMarker(message)) => {
77                return Err(PythonEnvironmentUpdateError::CurrentInvalid(message));
78            }
79            Err(error) => return Err(error.into()),
80        };
81        self.runner
82            .verify_identity(&self.uv_program, &plan.uv_version)
83            .map_err(|message| PythonEnvironmentUpdateError::Uv {
84                operation: "identity verification",
85                message,
86            })?;
87        let sdk_wheel_bytes =
88            read_verified_sdk(request.materialization.sdk_wheel, &plan.sdk_wheel_sha256)?;
89
90        let python_cache_root = plan
91            .directory
92            .parent()
93            .and_then(Path::parent)
94            .ok_or(PythonEnvironmentUpdateError::MissingCacheRoot)?;
95        ensure_real_directory(python_cache_root, true)?;
96        let candidate_parent = python_cache_root.join(format!("v{UPDATE_PLAN_VERSION}"));
97        ensure_real_directory(&candidate_parent, false)?;
98        let staging = update_staging_path(&candidate_parent, &plan.key)?;
99        fs::create_dir(&staging)?;
100        let sdk_wheel_name = request
101            .materialization
102            .sdk_wheel
103            .file_name()
104            .ok_or_else(|| {
105                PythonMaterializationError::IncompleteCache(
106                    "configured SDK wheel has no file name".to_owned(),
107                )
108            })?;
109        let staged_sdk_wheel = staging.join(sdk_wheel_name);
110        fs::write(&staged_sdk_wheel, sdk_wheel_bytes)?;
111        let staged_request = PythonEnvironmentUpdateRequest {
112            materialization: PythonMaterializationRequest {
113                metadata: request.materialization.metadata,
114                python_executable: request.materialization.python_executable,
115                sdk_wheel: &staged_sdk_wheel,
116                offline: request.materialization.offline,
117            },
118            provider_source_sha256: request.provider_source_sha256,
119        };
120
121        let result = self.resolve_and_prepare_update(
122            plan,
123            staged_request,
124            &source_sha256,
125            &candidate_parent,
126            &staging,
127        );
128        if result.is_err() {
129            let _ = fs::remove_dir_all(&staging);
130        }
131        let (outcome, candidate) = result?;
132        Ok(PythonEnvironmentUpdateReport {
133            outcome,
134            current,
135            candidate,
136        })
137    }
138
139    fn resolve_and_prepare_update(
140        &self,
141        plan: &PythonEnvironmentPlan,
142        request: PythonEnvironmentUpdateRequest<'_>,
143        source_sha256: &str,
144        candidate_parent: &Path,
145        staging: &Path,
146    ) -> Result<
147        (PythonEnvironmentUpdateOutcome, PreparedPythonEnvironment),
148        PythonEnvironmentUpdateError,
149    > {
150        fs::write(
151            staging.join("pyproject.toml"),
152            render_project(request.materialization.metadata),
153        )?;
154        let mut lock_args = vec![
155            OsString::from("lock"),
156            OsString::from("--upgrade"),
157            OsString::from("--project"),
158            OsString::from("."),
159            OsString::from("--python"),
160            request
161                .materialization
162                .python_executable
163                .as_os_str()
164                .to_owned(),
165        ];
166        if request.materialization.offline {
167            lock_args.push(OsString::from("--offline"));
168        }
169        self.update_uv("lock", staging, &lock_args)?;
170
171        let lockfile = staging.join("uv.lock");
172        let lock = fs::read(&lockfile)?;
173        let lock_sha256 = sha256_hex(&lock);
174        let candidate_key = resolved_candidate_key(
175            plan,
176            request.materialization.metadata,
177            source_sha256,
178            &lock_sha256,
179        );
180        let candidate_plan = PythonEnvironmentPlan {
181            key: candidate_key.clone(),
182            directory: candidate_parent.join(&candidate_key),
183            plan_version: UPDATE_PLAN_VERSION,
184            dependency_count: plan.dependency_count,
185            runtime: plan.runtime.clone(),
186            sdk_wheel_tag: plan.sdk_wheel_tag.clone(),
187            sdk_wheel_sha256: plan.sdk_wheel_sha256.clone(),
188            uv_version: plan.uv_version.clone(),
189        };
190
191        match self.open_verified(&candidate_plan) {
192            Ok(Some(candidate)) => {
193                validate_candidate_identity(&candidate, source_sha256, &plan.key)?;
194                fs::remove_dir_all(staging)?;
195                return Ok((PythonEnvironmentUpdateOutcome::Reused, candidate));
196            }
197            Ok(None) => {}
198            Err(error) => {
199                return Err(PythonEnvironmentUpdateError::CandidateInvalid(
200                    error.to_string(),
201                ));
202            }
203        }
204
205        let mut sync_args = vec![
206            OsString::from("sync"),
207            OsString::from("--project"),
208            OsString::from("."),
209            OsString::from("--locked"),
210            OsString::from("--no-install-project"),
211            OsString::from("--python"),
212            request
213                .materialization
214                .python_executable
215                .as_os_str()
216                .to_owned(),
217        ];
218        if request.materialization.offline {
219            sync_args.push(OsString::from("--offline"));
220        }
221        self.update_uv("sync", staging, &sync_args)?;
222        let python = environment_python(staging);
223        self.runner
224            .verify_python(&python, &candidate_plan.runtime)
225            .map_err(|message| PythonEnvironmentUpdateError::Uv {
226                operation: "Python runtime identity verification",
227                message,
228            })?;
229        self.update_uv(
230            "SDK install",
231            staging,
232            &[
233                OsString::from("pip"),
234                OsString::from("install"),
235                OsString::from("--python"),
236                python.as_os_str().to_owned(),
237                OsString::from("--offline"),
238                OsString::from("--no-deps"),
239                request.materialization.sdk_wheel.as_os_str().to_owned(),
240            ],
241        )?;
242        if !python.is_file() || !lockfile.is_file() {
243            return Err(PythonMaterializationError::IncompleteCache(
244                "update did not create both .venv Python and uv.lock".to_owned(),
245            )
246            .into());
247        }
248        let marker = ReadyMarker {
249            schema_version: READY_SCHEMA_VERSION,
250            environment_key: candidate_plan.key.clone(),
251            plan_version: candidate_plan.plan_version,
252            dependency_count: candidate_plan.dependency_count,
253            runtime: candidate_plan.runtime.clone(),
254            sdk_wheel_tag: candidate_plan.sdk_wheel_tag.clone(),
255            sdk_wheel_sha256: candidate_plan.sdk_wheel_sha256.clone(),
256            uv_version: candidate_plan.uv_version.clone(),
257            lock_sha256,
258            provider_source_sha256: Some(source_sha256.to_owned()),
259            input_plan_key: Some(plan.key.clone()),
260        };
261        fs::write(
262            staging.join(READY_FILE),
263            serde_json::to_vec_pretty(&marker).expect("update marker is serializable"),
264        )?;
265
266        match fs::rename(staging, &candidate_plan.directory) {
267            Ok(()) => {}
268            Err(_) if fs::symlink_metadata(&candidate_plan.directory).is_ok() => {
269                fs::remove_dir_all(staging)?;
270            }
271            Err(error) => return Err(error.into()),
272        }
273        let candidate = self.open_verified(&candidate_plan)?.ok_or(
274            PythonEnvironmentUpdateError::CandidateInvalid(candidate_key),
275        )?;
276        validate_candidate_identity(&candidate, source_sha256, &plan.key)?;
277        Ok((PythonEnvironmentUpdateOutcome::Prepared, candidate))
278    }
279
280    fn update_uv(
281        &self,
282        operation: &'static str,
283        current_dir: &Path,
284        args: &[OsString],
285    ) -> Result<(), PythonEnvironmentUpdateError> {
286        self.runner
287            .run(&self.uv_program, args, current_dir)
288            .map_err(|message| PythonEnvironmentUpdateError::Uv { operation, message })
289    }
290}
291
292fn ensure_real_directory(
293    path: &Path,
294    create_parents: bool,
295) -> Result<(), PythonEnvironmentUpdateError> {
296    match fs::symlink_metadata(path) {
297        Ok(metadata) => {
298            if metadata.file_type().is_symlink() || !metadata.is_dir() {
299                return Err(PythonEnvironmentUpdateError::UnsafeCachePath {
300                    path: path.to_path_buf(),
301                });
302            }
303        }
304        Err(error) if error.kind() == io::ErrorKind::NotFound => {
305            if create_parents {
306                fs::create_dir_all(path)?;
307            } else {
308                fs::create_dir(path)?;
309            }
310            let metadata = fs::symlink_metadata(path)?;
311            if metadata.file_type().is_symlink() || !metadata.is_dir() {
312                return Err(PythonEnvironmentUpdateError::UnsafeCachePath {
313                    path: path.to_path_buf(),
314                });
315            }
316        }
317        Err(error) => return Err(error.into()),
318    }
319    Ok(())
320}
321
322fn validate_candidate_identity(
323    candidate: &PreparedPythonEnvironment,
324    source_sha256: &str,
325    input_plan_key: &str,
326) -> Result<(), PythonEnvironmentUpdateError> {
327    if candidate.provider_source_sha256.as_deref() != Some(source_sha256)
328        || candidate.input_plan_key.as_deref() != Some(input_plan_key)
329    {
330        return Err(PythonEnvironmentUpdateError::CandidateInvalid(
331            "readiness identity does not match update request".to_owned(),
332        ));
333    }
334    Ok(())
335}
336
337fn normalize_digest(value: &str) -> Result<String, PythonEnvironmentUpdateError> {
338    let value = value.trim().to_ascii_lowercase();
339    if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
340        return Err(PythonEnvironmentUpdateError::InvalidSourceDigest);
341    }
342    Ok(value)
343}
344
345#[derive(Serialize)]
346struct ResolvedCandidateFingerprint<'a> {
347    policy_version: u32,
348    provider_source_sha256: &'a str,
349    input_plan_key: &'a str,
350    metadata: Option<&'a Pep723Metadata>,
351    runtime: &'a crate::python::environment::PythonRuntimeFingerprint,
352    sdk_wheel_tag: &'a crate::python::environment::PythonWheelTag,
353    sdk_wheel_sha256: &'a str,
354    uv_version: &'a str,
355    lock_sha256: &'a str,
356}
357
358fn resolved_candidate_key(
359    plan: &PythonEnvironmentPlan,
360    metadata: Option<&Pep723Metadata>,
361    source_sha256: &str,
362    lock_sha256: &str,
363) -> String {
364    let fingerprint = ResolvedCandidateFingerprint {
365        policy_version: UPDATE_PLAN_VERSION,
366        provider_source_sha256: source_sha256,
367        input_plan_key: &plan.key,
368        metadata,
369        runtime: &plan.runtime,
370        sdk_wheel_tag: &plan.sdk_wheel_tag,
371        sdk_wheel_sha256: &plan.sdk_wheel_sha256,
372        uv_version: &plan.uv_version,
373        lock_sha256,
374    };
375    sha256_hex(&serde_json::to_vec(&fingerprint).expect("candidate fingerprint is serializable"))
376}
377
378fn update_staging_path(parent: &Path, input_plan_key: &str) -> io::Result<PathBuf> {
379    loop {
380        let sequence = UPDATE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
381        let candidate = parent.join(format!(
382            ".{input_plan_key}.update-{}-{sequence}",
383            std::process::id()
384        ));
385        match fs::symlink_metadata(&candidate) {
386            Ok(_) => continue,
387            Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(candidate),
388            Err(error) => return Err(error),
389        }
390    }
391}
soma_provider_adapters/python/materializer_update.rs

soma_provider_adapters/python/
materializer_update.rs
