1/*
2 * SPDX-FileCopyrightText: 2026 Copyright (c) Contributors to the Eclipse Foundation
3 *
4 * See the NOTICE file(s) distributed with this work for additional
5 * information regarding copyright ownership.
6 *
7 * This program and the accompanying materials are made available under the
8 * terms of the Apache License Version 2.0 which is available at
9 * https://www.apache.org/licenses/LICENSE-2.0
10 *
11 * SPDX-License-Identifier: Apache-2.0
12 */
13
14use std::{sync::Arc, time::Duration};
15
16use cda_interfaces::{
17 HashMap,
18 communication_control::{
19 ActivationCause, CommunicationAccess, DisableCommunication, DisableError, DisableGuard,
20 DisableReason, PostUpdateCommunicationMode,
21 },
22 http_protection::registry::{
23 HttpProtectionConfig, HttpProtectionReason, HttpProtectionRegistry, HttpStatusCode,
24 OwnedHttpProtection,
25 },
26 runtime_update_api::{
27 ExecutionMode, ExecutionStatus, LockStateProvider, RuntimeReloaderPlugin,
28 RuntimeUpdateError, RuntimeUpdateSecurityPlugin, UpdateCollections, UpdateExecution,
29 },
30 storage_api::{CollectionName, Storage},
31};
32use tokio::sync::RwLock;
33
34pub(crate) struct ExecutionParams<'a, S, R: ?Sized, T, L> {
35 pub(crate) storage: &'a Arc<S>,
36 pub(crate) security_handler: &'a Arc<T>,
37 pub(crate) reload_handler: &'a Arc<R>,
38 pub(crate) executions: &'a Arc<RwLock<HashMap<String, UpdateExecution>>>,
39 pub(crate) communication_disable: &'a Arc<dyn DisableCommunication>,
40 pub(crate) communication_access: &'a Arc<dyn CommunicationAccess>,
41 pub(crate) http_protections: &'a HttpProtectionRegistry,
42 pub(crate) update_retry_after: Duration,
43 pub(crate) post_update_mode: PostUpdateCommunicationMode,
44 pub(crate) mdd_decompress: bool,
45 pub(crate) lock_state_provider: &'a L,
46}
47
48fn http_protection_config_for_update(retry_after: Duration) -> HttpProtectionConfig {
49 HttpProtectionConfig::new(
50 HttpProtectionReason::UpdateInProgress,
51 HttpStatusCode::CONFLICT,
52 "Update in progress",
53 )
54 .with_exempt_routes(cda_sovd::routes_accessible_during_update())
55 .with_retry_after(retry_after)
56}
57
58pub(crate) async fn start_execution<S, R, T, L>(
59 params: &ExecutionParams<'_, S, R, T, L>,
60 mode: ExecutionMode,
61) -> Result<String, RuntimeUpdateError>
62where
63 S: Storage + Send + Sync + 'static,
64 R: RuntimeReloaderPlugin + ?Sized,
65 T: RuntimeUpdateSecurityPlugin<L, S::CollectionHandle>,
66 L: LockStateProvider,
67{
68 let (protection, disable_lease) = acquire_execution_guards(params).await?;
69 let collections = match load_update_collections(&**params.storage).await {
70 Ok(collections) => collections,
71 Err(error) => return Err(reject_execution(error, protection, disable_lease).await),
72 };
73 let (protection, disable_lease) =
74 validate_execution_preconditions(params, mode, &collections, protection, disable_lease)
75 .await?;
76 let execution_id = register_execution(params.executions, mode).await;
77
78 spawn_execution(
79 mode,
80 execution_id.clone(),
81 Arc::clone(params.storage),
82 Arc::clone(params.reload_handler),
83 Arc::clone(params.executions),
84 params.mdd_decompress,
85 params.post_update_mode.clone(),
86 Arc::clone(params.communication_access),
87 disable_lease,
88 protection,
89 );
90
91 Ok(execution_id)
92}
93
94async fn acquire_execution_guards<S, R: ?Sized, T, L>(
95 params: &ExecutionParams<'_, S, R, T, L>,
96) -> Result<(OwnedHttpProtection, Box<dyn DisableGuard>), RuntimeUpdateError> {
97 let protection = params
98 .http_protections
99 .protect(http_protection_config_for_update(params.update_retry_after))
100 .map_err(|error| {
101 // Built in-process, so this is a programming error. Refuse the
102 // execution rather than run it unprotected.
103 RuntimeUpdateError::FatalError(format!(
104 "Invalid HTTP protection configuration for update: {error}"
105 ))
106 })?;
107 let disable_lease = params
108 .communication_disable
109 .disable(DisableReason::RuntimeUpdate)
110 .await
111 .map_err(|error| match error {
112 DisableError::Conflict => RuntimeUpdateError::ExecutionConflict,
113 DisableError::InUse => RuntimeUpdateError::OperationsInProgress(
114 "another operation is running (i.e. flash transfer)".to_string(),
115 ),
116 DisableError::Failed(failure) => {
117 RuntimeUpdateError::CommunicationFailure(failure.to_string())
118 }
119 })?;
120
121 Ok((protection, disable_lease))
122}
123
124async fn load_update_collections<S>(
125 storage: &S,
126) -> Result<UpdateCollections<S::CollectionHandle>, RuntimeUpdateError>
127where
128 S: Storage + Send + Sync + 'static,
129{
130 Ok(UpdateCollections {
131 pending_mdd: crate::operations::try_get_collection(
132 storage,
133 &CollectionName::DiagnosticDatabaseNextUpdate,
134 )
135 .await?,
136 current_mdd: crate::operations::try_get_collection(
137 storage,
138 &CollectionName::DiagnosticDatabase,
139 )
140 .await?,
141 })
142}
143
144async fn validate_execution_preconditions<S, R, T, L>(
145 params: &ExecutionParams<'_, S, R, T, L>,
146 mode: ExecutionMode,
147 collections: &UpdateCollections<S::CollectionHandle>,
148 protection: OwnedHttpProtection,
149 disable_lease: Box<dyn DisableGuard>,
150) -> Result<(OwnedHttpProtection, Box<dyn DisableGuard>), RuntimeUpdateError>
151where
152 S: Storage + Send + Sync + 'static,
153 R: RuntimeReloaderPlugin + ?Sized,
154 T: RuntimeUpdateSecurityPlugin<L, S::CollectionHandle>,
155 L: LockStateProvider,
156{
157 if let Err(error) = params
158 .security_handler
159 .check_apply_allowed(params.lock_state_provider, collections)
160 .await
161 {
162 return Err(reject_execution(error, protection, disable_lease).await);
163 }
164
165 if mode == ExecutionMode::Rollback {
166 match crate::storage::is_backup_empty(&**params.storage).await {
167 Ok(true) => {
168 return Err(reject_execution(
169 RuntimeUpdateError::NoBackup,
170 protection,
171 disable_lease,
172 )
173 .await);
174 }
175 Err(error) => return Err(reject_execution(error, protection, disable_lease).await),
176 Ok(false) => {}
177 }
178 }
179
180 // For Apply: verify there is at least one pending NextUpdate collection before
181 // accepting the request. This mirrors the Rollback check above, and must happen
182 // before spawning the task so that the 404 is returned synchronously instead of
183 // 202 being sent with a later Failed status (execute_apply's own check runs
184 // inside the spawned task and is too late to affect the HTTP response).
185 if mode == ExecutionMode::Apply && collections.pending_mdd.is_none() {
186 return Err(reject_execution(
187 RuntimeUpdateError::NoPendingUpdate,
188 protection,
189 disable_lease,
190 )
191 .await);
192 }
193
194 Ok((protection, disable_lease))
195}
196
197async fn reject_execution(
198 error: RuntimeUpdateError,
199 protection: OwnedHttpProtection,
200 disable_lease: Box<dyn DisableGuard>,
201) -> RuntimeUpdateError {
202 // Resume first, drop the update protection second, as `spawn_execution`'s
203 // completion path does. The other order leaves a window with neither the
204 // protection nor a restored transport, where requests get a handler-level
205 // communication error instead of the `409` the caller retries against.
206 if let Err(release_failure) = disable_lease.release().await {
207 tracing::error!(
208 error = %error,
209 release_failure = %release_failure,
210 "Failed to resume transport after rejecting runtime update"
211 );
212 }
213 drop(protection);
214 error
215}
216
217async fn register_execution(
218 executions: &Arc<RwLock<HashMap<String, UpdateExecution>>>,
219 mode: ExecutionMode,
220) -> String {
221 let execution_id = uuid::Uuid::new_v4().to_string();
222 let mut execs = executions.write().await;
223 execs.retain(|_, execution| execution.status == ExecutionStatus::Running);
224 execs.insert(
225 execution_id.clone(),
226 UpdateExecution {
227 id: execution_id.clone(),
228 mode,
229 status: ExecutionStatus::Running,
230 },
231 );
232 execution_id
233}
234
235#[allow(
236 clippy::too_many_arguments,
237 reason = "Spawned task inputs must be owned"
238)]
[docs]239/// [[ dimpl~post-update-deferred-communication, Post-update deferred communication handling, dimpl ]]
240fn spawn_execution<S, R>(
241 mode: ExecutionMode,
242 execution_id: String,
243 storage: Arc<S>,
244 reload_handler: Arc<R>,
245 executions: Arc<RwLock<HashMap<String, UpdateExecution>>>,
246 mdd_decompress: bool,
247 post_update_mode: PostUpdateCommunicationMode,
248 communication_access: Arc<dyn CommunicationAccess>,
249 disable_lease: Box<dyn DisableGuard>,
250 protection: OwnedHttpProtection,
251) where
252 S: Storage + Send + Sync + 'static,
253 R: RuntimeReloaderPlugin + ?Sized,
254{
255 let supervised_executions = Arc::clone(&executions);
256 let supervised_execution_id = execution_id.clone();
257
258 let handle = cda_interfaces::spawn_named!(&format!("runtime-update-{mode:?}"), async move {
259 let result = execute_operation(mode, &*storage, &*reload_handler, mdd_decompress).await;
260
261 if let Err(error) = &result {
262 tracing::error!(
263 execution_id = %execution_id,
264 mode = ?mode,
265 error = %error,
266 "Runtime update execution failed"
267 );
268 }
269
270 // Only a successful Apply/Rollback defers. A failed update or a Cleanup
271 // leaves communication as it found it.
272 let should_defer = result.is_ok()
273 && matches!(mode, ExecutionMode::Apply | ExecutionMode::Rollback)
274 && matches!(post_update_mode, PostUpdateCommunicationMode::Deferred);
275 if should_defer {
276 drop(disable_lease);
277 } else {
278 if let Err(failure) = disable_lease.release().await {
279 tracing::error!(
280 execution_id = %execution_id,
281 mode = ?mode,
282 error = %failure,
283 "Failed to resume transport after runtime update"
284 );
285 }
286 // Releasing only restores what the lease displaced, so an update
287 // started from a deferred runtime comes back deferred. Requesting
288 // the configured end state through `CommunicationAccess` leaves
289 // `init_mode` the final word, and is a no-op when the release
290 // already resumed.
291 communication_access.request_activate(ActivationCause::DisableRelease);
292 }
293 drop(protection);
294
295 // Published last. A client resumes normal traffic once it reads the
296 // terminal status, so publishing before the release above would let a
297 // caller that saw `completed` get the update's `409` on its next
298 // request.
299 let mut map = executions.write().await;
300 if let Some(execution) = map.get_mut(&execution_id) {
301 execution.status = match result {
302 Ok(()) => ExecutionStatus::Completed,
303 Err(ref error) => ExecutionStatus::Failed(error.to_string()),
304 };
305 } else {
306 tracing::error!(
307 execution_id = %execution_id,
308 mode = ?mode,
309 "Runtime update execution completed without an execution record"
310 );
311 }
312 drop(map);
313 });
314
315 // A panic in `execute_operation` unwinds past the status write and leaves
316 // the execution record `Running` forever. This supervisor awaits the
317 // `JoinHandle` and marks the execution `Failed` when the task ended without
318 // a terminal status.
319 cda_interfaces::spawn_named!(&format!("runtime-update-{mode:?}-supervisor"), async move {
320 if let Err(join_error) = handle.await {
321 tracing::error!(
322 execution_id = %supervised_execution_id,
323 mode = ?mode,
324 error = %join_error,
325 "runtime update task ended without completing normally"
326 );
327 let mut map = supervised_executions.write().await;
328 if let Some(execution) = map.get_mut(&supervised_execution_id)
329 && execution.status == ExecutionStatus::Running
330 {
331 execution.status = ExecutionStatus::Failed(format!(
332 "execution task ended abnormally: {join_error}"
333 ));
334 }
335 }
336 });
337}
338
339pub(crate) async fn get_execution_status(
340 executions: &Arc<RwLock<HashMap<String, UpdateExecution>>>,
341 execution_id: &str,
342) -> Option<UpdateExecution> {
343 let execs = executions.read().await;
344 execs.get(execution_id).cloned()
345}
346
347async fn execute_operation<S, R>(
348 mode: ExecutionMode,
349 storage: &S,
350 reload_handler: &R,
351 mdd_decompress: bool,
352) -> Result<(), RuntimeUpdateError>
353where
354 S: Storage + Send + Sync + 'static,
355 R: RuntimeReloaderPlugin + ?Sized,
356{
357 match mode {
358 ExecutionMode::Apply => {
359 crate::operations::apply::execute_apply(storage, reload_handler, mdd_decompress).await
360 }
361 ExecutionMode::Rollback => {
362 crate::operations::rollback::execute_rollback(storage, reload_handler).await
363 }
364 ExecutionMode::Cleanup => crate::operations::cleanup::execute_cleanup(storage).await,
365 }
366}
367
368#[cfg(test)]
369mod tests {
370 use std::{sync::Arc, time::Duration};
371
372 use cda_interfaces::{
373 HashMap,
374 communication_control::{
375 CommunicationAccess, PostUpdateCommunicationMode, TransportControl, TransportState,
376 error::CommControlError,
377 },
378 http_protection::registry::{HttpProtectionRegistry, HttpRestrictionGuard},
379 runtime_update_api::{ExecutionMode, ExecutionStatus, RuntimeUpdateError, UpdateExecution},
380 storage_api::CollectionName,
381 };
382 use cda_plugin_communication_management::lifecycle::{
383 communication_disable_for_test, disable::DisableCommunication,
384 enabled_communication_access_for_test,
385 };
386 use cda_storage::LocalStorage;
387 use tokio::sync::RwLock;
388
389 use crate::test_utils::{
390 MockLockProvider, MockSecurityHandler, NoopReloadHandler, StubTransport, make_storage,
391 write_test_file,
392 };
393
394 /// A transport whose resume (`enable`) parks until the test lets it through,
395 /// making the window between "the guards are coming down" and "they are
396 /// down" long enough to observe.
397 struct GatedTransport {
398 entered: tokio::sync::mpsc::UnboundedSender<()>,
399 gate: Arc<tokio::sync::Semaphore>,
400 transport: Arc<StubTransport>,
401 }
402
403 /// Test-side handle to [`GatedTransport`]'s resume.
404 struct ResumeGate {
405 entered: tokio::sync::mpsc::UnboundedReceiver<()>,
406 gate: Arc<tokio::sync::Semaphore>,
407 }
408
409 impl GatedTransport {
410 fn new() -> (Arc<Self>, ResumeGate) {
411 let (entered_tx, entered_rx) = tokio::sync::mpsc::unbounded_channel();
412 let gate = Arc::new(tokio::sync::Semaphore::new(0));
413 let transport = Arc::new(Self {
414 entered: entered_tx,
415 gate: Arc::clone(&gate),
416 transport: StubTransport::with_state(TransportState::Enabled),
417 });
418 let resume = ResumeGate {
419 entered: entered_rx,
420 gate,
421 };
422 (transport, resume)
423 }
424 }
425
426 impl ResumeGate {
427 /// Resolves once the resume has started and is parked on the gate.
428 async fn entered(&mut self) {
429 self.entered
430 .recv()
431 .await
432 .expect("transport resume must be reached");
433 }
434
435 /// Lets the parked resume run to completion.
436 fn release(&self) {
437 self.gate.add_permits(1);
438 }
439 }
440
441 #[async_trait::async_trait]
442 impl TransportControl for GatedTransport {
443 async fn enable(&self) -> Result<(), CommControlError> {
444 let _ = self.entered.send(());
445 let _permit = self
446 .gate
447 .acquire()
448 .await
449 .expect("gate must not be closed while a resume is parked on it");
450 self.transport.enable().await
451 }
452
453 async fn disable(&self) -> Result<(), CommControlError> {
454 self.transport.disable().await
455 }
456
457 async fn state(&self) -> TransportState {
458 self.transport.state().await
459 }
460 }
461
462 struct TestFixture {
463 storage: Arc<LocalStorage>,
464 security_handler: Arc<MockSecurityHandler>,
465 reload_handler: Arc<NoopReloadHandler>,
466 lock_provider: MockLockProvider,
467 executions: Arc<RwLock<HashMap<String, UpdateExecution>>>,
468 communication_disable: Arc<dyn DisableCommunication>,
469 communication_access: Arc<dyn CommunicationAccess>,
470 transport: Arc<StubTransport>,
471 http_restriction_manager: HttpProtectionRegistry,
472 _dir: tempfile::TempDir,
473 }
474
475 impl TestFixture {
476 fn params(
477 &self,
478 ) -> super::ExecutionParams<
479 '_,
480 LocalStorage,
481 NoopReloadHandler,
482 MockSecurityHandler,
483 MockLockProvider,
484 > {
485 super::ExecutionParams {
486 communication_access: &self.communication_access,
487 storage: &self.storage,
488 security_handler: &self.security_handler,
489 reload_handler: &self.reload_handler,
490 executions: &self.executions,
491 communication_disable: &self.communication_disable,
492 http_protections: &self.http_restriction_manager,
493 update_retry_after: Duration::from_secs(1),
494 post_update_mode: PostUpdateCommunicationMode::Enabled,
495 mdd_decompress: false,
496 lock_state_provider: &self.lock_provider,
497 }
498 }
499
500 fn params_with_post_update_mode(
501 &self,
502 post_update_mode: PostUpdateCommunicationMode,
503 ) -> super::ExecutionParams<
504 '_,
505 LocalStorage,
506 NoopReloadHandler,
507 MockSecurityHandler,
508 MockLockProvider,
509 > {
510 super::ExecutionParams {
511 post_update_mode,
512 ..self.params()
513 }
514 }
515 }
516
517 fn make_fixture() -> TestFixture {
518 let (storage, dir) = make_storage();
519 let transport = StubTransport::new();
520 let mgr = HttpProtectionRegistry::new();
521 let communication_disable =
522 communication_disable_for_test(Arc::<StubTransport>::clone(&transport), true);
523 TestFixture {
524 storage: Arc::new(storage),
525 security_handler: Arc::new(MockSecurityHandler::new()),
526 reload_handler: Arc::new(NoopReloadHandler),
527 lock_provider: MockLockProvider {
528 owner: Some("test-user".to_owned()),
529 has_conflicts: false,
530 },
531 executions: Arc::new(RwLock::new(HashMap::default())),
532 communication_disable,
533 communication_access: enabled_communication_access_for_test(),
534 transport,
535 http_restriction_manager: mgr,
536 _dir: dir,
537 }
538 }
539
540 async fn poll_until_terminal(
541 executions: &Arc<RwLock<HashMap<String, UpdateExecution>>>,
542 exec_id: &str,
543 ) -> ExecutionStatus {
544 let deadline = tokio::time::Instant::now()
545 .checked_add(tokio::time::Duration::from_secs(5))
546 .unwrap();
547 loop {
548 tokio::task::yield_now().await;
549 if let Some(exec) = super::get_execution_status(executions, exec_id).await
550 && exec.status != ExecutionStatus::Running
551 {
552 return exec.status;
553 }
554 assert!(
555 tokio::time::Instant::now() < deadline,
556 "Execution did not complete within 5 seconds"
557 );
558 }
559 }
560
561 #[tokio::test]
562 async fn start_execution_apply_returns_execution_id() {
563 let f = make_fixture();
564 write_test_file(
565 &f.storage,
566 &CollectionName::DiagnosticDatabaseNextUpdate,
567 "ecu.mdd",
568 b"mdd_data",
569 )
570 .await;
571
572 let exec_id = super::start_execution(&f.params(), ExecutionMode::Apply)
573 .await
574 .unwrap();
575 assert!(!exec_id.is_empty());
576
577 let status = super::get_execution_status(&f.executions, &exec_id).await;
578 assert!(status.is_some());
579 }
580
[docs]581 /// [[ test~deferred-post-update-apply, Successful apply leaves communication disabled in deferred post-update mode, test ]]
582 #[tokio::test]
583 async fn successful_apply_leaves_communication_disabled_in_deferred_mode() {
584 let f = make_fixture();
585 write_test_file(
586 &f.storage,
587 &CollectionName::DiagnosticDatabaseNextUpdate,
588 "ecu.mdd",
589 b"mdd_data",
590 )
591 .await;
592
593 let params = f.params_with_post_update_mode(PostUpdateCommunicationMode::Deferred);
594 let exec_id = super::start_execution(¶ms, ExecutionMode::Apply)
595 .await
596 .unwrap();
597 assert_eq!(
598 poll_until_terminal(&f.executions, &exec_id).await,
599 ExecutionStatus::Completed
600 );
601 assert_eq!(
602 TransportControl::state(&*f.transport).await,
603 TransportState::Disabled
604 );
605 assert!(!f.http_restriction_manager.is_active());
606 }
607
[docs]608 /// [[ test~deferred-post-update-rollback, Successful rollback leaves communication disabled in deferred post-update mode, test ]]
609 #[tokio::test]
610 async fn successful_rollback_leaves_communication_disabled_in_deferred_mode() {
611 let f = make_fixture();
612 write_test_file(
613 &f.storage,
614 &CollectionName::DiagnosticDatabaseBackup,
615 "ecu.mdd",
616 b"backup_data",
617 )
618 .await;
619
620 let params = f.params_with_post_update_mode(PostUpdateCommunicationMode::Deferred);
621 let exec_id = super::start_execution(¶ms, ExecutionMode::Rollback)
622 .await
623 .unwrap();
624 assert_eq!(
625 poll_until_terminal(&f.executions, &exec_id).await,
626 ExecutionStatus::Completed
627 );
628 assert_eq!(
629 TransportControl::state(&*f.transport).await,
630 TransportState::Disabled
631 );
632 assert!(!f.http_restriction_manager.is_active());
633 }
634
635 #[tokio::test]
636 async fn start_execution_rollback() {
637 let f = make_fixture();
638 write_test_file(
639 &f.storage,
640 &CollectionName::DiagnosticDatabaseBackup,
641 "ecu.mdd",
642 b"backup_data",
643 )
644 .await;
645
646 let exec_id = super::start_execution(&f.params(), ExecutionMode::Rollback)
647 .await
648 .unwrap();
649 assert!(!exec_id.is_empty());
650 }
651
652 #[tokio::test]
653 async fn start_execution_apply_with_no_pending_update_rejected_synchronously() {
654 let f = make_fixture();
655
656 // Nothing seeded into DiagnosticDatabaseNextUpdate:
657 // Apply must be rejected synchronously (i.e. `start_execution` itself returns
658 // an error) rather than accepted (202-equivalent execution id) only to fail
659 // later inside the spawned task.
660 let result = super::start_execution(&f.params(), ExecutionMode::Apply).await;
661
662 assert!(
663 matches!(result, Err(RuntimeUpdateError::NoPendingUpdate)),
664 "expected NoPendingUpdate, got: {result:?}"
665 );
666 assert!(
667 f.executions.read().await.is_empty(),
668 "no execution should have been recorded for a synchronously-rejected apply"
669 );
670 // Transport should be re-enabled (restriction removed) after the error path.
671 assert!(
672 !f.http_restriction_manager.is_active(),
673 "update restriction must be released after a synchronously-rejected apply"
674 );
675 }
676
677 #[tokio::test]
678 async fn start_execution_cleanup_succeeds() {
679 let f = make_fixture();
680 let exec_id = super::start_execution(&f.params(), ExecutionMode::Cleanup)
681 .await
682 .unwrap();
683 assert!(!exec_id.is_empty());
684
685 let status = poll_until_terminal(&f.executions, &exec_id).await;
686 assert_eq!(status, ExecutionStatus::Completed);
687 }
688
689 #[tokio::test]
690 async fn terminal_status_is_only_published_after_the_update_guards_are_released() {
691 // A client resumes ordinary requests as soon as it reads a terminal
692 // status, so that status must not appear while the update's `409`
693 // protection is still up. `GatedTransport` holds the resume open to make
694 // that window observable. With an instant transport the tail of the
695 // execution task runs in a single poll.
696 let (transport, mut resume) = GatedTransport::new();
697 let mut f = make_fixture();
698 f.communication_disable = communication_disable_for_test(transport, true);
699
700 let exec_id = super::start_execution(&f.params(), ExecutionMode::Cleanup)
701 .await
702 .unwrap();
703
704 resume.entered().await;
705 assert_eq!(
706 super::get_execution_status(&f.executions, &exec_id)
707 .await
708 .map(|execution| execution.status),
709 Some(ExecutionStatus::Running),
710 "the execution must still read as running while its guards are being released"
711 );
712 assert!(
713 f.http_restriction_manager.is_active(),
714 "sanity check: the update restriction is still up at this point"
715 );
716
717 resume.release();
718 let status = poll_until_terminal(&f.executions, &exec_id).await;
719 assert_eq!(status, ExecutionStatus::Completed);
720 assert!(
721 !f.http_restriction_manager.is_active(),
722 "update restriction must already be released once the execution reports a terminal \
723 status"
724 );
725 }
726
727 #[tokio::test]
728 async fn get_execution_status_unknown_id_returns_none() {
729 let f = make_fixture();
730 let result = super::get_execution_status(&f.executions, "nonexistent-id").await;
731 assert!(result.is_none());
732 }
733
734 #[tokio::test]
735 async fn start_execution_allowed_when_previous_completed() {
736 let f = make_fixture();
737 {
738 let mut execs = f.executions.write().await;
739 execs.insert(
740 "prev".to_string(),
741 UpdateExecution {
742 id: "prev".to_string(),
743 mode: ExecutionMode::Cleanup,
744 status: ExecutionStatus::Completed,
745 },
746 );
747 }
748
749 let exec_id = super::start_execution(&f.params(), ExecutionMode::Cleanup)
750 .await
751 .unwrap();
752 assert!(!exec_id.is_empty());
753 }
754
755 #[tokio::test]
756 async fn previous_execution_removed_when_new_one_starts() {
757 let f = make_fixture();
758 let first_id = super::start_execution(&f.params(), ExecutionMode::Cleanup)
759 .await
760 .unwrap();
761
762 poll_until_terminal(&f.executions, &first_id).await;
763
764 let _second_id = super::start_execution(&f.params(), ExecutionMode::Cleanup)
765 .await
766 .unwrap();
767
768 let status = super::get_execution_status(&f.executions, &first_id).await;
769 assert!(
770 status.is_none(),
771 "expected first execution to be removed when second execution started"
772 );
773 }
774
775 #[tokio::test]
776 async fn execution_transitions_to_failed_on_error() {
777 let f = make_fixture();
778 write_test_file(
779 &f.storage,
780 &CollectionName::DiagnosticDatabaseNextUpdate,
781 "ecu.mdd",
782 b"mdd_data",
783 )
784 .await;
785 let failing_reload_handler = Arc::new(crate::test_utils::FailingReloadHandler);
786 let communication_access = enabled_communication_access_for_test();
787 let params = super::ExecutionParams {
788 communication_access: &communication_access,
789 storage: &f.storage,
790 security_handler: &f.security_handler,
791 reload_handler: &failing_reload_handler,
792 executions: &f.executions,
793 communication_disable: &f.communication_disable,
794 http_protections: &f.http_restriction_manager,
795 update_retry_after: Duration::from_secs(1),
796 post_update_mode: PostUpdateCommunicationMode::Enabled,
797 mdd_decompress: false,
798 lock_state_provider: &f.lock_provider,
799 };
800
801 let exec_id = super::start_execution(¶ms, ExecutionMode::Apply)
802 .await
803 .unwrap();
804
805 let status = poll_until_terminal(&f.executions, &exec_id).await;
806 assert!(matches!(status, ExecutionStatus::Failed(_)));
807 // After failure the spawned task re-enables, removing the restriction.
808 assert!(!f.http_restriction_manager.is_active());
809 }
810
811 #[tokio::test]
812 async fn execution_transitions_to_failed_when_task_panics() {
813 let f = make_fixture();
814 write_test_file(
815 &f.storage,
816 &CollectionName::DiagnosticDatabaseNextUpdate,
817 "ecu.mdd",
818 b"mdd_data",
819 )
820 .await;
821 let panicking_reload_handler = Arc::new(crate::test_utils::PanickingReloadHandler);
822 let communication_access = enabled_communication_access_for_test();
823 let params = super::ExecutionParams {
824 communication_access: &communication_access,
825 storage: &f.storage,
826 security_handler: &f.security_handler,
827 reload_handler: &panicking_reload_handler,
828 executions: &f.executions,
829 communication_disable: &f.communication_disable,
830 http_protections: &f.http_restriction_manager,
831 update_retry_after: Duration::from_secs(1),
832 post_update_mode: PostUpdateCommunicationMode::Enabled,
833 mdd_decompress: false,
834 lock_state_provider: &f.lock_provider,
835 };
836
837 let exec_id = super::start_execution(¶ms, ExecutionMode::Apply)
838 .await
839 .unwrap();
840
841 // Without the supervisor task the execution task's panic would leave no
842 // terminal status, and this would hang until poll_until_terminal's
843 // deadline.
844 let status = poll_until_terminal(&f.executions, &exec_id).await;
845 assert!(
846 matches!(status, ExecutionStatus::Failed(_)),
847 "expected Failed after the execution task panicked, got: {status:?}"
848 );
849 }
850}
Back to top