aion_server/api/handlers/
deploy.rs

1//! Shared operator-deploy handlers used by both transports.
2//!
3//! Every operation authorizes via [`crate::deploy::DeployGuard`] before any
4//! handler logic runs, then maps engine outcomes onto the deploy wire
5//! contract: `deploy_denied` for authorization, `version_pinned` for
6//! route-active/pinned refusals, `invalid_input` for malformed archives and
7//! the same-hash-different-manifest tripwire, `not_found` for unknown
8//! `(type, version)`, and 503/`Unavailable` for drain/shutdown windows.
9//! Mutations emit one structured audit line and the deploy metrics.
10
11use aion::EngineError;
12use aion_package::{ContentHash, ExtractionLimits, Package, PackageError};
13use aion_proto::{
14    ProtoListVersionsResponse, ProtoLoadPackageResponse, ProtoRouteVersionRequest,
15    ProtoRouteVersionResponse, ProtoUnloadVersionRequest, ProtoUnloadVersionResponse,
16    ProtoWorkflowVersion, WireError,
17};
18
19use crate::config::{DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED, DEPLOY_MAX_INFLATED_BYTES_REQUIRED};
20use crate::{CallerIdentity, ServerState};
21
22/// Deploy failure classes the transports must render distinctly: 503 vs 413
23/// vs the regular wire-code mapping.
24#[derive(Debug)]
25pub enum DeployApiError {
26    /// The server is draining or the engine is shutting down (503/`Unavailable`).
27    Unavailable(WireError),
28    /// The uploaded archive exceeds `deploy.max_archive_bytes`, or its
29    /// contents inflate past `deploy.max_inflated_bytes`
30    /// (413/`InvalidArgument`).
31    ArchiveTooLarge(WireError),
32    /// Mapped wire failure rendered through the standard code tables.
33    Wire(WireError),
34}
35
36impl DeployApiError {
37    /// Borrow the carried wire error regardless of class.
38    #[must_use]
39    pub const fn wire(&self) -> &WireError {
40        match self {
41            Self::Unavailable(wire) | Self::ArchiveTooLarge(wire) | Self::Wire(wire) => wire,
42        }
43    }
44}
45
46/// Handles a deploy archive upload (`LoadPackage` / `POST /deploy/packages`).
47///
48/// Idempotency is specified behavior: re-sending a resident archive succeeds
49/// with `freshly_loaded = false`, and `route_changed` reports whether the
50/// call re-pointed routing.
51///
52/// # Errors
53///
54/// Returns [`DeployApiError`] for authorization denials, drain/shutdown
55/// refusals, oversized or malformed archives, the manifest-mismatch
56/// tripwire, and engine failures.
57pub async fn load_package(
58    state: &ServerState,
59    caller: &CallerIdentity,
60    transport: &'static str,
61    archive: Vec<u8>,
62) -> Result<ProtoLoadPackageResponse, DeployApiError> {
63    authorize_mutation(state, caller, transport, "deploy.load")?;
64    enforce_archive_ceiling(state, archive.len())?;
65    let inflate_ceiling = inflate_ceiling(state)?;
66
67    // Network input: extraction is bounded by the operator's inflate ceiling
68    // so a DEFLATE bomb under the upload ceiling cannot exhaust memory.
69    let package = match Package::load_from_bytes(
70        &archive,
71        ExtractionLimits::bounded(inflate_ceiling),
72    ) {
73        Ok(package) => package,
74        Err(PackageError::InflatedSizeExceeded { limit }) => {
75            let wire = WireError::invalid_input(format!(
76                "archive contents inflate past the deploy.max_inflated_bytes limit of {limit} bytes; raise deploy.max_inflated_bytes (or AION_DEPLOY_MAX_INFLATED_BYTES) if this package size is intended"
77            ))
78            .with_error_type("Package");
79            return Err(refused(
80                state,
81                caller,
82                transport,
83                "deploy.load",
84                None,
85                DeployApiError::ArchiveTooLarge(wire),
86            ));
87        }
88        Err(error) => {
89            let wire = WireError::invalid_input(format!("archive rejected: {error}"))
90                .with_error_type("Package");
91            return Err(refused(
92                state,
93                caller,
94                transport,
95                "deploy.load",
96                None,
97                DeployApiError::Wire(wire),
98            ));
99        }
100    };
101    let engine = engine_handle(state)?;
102    // Taken BEFORE the load, which consumes the package. This is the only place
103    // on this side of the wire the `harness` section exists: it is deliberately
104    // not flow meaning, so it never reaches the identity-bound contract, and it
105    // rides in only as the archive's AWL provenance.
106    let awl = package.awl().cloned();
107    match engine.load_package(package).await {
108        Ok(outcome) => {
109            let workflow_type = outcome.record.workflow_type().to_owned();
110            let content_hash = outcome.record.version().to_string();
111            let audit_outcome = if outcome.freshly_loaded {
112                "loaded"
113            } else {
114                "idempotent"
115            };
116            tracing::info!(
117                operation = "deploy.load",
118                subject = caller.subject(),
119                grant_source = caller.grant_source().label(),
120                transport,
121                workflow_type = %workflow_type,
122                content_hash = %content_hash,
123                outcome = audit_outcome,
124                freshly_loaded = outcome.freshly_loaded,
125                route_changed = outcome.route_changed,
126                "deploy mutation applied"
127            );
128            record_mutation_metrics(state, "deploy.load", audit_outcome, &workflow_type);
129            let auto_workers = auto_provision(state, awl.as_ref(), &workflow_type).await;
130            Ok(ProtoLoadPackageResponse {
131                workflow_type,
132                content_hash,
133                deployed_entry_module: outcome.record.deployed_entry_module().to_owned(),
134                entry_function: outcome.record.entry_function().to_owned(),
135                freshly_loaded: outcome.freshly_loaded,
136                route_changed: outcome.route_changed,
137                superseded_versions: outcome
138                    .superseded_versions
139                    .iter()
140                    .map(ToString::to_string)
141                    .collect(),
142                auto_workers,
143            })
144        }
145        Err(error) => Err(map_engine_refusal(
146            state,
147            caller,
148            transport,
149            "deploy.load",
150            None,
151            error,
152        )),
153    }
154}
155
156/// Handles the deploy read model (`ListVersions` / `GET /deploy/versions`).
157///
158/// Listing keeps serving during drain: it is the operator's view of a
159/// rollout, not new work admission.
160///
161/// # Errors
162///
163/// Returns [`DeployApiError`] for authorization denials and engine failures.
164pub fn list_versions(
165    state: &ServerState,
166    caller: &CallerIdentity,
167    transport: &'static str,
168) -> Result<ProtoListVersionsResponse, DeployApiError> {
169    let guard = state.deploy_guard();
170    if let Err(error) = guard.authorize(caller) {
171        return Err(denied(state, caller, transport, "deploy.list", &error));
172    }
173    let engine = engine_handle(state)?;
174    let versions = engine
175        .list_workflow_versions()
176        .map_err(|error| DeployApiError::Wire(crate::ServerError::from(error).to_wire_error()))?;
177    Ok(ProtoListVersionsResponse {
178        versions: versions
179            .into_iter()
180            .map(|info| ProtoWorkflowVersion {
181                workflow_type: info.workflow_type,
182                content_hash: info.content_hash.to_string(),
183                deployed_entry_module: info.deployed_entry_module,
184                entry_function: info.entry_function,
185                manifest_version: info.manifest_version.as_str().to_owned(),
186                loaded_at: info.loaded_at.to_rfc3339(),
187                route_active: info.route_active,
188            })
189            .collect(),
190    })
191}
192
193/// Handles a route re-point (`RouteVersion` / `POST /deploy/route`).
194///
195/// # Errors
196///
197/// Returns [`DeployApiError`] for authorization denials, drain/shutdown
198/// refusals, malformed hashes, unknown versions, and engine failures.
199pub async fn route_version(
200    state: &ServerState,
201    caller: &CallerIdentity,
202    transport: &'static str,
203    request: ProtoRouteVersionRequest,
204) -> Result<ProtoRouteVersionResponse, DeployApiError> {
205    authorize_mutation(state, caller, transport, "deploy.route")?;
206    let (workflow_type, version) = decode_version_target(
207        state,
208        caller,
209        transport,
210        "deploy.route",
211        &request.workflow_type,
212        &request.content_hash,
213    )?;
214    let engine = engine_handle(state)?;
215    match engine
216        .route_workflow_version(&workflow_type, &version)
217        .await
218    {
219        Ok(()) => {
220            tracing::info!(
221                operation = "deploy.route",
222                subject = caller.subject(),
223                grant_source = caller.grant_source().label(),
224                transport,
225                workflow_type = %workflow_type,
226                content_hash = %version,
227                outcome = "rerouted",
228                "deploy mutation applied"
229            );
230            record_mutation_metrics(state, "deploy.route", "rerouted", &workflow_type);
231            Ok(ProtoRouteVersionResponse {})
232        }
233        Err(error) => Err(map_engine_refusal(
234            state,
235            caller,
236            transport,
237            "deploy.route",
238            Some((&workflow_type, &version)),
239            error,
240        )),
241    }
242}
243
244/// Handles a version unload (`UnloadVersion` / `POST /deploy/unload`).
245///
246/// # Errors
247///
248/// Returns [`DeployApiError`] for authorization denials, drain/shutdown
249/// refusals, malformed hashes, unknown versions, pinned/route-active
250/// refusals, and engine failures.
251pub async fn unload_version(
252    state: &ServerState,
253    caller: &CallerIdentity,
254    transport: &'static str,
255    request: ProtoUnloadVersionRequest,
256) -> Result<ProtoUnloadVersionResponse, DeployApiError> {
257    authorize_mutation(state, caller, transport, "deploy.unload")?;
258    let (workflow_type, version) = decode_version_target(
259        state,
260        caller,
261        transport,
262        "deploy.unload",
263        &request.workflow_type,
264        &request.content_hash,
265    )?;
266    let engine = engine_handle(state)?;
267    match engine
268        .unload_workflow_version(&workflow_type, &version)
269        .await
270    {
271        Ok(()) => {
272            tracing::info!(
273                operation = "deploy.unload",
274                subject = caller.subject(),
275                grant_source = caller.grant_source().label(),
276                transport,
277                workflow_type = %workflow_type,
278                content_hash = %version,
279                outcome = "unloaded",
280                "deploy mutation applied"
281            );
282            record_mutation_metrics(state, "deploy.unload", "unloaded", &workflow_type);
283            Ok(ProtoUnloadVersionResponse {})
284        }
285        Err(error) => Err(map_engine_refusal(
286            state,
287            caller,
288            transport,
289            "deploy.unload",
290            Some((&workflow_type, &version)),
291            error,
292        )),
293    }
294}
295
296/// Stand a built-in agent worker up for every queue the deployed document
297/// declares one on, and report what happened on every arm.
298///
299/// An archive that carries no AWL provenance reports NOTHING rather than a
300/// failure: a `.aion` built from a Gleam project has no document to read a
301/// `harness` section out of, and there is nothing there to have gone wrong.
302/// That is the same absence as a document with no section — no queue asked for
303/// a worker — so it renders identically.
304async fn auto_provision(
305    state: &ServerState,
306    awl: Option<&aion_package::AwlSource>,
307    workflow_type: &str,
308) -> Vec<aion_proto::ProtoAutoWorker> {
309    let Some(awl) = awl else {
310        return Vec::new();
311    };
312    let root = match crate::worker::auto_provision::document_root() {
313        Ok(root) => root,
314        Err(error) => {
315            // The package loaded; only the worker could not be provisioned.
316            // Reported to the caller rather than logged and dropped, because
317            // the deploy they just ran is the moment they can act on it.
318            tracing::error!(
319                operation = "worker.auto_provision",
320                workflow_type,
321                %error,
322                "no built-in agent worker could be provisioned for this document"
323            );
324            return vec![aion_proto::ProtoAutoWorker {
325                task_queue: String::new(),
326                decision: "failed".to_owned(),
327                deployment: String::new(),
328                detail: error.to_string(),
329            }];
330        }
331    };
332    crate::worker::auto_provision::provision(state, &root, awl, workflow_type)
333        .await
334        .into_iter()
335        .map(|outcome| aion_proto::ProtoAutoWorker {
336            task_queue: outcome.task_queue,
337            decision: outcome.decision.token().to_owned(),
338            deployment: outcome.deployment.unwrap_or_default(),
339            detail: outcome.detail,
340        })
341        .collect()
342}
343
344/// Authorization plus drain gate shared by every deploy mutation.
345fn authorize_mutation(
346    state: &ServerState,
347    caller: &CallerIdentity,
348    transport: &'static str,
349    operation: &'static str,
350) -> Result<(), DeployApiError> {
351    let guard = state.deploy_guard();
352    if let Err(error) = guard.authorize(caller) {
353        return Err(denied(state, caller, transport, operation, &error));
354    }
355    if state.drain_state().is_draining() {
356        return Err(DeployApiError::Unavailable(WireError::backend(
357            "server is draining and not accepting deploy mutations",
358        )));
359    }
360    Ok(())
361}
362
363/// Records, logs, and wraps an authorization denial. Denied calls never
364/// reach the engine.
365fn denied(
366    state: &ServerState,
367    caller: &CallerIdentity,
368    transport: &'static str,
369    operation: &'static str,
370    error: &crate::ServerError,
371) -> DeployApiError {
372    let wire = error.to_wire_error();
373    tracing::warn!(
374        operation,
375        subject = caller.subject(),
376        grant_source = caller.grant_source().label(),
377        transport,
378        reason = %wire.message,
379        "deploy operation denied"
380    );
381    if let Some(metrics) = state.metrics() {
382        metrics.deploy_denied(transport);
383    }
384    DeployApiError::Wire(wire)
385}
386
387/// Enforces the operator-configured archive ceiling, naming the config key.
388fn enforce_archive_ceiling(state: &ServerState, archive_len: usize) -> Result<(), DeployApiError> {
389    let Some(limit) = state.runtime_config().deploy.max_archive_bytes else {
390        // The deploy surface is only mounted when validation proved the
391        // ceiling present; reaching this is a wiring bug, never a caller
392        // error, and it must fail loudly rather than admit unbounded bodies.
393        return Err(DeployApiError::Wire(WireError::backend(
394            DEPLOY_MAX_ARCHIVE_BYTES_REQUIRED,
395        )));
396    };
397    if archive_len as u64 > limit {
398        return Err(DeployApiError::ArchiveTooLarge(WireError::invalid_input(
399            format!(
400                "archive is {archive_len} bytes, exceeding the deploy.max_archive_bytes limit of {limit} bytes; raise deploy.max_archive_bytes (or AION_DEPLOY_MAX_ARCHIVE_BYTES) if this package size is intended"
401            ),
402        )));
403    }
404    Ok(())
405}
406
407/// Resolves the operator-configured inflate ceiling for archive extraction.
408fn inflate_ceiling(state: &ServerState) -> Result<u64, DeployApiError> {
409    state
410        .runtime_config()
411        .deploy
412        .max_inflated_bytes
413        .ok_or_else(|| {
414            // The deploy surface is only mounted when validation proved the
415            // ceiling present; reaching this is a wiring bug, never a caller
416            // error, and it must fail loudly rather than extract unbounded.
417            DeployApiError::Wire(WireError::backend(DEPLOY_MAX_INFLATED_BYTES_REQUIRED))
418        })
419}
420
421/// Decodes a `(workflow_type, content_hash)` target, refusing malformed input.
422fn decode_version_target(
423    state: &ServerState,
424    caller: &CallerIdentity,
425    transport: &'static str,
426    operation: &'static str,
427    workflow_type: &str,
428    content_hash: &str,
429) -> Result<(String, ContentHash), DeployApiError> {
430    if workflow_type.is_empty() {
431        let wire = WireError::invalid_input("workflow_type must not be empty");
432        return Err(refused(
433            state,
434            caller,
435            transport,
436            operation,
437            None,
438            DeployApiError::Wire(wire),
439        ));
440    }
441    match content_hash.parse::<ContentHash>() {
442        Ok(version) => Ok((workflow_type.to_owned(), version)),
443        Err(error) => {
444            let wire = WireError::invalid_input(format!(
445                "content_hash `{content_hash}` is not a canonical content hash: {error}"
446            ));
447            Err(refused(
448                state,
449                caller,
450                transport,
451                operation,
452                None,
453                DeployApiError::Wire(wire),
454            ))
455        }
456    }
457}
458
459/// Maps an engine failure onto the deploy wire contract, emitting the audit
460/// line and refusal metrics.
461fn map_engine_refusal(
462    state: &ServerState,
463    caller: &CallerIdentity,
464    transport: &'static str,
465    operation: &'static str,
466    target: Option<(&str, &ContentHash)>,
467    error: EngineError,
468) -> DeployApiError {
469    let mapped = match error {
470        EngineError::ShuttingDown => DeployApiError::Unavailable(
471            WireError::backend(error.to_string()).with_error_type("ShuttingDown"),
472        ),
473        // On the deploy path archive/validation/collision/registration
474        // failures are caller-correctable input problems, not backend faults.
475        EngineError::Load { .. } => DeployApiError::Wire(
476            WireError::invalid_input(error.to_string()).with_error_type("Load"),
477        ),
478        EngineError::Package(_) => DeployApiError::Wire(
479            WireError::invalid_input(error.to_string()).with_error_type("Package"),
480        ),
481        // UnknownVersion -> not_found, VersionPinned/RouteActive ->
482        // version_pinned, ManifestMismatch -> invalid_input via the central
483        // ServerError mapping; refusal prose passes through verbatim.
484        other => DeployApiError::Wire(crate::ServerError::from(other).to_wire_error()),
485    };
486    let wire = mapped.wire();
487    let outcome = refusal_outcome(&mapped);
488    tracing::info!(
489        operation,
490        subject = caller.subject(),
491        grant_source = caller.grant_source().label(),
492        transport,
493        workflow_type = target.map(|(workflow_type, _)| workflow_type),
494        content_hash = target.map(|(_, version)| version.to_string()).as_deref(),
495        outcome,
496        reason = %wire.message,
497        "deploy mutation refused"
498    );
499    if let Some(metrics) = state.metrics() {
500        metrics.deploy_operation(operation, outcome);
501    }
502    mapped
503}
504
505/// Records, logs, and wraps an adapter-level refusal (malformed or
506/// oversized/over-inflating input), pre-mapped onto its wire class.
507fn refused(
508    state: &ServerState,
509    caller: &CallerIdentity,
510    transport: &'static str,
511    operation: &'static str,
512    target: Option<(&str, &ContentHash)>,
513    mapped: DeployApiError,
514) -> DeployApiError {
515    let outcome = refusal_outcome(&mapped);
516    tracing::info!(
517        operation,
518        subject = caller.subject(),
519        grant_source = caller.grant_source().label(),
520        transport,
521        workflow_type = target.map(|(workflow_type, _)| workflow_type),
522        content_hash = target.map(|(_, version)| version.to_string()).as_deref(),
523        outcome,
524        reason = %mapped.wire().message,
525        "deploy mutation refused"
526    );
527    if let Some(metrics) = state.metrics() {
528        metrics.deploy_operation(operation, outcome);
529    }
530    mapped
531}
532
533/// Stable refusal-class label for audit lines and the outcome metric.
534fn refusal_outcome(error: &DeployApiError) -> &'static str {
535    match error {
536        DeployApiError::Unavailable(_) => "unavailable",
537        DeployApiError::ArchiveTooLarge(_) | DeployApiError::Wire(_) => error.wire().code.as_str(),
538    }
539}
540
541fn engine_handle(state: &ServerState) -> Result<std::sync::Arc<aion::Engine>, DeployApiError> {
542    state
543        .deploy_guard()
544        .engine()
545        .map(std::sync::Arc::clone)
546        .map_err(|error| DeployApiError::Wire(error.to_wire_error()))
547}
548
549/// Counter + gauge updates for one applied mutation. The gauge is set from
550/// the post-operation listing for the affected workflow type (0 when the
551/// last version of a type was unloaded).
552fn record_mutation_metrics(
553    state: &ServerState,
554    operation: &'static str,
555    outcome: &'static str,
556    workflow_type: &str,
557) {
558    let Some(metrics) = state.metrics() else {
559        return;
560    };
561    metrics.deploy_operation(operation, outcome);
562    let Ok(engine) = state.deploy_guard().engine().map(std::sync::Arc::clone) else {
563        return;
564    };
565    match engine.list_workflow_versions() {
566        Ok(versions) => {
567            let count = versions
568                .iter()
569                .filter(|info| info.workflow_type == workflow_type)
570                .count();
571            let count = i64::try_from(count).unwrap_or(i64::MAX);
572            metrics.set_loaded_workflow_versions(workflow_type, count);
573        }
574        Err(error) => {
575            tracing::warn!(
576                operation,
577                workflow_type,
578                %error,
579                "post-operation version listing failed; loaded-version gauge not updated"
580            );
581        }
582    }
583}
aion_server/api/handlers/deploy.rs

aion_server/api/handlers/
deploy.rs
