Phase 6e: expose refs as FUSE files #78
+116
-3
@@ -42,10 +42,43 @@ mod sync;
|
||||
mod zfs;
|
||||
|
||||
use cluster::blob::{BlobId, BlobStore};
|
||||
use cluster::refs::RefStore;
|
||||
use cluster::snapshot::SnapshotStore;
|
||||
use cluster::tags::TagStore;
|
||||
|
||||
const TTL: Duration = Duration::from_secs(1);
|
||||
|
||||
fn hex64(bytes: &[u8; 32]) -> String {
|
||||
let mut s = String::with_capacity(64);
|
||||
for b in bytes {
|
||||
s.push_str(&format!("{b:02x}"));
|
||||
}
|
||||
s
|
||||
}
|
||||
|
||||
fn decode_hex_key(name: &str) -> Option<[u8; 32]> {
|
||||
if name.len() != 64 {
|
||||
return None;
|
||||
}
|
||||
let bytes = name.as_bytes();
|
||||
let mut out = [0u8; 32];
|
||||
for i in 0..32 {
|
||||
let hi = decode_nibble(bytes[i * 2])?;
|
||||
let lo = decode_nibble(bytes[i * 2 + 1])?;
|
||||
out[i] = (hi << 4) | lo;
|
||||
}
|
||||
Some(out)
|
||||
}
|
||||
|
||||
fn decode_nibble(b: u8) -> Option<u8> {
|
||||
match b {
|
||||
b'0'..=b'9' => Some(b - b'0'),
|
||||
b'a'..=b'f' => Some(b - b'a' + 10),
|
||||
b'A'..=b'F' => Some(b - b'A' + 10),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
// Inode number layout:
|
||||
// * 1 = root ("/")
|
||||
// * 2 = "/blobs" directory
|
||||
@@ -56,6 +89,7 @@ const ROOT_INO: u64 = 1;
|
||||
const BLOBS_DIR_INO: u64 = 2;
|
||||
const SNAPSHOTS_DIR_INO: u64 = 3;
|
||||
const TAGS_DIR_INO: u64 = 4;
|
||||
const REFS_DIR_INO: u64 = 5;
|
||||
// Tag file inodes: 1_000..9_999.
|
||||
const FIRST_TAG_INO: u64 = 1_000;
|
||||
// Snapshot dir inodes: 10_000..99_999 (up to 90k snapshots).
|
||||
@@ -68,6 +102,7 @@ struct ClawFuse {
|
||||
store: BlobStore,
|
||||
snapshots: SnapshotStore,
|
||||
tags: TagStore,
|
||||
refs: RefStore,
|
||||
/// Runtime for async store calls. fuser is sync so we
|
||||
/// block_on inside each callback.
|
||||
runtime: tokio::runtime::Runtime,
|
||||
@@ -90,11 +125,17 @@ struct ClawFuse {
|
||||
}
|
||||
|
||||
impl ClawFuse {
|
||||
fn new(store: BlobStore, snapshots: SnapshotStore, tags: TagStore) -> Result<Self> {
|
||||
fn new(
|
||||
store: BlobStore,
|
||||
snapshots: SnapshotStore,
|
||||
tags: TagStore,
|
||||
refs: RefStore,
|
||||
) -> Result<Self> {
|
||||
Ok(Self {
|
||||
store,
|
||||
snapshots,
|
||||
tags,
|
||||
refs,
|
||||
runtime: tokio::runtime::Builder::new_current_thread()
|
||||
.enable_all()
|
||||
.build()?,
|
||||
@@ -248,6 +289,49 @@ impl Filesystem for ClawFuse {
|
||||
ROOT_INO if name == "tags" => {
|
||||
reply.entry(&TTL, &Self::dir_attr(TAGS_DIR_INO), 0);
|
||||
}
|
||||
ROOT_INO if name == "refs" => {
|
||||
reply.entry(&TTL, &Self::dir_attr(REFS_DIR_INO), 0);
|
||||
}
|
||||
REFS_DIR_INO => {
|
||||
// /refs/<64-hex> → file whose bytes are the assembled
|
||||
// blob that the fingerprint currently points at.
|
||||
let key = match decode_hex_key(name) {
|
||||
Some(k) => k,
|
||||
None => {
|
||||
reply.error(libc::ENOENT);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let value = match self.runtime.block_on(async {
|
||||
if let Ok(Some(s)) = self.refs.get_stamped(&key).await {
|
||||
return Some(s.value);
|
||||
}
|
||||
self.refs.get(&key).await.ok().flatten()
|
||||
}) {
|
||||
Some(v) => v,
|
||||
None => {
|
||||
reply.error(libc::ENOENT);
|
||||
return;
|
||||
}
|
||||
};
|
||||
let blob_id = BlobId::from_bytes(value);
|
||||
let size = match self
|
||||
.runtime
|
||||
.block_on(self.store.load_manifest(&blob_id))
|
||||
.ok()
|
||||
.flatten()
|
||||
{
|
||||
Some(m) => m.total_size,
|
||||
None => {
|
||||
reply.error(libc::ENOENT);
|
||||
return;
|
||||
}
|
||||
};
|
||||
// Ref files piggy-back the blob inode — content is
|
||||
// identical, no reason for separate ino.
|
||||
let ino = self.alloc_blob_ino(&blob_id.to_hex());
|
||||
reply.entry(&TTL, &Self::file_attr(ino, size), 0);
|
||||
}
|
||||
TAGS_DIR_INO => {
|
||||
// Find a tag whose sanitized name matches `name` AND
|
||||
// whose value blob exists on disk.
|
||||
@@ -356,7 +440,7 @@ impl Filesystem for ClawFuse {
|
||||
|
||||
fn getattr(&mut self, _req: &Request, ino: u64, _fh: Option<u64>, reply: ReplyAttr) {
|
||||
match ino {
|
||||
ROOT_INO | BLOBS_DIR_INO | SNAPSHOTS_DIR_INO | TAGS_DIR_INO => {
|
||||
ROOT_INO | BLOBS_DIR_INO | SNAPSHOTS_DIR_INO | TAGS_DIR_INO | REFS_DIR_INO => {
|
||||
reply.attr(&TTL, &Self::dir_attr(ino))
|
||||
}
|
||||
n if (FIRST_SNAPSHOT_INO..FIRST_BLOB_INO).contains(&n) => {
|
||||
@@ -431,6 +515,33 @@ impl Filesystem for ClawFuse {
|
||||
entries.push((BLOBS_DIR_INO, FileType::Directory, "blobs".into()));
|
||||
entries.push((SNAPSHOTS_DIR_INO, FileType::Directory, "snapshots".into()));
|
||||
entries.push((TAGS_DIR_INO, FileType::Directory, "tags".into()));
|
||||
entries.push((REFS_DIR_INO, FileType::Directory, "refs".into()));
|
||||
}
|
||||
REFS_DIR_INO => {
|
||||
entries.push((REFS_DIR_INO, FileType::Directory, ".".into()));
|
||||
entries.push((ROOT_INO, FileType::Directory, "..".into()));
|
||||
let all = match self.runtime.block_on(self.refs.list()) {
|
||||
Ok(v) => v,
|
||||
Err(_) => {
|
||||
reply.error(libc::EIO);
|
||||
return;
|
||||
}
|
||||
};
|
||||
for (key, value) in all {
|
||||
let blob_id = BlobId::from_bytes(value);
|
||||
let exists = self
|
||||
.runtime
|
||||
.block_on(self.store.load_manifest(&blob_id))
|
||||
.ok()
|
||||
.flatten()
|
||||
.is_some();
|
||||
if !exists {
|
||||
continue;
|
||||
}
|
||||
let hex = hex64(&key);
|
||||
let ino = self.alloc_blob_ino(&blob_id.to_hex());
|
||||
entries.push((ino, FileType::RegularFile, hex));
|
||||
}
|
||||
}
|
||||
TAGS_DIR_INO => {
|
||||
entries.push((TAGS_DIR_INO, FileType::Directory, ".".into()));
|
||||
@@ -639,7 +750,9 @@ fn main() -> Result<()> {
|
||||
let tag_dir = cli.data_dir.join("tags-db");
|
||||
let tags = TagStore::open(tag_dir.clone())
|
||||
.with_context(|| format!("opening tag store at {}", tag_dir.display()))?;
|
||||
let fs = ClawFuse::new(store, snapshots, tags)?;
|
||||
let refs = RefStore::open(cli.data_dir.clone())
|
||||
.with_context(|| format!("opening ref store at {}", cli.data_dir.display()))?;
|
||||
let fs = ClawFuse::new(store, snapshots, tags, refs)?;
|
||||
|
||||
let mut opts = vec![
|
||||
MountOption::RO,
|
||||
|
||||
@@ -220,6 +220,85 @@ impl RefStore {
|
||||
.join(format!("{hex}.ref"))
|
||||
}
|
||||
|
||||
/// Phase 6e (2026-07-14): enumerate every `(key, value)` in
|
||||
/// both the legacy `refs/` layer and the stamped `refs-v2/`
|
||||
/// layer. Bounded by (refs on disk × 32 bytes) — cheap even
|
||||
/// with 100k refs.
|
||||
///
|
||||
/// Sorted by key hex for deterministic output.
|
||||
pub async fn list(&self) -> Result<Vec<(RefKey, RefValue)>> {
|
||||
let mut out: std::collections::HashMap<RefKey, RefValue> =
|
||||
std::collections::HashMap::new();
|
||||
// Legacy layer.
|
||||
let legacy_root = self.root.join("refs");
|
||||
if legacy_root.is_dir() {
|
||||
let mut top = tokio::fs::read_dir(&legacy_root).await?;
|
||||
while let Some(bucket) = top.next_entry().await? {
|
||||
if !bucket.file_type().await?.is_dir() {
|
||||
continue;
|
||||
}
|
||||
let mut inner = tokio::fs::read_dir(bucket.path()).await?;
|
||||
while let Some(entry) = inner.next_entry().await? {
|
||||
if !entry.file_type().await?.is_file() {
|
||||
continue;
|
||||
}
|
||||
let name = entry.file_name();
|
||||
let hex = match name.to_str().and_then(|s| s.strip_suffix(".ref")) {
|
||||
Some(h) if h.len() == 64 => h.to_string(),
|
||||
_ => continue,
|
||||
};
|
||||
let key = match decode_hex32(&hex) {
|
||||
Some(k) => k,
|
||||
None => continue,
|
||||
};
|
||||
let bytes = match tokio::fs::read(entry.path()).await {
|
||||
Ok(b) if b.len() == 32 => b,
|
||||
_ => continue,
|
||||
};
|
||||
let mut val = [0u8; 32];
|
||||
val.copy_from_slice(&bytes);
|
||||
out.insert(key, val);
|
||||
}
|
||||
}
|
||||
}
|
||||
// Stamped layer (Phase 3a+). Structure: refs-v2/<hh>/<hex>.svref
|
||||
let stamped_root = self.root.join("refs-v2");
|
||||
if stamped_root.is_dir() {
|
||||
let mut top = tokio::fs::read_dir(&stamped_root).await?;
|
||||
while let Some(bucket) = top.next_entry().await? {
|
||||
if !bucket.file_type().await?.is_dir() {
|
||||
continue;
|
||||
}
|
||||
let mut inner = tokio::fs::read_dir(bucket.path()).await?;
|
||||
while let Some(entry) = inner.next_entry().await? {
|
||||
if !entry.file_type().await?.is_file() {
|
||||
continue;
|
||||
}
|
||||
let name = entry.file_name();
|
||||
let hex = match name.to_str().and_then(|s| s.strip_suffix(".svref")) {
|
||||
Some(h) if h.len() == 64 => h.to_string(),
|
||||
_ => continue,
|
||||
};
|
||||
let key = match decode_hex32(&hex) {
|
||||
Some(k) => k,
|
||||
None => continue,
|
||||
};
|
||||
let bytes = match tokio::fs::read(entry.path()).await {
|
||||
Ok(b) => b,
|
||||
Err(_) => continue,
|
||||
};
|
||||
if let Ok(stamped) = StampedRef::from_bytes(&bytes) {
|
||||
// Stamped wins on conflict — modern write path.
|
||||
out.insert(key, stamped.value);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let mut pairs: Vec<_> = out.into_iter().collect();
|
||||
pairs.sort_by(|a, b| a.0.cmp(&b.0));
|
||||
Ok(pairs)
|
||||
}
|
||||
|
||||
/// Phase 3 (2026-07-13): Lamport-stamped ref lookup. Reads from
|
||||
/// the `refs-v2/` directory (separate namespace from the raw
|
||||
/// `refs/` set) so the two lookup surfaces don't interfere.
|
||||
@@ -313,6 +392,29 @@ fn hex32(bytes: &[u8; 32]) -> String {
|
||||
out
|
||||
}
|
||||
|
||||
fn decode_hex32(s: &str) -> Option<[u8; 32]> {
|
||||
if s.len() != 64 {
|
||||
return None;
|
||||
}
|
||||
let bytes = s.as_bytes();
|
||||
let mut out = [0u8; 32];
|
||||
for i in 0..32 {
|
||||
let hi = decode_nibble(bytes[i * 2])?;
|
||||
let lo = decode_nibble(bytes[i * 2 + 1])?;
|
||||
out[i] = (hi << 4) | lo;
|
||||
}
|
||||
Some(out)
|
||||
}
|
||||
|
||||
fn decode_nibble(b: u8) -> Option<u8> {
|
||||
match b {
|
||||
b'0'..=b'9' => Some(b - b'0'),
|
||||
b'a'..=b'f' => Some(b - b'a' + 10),
|
||||
b'A'..=b'F' => Some(b - b'A' + 10),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
Reference in New Issue
Block a user