powdb_sync/
metadata.rs

1use std::collections::HashSet;
2use std::fs::{self, File, OpenOptions};
3use std::io::{self, Write};
4use std::path::{Path, PathBuf};
5use std::sync::atomic::{AtomicU64, Ordering};
6use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
7
8use powdb_storage::create_data_dir_secure;
9use serde::{Deserialize, Serialize};
10
11use crate::segment::{read_units_since, SegmentIdentity};
12
13pub const SYNC_STATE_DIR: &str = ".powdb-sync";
14pub const IDENTITY_FILE: &str = "identity.json";
15pub const REPLICA_CURSORS_FILE: &str = "replica-cursors.json";
16pub const SYNC_METADATA_FORMAT_VERSION: u32 = 1;
17
18static TEMP_COUNTER: AtomicU64 = AtomicU64::new(0);
19const CURSOR_LOCK_FILE: &str = ".replica-cursors.lock";
20const CURSOR_LOCK_TIMEOUT: Duration = Duration::from_secs(5);
21const CURSOR_LOCK_RETRY: Duration = Duration::from_millis(5);
22const CURSOR_LOCK_STALE_AFTER: Duration = Duration::from_secs(30);
23
24#[derive(Debug, Clone, Copy, PartialEq, Eq)]
25pub struct DatabaseIdentity {
26    pub database_id: [u8; 16],
27    pub primary_generation: u64,
28}
29
30impl DatabaseIdentity {
31    pub fn segment_identity(self) -> SegmentIdentity {
32        SegmentIdentity::current(self.database_id, self.primary_generation)
33    }
34}
35
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
37pub struct ReplicaCursor {
38    pub replica_id: String,
39    pub applied_lsn: u64,
40    pub updated_unix_secs: u64,
41    pub active: bool,
42}
43
44impl ReplicaCursor {
45    pub fn active(replica_id: impl Into<String>, applied_lsn: u64) -> Self {
46        Self {
47            replica_id: replica_id.into(),
48            applied_lsn,
49            updated_unix_secs: now_unix_secs(),
50            active: true,
51        }
52    }
53
54    pub fn next_required_lsn(&self) -> io::Result<u64> {
55        self.applied_lsn
56            .checked_add(1)
57            .ok_or_else(|| invalid_data("replica cursor LSN overflow"))
58    }
59}
60
61#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
62pub struct IdentitySnapshot {
63    pub format_version: u32,
64    pub database_id: String,
65    pub primary_generation: u64,
66    pub created_unix_secs: u64,
67}
68
69impl IdentitySnapshot {
70    pub fn from_identity(identity: DatabaseIdentity, created_unix_secs: u64) -> Self {
71        Self {
72            format_version: SYNC_METADATA_FORMAT_VERSION,
73            database_id: encode_hex_16(identity.database_id),
74            primary_generation: identity.primary_generation,
75            created_unix_secs,
76        }
77    }
78
79    pub fn identity(&self) -> io::Result<DatabaseIdentity> {
80        if self.format_version != SYNC_METADATA_FORMAT_VERSION {
81            return Err(invalid_data(format!(
82                "unsupported sync identity format {}",
83                self.format_version
84            )));
85        }
86        let identity = DatabaseIdentity {
87            database_id: decode_hex_16(&self.database_id)?,
88            primary_generation: self.primary_generation,
89        };
90        validate_identity(identity)?;
91        Ok(identity)
92    }
93
94    pub fn validate(&self) -> io::Result<()> {
95        self.identity().map(|_| ())
96    }
97}
98
99#[derive(Debug, Clone, Serialize, Deserialize)]
100struct CursorFile {
101    format_version: u32,
102    cursors: Vec<ReplicaCursor>,
103}
104
105#[derive(Debug, Clone, Serialize, Deserialize)]
106struct CursorLockFile {
107    format_version: u32,
108    owner_pid: u32,
109    created_unix_secs: u64,
110}
111
112pub fn sync_state_dir(data_dir: &Path) -> PathBuf {
113    data_dir.join(SYNC_STATE_DIR)
114}
115
116pub fn open_or_create_identity(data_dir: &Path) -> io::Result<DatabaseIdentity> {
117    let state_dir = sync_state_dir(data_dir);
118    create_data_dir_secure(&state_dir)?;
119
120    match read_identity(data_dir) {
121        Ok(identity) => return Ok(identity),
122        Err(err) if err.kind() == io::ErrorKind::NotFound => {}
123        Err(err) => return Err(err),
124    }
125
126    let identity = DatabaseIdentity {
127        database_id: generate_database_id()?,
128        primary_generation: 1,
129    };
130    let file = IdentitySnapshot::from_identity(identity, now_unix_secs());
131
132    match write_new_identity_file(&state_dir, &file) {
133        Ok(()) => Ok(identity),
134        Err(err) if err.kind() == io::ErrorKind::AlreadyExists => read_identity(data_dir),
135        Err(err) => Err(err),
136    }
137}
138
139pub fn read_identity(data_dir: &Path) -> io::Result<DatabaseIdentity> {
140    let Some(snapshot) = read_identity_snapshot(data_dir)? else {
141        return Err(io::Error::new(
142            io::ErrorKind::NotFound,
143            "sync identity not found",
144        ));
145    };
146    snapshot.identity()
147}
148
149pub fn read_identity_snapshot(data_dir: &Path) -> io::Result<Option<IdentitySnapshot>> {
150    let bytes = fs::read(sync_state_dir(data_dir).join(IDENTITY_FILE))?;
151    let snapshot: IdentitySnapshot = serde_json::from_slice(&bytes).map_err(invalid_data)?;
152    snapshot.validate()?;
153    Ok(Some(snapshot))
154}
155
156pub fn read_identity_snapshot_if_exists(data_dir: &Path) -> io::Result<Option<IdentitySnapshot>> {
157    match read_identity_snapshot(data_dir) {
158        Ok(snapshot) => Ok(snapshot),
159        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(None),
160        Err(err) => Err(err),
161    }
162}
163
164pub fn write_identity_snapshot(data_dir: &Path, snapshot: &IdentitySnapshot) -> io::Result<()> {
165    snapshot.validate()?;
166    let state_dir = sync_state_dir(data_dir);
167    create_data_dir_secure(&state_dir)?;
168    match write_new_identity_file(&state_dir, snapshot) {
169        Ok(()) => Ok(()),
170        Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {
171            let Some(existing) = read_identity_snapshot(data_dir)? else {
172                return Err(io::Error::new(
173                    io::ErrorKind::AlreadyExists,
174                    "sync identity already exists but could not be read",
175                ));
176            };
177            if existing == *snapshot {
178                Ok(())
179            } else {
180                Err(io::Error::new(
181                    io::ErrorKind::AlreadyExists,
182                    "sync identity already exists with different database history",
183                ))
184            }
185        }
186        Err(err) => Err(err),
187    }
188}
189
190pub fn read_replica_cursors(data_dir: &Path) -> io::Result<Vec<ReplicaCursor>> {
191    read_replica_cursors_unlocked(data_dir)
192}
193
194pub(crate) fn read_replica_cursors_unlocked(data_dir: &Path) -> io::Result<Vec<ReplicaCursor>> {
195    let path = sync_state_dir(data_dir).join(REPLICA_CURSORS_FILE);
196    let bytes = match fs::read(path) {
197        Ok(bytes) => bytes,
198        Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
199        Err(err) => return Err(err),
200    };
201    let file: CursorFile = serde_json::from_slice(&bytes).map_err(invalid_data)?;
202    validate_cursor_file(&file)?;
203    Ok(file.cursors)
204}
205
206pub fn write_replica_cursors(data_dir: &Path, mut cursors: Vec<ReplicaCursor>) -> io::Result<()> {
207    validate_cursors_for_write(&cursors)?;
208    cursors.sort_by(|a, b| a.replica_id.cmp(&b.replica_id));
209    let state_dir = sync_state_dir(data_dir);
210    create_data_dir_secure(&state_dir)?;
211    let _lock = acquire_cursor_lock(&state_dir)?;
212    validate_active_cursors_against_retained_tail(data_dir, &cursors)?;
213    write_replica_cursors_unlocked(&state_dir, cursors)
214}
215
216fn write_replica_cursors_unlocked(state_dir: &Path, cursors: Vec<ReplicaCursor>) -> io::Result<()> {
217    let file = CursorFile {
218        format_version: SYNC_METADATA_FORMAT_VERSION,
219        cursors,
220    };
221    atomic_replace_json(state_dir, REPLICA_CURSORS_FILE, &file)
222}
223
224pub(crate) fn replace_replica_cursors_unlocked(
225    data_dir: &Path,
226    mut cursors: Vec<ReplicaCursor>,
227) -> io::Result<()> {
228    validate_cursors_for_write(&cursors)?;
229    cursors.sort_by(|a, b| a.replica_id.cmp(&b.replica_id));
230    let state_dir = sync_state_dir(data_dir);
231    create_data_dir_secure(&state_dir)?;
232    write_replica_cursors_unlocked(&state_dir, cursors)
233}
234
235pub fn upsert_replica_cursor(data_dir: &Path, cursor: ReplicaCursor) -> io::Result<()> {
236    validate_cursor_for_write(&cursor)?;
237    let state_dir = sync_state_dir(data_dir);
238    create_data_dir_secure(&state_dir)?;
239    let _lock = acquire_cursor_lock(&state_dir)?;
240    validate_active_cursor_against_retained_tail(data_dir, &cursor)?;
241    let mut cursors = read_replica_cursors_unlocked(data_dir)?;
242    if let Some(existing) = cursors
243        .iter_mut()
244        .find(|existing| existing.replica_id == cursor.replica_id)
245    {
246        *existing = cursor;
247    } else {
248        cursors.push(cursor);
249    }
250    validate_cursors_for_write(&cursors)?;
251    cursors.sort_by(|a, b| a.replica_id.cmp(&b.replica_id));
252    write_replica_cursors_unlocked(&state_dir, cursors)
253}
254
255pub fn register_bootstrap_cursor(
256    data_dir: &Path,
257    replica_id: &str,
258    snapshot_lsn: u64,
259    required_through_lsn: u64,
260) -> io::Result<ReplicaCursor> {
261    if required_through_lsn < snapshot_lsn {
262        return Err(invalid_input(format!(
263            "remote LSN {required_through_lsn} is behind snapshot LSN {snapshot_lsn}"
264        )));
265    }
266    validate_replica_id(replica_id).map_err(invalid_input)?;
267    let cursor = ReplicaCursor::active(replica_id, snapshot_lsn);
268    validate_cursor_for_write(&cursor)?;
269
270    let state_dir = sync_state_dir(data_dir);
271    create_data_dir_secure(&state_dir)?;
272    let _lock = acquire_cursor_lock(&state_dir)?;
273
274    let identity = read_identity(data_dir)?;
275    crate::validate_retained_tail_available(
276        &crate::retained_segments_dir(data_dir),
277        identity.segment_identity(),
278        snapshot_lsn,
279        required_through_lsn,
280    )
281    .map_err(|err| {
282        if matches!(err.kind(), io::ErrorKind::InvalidData)
283            && (err.to_string().contains("gap") || err.to_string().contains("missing"))
284        {
285            io::Error::new(
286                io::ErrorKind::InvalidInput,
287                format!("retained history is unavailable; rebootstrap required: {err}"),
288            )
289        } else {
290            err
291        }
292    })?;
293
294    let mut cursors = read_replica_cursors_unlocked(data_dir)?;
295    if cursors
296        .iter()
297        .any(|existing| existing.replica_id == replica_id && existing.active)
298    {
299        return Err(io::Error::new(
300            io::ErrorKind::AlreadyExists,
301            "replica cursor is already active; retire it before bootstrap",
302        ));
303    }
304    if let Some(existing) = cursors
305        .iter_mut()
306        .find(|existing| existing.replica_id == replica_id)
307    {
308        *existing = cursor.clone();
309    } else {
310        cursors.push(cursor.clone());
311    }
312    validate_cursors_for_write(&cursors)?;
313    cursors.sort_by(|a, b| a.replica_id.cmp(&b.replica_id));
314    write_replica_cursors_unlocked(&state_dir, cursors)?;
315    Ok(cursor)
316}
317
318pub(crate) fn with_cursor_metadata_lock<T>(
319    data_dir: &Path,
320    f: impl FnOnce() -> io::Result<T>,
321) -> io::Result<T> {
322    let state_dir = sync_state_dir(data_dir);
323    create_data_dir_secure(&state_dir)?;
324    let _lock = acquire_cursor_lock(&state_dir)?;
325    f()
326}
327
328pub fn retire_replica_cursor(
329    data_dir: &Path,
330    replica_id: &str,
331    updated_unix_secs: u64,
332) -> io::Result<()> {
333    validate_replica_id(replica_id).map_err(invalid_input)?;
334    let state_dir = sync_state_dir(data_dir);
335    create_data_dir_secure(&state_dir)?;
336    let _lock = acquire_cursor_lock(&state_dir)?;
337    let mut cursors = read_replica_cursors_unlocked(data_dir)?;
338    let Some(cursor) = cursors
339        .iter_mut()
340        .find(|cursor| cursor.replica_id == replica_id)
341    else {
342        return Err(io::Error::new(
343            io::ErrorKind::NotFound,
344            "replica cursor not found",
345        ));
346    };
347    cursor.active = false;
348    cursor.updated_unix_secs = updated_unix_secs;
349    write_replica_cursors_unlocked(&state_dir, cursors)
350}
351
352/// Return the smallest next-required LSN across active replica cursors.
353///
354/// This is an advisory snapshot. Do not use it by itself as authority to delete
355/// retained history because cursor publication can race an unlocked caller.
356/// Use `prune_retained_segments_for_cursors` for deletion; it holds the cursor
357/// metadata lock from cursor-floor read through segment removal.
358pub fn minimum_retained_lsn(data_dir: &Path) -> io::Result<Option<u64>> {
359    let mut min_lsn: Option<u64> = None;
360    for cursor in read_replica_cursors(data_dir)? {
361        if !cursor.active {
362            continue;
363        }
364        let next_lsn = cursor.next_required_lsn()?;
365        min_lsn = Some(match min_lsn {
366            Some(current) => current.min(next_lsn),
367            None => next_lsn,
368        });
369    }
370    Ok(min_lsn)
371}
372
373fn write_new_identity_file(state_dir: &Path, file: &IdentitySnapshot) -> io::Result<()> {
374    let final_path = state_dir.join(IDENTITY_FILE);
375    let temp_path = temp_path(state_dir, IDENTITY_FILE);
376    let bytes = serde_json::to_vec_pretty(file).map_err(io::Error::other)?;
377    let result: io::Result<()> = (|| {
378        let mut temp = OpenOptions::new()
379            .write(true)
380            .create_new(true)
381            .open(&temp_path)?;
382        temp.write_all(&bytes)?;
383        temp.sync_all()?;
384        drop(temp);
385        fs::hard_link(&temp_path, &final_path)?;
386        fsync_dir(state_dir)?;
387        let _ = fs::remove_file(&temp_path);
388        let _ = fsync_dir(state_dir);
389        Ok(())
390    })();
391    if result.is_err() {
392        let _ = fs::remove_file(&temp_path);
393    }
394    result
395}
396
397pub(crate) fn atomic_replace_json<T: Serialize>(
398    state_dir: &Path,
399    file_name: &str,
400    value: &T,
401) -> io::Result<()> {
402    let final_path = state_dir.join(file_name);
403    let temp_path = temp_path(state_dir, file_name);
404    let bytes = serde_json::to_vec_pretty(value).map_err(io::Error::other)?;
405    let result: io::Result<()> = (|| {
406        let mut temp = OpenOptions::new()
407            .write(true)
408            .create_new(true)
409            .open(&temp_path)?;
410        temp.write_all(&bytes)?;
411        temp.sync_all()?;
412        drop(temp);
413        fs::rename(&temp_path, &final_path)?;
414        fsync_dir(state_dir)?;
415        Ok(())
416    })();
417    if result.is_err() {
418        let _ = fs::remove_file(&temp_path);
419    }
420    result
421}
422
423struct CursorLock {
424    path: PathBuf,
425    dir: PathBuf,
426}
427
428impl Drop for CursorLock {
429    fn drop(&mut self) {
430        let _ = fs::remove_file(&self.path);
431        let _ = fsync_dir(&self.dir);
432    }
433}
434
435fn acquire_cursor_lock(state_dir: &Path) -> io::Result<CursorLock> {
436    let path = state_dir.join(CURSOR_LOCK_FILE);
437    let start = Instant::now();
438    loop {
439        match OpenOptions::new().write(true).create_new(true).open(&path) {
440            Ok(mut file) => {
441                let result: io::Result<()> = (|| {
442                    let owner = CursorLockFile {
443                        format_version: SYNC_METADATA_FORMAT_VERSION,
444                        owner_pid: std::process::id(),
445                        created_unix_secs: now_unix_secs(),
446                    };
447                    let bytes = serde_json::to_vec_pretty(&owner).map_err(io::Error::other)?;
448                    file.write_all(&bytes)?;
449                    file.write_all(b"\n")?;
450                    file.sync_all()?;
451                    fsync_dir(state_dir)
452                })();
453                if let Err(err) = result {
454                    let _ = fs::remove_file(&path);
455                    let _ = fsync_dir(state_dir);
456                    return Err(err);
457                }
458                return Ok(CursorLock {
459                    path,
460                    dir: state_dir.to_path_buf(),
461                });
462            }
463            Err(err) if err.kind() == io::ErrorKind::AlreadyExists => {
464                if reclaim_stale_cursor_lock(&path, state_dir)? {
465                    continue;
466                }
467                if start.elapsed() >= CURSOR_LOCK_TIMEOUT {
468                    return Err(io::Error::new(
469                        io::ErrorKind::WouldBlock,
470                        "timed out waiting for sync cursor metadata lock",
471                    ));
472                }
473                std::thread::sleep(CURSOR_LOCK_RETRY);
474            }
475            Err(err) => return Err(err),
476        }
477    }
478}
479
480fn reclaim_stale_cursor_lock(path: &Path, state_dir: &Path) -> io::Result<bool> {
481    let bytes = match fs::read(path) {
482        Ok(bytes) => bytes,
483        Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(true),
484        Err(err) => return Err(err),
485    };
486    let stale = match serde_json::from_slice::<CursorLockFile>(&bytes) {
487        Ok(owner) => cursor_lock_owner_stale(&owner),
488        Err(_) => lock_file_age(path)?
489            .map(|age| age >= CURSOR_LOCK_STALE_AFTER)
490            .unwrap_or(false),
491    };
492    if !stale {
493        return Ok(false);
494    }
495    match fs::remove_file(path) {
496        Ok(()) => {
497            fsync_dir(state_dir)?;
498            Ok(true)
499        }
500        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(true),
501        Err(err) => Err(err),
502    }
503}
504
505fn cursor_lock_owner_stale(owner: &CursorLockFile) -> bool {
506    if owner.format_version != SYNC_METADATA_FORMAT_VERSION {
507        return lock_owner_age(owner)
508            .map(|age| age >= CURSOR_LOCK_STALE_AFTER)
509            .unwrap_or(false);
510    }
511    if owner.owner_pid == 0 {
512        return true;
513    }
514    if !owner_process_alive(owner.owner_pid) {
515        return true;
516    }
517    lock_owner_stale_without_process_probe(owner)
518}
519
520#[cfg(unix)]
521fn lock_owner_stale_without_process_probe(_owner: &CursorLockFile) -> bool {
522    false
523}
524
525#[cfg(not(unix))]
526fn lock_owner_stale_without_process_probe(owner: &CursorLockFile) -> bool {
527    lock_owner_age(owner)
528        .map(|age| age >= CURSOR_LOCK_STALE_AFTER)
529        .unwrap_or(false)
530}
531
532#[cfg(unix)]
533fn owner_process_alive(pid: u32) -> bool {
534    if pid > libc::pid_t::MAX as u32 {
535        return false;
536    }
537    let rc = unsafe { libc::kill(pid as libc::pid_t, 0) };
538    if rc == 0 {
539        return true;
540    }
541    let err = io::Error::last_os_error();
542    err.raw_os_error() != Some(libc::ESRCH)
543}
544
545#[cfg(not(unix))]
546fn owner_process_alive(_pid: u32) -> bool {
547    true
548}
549
550fn lock_owner_age(owner: &CursorLockFile) -> Option<Duration> {
551    Some(Duration::from_secs(
552        now_unix_secs().checked_sub(owner.created_unix_secs)?,
553    ))
554}
555
556fn lock_file_age(path: &Path) -> io::Result<Option<Duration>> {
557    let modified = fs::metadata(path)?.modified()?;
558    Ok(SystemTime::now().duration_since(modified).ok())
559}
560
561fn validate_identity(identity: DatabaseIdentity) -> io::Result<()> {
562    if identity.primary_generation == 0 {
563        return Err(invalid_data("primary generation must be non-zero"));
564    }
565    if identity.database_id == [0; 16] {
566        return Err(invalid_data("database id must be non-zero"));
567    }
568    Ok(())
569}
570
571fn validate_cursor_file(file: &CursorFile) -> io::Result<()> {
572    if file.format_version != SYNC_METADATA_FORMAT_VERSION {
573        return Err(invalid_data(format!(
574            "unsupported sync cursor format {}",
575            file.format_version
576        )));
577    }
578    validate_cursors(&file.cursors, io::ErrorKind::InvalidData)
579}
580
581fn validate_cursors_for_write(cursors: &[ReplicaCursor]) -> io::Result<()> {
582    validate_cursors(cursors, io::ErrorKind::InvalidInput)
583}
584
585fn validate_cursor_for_write(cursor: &ReplicaCursor) -> io::Result<()> {
586    validate_replica_id(&cursor.replica_id).map_err(invalid_input)?;
587    cursor
588        .next_required_lsn()
589        .map_err(|err| invalid_input(err.to_string()))?;
590    Ok(())
591}
592
593fn validate_active_cursors_against_retained_tail(
594    data_dir: &Path,
595    cursors: &[ReplicaCursor],
596) -> io::Result<()> {
597    for cursor in cursors {
598        validate_active_cursor_against_retained_tail(data_dir, cursor)?;
599    }
600    Ok(())
601}
602
603fn validate_active_cursor_against_retained_tail(
604    data_dir: &Path,
605    cursor: &ReplicaCursor,
606) -> io::Result<()> {
607    if !cursor.active {
608        return Ok(());
609    }
610    let identity = match read_identity(data_dir) {
611        Ok(identity) => identity,
612        Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(()),
613        Err(err) => return Err(err),
614    };
615    let segment_dir = crate::retained_segments_dir(data_dir);
616    if crate::list_segment_files(&segment_dir)?.is_empty() {
617        return Ok(());
618    }
619    match read_units_since(
620        &segment_dir,
621        identity.segment_identity(),
622        cursor.applied_lsn,
623        1,
624    ) {
625        Ok(_) => Ok(()),
626        Err(err) if err.to_string().contains("gap") => Err(io::Error::new(
627            io::ErrorKind::InvalidInput,
628            "replica cursor is behind retained history; rebootstrap required",
629        )),
630        Err(err) => Err(err),
631    }
632}
633
634fn validate_cursors(cursors: &[ReplicaCursor], kind: io::ErrorKind) -> io::Result<()> {
635    let mut seen = HashSet::new();
636    for cursor in cursors {
637        validate_replica_id(&cursor.replica_id)
638            .map_err(|err| io::Error::new(kind, err.to_string()))?;
639        cursor
640            .next_required_lsn()
641            .map_err(|err| io::Error::new(kind, err.to_string()))?;
642        if !seen.insert(cursor.replica_id.as_str()) {
643            return Err(io::Error::new(kind, "duplicate replica cursor id"));
644        }
645    }
646    Ok(())
647}
648
649fn validate_replica_id(replica_id: &str) -> Result<(), &'static str> {
650    if replica_id.is_empty() {
651        return Err("replica id must be non-empty");
652    }
653    if replica_id.len() > 128 {
654        return Err("replica id must be at most 128 bytes");
655    }
656    if !replica_id
657        .bytes()
658        .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b':'))
659    {
660        return Err("replica id contains unsupported characters");
661    }
662    Ok(())
663}
664
665fn generate_database_id() -> io::Result<[u8; 16]> {
666    let mut id = [0u8; 16];
667    getrandom::fill(&mut id).map_err(|err| {
668        io::Error::other(format!(
669            "secure OS random source failed while creating sync database id: {err}"
670        ))
671    })?;
672    if id == [0; 16] {
673        return Err(io::Error::other(
674            "secure OS random source returned an all-zero sync database id",
675        ));
676    }
677    Ok(id)
678}
679
680fn encode_hex_16(bytes: [u8; 16]) -> String {
681    let mut out = String::with_capacity(32);
682    for byte in bytes {
683        out.push(hex_digit(byte >> 4));
684        out.push(hex_digit(byte & 0x0f));
685    }
686    out
687}
688
689fn decode_hex_16(s: &str) -> io::Result<[u8; 16]> {
690    if s.len() != 32 {
691        return Err(invalid_data("database id must be 32 hex characters"));
692    }
693    let mut out = [0u8; 16];
694    let bytes = s.as_bytes();
695    for i in 0..16 {
696        let hi = hex_value(bytes[i * 2])?;
697        let lo = hex_value(bytes[i * 2 + 1])?;
698        out[i] = (hi << 4) | lo;
699    }
700    Ok(out)
701}
702
703fn hex_digit(nibble: u8) -> char {
704    match nibble {
705        0..=9 => (b'0' + nibble) as char,
706        10..=15 => (b'a' + (nibble - 10)) as char,
707        _ => unreachable!("nibble is always four bits"),
708    }
709}
710
711fn hex_value(byte: u8) -> io::Result<u8> {
712    match byte {
713        b'0'..=b'9' => Ok(byte - b'0'),
714        b'a'..=b'f' => Ok(byte - b'a' + 10),
715        b'A'..=b'F' => Ok(byte - b'A' + 10),
716        _ => Err(invalid_data("database id contains non-hex byte")),
717    }
718}
719
720fn temp_path(dir: &Path, file_name: &str) -> PathBuf {
721    let counter = TEMP_COUNTER.fetch_add(1, Ordering::Relaxed);
722    dir.join(format!(
723        ".{file_name}.tmp.{}.{}.{}",
724        std::process::id(),
725        now_nanos(),
726        counter
727    ))
728}
729
730pub(crate) fn now_unix_secs() -> u64 {
731    SystemTime::now()
732        .duration_since(UNIX_EPOCH)
733        .map(|duration| duration.as_secs())
734        .unwrap_or(0)
735}
736
737fn now_nanos() -> u128 {
738    SystemTime::now()
739        .duration_since(UNIX_EPOCH)
740        .map(|duration| duration.as_nanos())
741        .unwrap_or(0)
742}
743
744#[cfg(unix)]
745fn fsync_dir(dir: &Path) -> io::Result<()> {
746    File::open(dir)?.sync_all()
747}
748
749#[cfg(not(unix))]
750fn fsync_dir(_dir: &Path) -> io::Result<()> {
751    Ok(())
752}
753
754fn invalid_input(message: impl ToString) -> io::Error {
755    io::Error::new(io::ErrorKind::InvalidInput, message.to_string())
756}
757
758fn invalid_data(message: impl ToString) -> io::Error {
759    io::Error::new(io::ErrorKind::InvalidData, message.to_string())
760}
761
762#[cfg(test)]
763mod tests {
764    use super::*;
765    use std::sync::{Arc, Barrier};
766
767    #[test]
768    fn identity_is_created_once_and_reused() {
769        let dir = tempfile::tempdir().unwrap();
770        let first = open_or_create_identity(dir.path()).unwrap();
771        let second = open_or_create_identity(dir.path()).unwrap();
772
773        assert_eq!(first, second);
774        assert_ne!(first.database_id, [0; 16]);
775        assert_eq!(first.primary_generation, 1);
776        assert_eq!(read_identity(dir.path()).unwrap(), first);
777        assert_eq!(
778            first.segment_identity(),
779            SegmentIdentity::current(first.database_id, 1)
780        );
781    }
782
783    #[test]
784    fn concurrent_identity_creation_uses_one_winner() {
785        let dir = tempfile::tempdir().unwrap();
786        let barrier = Arc::new(Barrier::new(8));
787        let mut handles = Vec::new();
788
789        for _ in 0..8 {
790            let data_dir = dir.path().to_path_buf();
791            let barrier = Arc::clone(&barrier);
792            handles.push(std::thread::spawn(move || {
793                barrier.wait();
794                open_or_create_identity(&data_dir)
795            }));
796        }
797
798        let mut identities = Vec::new();
799        for handle in handles {
800            identities.push(handle.join().unwrap().unwrap());
801        }
802
803        for identity in &identities[1..] {
804            assert_eq!(*identity, identities[0]);
805        }
806    }
807
808    #[test]
809    fn corrupt_identity_fails_closed() {
810        let dir = tempfile::tempdir().unwrap();
811        let state_dir = sync_state_dir(dir.path());
812        create_data_dir_secure(&state_dir).unwrap();
813        fs::write(
814            state_dir.join(IDENTITY_FILE),
815            br#"{"format_version":1,"database_id":"not-hex","primary_generation":1,"created_unix_secs":1}"#,
816        )
817        .unwrap();
818
819        let err = read_identity(dir.path()).unwrap_err();
820        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
821    }
822
823    #[test]
824    fn cursors_round_trip_and_minimum_retained_lsn_uses_active_replicas() {
825        let dir = tempfile::tempdir().unwrap();
826        write_replica_cursors(
827            dir.path(),
828            vec![
829                ReplicaCursor {
830                    replica_id: "replica-b".into(),
831                    applied_lsn: 40,
832                    updated_unix_secs: 2,
833                    active: true,
834                },
835                ReplicaCursor {
836                    replica_id: "replica-a".into(),
837                    applied_lsn: 10,
838                    updated_unix_secs: 1,
839                    active: true,
840                },
841                ReplicaCursor {
842                    replica_id: "replica-retired".into(),
843                    applied_lsn: 1,
844                    updated_unix_secs: 3,
845                    active: false,
846                },
847            ],
848        )
849        .unwrap();
850
851        let cursors = read_replica_cursors(dir.path()).unwrap();
852        assert_eq!(cursors[0].replica_id, "replica-a");
853        assert_eq!(cursors[1].replica_id, "replica-b");
854        assert_eq!(minimum_retained_lsn(dir.path()).unwrap(), Some(11));
855    }
856
857    #[test]
858    fn upsert_and_retire_cursor_update_minimum_lsn() {
859        let dir = tempfile::tempdir().unwrap();
860        upsert_replica_cursor(dir.path(), ReplicaCursor::active("replica-a", 5)).unwrap();
861        upsert_replica_cursor(dir.path(), ReplicaCursor::active("replica-b", 9)).unwrap();
862        assert_eq!(minimum_retained_lsn(dir.path()).unwrap(), Some(6));
863
864        upsert_replica_cursor(dir.path(), ReplicaCursor::active("replica-a", 12)).unwrap();
865        assert_eq!(minimum_retained_lsn(dir.path()).unwrap(), Some(10));
866
867        retire_replica_cursor(dir.path(), "replica-b", 100).unwrap();
868        assert_eq!(minimum_retained_lsn(dir.path()).unwrap(), Some(13));
869    }
870
871    #[test]
872    fn concurrent_upserts_keep_all_active_replicas() {
873        let dir = tempfile::tempdir().unwrap();
874        let barrier = Arc::new(Barrier::new(16));
875        let mut handles = Vec::new();
876
877        for i in 0..16 {
878            let data_dir = dir.path().to_path_buf();
879            let barrier = Arc::clone(&barrier);
880            handles.push(std::thread::spawn(move || {
881                barrier.wait();
882                upsert_replica_cursor(
883                    &data_dir,
884                    ReplicaCursor::active(format!("replica-{i:02}"), 100 + i),
885                )
886            }));
887        }
888
889        for handle in handles {
890            handle.join().unwrap().unwrap();
891        }
892
893        let cursors = read_replica_cursors(dir.path()).unwrap();
894        assert_eq!(cursors.len(), 16);
895        assert_eq!(minimum_retained_lsn(dir.path()).unwrap(), Some(101));
896    }
897
898    #[test]
899    fn stale_cursor_lock_is_reclaimed() {
900        let dir = tempfile::tempdir().unwrap();
901        let state_dir = sync_state_dir(dir.path());
902        create_data_dir_secure(&state_dir).unwrap();
903        let stale_owner = CursorLockFile {
904            format_version: SYNC_METADATA_FORMAT_VERSION,
905            owner_pid: 0,
906            created_unix_secs: now_unix_secs(),
907        };
908        fs::write(
909            state_dir.join(CURSOR_LOCK_FILE),
910            serde_json::to_vec_pretty(&stale_owner).unwrap(),
911        )
912        .unwrap();
913        fsync_dir(&state_dir).unwrap();
914
915        upsert_replica_cursor(dir.path(), ReplicaCursor::active("replica-a", 7)).unwrap();
916
917        let cursors = read_replica_cursors(dir.path()).unwrap();
918        assert_eq!(cursors.len(), 1);
919        assert_eq!(cursors[0].replica_id, "replica-a");
920        assert!(!state_dir.join(CURSOR_LOCK_FILE).exists());
921    }
922
923    #[test]
924    fn cursor_validation_rejects_duplicate_bad_and_overflowing_ids() {
925        let dir = tempfile::tempdir().unwrap();
926        let err = write_replica_cursors(
927            dir.path(),
928            vec![
929                ReplicaCursor::active("replica-a", 1),
930                ReplicaCursor::active("replica-a", 2),
931            ],
932        )
933        .unwrap_err();
934        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
935
936        let err =
937            upsert_replica_cursor(dir.path(), ReplicaCursor::active("../bad", 1)).unwrap_err();
938        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
939
940        let err = upsert_replica_cursor(dir.path(), ReplicaCursor::active("replica-c", u64::MAX))
941            .unwrap_err();
942        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
943    }
944
945    #[test]
946    fn corrupt_cursor_file_fails_closed() {
947        let dir = tempfile::tempdir().unwrap();
948        let state_dir = sync_state_dir(dir.path());
949        create_data_dir_secure(&state_dir).unwrap();
950        fs::write(
951            state_dir.join(REPLICA_CURSORS_FILE),
952            br#"{"format_version":999,"cursors":[]}"#,
953        )
954        .unwrap();
955
956        let err = read_replica_cursors(dir.path()).unwrap_err();
957        assert_eq!(err.kind(), io::ErrorKind::InvalidData);
958    }
959}
powdb_sync/metadata.rs

powdb_sync/
metadata.rs
