rocketmq_auth/runtime/
startup.rs

1// Copyright 2026 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::Arc;
16use std::time::Duration;
17
18use rocketmq_runtime::ChildServiceContext;
19use rocketmq_runtime::MetadataDeadline;
20use rocketmq_runtime::MetadataIoActor;
21use rocketmq_runtime::ShutdownDeadline;
22use tokio::sync::Mutex;
23
24use crate::provider_owner::ProviderOwner;
25use crate::AuthFailureKind;
26use crate::AuthOperation;
27use crate::AuthServiceError;
28use crate::AuthServiceResult;
29
30/// Startup failure with the original cause and any incomplete cleanup retained.
31///
32/// Obtain this value from the source of an [`AuthServiceError`]. A failed or
33/// timed-out rollback retains its providers and I/O owner for an explicit retry;
34/// dropping the error does not perform asynchronous cleanup.
35pub struct AuthStartupFailure {
36    primary: AuthServiceError,
37    cleanup_error: Option<AuthServiceError>,
38    cleanup: StartupCleanup,
39}
40
41impl AuthStartupFailure {
42    pub fn primary(&self) -> &AuthServiceError {
43        &self.primary
44    }
45
46    pub fn cleanup_error(&self) -> Option<&AuthServiceError> {
47        self.cleanup_error.as_ref()
48    }
49
50    /// Resumes rollback using the retained owner and one absolute deadline.
51    ///
52    /// # Errors
53    ///
54    /// Returns a typed timeout or cleanup failure while resources still need
55    /// finalization. Successful controls are not closed again on subsequent calls.
56    pub async fn retry_cleanup_until(&self, deadline: ShutdownDeadline) -> AuthServiceResult<()> {
57        self.cleanup.run(deadline).await
58    }
59}
60
61impl std::fmt::Debug for AuthStartupFailure {
62    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63        f.debug_struct("AuthStartupFailure")
64            .field("primary", &self.primary)
65            .field("cleanup_error", &self.cleanup_error)
66            .finish_non_exhaustive()
67    }
68}
69
70impl std::fmt::Display for AuthStartupFailure {
71    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
72        f.write_str("authentication runtime startup failed")
73    }
74}
75
76impl std::error::Error for AuthStartupFailure {
77    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
78        Some(&self.primary)
79    }
80}
81
82struct StartupCleanup {
83    provider: Option<Arc<ProviderOwner>>,
84    metadata_io: Option<MetadataIoActor>,
85    context: ChildServiceContext,
86    completed: Mutex<bool>,
87}
88
89impl StartupCleanup {
90    async fn run(&self, deadline: ShutdownDeadline) -> AuthServiceResult<()> {
91        let at = tokio::time::Instant::from_std(deadline.instant());
92        let timeout = || AuthServiceError::new(AuthOperation::MaintainService, AuthFailureKind::Timeout);
93        let mut completed = tokio::time::timeout_at(at, self.completed.lock())
94            .await
95            .map_err(|_| timeout())?;
96        if *completed {
97            return Ok(());
98        }
99        if let Some(provider) = &self.provider {
100            // Keep the I/O lane alive until provider cleanup finishes, including
101            // after a timeout. Providers may need it to release their resources.
102            provider.shutdown_until(deadline, false).await?;
103        }
104        if let Some(metadata_io) = &self.metadata_io {
105            let report = metadata_io.shutdown_until(MetadataDeadline::at(at)).await;
106            if report.timed_out || report.pending_operations != 0 {
107                return Err(timeout());
108            }
109        }
110        let report = self.context.task_group().shutdown(deadline.remaining()).await;
111        if !report.is_healthy() {
112            return Err(timeout());
113        }
114        *completed = true;
115        Ok(())
116    }
117}
118
119pub(super) async fn rollback(
120    primary: AuthServiceError,
121    provider: Option<Arc<ProviderOwner>>,
122    metadata_io: Option<MetadataIoActor>,
123    context: ChildServiceContext,
124) -> AuthServiceError {
125    let cleanup = StartupCleanup {
126        provider,
127        metadata_io,
128        context,
129        completed: Mutex::new(false),
130    };
131    let cleanup_error = cleanup.run(ShutdownDeadline::after(Duration::from_secs(5))).await.err();
132    AuthServiceError::with_source(
133        primary.operation(),
134        primary.kind(),
135        AuthStartupFailure {
136            primary,
137            cleanup_error,
138            cleanup,
139        },
140    )
141}
rocketmq_auth/runtime/startup.rs

rocketmq_auth/runtime/
startup.rs
