diff --git a/Cargo.lock b/Cargo.lock index 3eb1fa1f..f68047db 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1851,6 +1851,7 @@ name = "hive-forge-notify" version = "0.1.0" dependencies = [ "anyhow", + "backon", "forgejo-api", "hive-agent-sock", "hive-sock-client", diff --git a/hive-forge-notify/Cargo.toml b/hive-forge-notify/Cargo.toml index cb2bb303..4f967564 100644 --- a/hive-forge-notify/Cargo.toml +++ b/hive-forge-notify/Cargo.toml @@ -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"] } diff --git a/hive-forge-notify/src/bin/hive-forge-notify/main.rs b/hive-forge-notify/src/bin/hive-forge-notify/main.rs index 7c2003fc..9e2c2ac5 100644 --- a/hive-forge-notify/src/bin/hive-forge-notify/main.rs +++ b/hive-forge-notify/src/bin/hive-forge-notify/main.rs @@ -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 { diff --git a/hive-forge-notify/src/bin/hive-github-notify/main.rs b/hive-forge-notify/src/bin/hive-github-notify/main.rs index c479b8d0..11fed9fd 100644 --- a/hive-forge-notify/src/bin/hive-github-notify/main.rs +++ b/hive-forge-notify/src/bin/hive-github-notify/main.rs @@ -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() diff --git a/hive-forge-notify/src/notify.rs b/hive-forge-notify/src/notify.rs index 2e9bbee0..35e62857 100644 --- a/hive-forge-notify/src/notify.rs +++ b/hive-forge-notify/src/notify.rs @@ -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(what: &str, read: F) -> Option +where + F: Fn() -> Fut, + Fut: Future>, +{ + (|| 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 ` 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 { + 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 diff --git a/swarm-controller/src/agent_permissions.rs b/swarm-controller/src/agent_permissions.rs index 919f726a..b459d860 100644 --- a/swarm-controller/src/agent_permissions.rs +++ b/swarm-controller/src/agent_permissions.rs @@ -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, hives: Vec) { 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(); diff --git a/swarm-controller/src/read_policy.rs b/swarm-controller/src/read_policy.rs index 533c49fb..61777762 100644 --- a/swarm-controller/src/read_policy.rs +++ b/swarm-controller/src/read_policy.rs @@ -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>, { - 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); }