harmos_storage/manager/
shutdown.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17
18// Third Party Imports
19
20// Local Imports
21use super::*;
22
23/* === Constants === */
24
25/* === Macros === */
26
27/* === Type Aliases === */
28
29/* === Structs === */
30
31/* === Enums === */
32
33/* === Traits === */
34
35/* === Implementations === */
36
37impl<C> StorageManager<C> {
38    #[tracing::instrument(name = "storage.manager.shutdown.close", level = "debug", skip_all, fields(outcome = "cancelled"))]
39    pub async fn close(&self) -> Result<()> {
40        let result = async {
41            {
42                let mut state = self.inner.state.lock().unwrap();
43                state.status.closing = true;
44                state.close_followers();
45            }
46            self.inner.notify();
47
48            let mut lifecycle =
49                tracing::Instrument::instrument(self.inner.lifecycle.write(), tracing::trace_span!("storage.lock.wait")).await;
50            let ready = {
51                let state = self.inner.state.lock().unwrap();
52                state.ready && !state.status.closed
53            };
54            if ready {
55                self.settle_transition().await;
56                self.settle_files().await?;
57                self.settle_deletions().await?;
58                let files = self.capture_files(None)?;
59                self.flush_domain(Domain::Data, None).await?;
60                self.persist_files(&files).await?;
61                self.retire_files();
62                self.flush_recovery().await?;
63            }
64            if ready {
65                self.release_snapshots().await?;
66            }
67            // Closing prevents new maintenance; let an in-flight backend call settle
68            // before releasing its physical owner, including uncertain failed calls.
69            let mut changed = self.inner.changed.subscribe();
70            loop {
71                {
72                    let state = self.inner.state.lock().unwrap();
73                    if !state.pruning_objects {
74                        break;
75                    }
76                    if !state.status.running {
77                        return Err(Error::NotReady { domain: "storage worker" });
78                    }
79                }
80                changed.changed().await.map_err(|_| Error::NotReady { domain: "storage" })?;
81            }
82            if lifecycle.primary_touched && !lifecycle.primary_closed {
83                tracing::Instrument::instrument(
84                    self.inner.primary.close(),
85                    tracing::debug_span!("storage.backend", method = "primary.close"),
86                )
87                .await?;
88                lifecycle.primary_closed = true;
89            }
90            if lifecycle.recovery_touched && !lifecycle.recovery_closed {
91                tracing::Instrument::instrument(
92                    self.inner.recovery.close(),
93                    tracing::debug_span!("storage.backend", method = "recovery.close"),
94                )
95                .await?;
96                lifecycle.recovery_closed = true;
97            }
98            // Release Recovered Files Even When Startup Failed Before Readiness.
99            self.retire_files();
100            {
101                let mut state = self.inner.state.lock().unwrap();
102                state.receipts.clear();
103                state.status.closed = true;
104                state.collect();
105            }
106            self.inner.notify();
107            if let Some(owner) = &lifecycle.owner {
108                owner.stop().await?;
109            }
110            lifecycle.owner = None;
111            // A supervisor stopped before its first poll never enters the worker's
112            // running guard. The completed join is authoritative in either case.
113            self.inner.state.lock().unwrap().status.running = false;
114            self.inner.notify();
115            Ok(())
116        }
117        .await;
118        tracing::Span::current().record("outcome", if result.is_ok() { "completed" } else { "failed" });
119        result
120    }
121
122    pub(super) fn check_raw_recovery(&self) -> Result<()> {
123        if self.inner.state.lock().unwrap().journal_saved.is_some() {
124            return Err(Error::InvalidInput { field: "recovery ownership", reason: "managed journal owns recovery writes" });
125        }
126        Ok(())
127    }
128}
129
130/* === Public Functions === */
131
132/* === Private Functions === */
133
134/* === Tests === */
harmos_storage/manager/shutdown.rs

harmos_storage/manager/
shutdown.rs
