aft/pins/
mod.rs

1//! Durable generation pins used while a view is assembled or read.
2//!
3//! An assembly pin is created before its first blob put so a concurrent sweep
4//! can keep every prospective blob alive until publication finishes.
5
6use std::fmt;
7use std::fs::{self, File, OpenOptions};
8use std::io::{self, Write};
9use std::path::{Path, PathBuf};
10use std::time::{SystemTime, UNIX_EPOCH};
11
12use serde::{Deserialize, Serialize};
13
14use crate::blob_store::FullKey;
15use crate::fs_lock;
16use crate::root_cache::{self, ReadMarker};
17
18/// A pin remains live for thirty minutes after its most recent successful renewal.
19pub const PIN_TTL_MS: u64 = 30 * 60 * 1_000;
20/// Assemblers renew before a third of the pin lifetime has elapsed.
21pub const PIN_RENEW_INTERVAL_MS: u64 = PIN_TTL_MS / 3;
22
23#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
24pub struct PinOwner {
25    pub pid: u32,
26    pub start_time: u64,
27}
28
29#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
30pub struct PinMetadata {
31    pub family: String,
32    pub view: String,
33    pub generation: String,
34    pub owner: PinOwner,
35    pub created_at: u64,
36    pub renewed_at: u64,
37}
38
39#[derive(Debug)]
40pub enum PinError {
41    Io(io::Error),
42    Serialize(serde_json::Error),
43    InvalidGeneration(String),
44}
45
46impl fmt::Display for PinError {
47    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
48        match self {
49            Self::Io(error) => write!(f, "pin I/O error: {error}"),
50            Self::Serialize(error) => write!(f, "pin serialization error: {error}"),
51            Self::InvalidGeneration(generation) => {
52                write!(f, "invalid pin generation `{generation}`")
53            }
54        }
55    }
56}
57
58impl std::error::Error for PinError {
59    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
60        match self {
61            Self::Io(error) => Some(error),
62            Self::Serialize(error) => Some(error),
63            Self::InvalidGeneration(_) => None,
64        }
65    }
66}
67
68impl From<io::Error> for PinError {
69    fn from(error: io::Error) -> Self {
70        Self::Io(error)
71    }
72}
73
74impl From<serde_json::Error> for PinError {
75    fn from(error: serde_json::Error) -> Self {
76        Self::Serialize(error)
77    }
78}
79
80/// A durable pin around an in-progress assembly.
81#[derive(Debug)]
82pub struct AssemblyPin {
83    keys_path: PathBuf,
84    metadata_path: PathBuf,
85    metadata: PinMetadata,
86    released: bool,
87}
88
89impl AssemblyPin {
90    /// Creates and syncs `pins/<generation>.keys` before making the pin visible.
91    /// The caller must create this guard before its first blob put.
92    pub fn create(
93        view_dir: &Path,
94        family: impl Into<String>,
95        view: impl Into<String>,
96        generation: impl Into<String>,
97        keys: &[FullKey],
98    ) -> Result<Self, PinError> {
99        let family = family.into();
100        let view = view.into();
101        let generation = generation.into();
102        validate_generation(&generation)?;
103        let pins_dir = view_dir.join("pins");
104        fs::create_dir_all(&pins_dir)?;
105
106        let keys_path = pins_dir.join(format!("{generation}.keys"));
107        write_keys(&keys_path, keys)?;
108        let now = now_ms();
109        let metadata = PinMetadata {
110            family,
111            view,
112            generation: generation.clone(),
113            owner: PinOwner {
114                pid: std::process::id(),
115                start_time: root_cache::process_start_time_ms(std::process::id()).unwrap_or(now),
116            },
117            created_at: now,
118            renewed_at: now,
119        };
120        let metadata_path = pins_dir.join(format!("{generation}.json"));
121        write_metadata(&metadata_path, &metadata)?;
122        Ok(Self {
123            keys_path,
124            metadata_path,
125            metadata,
126            released: false,
127        })
128    }
129
130    pub fn metadata(&self) -> &PinMetadata {
131        &self.metadata
132    }
133
134    pub fn keys_path(&self) -> &Path {
135        &self.keys_path
136    }
137
138    /// Renews the pin when a put is due. A renewal error is returned before the
139    /// caller's put closure runs, so an assembly cannot publish after losing its pin.
140    pub fn put<T>(&mut self, put: impl FnOnce() -> Result<T, PinError>) -> Result<T, PinError> {
141        self.renew_if_due()?;
142        put()
143    }
144
145    pub fn renew_if_due(&mut self) -> Result<(), PinError> {
146        let now = now_ms();
147        if now.saturating_sub(self.metadata.renewed_at) >= PIN_RENEW_INTERVAL_MS {
148            self.metadata.renewed_at = now;
149            write_metadata(&self.metadata_path, &self.metadata)?;
150        }
151        Ok(())
152    }
153
154    /// Removes both parts of the pin once publication has completed or aborted.
155    pub fn release(&mut self) {
156        if self.released {
157            return;
158        }
159        let _ = fs::remove_file(&self.metadata_path);
160        let _ = fs::remove_file(&self.keys_path);
161        fs_lock::sync_parent(&self.metadata_path);
162        self.released = true;
163    }
164}
165
166impl Drop for AssemblyPin {
167    fn drop(&mut self) {
168        self.release();
169    }
170}
171
172/// Pins a view generation for an in-flight query and removes the marker on drop.
173/// Existing read-marker sweeping reclaims markers left by dead owners.
174#[derive(Debug)]
175pub struct QueryPin {
176    marker: ReadMarker,
177}
178
179impl QueryPin {
180    pub fn acquire(view_dir: &Path, generation: &str) -> Result<Self, PinError> {
181        Ok(Self {
182            marker: ReadMarker::create(view_dir, generation)?,
183        })
184    }
185
186    pub fn touch_if_due(&self) -> Result<(), PinError> {
187        self.marker.touch_if_due()?;
188        Ok(())
189    }
190
191    pub fn path(&self) -> &Path {
192        self.marker.path()
193    }
194}
195
196pub(crate) fn pin_paths(view_dir: &Path, generation: &str) -> (PathBuf, PathBuf) {
197    let pins_dir = view_dir.join("pins");
198    (
199        pins_dir.join(format!("{generation}.json")),
200        pins_dir.join(format!("{generation}.keys")),
201    )
202}
203
204pub(crate) fn read_keys(path: &Path) -> Result<Vec<[u8; 32]>, PinError> {
205    let contents = fs::read_to_string(path)?;
206    contents.lines().map(parse_hex_key).collect()
207}
208
209pub(crate) fn owner_is_live(owner: &PinOwner) -> bool {
210    root_cache::process_start_time_ms(owner.pid)
211        .map(|actual| actual == owner.start_time)
212        .unwrap_or_else(|| crate::fs_lock::process_alive(owner.pid))
213}
214
215pub(crate) fn now_ms() -> u64 {
216    SystemTime::now()
217        .duration_since(UNIX_EPOCH)
218        .unwrap_or_default()
219        .as_millis() as u64
220}
221
222fn validate_generation(generation: &str) -> Result<(), PinError> {
223    if generation.is_empty()
224        || generation == "."
225        || generation == ".."
226        || generation.contains(['/', '\\'])
227    {
228        return Err(PinError::InvalidGeneration(generation.to_owned()));
229    }
230    Ok(())
231}
232
233fn write_keys(path: &Path, keys: &[FullKey]) -> Result<(), PinError> {
234    let mut encoded = keys.iter().map(FullKey::to_hex).collect::<Vec<_>>();
235    encoded.sort_unstable();
236    encoded.dedup();
237    let mut file = create_private(path)?;
238    for key in encoded {
239        writeln!(file, "{key}")?;
240    }
241    file.sync_all()?;
242    drop(file);
243    fs_lock::sync_parent(path);
244    Ok(())
245}
246
247fn write_metadata(path: &Path, metadata: &PinMetadata) -> Result<(), PinError> {
248    let temporary = path.with_extension(format!("json.tmp.{}.{}", std::process::id(), now_ms()));
249    let result = (|| {
250        let mut file = create_private(&temporary)?;
251        serde_json::to_writer(&mut file, metadata)?;
252        file.write_all(b"\n")?;
253        file.sync_all()?;
254        drop(file);
255        fs_lock::rename_over(&temporary, path)?;
256        fs_lock::sync_parent(path);
257        Ok(())
258    })();
259    if result.is_err() {
260        let _ = fs::remove_file(&temporary);
261    }
262    result
263}
264
265fn create_private(path: &Path) -> io::Result<File> {
266    #[cfg(unix)]
267    {
268        use std::os::unix::fs::OpenOptionsExt;
269        return OpenOptions::new()
270            .write(true)
271            .create_new(true)
272            .mode(0o600)
273            .open(path);
274    }
275    #[cfg(not(unix))]
276    OpenOptions::new().write(true).create_new(true).open(path)
277}
278
279fn parse_hex_key(value: &str) -> Result<[u8; 32], PinError> {
280    if value.len() != 64 {
281        return Err(PinError::InvalidGeneration(value.to_owned()));
282    }
283    let mut key = [0; 32];
284    for (index, byte) in key.iter_mut().enumerate() {
285        *byte = u8::from_str_radix(&value[index * 2..index * 2 + 2], 16)
286            .map_err(|_| PinError::InvalidGeneration(value.to_owned()))?;
287    }
288    Ok(key)
289}
290
291#[cfg(test)]
292mod tests {
293    use super::*;
294
295    #[test]
296    fn failed_renewal_stops_the_next_put() {
297        let view = tempfile::tempdir().expect("create view");
298        let mut pin = AssemblyPin::create(view.path(), "family", "view", "generation", &[])
299            .expect("create pin");
300        pin.metadata.renewed_at = now_ms().saturating_sub(PIN_RENEW_INTERVAL_MS);
301        fs::remove_file(&pin.metadata_path).expect("remove pin metadata");
302        fs::create_dir(&pin.metadata_path).expect("make renewal destination invalid");
303
304        let mut put_called = false;
305        let result = pin.put(|| {
306            put_called = true;
307            Ok(())
308        });
309        assert!(result.is_err());
310        assert!(!put_called, "a failed renewal must stop the put");
311    }
312}
aft/pins/mod.rs

aft/pins/
mod.rs
