vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-controller: re-read the forge token from the store when the forge rejects it

Browse source 
The controller read its forge token from the swarm secret store once, when
its client slot filled, and kept sending it until the daemon restarted. A
token the forge re-minted or revoked left every forge-backed path failing
with 401 until someone restarted the controller, and a restart drops the
runtime-only job graph.

Every forge request now goes through `forge::api::Api::call`. On a 401 it
asks a `StoredSecret` (the queue clients' debounced, single-flight re-read
with backoff) for a token to try instead, rebuilds the forgejo client with
it, and sends the request once more. A second 401, or a store still holding
the refused token, surfaces the error as before. Only 401 counts: unlike an
IdP, the forge answers 400 to a malformed request, and a re-read started by
one would put a real 401 that follows inside the backoff.

forgejo-api bakes the token into its own reqwest client's default headers
and takes no client or middleware from outside, so there is no single
request function to hook. Each `self.api.<endpoint>(..)` site becomes
`self.api.call(|api| api.<endpoint>(..).into_future())`; option structs a
site moved into the call are cloned, since the closure can run twice. The
`OnceLock<Arc<Client>>` slot and its `get()` callers are unchanged.

Refs #5017
This commit is contained in:
atlas 2026-10-13 21:38:35 +02:00 • committed by mara
commit 6ba29c4d33
11 changed files with 638 additions and 192 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/swarm/credentials.md
View file

@ -62,7 +62,7 @@ of the cell says how.
| `swarm/agents/<agent>/matrix-recovery/<account>` | `swarm-controller`, a random passphrase for every account listed under the agent's `matrix/`, in a five-minute pass | the agent container itself, under the certificate its hive passed in, when `hive-matrix-daemon` brings the account up | ❌ never replaced: the account's server-side key backup opens only with the passphrase it was enabled with, so the pass writes only to an empty path | ❌ read when the account comes up; an account whose passphrase lands later turns backup on at the next daemon start | | `swarm/agents/<agent>/matrix-recovery/<account>` | `swarm-controller`, a random passphrase for every account listed under the agent's `matrix/`, in a five-minute pass | the agent container itself, under the certificate its hive passed in, when `hive-matrix-daemon` brings the account up | ❌ never replaced: the account's server-side key backup opens only with the passphrase it was enabled with, so the pass writes only to an empty path | ❌ read when the account comes up; an account whose passphrase lands later turns backup on at the next daemon start |
| `swarm/controller/swarm-controller/matrix/appservice-token` | `swarm-matrix-ctl`, inside the `hive-matrix` container, once | `swarm-controller`, under its own certificate | ❌ `swarm-matrix-ctl` mints it once; the container keeps its copy and republishes it when the store's differs | ✅ the controller reads it on every five-minute matrix pass | | `swarm/controller/swarm-controller/matrix/appservice-token` | `swarm-matrix-ctl`, inside the `hive-matrix` container, once | `swarm-controller`, under its own certificate | ❌ `swarm-matrix-ctl` mints it once; the container keeps its copy and republishes it when the store's differs | ✅ the controller reads it on every five-minute matrix pass |
| `swarm/controller/swarm-controller/oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | `swarm-controller`, under its own certificate, at start and on refusal | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ✅ read at start and held in memory; read again when authelia refuses it, at most once per backoff of 5 s doubling to 5 min while the store holds the refused value | | `swarm/controller/swarm-controller/oidc/client` | authelia, at its first boot; `swarm-secret-publish` copies it in | `swarm-controller`, under its own certificate, at start and on refusal | ❌ authelia mints it once. A re-mint is republished by `swarm-secret-publish`'s path unit | ✅ read at start and held in memory; read again when authelia refuses it, at most once per backoff of 5 s doubling to 5 min while the store holds the refused value |
| `swarm/controller/swarm-controller/forge/token` | the forge, inside the `hive-forge` container, once; `swarm-bao-forge-publisher` on the forge's host copies it in under its own certificate, with a write-only grant | `swarm-controller`, under its own certificate, at start | ❌ the forge mints it once and keeps its copy in the container; the publisher writes that copy each time the forge container starts and each time the mint writes the file | ❌ the controller reads it at start, and in the background with backoff until it first finds it; a replaced token reaches it with the next restart | | `swarm/controller/swarm-controller/forge/token` | the forge, inside the `hive-forge` container, once; `swarm-bao-forge-publisher` on the forge's host copies it in under its own certificate, with a write-only grant | `swarm-controller`, under its own certificate, at start and on refusal | ❌ the forge mints it once and keeps its copy in the container; the publisher writes that copy each time the forge container starts and each time the mint writes the file | ✅ the controller reads it at start, and in the background with backoff until it first finds it. When the forge answers 401 it reads it again and sends the request once more with the new token, at most once per backoff of 5 s doubling to 5 min while the store holds the refused value |
| `swarm/agents/<agent>/bao-mtls` | the store's agent PKI mount (`deploy.bao.agentPkiMountPath`), which generates the key, at `swarm-controller`'s request at agent creation | `hive-c0re`, under the hive's own certificate, when it writes the agent's container config; then `hive-agent-bao-identity` in the agent container, under the leaf it holds | ✅ `swarm-controller`'s five-minute pass re-issues a live agent's leaf once it's past half its validity (45 of 90 days, read from the certificate itself) | ✅ `hive-agent-bao-identity` re-reads it every 10 minutes and swaps a new leaf into `/run/hive-agent-bao-identity/`, which every store login in the container presents. It reads under the leaf it holds, so it renews only while that leaf is valid | | `swarm/agents/<agent>/bao-mtls` | the store's agent PKI mount (`deploy.bao.agentPkiMountPath`), which generates the key, at `swarm-controller`'s request at agent creation | `hive-c0re`, under the hive's own certificate, when it writes the agent's container config; then `hive-agent-bao-identity` in the agent container, under the leaf it holds | ✅ `swarm-controller`'s five-minute pass re-issues a live agent's leaf once it's past half its validity (45 of 90 days, read from the certificate itself) | ✅ `hive-agent-bao-identity` re-reads it every 10 minutes and swaps a new leaf into `/run/hive-agent-bao-identity/`, which every store login in the container presents. It reads under the leaf it holds, so it renews only while that leaf is valid |
| `swarm/agents/<agent>/queue` | `swarm-controller`, at agent creation | `hive-agent` in the agent container, under the agent's own certificate, held in memory — the identity it presents to the swarm queue, naming that one agent rather than its hive; `swarm-logs` in the agent container, under the same certificate, on every query, as the bearer it presents at the log store; `swarm-nats-auth` and `swarm-logs-auth`, each under its own certificate, to check a presented secret | ✅ `swarm-controller`'s five-minute pass re-mints a live agent's secret once it's 45 days old by `minted_at` on the stored object; a secret with no `minted_at` gets one stamped, value unchanged. The pass skips agents declared `Destroyed` — declaring an agent destroyed deletes every version of the path instead, the undo of the mint rather than another one | ✅ `hive-agent` reads the path before its first connect and again on every reconnect attempt, so a reconnect after a re-mint presents the new secret. An open connection keeps the secret it connected with; after a revocation the agent keeps retrying under the queue client's backoff. `swarm-logs` reads it on every query and once more after a 401. `swarm-logs-auth` keeps granting a secret it verified for `deploy.victorialogs.authVerdictTtl` (60 s by default) without reading the path, so a replaced or deleted secret still works at the log store for up to that long | | `swarm/agents/<agent>/queue` | `swarm-controller`, at agent creation | `hive-agent` in the agent container, under the agent's own certificate, held in memory — the identity it presents to the swarm queue, naming that one agent rather than its hive; `swarm-logs` in the agent container, under the same certificate, on every query, as the bearer it presents at the log store; `swarm-nats-auth` and `swarm-logs-auth`, each under its own certificate, to check a presented secret | ✅ `swarm-controller`'s five-minute pass re-mints a live agent's secret once it's 45 days old by `minted_at` on the stored object; a secret with no `minted_at` gets one stamped, value unchanged. The pass skips agents declared `Destroyed` — declaring an agent destroyed deletes every version of the path instead, the undo of the mint rather than another one | ✅ `hive-agent` reads the path before its first connect and again on every reconnect attempt, so a reconnect after a re-mint presents the new secret. An open connection keeps the secret it connected with; after a revocation the agent keeps retrying under the queue client's backoff. `swarm-logs` reads it on every query and once more after a 401. `swarm-logs-auth` keeps granting a secret it verified for `deploy.victorialogs.authVerdictTtl` (60 s by default) without reading the path, so a replaced or deleted secret still works at the log store for up to that long |
| `swarm/agents/<agent>/forge-token` | `swarm-controller`, at agent creation and in a pass every 5 minutes over every agent with a store identity | the agent container itself, under its own certificate, fetched to `/run/hive-agent-forge-token/token` | ✅ the controller re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes) | ✅ the agent re-fetches on a 10-minute timer | | `swarm/agents/<agent>/forge-token` | `swarm-controller`, at agent creation and in a pass every 5 minutes over every agent with a store identity | the agent container itself, under its own certificate, fetched to `/run/hive-agent-forge-token/token` | ✅ the controller re-mints when the stored token is missing or no longer matches the forge (last eight characters and scopes) | ✅ the agent re-fetches on a 10-minute timer |

300
swarm-controller/src/forge.rs
View file

@ -16,7 +16,10 @@
//! there, later from a background re-read when it is not. Every caller reads //! there, later from a background re-read when it is not. Every caller reads
//! the slot and treats an empty one as "no forge access here" — the same //! the slot and treats an empty one as "no forge access here" — the same
//! graceful-absence shape the queue coordinates use — so the daemon serves //! graceful-absence shape the queue coordinates use — so the daemon serves
//! from the start and gains forge access without a restart. //! from the start and gains forge access without a restart. Once filled, the
//! client reads the token again when the forge answers 401 and sends the
//! request once more with it (`forge/api.rs`), so a replaced token needs no restart
//! either.
use anyhow::{Context, Result}; use anyhow::{Context, Result};
use forgejo_api::structs::{ use forgejo_api::structs::{
@ -28,7 +31,7 @@ use forgejo_api::structs::{
RepoDownloadPullDiffOrPatchQuery, RepoGetContentsQuery, RepoGetPullRequestCommitsQuery, RepoDownloadPullDiffOrPatchQuery, RepoGetContentsQuery, RepoGetPullRequestCommitsQuery,
RepoListPullRequestsQuery, RepoListPullRequestsQueryState, RepoSearchQuery, StateType, RepoListPullRequestsQuery, RepoListPullRequestsQueryState, RepoSearchQuery, StateType,
}; };
use forgejo_api::{ApiErrorKind, Auth, Forgejo, ForgejoError}; use forgejo_api::{ApiErrorKind, ForgejoError};
use futures_util::{StreamExt as _, TryStreamExt as _}; use futures_util::{StreamExt as _, TryStreamExt as _};
use reqwest::StatusCode; use reqwest::StatusCode;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
@ -38,8 +41,10 @@ use utoipa::ToSchema;
use crate::webhook::DeliveryKind; use crate::webhook::DeliveryKind;
use swarm_queue_client::review_classifier::Commit; use swarm_queue_client::review_classifier::Commit;
use swarm_queue_client::stored_secret::StoredSecret;
pub mod agent_token; pub mod agent_token;
mod api;
pub mod builder_account; pub mod builder_account;
pub mod classifier_account; pub mod classifier_account;
pub mod docs_grant; pub mod docs_grant;
@ -229,7 +234,7 @@ async fn fetch_token() -> Result<String> {
/// Typed forgejo client, built once at startup from env. /// Typed forgejo client, built once at startup from env.
pub struct Client { pub struct Client {
api: Forgejo, api: api::Api,
/// The forge's base URL, kept so a client authenticated as someone /// The forge's base URL, kept so a client authenticated as someone
/// else can be built against the same forge (see /// else can be built against the same forge (see
/// [`agent_token`]'s read-back). /// [`agent_token`]'s read-back).
@ -330,9 +335,6 @@ pub(crate) async fn within_forge_deadline<T>(
/// until it does. Either way the daemon starts, and every forge-backed path /// until it does. Either way the daemon starts, and every forge-backed path
/// reads `slot`, so it answers "no forge" until the token lands and works /// reads `slot`, so it answers "no forge" until the token lands and works
/// from then on. /// from then on.
///
/// A token replaced in the store after the slot is filled reaches this daemon
/// with its next restart.
pub async fn enable( pub async fn enable(
url: url::Url, url: url::Url,
slot: Arc<OnceLock<Arc<Client>>>, slot: Arc<OnceLock<Arc<Client>>>,
@ -362,14 +364,16 @@ pub async fn enable(
impl Client { impl Client {
/// One read of the token from the secret store, and a client for `url` /// One read of the token from the secret store, and a client for `url`
/// holding it. /// holding it, which reads it again when the forge refuses it.
/// ///
/// # Errors /// # Errors
/// The store refusing the login, nothing at the path because the forge's /// The store refusing the login, nothing at the path because the forge's
/// host has not published it yet, or the client failing to build. /// host has not published it yet, or the client failing to build.
async fn connect(url: &url::Url) -> Result<Self> { async fn connect(url: &url::Url) -> Result<Self> {
let token = fetch_token().await?; let token = StoredSecret::new("swarm-controller-forge", fetch_token().await?, || async {
let api = Forgejo::new(Auth::Token(&token), url.clone()).context("build forgejo client")?; fetch_token().await.map_err(Into::into)
});
let api = api::Api::new(url.clone(), token).context("build forgejo client")?;
Ok(Self { Ok(Self {
api, api,
url: url.clone(), url: url.clone(),
@ -403,7 +407,10 @@ impl Client {
async fn ensure_org_repo(&self, name: &str) -> Result<()> { async fn ensure_org_repo(&self, name: &str) -> Result<()> {
match self match self
.api .api
.create_org_repo(CONFIG_ORG, Self::repo_option(name, true)) .call(|api| {
api.create_org_repo(CONFIG_ORG, Self::repo_option(name, true))
.into_future()
})
.await .await
{ {
Ok(_) => { Ok(_) => {
@ -412,7 +419,11 @@ impl Client {
} }
Err(e) => { Err(e) => {
let existing = is_ambiguous_validation_failure(&e) let existing = is_ambiguous_validation_failure(&e)
&& self.api.repo_get(CONFIG_ORG, name).await.is_ok(); && self
.api
.call(|api| api.repo_get(CONFIG_ORG, name).into_future())
.await
.is_ok();
if folds_into_success(&e, existing) { if folds_into_success(&e, existing) {
tracing::debug!(%name, "swarm forge: repo already exists"); tracing::debug!(%name, "swarm forge: repo already exists");
Ok(()) Ok(())
@ -448,14 +459,17 @@ impl Client {
) -> Result<()> { ) -> Result<()> {
let res = self let res = self
.api .api
.repo_add_collaborator( .call(|api| {
owner, api.repo_add_collaborator(
repo, owner,
user, repo,
AddCollaboratorOption { user,
permission: Some(permission), AddCollaboratorOption {
}, permission: Some(permission),
) },
)
.into_future()
})
.await; .await;
match res { match res {
Ok(()) => {} Ok(()) => {}
@ -525,7 +539,10 @@ impl Client {
}; };
let Err(create_err) = self let Err(create_err) = self
.api .api
.repo_create_branch_protection(CONFIG_ORG, repo, rule) .call(|api| {
api.repo_create_branch_protection(CONFIG_ORG, repo, rule.clone())
.into_future()
})
.await .await
else { else {
tracing::info!(%repo, "swarm forge: applied operator branch protection"); tracing::info!(%repo, "swarm forge: applied operator branch protection");
@ -533,7 +550,10 @@ impl Client {
}; };
match self match self
.api .api
.repo_get_branch_protection(CONFIG_ORG, repo, "main") .call(|api| {
api.repo_get_branch_protection(CONFIG_ORG, repo, "main")
.into_future()
})
.await .await
{ {
Ok(_) => { Ok(_) => {
@ -635,7 +655,7 @@ impl Client {
async fn disable_repo_creation(&self, agent: &str) -> Result<()> { async fn disable_repo_creation(&self, agent: &str) -> Result<()> {
let user = self let user = self
.api .api
.user_get(agent) .call(|api| api.user_get(agent).into_future())
.await .await
.with_context(|| format!("read forge user {agent} before locking it down"))?; .with_context(|| format!("read forge user {agent} before locking it down"))?;
let expected = agent_email(agent); let expected = agent_email(agent);
@ -648,7 +668,10 @@ impl Client {
); );
} }
self.api self.api
.admin_edit_user(agent, repo_creation_lockdown(agent)) .call(|api| {
api.admin_edit_user(agent, repo_creation_lockdown(agent))
.into_future()
})
.await .await
.with_context(|| { .with_context(|| {
format!("disable repo creation for forge user {agent} (max_repo_creation = 0)") format!("disable repo creation for forge user {agent} (max_repo_creation = 0)")
@ -671,18 +694,21 @@ impl Client {
.context("generating a throwaway password for a forge account")?; .context("generating a throwaway password for a forge account")?;
let res = self let res = self
.api .api
.admin_create_user(CreateUserOption { .call(|api| {
created_at: None, api.admin_create_user(CreateUserOption {
email: email.to_owned(), created_at: None,
full_name: None, email: email.to_owned(),
login_name: None, full_name: None,
must_change_password: Some(false), login_name: None,
password: Some(password), must_change_password: Some(false),
restricted: None, password: Some(password.clone()),
send_notify: None, restricted: None,
source_id: None, send_notify: None,
username: login.to_owned(), source_id: None,
visibility: None, username: login.to_owned(),
visibility: None,
})
.into_future()
}) })
.await; .await;
match res { match res {
@ -691,8 +717,12 @@ impl Client {
Ok(()) Ok(())
} }
Err(e) => { Err(e) => {
let existing = let existing = is_ambiguous_validation_failure(&e)
is_ambiguous_validation_failure(&e) && self.api.user_get(login).await.is_ok(); && self
.api
.call(|api| api.user_get(login).into_future())
.await
.is_ok();
if folds_into_success(&e, existing) { if folds_into_success(&e, existing) {
tracing::debug!(%login, "swarm forge: forge user already exists"); tracing::debug!(%login, "swarm forge: forge user already exists");
Ok(()) Ok(())
@ -747,21 +777,24 @@ impl Client {
}, },
]; ];
self.api self.api
.repo_change_files( .call(|api| {
CONFIG_ORG, api.repo_change_files(
repo, CONFIG_ORG,
ChangeFilesOptions { repo,
author: None, ChangeFilesOptions {
branch: None, author: None,
committer: None, branch: None,
dates: None, committer: None,
files, dates: None,
force_overwrite_new_branch: None, files: files.clone(),
message: Some("swarm-controller: seed agent config".to_owned()), force_overwrite_new_branch: None,
new_branch: None, message: Some("swarm-controller: seed agent config".to_owned()),
signoff: None, new_branch: None,
}, signoff: None,
) },
)
.into_future()
})
.await .await
.with_context(|| format!("seed config files in {CONFIG_ORG}/{repo}"))?; .with_context(|| format!("seed config files in {CONFIG_ORG}/{repo}"))?;
tracing::info!(%repo, %agent, "swarm forge: seeded agent.nix + flake.nix"); tracing::info!(%repo, %agent, "swarm forge: seeded agent.nix + flake.nix");
@ -791,7 +824,7 @@ impl Client {
async fn file_exists(&self, repo: &str, path: &str) -> Result<bool> { async fn file_exists(&self, repo: &str, path: &str) -> Result<bool> {
let is_empty = self let is_empty = self
.api .api
.repo_get(CONFIG_ORG, repo) .call(|api| api.repo_get(CONFIG_ORG, repo).into_future())
.await .await
.with_context(|| format!("get repo {CONFIG_ORG}/{repo}"))? .with_context(|| format!("get repo {CONFIG_ORG}/{repo}"))?
.empty .empty
@ -801,7 +834,10 @@ impl Client {
} }
match self match self
.api .api
.repo_get_contents(CONFIG_ORG, repo, path, RepoGetContentsQuery::default()) .call(|api| {
api.repo_get_contents(CONFIG_ORG, repo, path, RepoGetContentsQuery::default())
.into_future()
})
.await .await
{ {
Ok(_) => Ok(true), Ok(_) => Ok(true),
@ -822,8 +858,7 @@ impl Client {
) -> Result<std::collections::HashMap<String, ConfigPrStatus>> { ) -> Result<std::collections::HashMap<String, ConfigPrStatus>> {
let repos = self let repos = self
.api .api
.org_list_repos(CONFIG_ORG) .call(|api| Box::pin(api.org_list_repos(CONFIG_ORG).all()))
.all()
.await .await
.with_context(|| format!("list repos in {CONFIG_ORG}"))?; .with_context(|| format!("list repos in {CONFIG_ORG}"))?;
@ -843,8 +878,12 @@ impl Client {
}; };
let prs = match self let prs = match self
.api .api
.repo_list_pull_requests(CONFIG_ORG, agent, query) .call(|api| {
.all() Box::pin(
api.repo_list_pull_requests(CONFIG_ORG, agent, query.clone())
.all(),
)
})
.await .await
{ {
Ok(prs) => prs, Ok(prs) => prs,
@ -953,9 +992,12 @@ impl Client {
let repos = Self::page_search_results(PAGE_SIZE, MAX_PAGES, |page| async move { let repos = Self::page_search_results(PAGE_SIZE, MAX_PAGES, |page| async move {
Ok(self Ok(self
.api .api
.repo_search(RepoSearchQuery::default()) .call(|api| {
.page(page) api.repo_search(RepoSearchQuery::default())
.page_size(PAGE_SIZE) .page(page)
.page_size(PAGE_SIZE)
.into_future()
})
.await .await
.context("search repos")? .context("search repos")?
.data) .data)
@ -979,8 +1021,12 @@ impl Client {
}; };
let pulls = self let pulls = self
.api .api
.repo_list_pull_requests(owner, repo, query) .call(|api| {
.all() Box::pin(
api.repo_list_pull_requests(owner, repo, query.clone())
.all(),
)
})
.await .await
.with_context(|| format!("list open pull requests in {owner}/{repo}"))?; .with_context(|| format!("list open pull requests in {owner}/{repo}"))?;
Ok(open_pulls(pulls)) Ok(open_pulls(pulls))
@ -1019,16 +1065,20 @@ impl Client {
let issues = self let issues = self
.api .api
.issue_list_issues( .call(|api| {
owner, Box::pin(
repo, api.issue_list_issues(
IssueListIssuesQuery { owner,
state: Some(IssueListIssuesQueryState::Open), repo,
r#type: Some(IssueListIssuesQueryType::Issues), IssueListIssuesQuery {
..Default::default() state: Some(IssueListIssuesQueryState::Open),
}, r#type: Some(IssueListIssuesQueryType::Issues),
) ..Default::default()
.all() },
)
.all(),
)
})
.await .await
.with_context(|| format!("list open issues in {owner}/{repo}"))?; .with_context(|| format!("list open issues in {owner}/{repo}"))?;
@ -1045,7 +1095,10 @@ impl Client {
.map(|number| async move { .map(|number| async move {
let blockers = self let blockers = self
.api .api
.issue_list_issue_dependencies(owner, repo, number) .call(|api| {
api.issue_list_issue_dependencies(owner, repo, number)
.into_future()
})
.await .await
.with_context(|| format!("list dependencies of {owner}/{repo}#{number}"))?; .with_context(|| format!("list dependencies of {owner}/{repo}#{number}"))?;
Ok::<_, anyhow::Error>((number, blockers)) Ok::<_, anyhow::Error>((number, blockers))
@ -1194,7 +1247,7 @@ impl Client {
) -> Result<PullForReview> { ) -> Result<PullForReview> {
let pull = self let pull = self
.api .api
.repo_get_pull_request(owner, repo, number) .call(|api| api.repo_get_pull_request(owner, repo, number).into_future())
.await .await
.with_context(|| format!("read PR {number} of {owner}/{repo}"))?; .with_context(|| format!("read PR {number} of {owner}/{repo}"))?;
let head_sha = pull let head_sha = pull
@ -1203,16 +1256,20 @@ impl Client {
.with_context(|| format!("PR {number} of {owner}/{repo} names no head commit"))?; .with_context(|| format!("PR {number} of {owner}/{repo} names no head commit"))?;
let commits = self let commits = self
.api .api
.repo_get_pull_request_commits( .call(|api| {
owner, Box::pin(
repo, api.repo_get_pull_request_commits(
number, owner,
RepoGetPullRequestCommitsQuery { repo,
verification: Some(false), number,
files: Some(false), RepoGetPullRequestCommitsQuery {
}, verification: Some(false),
) files: Some(false),
.all() },
)
.all(),
)
})
.await .await
.with_context(|| format!("list the commits of PR {number} of {owner}/{repo}"))? .with_context(|| format!("list the commits of PR {number} of {owner}/{repo}"))?
.into_iter() .into_iter()
@ -1223,13 +1280,16 @@ impl Client {
.collect(); .collect();
let diff = self let diff = self
.api .api
.repo_download_pull_diff_or_patch( .call(|api| {
owner, api.repo_download_pull_diff_or_patch(
repo, owner,
number, repo,
"diff", number,
RepoDownloadPullDiffOrPatchQuery::default(), "diff",
) RepoDownloadPullDiffOrPatchQuery::default(),
)
.into_future()
})
.await .await
.with_context(|| format!("read the diff of PR {number} of {owner}/{repo}"))?; .with_context(|| format!("read the diff of PR {number} of {owner}/{repo}"))?;
Ok(PullForReview { Ok(PullForReview {
@ -1259,7 +1319,7 @@ impl Client {
) -> Result<Option<DefaultBranchFiles>> { ) -> Result<Option<DefaultBranchFiles>> {
let info = self let info = self
.api .api
.repo_get(owner, repo) .call(|api| api.repo_get(owner, repo).into_future())
.await .await
.with_context(|| format!("get repo {owner}/{repo}"))?; .with_context(|| format!("get repo {owner}/{repo}"))?;
if info.empty.unwrap_or(false) { if info.empty.unwrap_or(false) {
@ -1268,7 +1328,11 @@ impl Client {
let branch = info let branch = info
.default_branch .default_branch
.with_context(|| format!("{owner}/{repo} names no default branch"))?; .with_context(|| format!("{owner}/{repo} names no default branch"))?;
let head = match self.api.repo_get_branch(owner, repo, &branch).await { let head = match self
.api
.call(|api| api.repo_get_branch(owner, repo, &branch).into_future())
.await
{
Ok(head) => head, Ok(head) => head,
Err(e) if is_repo_not_found(&e) => return Ok(None), Err(e) if is_repo_not_found(&e) => return Ok(None),
Err(e) => { Err(e) => {
@ -1339,7 +1403,10 @@ impl Client {
if !trees.contains_key(sha) { if !trees.contains_key(sha) {
let listing = self let listing = self
.api .api
.get_tree(owner, repo, sha, GetTreeQuery::default()) .call(|api| {
api.get_tree(owner, repo, sha, GetTreeQuery::default())
.into_future()
})
.await .await
.map_err(|e| anyhow::anyhow!("read tree {sha} of {owner}/{repo}: {e}"))?; .map_err(|e| anyhow::anyhow!("read tree {sha} of {owner}/{repo}: {e}"))?;
if listing.truncated.unwrap_or(false) { if listing.truncated.unwrap_or(false) {
@ -1359,7 +1426,7 @@ impl Client {
let blob = self let blob = self
.api .api
.get_blob(owner, repo, sha) .call(|api| api.get_blob(owner, repo, sha).into_future())
.await .await
.map_err(|e| anyhow::anyhow!("read blob {sha} of {owner}/{repo}: {e}"))?; .map_err(|e| anyhow::anyhow!("read blob {sha} of {owner}/{repo}: {e}"))?;
anyhow::ensure!( anyhow::ensure!(
@ -1568,11 +1635,17 @@ impl HookScope<'_> {
} }
/// The `url` config value of every hook currently on this scope. /// The `url` config value of every hook currently on this scope.
async fn list_hook_urls(&self, api: &Forgejo) -> Result<Vec<String>, ForgejoError> { async fn list_hook_urls(&self, api: &api::Api) -> Result<Vec<String>, ForgejoError> {
let hooks = match self { let hooks = match self {
Self::Repo { org, repo } => api.repo_list_hooks(org, repo).all().await?, Self::Repo { org, repo } => {
Self::Org { org } => api.org_list_hooks(org).send().await?, api.call(|api| Box::pin(api.repo_list_hooks(org, repo).all()))
Self::Instance => api.admin_list_hooks().await?, .await?
}
Self::Org { org } => {
api.call(|api| api.org_list_hooks(org).into_future())
.await?
}
Self::Instance => api.call(|api| api.admin_list_hooks().into_future()).await?,
}; };
Ok(hooks Ok(hooks
.iter() .iter()
@ -1580,11 +1653,24 @@ impl HookScope<'_> {
.collect()) .collect())
} }
async fn create_hook(&self, api: &Forgejo, hook: CreateHookOption) -> Result<(), ForgejoError> { async fn create_hook(
&self,
api: &api::Api,
hook: CreateHookOption,
) -> Result<(), ForgejoError> {
match self { match self {
Self::Repo { org, repo } => api.repo_create_hook(org, repo, hook).await.map(drop), Self::Repo { org, repo } => api
Self::Org { org } => api.org_create_hook(org, hook).await.map(drop), .call(|api| api.repo_create_hook(org, repo, hook.clone()).into_future())
Self::Instance => api.admin_create_hook(hook).await.map(drop), .await
.map(drop),
Self::Org { org } => api
.call(|api| api.org_create_hook(org, hook.clone()).into_future())
.await
.map(drop),
Self::Instance => api
.call(|api| api.admin_create_hook(hook.clone()).into_future())
.await
.map(drop),
} }
} }
} }
@ -1765,7 +1851,7 @@ fn conflict_error() -> ForgejoError {
impl Client { impl Client {
/// A client for a test's own forge at `url`, holding a placeholder token. /// A client for a test's own forge at `url`, holding a placeholder token.
pub(crate) fn for_test_forge(url: url::Url) -> Self { pub(crate) fn for_test_forge(url: url::Url) -> Self {
let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap(); let api = crate::forge::api::tests::fixed(url.clone(), "stub-token");
Self { api, url } Self { api, url }
} }
} }
@ -2041,7 +2127,7 @@ mod tests {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
tokio::spawn(async move { axum::serve(listener, app).await }); tokio::spawn(async move { axum::serve(listener, app).await });
let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap(); let api = crate::forge::api::tests::fixed(url.clone(), "stub-token");
(Client { api, url }, seen) (Client { api, url }, seen)
} }
@ -2328,7 +2414,7 @@ mod tests {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
tokio::spawn(async move { axum::serve(listener, app).await }); tokio::spawn(async move { axum::serve(listener, app).await });
let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap(); let api = crate::forge::api::tests::fixed(url.clone(), "stub-token");
(Client { api, url }, seen) (Client { api, url }, seen)
} }

31
swarm-controller/src/forge/agent_token.rs
View file

@ -180,7 +180,11 @@ impl Client {
/// Every access token the forge lists for `agent`, or `None` when the /// Every access token the forge lists for `agent`, or `None` when the
/// forge has no such user. /// forge has no such user.
pub(super) async fn list_agent_tokens(&self, agent: &str) -> Result<Option<Vec<AccessToken>>> { pub(super) async fn list_agent_tokens(&self, agent: &str) -> Result<Option<Vec<AccessToken>>> {
match self.api.admin_list_user_access_tokens(agent).all().await { match self
.api
.call(|api| Box::pin(api.admin_list_user_access_tokens(agent).all()))
.await
{
Ok(tokens) => Ok(Some(tokens)), Ok(tokens) => Ok(Some(tokens)),
Err(e) if is_not_found(&e) => Ok(None), Err(e) if is_not_found(&e) => Ok(None),
Err(e) => Err(e).with_context(|| format!("listing {agent}'s forge tokens")), Err(e) => Err(e).with_context(|| format!("listing {agent}'s forge tokens")),
@ -268,7 +272,11 @@ impl Client {
path: &str, path: &str,
) -> Result<forge::Credential> { ) -> Result<forge::Credential> {
if listed.iter().any(|t| t.name.as_deref() == Some(name)) { if listed.iter().any(|t| t.name.as_deref() == Some(name)) {
match self.api.admin_delete_user_access_token(user, name).await { match self
.api
.call(|api| api.admin_delete_user_access_token(user, name).into_future())
.await
{
Ok(()) => {} Ok(()) => {}
Err(e) if is_not_found(&e) => {} Err(e) if is_not_found(&e) => {}
Err(e) => { Err(e) => {
@ -278,14 +286,17 @@ impl Client {
} }
let created = self let created = self
.api .api
.admin_create_user_access_token( .call(|api| {
user, api.admin_create_user_access_token(
CreateAccessTokenOption { user,
name: name.to_owned(), CreateAccessTokenOption {
repositories: None, name: name.to_owned(),
scopes: Some(scopes.split(',').map(str::to_owned).collect()), repositories: None,
}, scopes: Some(scopes.split(',').map(str::to_owned).collect()),
) },
)
.into_future()
})
.await .await
.with_context(|| format!("creating {user}'s {name} token"))?; .with_context(|| format!("creating {user}'s {name} token"))?;
let Some(value) = created.sha1.filter(|v| !v.is_empty()) else { let Some(value) = created.sha1.filter(|v| !v.is_empty()) else {

261
swarm-controller/src/forge/api.rs Normal file
View file

@ -0,0 +1,261 @@
//! The forge API under the controller's token, re-read from the swarm secret
//! store when the forge refuses it.
//!
//! [`Api::call`] runs one request. When the forge answers 401, it asks the
//! [`StoredSecret`] for a token to try instead and runs the request once more
//! with it; a second 401 is returned as it came. The `StoredSecret` runs one
//! store read at a time and backs off after a read that brings no new token,
//! so a burst of refusals costs the store one read.
use std::sync::{Arc, Mutex};
use forgejo_api::{ApiErrorKind, Auth, Forgejo, ForgejoError};
use futures_util::future::BoxFuture;
use reqwest::StatusCode;
use swarm_queue_client::stored_secret::StoredSecret;
/// One forge request, as [`Api::call`] runs it.
pub(crate) type Call<'a, R> = BoxFuture<'a, Result<R, ForgejoError>>;
pub(crate) struct Api {
url: url::Url,
token: StoredSecret,
/// A client sending the token it is paired with. Rebuilt when `token`
/// holds a different one.
built: Mutex<(String, Arc<Forgejo>)>,
}
impl Api {
/// A client for the forge at `url`, sending `token`'s current value.
///
/// # Errors
/// `url` is not http(s), or the token is not ASCII.
pub(crate) fn new(url: url::Url, token: StoredSecret) -> Result<Self, ForgejoError> {
let current = token.current();
let api = Arc::new(Forgejo::new(Auth::Token(&current), url.clone())?);
Ok(Self {
url,
token,
built: Mutex::new((current, api)),
})
}
/// The token in use and a client sending it.
fn current(&self) -> Result<(String, Arc<Forgejo>), ForgejoError> {
let token = self.token.current();
let mut built = self.built.lock().expect("unpoisoned");
if built.0 != token {
let api = Arc::new(Forgejo::new(Auth::Token(&token), self.url.clone())?);
*built = (token.clone(), api);
}
Ok((token, Arc::clone(&built.1)))
}
/// `request` against the forge, run a second time with a re-read token
/// when the forge refuses the first. The forge acts on nothing it answers
/// 401 to, so running a non-idempotent request again is safe.
///
/// Boxed, so a caller awaiting several of these does not carry each one's
/// `R` inline.
pub(crate) fn call<'s, R, F>(&'s self, request: F) -> Call<'s, R>
where
F: Fn(&Forgejo) -> Call<'_, R> + Send + 's,
R: Send + 's,
{
Box::pin(async move {
let (token, api) = self.current()?;
match request(&api).await {
Err(e) if refuses_the_token(&e) => {
if self.token.after_refusal(&token).await.is_none() {
return Err(e);
}
let (_, api) = self.current()?;
request(&api).await
}
done => done,
}
})
}
}
/// The HTTP status `e` reports, for the shapes a 401 can arrive in: endpoints
/// that declare a 401 parse its body into [`ApiErrorKind::Unauthorized`], the
/// rest report the bare status.
fn status(e: &ForgejoError) -> Option<StatusCode> {
match e {
ForgejoError::UnexpectedStatusCode(s) => Some(*s),
ForgejoError::ApiError(e) => match e.error_kind() {
ApiErrorKind::Unauthorized => Some(StatusCode::UNAUTHORIZED),
ApiErrorKind::Other(s) => Some(*s),
_ => None,
},
_ => None,
}
}
/// Whether the forge refused the token. Only a 401: to the forge a 400 is a
/// malformed request, and a re-read it started would find the same token and
/// put a real 401 right after it inside the re-read backoff.
fn refuses_the_token(e: &ForgejoError) -> bool {
status(e) == Some(StatusCode::UNAUTHORIZED)
}
#[cfg(test)]
pub(crate) mod tests {
use std::sync::atomic::{AtomicUsize, Ordering};
use super::*;
/// A forge answering 200 to `accepted` and 401 to any other token,
/// counting the requests it serves. The `held`-th request to arrive, from
/// 0, is answered only after one with `accepted` is.
async fn forge_accepting(
accepted: &'static str,
held: Option<usize>,
) -> (url::Url, Arc<AtomicUsize>) {
use axum::http::{HeaderMap, StatusCode, header::AUTHORIZATION, header::CONTENT_TYPE};
let served = Arc::new(AtomicUsize::new(0));
let counted = Arc::clone(&served);
let (accepted_once, seen_accepted) = tokio::sync::watch::channel(false);
let app = axum::Router::new().fallback(move |headers: HeaderMap| {
let counted = Arc::clone(&counted);
let (accepted_once, mut seen_accepted) = (accepted_once.clone(), seen_accepted.clone());
async move {
let n = counted.fetch_add(1, Ordering::SeqCst);
if held == Some(n) {
seen_accepted.wait_for(|seen| *seen).await.unwrap();
}
let sent = headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok());
if sent == Some(&format!("token {accepted}")) {
accepted_once.send_replace(true);
(StatusCode::OK, [(CONTENT_TYPE, "application/json")], "{}")
} else {
let body = r#"{"message":"token is required"}"#;
(
StatusCode::UNAUTHORIZED,
[(CONTENT_TYPE, "application/json")],
body,
)
}
}
});
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
tokio::spawn(async move { axum::serve(listener, app).await });
(url, served)
}
/// A token starting at `initial` whose store holds `stored`, counting the
/// reads, each of which takes a second.
fn token(initial: &str, stored: &'static str) -> (StoredSecret, Arc<AtomicUsize>) {
let reads = Arc::new(AtomicUsize::new(0));
let counted = Arc::clone(&reads);
let secret = StoredSecret::new("test", initial.to_owned(), move || {
let counted = Arc::clone(&counted);
async move {
counted.fetch_add(1, Ordering::SeqCst);
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
Ok(stored.to_owned())
}
});
(secret, reads)
}
/// A client for `url` holding `token`, whose store read always fails.
pub(crate) fn fixed(url: url::Url, token: &str) -> Api {
let token = StoredSecret::new("test", token.to_owned(), || async {
Err("no store in this test".into())
});
Api::new(url, token).unwrap()
}
async fn whoami(api: &Api) -> Result<forgejo_api::structs::User, ForgejoError> {
api.call(|api| api.user_get_current().into_future()).await
}
#[tokio::test(start_paused = true)]
async fn a_refused_token_is_re_read_and_the_request_sent_again_with_the_new_one() {
let (url, served) = forge_accepting("new", None).await;
let (token, reads) = token("old", "new");
let api = Api::new(url, token).unwrap();
whoami(&api).await.expect("accepted with the re-read token");
assert_eq!(reads.load(Ordering::SeqCst), 1);
assert_eq!(served.load(Ordering::SeqCst), 2);
whoami(&api).await.expect("the new token stays in use");
assert_eq!(reads.load(Ordering::SeqCst), 1);
assert_eq!(served.load(Ordering::SeqCst), 3);
}
#[tokio::test(start_paused = true)]
async fn a_re_read_token_the_forge_also_refuses_is_returned_as_the_error() {
let (url, served) = forge_accepting("neither", None).await;
let (token, reads) = token("old", "new");
let api = Api::new(url, token).unwrap();
let err = whoami(&api).await.expect_err("both tokens refused");
assert!(refuses_the_token(&err), "{err}");
assert_eq!(reads.load(Ordering::SeqCst), 1);
assert_eq!(served.load(Ordering::SeqCst), 2);
}
#[tokio::test(start_paused = true)]
async fn a_refused_token_still_in_the_store_is_not_sent_again() {
let (url, served) = forge_accepting("new", None).await;
let (token, reads) = token("old", "old");
let api = Api::new(url, token).unwrap();
let err = whoami(&api).await.expect_err("no other token to try");
assert!(refuses_the_token(&err), "{err}");
assert_eq!(reads.load(Ordering::SeqCst), 1);
assert_eq!(served.load(Ordering::SeqCst), 1);
}
#[tokio::test(start_paused = true)]
async fn concurrent_refusals_share_one_store_read() {
let (url, served) = forge_accepting("new", None).await;
let (token, reads) = token("old", "new");
let api = Api::new(url, token).unwrap();
let all = futures_util::future::join_all((0..5).map(|_| whoami(&api))).await;
assert!(all.iter().all(Result::is_ok), "{all:?}");
assert_eq!(reads.load(Ordering::SeqCst), 1);
assert_eq!(served.load(Ordering::SeqCst), 10);
}
// Real time: the held request waits on a socket, and paused time would
// jump the bound below while it does.
#[tokio::test]
async fn a_refusal_landing_after_the_re_read_is_sent_again_without_another_read() {
let (url, served) = forge_accepting("new", Some(1)).await;
let (token, reads) = token("old", "new");
let api = Api::new(url, token).unwrap();
let both = async { tokio::join!(whoami(&api), whoami(&api)) };
let (a, b) = tokio::time::timeout(std::time::Duration::from_secs(10), both)
.await
.expect("the held request is answered once the other is accepted");
a.expect("accepted with the re-read token");
b.expect("accepted with the re-read token");
assert_eq!(reads.load(Ordering::SeqCst), 1);
assert_eq!(served.load(Ordering::SeqCst), 4);
}
#[test]
fn only_a_401_counts_as_refusing_the_token() {
let unauthorized = forgejo_api::ApiError {
message: None,
kind: ApiErrorKind::Unauthorized,
};
assert!(refuses_the_token(&ForgejoError::ApiError(unauthorized)));
assert!(refuses_the_token(&ForgejoError::UnexpectedStatusCode(
StatusCode::UNAUTHORIZED
)));
let other = forgejo_api::ApiError {
message: None,
kind: ApiErrorKind::Other(StatusCode::UNAUTHORIZED),
};
assert!(refuses_the_token(&ForgejoError::ApiError(other)));
for refused in [StatusCode::BAD_REQUEST, StatusCode::FORBIDDEN] {
assert!(!refuses_the_token(&ForgejoError::UnexpectedStatusCode(
refused
)));
}
}
}

7
swarm-controller/src/forge/classifier_account.rs
View file

@ -176,7 +176,7 @@ impl Client {
self.create_user(LOGIN, EMAIL).await?; self.create_user(LOGIN, EMAIL).await?;
let user = self let user = self
.api .api
.user_get(LOGIN) .call(|api| api.user_get(LOGIN).into_future())
.await .await
.with_context(|| format!("read forge user {LOGIN}"))?; .with_context(|| format!("read forge user {LOGIN}"))?;
if user.email.as_deref() != Some(EMAIL) { if user.email.as_deref() != Some(EMAIL) {
@ -187,7 +187,10 @@ impl Client {
); );
} }
self.api self.api
.admin_edit_user(LOGIN, super::repo_creation_lockdown(LOGIN)) .call(|api| {
api.admin_edit_user(LOGIN, super::repo_creation_lockdown(LOGIN))
.into_future()
})
.await .await
.with_context(|| format!("disable repo creation for forge user {LOGIN}"))?; .with_context(|| format!("disable repo creation for forge user {LOGIN}"))?;
Ok(()) Ok(())

19
swarm-controller/src/forge/docs_grant.rs
View file

@ -61,7 +61,10 @@ impl Client {
pub async fn grant_docs_read(&self, agent: &str) -> Result<()> { pub async fn grant_docs_read(&self, agent: &str) -> Result<()> {
match self match self
.api .api
.repo_get_repo_permissions(SHARED_ORG, SHARED_DOCS_REPO, agent) .call(|api| {
api.repo_get_repo_permissions(SHARED_ORG, SHARED_DOCS_REPO, agent)
.into_future()
})
.await .await
{ {
Ok(p) if holds_read(&p) => return Ok(()), Ok(p) if holds_read(&p) => return Ok(()),
@ -86,7 +89,10 @@ impl Client {
async fn observe_grant(&self, agent: &str) -> Observed { async fn observe_grant(&self, agent: &str) -> Observed {
match self match self
.api .api
.repo_check_collaborator(SHARED_ORG, SHARED_DOCS_REPO, agent) .call(|api| {
api.repo_check_collaborator(SHARED_ORG, SHARED_DOCS_REPO, agent)
.into_future()
})
.await .await
{ {
Ok(()) => Observed::Granted, Ok(()) => Observed::Granted,
@ -102,7 +108,11 @@ impl Client {
/// say. A missing repo makes every collaborator check answer 404 too, so /// say. A missing repo makes every collaborator check answer 404 too, so
/// it has to be told apart from "not a collaborator" before they run. /// it has to be told apart from "not a collaborator" before they run.
async fn docs_repo_exists(&self) -> Option<bool> { async fn docs_repo_exists(&self) -> Option<bool> {
match self.api.repo_get(SHARED_ORG, SHARED_DOCS_REPO).await { match self
.api
.call(|api| api.repo_get(SHARED_ORG, SHARED_DOCS_REPO).into_future())
.await
{
Ok(_) => Some(true), Ok(_) => Some(true),
Err(e) if is_not_found(&e) => Some(false), Err(e) if is_not_found(&e) => Some(false),
Err(e) => { Err(e) => {
@ -206,7 +216,6 @@ mod tests {
use axum::extract::{Path, State}; use axum::extract::{Path, State};
use axum::http::{StatusCode, header::CONTENT_TYPE}; use axum::http::{StatusCode, header::CONTENT_TYPE};
use forgejo_api::{Auth, Forgejo};
use super::*; use super::*;
@ -255,7 +264,7 @@ mod tests {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
tokio::spawn(async move { axum::serve(listener, app).await }); tokio::spawn(async move { axum::serve(listener, app).await });
let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap(); let api = crate::forge::api::tests::fixed(url.clone(), "stub-token");
(Client { api, url }, docs) (Client { api, url }, docs)
} }

54
swarm-controller/src/forge/hive_token.rs
View file

@ -145,7 +145,7 @@ impl Client {
self.ensure_hive_user(hive, &login).await?; self.ensure_hive_user(hive, &login).await?;
let team = self.ensure_hives_team().await?; let team = self.ensure_hives_team().await?;
self.api self.api
.org_add_team_member(team, &login) .call(|api| api.org_add_team_member(team, &login).into_future())
.await .await
.with_context(|| format!("adding {login} to {CONFIG_ORG}/{HIVES_TEAM}"))?; .with_context(|| format!("adding {login} to {CONFIG_ORG}/{HIVES_TEAM}"))?;
@ -214,7 +214,7 @@ impl Client {
self.create_user(login, &hive_email(hive)).await?; self.create_user(login, &hive_email(hive)).await?;
let user = self let user = self
.api .api
.user_get(login) .call(|api| api.user_get(login).into_future())
.await .await
.with_context(|| format!("read forge user {login}"))?; .with_context(|| format!("read forge user {login}"))?;
if !is_hive_account(hive, user.email.as_deref()) { if !is_hive_account(hive, user.email.as_deref()) {
@ -226,7 +226,10 @@ impl Client {
); );
} }
self.api self.api
.admin_edit_user(login, super::repo_creation_lockdown(login)) .call(|api| {
api.admin_edit_user(login, super::repo_creation_lockdown(login))
.into_future()
})
.await .await
.with_context(|| format!("disable repo creation for forge user {login}"))?; .with_context(|| format!("disable repo creation for forge user {login}"))?;
Ok(()) Ok(())
@ -240,18 +243,21 @@ impl Client {
TeamSeen::Matches(id) => Ok(id), TeamSeen::Matches(id) => Ok(id),
TeamSeen::Drifted(id) => { TeamSeen::Drifted(id) => {
self.api self.api
.org_edit_team( .call(|api| {
id, api.org_edit_team(
EditTeamOption { id,
can_create_org_repo: Some(false), EditTeamOption {
description: Some(HIVES_TEAM_DESCRIPTION.to_owned()), can_create_org_repo: Some(false),
includes_all_repositories: Some(true), description: Some(HIVES_TEAM_DESCRIPTION.to_owned()),
name: HIVES_TEAM.to_owned(), includes_all_repositories: Some(true),
permission: Some(EditTeamOptionPermission::Read), name: HIVES_TEAM.to_owned(),
units: Some(HIVES_TEAM_UNITS.map(str::to_owned).to_vec()), permission: Some(EditTeamOptionPermission::Read),
units_map: None, units: Some(HIVES_TEAM_UNITS.map(str::to_owned).to_vec()),
}, units_map: None,
) },
)
.into_future()
})
.await .await
.with_context(|| { .with_context(|| {
format!("reconcile {CONFIG_ORG}/{HIVES_TEAM} team settings") format!("reconcile {CONFIG_ORG}/{HIVES_TEAM} team settings")
@ -274,7 +280,11 @@ impl Client {
units: Some(HIVES_TEAM_UNITS.map(str::to_owned).to_vec()), units: Some(HIVES_TEAM_UNITS.map(str::to_owned).to_vec()),
units_map: None, units_map: None,
}; };
match self.api.org_create_team(CONFIG_ORG, team).await { match self
.api
.call(|api| api.org_create_team(CONFIG_ORG, team.clone()).into_future())
.await
{
Ok(created) => { Ok(created) => {
tracing::info!("swarm forge: created {CONFIG_ORG}/{HIVES_TEAM} team"); tracing::info!("swarm forge: created {CONFIG_ORG}/{HIVES_TEAM} team");
created created
@ -293,7 +303,11 @@ impl Client {
} }
pub(super) async fn observe_hives_team(&self) -> TeamSeen { pub(super) async fn observe_hives_team(&self) -> TeamSeen {
match self.api.org_list_teams(CONFIG_ORG).await { match self
.api
.call(|api| api.org_list_teams(CONFIG_ORG).into_future())
.await
{
Ok((_, teams)) => teams Ok((_, teams)) => teams
.into_iter() .into_iter()
.find(|t| t.name.as_deref() == Some(HIVES_TEAM)) .find(|t| t.name.as_deref() == Some(HIVES_TEAM))
@ -342,7 +356,11 @@ impl Client {
}; };
let member = match team { let member = match team {
TeamSeen::Matches(id) | TeamSeen::Drifted(id) => { TeamSeen::Matches(id) | TeamSeen::Drifted(id) => {
match self.api.org_list_team_member(id, &login).await { match self
.api
.call(|api| api.org_list_team_member(id, &login).into_future())
.await
{
Ok(_) => Some(true), Ok(_) => Some(true),
Err(e) if is_not_found(&e) => Some(false), Err(e) if is_not_found(&e) => Some(false),
Err(e) => { Err(e) => {

10
swarm-controller/src/forge/legacy_tokens.rs
View file

@ -68,7 +68,11 @@ impl Client {
.filter_map(|t| t.name.as_deref()) .filter_map(|t| t.name.as_deref())
.filter(|n| is_legacy_token_name(n)) .filter(|n| is_legacy_token_name(n))
{ {
match self.api.admin_delete_user_access_token(user, name).await { match self
.api
.call(|api| api.admin_delete_user_access_token(user, name).into_future())
.await
{
Ok(()) => removed += 1, Ok(()) => removed += 1,
Err(e) if super::agent_token::is_not_found(&e) => {} Err(e) if super::agent_token::is_not_found(&e) => {}
Err(e) => tracing::warn!( Err(e) => tracing::warn!(
@ -113,8 +117,6 @@ mod tests {
use std::collections::BTreeMap; use std::collections::BTreeMap;
use std::sync::{Arc, Mutex}; use std::sync::{Arc, Mutex};
use forgejo_api::{Auth, Forgejo};
use super::super::agent_token::{AGENT_TOKEN_NAME, MintReason}; use super::super::agent_token::{AGENT_TOKEN_NAME, MintReason};
use super::*; use super::*;
@ -229,7 +231,7 @@ mod tests {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap(); let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
tokio::spawn(async move { axum::serve(listener, app).await }); tokio::spawn(async move { axum::serve(listener, app).await });
let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap(); let api = crate::forge::api::tests::fixed(url.clone(), "stub-token");
(Client { api, url }, deletes) (Client { api, url }, deletes)
} }

140
swarm-controller/src/forge/objects.rs
View file

@ -642,7 +642,11 @@ impl Client {
async fn observe(&self, desired: &Desired) -> Observed { async fn observe(&self, desired: &Desired) -> Observed {
let mut observed = Observed::default(); let mut observed = Observed::default();
for org in &desired.orgs { for org in &desired.orgs {
let org_seen = seen(self.api.org_get(org).await, &format!("org {org}"), drop); let org_seen = seen(
self.api.call(|api| api.org_get(org).into_future()).await,
&format!("org {org}"),
drop,
);
observed.orgs.insert(org.clone(), org_seen); observed.orgs.insert(org.clone(), org_seen);
} }
for org in desired.team_orgs() { for org in desired.team_orgs() {
@ -650,7 +654,9 @@ impl Client {
continue; continue;
} }
let teams = seen( let teams = seen(
self.api.org_list_teams(org).await, self.api
.call(|api| api.org_list_teams(org).into_future())
.await,
&format!("teams of {org}"), &format!("teams of {org}"),
|(_headers, teams)| teams, |(_headers, teams)| teams,
); );
@ -685,7 +691,9 @@ impl Client {
continue; continue;
} }
let repo = seen( let repo = seen(
self.api.repo_get(&owner, &name).await, self.api
.call(|api| api.repo_get(&owner, &name).into_future())
.await,
&format!("repo {owner}/{name}"), &format!("repo {owner}/{name}"),
|r| RepoState { |r| RepoState {
private: r.private.unwrap_or(false), private: r.private.unwrap_or(false),
@ -704,7 +712,11 @@ impl Client {
/// Read every config repo's `main` rule into `observed.config_rules`. /// Read every config repo's `main` rule into `observed.config_rules`.
async fn observe_config_rules(&self, observed: &mut Observed) { async fn observe_config_rules(&self, observed: &mut Observed) {
let repos = match self.api.org_list_repos(CONFIG_ORG).all().await { let repos = match self
.api
.call(|api| Box::pin(api.org_list_repos(CONFIG_ORG).all()))
.await
{
Ok(repos) => repos, Ok(repos) => repos,
Err(e) => { Err(e) => {
tracing::warn!(error = %e, "swarm forge objects: listing {CONFIG_ORG} repos failed; skipping their branch rules this pass"); tracing::warn!(error = %e, "swarm forge objects: listing {CONFIG_ORG} repos failed; skipping their branch rules this pass");
@ -715,7 +727,10 @@ impl Client {
for name in repos.into_iter().filter_map(|r| r.name) { for name in repos.into_iter().filter_map(|r| r.name) {
let rule = seen( let rule = seen(
self.api self.api
.repo_get_branch_protection(CONFIG_ORG, &name, "main") .call(|api| {
api.repo_get_branch_protection(CONFIG_ORG, &name, "main")
.into_future()
})
.await, .await,
&format!("branch protection of {CONFIG_ORG}/{name}"), &format!("branch protection of {CONFIG_ORG}/{name}"),
|rule| config_rule_matches(&rule), |rule| config_rule_matches(&rule),
@ -759,7 +774,15 @@ impl Client {
Action::ReconcileTeam { org, id } => self.reconcile_operators_team(org, *id).await, Action::ReconcileTeam { org, id } => self.reconcile_operators_team(org, *id).await,
Action::ConvergeConfigRule { repo } => { Action::ConvergeConfigRule { repo } => {
self.api self.api
.repo_edit_branch_protection(CONFIG_ORG, repo, "main", config_rule_edit()) .call(|api| {
api.repo_edit_branch_protection(
CONFIG_ORG,
repo,
"main",
config_rule_edit(),
)
.into_future()
})
.await .await
.with_context(|| format!("converge {CONFIG_ORG}/{repo} main branch rule"))?; .with_context(|| format!("converge {CONFIG_ORG}/{repo} main branch rule"))?;
tracing::info!(%repo, "swarm forge objects: config repo merge gate converged"); tracing::info!(%repo, "swarm forge objects: config repo merge gate converged");
@ -774,7 +797,7 @@ impl Client {
let mut edit = sparse_edit_repo_option(); let mut edit = sparse_edit_repo_option();
edit.private = Some(false); edit.private = Some(false);
self.api self.api
.repo_edit(owner, name, edit) .call(|api| api.repo_edit(owner, name, edit.clone()).into_future())
.await .await
.with_context(|| format!("edit {owner}/{name} (set public)"))?; .with_context(|| format!("edit {owner}/{name} (set public)"))?;
tracing::info!(%owner, %name, "swarm forge objects: repo set to public"); tracing::info!(%owner, %name, "swarm forge objects: repo set to public");
@ -790,7 +813,7 @@ impl Client {
let mut edit = sparse_edit_repo_option(); let mut edit = sparse_edit_repo_option();
edit.mirror_interval = Some(MIRROR_INTERVAL.to_owned()); edit.mirror_interval = Some(MIRROR_INTERVAL.to_owned());
self.api self.api
.repo_edit(owner, repo, edit) .call(|api| api.repo_edit(owner, repo, edit.clone()).into_future())
.await .await
.with_context(|| format!("set mirror_interval on {owner}/{repo}"))?; .with_context(|| format!("set mirror_interval on {owner}/{repo}"))?;
tracing::info!(%owner, %repo, interval = MIRROR_INTERVAL, "swarm forge objects: pull-mirror interval updated"); tracing::info!(%owner, %repo, interval = MIRROR_INTERVAL, "swarm forge objects: pull-mirror interval updated");
@ -816,14 +839,22 @@ impl Client {
visibility: None, visibility: None,
website: None, website: None,
}; };
match self.api.org_create(option).await { match self
.api
.call(|api| api.org_create(option.clone()).into_future())
.await
{
Ok(_) => { Ok(_) => {
tracing::info!(%org, "swarm forge objects: created org"); tracing::info!(%org, "swarm forge objects: created org");
Ok(()) Ok(())
} }
Err(e) => { Err(e) => {
let existing = let existing = is_ambiguous_validation_failure(&e)
is_ambiguous_validation_failure(&e) && self.api.org_get(org).await.is_ok(); && self
.api
.call(|api| api.org_get(org).into_future())
.await
.is_ok();
if folds_into_success(&e, existing) { if folds_into_success(&e, existing) {
Ok(()) Ok(())
} else { } else {
@ -853,7 +884,11 @@ impl Client {
units: Some(OPERATORS_TEAM_UNITS.map(str::to_owned).to_vec()), units: Some(OPERATORS_TEAM_UNITS.map(str::to_owned).to_vec()),
units_map: None, units_map: None,
}; };
match self.api.org_create_team(org, team).await { match self
.api
.call(|api| api.org_create_team(org, team.clone()).into_future())
.await
{
Ok(_) => { Ok(_) => {
tracing::info!(%org, "swarm forge objects: created {OPERATORS_TEAM} team"); tracing::info!(%org, "swarm forge objects: created {OPERATORS_TEAM} team");
Ok(()) Ok(())
@ -863,11 +898,15 @@ impl Client {
// listing; the next pass reconciles its settings if they differ. // listing; the next pass reconciles its settings if they differ.
Err(e) => { Err(e) => {
let existing = is_ambiguous_validation_failure(&e) let existing = is_ambiguous_validation_failure(&e)
&& self.api.org_list_teams(org).await.is_ok_and(|(_, teams)| { && self
teams .api
.iter() .call(|api| api.org_list_teams(org).into_future())
.any(|t| t.name.as_deref() == Some(OPERATORS_TEAM)) .await
}); .is_ok_and(|(_, teams)| {
teams
.iter()
.any(|t| t.name.as_deref() == Some(OPERATORS_TEAM))
});
if folds_into_success(&e, existing) { if folds_into_success(&e, existing) {
Ok(()) Ok(())
} else { } else {
@ -891,7 +930,7 @@ impl Client {
units_map: None, units_map: None,
}; };
self.api self.api
.org_edit_team(id, edit) .call(|api| api.org_edit_team(id, edit.clone()).into_future())
.await .await
.with_context(|| format!("reconcile {org}/{OPERATORS_TEAM} team settings"))?; .with_context(|| format!("reconcile {org}/{OPERATORS_TEAM} team settings"))?;
tracing::info!(%org, "swarm forge objects: reconciled {OPERATORS_TEAM} team settings"); tracing::info!(%org, "swarm forge objects: reconciled {OPERATORS_TEAM} team settings");
@ -902,7 +941,10 @@ impl Client {
async fn create_shared_repo(&self, owner: &str, name: &str, private: bool) -> Result<()> { async fn create_shared_repo(&self, owner: &str, name: &str, private: bool) -> Result<()> {
match self match self
.api .api
.create_org_repo(owner, Self::repo_option(name, private)) .call(|api| {
api.create_org_repo(owner, Self::repo_option(name, private))
.into_future()
})
.await .await
{ {
Ok(_) => { Ok(_) => {
@ -911,7 +953,11 @@ impl Client {
} }
Err(e) => { Err(e) => {
let existing = is_ambiguous_validation_failure(&e) let existing = is_ambiguous_validation_failure(&e)
&& self.api.repo_get(owner, name).await.is_ok(); && self
.api
.call(|api| api.repo_get(owner, name).into_future())
.await
.is_ok();
if folds_into_success(&e, existing) { if folds_into_success(&e, existing) {
Ok(()) Ok(())
} else { } else {
@ -933,21 +979,24 @@ impl Client {
sha: None, sha: None,
}]; }];
self.api self.api
.repo_change_files( .call(|api| {
owner, api.repo_change_files(
name, owner,
ChangeFilesOptions { name,
author: None, ChangeFilesOptions {
branch: None, author: None,
committer: None, branch: None,
dates: None, committer: None,
files, dates: None,
force_overwrite_new_branch: None, files: files.clone(),
message: Some("init: seed README".to_owned()), force_overwrite_new_branch: None,
new_branch: None, message: Some("init: seed README".to_owned()),
signoff: None, new_branch: None,
}, signoff: None,
) },
)
.into_future()
})
.await .await
.with_context(|| format!("seed README.md in {owner}/{name}"))?; .with_context(|| format!("seed README.md in {owner}/{name}"))?;
tracing::info!(%owner, %name, "swarm forge objects: seeded README.md"); tracing::info!(%owner, %name, "swarm forge objects: seeded README.md");
@ -984,7 +1033,11 @@ impl Client {
uid: None, uid: None,
wiki: None, wiki: None,
}; };
match self.api.repo_migrate(opts).await { match self
.api
.call(|api| api.repo_migrate(opts.clone()).into_future())
.await
{
Ok(_) => { Ok(_) => {
tracing::info!(%owner, %repo, %upstream, interval = MIRROR_INTERVAL, "swarm forge objects: created pull-mirror"); tracing::info!(%owner, %repo, %upstream, interval = MIRROR_INTERVAL, "swarm forge objects: created pull-mirror");
Ok(()) Ok(())
@ -1006,12 +1059,15 @@ impl Client {
.await .await
.with_context(|| format!("read {CONFIG_ORG} avatar PNG from {}", png.display()))?; .with_context(|| format!("read {CONFIG_ORG} avatar PNG from {}", png.display()))?;
self.api self.api
.org_update_avatar( .call(|api| {
CONFIG_ORG, api.org_update_avatar(
UpdateUserAvatarOption { CONFIG_ORG,
image: Some(base64::engine::general_purpose::STANDARD.encode(&bytes)), UpdateUserAvatarOption {
}, image: Some(base64::engine::general_purpose::STANDARD.encode(&bytes)),
) },
)
.into_future()
})
.await .await
.with_context(|| format!("set {CONFIG_ORG} avatar"))?; .with_context(|| format!("set {CONFIG_ORG} avatar"))?;
if let Err(e) = std::fs::write(marker, "") { if let Err(e) = std::fs::write(marker, "") {

2
swarm-controller/src/forge/runner_token.rs
View file

@ -35,7 +35,7 @@ impl Client {
pub async fn runner_registration_token(&self) -> Result<String> { pub async fn runner_registration_token(&self) -> Result<String> {
let answer = self let answer = self
.api .api
.admin_get_registration_token() .call(|api| api.admin_get_registration_token().into_future())
.await .await
.context("GET /admin/runners/registration-token")?; .context("GET /admin/runners/registration-token")?;
answer answer

4
swarm-controller/src/forge/site_admin.rs
View file

@ -75,7 +75,7 @@ impl Client {
/// # Errors /// # Errors
/// When the forge refuses the read or the edit. /// When the forge refuses the read or the edit.
pub async fn make_site_admin(&self, name: &str) -> Result<Plan> { pub async fn make_site_admin(&self, name: &str) -> Result<Plan> {
let user = match self.api.user_get(name).await { let user = match self.api.call(|api| api.user_get(name).into_future()).await {
Ok(user) => Some(user), Ok(user) => Some(user),
Err(e) if is_not_found(&e) => None, Err(e) if is_not_found(&e) => None,
Err(e) => return Err(e).with_context(|| format!("read forge user {name}")), Err(e) => return Err(e).with_context(|| format!("read forge user {name}")),
@ -83,7 +83,7 @@ impl Client {
let plan = plan(user.as_ref()); let plan = plan(user.as_ref());
if plan == Plan::Promote { if plan == Plan::Promote {
self.api self.api
.admin_edit_user(name, admin_edit()) .call(|api| api.admin_edit_user(name, admin_edit()).into_future())
.await .await
.with_context(|| format!("make forge user {name} a site admin"))?; .with_context(|| format!("make forge user {name} a site admin"))?;
tracing::info!(%name, "swarm forge: made the user a site admin"); tracing::info!(%name, "swarm forge: made the user a site admin");