Coverage Report

Created: 2025-05-30 16:37

next uncovered line (L), next uncovered region (R), next uncovered branch (B)
/build/source/nativelink-store/src/filesystem_store.rs
Line
Count
Source
1
// Copyright 2024 The NativeLink Authors. All rights reserved.
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
15
use core::fmt::{Debug, Formatter};
16
use core::pin::Pin;
17
use core::sync::atomic::{AtomicU64, Ordering};
18
use std::borrow::Cow;
19
use std::ffi::{OsStr, OsString};
20
use std::sync::{Arc, Weak};
21
use std::time::SystemTime;
22
23
use async_lock::RwLock;
24
use async_trait::async_trait;
25
use bytes::BytesMut;
26
use futures::stream::{StreamExt, TryStreamExt};
27
use futures::{Future, TryFutureExt};
28
use nativelink_config::stores::FilesystemSpec;
29
use nativelink_error::{Code, Error, ResultExt, make_err, make_input_err};
30
use nativelink_metric::MetricsComponent;
31
use nativelink_util::background_spawn;
32
use nativelink_util::buf_channel::{
33
    DropCloserReadHalf, DropCloserWriteHalf, make_buf_channel_pair,
34
};
35
use nativelink_util::common::{DigestInfo, fs};
36
use nativelink_util::evicting_map::{EvictingMap, LenEntry};
37
use nativelink_util::health_utils::{HealthRegistryBuilder, HealthStatus, HealthStatusIndicator};
38
use nativelink_util::store_trait::{
39
    StoreDriver, StoreKey, StoreKeyBorrow, StoreOptimizations, UploadSizeInfo,
40
};
41
use tokio::io::{AsyncReadExt, AsyncWriteExt, Take};
42
use tokio_stream::wrappers::ReadDirStream;
43
use tracing::{debug, error, warn};
44
45
use crate::cas_utils::is_zero_digest;
46
47
// Default size to allocate memory of the buffer when reading files.
48
const DEFAULT_BUFF_SIZE: usize = 32 * 1024;
49
// Default block size of all major filesystems is 4KB
50
const DEFAULT_BLOCK_SIZE: u64 = 4 * 1024;
51
52
pub const STR_FOLDER: &str = "s";
53
pub const DIGEST_FOLDER: &str = "d";
54
55
#[derive(Clone, Copy, Debug)]
56
pub enum FileType {
57
    Digest,
58
    String,
59
}
60
61
#[derive(Debug, MetricsComponent)]
62
pub struct SharedContext {
63
    // Used in testing to know how many active drop() spawns are running.
64
    // TODO(aaronmondal) It is probably a good idea to use a spin lock during
65
    // destruction of the store to ensure that all files are actually
66
    // deleted (similar to how it is done in tests).
67
    #[metric(help = "Number of active drop spawns")]
68
    pub active_drop_spawns: AtomicU64,
69
    #[metric(help = "Path to the configured temp path")]
70
    temp_path: String,
71
    #[metric(help = "Path to the configured content path")]
72
    content_path: String,
73
}
74
75
#[derive(Eq, PartialEq, Debug)]
76
enum PathType {
77
    Content,
78
    Temp,
79
    Custom(OsString),
80
}
81
82
/// [`EncodedFilePath`] stores the path to the file
83
/// including the context, path type and key to the file.
84
/// The whole [`StoreKey`] is stored as opposed to solely
85
/// the [`DigestInfo`] so that it is more usable for things
86
/// such as BEP -see Issue #1108
87
#[derive(Debug)]
88
pub struct EncodedFilePath {
89
    shared_context: Arc<SharedContext>,
90
    path_type: PathType,
91
    key: StoreKey<'static>,
92
}
93
94
impl EncodedFilePath {
95
    #[inline]
96
330
    fn get_file_path(&self) -> Cow<'_, OsStr> {
97
330
        get_file_path_raw(&self.path_type, self.shared_context.as_ref(), &self.key)
98
330
    }
99
}
100
101
#[inline]
102
447
fn get_file_path_raw<'a>(
103
447
    path_type: &'a PathType,
104
447
    shared_context: &SharedContext,
105
447
    key: &StoreKey<'a>,
106
447
) -> Cow<'a, OsStr> {
107
447
    let 
folder438
= match path_type {
108
195
        PathType::Content => &shared_context.content_path,
109
243
        PathType::Temp => &shared_context.temp_path,
110
9
        PathType::Custom(path) => return Cow::Borrowed(path),
111
    };
112
438
    Cow::Owned(to_full_path_from_key(folder, key))
113
447
}
114
115
impl Drop for EncodedFilePath {
116
118
    fn drop(&mut self) {
117
        // `drop()` can be called during shutdown, so we use `path_type` flag to know if the
118
        // file actually needs to be deleted.
119
118
        if self.path_type == PathType::Content {
  Branch (119:12): [True: 92, False: 26]
  Branch (119:12): [Folded - Ignored]
120
92
            return;
121
26
        }
122
123
26
        let file_path = self.get_file_path().to_os_string();
124
26
        let shared_context = self.shared_context.clone();
125
26
        shared_context
126
26
            .active_drop_spawns
127
26
            .fetch_add(1, Ordering::Relaxed);
128
26
        background_spawn!("filesystem_delete_file", async move 
{24
129
24
            debug!(?file_path, 
"File deleted"0
,);
130
24
            let 
result23
= fs::remove_file(&file_path)
131
24
                .await
132
23
                .err_tip(|| format!(
"Failed to remove file {}"0
,
file_path.display()0
));
133
23
            if let Err(
err0
) = result {
  Branch (133:20): [True: 0, False: 23]
  Branch (133:20): [Folded - Ignored]
134
0
                error!(?file_path, ?err, "Failed to delete file",);
135
23
            }
136
23
            shared_context
137
23
                .active_drop_spawns
138
23
                .fetch_sub(1, Ordering::Relaxed);
139
23
        });
140
118
    }
141
}
142
143
/// This creates the file path from the [`StoreKey`]. If
144
/// it is a string, the string, prefixed with [`STR_PREFIX`]
145
/// for backwards compatibility, is stored.
146
///
147
/// If it is a [`DigestInfo`], it is prefixed by [`DIGEST_PREFIX`]
148
/// followed by the string representation of a digest - the hash in hex,
149
/// a hyphen then the size in bytes
150
///
151
/// Previously, only the string representation of the [`DigestInfo`] was
152
/// used with no prefix
153
#[inline]
154
465
fn to_full_path_from_key(folder: &str, key: &StoreKey<'_>) -> OsString {
155
465
    match key {
156
3
        StoreKey::Str(str) => format!("{folder}/{STR_FOLDER}/{str}"),
157
462
        StoreKey::Digest(digest_info) => format!("{folder}/{DIGEST_FOLDER}/{digest_info}"),
158
    }
159
465
    .into()
160
465
}
161
162
pub trait FileEntry: LenEntry + Send + Sync + Debug + 'static {
163
    /// Responsible for creating the underlying `FileEntry`.
164
    fn create(data_size: u64, block_size: u64, encoded_file_path: RwLock<EncodedFilePath>) -> Self;
165
166
    /// Creates a (usually) temp file, opens it and returns the path to the temp file.
167
    fn make_and_open_file(
168
        block_size: u64,
169
        encoded_file_path: EncodedFilePath,
170
    ) -> impl Future<Output = Result<(Self, fs::FileSlot, OsString), Error>> + Send
171
    where
172
        Self: Sized;
173
174
    /// Returns the underlying reference to the size of the data in bytes
175
    fn data_size_mut(&mut self) -> &mut u64;
176
177
    /// Returns the actual size of the underlying file on the disk after accounting for filesystem block size.
178
    fn size_on_disk(&self) -> u64;
179
180
    /// Gets the underlying `EncodedfilePath`.
181
    fn get_encoded_file_path(&self) -> &RwLock<EncodedFilePath>;
182
183
    /// Returns a reader that will read part of the underlying file.
184
    fn read_file_part(
185
        &self,
186
        offset: u64,
187
        length: u64,
188
    ) -> impl Future<Output = Result<Take<fs::FileSlot>, Error>> + Send;
189
190
    /// This function is a safe way to extract the file name of the underlying file. To protect users from
191
    /// accidentally creating undefined behavior we encourage users to do the logic they need to do with
192
    /// the filename inside this function instead of extracting the filename and doing the logic outside.
193
    /// This is because the filename is not guaranteed to exist after this function returns, however inside
194
    /// the callback the file is always guaranteed to exist and immutable.
195
    /// DO NOT USE THIS FUNCTION TO EXTRACT THE FILENAME AND STORE IT FOR LATER USE.
196
    fn get_file_path_locked<
197
        T,
198
        Fut: Future<Output = Result<T, Error>> + Send,
199
        F: FnOnce(OsString) -> Fut + Send,
200
    >(
201
        &self,
202
        handler: F,
203
    ) -> impl Future<Output = Result<T, Error>> + Send;
204
}
205
206
pub struct FileEntryImpl {
207
    data_size: u64,
208
    block_size: u64,
209
    encoded_file_path: RwLock<EncodedFilePath>,
210
}
211
212
impl FileEntryImpl {
213
9
    pub fn get_shared_context_for_test(&mut self) -> Arc<SharedContext> {
214
9
        self.encoded_file_path.get_mut().shared_context.clone()
215
9
    }
216
}
217
218
impl FileEntry for FileEntryImpl {
219
118
    fn create(data_size: u64, block_size: u64, encoded_file_path: RwLock<EncodedFilePath>) -> Self {
220
118
        Self {
221
118
            data_size,
222
118
            block_size,
223
118
            encoded_file_path,
224
118
        }
225
118
    }
226
227
    /// This encapsulates the logic for the edge case of if the file fails to create
228
    /// the cleanup of the file is handled without creating a `FileEntry`, which would
229
    /// try to cleanup the file as well during `drop()`.
230
108
    async fn make_and_open_file(
231
108
        block_size: u64,
232
108
        encoded_file_path: EncodedFilePath,
233
108
    ) -> Result<(Self, fs::FileSlot, OsString), Error> {
234
108
        let temp_full_path = encoded_file_path.get_file_path().to_os_string();
235
108
        let temp_file_result = fs::create_file(temp_full_path.clone())
236
108
            .or_else(|mut err| async 
{0
237
0
                let remove_result = fs::remove_file(&temp_full_path).await.err_tip(|| {
238
0
                    format!(
239
0
                        "Failed to remove file {} in filesystem store",
240
0
                        temp_full_path.display()
241
                    )
242
0
                });
243
0
                if let Err(remove_err) = remove_result {
  Branch (243:24): [True: 0, False: 0]
  Branch (243:24): [True: 0, False: 0]
  Branch (243:24): [True: 0, False: 0]
  Branch (243:24): [Folded - Ignored]
  Branch (243:24): [True: 0, False: 0]
244
0
                    err = err.merge(remove_err);
245
0
                }
246
0
                warn!(?err, ?block_size, ?temp_full_path, "Failed to create file",);
247
0
                Err(err).err_tip(|| {
248
0
                    format!(
249
0
                        "Failed to create {} in filesystem store",
250
0
                        temp_full_path.display()
251
                    )
252
0
                })
253
0
            })
254
108
            .await
?0
;
255
256
108
        Ok((
257
108
            <Self as FileEntry>::create(
258
108
                0, /* Unknown yet, we will fill it in later */
259
108
                block_size,
260
108
                RwLock::new(encoded_file_path),
261
108
            ),
262
108
            temp_file_result,
263
108
            temp_full_path,
264
108
        ))
265
108
    }
266
267
108
    fn data_size_mut(&mut self) -> &mut u64 {
268
108
        &mut self.data_size
269
108
    }
270
271
245
    fn size_on_disk(&self) -> u64 {
272
245
        self.data_size.div_ceil(self.block_size) * self.block_size
273
245
    }
274
275
169
    fn get_encoded_file_path(&self) -> &RwLock<EncodedFilePath> {
276
169
        &self.encoded_file_path
277
169
    }
278
279
46
    fn read_file_part(
280
46
        &self,
281
46
        offset: u64,
282
46
        length: u64,
283
46
    ) -> impl Future<Output = Result<Take<fs::FileSlot>, Error>> + Send {
284
46
        self.get_file_path_locked(move |full_content_path| async move {
285
46
            let 
file44
= fs::open_file(&full_content_path, offset, length)
286
46
                .await
287
46
                .err_tip(|| 
{2
288
2
                    format!(
289
2
                        "Failed to open file in filesystem store {}",
290
2
                        full_content_path.display()
291
                    )
292
2
                })?;
293
44
            Ok(file)
294
92
        })
295
46
    }
296
297
52
    async fn get_file_path_locked<
298
52
        T,
299
52
        Fut: Future<Output = Result<T, Error>> + Send,
300
52
        F: FnOnce(OsString) -> Fut + Send,
301
52
    >(
302
52
        &self,
303
52
        handler: F,
304
52
    ) -> Result<T, Error> {
305
52
        let encoded_file_path = self.get_encoded_file_path().read().await;
306
52
        handler(encoded_file_path.get_file_path().to_os_string()).await
307
52
    }
308
}
309
310
impl Debug for FileEntryImpl {
311
0
    fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), core::fmt::Error> {
312
0
        f.debug_struct("FileEntryImpl")
313
0
            .field("data_size", &self.data_size)
314
0
            .field("encoded_file_path", &"<behind mutex>")
315
0
            .finish()
316
0
    }
317
}
318
319
135
fn make_temp_digest(mut digest: DigestInfo) -> DigestInfo {
320
    static DELETE_FILE_COUNTER: AtomicU64 = AtomicU64::new(0);
321
135
    let mut hash = *digest.packed_hash();
322
135
    hash[24..].clone_from_slice(
323
135
        &DELETE_FILE_COUNTER
324
135
            .fetch_add(1, Ordering::Relaxed)
325
135
            .to_le_bytes(),
326
    );
327
135
    digest.set_packed_hash(*hash);
328
135
    digest
329
135
}
330
331
135
fn make_temp_key(key: &StoreKey) -> StoreKey<'static> {
332
135
    StoreKey::Digest(make_temp_digest(key.borrow().into_digest()))
333
135
}
334
335
impl LenEntry for FileEntryImpl {
336
    #[inline]
337
243
    fn len(&self) -> u64 {
338
243
        self.size_on_disk()
339
243
    }
340
341
0
    fn is_empty(&self) -> bool {
342
0
        self.data_size == 0
343
0
    }
344
345
    // unref() only triggers when an item is removed from the eviction_map. It is possible
346
    // that another place in code has a reference to `FileEntryImpl` and may later read the
347
    // file. To support this edge case, we first move the file to a temp file and point
348
    // target file location to the new temp file. `unref()` should only ever be called once.
349
    #[inline]
350
28
    async fn unref(&self) {
351
        {
352
28
            let mut encoded_file_path = self.encoded_file_path.write().await;
353
28
            if encoded_file_path.path_type == PathType::Temp {
  Branch (353:16): [True: 0, False: 0]
  Branch (353:16): [True: 1, False: 9]
  Branch (353:16): [True: 0, False: 0]
  Branch (353:16): [True: 0, False: 0]
  Branch (353:16): [Folded - Ignored]
  Branch (353:16): [True: 0, False: 18]
354
                // We are already a temp file that is now marked for deletion on drop.
355
                // This is very rare, but most likely the rename into the content path failed.
356
1
                return;
357
27
            }
358
27
            let from_path = encoded_file_path.get_file_path();
359
27
            let new_key = make_temp_key(&encoded_file_path.key);
360
361
27
            let to_path =
362
27
                to_full_path_from_key(&encoded_file_path.shared_context.temp_path, &new_key);
363
364
27
            if let Err(
err2
) = fs::rename(&from_path, &to_path).await {
  Branch (364:20): [True: 0, False: 0]
  Branch (364:20): [True: 2, False: 7]
  Branch (364:20): [True: 0, False: 0]
  Branch (364:20): [True: 0, False: 0]
  Branch (364:20): [Folded - Ignored]
  Branch (364:20): [True: 0, False: 18]
365
2
                warn!(
366
2
                    key = ?encoded_file_path.key,
367
                    ?from_path,
368
                    ?to_path,
369
                    ?err,
370
2
                    "Failed to rename file",
371
                );
372
            } else {
373
25
                debug!(
374
0
                    key = ?encoded_file_path.key,
375
                    ?from_path,
376
                    ?to_path,
377
0
                    "Renamed file",
378
                );
379
25
                encoded_file_path.path_type = PathType::Temp;
380
25
                encoded_file_path.key = new_key;
381
            }
382
        }
383
28
    }
384
}
385
386
#[inline]
387
2
fn digest_from_filename(file_name: &str) -> Result<DigestInfo, Error> {
388
2
    let (hash, size) = file_name.split_once('-').err_tip(|| "")
?0
;
389
2
    let size = size.parse::<i64>()
?0
;
390
2
    DigestInfo::try_new(hash, size)
391
2
}
392
393
2
pub fn key_from_file(file_name: &str, file_type: FileType) -> Result<StoreKey<'_>, Error> {
394
2
    match file_type {
395
0
        FileType::String => Ok(StoreKey::new_str(file_name)),
396
2
        FileType::Digest => digest_from_filename(file_name).map(StoreKey::Digest),
397
    }
398
2
}
399
400
/// The number of files to read the metadata for at the same time when running
401
/// `add_files_to_cache`.
402
const SIMULTANEOUS_METADATA_READS: usize = 200;
403
404
41
async fn add_files_to_cache<Fe: FileEntry>(
405
41
    evicting_map: &EvictingMap<StoreKeyBorrow, Arc<Fe>, SystemTime>,
406
41
    anchor_time: &SystemTime,
407
41
    shared_context: &Arc<SharedContext>,
408
41
    block_size: u64,
409
41
    rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>,
410
41
) -> Result<(), Error> {
411
    #[expect(clippy::too_many_arguments)]
412
1
    async fn process_entry<Fe: FileEntry>(
413
1
        evicting_map: &EvictingMap<StoreKeyBorrow, Arc<Fe>, SystemTime>,
414
1
        file_name: &str,
415
1
        file_type: FileType,
416
1
        atime: SystemTime,
417
1
        data_size: u64,
418
1
        block_size: u64,
419
1
        anchor_time: &SystemTime,
420
1
        shared_context: &Arc<SharedContext>,
421
1
    ) -> Result<(), Error> {
422
1
        let key = key_from_file(file_name, file_type)
?0
;
423
424
1
        let file_entry = Fe::create(
425
1
            data_size,
426
1
            block_size,
427
1
            RwLock::new(EncodedFilePath {
428
1
                shared_context: shared_context.clone(),
429
1
                path_type: PathType::Content,
430
1
                key: key.borrow().into_owned(),
431
1
            }),
432
        );
433
1
        let time_since_anchor = anchor_time
434
1
            .duration_since(atime)
435
1
            .map_err(|_| make_input_err!("File access time newer than now"))
?0
;
436
1
        evicting_map
437
1
            .insert_with_time(
438
1
                key.into_owned().into(),
439
1
                Arc::new(file_entry),
440
1
                time_since_anchor.as_secs() as i32,
441
1
            )
442
1
            .await;
443
1
        Ok(())
444
1
    }
445
446
123
    async fn read_files(
447
123
        folder: Option<&str>,
448
123
        shared_context: &SharedContext,
449
123
    ) -> Result<Vec<(String, SystemTime, u64, bool)>, Error> {
450
        // Note: In Dec 2024 this is for backwards compatibility with the old
451
        // way files were stored on disk. Previously all files were in a single
452
        // folder regardless of the StoreKey type. This allows old versions of
453
        // nativelink file layout to be upgraded at startup time.
454
        // This logic can be removed once more time has passed.
455
123
        let read_dir = folder.map_or_else(
456
41
            || format!("{}/", shared_context.content_path),
457
82
            |folder| format!("{}/{folder}/", shared_context.content_path),
458
        );
459
460
123
        let (_permit, dir_handle) = fs::read_dir(read_dir)
461
123
            .await
462
123
            .err_tip(|| "Failed opening content directory for iterating in filesystem store")
?0
463
123
            .into_inner();
464
465
123
        let read_dir_stream = ReadDirStream::new(dir_handle);
466
123
        read_dir_stream
467
123
            .map(|dir_entry| async move 
{83
468
83
                let dir_entry = dir_entry.unwrap();
469
83
                let file_name = dir_entry.file_name().into_string().unwrap();
470
83
                let metadata = dir_entry
471
83
                    .metadata()
472
83
                    .await
473
83
                    .err_tip(|| "Failed to get metadata in filesystem store")
?0
;
474
                // We need to filter out folders - we do not want to try to cache the s and d folders.
475
83
                let is_file =
476
83
                    metadata.is_file() || !(
file_name == STR_FOLDER82
||
file_name == DIGEST_FOLDER41
);
  Branch (476:21): [True: 0, False: 0]
  Branch (476:45): [True: 0, False: 0]
  Branch (476:21): [True: 1, False: 32]
  Branch (476:45): [True: 16, False: 16]
  Branch (476:21): [True: 0, False: 4]
  Branch (476:45): [True: 2, False: 2]
  Branch (476:21): [Folded - Ignored]
  Branch (476:45): [Folded - Ignored]
  Branch (476:21): [True: 0, False: 46]
  Branch (476:45): [True: 23, False: 23]
477
                // Using access time is not perfect, but better than random. We do not update the
478
                // atime when a file is actually "touched", we rely on whatever the filesystem does
479
                // when we read the file (usually update on read).
480
83
                let atime = metadata
481
83
                    .accessed()
482
83
                    .or_else(|_| 
metadata0
.
modified0
())
483
83
                    .unwrap_or(SystemTime::UNIX_EPOCH);
484
83
                Result::<(String, SystemTime, u64, bool), Error>::Ok((
485
83
                    file_name,
486
83
                    atime,
487
83
                    metadata.len(),
488
83
                    is_file,
489
83
                ))
490
166
            })
491
123
            .buffer_unordered(SIMULTANEOUS_METADATA_READS)
492
123
            .try_collect()
493
123
            .await
494
123
    }
495
496
    /// Note: In Dec 2024 this is for backwards compatibility with the old
497
    /// way files were stored on disk. Previously all files were in a single
498
    /// folder regardless of the [`StoreKey`] type. This moves files from the old cache
499
    /// location to the new cache location, under [`DIGEST_FOLDER`].
500
41
    async fn move_old_cache(
501
41
        shared_context: &Arc<SharedContext>,
502
41
        rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>,
503
41
    ) -> Result<(), Error> {
504
41
        let file_infos = read_files(None, shared_context).await
?0
;
505
506
41
        let from_path = shared_context.content_path.to_string();
507
508
41
        let to_path = format!("{}/{DIGEST_FOLDER}", shared_context.content_path);
509
510
41
        for (
file_name0
, _, _, _) in file_infos.into_iter().filter(|x| x.3) {
511
0
            let from_file: OsString = format!("{from_path}/{file_name}").into();
512
0
            let to_file: OsString = format!("{to_path}/{file_name}").into();
513
514
0
            if let Err(err) = rename_fn(&from_file, &to_file) {
  Branch (514:20): [True: 0, False: 0]
  Branch (514:20): [True: 0, False: 0]
  Branch (514:20): [True: 0, False: 0]
  Branch (514:20): [Folded - Ignored]
  Branch (514:20): [True: 0, False: 0]
515
0
                warn!(?from_file, ?to_file, ?err, "Failed to rename file",);
516
            } else {
517
0
                debug!(?from_file, ?to_file, "Renamed file",);
518
            }
519
        }
520
41
        Ok(())
521
41
    }
522
523
82
    async fn add_files_to_cache<Fe: FileEntry>(
524
82
        evicting_map: &EvictingMap<StoreKeyBorrow, Arc<Fe>, SystemTime>,
525
82
        anchor_time: &SystemTime,
526
82
        shared_context: &Arc<SharedContext>,
527
82
        block_size: u64,
528
82
        folder: &str,
529
82
    ) -> Result<(), Error> {
530
82
        let file_infos = read_files(Some(folder), shared_context).await
?0
;
531
82
        let file_type = match folder {
532
82
            STR_FOLDER => 
FileType::String41
,
533
41
            DIGEST_FOLDER => FileType::Digest,
534
0
            _ => panic!("Invalid folder type"),
535
        };
536
537
82
        let path_root = format!("{}/{folder}", shared_context.content_path);
538
539
82
        for (
file_name1
,
atime1
,
data_size1
, _) in file_infos.into_iter().filter(|x| x.3) {
540
1
            let result = process_entry(
541
1
                evicting_map,
542
1
                &file_name,
543
1
                file_type,
544
1
                atime,
545
1
                data_size,
546
1
                block_size,
547
1
                anchor_time,
548
1
                shared_context,
549
1
            )
550
1
            .await;
551
1
            if let Err(
err0
) = result {
  Branch (551:20): [True: 0, False: 0]
  Branch (551:20): [True: 0, False: 0]
  Branch (551:20): [True: 0, False: 0]
  Branch (551:20): [True: 0, False: 0]
  Branch (551:20): [True: 0, False: 0]
  Branch (551:20): [True: 0, False: 0]
  Branch (551:20): [True: 0, False: 1]
  Branch (551:20): [True: 0, False: 0]
  Branch (551:20): [Folded - Ignored]
  Branch (551:20): [True: 0, False: 0]
552
0
                warn!(?file_name, ?err, "Failed to add file to eviction cache",);
553
                // Ignore result.
554
0
                drop(fs::remove_file(format!("{path_root}/{file_name}")).await);
555
1
            }
556
        }
557
82
        Ok(())
558
82
    }
559
560
41
    move_old_cache(shared_context, rename_fn).await
?0
;
561
562
41
    add_files_to_cache(
563
41
        evicting_map,
564
41
        anchor_time,
565
41
        shared_context,
566
41
        block_size,
567
41
        DIGEST_FOLDER,
568
41
    )
569
41
    .await
?0
;
570
571
41
    add_files_to_cache(
572
41
        evicting_map,
573
41
        anchor_time,
574
41
        shared_context,
575
41
        block_size,
576
41
        STR_FOLDER,
577
41
    )
578
41
    .await
?0
;
579
41
    Ok(())
580
41
}
581
582
41
async fn prune_temp_path(temp_path: &str) -> Result<(), Error> {
583
82
    async fn prune_temp_inner(temp_path: &str, subpath: &str) -> Result<(), Error> {
584
82
        let (_permit, dir_handle) = fs::read_dir(format!("{temp_path}/{subpath}"))
585
82
            .await
586
82
            .err_tip(
587
                || "Failed opening temp directory to prune partial downloads in filesystem store",
588
0
            )?
589
82
            .into_inner();
590
591
82
        let mut read_dir_stream = ReadDirStream::new(dir_handle);
592
82
        while let Some(
dir_entry0
) = read_dir_stream.next().await {
  Branch (592:19): [True: 0, False: 0]
  Branch (592:19): [True: 0, False: 32]
  Branch (592:19): [True: 0, False: 4]
  Branch (592:19): [Folded - Ignored]
  Branch (592:19): [True: 0, False: 46]
593
0
            let path = dir_entry?.path();
594
0
            if let Err(err) = fs::remove_file(&path).await {
  Branch (594:20): [True: 0, False: 0]
  Branch (594:20): [True: 0, False: 0]
  Branch (594:20): [True: 0, False: 0]
  Branch (594:20): [Folded - Ignored]
  Branch (594:20): [True: 0, False: 0]
595
0
                warn!(?path, ?err, "Failed to delete file",);
596
0
            }
597
        }
598
82
        Ok(())
599
82
    }
600
601
41
    prune_temp_inner(temp_path, STR_FOLDER).await
?0
;
602
41
    prune_temp_inner(temp_path, DIGEST_FOLDER).await
?0
;
603
41
    Ok(())
604
41
}
605
606
#[derive(Debug, MetricsComponent)]
607
pub struct FilesystemStore<Fe: FileEntry = FileEntryImpl> {
608
    #[metric]
609
    shared_context: Arc<SharedContext>,
610
    #[metric(group = "evicting_map")]
611
    evicting_map: Arc<EvictingMap<StoreKeyBorrow, Arc<Fe>, SystemTime>>,
612
    #[metric(help = "Block size of the configured filesystem")]
613
    block_size: u64,
614
    #[metric(help = "Size of the configured read buffer size")]
615
    read_buffer_size: usize,
616
    weak_self: Weak<Self>,
617
    rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>,
618
}
619
620
impl<Fe: FileEntry> FilesystemStore<Fe> {
621
34
    pub async fn new(spec: &FilesystemSpec) -> Result<Arc<Self>, Error> {
622
107
        
Self::new_with_timeout_and_rename_fn34
(
spec34
, |from, to| std::fs::rename(from, to)).
await34
623
34
    }
624
625
41
    pub async fn new_with_timeout_and_rename_fn(
626
41
        spec: &FilesystemSpec,
627
41
        rename_fn: fn(&OsStr, &OsStr) -> Result<(), std::io::Error>,
628
41
    ) -> Result<Arc<Self>, Error> {
629
82
        async fn create_subdirs(path: &str) -> Result<(), Error> {
630
82
            fs::create_dir_all(format!("{path}/{STR_FOLDER}"))
631
82
                .await
632
82
                .err_tip(|| format!(
"Failed to create directory {path}/{STR_FOLDER}"0
))
?0
;
633
82
            fs::create_dir_all(format!("{path}/{DIGEST_FOLDER}"))
634
82
                .await
635
82
                .err_tip(|| format!(
"Failed to create directory {path}/{DIGEST_FOLDER}"0
))
636
82
        }
637
638
41
        let now = SystemTime::now();
639
640
41
        let empty_policy = nativelink_config::stores::EvictionPolicy::default();
641
41
        let eviction_policy = spec.eviction_policy.as_ref().unwrap_or(&empty_policy);
642
41
        let evicting_map = Arc::new(EvictingMap::new(eviction_policy, now));
643
644
        // Create temp and content directories and the s and d subdirectories.
645
646
41
        create_subdirs(&spec.temp_path).await
?0
;
647
41
        create_subdirs(&spec.content_path).await
?0
;
648
649
41
        let shared_context = Arc::new(SharedContext {
650
41
            active_drop_spawns: AtomicU64::new(0),
651
41
            temp_path: spec.temp_path.clone(),
652
41
            content_path: spec.content_path.clone(),
653
41
        });
654
655
41
        let block_size = if spec.block_size == 0 {
  Branch (655:29): [True: 0, False: 0]
  Branch (655:29): [True: 0, False: 1]
  Branch (655:29): [True: 0, False: 1]
  Branch (655:29): [True: 1, False: 0]
  Branch (655:29): [True: 0, False: 1]
  Branch (655:29): [True: 1, False: 0]
  Branch (655:29): [True: 10, False: 1]
  Branch (655:29): [True: 2, False: 0]
  Branch (655:29): [Folded - Ignored]
  Branch (655:29): [True: 23, False: 0]
656
37
            DEFAULT_BLOCK_SIZE
657
        } else {
658
4
            spec.block_size
659
        };
660
41
        add_files_to_cache(
661
41
            evicting_map.as_ref(),
662
41
            &now,
663
41
            &shared_context,
664
41
            block_size,
665
41
            rename_fn,
666
41
        )
667
41
        .await
?0
;
668
41
        prune_temp_path(&shared_context.temp_path).await
?0
;
669
670
41
        let read_buffer_size = if spec.read_buffer_size == 0 {
  Branch (670:35): [True: 0, False: 0]
  Branch (670:35): [True: 1, False: 0]
  Branch (670:35): [True: 0, False: 1]
  Branch (670:35): [True: 1, False: 0]
  Branch (670:35): [True: 0, False: 1]
  Branch (670:35): [True: 1, False: 0]
  Branch (670:35): [True: 4, False: 7]
  Branch (670:35): [True: 2, False: 0]
  Branch (670:35): [Folded - Ignored]
  Branch (670:35): [True: 23, False: 0]
671
32
            DEFAULT_BUFF_SIZE
672
        } else {
673
9
            spec.read_buffer_size as usize
674
        };
675
41
        Ok(Arc::new_cyclic(|weak_self| Self {
676
41
            shared_context,
677
41
            evicting_map,
678
41
            block_size,
679
41
            read_buffer_size,
680
41
            weak_self: weak_self.clone(),
681
41
            rename_fn,
682
41
        }))
683
41
    }
684
685
24
    pub fn get_arc(&self) -> Option<Arc<Self>> {
686
24
        self.weak_self.upgrade()
687
24
    }
688
689
9
    pub async fn get_file_entry_for_digest(&self, digest: &DigestInfo) -> Result<Arc<Fe>, Error> {
690
9
        self.evicting_map
691
9
            .get::<StoreKey<'static>>(&digest.into())
692
9
            .await
693
9
            .ok_or_else(|| make_err!(
Code::NotFound0
, "{digest} not found in filesystem store"))
694
9
    }
695
696
108
    async fn update_file(
697
108
        self: Pin<&Self>,
698
108
        mut entry: Fe,
699
108
        mut temp_file: fs::FileSlot,
700
108
        final_key: StoreKey<'static>,
701
108
        mut reader: DropCloserReadHalf,
702
108
    ) -> Result<(), Error> {
703
108
        let mut data_size = 0;
704
        loop {
705
183
            let mut data = reader
706
183
                .recv()
707
183
                .await
708
183
                .err_tip(|| "Failed to receive data in filesystem store")
?0
;
709
183
            let data_len = data.len();
710
183
            if data_len == 0 {
  Branch (710:16): [True: 0, False: 0]
  Branch (710:16): [True: 2, False: 2]
  Branch (710:16): [True: 2, False: 2]
  Branch (710:16): [True: 2, False: 2]
  Branch (710:16): [True: 2, False: 2]
  Branch (710:16): [True: 1, False: 1]
  Branch (710:16): [True: 13, False: 11]
  Branch (710:16): [True: 0, False: 0]
  Branch (710:16): [Folded - Ignored]
  Branch (710:16): [True: 86, False: 55]
711
108
                break; // EOF.
712
75
            }
713
75
            temp_file
714
75
                .write_all_buf(&mut data)
715
75
                .await
716
75
                .err_tip(|| "Failed to write data into filesystem store")
?0
;
717
75
            data_size += data_len as u64;
718
        }
719
720
108
        temp_file
721
108
            .as_ref()
722
108
            .sync_all()
723
108
            .await
724
108
            .err_tip(|| "Failed to sync_data in filesystem store")
?0
;
725
726
108
        drop(temp_file);
727
728
108
        *entry.data_size_mut() = data_size;
729
108
        self.emplace_file(final_key, Arc::new(entry)).await
730
107
    }
731
732
117
    async fn emplace_file(&self, key: StoreKey<'static>, entry: Arc<Fe>) -> Result<(), Error> {
733
        // This sequence of events is quite ticky to understand due to the amount of triggers that
734
        // happen, async'ness of it and the locking. So here is a breakdown of what happens:
735
        // 1. Here will hold a write lock on any file operations of this FileEntry.
736
        // 2. Then insert the entry into the evicting map. This may trigger an eviction of other
737
        //    entries.
738
        // 3. Eviction triggers `unref()`, which grabs a write lock on the evicted FileEntry
739
        //    during the rename.
740
        // 4. It should be impossible for items to be added while eviction is happening, so there
741
        //    should not be a deadlock possibility. However, it is possible for the new FileEntry
742
        //    to be evicted before the file is moved into place. Eviction of the newly inserted
743
        //    item is not possible within the `insert()` call because the write lock inside the
744
        //    eviction map. If an eviction of new item happens after `insert()` but before
745
        //    `rename()` then we get to finish our operation because the `unref()` of the new item
746
        //    will be blocked on us because we currently have the lock.
747
        // 5. Move the file into place. Since we hold a write lock still anyone that gets our new
748
        //    FileEntry (which has not yet been placed on disk) will not be able to read the file's
749
        //    contents until we release the lock.
750
117
        let evicting_map = self.evicting_map.clone();
751
117
        let rename_fn = self.rename_fn;
752
753
        // We need to guarantee that this will get to the end even if the parent future is dropped.
754
        // See: https://github.com/TraceMachina/nativelink/issues/495
755
117
        background_spawn!("filesystem_store_emplace_file", async move {
756
117
            let mut encoded_file_path = entry.get_encoded_file_path().write().await;
757
117
            let final_path = get_file_path_raw(
758
117
                &PathType::Content,
759
117
                encoded_file_path.shared_context.as_ref(),
760
117
                &key,
761
            );
762
763
117
            evicting_map
764
117
                .insert(key.borrow().into_owned().into(), entry.clone())
765
117
                .await;
766
767
117
            let from_path = encoded_file_path.get_file_path();
768
            // Internally tokio spawns fs commands onto a blocking thread anyways.
769
            // Since we are already on a blocking thread, we just need the `fs` wrapper to manage
770
            // an open-file permit (ensure we don't open too many files at once).
771
117
            let result = (rename_fn)(&from_path, &final_path).err_tip(|| 
{1
772
1
                format!(
773
1
                    "Failed to rename temp file to final path {}",
774
1
                    final_path.display()
775
                )
776
1
            });
777
778
            // In the event our move from temp file to final file fails we need to ensure we remove
779
            // the entry from our map.
780
            // Remember: At this point it is possible for another thread to have a reference to
781
            // `entry`, so we can't delete the file, only drop() should ever delete files.
782
117
            if let Err(
err1
) = result {
  Branch (782:20): [True: 0, False: 0]
  Branch (782:20): [True: 0, False: 2]
  Branch (782:20): [True: 0, False: 2]
  Branch (782:20): [True: 0, False: 2]
  Branch (782:20): [True: 0, False: 2]
  Branch (782:20): [True: 1, False: 0]
  Branch (782:20): [True: 0, False: 15]
  Branch (782:20): [True: 0, False: 0]
  Branch (782:20): [Folded - Ignored]
  Branch (782:20): [True: 0, False: 93]
783
1
                error!(?err, ?from_path, ?final_path, "Failed to rename file",);
784
                // Warning: To prevent deadlock we need to release our lock or during `remove_if()`
785
                // it will call `unref()`, which triggers a write-lock on `encoded_file_path`.
786
1
                drop(encoded_file_path);
787
                // It is possible that the item in our map is no longer the item we inserted,
788
                // So, we need to conditionally remove it only if the pointers are the same.
789
790
1
                evicting_map
791
1
                    .remove_if(&key, |map_entry| Arc::<Fe>::ptr_eq(map_entry, &entry))
792
1
                    .await;
793
1
                return Err(err);
794
116
            }
795
116
            encoded_file_path.path_type = PathType::Content;
796
116
            encoded_file_path.key = key;
797
116
            Ok(())
798
117
        })
799
117
        .await
800
116
        .err_tip(|| "Failed to create spawn in filesystem store update_file")
?0
801
116
    }
802
}
803
804
#[async_trait]
805
impl<Fe: FileEntry> StoreDriver for FilesystemStore<Fe> {
806
    async fn has_with_results(
807
        self: Pin<&Self>,
808
        keys: &[StoreKey<'_>],
809
        results: &mut [Option<u64>],
810
170
    ) -> Result<(), Error> {
811
85
        self.evicting_map
812
85
            .sizes_for_keys::<_, StoreKey<'_>, &StoreKey<'_>>(
813
85
                keys.iter(),
814
85
                results,
815
85
                false, /* peek */
816
85
            )
817
85
            .await;
818
        // We need to do a special pass to ensure our zero files exist.
819
        // If our results failed and the result was a zero file, we need to
820
        // create the file by spec.
821
85
        for (key, result) in keys.iter().zip(results.iter_mut()) {
822
85
            if result.is_some() || 
!16
is_zero_digest16
(key.borrow()) {
  Branch (822:16): [True: 0, False: 0]
  Branch (822:36): [True: 0, False: 0]
  Branch (822:16): [True: 0, False: 0]
  Branch (822:36): [True: 0, False: 0]
  Branch (822:16): [True: 0, False: 0]
  Branch (822:36): [True: 0, False: 0]
  Branch (822:16): [True: 0, False: 0]
  Branch (822:36): [True: 0, False: 0]
  Branch (822:16): [True: 0, False: 0]
  Branch (822:36): [True: 0, False: 0]
  Branch (822:16): [True: 0, False: 1]
  Branch (822:36): [True: 1, False: 0]
  Branch (822:16): [True: 1, False: 2]
  Branch (822:36): [True: 0, False: 2]
  Branch (822:16): [True: 0, False: 0]
  Branch (822:36): [True: 0, False: 0]
  Branch (822:16): [Folded - Ignored]
  Branch (822:36): [Folded - Ignored]
  Branch (822:16): [True: 68, False: 13]
  Branch (822:36): [True: 13, False: 0]
823
83
                continue;
824
2
            }
825
2
            let (mut tx, rx) = make_buf_channel_pair();
826
2
            let send_eof_result = tx.send_eof();
827
2
            self.update(key.borrow(), rx, UploadSizeInfo::ExactSize(0))
828
2
                .await
829
2
                .err_tip(|| format!(
"Failed to create zero file for key {}"0
,
key0
.
as_str0
()))
830
2
                .merge(
831
2
                    send_eof_result
832
2
                        .err_tip(|| "Failed to send zero file EOF in filesystem store has"),
833
0
                )?;
834
835
2
            *result = Some(0);
836
        }
837
85
        Ok(())
838
170
    }
839
840
    async fn update(
841
        self: Pin<&Self>,
842
        key: StoreKey<'_>,
843
        reader: DropCloserReadHalf,
844
        _upload_size: UploadSizeInfo,
845
216
    ) -> Result<(), Error> {
846
108
        let temp_key = make_temp_key(&key);
847
108
        let (entry, temp_file, temp_full_path) = Fe::make_and_open_file(
848
108
            self.block_size,
849
108
            EncodedFilePath {
850
108
                shared_context: self.shared_context.clone(),
851
108
                path_type: PathType::Temp,
852
108
                key: temp_key,
853
108
            },
854
108
        )
855
108
        .await
?0
;
856
857
108
        self.update_file(entry, temp_file, key.into_owned(), reader)
858
108
            .await
859
107
            .err_tip(|| 
{1
860
1
                format!(
861
1
                    "While processing with temp file {}",
862
1
                    temp_full_path.display()
863
                )
864
1
            })
865
215
    }
866
867
84
    fn optimized_for(&self, optimization: StoreOptimizations) -> bool {
868
84
        optimization == StoreOptimizations::FileUpdates
869
84
    }
870
871
    async fn update_with_whole_file(
872
        self: Pin<&Self>,
873
        key: StoreKey<'_>,
874
        path: OsString,
875
        file: fs::FileSlot,
876
        upload_size: UploadSizeInfo,
877
18
    ) -> Result<Option<fs::FileSlot>, Error> {
878
9
        let file_size = match upload_size {
879
9
            UploadSizeInfo::ExactSize(size) => size,
880
0
            UploadSizeInfo::MaxSize(_) => file
881
0
                .as_ref()
882
0
                .metadata()
883
0
                .await
884
0
                .err_tip(|| format!("While reading metadata for {}", path.display()))?
885
0
                .len(),
886
        };
887
9
        let entry = Fe::create(
888
9
            file_size,
889
9
            self.block_size,
890
9
            RwLock::new(EncodedFilePath {
891
9
                shared_context: self.shared_context.clone(),
892
9
                path_type: PathType::Custom(path),
893
9
                key: key.borrow().into_owned(),
894
9
            }),
895
        );
896
        // We are done with the file, if we hold a reference to the file here, it could
897
        // result in a deadlock if `emplace_file()` also needs file descriptors.
898
9
        drop(file);
899
9
        self.emplace_file(key.into_owned(), Arc::new(entry))
900
9
            .await
901
9
            .err_tip(|| "Could not move file into store in upload_file_to_store, maybe dest is on different volume?")
?0
;
902
9
        return Ok(None);
903
18
    }
904
905
    async fn get_part(
906
        self: Pin<&Self>,
907
        key: StoreKey<'_>,
908
        writer: &mut DropCloserWriteHalf,
909
        offset: u64,
910
        length: Option<u64>,
911
120
    ) -> Result<(), Error> {
912
60
        if is_zero_digest(key.borrow()) {
  Branch (912:12): [True: 0, False: 0]
  Branch (912:12): [True: 0, False: 0]
  Branch (912:12): [True: 0, False: 1]
  Branch (912:12): [True: 0, False: 0]
  Branch (912:12): [True: 0, False: 1]
  Branch (912:12): [True: 0, False: 0]
  Branch (912:12): [True: 1, False: 4]
  Branch (912:12): [True: 0, False: 0]
  Branch (912:12): [Folded - Ignored]
  Branch (912:12): [True: 15, False: 38]
913
16
            self.has(key.borrow())
914
16
                .await
915
16
                .err_tip(|| "Failed to check if zero digest exists in filesystem store")
?0
;
916
16
            writer
917
16
                .send_eof()
918
16
                .err_tip(|| "Failed to send zero EOF in filesystem store get_part")
?0
;
919
16
            return Ok(());
920
44
        }
921
44
        let entry = self.evicting_map.get(&key).await.ok_or_else(|| 
{0
922
0
            make_err!(
923
0
                Code::NotFound,
924
                "{} not found in filesystem store here",
925
0
                key.as_str()
926
            )
927
0
        })?;
928
44
        let read_limit = length.unwrap_or(u64::MAX);
929
44
        let 
mut temp_file42
= entry.read_file_part(offset, read_limit).or_else(|err| async move
{2
930
            // If the file is not found, we need to remove it from the eviction map.
931
2
            if err.code == Code::NotFound {
  Branch (931:16): [True: 0, False: 0]
  Branch (931:16): [True: 0, False: 0]
  Branch (931:16): [True: 0, False: 0]
  Branch (931:16): [True: 0, False: 0]
  Branch (931:16): [True: 0, False: 0]
  Branch (931:16): [True: 0, False: 0]
  Branch (931:16): [True: 2, False: 0]
  Branch (931:16): [True: 0, False: 0]
  Branch (931:16): [Folded - Ignored]
  Branch (931:16): [True: 0, False: 0]
932
2
                error!(
933
                    ?err,
934
                    ?key,
935
2
                    "Entry was in our map, but not found on disk. Removing from map as a precaution, but process probably need restarted."
936
                );
937
2
                self.evicting_map.remove(&key).await;
938
0
            }
939
2
            Err(err)
940
44
        
}4
).await
?2
;
941
942
        loop {
943
1.12k
            let mut buf = BytesMut::with_capacity(self.read_buffer_size);
944
1.12k
            temp_file
945
1.12k
                .read_buf(&mut buf)
946
1.12k
                .await
947
1.12k
                .err_tip(|| "Failed to read data in filesystem store")
?0
;
948
1.12k
            if buf.is_empty() {
  Branch (948:16): [True: 0, False: 0]
  Branch (948:16): [True: 0, False: 0]
  Branch (948:16): [True: 1, False: 10]
  Branch (948:16): [True: 0, False: 0]
  Branch (948:16): [True: 1, False: 10]
  Branch (948:16): [True: 0, False: 0]
  Branch (948:16): [True: 1, False: 1.02k]
  Branch (948:16): [True: 0, False: 0]
  Branch (948:16): [Folded - Ignored]
  Branch (948:16): [True: 38, False: 38]
949
41
                break; // EOF.
950
1.08k
            }
951
1.08k
            writer
952
1.08k
                .send(buf.freeze())
953
1.08k
                .await
954
1.08k
                .err_tip(|| "Failed to send chunk in filesystem store get_part")
?0
;
955
        }
956
41
        writer
957
41
            .send_eof()
958
41
            .err_tip(|| "Filed to send EOF in filesystem store get_part")
?0
;
959
960
41
        Ok(())
961
119
    }
962
963
101
    fn inner_store(&self, _digest: Option<StoreKey>) -> &dyn StoreDriver {
964
101
        self
965
101
    }
966
967
24
    fn as_any<'a>(&'a self) -> &'a (dyn core::any::Any + Sync + Send + 'static) {
968
24
        self
969
24
    }
970
971
0
    fn as_any_arc(self: Arc<Self>) -> Arc<dyn core::any::Any + Sync + Send + 'static> {
972
0
        self
973
0
    }
974
975
0
    fn register_health(self: Arc<Self>, registry: &mut HealthRegistryBuilder) {
976
0
        registry.register_indicator(self);
977
0
    }
978
}
979
980
#[async_trait]
981
impl<Fe: FileEntry> HealthStatusIndicator for FilesystemStore<Fe> {
982
0
    fn get_name(&self) -> &'static str {
983
0
        "FilesystemStore"
984
0
    }
985
986
0
    async fn check_health(&self, namespace: Cow<'static, str>) -> HealthStatus {
987
0
        StoreDriver::check_health(Pin::new(self), namespace).await
988
0
    }
989
}