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    pub async fn close(&self) -> Result<()> {
39        {
40            let mut state = self.inner.state.lock().unwrap();
41            state.status.closing = true;
42            state.close_followers();
43        }
44        self.inner.notify();
45
46        let mut lifecycle = self.inner.lifecycle.write().await;
47        let ready = {
48            let state = self.inner.state.lock().unwrap();
49            state.ready && !state.status.closed
50        };
51        if ready {
52            self.settle_transition().await;
53            self.settle_files().await?;
54            self.settle_deletions().await?;
55            let files = self.capture_files(None)?;
56            self.flush_domain(Domain::Data, None).await?;
57            self.persist_files(&files).await?;
58            self.retire_files();
59            self.flush_recovery().await?;
60        }
61        if ready {
62            self.release_snapshots().await?;
63        }
64        // Closing prevents new maintenance; let an in-flight backend call settle
65        // before releasing its physical owner, including uncertain failed calls.
66        let mut changed = self.inner.changed.subscribe();
67        loop {
68            {
69                let state = self.inner.state.lock().unwrap();
70                if !state.pruning_objects {
71                    break;
72                }
73                if !state.status.running {
74                    return Err(Error::NotReady { domain: "storage worker" });
75                }
76            }
77            changed.changed().await.map_err(|_| Error::NotReady { domain: "storage" })?;
78        }
79        if lifecycle.primary_touched && !lifecycle.primary_closed {
80            self.inner.primary.close().await?;
81            lifecycle.primary_closed = true;
82        }
83        if lifecycle.recovery_touched && !lifecycle.recovery_closed {
84            self.inner.recovery.close().await?;
85            lifecycle.recovery_closed = true;
86        }
87        // Release Recovered Files Even When Startup Failed Before Readiness.
88        self.retire_files();
89        {
90            let mut state = self.inner.state.lock().unwrap();
91            state.receipts.clear();
92            state.status.closed = true;
93            state.collect();
94        }
95        self.inner.notify();
96        if let Some(owner) = &lifecycle.owner {
97            owner.stop().await?;
98        }
99        lifecycle.owner = None;
100        // A supervisor stopped before its first poll never enters the worker's
101        // running guard. The completed join is authoritative in either case.
102        self.inner.state.lock().unwrap().status.running = false;
103        self.inner.notify();
104        Ok(())
105    }
106
107    pub(super) fn check_raw_recovery(&self) -> Result<()> {
108        if self.inner.state.lock().unwrap().journal_saved.is_some() {
109            return Err(Error::InvalidInput { field: "recovery ownership", reason: "managed journal owns recovery writes" });
110        }
111        Ok(())
112    }
113}
114
115/* === Public Functions === */
116
117/* === Private Functions === */
118
119/* === Tests === */
harmos_storage/manager/shutdown.rs

harmos_storage/manager/
shutdown.rs
