Coverage Report

Created: 2026-10-26 14:29

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/build/source/nativelink-store/src/ref_store.rs
Line
Count
Source
1
// Copyright 2024 The NativeLink Authors. All rights reserved.
2
//
3
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (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
//    See LICENSE file for details
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
15
use core::cell::UnsafeCell;
16
use core::pin::Pin;
17
use std::sync::{Arc, Weak};
18
19
use async_trait::async_trait;
20
use nativelink_config::stores::RefSpec;
21
use nativelink_error::{Error, ResultExt, make_input_err};
22
use nativelink_metric::MetricsComponent;
23
use nativelink_util::buf_channel::{DropCloserReadHalf, DropCloserWriteHalf};
24
use nativelink_util::health_utils::{HealthStatusIndicator, default_health_status_indicator};
25
use nativelink_util::store_trait::{
26
    RemoveCallback, Store, StoreDriver, StoreKey, StoreLike, UploadSizeInfo,
27
};
28
use parking_lot::Mutex;
29
use tracing::{debug, error};
30
31
use crate::store_manager::StoreManager;
32
33
#[repr(C, align(8))]
34
#[derive(Debug)]
35
struct AlignedStoreCell(UnsafeCell<Option<Store>>);
36
37
#[derive(Debug)]
38
struct StoreReference {
39
    cell: AlignedStoreCell,
40
    mux: Mutex<()>,
41
}
42
43
unsafe impl Sync for StoreReference {}
44
45
#[derive(Debug, MetricsComponent)]
46
pub struct RefStore {
47
    #[metric(help = "The store we are referencing")]
48
    name: String,
49
    store_manager: Weak<StoreManager>,
50
    inner: StoreReference,
51
    remove_callbacks: Mutex<Vec<RemoveCallback>>,
52
}
53
54
impl RefStore {
55
37
    pub fn new(spec: &RefSpec, store_manager: Weak<StoreManager>) -> Arc<Self> {
56
37
        Arc::new(Self {
57
37
            name: spec.name.clone(),
58
37
            store_manager,
59
37
            inner: StoreReference {
60
37
                mux: Mutex::new(()),
61
37
                cell: AlignedStoreCell(UnsafeCell::new(None)),
62
37
            },
63
37
            remove_callbacks: Mutex::new(vec![]),
64
37
        })
65
37
    }
66
67
    #[inline]
68
218
    fn get_store(&self) -> Result<&Store, Error> {
69
218
        let ref_store = self.inner.cell.0.get();
70
        unsafe {
71
218
            if let Some(
ref store202
) = *ref_store {
72
202
                return Ok(store);
73
16
            }
74
        }
75
16
        Err(make_input_err!(
76
16
            "ref_store cannot get store '{}', was post_init called?",
77
16
            self.name
78
16
        ))
79
218
    }
80
}
81
82
#[async_trait]
83
impl StoreDriver for RefStore {
84
35
    async fn post_init(self: Arc<Self>) -> Result<(), Error> {
85
        debug!("Running post_init to get store");
86
        let ref_store = self.inner.cell.0.get();
87
        unsafe {
88
            if (*ref_store).is_some() {
89
                return Err(make_input_err!("post_init already called for RefStore"));
90
            }
91
        }
92
        let _lock = self.inner.mux.lock();
93
        let store_manager = self
94
            .store_manager
95
            .upgrade()
96
            .err_tip(|| "Store manager is gone")?;
97
        if let Some(store) = store_manager.get_store(&self.name) {
98
            let remove_callbacks = self.remove_callbacks.lock().clone();
99
            for callback in remove_callbacks {
100
                debug!(?callback, ?store, "Added callback to store");
101
                store.register_remove_callback(callback)?;
102
            }
103
            unsafe {
104
                *ref_store = Some(store);
105
            }
106
            Ok(())
107
        } else {
108
            Err(make_input_err!(
109
                "Failed to find store '{}' in StoreManager in RefStore",
110
                self.name
111
            ))
112
        }
113
35
    }
114
115
19
    async fn remove(self: Pin<&Self>, key: StoreKey<'_>) -> Result<bool, Error> {
116
        self.get_store()?.as_store_driver_pin().remove(key).await
117
19
    }
118
119
11
    async fn remove_cached(self: Pin<&Self>, key: StoreKey<'_>) -> Result<bool, Error> {
120
        self.get_store()?
121
            .as_store_driver_pin()
122
            .remove_cached(key)
123
            .await
124
11
    }
125
126
    async fn has_with_results(
127
        self: Pin<&Self>,
128
        keys: &[StoreKey<'_>],
129
        results: &mut [Option<u64>],
130
76
    ) -> Result<(), Error> {
131
        self.get_store()?.has_with_results(keys, results).await
132
76
    }
133
134
    async fn update(
135
        self: Pin<&Self>,
136
        key: StoreKey<'_>,
137
        reader: DropCloserReadHalf,
138
        size_info: UploadSizeInfo,
139
19
    ) -> Result<u64, Error> {
140
        self.get_store()?.update(key, reader, size_info).await
141
19
    }
142
143
    async fn get_part(
144
        self: Pin<&Self>,
145
        key: StoreKey<'_>,
146
        writer: &mut DropCloserWriteHalf,
147
        offset: u64,
148
        length: Option<u64>,
149
46
    ) -> Result<(), Error> {
150
        self.get_store()?
151
            .get_part(key, writer, offset, length)
152
            .await
153
46
    }
154
155
47
    fn inner_store(&self, key: Option<StoreKey>) -> &'_ dyn StoreDriver {
156
47
        match self.get_store() {
157
34
            Ok(store) => store.inner_store(key),
158
13
            Err(err) => {
159
13
                error!(?key, ?err, "Failed to get store for key");
160
13
                self
161
            }
162
        }
163
47
    }
164
165
0
    fn as_any<'a>(&'a self) -> &'a (dyn core::any::Any + Sync + Send + 'static) {
166
0
        self
167
0
    }
168
169
0
    fn as_any_arc(self: Arc<Self>) -> Arc<dyn core::any::Any + Sync + Send + 'static> {
170
0
        self
171
0
    }
172
173
0
    fn register_remove_callback(self: Arc<Self>, callback: RemoveCallback) -> Result<(), Error> {
174
0
        self.remove_callbacks.lock().push(callback.clone());
175
0
        let ref_store = self.inner.cell.0.get();
176
        unsafe {
177
0
            if let Some(ref store) = *ref_store {
178
0
                debug!(?callback, ?store, "New callback");
179
0
                store.register_remove_callback(callback)?;
180
            } else {
181
0
                debug!(?callback, "New callback, no store yet");
182
            }
183
        }
184
0
        Ok(())
185
0
    }
186
}
187
188
default_health_status_indicator!(RefStore);