feldera_types/
runtime_status.rs

1use crate::checkpoint::CheckpointMetadata;
2use crate::error::ErrorResponse;
3use actix_web::body::BoxBody;
4use actix_web::http::StatusCode;
5use actix_web::{HttpRequest, HttpResponse, HttpResponseBuilder, Responder, ResponseError};
6use bytemuck::NoUninit;
7use clap::ValueEnum;
8use serde::{Deserialize, Serialize};
9use serde_json::json;
10use std::fmt;
11use std::fmt::Display;
12use utoipa::ToSchema;
13
14/// Runtime status of the pipeline.
15///
16/// Of the statuses, only `Unavailable` is determined by the runner. All other statuses are
17/// determined by the pipeline and taken over by the runner.
18#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize, ToSchema, NoUninit)]
19#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
20#[repr(u8)]
21pub enum RuntimeStatus {
22    /// The runner was unable to determine the pipeline runtime status. This status is never
23    /// returned by the pipeline endpoint itself, but only determined by the runner.
24    ///
25    /// It can notably occur in two scenarios:
26    /// 1. The runner is unable to (in time) receive a response for its sent request to the
27    ///    pipeline `/status` endpoint, or it is unable to parse the response.
28    /// 2. The runner received back a `503 Service Unavailable` as a response to the request.
29    ///    This can occur for example if the pipeline is unable to acquire a lock necessary to
30    ///    determine whether it is in any of the other runtime statuses.
31    Unavailable,
32
33    /// The pipeline is waiting for initialization instructions from the
34    /// coordinator.
35    Coordination,
36
37    /// The pipeline is constantly pulling the latest checkpoint from S3 but not processing any inputs.
38    Standby,
39
40    /// The input and output connectors are establishing connections to their data sources and sinks
41    /// respectively.
42    Initializing,
43
44    /// The pipeline was modified since the last checkpoint. User approval is required before
45    /// bootstrapping can proceed.
46    AwaitingApproval,
47
48    /// The pipeline was modified since the last checkpoint, and is currently bootstrapping modified
49    /// views.
50    Bootstrapping,
51
52    /// Input records that were stored in the journal but were not yet processed, are being
53    /// processed first.
54    Replaying,
55
56    /// The input connectors are paused.
57    Paused,
58
59    /// The input connectors are running.
60    Running,
61
62    /// The pipeline finished checkpointing and pausing.
63    Suspended,
64
65    /// A concurrent bootstrap is in progress: the pre-existing views are live
66    /// and serving while new/modified views backfill in the background.
67    ConcurrentBootstrapping,
68
69    /// A concurrent bootstrap is in its cutover window: inputs are briefly
70    /// paused while the backfilled views are synchronized and brought online.
71    Synchronizing,
72}
73
74impl From<RuntimeDesiredStatus> for RuntimeStatus {
75    fn from(value: RuntimeDesiredStatus) -> Self {
76        match value {
77            RuntimeDesiredStatus::Unavailable => Self::Unavailable,
78            RuntimeDesiredStatus::Coordination => Self::Coordination,
79            RuntimeDesiredStatus::Standby => Self::Standby,
80            RuntimeDesiredStatus::Paused => Self::Paused,
81            RuntimeDesiredStatus::Running => Self::Running,
82            RuntimeDesiredStatus::Suspended => Self::Suspended,
83        }
84    }
85}
86
87#[derive(Debug, Clone, Copy, Eq, PartialEq, Deserialize, Serialize, ToSchema, ValueEnum)]
88#[cfg_attr(feature = "testing", derive(proptest_derive::Arbitrary))]
89pub enum RuntimeDesiredStatus {
90    Unavailable,
91    Coordination,
92    Standby,
93    Paused,
94    Running,
95    Suspended,
96}
97
98impl RuntimeDesiredStatus {
99    pub fn may_transition_to(&self, target: Self) -> bool {
100        match (*self, target) {
101            (old, new) if old == new => true,
102            (Self::Standby, Self::Paused | Self::Running) => true,
103            (Self::Paused, Self::Running | Self::Suspended) => true,
104            (Self::Running, Self::Paused | Self::Suspended) => true,
105            _ => false,
106        }
107    }
108
109    pub fn may_transition_to_at_startup(&self, target: Self) -> bool {
110        match (*self, target) {
111            (_, Self::Coordination) => true,
112            (Self::Suspended, _) => {
113                // A suspended pipeline must transition to "paused" or
114                // "running".
115                matches!(target, Self::Paused | Self::Running)
116            }
117            (old, new) if old.may_transition_to(new) => true,
118            _ => false,
119        }
120    }
121}
122
123/// Some of our JSON interface uses capitalized status names, like `Paused`, but
124/// other parts use snake-case names, like `paused`.  To support the latter with
125/// `serde`, use this module in the field declaration, e.g.:
126///
127/// ```ignore
128/// #[serde(with = "feldera_types::runtime_status::snake_case_runtime_desired_status")]
129/// ```
130pub mod snake_case_runtime_desired_status {
131    use serde::{Deserialize, Deserializer, Serialize, Serializer};
132
133    use crate::runtime_status::RuntimeDesiredStatus;
134
135    #[derive(Debug, Clone, Copy, Eq, PartialEq, Deserialize, Serialize)]
136    #[serde(rename_all = "snake_case")]
137    enum SnakeRuntimeDesiredStatus {
138        Unavailable,
139        Coordination,
140        Standby,
141        Paused,
142        Running,
143        Suspended,
144    }
145
146    impl From<RuntimeDesiredStatus> for SnakeRuntimeDesiredStatus {
147        fn from(value: RuntimeDesiredStatus) -> Self {
148            match value {
149                RuntimeDesiredStatus::Unavailable => SnakeRuntimeDesiredStatus::Unavailable,
150                RuntimeDesiredStatus::Coordination => SnakeRuntimeDesiredStatus::Coordination,
151                RuntimeDesiredStatus::Standby => SnakeRuntimeDesiredStatus::Standby,
152                RuntimeDesiredStatus::Paused => SnakeRuntimeDesiredStatus::Paused,
153                RuntimeDesiredStatus::Running => SnakeRuntimeDesiredStatus::Running,
154                RuntimeDesiredStatus::Suspended => SnakeRuntimeDesiredStatus::Suspended,
155            }
156        }
157    }
158
159    impl From<SnakeRuntimeDesiredStatus> for RuntimeDesiredStatus {
160        fn from(value: SnakeRuntimeDesiredStatus) -> Self {
161            match value {
162                SnakeRuntimeDesiredStatus::Unavailable => RuntimeDesiredStatus::Unavailable,
163                SnakeRuntimeDesiredStatus::Coordination => RuntimeDesiredStatus::Coordination,
164                SnakeRuntimeDesiredStatus::Standby => RuntimeDesiredStatus::Standby,
165                SnakeRuntimeDesiredStatus::Paused => RuntimeDesiredStatus::Paused,
166                SnakeRuntimeDesiredStatus::Running => RuntimeDesiredStatus::Running,
167                SnakeRuntimeDesiredStatus::Suspended => RuntimeDesiredStatus::Suspended,
168            }
169        }
170    }
171
172    pub fn serialize<S>(value: &RuntimeDesiredStatus, serializer: S) -> Result<S::Ok, S::Error>
173    where
174        S: Serializer,
175    {
176        SnakeRuntimeDesiredStatus::from(*value).serialize(serializer)
177    }
178
179    pub fn deserialize<'de, D>(deserializer: D) -> Result<RuntimeDesiredStatus, D::Error>
180    where
181        D: Deserializer<'de>,
182    {
183        SnakeRuntimeDesiredStatus::deserialize(deserializer).map(|status| status.into())
184    }
185}
186
187#[derive(
188    Debug, Default, Clone, Copy, Eq, PartialEq, Deserialize, Serialize, ToSchema, NoUninit,
189)]
190#[repr(u8)]
191#[serde(rename_all = "snake_case")]
192pub enum BootstrapPolicy {
193    Allow,
194    Reject,
195    #[default]
196    AwaitApproval,
197}
198
199impl TryFrom<Option<String>> for BootstrapPolicy {
200    type Error = ();
201
202    fn try_from(value: Option<String>) -> Result<Self, Self::Error> {
203        match value.as_deref() {
204            Some("allow") => Ok(Self::Allow),
205            Some("reject") => Ok(Self::Reject),
206            Some("await_approval") | None => Ok(Self::AwaitApproval),
207            _ => Err(()),
208        }
209    }
210}
211
212impl From<String> for BootstrapPolicy {
213    fn from(value: String) -> Self {
214        match value.as_str() {
215            "allow" => Self::Allow,
216            "reject" => Self::Reject,
217            "await_approval" => Self::AwaitApproval,
218            _ => panic!("Invalid 'bootstrap_policy' value: {value}"),
219        }
220    }
221}
222
223impl Display for BootstrapPolicy {
224    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
225        let s = match self {
226            BootstrapPolicy::Allow => "allow",
227            BootstrapPolicy::Reject => "reject",
228            BootstrapPolicy::AwaitApproval => "await_approval",
229        };
230        write!(f, "{s}")
231    }
232}
233
234/// Bootstrap-related configuration for a deployment start request.
235#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize, ToSchema, Default)]
236pub struct BootstrapConfig {
237    /// Bootstrap policy.
238    #[serde(default)]
239    pub bootstrap_policy: Option<BootstrapPolicy>,
240    /// Bootstrap the pipeline with output connectors disabled.
241    #[serde(default)]
242    pub silent_bootstrap: bool,
243    /// Bootstrap new and modified views concurrently, keeping the pre-existing
244    /// views live while the new ones backfill in the background.
245    ///
246    /// Mutually exclusive with `silent_bootstrap`. When set, a circuit that
247    /// cannot be bootstrapped concurrently fails the pipeline instead of
248    /// falling back to a stop-the-world bootstrap.
249    #[serde(default)]
250    pub concurrent_bootstrap: bool,
251}
252
253impl From<BootstrapPolicy> for BootstrapConfig {
254    fn from(bootstrap_policy: BootstrapPolicy) -> Self {
255        Self {
256            bootstrap_policy: Some(bootstrap_policy),
257            silent_bootstrap: false,
258            concurrent_bootstrap: false,
259        }
260    }
261}
262
263impl BootstrapConfig {
264    pub fn with_silent_bootstrap(self, silent_bootstrap: bool) -> Self {
265        Self {
266            silent_bootstrap,
267            ..self
268        }
269    }
270
271    pub fn with_concurrent_bootstrap(self, concurrent_bootstrap: bool) -> Self {
272        Self {
273            concurrent_bootstrap,
274            ..self
275        }
276    }
277
278    /// Validates that the bootstrap options are mutually consistent.
279    ///
280    /// `silent_bootstrap` and `concurrent_bootstrap` cannot both be set:
281    /// silent bootstrap suppresses outputs during a stop-the-world bootstrap,
282    /// whereas concurrent bootstrap keeps the old views (and their outputs)
283    /// live, so the two requests contradict each other.
284    pub fn validate(&self) -> Result<(), String> {
285        if self.silent_bootstrap && self.concurrent_bootstrap {
286            return Err(
287                "`silent_bootstrap` and `concurrent_bootstrap` are mutually exclusive; \
288                 set at most one"
289                    .to_string(),
290            );
291        }
292        Ok(())
293    }
294
295    /// Returns the bootstrap policy for an active deployment.
296    pub fn active_bootstrap_policy(&self) -> BootstrapPolicy {
297        self.bootstrap_policy
298            .expect("bootstrap policy must be set for an active deployment")
299    }
300}
301
302/// Details about pipeline storage, which are returned as part of the regular runtime status polling
303/// by the runner.
304#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, ToSchema)]
305pub struct StorageStatusDetails {
306    /// Present checkpoints.
307    pub checkpoints: Vec<CheckpointMetadata>,
308}
309
310#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
311pub struct ExtendedRuntimeStatus {
312    /// Runtime status of the pipeline.
313    pub runtime_status: RuntimeStatus,
314
315    /// Human-readable details about the runtime status. Its content can contain for instance an
316    /// explanation why it is in this status and any other additional information about it (e.g.,
317    /// progress).
318    pub runtime_status_details: serde_json::Value,
319
320    /// Runtime desired status of the pipeline.
321    pub runtime_desired_status: RuntimeDesiredStatus,
322
323    /// Details about the pipeline persistent storage.
324    ///
325    /// `None` indicates that the pipeline in its current runtime status is unable to check the
326    /// storage status details. Returning `None` _does not_ override the already existing storage
327    /// status details in the database of the runner.
328    pub storage_status_details: Option<StorageStatusDetails>,
329}
330
331impl Responder for ExtendedRuntimeStatus {
332    type Body = BoxBody;
333
334    fn respond_to(self, _req: &HttpRequest) -> HttpResponse<Self::Body> {
335        HttpResponseBuilder::new(StatusCode::OK).json(self)
336    }
337}
338
339impl From<ExtendedRuntimeStatus> for HttpResponse<BoxBody> {
340    fn from(value: ExtendedRuntimeStatus) -> Self {
341        HttpResponseBuilder::new(StatusCode::OK).json(value)
342    }
343}
344
345/// Error returned by the pipeline `/status` endpoint.
346#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
347pub struct ExtendedRuntimeStatusError {
348    /// Status code. Returning anything except `503 Service Unavailable` will cause the runner to
349    /// forcefully stop the pipeline.
350    #[serde(with = "status_code")]
351    pub status_code: StatusCode,
352
353    /// Error response.
354    pub error: ErrorResponse,
355}
356
357mod status_code {
358    use actix_web::http::StatusCode;
359    use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error};
360
361    pub fn serialize<S>(value: &StatusCode, serializer: S) -> Result<S::Ok, S::Error>
362    where
363        S: Serializer,
364    {
365        value.as_u16().serialize(serializer)
366    }
367
368    pub fn deserialize<'de, D>(deserializer: D) -> Result<StatusCode, D::Error>
369    where
370        D: Deserializer<'de>,
371    {
372        let value = u16::deserialize(deserializer)?;
373        StatusCode::from_u16(value).map_err(D::Error::custom)
374    }
375}
376
377impl Display for ExtendedRuntimeStatusError {
378    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
379        write!(f, "{}: {:?}", self.status_code, self.error)
380    }
381}
382
383impl ResponseError for ExtendedRuntimeStatusError {
384    fn status_code(&self) -> StatusCode {
385        self.status_code
386    }
387
388    fn error_response(&self) -> HttpResponse<BoxBody> {
389        HttpResponseBuilder::new(self.status_code()).json(self.error.clone())
390    }
391}
392
393/// Details about the current runtime status. The fields in this struct should all be **optional**
394/// and set only by a runtime status when they are known. Otherwise, they can just be set `None`.
395#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, Default, Eq, ToSchema)]
396pub struct RuntimeStatusDetails {
397    /// Free form text giving an explanation why it is currently in this runtime status.
398    ///
399    /// Specifically useful for: `Unavailable`, `Initializing`.
400    #[serde(skip_serializing_if = "Option::is_none")]
401    pub reason: Option<String>,
402
403    /// Statistics across all connectors.
404    ///
405    /// Specifically useful for: `Paused`, `Running`.
406    #[serde(skip_serializing_if = "Option::is_none")]
407    pub connector_stats: Option<ConnectorStats>,
408
409    /// The diff which is awaiting approval.
410    ///
411    /// Specifically useful for: `AwaitingApproval`.
412    #[serde(skip_serializing_if = "Option::is_none")]
413    pub approval_diff: Option<serde_json::Value>,
414    // Backward compatibility: in older versions, the approval diff was the runtime status details
415    // value itself. To distinguish between the old and new version, clients can check if the
416    // `program_diff` field (one of the fields within the `approval_diff`) is present if they
417    // expect there to be a diff (i.e., when the runtime status is `AwaitingApproval`). As such,
418    // `program_diff` is a reserved field name that cannot be added here in the future.
419}
420
421impl RuntimeStatusDetails {
422    pub fn new_only_reason(reason: &str) -> Self {
423        Self {
424            reason: Some(reason.to_string()),
425            ..Self::default()
426        }
427    }
428
429    /// Serializes the runtime status details to JSON. If the serialization errors, an error JSON
430    /// is returned instead. This makes sure this method does not panic unexpectedly and that the
431    /// error bubbles up. The details are only supplementary information, and as such are not
432    /// critical to operation.
433    pub fn serialize_guaranteed(self) -> serde_json::Value {
434        serde_json::to_value(self).unwrap_or_else(|e| {
435            json!({
436                "reason": format!("unable to serialize runtime status details due to: {e}")
437            })
438        })
439    }
440}
441
442/// Statistics across all connectors.
443#[derive(Serialize, Deserialize, ToSchema, Eq, PartialEq, Debug, Clone)]
444pub struct ConnectorStats {
445    /// Total number of errors across all connectors.
446    ///
447    /// - `num_transport_errors` from all input connectors
448    /// - `num_parse_errors` from all input connectors
449    /// - `num_encode_errors` from all output connectors
450    /// - `num_transport_errors` from all output connectors
451    pub num_errors: u64,
452}
feldera_types/runtime_status.rs

feldera_types/
runtime_status.rs
