Compare commits
6
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2f7eabf034 | ||
|
|
7902c2e395 | ||
|
|
3d504ee6b3 | ||
|
|
8407c99ba4 | ||
|
|
088b88b225 | ||
|
|
c3f6b500fc |
@@ -32,7 +32,7 @@ install-systemd:
|
|||||||
install-dashboard:
|
install-dashboard:
|
||||||
cd dashboard && npm ci && npm run build
|
cd dashboard && npm ci && npm run build
|
||||||
install -dm755 $(INSTALL_STATIC)
|
install -dm755 $(INSTALL_STATIC)
|
||||||
cp -r dashboard/dist/. $(INSTALL_STATIC)/
|
cp -r claw-store/static/. $(INSTALL_STATIC)/
|
||||||
@echo "Dashboard installed to $(INSTALL_STATIC)"
|
@echo "Dashboard installed to $(INSTALL_STATIC)"
|
||||||
|
|
||||||
## Install node-specific config (NODE=architect|tank)
|
## Install node-specific config (NODE=architect|tank)
|
||||||
|
|||||||
@@ -128,6 +128,7 @@ mod tests {
|
|||||||
peer: None,
|
peer: None,
|
||||||
cluster: None,
|
cluster: None,
|
||||||
api_token: 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()
|
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).
|
// 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!(
|
let new_content = format!(
|
||||||
"# claw-store managed — do not edit manually\n\
|
"# 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()))
|
.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`,
|
/// Remove a TOML section `[name]` and all its key=value lines from `src`,
|
||||||
/// stopping at the next `[section]` header or EOF.
|
/// stopping at the next `[section]` header or EOF.
|
||||||
fn strip_section(src: &str, name: &str) -> String {
|
fn strip_section(src: &str, name: &str) -> String {
|
||||||
@@ -102,6 +123,46 @@ mod tests {
|
|||||||
assert!(verify_cargo_config(&warm, &hot).unwrap());
|
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]
|
#[test]
|
||||||
fn test_verify_cargo_config_detects_missing() {
|
fn test_verify_cargo_config_detects_missing() {
|
||||||
let dir = TempDir::new().unwrap();
|
let dir = TempDir::new().unwrap();
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ pub mod metrics;
|
|||||||
pub mod prom;
|
pub mod prom;
|
||||||
pub mod ref_tracking;
|
pub mod ref_tracking;
|
||||||
pub mod refs;
|
pub mod refs;
|
||||||
|
pub mod repo_ensure;
|
||||||
pub mod rpc;
|
pub mod rpc;
|
||||||
pub mod services;
|
pub mod services;
|
||||||
pub mod snapshot;
|
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.
|
/// `payload`: empty.
|
||||||
/// Reply: JSON `DashboardStorageReply`.
|
/// Reply: JSON `DashboardStorageReply`.
|
||||||
DashboardStorage = 0x1d,
|
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 {
|
impl Method {
|
||||||
@@ -236,6 +253,8 @@ impl Method {
|
|||||||
0x1b => Some(Method::GetTagExpiry),
|
0x1b => Some(Method::GetTagExpiry),
|
||||||
0x1c => Some(Method::DashboardStatus),
|
0x1c => Some(Method::DashboardStatus),
|
||||||
0x1d => Some(Method::DashboardStorage),
|
0x1d => Some(Method::DashboardStorage),
|
||||||
|
0x1e => Some(Method::RepoEnsure),
|
||||||
|
0x1f => Some(Method::RepoRelease),
|
||||||
_ => None,
|
_ => None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -743,6 +762,10 @@ pub struct RpcRouter {
|
|||||||
/// `GetRefLocal` (strict local-only). Set at daemon startup by
|
/// `GetRefLocal` (strict local-only). Set at daemon startup by
|
||||||
/// `ClusterServices` when a NodeIdentity is available.
|
/// `ClusterServices` when a NodeIdentity is available.
|
||||||
outbound_client: Option<Arc<crate::cluster::transport::QuicClient>>,
|
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
|
/// Result of dispatching a request: either a real reply (`Ok`) or a
|
||||||
@@ -764,9 +787,17 @@ impl RpcRouter {
|
|||||||
local_name,
|
local_name,
|
||||||
local_zone,
|
local_zone,
|
||||||
outbound_client: None,
|
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
|
/// Enable ref-forwarding on `GetRef` misses by installing the
|
||||||
/// outbound QUIC client the router will use to dial peers. See
|
/// outbound QUIC client the router will use to dial peers. See
|
||||||
/// the field's rustdoc for the semantics.
|
/// the field's rustdoc for the semantics.
|
||||||
@@ -1108,6 +1139,43 @@ impl RpcRouter {
|
|||||||
.context("encoding DashboardStorageReply as JSON")?;
|
.context("encoding DashboardStorageReply as JSON")?;
|
||||||
Ok(HandlerOutcome::Reply(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 => {
|
Method::BlobStat => {
|
||||||
let store = match &self.blob_store {
|
let store = match &self.blob_store {
|
||||||
Some(s) => s,
|
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")
|
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
|
/// Recognise a single-byte reply as one of our error codes. Returns
|
||||||
/// `None` for any other single-byte value (which is a valid reply,
|
/// `None` for any other single-byte value (which is a valid reply,
|
||||||
/// just an unusually short one).
|
/// just an unusually short one).
|
||||||
|
|||||||
@@ -226,6 +226,15 @@ impl ClusterServices {
|
|||||||
r = r.with_tag_store(store.clone());
|
r = r.with_tag_store(store.clone());
|
||||||
}
|
}
|
||||||
r = r.with_outbound_client(outbound_client);
|
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 router = Arc::new(r);
|
||||||
|
|
||||||
let server =
|
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) ──────────────────────────
|
// ── write-through fan-out (Phase 9 F1) ──────────────────────────
|
||||||
@@ -724,6 +752,217 @@ async fn fanout_delete_tag(s: &V2State, name: &str) -> Vec<PeerResult> {
|
|||||||
out
|
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 ─────────────────────────────────────────────
|
// ── auth middleware ─────────────────────────────────────────────
|
||||||
|
|
||||||
/// Resolve the request's bearer against the aggregator's known
|
/// 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/pin", post(handle_session_pin))
|
||||||
.route("/api/v2/sessions/:id/renew", post(handle_renew_session))
|
.route("/api/v2/sessions/:id/renew", post(handle_renew_session))
|
||||||
.route("/api/v2/sessions/:id/commit", post(handle_commit_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))
|
.route_layer(axum::middleware::from_fn_with_state(state.clone(), v2_auth))
|
||||||
.with_state(state)
|
.with_state(state)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -19,6 +19,33 @@ archive_path = "/data/archive"
|
|||||||
zfs_dataset = "data/archive"
|
zfs_dataset = "data/archive"
|
||||||
retain_weeks = 12
|
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]
|
[replication]
|
||||||
receive_from_peer = true
|
receive_from_peer = true
|
||||||
peer_user = "osobh"
|
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_days = 7
|
||||||
snapshot_retain_weeks = 4
|
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]
|
[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_host = "10.10.0.9"
|
||||||
send_to_user = "osobh"
|
send_to_user = "osobh"
|
||||||
cold_dataset_on_peer = "data/archive/tank-projects"
|
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"
|
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]
|
[Service]
|
||||||
Type=simple
|
Type=simple
|
||||||
User=osobh
|
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
|
Restart=on-failure
|
||||||
RestartSec=15
|
RestartSec=15
|
||||||
Environment=RUST_LOG=info
|
Environment=RUST_LOG=info
|
||||||
|
|||||||
@@ -1,15 +1,19 @@
|
|||||||
[Unit]
|
[Unit]
|
||||||
Description=claw-store fleet storage daemon
|
Description=clawstor cluster daemon (gossip + QUIC blob store + ZFS snapshots)
|
||||||
After=zfs-mount.service network.target
|
After=network-online.target
|
||||||
Wants=zfs-mount.service
|
Wants=network-online.target
|
||||||
|
|
||||||
[Service]
|
[Service]
|
||||||
Type=simple
|
Type=simple
|
||||||
User=osobh
|
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
|
ExecStart=/usr/local/bin/claw-store daemon
|
||||||
Restart=on-failure
|
Restart=on-failure
|
||||||
RestartSec=30
|
RestartSec=15
|
||||||
Environment=RUST_LOG=info
|
TimeoutStopSec=60
|
||||||
|
ProtectSystem=strict
|
||||||
|
ReadWritePaths=/var/lib/claw-store /hot/targets /slab/projects /home/osobh/clawstor-deploy
|
||||||
|
|
||||||
[Install]
|
[Install]
|
||||||
WantedBy=multi-user.target
|
WantedBy=multi-user.target
|
||||||
|
|||||||
Reference in New Issue
Block a user