clawverse/clawstor
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 9 R1b: aggregator fan-out for RepoEnsure/RepoRelease

Browse Source 
Layers HTTP over the R1a per-peer primitive so external callers
(clawmates, gitea runners, ops tooling) speak one URL to the
aggregator instead of dialing every peer.

Endpoints (require v2 auth, same middleware as tags/sessions):
  POST /api/v2/repos/ensure   {url, git_ref, workspace?}
  POST /api/v2/repos/release  {url, git_ref, workspace?}

Namespaced tokens are pinned to their own workspace (workspace omitted
in body → derived from token; explicit mismatch → 403). Admin/open
callers must supply workspace explicitly.

Reply shape mirrors the tag fan-out (FanoutReply) with per-peer
{peer, ok, path?, head_sha?, cached?, removed?, error?}. all_ok is
true iff every peer succeeded.

Also adds client wrappers call_repo_ensure/call_repo_release in
cluster/rpc/client.rs used by the aggregator's fan-out.
This commit is contained in:
Omar Sobh
2026-07-15 04:08:35 -07:00
parent 5c1d962bf2
commit 5c9bc7eb9c
2 changed files with 278 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
claw-store/src/serve_v2.rs
+241
View File
@@ -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)
}