submilli_engine/stdlib/git/
mod.rs

1//! Capability-controlled Git operations. No subprocesses or guest-visible credentials.
2use crate::runtime::effects;
3use crate::runtime::host::permission_denied;
4pub(crate) mod class;
5pub mod declaration;
6mod diff;
7mod history;
8mod index_limits;
9mod install;
10mod location;
11mod lock;
12pub(crate) mod log_cache;
13mod metadata_scan;
14mod meter;
15mod object;
16mod operations;
17mod pack_index_check;
18mod pack_limits;
19mod stage;
20mod storage;
21mod transport;
22mod work;
23mod worker;
24
25use crate::runtime::Vfs;
26use crate::runtime::decision::{CallOrigin, CallSite, EntryPath};
27use crate::runtime::fuel;
28use crate::runtime::{HttpClient, SecretProvider, SecurityCheck, StoreData};
29use crate::stdlib::shared::running_package;
30use serde::{Serialize, de::DeserializeOwned};
31use serde_json::Value;
32use std::sync::{
33    Arc, Mutex, OnceLock,
34    atomic::{AtomicBool, AtomicU64, Ordering},
35};
36use wasmtime::{Caller, Result, bail};
37
38static WORKERS: OnceLock<Arc<tokio::sync::Semaphore>> = OnceLock::new();
39
40pub const MODULE_NAME: &str = "submilli:git";
41pub use declaration::package_declaration;
42pub use install::install;
43pub(crate) use transport::canonical_url;
44
45#[derive(Clone, Debug)]
46pub struct GitConfig {
47    pub name: String,
48    pub email: String,
49    pub username: Option<String>,
50}
51
52#[derive(Clone)]
53struct Job {
54    op: String,
55    path: String,
56    caller: String,
57    config: GitConfig,
58    security: Arc<dyn SecurityCheck>,
59    secrets: Arc<dyn SecretProvider>,
60    http: Arc<dyn HttpClient>,
61    runtime: tokio::runtime::Handle,
62    /// When the operation times out, waits included.
63    deadline: tokio::time::Instant,
64    cancelled: Arc<AtomicBool>,
65    max_bytes: u64,
66    transferred: Arc<AtomicU64>,
67    /// The worker's own file and object work, charged when it returns.
68    meter: Arc<meter::Meter>,
69    algorithm_fuel: Arc<work::AlgorithmWork>,
70    denial: Arc<Mutex<Option<(String, String)>>>,
71    history_cache: Option<Arc<log_cache::Cache>>,
72    /// Where in the program, and in a package, the operation started; captured only
73    /// while recording.
74    origin: CallOrigin,
75}
76impl Job {
77    fn remote_capability(&self) -> &'static str {
78        if self.op == "clone" {
79            "git.clone"
80        } else {
81            "git.fetch"
82        }
83    }
84
85    fn check(&self, capability: &str, context: Value) -> Result<()> {
86        self.check_cancelled()?;
87        // Git's decisions are made on a worker thread, after the program's stack is gone;
88        // they reuse the origin the host call captured on entry.
89        let site = CallSite::new(
90            self.security
91                .recorder()
92                .map(|recorder| recorder.begin_call(&self.caller, capability, self.origin.clone())),
93            EntryPath::Git,
94        );
95        crate::stdlib::shared::check_and_audit(
96            self.security.as_ref(),
97            &self.caller,
98            capability,
99            &context,
100            "/",
101            site,
102        )
103        .map_err(|reason| {
104            // Poison may leave the denied capability and reason partly updated.
105            // AGENTS.md permits poisoned-lock panics rather than recovering this
106            // attribution; it does not permit the panic that caused poisoning.
107            let mut denial = self.denial.lock().expect("git worker denial lock poisoned");
108            *denial = Some((capability.to_owned(), reason.clone()));
109            permission_denied(&self.caller, capability, reason)
110        })
111    }
112
113    /// One of the workers Git shares across the process, waited for until
114    /// the operation's deadline or its cancellation.
115    fn worker_permit(&self) -> Result<tokio::sync::OwnedSemaphorePermit> {
116        let workers = WORKERS
117            .get_or_init(|| Arc::new(tokio::sync::Semaphore::new(4)))
118            .clone();
119        // Polled from the worker's own thread, which is not the runtime's.
120        loop {
121            match workers.clone().try_acquire_owned() {
122                Ok(permit) => return Ok(permit),
123                Err(tokio::sync::TryAcquireError::Closed) => {
124                    return Err(crate::runtime::host::fatal_host_error(
125                        "git: workers closed",
126                    ));
127                }
128                Err(tokio::sync::TryAcquireError::NoPermits) => {}
129            }
130            self.check_cancelled()?;
131            if tokio::time::Instant::now() >= self.deadline {
132                bail!("git: operation timed out");
133            }
134            std::thread::sleep(std::time::Duration::from_millis(10));
135        }
136    }
137
138    fn check_cancelled(&self) -> Result<()> {
139        if self.cancelled.load(Ordering::Relaxed) {
140            bail!("git: operation cancelled");
141        }
142        Ok(())
143    }
144}
145struct CancelOnDrop(Arc<AtomicBool>);
146impl Drop for CancelOnDrop {
147    fn drop(&mut self) {
148        self.0.store(true, Ordering::Relaxed);
149    }
150}
151
152fn open(vfs: &Vfs, job: &Job, (): ()) -> Result<()> {
153    let (_permit, _snapshot) = worker::open_repository(vfs, job)?;
154    Ok(())
155}
156
157fn status(vfs: &Vfs, job: &Job, (): ()) -> Result<Value> {
158    let (_permit, snapshot) = worker::open_repository(vfs, job)?;
159    operations::status(&snapshot)
160}
161
162fn branches(vfs: &Vfs, job: &Job, (): ()) -> Result<Value> {
163    let (_permit, snapshot) = worker::open_repository(vfs, job)?;
164    operations::branches(&snapshot)
165}
166
167fn remotes(vfs: &Vfs, job: &Job, (): ()) -> Result<Value> {
168    let (_permit, snapshot) = worker::open_repository(vfs, job)?;
169    operations::remotes(&snapshot)
170}
171
172fn log(vfs: &Vfs, job: &Job, arguments: Value) -> Result<Value> {
173    worker::validate_branch_options(&arguments)?;
174    let (_permit, snapshot) = worker::open_repository(vfs, job)?;
175    operations::log(&snapshot, &arguments)
176}
177
178fn diff(vfs: &Vfs, job: &Job, arguments: Value) -> Result<Value> {
179    worker::validate_branch_options(&arguments)?;
180    let (_permit, snapshot) = worker::open_repository(vfs, job)?;
181    diff::read(&snapshot, &arguments)
182}
183
184fn show(vfs: &Vfs, job: &Job, arguments: ShowArguments) -> Result<effects::OutcomeBytes> {
185    let (_permit, snapshot) = worker::open_repository(vfs, job)?;
186    operations::show(&snapshot, &arguments.revision, &arguments.path).map(effects::OutcomeBytes)
187}
188
189fn init(vfs: &Vfs, job: &Job, arguments: Value) -> Result<()> {
190    let branch = worker::requested_branch(&arguments, "main")?;
191    job.check("git.init", serde_json::json!({"path":job.path}))?;
192    let (_permit, snapshot) = worker::create_repository(vfs, job, branch)?;
193    job.check_cancelled()?;
194    job.algorithm_fuel.before_effect();
195    snapshot.publish()?;
196    Ok(())
197}
198
199fn clone_repository(vfs: &Vfs, job: &Job, arguments: CloneArguments) -> Result<()> {
200    let branch = worker::requested_branch(&arguments.options, "")?;
201    let url = transport::canonical_url(&arguments.url)?;
202    job.check(
203        "git.clone",
204        serde_json::json!({"path":job.path,"remote":url,"remoteName":"origin","branch":branch}),
205    )?;
206    let (_permit, mut snapshot) = worker::create_clone_repository(vfs, job, branch)?;
207    worker::clone_repository(&mut snapshot, job, &arguments.url, branch)?;
208    job.check_cancelled()?;
209    job.algorithm_fuel.before_effect();
210    snapshot.publish()?;
211    Ok(())
212}
213
214fn add(vfs: &Vfs, job: &Job, arguments: Vec<String>) -> Result<()> {
215    let (_permit, snapshot) = worker::open_repository_for_write(vfs, job)?;
216    operations::add(&snapshot, &arguments)?;
217    job.check_cancelled()?;
218    job.algorithm_fuel.before_effect();
219    snapshot.publish()?;
220    Ok(())
221}
222
223fn commit(vfs: &Vfs, job: &Job, arguments: String) -> Result<String> {
224    let (_permit, snapshot) = worker::open_repository_for_write(vfs, job)?;
225    job.check(
226        "git.commit",
227        serde_json::json!({"path":job.path,"branch":operations::current_branch(&snapshot)?}),
228    )?;
229    let id = operations::commit(&snapshot, &arguments, &job.config)?;
230    job.check_cancelled()?;
231    job.algorithm_fuel.before_effect();
232    snapshot.publish()?;
233    Ok(id)
234}
235
236fn create_branch(vfs: &Vfs, job: &Job, arguments: BranchArguments) -> Result<()> {
237    let (_permit, snapshot) = worker::open_repository_for_write(vfs, job)?;
238    operations::create_branch(&snapshot, &arguments.name, &arguments.start)?;
239    job.check_cancelled()?;
240    job.algorithm_fuel.before_effect();
241    snapshot.publish()?;
242    Ok(())
243}
244
245fn switch_branch(vfs: &Vfs, job: &Job, arguments: String) -> Result<()> {
246    let (_permit, snapshot) = worker::open_repository_for_write(vfs, job)?;
247    operations::checkout(&snapshot, &arguments)?;
248    job.check_cancelled()?;
249    job.algorithm_fuel.before_effect();
250    snapshot.publish()?;
251    Ok(())
252}
253
254fn add_remote(vfs: &Vfs, job: &Job, arguments: RemoteArguments) -> Result<()> {
255    let (_permit, mut snapshot) = worker::open_repository_for_write(vfs, job)?;
256    operations::add_remote(&mut snapshot, &arguments.name, &arguments.url)?;
257    job.check_cancelled()?;
258    job.algorithm_fuel.before_effect();
259    snapshot.publish()?;
260    Ok(())
261}
262
263fn set_remote_url(vfs: &Vfs, job: &Job, arguments: RemoteArguments) -> Result<()> {
264    let (_permit, mut snapshot) = worker::open_repository_for_write(vfs, job)?;
265    operations::set_remote_url(&mut snapshot, &arguments.name, &arguments.url)?;
266    job.check_cancelled()?;
267    job.algorithm_fuel.before_effect();
268    snapshot.publish()?;
269    Ok(())
270}
271
272fn fetch(vfs: &Vfs, job: &Job, arguments: FetchArguments) -> Result<Value> {
273    let (_permit, snapshot) = worker::open_repository_for_write(vfs, job)?;
274    let branches = transport::fetch(&snapshot, job, &arguments.remote, &arguments.branch)?;
275    job.check_cancelled()?;
276    job.algorithm_fuel.before_effect();
277    snapshot.publish()?;
278    Ok(serde_json::json!({"branches":branches.branches}))
279}
280
281fn pull(vfs: &Vfs, job: &Job, arguments: FetchArguments) -> Result<Value> {
282    let (_permit, snapshot) = worker::open_repository_for_write(vfs, job)?;
283    let result = transport::pull(&snapshot, job, &arguments.remote, &arguments.branch)?;
284    job.check_cancelled()?;
285    job.algorithm_fuel.before_effect();
286    snapshot.publish()?;
287    Ok(result)
288}
289
290#[derive(Serialize)]
291struct ShowArguments {
292    revision: String,
293    path: String,
294}
295#[derive(Serialize)]
296struct CloneArguments {
297    url: String,
298    options: Value,
299}
300#[derive(Serialize)]
301struct BranchArguments {
302    name: String,
303    start: String,
304}
305#[derive(Serialize)]
306struct RemoteArguments {
307    name: String,
308    url: String,
309}
310#[derive(Serialize)]
311struct FetchArguments {
312    remote: String,
313    branch: String,
314}
315
316#[derive(Serialize)]
317struct GitRequest<A> {
318    operation: &'static str,
319    path: String,
320    arguments: A,
321}
322
323/// A single call's deadline and native-memory reservation. The blocking worker
324/// retains the reservation if the awaiting guest is cancelled.
325struct GitCall {
326    operation: &'static str,
327    config: GitConfig,
328    deadline: tokio::time::Instant,
329    budget: Arc<WorkingBudget>,
330    max_bytes: u64,
331}
332impl GitCall {
333    fn new(caller: &Caller<'_, StoreData>, operation: &'static str) -> Result<Self> {
334        let deadline = tokio::time::Instant::now()
335            .checked_add(std::time::Duration::from_secs(60))
336            .ok_or_else(|| {
337                crate::runtime::host::fatal_host_error("git deadline is out of range")
338            })?;
339        let config = caller.data().git.clone().ok_or_else(|| {
340            wasmtime::Error::msg("submilli:git is disabled; configure the blueprint git block")
341        })?;
342        let budget = Arc::new(WorkingBudget::reserve(
343            &caller.data().tenant_limits,
344            operation,
345        )?);
346        let max_bytes = budget.max_bytes(operation);
347        Ok(Self {
348            operation,
349            config,
350            deadline,
351            budget,
352            max_bytes,
353        })
354    }
355
356    async fn execute<A, T, F>(
357        &self,
358        caller: &mut Caller<'_, StoreData>,
359        path: String,
360        arguments: A,
361        operation: F,
362    ) -> Result<T>
363    where
364        A: Serialize + Send + 'static,
365        T: Serialize + DeserializeOwned + Send + 'static,
366        F: FnOnce(&Vfs, &Job, A) -> Result<T> + Send + 'static,
367    {
368        let op = self.operation;
369        let config = self.config.clone();
370        let deadline = self.deadline;
371        let max_bytes = self.max_bytes;
372        let worker_budget = Arc::clone(&self.budget);
373        let request = GitRequest {
374            operation: op,
375            path,
376            arguments,
377        };
378        effects::execute(caller, "git", request, move |caller, _, request| {
379            Box::pin(async move {
380                let path = request.path;
381                let args = request.arguments;
382                let budget = worker_budget;
383                let principal = running_package(&*caller).map_err(|_| {
384                    crate::runtime::host::fatal_host_error("git: cannot identify caller")
385                })?;
386                let cancelled = Arc::new(AtomicBool::new(false));
387                let cancel_guard = CancelOnDrop(cancelled.clone());
388                // The worker stops itself once its work passes what the run has left.
389                let meter = Arc::new(meter::Meter::new(
390                    fuel::remaining_fuel(&mut *caller)?,
391                    Some(cancelled.clone()),
392                ));
393                let job = Job {
394                    op: op.to_owned(),
395                    path: path.clone(),
396                    caller: principal,
397                    config,
398                    security: caller.data().security_check.clone(),
399                    secrets: caller.data().secret_provider.clone(),
400                    http: caller.data().http_client.clone(),
401                    runtime: tokio::runtime::Handle::try_current().map_err(|_| {
402                        crate::runtime::host::fatal_host_error(
403                            "git worker requires a Tokio runtime",
404                        )
405                    })?,
406                    deadline,
407                    cancelled,
408                    max_bytes,
409                    transferred: Arc::new(AtomicU64::new(0)),
410                    meter: Arc::clone(&meter),
411                    algorithm_fuel: Arc::new(work::AlgorithmWork::with_meter(Arc::clone(&meter))),
412                    denial: Arc::new(Mutex::new(None)),
413                    history_cache: caller.data().git_history.clone(),
414                    origin: crate::stdlib::shared::call_origin(&*caller),
415                };
416                let vfs = caller.data().vfs.clone();
417                let denial = job.denial.clone();
418                let principal = job.caller.clone();
419                let worker = async {
420                    caller
421                        .data_mut()
422                        .blocking_work
423                        .spawn(move || {
424                            job.check_cancelled()?;
425                            let _budget = budget;
426                            operation(&vfs, &job, args)
427                        })
428                        .await
429                        .map_err(blocking_worker_failure)?
430                };
431                let outcome = finish_worker(worker, &cancel_guard.0, deadline).await;
432                // The work is done, and may be published, whether or not it succeeded:
433                // it is settled, not refused.
434                meter.settle(&mut *caller)?;
435                if let Err(error) = &outcome
436                    && meter.is_exhausted()
437                    && !crate::runtime::host::ends_the_run(error)
438                    && !error.is::<stage::NeedsHostRecovery>()
439                {
440                    // Stopped for fuel before publishing: the run ends as a Wasm loop
441                    // out of fuel would, and the program cannot catch it. A host failure,
442                    // or a publication that needs recovery, says more and goes first.
443                    return Err(wasmtime::Trap::OutOfFuel.into());
444                }
445
446                let outcome = match outcome {
447                    Err(error)
448                        if crate::runtime::host::ends_the_run(&error)
449                            || error.is::<stage::NeedsHostRecovery>() =>
450                    {
451                        if error.is::<stage::NeedsHostRecovery>() {
452                            return Err(crate::runtime::host::fatal_host_error(error));
453                        }
454                        return Err(error);
455                    }
456                    other => other,
457                };
458                // Poison may leave denial attribution partly updated by the worker. AGENTS.md
459                // permits this poisoned-lock panic instead of recovery, including after worker
460                // cleanup; it does not permit the panic that caused poisoning.
461                let denial = denial.lock().expect("git: denial lock poisoned").take();
462                drop(cancel_guard);
463                if let Some((capability, reason)) = denial {
464                    return Err(permission_denied(&principal, &capability, reason));
465                }
466                outcome
467            })
468        })
469        .await
470    }
471}
472async fn before_deadline<T>(
473    deadline: tokio::time::Instant,
474    future: impl std::future::Future<Output = T>,
475) -> Result<T> {
476    if tokio::time::Instant::now() >= deadline {
477        bail!("git: operation timed out");
478    }
479    tokio::time::timeout_at(deadline, future)
480        .await
481        .map_err(|_| wasmtime::Error::msg("git: operation timed out"))
482}
483
484async fn finish_worker<T>(
485    worker: impl std::future::Future<Output = Result<T>>,
486    cancelled: &AtomicBool,
487    deadline: tokio::time::Instant,
488) -> Result<T> {
489    tokio::pin!(worker);
490    match before_deadline(deadline, &mut worker).await {
491        Ok(result) => result,
492        Err(error) => {
493            cancelled.store(true, Ordering::Relaxed);
494            // A catchable timeout must not let guest code resume while the
495            // worker is still publishing or rolling back repository changes.
496            match worker.await {
497                Err(worker_error)
498                    if crate::runtime::host::ends_the_run(&worker_error)
499                        || worker_error.is::<stage::NeedsHostRecovery>() =>
500                {
501                    Err(worker_error)
502                }
503                _ => Err(error),
504            }
505        }
506    }
507}
508
509fn blocking_worker_failure(error: crate::runtime::blocking::BlockingWorkError) -> wasmtime::Error {
510    crate::runtime::host::fatal_host_error("git blocking worker failed").context(error)
511}
512
513/// The least Git works with: what opening a repository and a fetch's index
514/// buffers take, measured at under 1 MB, with room to spare.
515const MIN_WORKING_BYTES: u64 = 4 * 1024 * 1024;
516
517/// Working memory reserved for one Git operation against the same cap as
518/// Wasm, while the program waits for it.
519///
520/// gix runs natively, outside the metered heap, so the operation reserves
521/// three quarters of what the run has free, and bounds what it holds by
522/// counting: trees, the index, worktree paths, a blob, a diff, a fetched
523/// pack's delta chains. Each count is held to the reservation divided by
524/// what the operation was measured to need per byte counted, beyond its
525/// fixed working set (`tests/git_memory.rs`).
526struct WorkingBudget {
527    bytes: u64,
528    counter: Arc<std::sync::atomic::AtomicU64>,
529}
530impl WorkingBudget {
531    fn reserve(limits: &crate::runtime::limits::TenantLimits, op: &str) -> Result<Self> {
532        let available = limits
533            .max_total_bytes
534            .saturating_sub(limits.observed_bytes())
535            .saturating_sub(limits.host_attached_bytes());
536        let bytes = available / 4 * 3;
537        if bytes < MIN_WORKING_BYTES {
538            const MB: u64 = 1024 * 1024;
539            bail!(
540                "git.{op} needs at least {} MB of working memory; {} MB is free under \
541                 max_execution_memory ({} MB). Raise max_execution_memory",
542                MIN_WORKING_BYTES * 4 / 3 / MB,
543                available / MB,
544                limits.max_total_bytes / MB
545            );
546        }
547        limits.charge_host_bytes(bytes)?;
548        Ok(Self {
549            bytes,
550            counter: limits.host_attached_counter(),
551        })
552    }
553
554    /// What `op` may count while it works: the reservation divided by what
555    /// it was measured to hold per byte counted. A diff holds its patch two
556    /// or three times over as it is built and encoded; everything else
557    /// holds at most about three times what it counts.
558    fn max_bytes(&self, op: &str) -> u64 {
559        let ratio = if op == "diff" { 8 } else { 4 };
560        (self.bytes.saturating_sub(MIN_WORKING_BYTES / 4) / ratio).min(storage::MAX_WORKING_BYTES)
561    }
562}
563impl Drop for WorkingBudget {
564    fn drop(&mut self) {
565        self.counter.fetch_sub(self.bytes, Ordering::Relaxed);
566    }
567}
568
569#[cfg(test)]
570mod tests;
571
572#[cfg(test)]
573mod integration_tests;
submilli_engine/stdlib/git/mod.rs

submilli_engine/stdlib/git/
mod.rs
