liminal-server 0.19.1

Standalone server for the liminal messaging bus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

use std::error::Error;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};

#[cfg(unix)]
use std::os::unix::fs::MetadataExt;

type FixtureResult<T> = Result<T, Box<dyn Error>>;

static NEXT_COPY: AtomicU64 = AtomicU64::new(0);

#[derive(Debug, Default)]
struct CopyCounts {
    files: u64,
    bytes: u64,
    directories: u64,
}

fn copy_directory(source: &Path, destination: &Path) -> FixtureResult<CopyCounts> {
    let mut counts = CopyCounts::default();
    for entry in fs::read_dir(source)? {
        let entry = entry?;
        let source = entry.path();
        let destination = destination.join(entry.file_name());
        let kind = entry.file_type()?;
        if kind.is_dir() {
            fs::create_dir(&destination)?;
            fs::set_permissions(&destination, entry.metadata()?.permissions())?;
            let nested = copy_directory(&source, &destination)?;
            counts.files = counts
                .files
                .checked_add(nested.files)
                .ok_or("fixture file count overflow")?;
            counts.bytes = counts
                .bytes
                .checked_add(nested.bytes)
                .ok_or("fixture byte count overflow")?;
            counts.directories = counts
                .directories
                .checked_add(nested.directories)
                .and_then(|count| count.checked_add(1))
                .ok_or("fixture directory count overflow")?;
        } else if kind.is_file() {
            let bytes = fs::copy(&source, &destination)?;
            counts.files = counts
                .files
                .checked_add(1)
                .ok_or("fixture file count overflow")?;
            counts.bytes = counts
                .bytes
                .checked_add(bytes)
                .ok_or("fixture byte count overflow")?;
            let before = fs::metadata(&source)?;
            let after = fs::metadata(&destination)?;
            #[cfg(unix)]
            if before.dev() == after.dev() && before.ino() == after.ino() {
                return Err("persistence fixture copy shares its source inode".into());
            }
            if before.len() != bytes
                || after.len() != bytes
                || fs::read(&source)? != fs::read(&destination)?
            {
                return Err("persistence fixture copy differs from its source bytes".into());
            }
        } else {
            return Err("persistence fixture contains a link or non-regular file".into());
        }
    }
    Ok(counts)
}

pub struct Persistence {
    path: PathBuf,
}

impl Persistence {
    pub(super) fn copied() -> FixtureResult<Self> {
        let root = std::env::var_os("LIMINAL_DURABILITY_FIXTURES")
            .ok_or("persistence fixture setup did not provide LIMINAL_DURABILITY_FIXTURES")?;
        let source = Path::new(&root).join("full-services");
        if !fs::symlink_metadata(&source)?.file_type().is_dir() {
            return Err("full-services persistence fixture image is not a directory".into());
        }
        let number = NEXT_COPY
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |count| {
                count.checked_add(1)
            })
            .map_err(|_| "persistence fixture copy counter overflow")?;
        let path = Path::new(&root).join(format!(
            "full-services-copy-{}-{number}",
            std::process::id()
        ));
        fs::create_dir(&path)?;
        let fixture = Self { path };
        let counts = copy_directory(&source, fixture.path())?;
        if counts.files == 0
            || counts.bytes == 0
            || !fixture.path().join("durability/config.json").is_file()
        {
            return Err("full-services persistence fixture has no database configuration".into());
        }
        println!("full-services fixture copy: {counts:?}");
        Ok(fixture)
    }

    #[must_use]
    pub(super) fn path(&self) -> &Path {
        &self.path
    }
}

impl Drop for Persistence {
    fn drop(&mut self) {
        if std::thread::panicking() {
            eprintln!(
                "persistence fixture unwind retained {}",
                self.path.display()
            );
        } else {
            let removed = fs::remove_dir_all(&self.path);
            assert!(
                removed.is_ok(),
                "persistence fixture removal failed at {}: {removed:?}",
                self.path.display()
            );
        }
    }
}