Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2f7eabf034 | ||
|
|
7902c2e395 | ||
|
|
3d504ee6b3 | ||
|
|
8407c99ba4 | ||
|
|
088b88b225 | ||
|
|
c3f6b500fc |
@@ -32,7 +32,7 @@ install-systemd:
|
||||
install-dashboard:
|
||||
cd dashboard && npm ci && npm run build
|
||||
install -dm755 $(INSTALL_STATIC)
|
||||
cp -r dashboard/dist/. $(INSTALL_STATIC)/
|
||||
cp -r claw-store/static/. $(INSTALL_STATIC)/
|
||||
@echo "Dashboard installed to $(INSTALL_STATIC)"
|
||||
|
||||
## Install node-specific config (NODE=architect|tank)
|
||||
|
||||
@@ -128,6 +128,7 @@ mod tests {
|
||||
peer: None,
|
||||
cluster: None,
|
||||
api_token: None,
|
||||
aggregator: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -21,8 +21,14 @@ pub fn write_cargo_config(warm_path: &Path, hot_target_path: &Path) -> Result<()
|
||||
String::new()
|
||||
};
|
||||
|
||||
// Drop any leading copies of our own marker comment before stripping the
|
||||
// [build] section — it sits above the [build] header, outside the range
|
||||
// strip_section tracks, so without this it would survive every
|
||||
// regenerate cycle and duplicate one more time.
|
||||
let existing = strip_leading_marker(&existing);
|
||||
|
||||
// Remove old [build] block (from its header to the next section or EOF).
|
||||
let stripped = strip_section(&existing, "build");
|
||||
let stripped = strip_section(existing, "build");
|
||||
|
||||
let new_content = format!(
|
||||
"# claw-store managed — do not edit manually\n\
|
||||
@@ -37,6 +43,21 @@ pub fn write_cargo_config(warm_path: &Path, hot_target_path: &Path) -> Result<()
|
||||
.with_context(|| format!("writing {}", config_path.display()))
|
||||
}
|
||||
|
||||
const MANAGED_MARKER: &str = "# claw-store managed — do not edit manually";
|
||||
|
||||
/// Strip leading copies of `MANAGED_MARKER`, one per line, from the start of `src`.
|
||||
fn strip_leading_marker(src: &str) -> &str {
|
||||
let mut rest = src;
|
||||
while let Some(line_end) = rest.find('\n') {
|
||||
if rest[..line_end].trim() == MANAGED_MARKER {
|
||||
rest = &rest[line_end + 1..];
|
||||
} else {
|
||||
break;
|
||||
}
|
||||
}
|
||||
rest
|
||||
}
|
||||
|
||||
/// Remove a TOML section `[name]` and all its key=value lines from `src`,
|
||||
/// stopping at the next `[section]` header or EOF.
|
||||
fn strip_section(src: &str, name: &str) -> String {
|
||||
@@ -102,6 +123,46 @@ mod tests {
|
||||
assert!(verify_cargo_config(&warm, &hot).unwrap());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_write_cargo_config_idempotent_no_duplicate_marker() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let warm = dir.path().join("proj");
|
||||
std::fs::create_dir_all(&warm).unwrap();
|
||||
let hot: std::path::PathBuf = "/hot/targets/proj".into();
|
||||
|
||||
write_cargo_config(&warm, &hot).unwrap();
|
||||
write_cargo_config(&warm, &hot).unwrap();
|
||||
write_cargo_config(&warm, &hot).unwrap();
|
||||
|
||||
let content = std::fs::read_to_string(warm.join(".cargo/config.toml")).unwrap();
|
||||
assert_eq!(content.matches("claw-store managed").count(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_write_cargo_config_heals_existing_duplicate_marker() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let warm = dir.path().join("proj");
|
||||
let cargo_dir = warm.join(".cargo");
|
||||
std::fs::create_dir_all(&cargo_dir).unwrap();
|
||||
std::fs::write(
|
||||
cargo_dir.join("config.toml"),
|
||||
"# claw-store managed — do not edit manually\n\
|
||||
[build]\n\
|
||||
target-dir = \"/hot/targets/proj\"\n\
|
||||
# claw-store managed — do not edit manually\n\
|
||||
[env]\n\
|
||||
FOO = \"bar\"\n",
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
let hot: std::path::PathBuf = "/hot/targets/proj".into();
|
||||
write_cargo_config(&warm, &hot).unwrap();
|
||||
|
||||
let content = std::fs::read_to_string(cargo_dir.join("config.toml")).unwrap();
|
||||
assert_eq!(content.matches("claw-store managed").count(), 1);
|
||||
assert!(content.contains("FOO = \"bar\""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_verify_cargo_config_detects_missing() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
|
||||
@@ -23,6 +23,7 @@ pub mod metrics;
|
||||
pub mod prom;
|
||||
pub mod ref_tracking;
|
||||
pub mod refs;
|
||||
pub mod repo_ensure;
|
||||
pub mod rpc;
|
||||
pub mod services;
|
||||
pub mod snapshot;
|
||||
|
||||
@@ -0,0 +1,335 @@
|
||||
//! Phase 9 R1a: peer-side git materialization.
|
||||
//!
|
||||
//! Two RPCs give the aggregator (or any authorized client) a way to
|
||||
//! ensure a `(url, git_ref)` pair is checked out on this node under a
|
||||
//! caller-provided workspace namespace, and to release it later.
|
||||
//!
|
||||
//! Nothing here fans out to peers — the aggregator layer is
|
||||
//! responsible for calling every node. This module is intentionally
|
||||
//! narrow: one node, one path derivation, one shallow clone.
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::Duration;
|
||||
use tokio::process::Command;
|
||||
use tokio::time::timeout;
|
||||
|
||||
/// Cap on how long a clone can take. Shallow clones of even a large
|
||||
/// repo over LAN complete in seconds; anything past 5 minutes is a
|
||||
/// stuck network or a pathologically large tree.
|
||||
const CLONE_TIMEOUT: Duration = Duration::from_secs(300);
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct RepoEnsureRequest {
|
||||
pub url: String,
|
||||
pub git_ref: String,
|
||||
pub workspace: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct RepoEnsureReply {
|
||||
pub path: String,
|
||||
pub head_sha: String,
|
||||
/// True when the checkout was already present with a valid `.git`
|
||||
/// and no reclone was needed.
|
||||
pub cached: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct RepoReleaseRequest {
|
||||
pub url: String,
|
||||
pub git_ref: String,
|
||||
pub workspace: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
|
||||
pub struct RepoReleaseReply {
|
||||
/// True when the path existed and was removed. False when nothing
|
||||
/// was on disk to begin with (still success from the caller's POV).
|
||||
pub removed: bool,
|
||||
}
|
||||
|
||||
/// Sanitize an untrusted path component so it can never escape its
|
||||
/// parent. Replaces `/`, `\`, `..`, control characters, and leading
|
||||
/// dots. The result is always non-empty and safe to `join` under a
|
||||
/// known root.
|
||||
fn sanitize_component(input: &str) -> String {
|
||||
if input.is_empty() {
|
||||
return "_".to_string();
|
||||
}
|
||||
let mut out = String::with_capacity(input.len());
|
||||
for ch in input.chars() {
|
||||
let mapped = match ch {
|
||||
'/' | '\\' | ':' | '\0' => '_',
|
||||
c if c.is_control() => '_',
|
||||
c => c,
|
||||
};
|
||||
out.push(mapped);
|
||||
}
|
||||
// Reject traversal — after char-mapping we could still have "..".
|
||||
// Collapse any run of dots at either end to `_` prefix.
|
||||
let trimmed = out.trim_matches('.');
|
||||
if trimmed.is_empty() {
|
||||
return "_".to_string();
|
||||
}
|
||||
// Replace embedded ".." segments defensively.
|
||||
trimmed.replace("..", "__")
|
||||
}
|
||||
|
||||
/// Deterministic on-disk path for a `(workspace, url, git_ref)` triple.
|
||||
/// The URL is hashed so we don't leak credentials or full URLs into
|
||||
/// directory names; the ref is sanitized so branch names with slashes
|
||||
/// (`feature/x`) don't create nested dirs.
|
||||
pub fn derive_path(repo_root: &Path, workspace: &str, url: &str, git_ref: &str) -> PathBuf {
|
||||
let ws = sanitize_component(workspace);
|
||||
let ref_slug = sanitize_component(git_ref);
|
||||
let hex = blake3::hash(url.as_bytes()).to_hex();
|
||||
let leaf = format!("{}-{}", &hex.as_str()[..16], ref_slug);
|
||||
repo_root.join(ws).join(leaf)
|
||||
}
|
||||
|
||||
/// Run `git -C <path> rev-parse HEAD` and return the resulting sha
|
||||
/// as a lowercase hex string. `Err` on any failure (including path
|
||||
/// not being a git repo).
|
||||
async fn read_head_sha(path: &Path) -> Result<String> {
|
||||
let output = Command::new("git")
|
||||
.arg("-C")
|
||||
.arg(path)
|
||||
.args(["rev-parse", "HEAD"])
|
||||
.output()
|
||||
.await
|
||||
.context("spawning `git rev-parse`")?;
|
||||
if !output.status.success() {
|
||||
anyhow::bail!(
|
||||
"git rev-parse failed at {}: {}",
|
||||
path.display(),
|
||||
String::from_utf8_lossy(&output.stderr).trim()
|
||||
);
|
||||
}
|
||||
let sha = String::from_utf8(output.stdout)
|
||||
.context("git rev-parse output was not utf-8")?
|
||||
.trim()
|
||||
.to_string();
|
||||
Ok(sha)
|
||||
}
|
||||
|
||||
/// Ensure `(url, git_ref)` is materialized under `repo_root`. When a
|
||||
/// prior checkout already exists and its `.git` resolves cleanly, we
|
||||
/// treat that as a cache hit and return without touching disk. On any
|
||||
/// mismatch we remove-and-reclone.
|
||||
pub async fn ensure_repo(repo_root: &Path, req: &RepoEnsureRequest) -> Result<RepoEnsureReply> {
|
||||
let path = derive_path(repo_root, &req.workspace, &req.url, &req.git_ref);
|
||||
|
||||
if path.join(".git").exists() {
|
||||
if let Ok(sha) = read_head_sha(&path).await {
|
||||
return Ok(RepoEnsureReply {
|
||||
path: path.display().to_string(),
|
||||
head_sha: sha,
|
||||
cached: true,
|
||||
});
|
||||
}
|
||||
// .git present but rev-parse failed — treat as corrupt and
|
||||
// reclone.
|
||||
}
|
||||
|
||||
// Ensure any partial prior attempt is cleared before we clone.
|
||||
if path.exists() {
|
||||
tokio::fs::remove_dir_all(&path)
|
||||
.await
|
||||
.with_context(|| format!("removing stale checkout at {}", path.display()))?;
|
||||
}
|
||||
if let Some(parent) = path.parent() {
|
||||
tokio::fs::create_dir_all(parent)
|
||||
.await
|
||||
.with_context(|| format!("creating parent {}", parent.display()))?;
|
||||
}
|
||||
|
||||
let clone = Command::new("git")
|
||||
.args(["clone", "--depth", "1", "--branch", &req.git_ref])
|
||||
.arg(&req.url)
|
||||
.arg(&path)
|
||||
.output();
|
||||
|
||||
let output = timeout(CLONE_TIMEOUT, clone)
|
||||
.await
|
||||
.with_context(|| format!("git clone timed out after {:?}", CLONE_TIMEOUT))?
|
||||
.context("spawning `git clone`")?;
|
||||
|
||||
if !output.status.success() {
|
||||
// Best-effort cleanup so we don't leave a half-clone behind.
|
||||
let _ = tokio::fs::remove_dir_all(&path).await;
|
||||
anyhow::bail!(
|
||||
"git clone failed for {}@{}: {}",
|
||||
req.url,
|
||||
req.git_ref,
|
||||
String::from_utf8_lossy(&output.stderr).trim()
|
||||
);
|
||||
}
|
||||
|
||||
let sha = read_head_sha(&path).await?;
|
||||
Ok(RepoEnsureReply {
|
||||
path: path.display().to_string(),
|
||||
head_sha: sha,
|
||||
cached: false,
|
||||
})
|
||||
}
|
||||
|
||||
/// Remove the on-disk checkout for `(url, git_ref)` under `repo_root`.
|
||||
/// A missing path is not an error — the caller's precondition
|
||||
/// (nothing at this key) is already satisfied.
|
||||
pub async fn release_repo(
|
||||
repo_root: &Path,
|
||||
req: &RepoReleaseRequest,
|
||||
) -> Result<RepoReleaseReply> {
|
||||
let path = derive_path(repo_root, &req.workspace, &req.url, &req.git_ref);
|
||||
if !path.exists() {
|
||||
return Ok(RepoReleaseReply { removed: false });
|
||||
}
|
||||
tokio::fs::remove_dir_all(&path)
|
||||
.await
|
||||
.with_context(|| format!("removing checkout at {}", path.display()))?;
|
||||
Ok(RepoReleaseReply { removed: true })
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn derive_path_is_deterministic() {
|
||||
let root = PathBuf::from("/tmp/repos");
|
||||
let a = derive_path(&root, "workspace:abc", "https://example.com/x.git", "main");
|
||||
let b = derive_path(&root, "workspace:abc", "https://example.com/x.git", "main");
|
||||
assert_eq!(a, b);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn derive_path_differs_by_ref() {
|
||||
let root = PathBuf::from("/tmp/repos");
|
||||
let a = derive_path(&root, "ws", "https://example.com/x.git", "main");
|
||||
let b = derive_path(&root, "ws", "https://example.com/x.git", "develop");
|
||||
assert_ne!(a, b);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn derive_path_differs_by_url() {
|
||||
let root = PathBuf::from("/tmp/repos");
|
||||
let a = derive_path(&root, "ws", "https://example.com/x.git", "main");
|
||||
let b = derive_path(&root, "ws", "https://example.com/y.git", "main");
|
||||
assert_ne!(a, b);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn derive_path_traversal_safe() {
|
||||
let root = PathBuf::from("/tmp/repos");
|
||||
// Malicious workspace tries to escape the root.
|
||||
let p = derive_path(&root, "../etc", "https://example.com/x.git", "main");
|
||||
assert!(
|
||||
p.starts_with(&root),
|
||||
"sanitized workspace must stay under root: got {}",
|
||||
p.display()
|
||||
);
|
||||
assert!(!p.to_string_lossy().contains(".."));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn derive_path_slashy_ref_is_flat() {
|
||||
let root = PathBuf::from("/tmp/repos");
|
||||
let p = derive_path(&root, "ws", "https://example.com/x.git", "feature/nested");
|
||||
// Ref becomes part of a single filename, not a nested dir.
|
||||
assert_eq!(p.parent().unwrap().parent().unwrap(), root.as_path());
|
||||
}
|
||||
|
||||
/// End-to-end: seed a bare git repo in a tempdir, ensure it into
|
||||
/// a fresh repo_root, verify cached=true on the second call, then
|
||||
/// release. Requires `git` on PATH; marked `#[ignore]` so CI
|
||||
/// without git-installed runners skips it silently.
|
||||
#[tokio::test]
|
||||
#[ignore]
|
||||
async fn ensure_then_cached_then_release() {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let source = tmp.path().join("source.git");
|
||||
let repo_root = tmp.path().join("repos");
|
||||
|
||||
// Init a bare-ish source repo with one commit on branch `main`.
|
||||
let seed_dir = tmp.path().join("seed");
|
||||
std::fs::create_dir_all(&seed_dir).unwrap();
|
||||
let git = |args: &[&str], cwd: &std::path::Path| {
|
||||
let out = std::process::Command::new("git")
|
||||
.args(args)
|
||||
.current_dir(cwd)
|
||||
.output()
|
||||
.unwrap();
|
||||
assert!(
|
||||
out.status.success(),
|
||||
"git {:?} failed: {}",
|
||||
args,
|
||||
String::from_utf8_lossy(&out.stderr)
|
||||
);
|
||||
};
|
||||
git(&["init", "-b", "main"], &seed_dir);
|
||||
git(&["config", "user.email", "t@t"], &seed_dir);
|
||||
git(&["config", "user.name", "t"], &seed_dir);
|
||||
std::fs::write(seed_dir.join("README"), "hi").unwrap();
|
||||
git(&["add", "."], &seed_dir);
|
||||
git(&["commit", "-m", "seed"], &seed_dir);
|
||||
git(
|
||||
&["clone", "--bare", seed_dir.to_str().unwrap(), source.to_str().unwrap()],
|
||||
tmp.path(),
|
||||
);
|
||||
|
||||
let req = RepoEnsureRequest {
|
||||
url: format!("file://{}", source.display()),
|
||||
git_ref: "main".into(),
|
||||
workspace: "workspace:test".into(),
|
||||
};
|
||||
|
||||
// First ensure → fresh clone.
|
||||
let r1 = ensure_repo(&repo_root, &req).await.unwrap();
|
||||
assert!(!r1.cached, "first ensure should not be cached");
|
||||
assert!(!r1.head_sha.is_empty());
|
||||
assert!(std::path::Path::new(&r1.path).join(".git").exists());
|
||||
|
||||
// Second ensure → cache hit.
|
||||
let r2 = ensure_repo(&repo_root, &req).await.unwrap();
|
||||
assert!(r2.cached, "second ensure should hit cache");
|
||||
assert_eq!(r1.head_sha, r2.head_sha);
|
||||
assert_eq!(r1.path, r2.path);
|
||||
|
||||
// Release removes it.
|
||||
let rel = release_repo(
|
||||
&repo_root,
|
||||
&RepoReleaseRequest {
|
||||
url: req.url.clone(),
|
||||
git_ref: req.git_ref.clone(),
|
||||
workspace: req.workspace.clone(),
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(rel.removed);
|
||||
assert!(!std::path::Path::new(&r1.path).exists());
|
||||
|
||||
// Idempotent release.
|
||||
let rel2 = release_repo(
|
||||
&repo_root,
|
||||
&RepoReleaseRequest {
|
||||
url: req.url,
|
||||
git_ref: req.git_ref,
|
||||
workspace: req.workspace,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(!rel2.removed);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn sanitize_component_replaces_traversal() {
|
||||
assert_eq!(sanitize_component(".."), "_");
|
||||
assert_eq!(sanitize_component("../etc"), "_etc");
|
||||
assert_eq!(sanitize_component(""), "_");
|
||||
assert_eq!(sanitize_component("a/b\\c:d"), "a_b_c_d");
|
||||
}
|
||||
}
|
||||
@@ -200,6 +200,23 @@ pub enum Method {
|
||||
/// `payload`: empty.
|
||||
/// Reply: JSON `DashboardStorageReply`.
|
||||
DashboardStorage = 0x1d,
|
||||
/// Phase 9 R1a (2026-07-15): shallow-clone a `(url, git_ref)` on
|
||||
/// this peer under a caller-provided workspace namespace. The
|
||||
/// aggregator fans this out to every fleet node; this per-peer
|
||||
/// method is deliberately narrow — one clone, one path.
|
||||
///
|
||||
/// `payload`: JSON [`crate::cluster::repo_ensure::RepoEnsureRequest`].
|
||||
/// Reply: JSON [`crate::cluster::repo_ensure::RepoEnsureReply`].
|
||||
/// Requires [`RpcRouter::with_repo_root`]; else
|
||||
/// [`ErrorCode::NotConfigured`].
|
||||
RepoEnsure = 0x1e,
|
||||
/// Phase 9 R1a: inverse of [`Method::RepoEnsure`]. Removes the
|
||||
/// on-disk checkout for `(url, git_ref)` under the caller's
|
||||
/// workspace. Idempotent: absent checkout ⇒ `removed=false`.
|
||||
///
|
||||
/// `payload`: JSON [`crate::cluster::repo_ensure::RepoReleaseRequest`].
|
||||
/// Reply: JSON [`crate::cluster::repo_ensure::RepoReleaseReply`].
|
||||
RepoRelease = 0x1f,
|
||||
}
|
||||
|
||||
impl Method {
|
||||
@@ -236,6 +253,8 @@ impl Method {
|
||||
0x1b => Some(Method::GetTagExpiry),
|
||||
0x1c => Some(Method::DashboardStatus),
|
||||
0x1d => Some(Method::DashboardStorage),
|
||||
0x1e => Some(Method::RepoEnsure),
|
||||
0x1f => Some(Method::RepoRelease),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@@ -743,6 +762,10 @@ pub struct RpcRouter {
|
||||
/// `GetRefLocal` (strict local-only). Set at daemon startup by
|
||||
/// `ClusterServices` when a NodeIdentity is available.
|
||||
outbound_client: Option<Arc<crate::cluster::transport::QuicClient>>,
|
||||
/// Phase 9 R1a: root directory under which `RepoEnsure` materializes
|
||||
/// checkouts. `None` disables both `RepoEnsure` and `RepoRelease`
|
||||
/// (server returns `NotConfigured`).
|
||||
repo_root: Option<std::path::PathBuf>,
|
||||
}
|
||||
|
||||
/// Result of dispatching a request: either a real reply (`Ok`) or a
|
||||
@@ -764,9 +787,17 @@ impl RpcRouter {
|
||||
local_name,
|
||||
local_zone,
|
||||
outbound_client: None,
|
||||
repo_root: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Phase 9 R1a: enable `RepoEnsure` / `RepoRelease` by attaching a
|
||||
/// root directory. The directory is created on first use.
|
||||
pub fn with_repo_root(mut self, root: std::path::PathBuf) -> Self {
|
||||
self.repo_root = Some(root);
|
||||
self
|
||||
}
|
||||
|
||||
/// Enable ref-forwarding on `GetRef` misses by installing the
|
||||
/// outbound QUIC client the router will use to dial peers. See
|
||||
/// the field's rustdoc for the semantics.
|
||||
@@ -1108,6 +1139,43 @@ impl RpcRouter {
|
||||
.context("encoding DashboardStorageReply as JSON")?;
|
||||
Ok(HandlerOutcome::Reply(json))
|
||||
}
|
||||
Method::RepoEnsure => {
|
||||
let root = match &self.repo_root {
|
||||
Some(r) => r.clone(),
|
||||
None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
|
||||
};
|
||||
let req: crate::cluster::repo_ensure::RepoEnsureRequest =
|
||||
match serde_json::from_slice(payload) {
|
||||
Ok(r) => r,
|
||||
Err(_) => {
|
||||
return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
|
||||
}
|
||||
};
|
||||
let _ = tokio::fs::create_dir_all(&root).await;
|
||||
let reply =
|
||||
crate::cluster::repo_ensure::ensure_repo(&root, &req).await?;
|
||||
let json = serde_json::to_vec(&reply)
|
||||
.context("encoding RepoEnsureReply as JSON")?;
|
||||
Ok(HandlerOutcome::Reply(json))
|
||||
}
|
||||
Method::RepoRelease => {
|
||||
let root = match &self.repo_root {
|
||||
Some(r) => r.clone(),
|
||||
None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
|
||||
};
|
||||
let req: crate::cluster::repo_ensure::RepoReleaseRequest =
|
||||
match serde_json::from_slice(payload) {
|
||||
Ok(r) => r,
|
||||
Err(_) => {
|
||||
return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
|
||||
}
|
||||
};
|
||||
let reply =
|
||||
crate::cluster::repo_ensure::release_repo(&root, &req).await?;
|
||||
let json = serde_json::to_vec(&reply)
|
||||
.context("encoding RepoReleaseReply as JSON")?;
|
||||
Ok(HandlerOutcome::Reply(json))
|
||||
}
|
||||
Method::BlobStat => {
|
||||
let store = match &self.blob_store {
|
||||
Some(s) => s,
|
||||
|
||||
@@ -192,6 +192,43 @@ pub async fn call_dashboard_storage(conn: &Connection) -> Result<DashboardStorag
|
||||
serde_json::from_slice(&reply).context("decoding DashboardStorageReply JSON")
|
||||
}
|
||||
|
||||
/// Phase 9 R1b: convenience wrapper for [`Method::RepoEnsure`].
|
||||
/// Materializes `(url, git_ref)` on the connected peer under the
|
||||
/// caller-provided workspace namespace and returns the resulting
|
||||
/// on-disk path + head sha. Cached reply = the checkout was already
|
||||
/// present with a valid `.git`.
|
||||
pub async fn call_repo_ensure(
|
||||
conn: &Connection,
|
||||
req: &crate::cluster::repo_ensure::RepoEnsureRequest,
|
||||
) -> Result<crate::cluster::repo_ensure::RepoEnsureReply> {
|
||||
let payload = serde_json::to_vec(req).context("encoding RepoEnsureRequest")?;
|
||||
let reply = rpc_call(conn, Method::RepoEnsure, &payload).await?;
|
||||
if reply.len() == 1 {
|
||||
if let Some(code) = decode_error(reply[0]) {
|
||||
bail!("peer replied with error: {}", code.describe());
|
||||
}
|
||||
}
|
||||
serde_json::from_slice(&reply).context("decoding RepoEnsureReply JSON")
|
||||
}
|
||||
|
||||
/// Phase 9 R1b: convenience wrapper for [`Method::RepoRelease`].
|
||||
/// Removes the on-disk checkout for `(url, git_ref)` under the
|
||||
/// caller's workspace. `removed=false` when nothing was on disk to
|
||||
/// begin with (still `Ok`).
|
||||
pub async fn call_repo_release(
|
||||
conn: &Connection,
|
||||
req: &crate::cluster::repo_ensure::RepoReleaseRequest,
|
||||
) -> Result<crate::cluster::repo_ensure::RepoReleaseReply> {
|
||||
let payload = serde_json::to_vec(req).context("encoding RepoReleaseRequest")?;
|
||||
let reply = rpc_call(conn, Method::RepoRelease, &payload).await?;
|
||||
if reply.len() == 1 {
|
||||
if let Some(code) = decode_error(reply[0]) {
|
||||
bail!("peer replied with error: {}", code.describe());
|
||||
}
|
||||
}
|
||||
serde_json::from_slice(&reply).context("decoding RepoReleaseReply JSON")
|
||||
}
|
||||
|
||||
/// Recognise a single-byte reply as one of our error codes. Returns
|
||||
/// `None` for any other single-byte value (which is a valid reply,
|
||||
/// just an unusually short one).
|
||||
|
||||
@@ -226,6 +226,15 @@ impl ClusterServices {
|
||||
r = r.with_tag_store(store.clone());
|
||||
}
|
||||
r = r.with_outbound_client(outbound_client);
|
||||
// Phase 9 R1a wiring: enable RepoEnsure/RepoRelease when
|
||||
// the daemon has a blob_store_root (which is the
|
||||
// canonical anchor for all fleet on-disk state). Repos
|
||||
// materialize under <blob_store_root>/repos/<workspace>/…
|
||||
if let Some(root) = blob_store_root.as_ref() {
|
||||
let repo_root = root.join("repos");
|
||||
let _ = std::fs::create_dir_all(&repo_root);
|
||||
r = r.with_repo_root(repo_root);
|
||||
}
|
||||
let router = Arc::new(r);
|
||||
|
||||
let server =
|
||||
|
||||
@@ -542,6 +542,34 @@ impl AuthedCaller {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Phase 9 R1b: returns the workspace this caller may act on for
|
||||
/// repo-ensure operations. Namespaced callers are pinned to their
|
||||
/// own namespace; admin/open callers must provide one explicitly
|
||||
/// in the request body. Explicit user-supplied workspace is only
|
||||
/// honored for admin/open — a namespaced caller supplying a
|
||||
/// mismatched workspace is a forbidden write.
|
||||
pub fn resolve_workspace(&self, requested: Option<&str>) -> Result<String, (StatusCode, String)> {
|
||||
match self {
|
||||
AuthedCaller::Admin | AuthedCaller::Open => match requested {
|
||||
Some(w) if !w.is_empty() => Ok(w.to_string()),
|
||||
_ => Err((
|
||||
StatusCode::BAD_REQUEST,
|
||||
"workspace is required for unnamespaced callers".to_string(),
|
||||
)),
|
||||
},
|
||||
AuthedCaller::Namespaced { namespace } => match requested {
|
||||
None => Ok(namespace.clone()),
|
||||
Some(w) if w == namespace => Ok(namespace.clone()),
|
||||
Some(w) => Err((
|
||||
StatusCode::FORBIDDEN,
|
||||
format!(
|
||||
"workspace \"{w}\" is outside your namespace \"{namespace}\""
|
||||
),
|
||||
)),
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── write-through fan-out (Phase 9 F1) ──────────────────────────
|
||||
@@ -724,6 +752,217 @@ async fn fanout_delete_tag(s: &V2State, name: &str) -> Vec<PeerResult> {
|
||||
out
|
||||
}
|
||||
|
||||
// ── Phase 9 R1b: repo-ensure fan-out ────────────────────────────
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct RepoEnsureBody {
|
||||
pub url: String,
|
||||
pub git_ref: String,
|
||||
/// Optional for namespaced tokens (server pins to the caller's
|
||||
/// namespace); required for admin/open callers.
|
||||
#[serde(default)]
|
||||
pub workspace: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct RepoPeerResult {
|
||||
pub peer: String,
|
||||
pub ok: bool,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub path: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub head_sha: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub cached: Option<bool>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub removed: Option<bool>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub error: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct RepoFanoutReply {
|
||||
pub url: String,
|
||||
pub git_ref: String,
|
||||
pub workspace: String,
|
||||
pub peers: Vec<RepoPeerResult>,
|
||||
pub all_ok: bool,
|
||||
}
|
||||
|
||||
async fn handle_repos_ensure(
|
||||
State(s): State<Arc<V2State>>,
|
||||
axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
|
||||
Json(body): Json<RepoEnsureBody>,
|
||||
) -> Result<Json<RepoFanoutReply>, (StatusCode, String)> {
|
||||
if body.url.trim().is_empty() {
|
||||
return Err((StatusCode::BAD_REQUEST, "url cannot be empty".into()));
|
||||
}
|
||||
if body.git_ref.trim().is_empty() {
|
||||
return Err((StatusCode::BAD_REQUEST, "git_ref cannot be empty".into()));
|
||||
}
|
||||
let workspace = caller.resolve_workspace(body.workspace.as_deref())?;
|
||||
let req = crate::cluster::repo_ensure::RepoEnsureRequest {
|
||||
url: body.url.clone(),
|
||||
git_ref: body.git_ref.clone(),
|
||||
workspace: workspace.clone(),
|
||||
};
|
||||
let results = fanout_repo_ensure(&s, &req).await;
|
||||
let all_ok = results.iter().all(|r| r.ok);
|
||||
Ok(Json(RepoFanoutReply {
|
||||
url: body.url,
|
||||
git_ref: body.git_ref,
|
||||
workspace,
|
||||
peers: results,
|
||||
all_ok,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn handle_repos_release(
|
||||
State(s): State<Arc<V2State>>,
|
||||
axum::extract::Extension(caller): axum::extract::Extension<AuthedCaller>,
|
||||
Json(body): Json<RepoEnsureBody>,
|
||||
) -> Result<Json<RepoFanoutReply>, (StatusCode, String)> {
|
||||
if body.url.trim().is_empty() {
|
||||
return Err((StatusCode::BAD_REQUEST, "url cannot be empty".into()));
|
||||
}
|
||||
if body.git_ref.trim().is_empty() {
|
||||
return Err((StatusCode::BAD_REQUEST, "git_ref cannot be empty".into()));
|
||||
}
|
||||
let workspace = caller.resolve_workspace(body.workspace.as_deref())?;
|
||||
let req = crate::cluster::repo_ensure::RepoReleaseRequest {
|
||||
url: body.url.clone(),
|
||||
git_ref: body.git_ref.clone(),
|
||||
workspace: workspace.clone(),
|
||||
};
|
||||
let results = fanout_repo_release(&s, &req).await;
|
||||
let all_ok = results.iter().all(|r| r.ok);
|
||||
Ok(Json(RepoFanoutReply {
|
||||
url: body.url,
|
||||
git_ref: body.git_ref,
|
||||
workspace,
|
||||
peers: results,
|
||||
all_ok,
|
||||
}))
|
||||
}
|
||||
|
||||
async fn fanout_repo_ensure(
|
||||
s: &V2State,
|
||||
req: &crate::cluster::repo_ensure::RepoEnsureRequest,
|
||||
) -> Vec<RepoPeerResult> {
|
||||
let mut set = JoinSet::new();
|
||||
for peer in &s.peers {
|
||||
let peer = peer.clone();
|
||||
let req = req.clone();
|
||||
let s = s.clone_shallow();
|
||||
set.spawn(async move {
|
||||
let peer_name = peer.name.clone();
|
||||
let res = async {
|
||||
let conn = s.dial(&peer).await?;
|
||||
let r = crate::cluster::rpc::call_repo_ensure(&conn, &req).await;
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
r
|
||||
}
|
||||
.await;
|
||||
match res {
|
||||
Ok(reply) => RepoPeerResult {
|
||||
peer: peer_name,
|
||||
ok: true,
|
||||
path: Some(reply.path),
|
||||
head_sha: Some(reply.head_sha),
|
||||
cached: Some(reply.cached),
|
||||
removed: None,
|
||||
error: None,
|
||||
},
|
||||
Err(e) => RepoPeerResult {
|
||||
peer: peer_name,
|
||||
ok: false,
|
||||
path: None,
|
||||
head_sha: None,
|
||||
cached: None,
|
||||
removed: None,
|
||||
error: Some(e.to_string()),
|
||||
},
|
||||
}
|
||||
});
|
||||
}
|
||||
let mut out = Vec::new();
|
||||
while let Some(joined) = set.join_next().await {
|
||||
match joined {
|
||||
Ok(r) => out.push(r),
|
||||
Err(e) => out.push(RepoPeerResult {
|
||||
peer: "<join-error>".into(),
|
||||
ok: false,
|
||||
path: None,
|
||||
head_sha: None,
|
||||
cached: None,
|
||||
removed: None,
|
||||
error: Some(e.to_string()),
|
||||
}),
|
||||
}
|
||||
}
|
||||
out.sort_by(|a, b| a.peer.cmp(&b.peer));
|
||||
out
|
||||
}
|
||||
|
||||
async fn fanout_repo_release(
|
||||
s: &V2State,
|
||||
req: &crate::cluster::repo_ensure::RepoReleaseRequest,
|
||||
) -> Vec<RepoPeerResult> {
|
||||
let mut set = JoinSet::new();
|
||||
for peer in &s.peers {
|
||||
let peer = peer.clone();
|
||||
let req = req.clone();
|
||||
let s = s.clone_shallow();
|
||||
set.spawn(async move {
|
||||
let peer_name = peer.name.clone();
|
||||
let res = async {
|
||||
let conn = s.dial(&peer).await?;
|
||||
let r = crate::cluster::rpc::call_repo_release(&conn, &req).await;
|
||||
conn.close(quinn::VarInt::from_u32(0), b"done");
|
||||
r
|
||||
}
|
||||
.await;
|
||||
match res {
|
||||
Ok(reply) => RepoPeerResult {
|
||||
peer: peer_name,
|
||||
ok: true,
|
||||
path: None,
|
||||
head_sha: None,
|
||||
cached: None,
|
||||
removed: Some(reply.removed),
|
||||
error: None,
|
||||
},
|
||||
Err(e) => RepoPeerResult {
|
||||
peer: peer_name,
|
||||
ok: false,
|
||||
path: None,
|
||||
head_sha: None,
|
||||
cached: None,
|
||||
removed: None,
|
||||
error: Some(e.to_string()),
|
||||
},
|
||||
}
|
||||
});
|
||||
}
|
||||
let mut out = Vec::new();
|
||||
while let Some(joined) = set.join_next().await {
|
||||
match joined {
|
||||
Ok(r) => out.push(r),
|
||||
Err(e) => out.push(RepoPeerResult {
|
||||
peer: "<join-error>".into(),
|
||||
ok: false,
|
||||
path: None,
|
||||
head_sha: None,
|
||||
cached: None,
|
||||
removed: None,
|
||||
error: Some(e.to_string()),
|
||||
}),
|
||||
}
|
||||
}
|
||||
out.sort_by(|a, b| a.peer.cmp(&b.peer));
|
||||
out
|
||||
}
|
||||
|
||||
// ── auth middleware ─────────────────────────────────────────────
|
||||
|
||||
/// Resolve the request's bearer against the aggregator's known
|
||||
@@ -1085,6 +1324,8 @@ pub fn build(state: Arc<V2State>) -> Router {
|
||||
.route("/api/v2/sessions/:id/pin", post(handle_session_pin))
|
||||
.route("/api/v2/sessions/:id/renew", post(handle_renew_session))
|
||||
.route("/api/v2/sessions/:id/commit", post(handle_commit_session))
|
||||
.route("/api/v2/repos/ensure", post(handle_repos_ensure))
|
||||
.route("/api/v2/repos/release", post(handle_repos_release))
|
||||
.route_layer(axum::middleware::from_fn_with_state(state.clone(), v2_auth))
|
||||
.with_state(state)
|
||||
}
|
||||
|
||||
@@ -19,6 +19,33 @@ archive_path = "/data/archive"
|
||||
zfs_dataset = "data/archive"
|
||||
retain_weeks = 12
|
||||
|
||||
[cluster]
|
||||
zone = "fabric-10g"
|
||||
bind_lan = "10.0.0.13:7701"
|
||||
prom_bind = "0.0.0.0:7703"
|
||||
bind_rpc_lan = "10.0.0.13:7702"
|
||||
bind_rpc_tailscale = "100.104.171.32:7702"
|
||||
blob_store_root = "/home/osobh/clawstor-deploy/data"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "tank"
|
||||
zone = "fabric-10g"
|
||||
lan_addr = "10.0.0.14:7701"
|
||||
rpc_lan_addr = "10.10.0.10:7702"
|
||||
tailscale_addr = "100.108.129.81:7702"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "morpheus"
|
||||
zone = "lan-1g"
|
||||
lan_addr = "10.0.0.5:7701"
|
||||
rpc_lan_addr = "10.0.0.5:7702"
|
||||
tailscale_addr = "100.123.224.84:7702"
|
||||
|
||||
[cluster.tls]
|
||||
ca_cert = "/home/osobh/clawstor-deploy/tls/ca.crt"
|
||||
node_cert = "/home/osobh/clawstor-deploy/tls/node.crt"
|
||||
node_key = "/home/osobh/clawstor-deploy/tls/node.key"
|
||||
|
||||
[replication]
|
||||
receive_from_peer = true
|
||||
peer_user = "osobh"
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
[node]
|
||||
name = "morpheus"
|
||||
role = "secondary"
|
||||
|
||||
[hot]
|
||||
path = "/hot/targets"
|
||||
max_gb = 80
|
||||
stale_hours = 48
|
||||
|
||||
[warm]
|
||||
projects_path = "/slab/projects"
|
||||
zfs_dataset = "none"
|
||||
snapshot_retain_hours = 24
|
||||
snapshot_retain_days = 7
|
||||
snapshot_retain_weeks = 4
|
||||
|
||||
[cluster]
|
||||
zone = "lan-1g"
|
||||
# Morpheus has no direct 10G to Architect/Tank — use main LAN for all traffic
|
||||
bind_lan = "10.0.0.5:7701"
|
||||
prom_bind = "0.0.0.0:7703"
|
||||
bind_rpc_lan = "10.0.0.5:7702"
|
||||
bind_rpc_tailscale = "100.123.224.84:7702"
|
||||
blob_store_root = "/home/osobh/clawstor-deploy/data"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "architect"
|
||||
zone = "fabric-10g"
|
||||
lan_addr = "10.0.0.13:7701"
|
||||
rpc_lan_addr = "10.0.0.13:7702"
|
||||
tailscale_addr = "100.104.171.32:7702"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "tank"
|
||||
zone = "fabric-10g"
|
||||
lan_addr = "10.0.0.14:7701"
|
||||
rpc_lan_addr = "10.0.0.14:7702"
|
||||
tailscale_addr = "100.108.129.81:7702"
|
||||
|
||||
[cluster.tls]
|
||||
ca_cert = "/home/osobh/clawstor-deploy/tls/ca.crt"
|
||||
node_cert = "/home/osobh/clawstor-deploy/tls/node.crt"
|
||||
node_key = "/home/osobh/clawstor-deploy/tls/node.key"
|
||||
+27
-5
@@ -14,13 +14,35 @@ snapshot_retain_hours = 24
|
||||
snapshot_retain_days = 7
|
||||
snapshot_retain_weeks = 4
|
||||
|
||||
[cluster]
|
||||
zone = "fabric-10g"
|
||||
bind_lan = "10.0.0.14:7701"
|
||||
prom_bind = "0.0.0.0:7703"
|
||||
bind_rpc_lan = "10.0.0.14:7702"
|
||||
bind_rpc_tailscale = "100.108.129.81:7702"
|
||||
blob_store_root = "/home/osobh/clawstor-deploy/data"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "architect"
|
||||
zone = "fabric-10g"
|
||||
lan_addr = "10.0.0.13:7701"
|
||||
rpc_lan_addr = "10.10.0.9:7702"
|
||||
tailscale_addr = "100.104.171.32:7702"
|
||||
|
||||
[[cluster.peers]]
|
||||
name = "morpheus"
|
||||
zone = "lan-1g"
|
||||
lan_addr = "10.0.0.5:7701"
|
||||
rpc_lan_addr = "10.0.0.5:7702"
|
||||
tailscale_addr = "100.123.224.84:7702"
|
||||
|
||||
[cluster.tls]
|
||||
ca_cert = "/home/osobh/clawstor-deploy/tls/ca.crt"
|
||||
node_cert = "/home/osobh/clawstor-deploy/tls/node.crt"
|
||||
node_key = "/home/osobh/clawstor-deploy/tls/node.key"
|
||||
|
||||
[replication]
|
||||
# 10.10.0.9 is architect-fab-tank — the dedicated 10G fabric link between
|
||||
# the two nodes. Intentionally used for replication to maximise bandwidth;
|
||||
# architect's primary LAN address is 10.0.0.13.
|
||||
send_to_host = "10.10.0.9"
|
||||
send_to_user = "osobh"
|
||||
cold_dataset_on_peer = "data/archive/tank-projects"
|
||||
# nightly_at is reserved for future use; replication schedule is currently
|
||||
# controlled by the claw-store-replicate.timer systemd unit.
|
||||
nightly_at = "03:30"
|
||||
|
||||
@@ -0,0 +1,17 @@
|
||||
[Unit]
|
||||
Description=clawstor FUSE mount
|
||||
After=claw-store.service
|
||||
Requires=claw-store.service
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=osobh
|
||||
Environment=PATH=/home/osobh/.cargo/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
|
||||
ExecStartPre=/bin/mkdir -p /home/osobh/clawstor-mount
|
||||
ExecStart=/usr/local/bin/claw-fuse --data-dir /home/osobh/clawstor-deploy/data --mount /home/osobh/clawstor-mount
|
||||
ExecStop=/bin/fusermount -u /home/osobh/clawstor-mount
|
||||
Restart=on-failure
|
||||
RestartSec=30
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -11,7 +11,7 @@ Wants=network-online.target
|
||||
[Service]
|
||||
Type=simple
|
||||
User=osobh
|
||||
ExecStart=/usr/local/bin/claw-store serve --port 7700 --static-dir /usr/share/claw-store/static
|
||||
ExecStart=/usr/local/bin/claw-store serve --port 7700 --static-dir /usr/share/claw-store/static --v2-static-dir /usr/share/claw-store/v2
|
||||
Restart=on-failure
|
||||
RestartSec=15
|
||||
Environment=RUST_LOG=info
|
||||
|
||||
@@ -1,15 +1,19 @@
|
||||
[Unit]
|
||||
Description=claw-store fleet storage daemon
|
||||
After=zfs-mount.service network.target
|
||||
Wants=zfs-mount.service
|
||||
Description=clawstor cluster daemon (gossip + QUIC blob store + ZFS snapshots)
|
||||
After=network-online.target
|
||||
Wants=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
User=osobh
|
||||
Environment=PATH=/home/osobh/.cargo/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
|
||||
Environment=RUST_LOG=info
|
||||
ExecStart=/usr/local/bin/claw-store daemon
|
||||
Restart=on-failure
|
||||
RestartSec=30
|
||||
Environment=RUST_LOG=info
|
||||
RestartSec=15
|
||||
TimeoutStopSec=60
|
||||
ProtectSystem=strict
|
||||
ReadWritePaths=/var/lib/claw-store /hot/targets /slab/projects /home/osobh/clawstor-deploy
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
|
||||
Reference in New Issue
Block a user