Phase 6e: expose refs as FUSE files #78
+116
-3
@@ -42,10 +42,43 @@ mod sync;
|
|||||||
mod zfs;
|
mod zfs;
|
||||||
|
|
||||||
use cluster::blob::{BlobId, BlobStore};
|
use cluster::blob::{BlobId, BlobStore};
|
||||||
|
use cluster::refs::RefStore;
|
||||||
use cluster::snapshot::SnapshotStore;
|
use cluster::snapshot::SnapshotStore;
|
||||||
use cluster::tags::TagStore;
|
use cluster::tags::TagStore;
|
||||||
|
|
||||||
const TTL: Duration = Duration::from_secs(1);
|
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:
|
// Inode number layout:
|
||||||
// * 1 = root ("/")
|
// * 1 = root ("/")
|
||||||
// * 2 = "/blobs" directory
|
// * 2 = "/blobs" directory
|
||||||
@@ -56,6 +89,7 @@ const ROOT_INO: u64 = 1;
|
|||||||
const BLOBS_DIR_INO: u64 = 2;
|
const BLOBS_DIR_INO: u64 = 2;
|
||||||
const SNAPSHOTS_DIR_INO: u64 = 3;
|
const SNAPSHOTS_DIR_INO: u64 = 3;
|
||||||
const TAGS_DIR_INO: u64 = 4;
|
const TAGS_DIR_INO: u64 = 4;
|
||||||
|
const REFS_DIR_INO: u64 = 5;
|
||||||
// Tag file inodes: 1_000..9_999.
|
// Tag file inodes: 1_000..9_999.
|
||||||
const FIRST_TAG_INO: u64 = 1_000;
|
const FIRST_TAG_INO: u64 = 1_000;
|
||||||
// Snapshot dir inodes: 10_000..99_999 (up to 90k snapshots).
|
// Snapshot dir inodes: 10_000..99_999 (up to 90k snapshots).
|
||||||
@@ -68,6 +102,7 @@ struct ClawFuse {
|
|||||||
store: BlobStore,
|
store: BlobStore,
|
||||||
snapshots: SnapshotStore,
|
snapshots: SnapshotStore,
|
||||||
tags: TagStore,
|
tags: TagStore,
|
||||||
|
refs: RefStore,
|
||||||
/// Runtime for async store calls. fuser is sync so we
|
/// Runtime for async store calls. fuser is sync so we
|
||||||
/// block_on inside each callback.
|
/// block_on inside each callback.
|
||||||
runtime: tokio::runtime::Runtime,
|
runtime: tokio::runtime::Runtime,
|
||||||
@@ -90,11 +125,17 @@ struct ClawFuse {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl 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 {
|
Ok(Self {
|
||||||
store,
|
store,
|
||||||
snapshots,
|
snapshots,
|
||||||
tags,
|
tags,
|
||||||
|
refs,
|
||||||
runtime: tokio::runtime::Builder::new_current_thread()
|
runtime: tokio::runtime::Builder::new_current_thread()
|
||||||
.enable_all()
|
.enable_all()
|
||||||
.build()?,
|
.build()?,
|
||||||
@@ -248,6 +289,49 @@ impl Filesystem for ClawFuse {
|
|||||||
ROOT_INO if name == "tags" => {
|
ROOT_INO if name == "tags" => {
|
||||||
reply.entry(&TTL, &Self::dir_attr(TAGS_DIR_INO), 0);
|
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 => {
|
TAGS_DIR_INO => {
|
||||||
// Find a tag whose sanitized name matches `name` AND
|
// Find a tag whose sanitized name matches `name` AND
|
||||||
// whose value blob exists on disk.
|
// 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) {
|
fn getattr(&mut self, _req: &Request, ino: u64, _fh: Option<u64>, reply: ReplyAttr) {
|
||||||
match ino {
|
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))
|
reply.attr(&TTL, &Self::dir_attr(ino))
|
||||||
}
|
}
|
||||||
n if (FIRST_SNAPSHOT_INO..FIRST_BLOB_INO).contains(&n) => {
|
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((BLOBS_DIR_INO, FileType::Directory, "blobs".into()));
|
||||||
entries.push((SNAPSHOTS_DIR_INO, FileType::Directory, "snapshots".into()));
|
entries.push((SNAPSHOTS_DIR_INO, FileType::Directory, "snapshots".into()));
|
||||||
entries.push((TAGS_DIR_INO, FileType::Directory, "tags".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 => {
|
TAGS_DIR_INO => {
|
||||||
entries.push((TAGS_DIR_INO, FileType::Directory, ".".into()));
|
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 tag_dir = cli.data_dir.join("tags-db");
|
||||||
let tags = TagStore::open(tag_dir.clone())
|
let tags = TagStore::open(tag_dir.clone())
|
||||||
.with_context(|| format!("opening tag store at {}", tag_dir.display()))?;
|
.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![
|
let mut opts = vec![
|
||||||
MountOption::RO,
|
MountOption::RO,
|
||||||
|
|||||||
@@ -220,6 +220,85 @@ impl RefStore {
|
|||||||
.join(format!("{hex}.ref"))
|
.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
|
/// Phase 3 (2026-07-13): Lamport-stamped ref lookup. Reads from
|
||||||
/// the `refs-v2/` directory (separate namespace from the raw
|
/// the `refs-v2/` directory (separate namespace from the raw
|
||||||
/// `refs/` set) so the two lookup surfaces don't interfere.
|
/// `refs/` set) so the two lookup surfaces don't interfere.
|
||||||
@@ -313,6 +392,29 @@ fn hex32(bytes: &[u8; 32]) -> String {
|
|||||||
out
|
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)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|||||||
Reference in New Issue
Block a user