platform_provider/
runtime.rs

1use crate::ProviderHostEffectCoordinator;
2use crate::config::ProviderConfig;
3use crate::invocation::{self, InvocationContext};
4use crate::protocol::{ProviderFunctionInvokeRequest, ProviderFunctionInvokeResponse};
5use crate::protocol::{ProviderInvocationMode, ProviderOperationKind};
6use crate::validation::validate_path_segment;
7use platform_core::{AppError, AppResult, ErrorCode, ExecutionContext};
8use platform_runtime::{
9    FunctionHandlerObservability, FunctionTerminalObservation, RuntimeFunction,
10};
11use serde_json::Value;
12use std::time::Duration;
13
14#[derive(Debug, Clone)]
15pub struct ProviderRuntimeFunction {
16    client: reqwest::Client,
17    config: ProviderConfig,
18    function_name: String,
19    effects: ProviderHostEffectCoordinator,
20}
21
22impl ProviderRuntimeFunction {
23    pub fn new(
24        config: ProviderConfig,
25        function_name: impl Into<String>,
26        effects: ProviderHostEffectCoordinator,
27    ) -> AppResult<Self> {
28        if config.name.trim().is_empty() {
29            return Err(AppError::new(
30                ErrorCode::Validation,
31                "provider runtime Module identity must be non-empty",
32            ));
33        }
34        let function_name = function_name.into();
35        validate_function_name(&function_name)?;
36        let client = reqwest::Client::builder()
37            .timeout(Duration::from_millis(config.timeout_ms))
38            .build()
39            .map_err(|error| {
40                AppError::new(
41                    ErrorCode::Internal,
42                    format!("failed to build provider runtime client: {error}"),
43                )
44            })?;
45        Ok(Self {
46            client,
47            config,
48            function_name,
49            effects,
50        })
51    }
52
53    pub async fn invoke(&self, ctx: ExecutionContext, input: Value) -> AppResult<Value> {
54        let function_run_id = ctx.execution_id.0.clone();
55        let invocation_id = runtime_invocation_id(&function_run_id, ctx.attempt);
56        let request_body = ProviderFunctionInvokeRequest {
57            request_id: invocation_id.clone(),
58            function_run_id,
59            function_name: self.function_name.clone(),
60            attempt: ctx.attempt,
61            correlation_id: ctx.correlation_id.0,
62            causation_id: ctx.causation_id,
63            actor: ctx.actor.clone(),
64            trace: ctx.trace.clone(),
65            input,
66        };
67        let invocation = invocation::build(
68            &self.config,
69            ProviderOperationKind::RuntimeFunction,
70            &self.function_name,
71            "1",
72            ProviderInvocationMode::Durable,
73            InvocationContext {
74                invocation_id,
75                request_id: request_body.request_id.clone(),
76                attempt: request_body.attempt,
77                actor: request_body.actor.clone(),
78                tenant_id: ctx.tenant_id.map(|tenant| tenant.0),
79                correlation_id: request_body.correlation_id.clone(),
80                causation_id: request_body.causation_id.clone(),
81                trace: request_body.trace.clone(),
82            },
83            serde_json::to_value(request_body).map_err(|error| {
84                AppError::new(
85                    ErrorCode::Internal,
86                    format!("encode Provider payload: {error}"),
87                )
88            })?,
89        )?;
90        let outcome = invocation::send(
91            &self.client,
92            &self.config,
93            &self.effects,
94            "runtime:invoke",
95            &invocation,
96        )
97        .await?;
98        let value = invocation::result(&invocation, outcome)?;
99        let response: ProviderFunctionInvokeResponse =
100            serde_json::from_value(value).map_err(|error| {
101                AppError::new(
102                    ErrorCode::ExternalDependency,
103                    format!("Provider runtime result violated its contract: {error}"),
104                )
105            })?;
106        Ok(response.output)
107    }
108}
109
110#[async_trait::async_trait]
111impl RuntimeFunction for ProviderRuntimeFunction {
112    async fn call(&self, ctx: ExecutionContext, input: Value) -> AppResult<Value> {
113        self.invoke(ctx, input).await
114    }
115
116    fn observability(&self) -> Option<FunctionHandlerObservability> {
117        Some(FunctionHandlerObservability::new(
118            "provider_runtime",
119            serde_json::json!({
120                "module_name": &self.config.name,
121                "function_name": &self.function_name,
122                "provider_path": format!("/exports/{}/runtime:invoke", self.config.export_key),
123                "timeout_ms": self.config.timeout_ms,
124            }),
125        ))
126    }
127
128    fn terminal_observation(&self) -> Option<FunctionTerminalObservation> {
129        Some(FunctionTerminalObservation::new(self.config.name.clone()))
130    }
131}
132
133pub(crate) fn validate_function_name(function_name: &str) -> AppResult<()> {
134    validate_path_segment(
135        function_name,
136        "provider runtime function name must be a stable path segment",
137    )
138}
139
140fn runtime_invocation_id(function_run_id: &str, attempt: u32) -> String {
141    format!("{function_run_id}:attempt:{attempt}")
142}
143
144#[cfg(test)]
145mod tests {
146    use super::runtime_invocation_id;
147
148    #[test]
149    fn outer_invocation_identity_is_stable_per_function_run_attempt() {
150        assert_eq!(
151            runtime_invocation_id("fnrun-1", 2),
152            runtime_invocation_id("fnrun-1", 2)
153        );
154        assert_ne!(
155            runtime_invocation_id("fnrun-1", 1),
156            runtime_invocation_id("fnrun-1", 2)
157        );
158        assert_ne!(
159            runtime_invocation_id("fnrun-1", 1),
160            runtime_invocation_id("fnrun-2", 1)
161        );
162    }
163}
platform_provider/runtime.rs

platform_provider/
runtime.rs
