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

retry the remaining plain schedules with backon

Browse source
This commit is contained in:
damocles 2026-10-17 15:25:22 +02:00 • committed by mara
commit ed76419be7
7 changed files with 132 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1851,6 +1851,7 @@ name = "hive-forge-notify"
version = "0.1.0"
dependencies = [
"anyhow",
"backon",
"forgejo-api",
"hive-agent-sock",
"hive-sock-client",

6
hive-forge-notify/Cargo.toml
View file

@ -15,6 +15,7 @@ workspace = true
[dependencies]
anyhow.workspace = true
backon.workspace = true
forgejo-api.workspace = true
hive-agent-sock.workspace = true
hive-sock-client.workspace = true
@ -26,3 +27,8 @@ tokio.workspace = true
tracing.workspace = true
tracing-subscriber.workspace = true
url.workspace = true
[dev-dependencies]
# `test-util` for `#[tokio::test(start_paused = true)]`: the token wait is
# thirty seconds a read, ten minutes in all.
tokio = { workspace = true, features = ["test-util"] }

27
hive-forge-notify/src/bin/hive-forge-notify/main.rs
View file

@ -12,9 +12,7 @@ use std::collections::HashMap;
use std::path::Path;
use std::time::Duration;
use hive_forge_notify::notify::{
POLL_INTERVAL_SECS, TOKEN_RETRY_MAX, TOKEN_RETRY_SECS, poll_once, resolve_own_login,
};
use hive_forge_notify::notify::{POLL_INTERVAL_SECS, poll_once, resolve_own_login, wait_for_token};
use hive_forge_notify::{HTTP_TIMEOUT_SECS, TODO_SOCKET_RETRY, agent_socket, init_tracing};
use source::ForgejoSource;
use tracing::{debug, info, warn};
@ -41,22 +39,13 @@ async fn forgejo_loop(state_dir: String, socket: std::path::PathBuf) {
// Retry reading the token to handle races where the agent's fetch of
// it from the swarm secret store lands after this unit starts, or where
// a parent-container chown briefly makes the file unreadable.
let mut token = {
let mut attempts = 0u32;
loop {
if let Some(t) = hive_forge_notify::read_first_token(&token_paths) {
break t;
}
debug!(?token_paths, "forge_notify: no forge token yet");
attempts += 1;
if attempts >= TOKEN_RETRY_MAX {
debug!(
"forge_notify: token not available after {TOKEN_RETRY_MAX} retries — disabled"
);
return;
}
tokio::time::sleep(Duration::from_secs(TOKEN_RETRY_SECS)).await;
}
let Some(mut token) = wait_for_token("forge token", || async {
hive_forge_notify::read_first_token(&token_paths)
})
.await
else {
debug!(?token_paths, "forge_notify: token not available — disabled");
return;
};
let Some(mut source) = ForgejoSource::new(&forge_url, &token) else {

26
hive-forge-notify/src/bin/hive-github-notify/main.rs
View file

@ -24,9 +24,7 @@ mod source;
use std::collections::HashMap;
use std::time::Duration;
use hive_forge_notify::notify::{
POLL_INTERVAL_SECS, TOKEN_RETRY_MAX, TOKEN_RETRY_SECS, poll_once, resolve_own_login,
};
use hive_forge_notify::notify::{POLL_INTERVAL_SECS, poll_once, resolve_own_login, wait_for_token};
use hive_forge_notify::{HTTP_TIMEOUT_SECS, agent_socket, init_tracing};
use source::GithubSource;
use tracing::{debug, info, warn};
@ -50,20 +48,14 @@ async fn main() {
/// common case for an agent that has the unit but no account.
async fn github_loop(state_dir: String, socket: std::path::PathBuf) {
let token_path = format!("{state_dir}/github-token");
let mut attempts = 0u32;
let token = loop {
match tokio::fs::read_to_string(&token_path).await {
Ok(t) if !t.trim().is_empty() => break t.trim().to_owned(),
_ => debug!("forge_notify: no github token at {token_path} yet"),
}
attempts += 1;
if attempts >= TOKEN_RETRY_MAX {
debug!(
"forge_notify: no github token after {TOKEN_RETRY_MAX} retries — github disabled"
);
return;
}
tokio::time::sleep(Duration::from_secs(TOKEN_RETRY_SECS)).await;
let Some(token) = wait_for_token("github token", || async {
let text = tokio::fs::read_to_string(&token_path).await.ok()?;
Some(text.trim().to_owned()).filter(|t| !t.is_empty())
})
.await
else {
debug!("forge_notify: no github token at {token_path} — github disabled");
return;
};
let client = match reqwest::Client::builder()

69
hive-forge-notify/src/notify.rs
View file

@ -27,9 +27,11 @@
use std::collections::{HashMap, HashSet};
use std::fmt::Write as _;
use std::future::Future;
use std::path::Path;
use std::time::Duration;
use backon::{ConstantBuilder, Retryable as _};
use forgejo_api::structs::NotifySubjectType;
use serde::{Deserialize, Deserializer};
use time::OffsetDateTime;
@ -54,10 +56,30 @@ const BODY_TRUNCATE: usize = 500;
/// How long to wait between token-read retries when the token file is
/// missing or unreadable at startup (e.g. hive-priv hasn't provisioned
/// it yet, or a chown race left it temporarily root-owned).
pub const TOKEN_RETRY_SECS: u64 = 30;
/// Give up waiting for the token after this many retries (~10 minutes).
const TOKEN_RETRY_SECS: u64 = 30;
/// Give up waiting for the token after this many reads (~10 minutes).
/// Avoids an infinite wait on agents that genuinely have no forge account.
pub const TOKEN_RETRY_MAX: u32 = 20;
const TOKEN_READS_MAX: usize = 20;
/// Read a token that another process writes out of band, waiting for it to
/// appear: `read` is called up to 20 times, 30 seconds apart (about ten
/// minutes). `None` when it never did, which is the normal answer for an agent
/// with no account for this source. `what` names the token in the log.
pub async fn wait_for_token<F, Fut>(what: &str, read: F) -> Option<String>
where
F: Fn() -> Fut,
Fut: Future<Output = Option<String>>,
{
(|| async { read().await.ok_or(()) })
.retry(
ConstantBuilder::default()
.with_delay(Duration::from_secs(TOKEN_RETRY_SECS))
.with_max_times(TOKEN_READS_MAX - 1),
)
.notify(|(), _| debug!("forge_notify: no {what} yet"))
.await
.ok()
}
/// How close (seconds) the notification's event time must be to a
/// subject's `created_at` for us to call it a genuine creation and emit
/// a `new <kind>` label. Later activity that lands on the state-change
@ -1186,6 +1208,47 @@ mod tests {
assert!(should_deliver(&delivered, "99", "2026-06-22T16:00:00Z"));
}
/// A token that shows up on the `n`th read.
fn token_on_read(n: usize, reads: &std::sync::atomic::AtomicUsize) -> Option<String> {
let read = reads.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1;
(read >= n).then(|| "secret".to_owned())
}
#[tokio::test(start_paused = true)]
async fn a_token_that_is_already_there_costs_no_wait() {
let reads = std::sync::atomic::AtomicUsize::new(0);
let started = tokio::time::Instant::now();
let got = wait_for_token("token", || async { token_on_read(1, &reads) }).await;
assert_eq!(got.as_deref(), Some("secret"));
assert_eq!(started.elapsed(), Duration::ZERO);
}
#[tokio::test(start_paused = true)]
async fn a_token_that_arrives_late_is_waited_for() {
let reads = std::sync::atomic::AtomicUsize::new(0);
let started = tokio::time::Instant::now();
let got = wait_for_token("token", || async { token_on_read(4, &reads) }).await;
assert_eq!(got.as_deref(), Some("secret"));
assert_eq!(started.elapsed(), Duration::from_secs(3 * TOKEN_RETRY_SECS));
}
#[tokio::test(start_paused = true)]
async fn a_token_that_never_comes_is_given_up_on_after_the_bound() {
let reads = std::sync::atomic::AtomicUsize::new(0);
let started = tokio::time::Instant::now();
let got = wait_for_token("token", || async { token_on_read(usize::MAX, &reads) }).await;
assert_eq!(got, None);
assert_eq!(
reads.load(std::sync::atomic::Ordering::SeqCst),
TOKEN_READS_MAX
);
assert_eq!(
started.elapsed(),
Duration::from_secs((TOKEN_READS_MAX as u64 - 1) * TOKEN_RETRY_SECS),
"no wait after the last read"
);
}
#[test]
fn escape_md_headings_escapes_top_level_atx() {
// Argus reviews start with `## argus review`, which would

52
swarm-controller/src/agent_permissions.rs
View file

@ -10,10 +10,12 @@
use std::collections::BTreeSet;
use std::sync::Arc;
use std::time::Duration;
use axum::Json;
use axum::extract::{Path, State};
use axum::http::StatusCode;
use backon::{ConstantBuilder, Retryable as _};
use futures_util::StreamExt as _;
use hive_sh4re::permissions::Permission;
use problem_details::ProblemDetails;
@ -335,32 +337,44 @@ pub(crate) async fn restore(client: &async_nats::Client, agent: &str, replaced:
/// the queue is not serving it, rather than that nothing is set.
pub(crate) fn spawn(client: async_nats::Client, wanted: Arc<WantedWriter>, hives: Vec<String>) {
tokio::spawn(async move {
const RETRY: std::time::Duration = std::time::Duration::from_secs(5);
let store = loop {
if swarm_queue_client::ensure_connected(&client).is_ok() {
match agent_permissions::open_or_create(&client).await {
Ok(store) => break store,
Err(e) => tracing::warn!(
error = swarm_queue_client::chain(&e),
"creating the agent-permissions bucket failed; retrying"
),
}
const RETRY: Duration = Duration::from_secs(5);
// The first two retry until they work: the queue is the only thing either waits on.
let every_five_seconds = || {
ConstantBuilder::default()
.with_delay(RETRY)
.without_max_times()
};
let Ok(store) = (|| async {
swarm_queue_client::ensure_connected(&client)?;
agent_permissions::open_or_create(&client).await
})
.retry(every_five_seconds())
.notify(|e: &swarm_queue_client::Error, _| {
// Not connected is the queue being late, and it says so itself.
if !matches!(e, swarm_queue_client::Error::NotConnected(_)) {
tracing::warn!(
error = swarm_queue_client::chain(e),
"creating the agent-permissions bucket failed; retrying"
);
}
tokio::time::sleep(RETRY).await;
})
.await
else {
return;
};
tokio::spawn({
let (store, wanted, hives) = (store.clone(), Arc::clone(&wanted), hives.clone());
async move {
loop {
match seed_ruth_grant(&store, &wanted, &hives).await {
Ok(()) => break,
Err(e) => tracing::warn!(
// Retries without end, so it answers only once it has worked.
let _ = (|| seed_ruth_grant(&store, &wanted, &hives))
.retry(every_five_seconds())
.notify(|e: &anyhow::Error, _| {
tracing::warn!(
error = %format!("{e:#}"),
"agent permissions: seeding ruth's grant failed; retrying"
),
}
tokio::time::sleep(RETRY).await;
}
);
})
.await;
}
});
let subject = agent_permissions::import_subject_wildcard();

15
swarm-controller/src/read_policy.rs
View file

@ -29,6 +29,7 @@
use std::{future::Future, time::Duration};
use anyhow::{Context, Result};
use backon::{BackoffBuilder as _, ConstantBuilder};
use swarm_secret_client::{Error, SecretStore, client::DEFAULT_CERT_MOUNT, policy};
/// File holding the authority hives are issued from, as
@ -111,12 +112,14 @@ where
F: FnMut() -> Fut,
Fut: Future<Output = Result<(), Unreachable>>,
{
let mut ticker = tokio::time::interval(interval);
// `interval`'s first tick is immediate, and that pass is the one the caller
// already ran before deciding to retry at all.
ticker.tick().await;
for n in 1..=attempts {
ticker.tick().await;
// One delay per attempt, waited before it: the pass the caller already ran
// before deciding to retry at all is the attempt that needs no wait.
let delays = ConstantBuilder::default()
.with_delay(interval)
.with_max_times(attempts as usize)
.build();
for (n, delay) in (1..).zip(delays) {
tokio::time::sleep(delay).await;
if attempt().await.is_ok() {
return Some(n);
}