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

submilli_engine/stdlib/git/
mod.rs
