From a1d1748c55aac2a8f29deb7f95afeacd1d378807 Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 21 Jul 2026 19:55:38 +0200 Subject: [PATCH 1/3] Rotate GitHub tokens round-robin across posts The daemon accepts github_token_files, a list of files each holding one personal access token. Every history record now carries the SHA-256 hash of the token that made the attempt; each post uses the successor of the token whose hash the latest record carries, so rotation survives restarts and token file renames without logging any secret. A single inline token still works as a pool of one, and hist shows the first eight hash characters per record. Co-Authored-By: Claude Fable 5 --- Cargo.lock | 66 +++++++++++++ crates/comenq/src/client.rs | 1 + crates/comenq/src/output.rs | 12 ++- crates/comenqd/Cargo.toml | 1 + crates/comenqd/src/config.rs | 41 +++++++- crates/comenqd/src/config/tests.rs | 88 +++++++++++++++++ crates/comenqd/src/config/tokens.rs | 54 ++++++++++ crates/comenqd/src/daemon.rs | 3 + crates/comenqd/src/queue.rs | 29 ++++-- crates/comenqd/src/store/history.rs | 23 ++++- crates/comenqd/src/store/tests.rs | 8 +- crates/comenqd/src/supervisor.rs | 21 ++-- crates/comenqd/src/supervisor/tests.rs | 7 +- crates/comenqd/src/worker.rs | 87 +++++++++++++++-- crates/comenqd/src/worker/tests.rs | 60 ++++++++++++ crates/comenqd/tests/daemon.rs | 23 ++++- src/protocol.rs | 3 + test-support/src/daemon.rs | 18 +++- test-support/src/lib.rs | 2 +- tests/features/worker.feature | 16 +++ tests/steps/history_steps.rs | 11 ++- tests/steps/mod.rs | 1 + tests/steps/worker_steps.rs | 130 +++++++++++++++++-------- tests/steps/worker_token_steps.rs | 114 ++++++++++++++++++++++ 24 files changed, 735 insertions(+), 84 deletions(-) create mode 100644 crates/comenqd/src/config/tokens.rs create mode 100644 tests/steps/worker_token_steps.rs diff --git a/Cargo.lock b/Cargo.lock index 5bf9a92..2799fee 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -189,6 +189,15 @@ version = "2.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1b8e56985ec62d17e9c1001dc89c88ecd7dc08e47eba5ec7c29c7b5eeecde967" +[[package]] +name = "block-buffer" +version = "0.10.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" +dependencies = [ + "generic-array", +] + [[package]] name = "bstr" version = "1.12.0" @@ -393,6 +402,7 @@ dependencies = [ "serde", "serde_json", "serial_test", + "sha2", "tempfile", "test-support", "thiserror 1.0.69", @@ -432,6 +442,15 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -457,6 +476,16 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" +[[package]] +name = "crypto-common" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a" +dependencies = [ + "generic-array", + "typenum", +] + [[package]] name = "cucumber" version = "0.20.2" @@ -568,6 +597,16 @@ dependencies = [ "syn", ] +[[package]] +name = "digest" +version = "0.10.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" +dependencies = [ + "block-buffer", + "crypto-common", +] + [[package]] name = "directories" version = "6.0.0" @@ -776,6 +815,16 @@ dependencies = [ "slab", ] +[[package]] +name = "generic-array" +version = "0.14.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" +dependencies = [ + "typenum", + "version_check", +] + [[package]] name = "getrandom" version = "0.2.16" @@ -2199,6 +2248,17 @@ dependencies = [ "syn", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -2780,6 +2840,12 @@ dependencies = [ "syn", ] +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + [[package]] name = "uncased" version = "0.9.10" diff --git a/crates/comenq/src/client.rs b/crates/comenq/src/client.rs index d16a2d6..f6aaa77 100644 --- a/crates/comenq/src/client.rs +++ b/crates/comenq/src/client.rs @@ -217,6 +217,7 @@ mod tests { posted_at: 1_000, success: true, error: None, + token_hash: "ab".repeat(32), owner: "octocat".into(), repo: "hello-world".into(), pr_number: 1, diff --git a/crates/comenq/src/output.rs b/crates/comenq/src/output.rs index 4b7f24c..75782df 100644 --- a/crates/comenq/src/output.rs +++ b/crates/comenq/src/output.rs @@ -86,16 +86,19 @@ pub fn format_age(seconds: u64) -> String { /// Render one `hist` line for a past posting attempt. /// -/// `now` is the current Unix time, used to show the attempt's age. Failed +/// `now` is the current Unix time, used to show the attempt's age. The +/// posting token appears as the first eight characters of its hash. Failed /// attempts carry the failure description, collapsed to one line. pub(crate) fn render_history(entry: &HistoryEntry, now: u64) -> String { let age = format_age(now.saturating_sub(entry.posted_at)); let status = if entry.success { "ok" } else { "FAIL" }; + let token: String = entry.token_hash.chars().take(8).collect(); let mut line = format!( - "{} {:>11} {:<4} {}/{}#{} {}", + "{} {:>11} {:<4} {:<8} {}/{}#{} {}", entry.id, age, status, + token, entry.owner, entry.repo, entry.pr_number, @@ -203,6 +206,7 @@ mod tests { posted_at: 1_000, success, error: error.map(str::to_owned), + token_hash: "cafe0123".repeat(8), owner: "octocat".into(), repo: "hello-world".into(), pr_number: 7, @@ -224,7 +228,7 @@ mod tests { let line = render_history(&history(true, None), 1_150); assert_eq!( line, - "1a2b3c4d 2m 30s ago ok octocat/hello-world#7 Hi there" + "1a2b3c4d 2m 30s ago ok cafe0123 octocat/hello-world#7 Hi there" ); } @@ -233,7 +237,7 @@ mod tests { let line = render_history(&history(false, Some("timeout\nafter 10s")), 1_045); assert_eq!( line, - "1a2b3c4d 45s ago FAIL octocat/hello-world#7 Hi there — timeout after 10s" + "1a2b3c4d 45s ago FAIL cafe0123 octocat/hello-world#7 Hi there — timeout after 10s" ); } diff --git a/crates/comenqd/Cargo.toml b/crates/comenqd/Cargo.toml index 21ec12e..32dd790 100644 --- a/crates/comenqd/Cargo.toml +++ b/crates/comenqd/Cargo.toml @@ -27,6 +27,7 @@ figment = { version = "0.10", default-features = false, features = ["env", "toml test-support = { workspace = true, optional = true } backon = { workspace = true } rand = { workspace = true } +sha2 = "0.10" uuid = { workspace = true, features = ["v4"] } [dev-dependencies] diff --git a/crates/comenqd/src/config.rs b/crates/comenqd/src/config.rs index 11d05bd..d3b7bd0 100644 --- a/crates/comenqd/src/config.rs +++ b/crates/comenqd/src/config.rs @@ -9,6 +9,9 @@ use serde::{Deserialize, Serialize}; use std::io; use std::path::{Path, PathBuf}; +mod tokens; +pub use tokens::NamedToken; + /// Default queue directory when none is provided. const DEFAULT_QUEUE_PATH: &str = "/var/lib/comenq/queue"; /// Default cooldown in seconds between comment posts. @@ -41,6 +44,14 @@ pub struct Config { /// `github_token_file = "${CREDENTIALS_DIRECTORY}/token"`. #[serde(default)] pub github_token_file: Option, + /// Paths to files each containing one GitHub Personal Access Token. + /// + /// When non-empty, the daemon posts comments with these tokens in + /// round-robin rotation: each post uses the successor of the token whose + /// hash was recorded against the most recent history record. The files + /// are read at startup; each must hold a non-empty token. + #[serde(default)] + pub github_token_files: Vec, /// Path to the Unix Domain Socket. /// /// Defaults to `$XDG_RUNTIME_DIR/comenq/comenq.sock` when a user runtime @@ -98,6 +109,7 @@ impl From for Config { Self { github_token, github_token_file: None, + github_token_files: Vec::new(), socket_path, queue_path, cooldown_period_seconds, @@ -137,6 +149,7 @@ impl From<&test_support::daemon::TestConfig> for Config { Self { github_token: value.github_token.clone(), github_token_file: None, + github_token_files: Vec::new(), socket_path: value.socket_path.clone(), queue_path: value.queue_path.clone(), cooldown_period_seconds: value.cooldown_period_seconds, @@ -298,13 +311,31 @@ impl Config { } self.github_token = token.to_owned(); } - if self.github_token.is_empty() { - return Err(ortho_config::OrthoError::Validation { + if !self.github_token_files.is_empty() { + // Fail fast: every rotation token must be readable at startup. + self.github_tokens() + .map(|_| ()) + .map_err(|e| ortho_config::OrthoError::Validation { + key: "github_token_files".into(), + message: e.to_string(), + }) + } else if self.github_token.is_empty() { + Err(ortho_config::OrthoError::Validation { key: "github_token".into(), - message: "provide github_token or github_token_file".into(), - }); + message: "provide github_token, github_token_file, or github_token_files".into(), + }) + } else { + Ok(()) } - Ok(()) + } + + /// The rotation pool of GitHub tokens, in configured order. + /// + /// Reads each path in [`Config::github_token_files`], naming the token + /// after its file. When no rotation files are configured, the single + /// [`Config::github_token`] forms a pool of one named `default`. + pub fn github_tokens(&self) -> io::Result> { + tokens::resolve_pool(&self.github_token_files, &self.github_token) } } diff --git a/crates/comenqd/src/config/tests.rs b/crates/comenqd/src/config/tests.rs index 8322b58..ded3a7a 100644 --- a/crates/comenqd/src/config/tests.rs +++ b/crates/comenqd/src/config/tests.rs @@ -261,3 +261,91 @@ fn converts_from_test_config(#[case] conv: fn(&test_support::daemon::TestConfig) test_cfg.github_api_timeout_secs ); } + +#[rstest] +#[serial_test::serial] +fn token_files_form_the_rotation_pool() { + let dir = tempdir().expect("create tempdir"); + let first = dir.path().join("pandalump-token"); + let second = dir.path().join("buzzybee-token"); + fs::write(&first, "alpha-token\n").expect("write first token"); + fs::write(&second, "bravo-token\n").expect("write second token"); + let path = dir.path().join("config.toml"); + fs::write( + &path, + format!( + "github_token_files=['{}','{}']", + first.display(), + second.display() + ), + ) + .expect("write config fixture"); + let cfg = Config::from_file(&path).expect("load config"); + let tokens = cfg.github_tokens().expect("resolve tokens"); + let summary: Vec<(&str, &str)> = tokens + .iter() + .map(|named| (named.name.as_str(), named.token.as_str())) + .collect(); + assert_eq!( + summary, + vec![ + ("pandalump-token", "alpha-token"), + ("buzzybee-token", "bravo-token"), + ] + ); +} + +#[rstest] +#[serial_test::serial] +fn token_files_suffice_without_an_inline_token() { + let dir = tempdir().expect("create tempdir"); + let file = dir.path().join("only-token"); + fs::write(&file, "solo\n").expect("write token"); + let path = dir.path().join("config.toml"); + fs::write(&path, format!("github_token_files=['{}']", file.display())) + .expect("write config fixture"); + let cfg = Config::from_file(&path).expect("token files alone should satisfy validation"); + assert!(cfg.github_token.is_empty()); +} + +#[rstest] +#[serial_test::serial] +fn empty_rotation_token_file_errors() { + let dir = tempdir().expect("create tempdir"); + let file = dir.path().join("empty-token"); + fs::write(&file, " \n").expect("write blank token"); + let path = dir.path().join("config.toml"); + fs::write(&path, format!("github_token_files=['{}']", file.display())) + .expect("write config fixture"); + assert!(Config::from_file(&path).is_err()); +} + +#[rstest] +#[serial_test::serial] +fn missing_rotation_token_file_errors() { + let dir = tempdir().expect("create tempdir"); + let path = dir.path().join("config.toml"); + fs::write( + &path, + format!( + "github_token_files=['{}']", + dir.path().join("nope-token").display() + ), + ) + .expect("write config fixture"); + assert!(Config::from_file(&path).is_err()); +} + +#[rstest] +#[serial_test::serial] +fn single_inline_token_forms_a_default_pool() { + let dir = tempdir().expect("create tempdir"); + let path = dir.path().join("config.toml"); + fs::write(&path, "github_token='abc'").expect("write config fixture"); + let cfg = Config::from_file(&path).expect("load config"); + let tokens = cfg.github_tokens().expect("resolve tokens"); + assert_eq!(tokens.len(), 1); + let named = tokens.first().expect("single token"); + assert_eq!(named.name, "default"); + assert_eq!(named.token, "abc"); +} diff --git a/crates/comenqd/src/config/tokens.rs b/crates/comenqd/src/config/tokens.rs new file mode 100644 index 0000000..446c6af --- /dev/null +++ b/crates/comenqd/src/config/tokens.rs @@ -0,0 +1,54 @@ +//! Resolution of the GitHub token rotation pool. +//! +//! Reads the configured token files into named tokens; the worker rotates +//! through the pool round-robin when posting comments. + +use std::io; +use std::path::PathBuf; + +/// A GitHub token paired with the name of the file it was read from. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct NamedToken { + /// The token file's name, or `default` for a single inline token. + pub name: String, + /// The token value. + pub token: String, +} + +/// Resolve the rotation pool from `files`, falling back to `inline`. +/// +/// Reads each path in `files`, naming the token after its file; each file +/// must hold a non-empty token. When `files` is empty, `inline` forms a +/// pool of one named `default`. +pub(super) fn resolve_pool(files: &[PathBuf], inline: &str) -> io::Result> { + if files.is_empty() { + return Ok(vec![NamedToken { + name: "default".to_owned(), + token: inline.to_owned(), + }]); + } + files + .iter() + .map(|file| { + let token = std::fs::read_to_string(file).map_err(|e| { + io::Error::new(e.kind(), format!("token file {}: {e}", file.display())) + })?; + let token = token.trim(); + if token.is_empty() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("token file {} is empty", file.display()), + )); + } + let name = file + .file_name() + .and_then(|name| name.to_str()) + .unwrap_or("token") + .to_owned(); + Ok(NamedToken { + name, + token: token.to_owned(), + }) + }) + .collect() +} diff --git a/crates/comenqd/src/daemon.rs b/crates/comenqd/src/daemon.rs index 9fec288..701bdd3 100644 --- a/crates/comenqd/src/daemon.rs +++ b/crates/comenqd/src/daemon.rs @@ -16,6 +16,9 @@ pub use crate::queue::SharedQueue; /// Run the worker that drains the queue and talks to the GitHub API. pub use crate::worker::run_worker; +/// A GitHub client paired with its token's name and hash, and the hash +/// helper used to identify tokens in the posting history. +pub use crate::worker::{TokenClient, token_hash}; /// Control handle and lifecycle hooks for the worker task. pub use crate::worker::{WorkerControl, WorkerHooks}; diff --git a/crates/comenqd/src/queue.rs b/crates/comenqd/src/queue.rs index 434d97f..f8fcc75 100644 --- a/crates/comenqd/src/queue.rs +++ b/crates/comenqd/src/queue.rs @@ -63,23 +63,40 @@ impl SharedQueue { } /// Remove the posted entry, recording the posting time and a history - /// record of the success. - pub async fn complete(&self, entry: &StoredEntry) -> StoreResult<()> { + /// record of the success, tagged with the posting token's hash. + pub async fn complete(&self, entry: &StoredEntry, token_hash: &str) -> StoreResult<()> { let store = self.store.lock().await; let now = unix_now(); store.complete(&entry.id, now)?; - store.append_history(&HistoryRecord::success(entry, now)) + store.append_history(&HistoryRecord::success(entry, token_hash, now)) } - /// Record a failed posting attempt in the history log. + /// Record a failed posting attempt in the history log, tagged with the + /// hash of the token that attempted it. /// /// The entry itself stays queued; the worker retries it after a /// cooldown. - pub async fn record_failure(&self, entry: &StoredEntry, error: &str) -> StoreResult<()> { + pub async fn record_failure( + &self, + entry: &StoredEntry, + token_hash: &str, + error: &str, + ) -> StoreResult<()> { self.store .lock() .await - .append_history(&HistoryRecord::failure(entry, unix_now(), error)) + .append_history(&HistoryRecord::failure( + entry, + token_hash, + unix_now(), + error, + )) + } + + /// Hash of the token used by the most recent posting attempt, when any. + pub async fn last_token_hash(&self) -> StoreResult> { + let history = self.store.lock().await.history()?; + Ok(history.last().map(|record| record.token_hash.clone())) } /// Execute a protocol request and produce the reply. diff --git a/crates/comenqd/src/store/history.rs b/crates/comenqd/src/store/history.rs index f28c4de..d6b2c33 100644 --- a/crates/comenqd/src/store/history.rs +++ b/crates/comenqd/src/store/history.rs @@ -24,31 +24,45 @@ pub struct HistoryRecord { /// Failure description when `success` is false. #[serde(default, skip_serializing_if = "Option::is_none")] pub error: Option, + /// Hex SHA-256 of the token that made the attempt. + /// + /// The worker reads the latest record's hash to pick the next token in + /// the round-robin rotation. + pub token_hash: String, /// The comment that was posted (or attempted). pub request: CommentRequest, } impl HistoryRecord { - /// Record a successful post of `entry` at `posted_at`. + /// Record a successful post of `entry` by the token hashing to + /// `token_hash` at `posted_at`. #[must_use] - pub fn success(entry: &StoredEntry, posted_at: u64) -> Self { + pub fn success(entry: &StoredEntry, token_hash: &str, posted_at: u64) -> Self { Self { id: entry.id.clone(), posted_at, success: true, error: None, + token_hash: token_hash.to_owned(), request: entry.request.clone(), } } - /// Record a failed posting attempt of `entry` at `posted_at`. + /// Record a failed posting attempt of `entry` by the token hashing to + /// `token_hash` at `posted_at`. #[must_use] - pub fn failure(entry: &StoredEntry, posted_at: u64, error: impl Into) -> Self { + pub fn failure( + entry: &StoredEntry, + token_hash: &str, + posted_at: u64, + error: impl Into, + ) -> Self { Self { id: entry.id.clone(), posted_at, success: false, error: Some(error.into()), + token_hash: token_hash.to_owned(), request: entry.request.clone(), } } @@ -61,6 +75,7 @@ impl HistoryRecord { posted_at: self.posted_at, success: self.success, error: self.error.clone(), + token_hash: self.token_hash.clone(), owner: self.request.owner.clone(), repo: self.request.repo.clone(), pr_number: self.request.pr_number, diff --git a/crates/comenqd/src/store/tests.rs b/crates/comenqd/src/store/tests.rs index 40d8e93..9e486e6 100644 --- a/crates/comenqd/src/store/tests.rs +++ b/crates/comenqd/src/store/tests.rs @@ -293,8 +293,8 @@ fn history_records_round_trip_in_order() { let dir = TempDir::new().expect("tempdir"); let store = open_store(&dir); let entry = store.put(request("a"), &immediate(0), 1000).expect("put"); - let success = HistoryRecord::success(&entry, 1600); - let failure = HistoryRecord::failure(&entry, 2200, "timeout"); + let success = HistoryRecord::success(&entry, "hash-a", 1600); + let failure = HistoryRecord::failure(&entry, "hash-b", 2200, "timeout"); store.append_history(&success).expect("append success"); store.append_history(&failure).expect("append failure"); let records = store.history().expect("read history"); @@ -316,14 +316,14 @@ fn history_skips_malformed_lines() { let store = open_store(&dir); let entry = store.put(request("a"), &immediate(0), 1000).expect("put"); store - .append_history(&HistoryRecord::success(&entry, 1600)) + .append_history(&HistoryRecord::success(&entry, "hash-a", 1600)) .expect("append success"); let log = dir.path().join("history.jsonl"); let mut text = std::fs::read_to_string(&log).expect("read log"); text.push_str("not json\n"); std::fs::write(&log, text).expect("corrupt log"); store - .append_history(&HistoryRecord::failure(&entry, 2200, "timeout")) + .append_history(&HistoryRecord::failure(&entry, "hash-b", 2200, "timeout")) .expect("append failure"); let records = store.history().expect("read history"); assert_eq!(records.len(), 2, "the corrupt line must not hide records"); diff --git a/crates/comenqd/src/supervisor.rs b/crates/comenqd/src/supervisor.rs index 65abf3b..49c4a59 100644 --- a/crates/comenqd/src/supervisor.rs +++ b/crates/comenqd/src/supervisor.rs @@ -5,7 +5,6 @@ use crate::config::Config; use backon::{ExponentialBackoff, ExponentialBuilder}; -use octocrab::Octocrab; use std::path::Path; use std::sync::Arc; use std::time::Duration; @@ -18,7 +17,7 @@ use tokio::sync::watch; use crate::listener::run_listener; use crate::queue::SharedQueue; use crate::store::StoreError; -use crate::worker::{WorkerControl, WorkerHooks, build_octocrab, run_worker}; +use crate::worker::{TokenClient, WorkerControl, WorkerHooks, build_octocrab, run_worker}; #[derive(Debug, Error)] pub enum SupervisorError { @@ -110,14 +109,22 @@ async fn supervise_task( pub async fn run(config: Config) -> Result<()> { ensure_queue_dir(&config.queue_path).await?; tracing::info!(queue = %config.queue_path.display(), "Queue directory prepared"); - let octocrab = Arc::new(build_octocrab(&config.github_token)?); + let mut clients = Vec::new(); + for named in config.github_tokens()? { + clients.push(TokenClient::new( + named.name, + &named.token, + Arc::new(build_octocrab(&named.token)?), + )); + } + let clients = Arc::new(clients); let cfg = Arc::new(config); let queue = SharedQueue::open(cfg.clone())?; let (shutdown_tx, shutdown_rx) = watch::channel(()); // Initial task spawns and backoff builders. let listener = spawn_listener(queue.clone(), shutdown_rx.clone()); - let worker = spawn_worker(queue.clone(), octocrab.clone(), shutdown_rx.clone()); + let worker = spawn_worker(queue.clone(), clients.clone(), shutdown_rx.clone()); let min_delay = Duration::from_millis(cfg.restart_min_delay_ms); let listener_backoff = backoff(min_delay); let worker_backoff = backoff(min_delay); @@ -168,7 +175,7 @@ pub async fn run(config: Config) -> Result<()> { "worker", worker, worker_backoff, - || spawn_worker(queue.clone(), octocrab.clone(), shutdown_worker.clone()), + || spawn_worker(queue.clone(), clients.clone(), shutdown_worker.clone()), shutdown_worker.clone(), || backoff(min_delay), ), @@ -186,11 +193,11 @@ fn spawn_listener( fn spawn_worker( queue: Arc, - octocrab: Arc, + clients: Arc>, shutdown: watch::Receiver<()>, ) -> tokio::task::JoinHandle> { let control = WorkerControl::new(shutdown, WorkerHooks::default()); - tokio::spawn(run_worker(queue, octocrab, control)) + tokio::spawn(run_worker(queue, clients, control)) } /// Log any failure from a supervised task. diff --git a/crates/comenqd/src/supervisor/tests.rs b/crates/comenqd/src/supervisor/tests.rs index 6cd261b..0ac2802 100644 --- a/crates/comenqd/src/supervisor/tests.rs +++ b/crates/comenqd/src/supervisor/tests.rs @@ -92,8 +92,13 @@ async fn worker_starts_and_stops_cleanly() { let queue = crate::queue::SharedQueue::open(cfg).expect("open shared queue"); let octocrab = std::sync::Arc::new(crate::worker::build_octocrab("token").expect("build octocrab")); + let clients = std::sync::Arc::new(vec![crate::worker::TokenClient::new( + "test-token", + "token", + octocrab, + )]); let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(()); - let handle = super::spawn_worker(queue, octocrab, shutdown_rx); + let handle = super::spawn_worker(queue, clients, shutdown_rx); shutdown_tx.send(()).expect("signal shutdown"); let res = tokio::time::timeout(std::time::Duration::from_secs(5), handle) diff --git a/crates/comenqd/src/worker.rs b/crates/comenqd/src/worker.rs index 9d464d4..e6d6279 100644 --- a/crates/comenqd/src/worker.rs +++ b/crates/comenqd/src/worker.rs @@ -4,12 +4,18 @@ //! posting time arrives. Cooldowns always run in full; each entry's random //! flutter was fixed when it was enqueued, so the projected schedule reported //! to clients matches what the worker executes. +//! +//! Posts rotate round-robin through the configured token pool: each post +//! uses the successor of the token whose hash the most recent history +//! record carries, so rotation survives restarts and redeployments. use crate::config::Config; use crate::queue::SharedQueue; -use anyhow::Result; +use anyhow::{Context as _, Result}; use comenq_lib::CommentRequest; use octocrab::Octocrab; +use sha2::{Digest as _, Sha256}; +use std::fmt::Write as _; use std::sync::Arc; use std::time::Duration; use thiserror::Error; @@ -34,6 +40,63 @@ pub(crate) fn build_octocrab(token: &str) -> octocrab::Result { .build() } +/// Hex-encoded SHA-256 hash of a token value. +/// +/// History records identify tokens by this hash rather than by name or +/// value, so rotation state survives token files being renamed and the log +/// never holds a secret. +/// +/// # Examples +/// +/// ```rust +/// let hash = comenqd::daemon::token_hash("s3cret"); +/// assert_eq!(hash.len(), 64); +/// assert_eq!(hash, comenqd::daemon::token_hash("s3cret")); +/// ``` +#[must_use] +pub fn token_hash(token: &str) -> String { + let digest = Sha256::digest(token.as_bytes()); + let mut hex = String::with_capacity(64); + for byte in digest { + let _ = write!(hex, "{byte:02x}"); + } + hex +} + +/// A GitHub client paired with its token's name and hash. +#[derive(Clone)] +pub struct TokenClient { + /// Token file name, used only for logging. + pub name: String, + /// Hex SHA-256 of the token value, recorded in the history. + pub hash: String, + /// Client authenticated with the token. + pub octocrab: Arc, +} + +impl TokenClient { + /// Pair `octocrab` (authenticated with `token`) with the token's + /// identifying name and hash. + #[must_use] + pub fn new(name: impl Into, token: &str, octocrab: Arc) -> Self { + Self { + name: name.into(), + hash: token_hash(token), + octocrab, + } + } +} + +/// Index of the token to use for the next post. +/// +/// The successor of the token whose hash matches `last_hash`; the first +/// token when the history is empty or names a token no longer configured. +fn next_token_index(clients: &[TokenClient], last_hash: Option<&str>) -> usize { + last_hash + .and_then(|hash| clients.iter().position(|client| client.hash == hash)) + .map_or(0, |index| index.saturating_add(1) % clients.len().max(1)) +} + async fn post_comment( octocrab: &Octocrab, request: &CommentRequest, @@ -159,12 +222,17 @@ impl WorkerControl { /// /// The worker recomputes the head entry's due time on every iteration, so /// queue mutations (put, bump, bust, del) take effect immediately: the shared -/// queue's change signal interrupts any wait. +/// queue's change signal interrupts any wait. Each post rotates to the next +/// token in `clients` after the one the latest history record names. pub async fn run_worker( queue: Arc, - octocrab: Arc, + clients: Arc>, mut control: WorkerControl, ) -> Result<()> { + anyhow::ensure!( + !clients.is_empty(), + "the worker needs at least one GitHub token" + ); let hooks = &control.hooks; let shutdown = &mut control.shutdown; let config = queue.config().clone(); @@ -188,21 +256,28 @@ pub async fn run_worker( continue; } hooks.notify_enqueued(); - match post_comment(&octocrab, &entry.request, &config).await { + let last_hash = queue.last_token_hash().await?; + let index = next_token_index(&clients, last_hash.as_deref()); + let client = clients.get(index).context("token index in range")?; + match post_comment(&client.octocrab, &entry.request, &config).await { Ok(()) => { - queue.complete(&entry).await?; + queue.complete(&entry, &client.hash).await?; } Err(e) => { tracing::error!( error = %e, id = %entry.id, + token = %client.name, owner = %entry.request.owner, repo = %entry.request.repo, pr = entry.request.pr_number, "GitHub API call failed; will retry after cooldown", ); // A history write failure must not stop the retry loop. - if let Err(log_error) = queue.record_failure(&entry, &e.to_string()).await { + if let Err(log_error) = queue + .record_failure(&entry, &client.hash, &e.to_string()) + .await + { tracing::error!( error = %log_error, id = %entry.id, diff --git a/crates/comenqd/src/worker/tests.rs b/crates/comenqd/src/worker/tests.rs index 750e299..fd7fea9 100644 --- a/crates/comenqd/src/worker/tests.rs +++ b/crates/comenqd/src/worker/tests.rs @@ -105,3 +105,63 @@ async fn notify_one_buffers_permit_when_no_waiters() { "second waiter should timeout with no remaining permit" ); } + +mod token_rotation { + //! Tests for token hashing and round-robin selection. + + use super::super::{TokenClient, build_octocrab, next_token_index, token_hash}; + use rstest::rstest; + use std::sync::Arc; + + fn clients() -> Vec { + ["alpha-token", "bravo-token"] + .iter() + .map(|token| { + let octocrab = Arc::new(build_octocrab(token).expect("build octocrab")); + TokenClient::new(format!("{token}-file"), token, octocrab) + }) + .collect() + } + + #[rstest] + fn token_hash_is_deterministic_hex_sha256() { + let hash = token_hash("s3cret"); + assert_eq!(hash.len(), 64); + assert!(hash.chars().all(|c| c.is_ascii_hexdigit())); + assert_eq!(hash, token_hash("s3cret")); + assert_ne!(hash, token_hash("other")); + // Known SHA-256 of "s3cret" pins the algorithm. + assert_eq!( + hash, + "1ec1c26b50d5d3c58d9583181af8076655fe00756bf7285940ba3670f99fcba0" + ); + } + + #[rstest] + #[tokio::test] + async fn empty_history_selects_the_first_token() { + assert_eq!(next_token_index(&clients(), None), 0); + } + + #[rstest] + #[tokio::test] + async fn rotation_advances_past_the_last_used_token() { + let clients = clients(); + let first_hash = token_hash("alpha-token"); + assert_eq!(next_token_index(&clients, Some(&first_hash)), 1); + } + + #[rstest] + #[tokio::test] + async fn rotation_wraps_from_the_last_token_to_the_first() { + let clients = clients(); + let second_hash = token_hash("bravo-token"); + assert_eq!(next_token_index(&clients, Some(&second_hash)), 0); + } + + #[rstest] + #[tokio::test] + async fn unknown_hashes_fall_back_to_the_first_token() { + assert_eq!(next_token_index(&clients(), Some("not-a-hash")), 0); + } +} diff --git a/crates/comenqd/tests/daemon.rs b/crates/comenqd/tests/daemon.rs index 436bffa..f3b04b8 100644 --- a/crates/comenqd/tests/daemon.rs +++ b/crates/comenqd/tests/daemon.rs @@ -5,7 +5,7 @@ mod util; use comenq_lib::protocol::{PendingEntry, Request, Response}; use comenqd::config::Config; use comenqd::daemon::{ - SharedQueue, WorkerControl, WorkerHooks, + SharedQueue, TokenClient, WorkerControl, WorkerHooks, listener::{handle_client, prepare_listener, run_listener}, run, run_worker, }; @@ -28,6 +28,11 @@ use util::{TestComplexity, TimeoutConfig, join_err, timeout_with_retries}; const TEST_COOLDOWN_SECONDS: u64 = 1; +/// Wrap a mock-server client as the worker's single-token pool. +fn single_client(octo: Arc) -> Arc> { + Arc::new(vec![TokenClient::new("test-token", "t", octo)]) +} + /// Convert a test configuration into the runtime `Config`. /// /// The conversion normally relies on the crate's `test-support` feature. @@ -308,7 +313,11 @@ mod worker_tests { drained: None, }, }; - let h = tokio::spawn(run_worker(ctx.queue.clone(), ctx.octo, control)); + let h = tokio::spawn(run_worker( + ctx.queue.clone(), + single_client(ctx.octo), + control, + )); timeout_with_retries(DRAINED_NOTIFICATION, "worker idle notification", || { let idle = idle.clone(); @@ -368,7 +377,11 @@ mod worker_tests { drained: None, }, }; - let h = tokio::spawn(run_worker(ctx.queue.clone(), ctx.octo, control)); + let h = tokio::spawn(run_worker( + ctx.queue.clone(), + single_client(ctx.octo), + control, + )); timeout_with_retries(WORKER_SUCCESS, "worker enqueued", || { let enqueued = enqueued.clone(); @@ -426,7 +439,7 @@ mod worker_tests { hooks: WorkerHooks::default(), }; - let h = tokio::spawn(run_worker(queue, octo, control)); + let h = tokio::spawn(run_worker(queue, single_client(octo), control)); // Give worker time to park on the change signal sleep(Duration::from_millis(50)).await; @@ -489,7 +502,7 @@ mod worker_tests { }, }; - let h = tokio::spawn(run_worker(queue, octo, control)); + let h = tokio::spawn(run_worker(queue, single_client(octo), control)); // Wait for idle notification (first entry posted; worker now waits a // full cooldown before the second). diff --git a/src/protocol.rs b/src/protocol.rs index f40f280..9371d75 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -75,6 +75,8 @@ pub struct HistoryEntry { /// Failure description when `success` is false. #[serde(default, skip_serializing_if = "Option::is_none")] pub error: Option, + /// Hex SHA-256 of the token that made the attempt. + pub token_hash: String, /// Repository owner. pub owner: String, /// Repository name. @@ -181,6 +183,7 @@ mod tests { posted_at: 1_700_000_000, success: false, error: Some("timeout".into()), + token_hash: "ab".repeat(32), owner: "octocat".into(), repo: "hello-world".into(), pr_number: 7, diff --git a/test-support/src/daemon.rs b/test-support/src/daemon.rs index 48b753c..f9a0f87 100644 --- a/test-support/src/daemon.rs +++ b/test-support/src/daemon.rs @@ -108,9 +108,25 @@ impl TestConfig { /// The error is boxed to keep the `Err` variant small /// (`clippy::result_large_err`). pub fn octocrab_for(server: &MockServer) -> Result, Box> { + octocrab_with_token(server, "t") +} + +/// Construct an [`Octocrab`] client for a [`MockServer`] using `token`. +/// +/// Like [`octocrab_for`], but authenticating with the supplied token so +/// tests can distinguish requests made by different tokens. +/// +/// # Errors +/// +/// Returns an error when the mock server URI cannot be parsed or the client +/// cannot be built. +pub fn octocrab_with_token( + server: &MockServer, + token: &str, +) -> Result, Box> { Ok(Arc::new( Octocrab::builder() - .personal_token(String::from("t")) + .personal_token(token.to_owned()) .base_uri(server.uri()) .map_err(Box::new)? .build() diff --git a/test-support/src/lib.rs b/test-support/src/lib.rs index 5bfab22..cf314c4 100644 --- a/test-support/src/lib.rs +++ b/test-support/src/lib.rs @@ -8,7 +8,7 @@ mod workflow; pub use workflow::uses_shared_release_actions; // Re-exports from daemon module (added in main) -pub use daemon::{TestConfig, octocrab_for, temp_config}; +pub use daemon::{TestConfig, octocrab_for, octocrab_with_token, temp_config}; pub use env_guard::EnvVarGuard; // Re-exports from util module with documentation (from your branch) diff --git a/tests/features/worker.feature b/tests/features/worker.feature index 17134d0..b872268 100644 --- a/tests/features/worker.feature +++ b/tests/features/worker.feature @@ -7,6 +7,22 @@ Feature: Worker task Then the comment is posted And the posting history records the success + Scenario: posts rotate through the configured tokens + Given two queued comment requests + And two GitHub tokens are configured + And GitHub returns success + When the worker drains the queue + Then the posting history alternates between the two tokens + + Scenario: rotation resumes from the recorded history + Given a queued comment request + And two GitHub tokens are configured + And the history already records a post with the second token + And GitHub returns success + When the worker runs briefly + Then the comment is posted + And the newest history record uses the first token + Scenario: API failure requeues job Given a queued comment request And GitHub returns an error diff --git a/tests/steps/history_steps.rs b/tests/steps/history_steps.rs index 577d34b..287c43c 100644 --- a/tests/steps/history_steps.rs +++ b/tests/steps/history_steps.rs @@ -16,6 +16,9 @@ use cucumber::{World, given, then, when}; use tempfile::TempDir; use test_support::temp_config; +/// Token hash recorded against simulated posting attempts. +const FIRST_TOKEN_HASH: &str = "feedbead0011"; + #[derive(Default, World)] pub struct HistoryWorld { dir: Option, @@ -122,7 +125,11 @@ async fn three_comments_queued( #[when("the head comment is posted successfully")] async fn head_posted(world: &mut HistoryWorld) -> anyhow::Result<()> { let entry = world.head().await?; - world.queue()?.complete(&entry).await.context("complete")?; + world + .queue()? + .complete(&entry, FIRST_TOKEN_HASH) + .await + .context("complete")?; world.recorded.push(entry.id); Ok(()) } @@ -146,7 +153,7 @@ async fn head_fails(world: &mut HistoryWorld, error: String) -> anyhow::Result<( let entry = world.head().await?; world .queue()? - .record_failure(&entry, &error) + .record_failure(&entry, FIRST_TOKEN_HASH, &error) .await .context("record failure")?; world.recorded.push(entry.id); diff --git a/tests/steps/mod.rs b/tests/steps/mod.rs index 1941ae9..0feb347 100644 --- a/tests/steps/mod.rs +++ b/tests/steps/mod.rs @@ -19,3 +19,4 @@ pub mod release_steps; pub use release_steps::ReleaseWorld; pub mod worker_steps; pub use worker_steps::WorkerWorld; +pub mod worker_token_steps; diff --git a/tests/steps/worker_steps.rs b/tests/steps/worker_steps.rs index 3ce808e..c8d41b3 100644 --- a/tests/steps/worker_steps.rs +++ b/tests/steps/worker_steps.rs @@ -13,10 +13,10 @@ use anyhow::Context as _; use comenq_lib::CommentRequest; use comenq_lib::protocol::{HistoryEntry, PendingEntry, Request, Response}; use comenqd::config::Config; -use comenqd::daemon::{SharedQueue, WorkerControl, WorkerHooks, run_worker}; +use comenqd::daemon::{SharedQueue, TokenClient, WorkerControl, WorkerHooks, run_worker}; use cucumber::{World, given, then, when}; use tempfile::TempDir; -use test_support::{octocrab_for, temp_config}; +use test_support::{octocrab_with_token, temp_config}; use tokio::sync::{Notify, watch}; use tokio::time::timeout; use wiremock::matchers::{method, path}; @@ -32,11 +32,15 @@ fn coverage_timeout_multiplier() -> u32 { } } +/// Token pool used when no rotation is configured: one anonymous token. +const DEFAULT_TOKENS: [(&str, &str); 1] = [("test-token", "t")]; #[derive(World, Default)] pub struct WorkerWorld { dir: Option, - queue: Option>, - server: Option, + pub(crate) queue: Option>, + pub(crate) server: Option, + /// Rotation pool (name, token value); empty means the single default. + pub(crate) tokens: Vec<(String, String)>, shutdown: Option>, handle: Option>, } @@ -59,7 +63,7 @@ impl Drop for WorkerWorld { } impl WorkerWorld { - async fn shutdown_and_join(&mut self) { + pub(crate) async fn shutdown_and_join(&mut self) { if let Some(tx) = self.shutdown.take() { let _ = tx.send(()); } @@ -81,7 +85,7 @@ async fn listed_entries(queue: &Arc) -> anyhow::Result) -> anyhow::Result> { +pub(crate) async fn listed_history(queue: &Arc) -> anyhow::Result> { match queue.execute(Request::Hist { limit: None }).await { Response::Ok { history: Some(history), @@ -91,31 +95,42 @@ async fn listed_history(queue: &Arc) -> anyhow::Result anyhow::Result<()> { +async fn queue_with_entries(world: &mut WorkerWorld, bodies: &[&str]) -> anyhow::Result<()> { let dir = TempDir::new().context("tempdir")?; let cfg = Arc::new(Config::from(temp_config(&dir).with_cooldown(0))); let queue = SharedQueue::open(cfg).context("open queue")?; - let response = queue - .execute(Request::Put { - request: CommentRequest { - owner: "o".into(), - repo: "r".into(), - pr_number: 1, - body: "b".into(), - }, - immediate: true, - }) - .await; - anyhow::ensure!( - matches!(response, Response::Ok { .. }), - "put should succeed, got {response:?}" - ); + for body in bodies { + let response = queue + .execute(Request::Put { + request: CommentRequest { + owner: "o".into(), + repo: "r".into(), + pr_number: 1, + body: (*body).to_owned(), + }, + immediate: true, + }) + .await; + anyhow::ensure!( + matches!(response, Response::Ok { .. }), + "put should succeed, got {response:?}" + ); + } world.dir = Some(dir); world.queue = Some(queue); Ok(()) } +#[given("a queued comment request")] +async fn queued_request(world: &mut WorkerWorld) -> anyhow::Result<()> { + queue_with_entries(world, &["b"]).await +} + +#[given("two queued comment requests")] +async fn two_queued_requests(world: &mut WorkerWorld) -> anyhow::Result<()> { + queue_with_entries(world, &["first", "second"]).await +} + #[given("GitHub returns success")] async fn github_success(world: &mut WorkerWorld) -> anyhow::Result<()> { let server = MockServer::start().await; @@ -145,8 +160,33 @@ async fn github_error(world: &mut WorkerWorld) { world.server = Some(server); } -#[when("the worker runs briefly")] -async fn worker_runs(world: &mut WorkerWorld) -> anyhow::Result<()> { +/// Which worker milestone a scenario waits for before asserting. +enum WaitFor { + /// One processing pass has completed. + Idle, + /// The queue is empty and the worker is parked. + Drained, +} + +fn token_clients(world: &WorkerWorld, server: &MockServer) -> anyhow::Result> { + let pool: Vec<(String, String)> = if world.tokens.is_empty() { + DEFAULT_TOKENS + .iter() + .map(|(name, value)| ((*name).to_owned(), (*value).to_owned())) + .collect() + } else { + world.tokens.clone() + }; + pool.into_iter() + .map(|(name, value)| { + let octocrab = octocrab_with_token(server, &value) + .context("octocrab client should build for the mock server")?; + Ok(TokenClient::new(name, &value, octocrab)) + }) + .collect() +} + +async fn start_worker(world: &mut WorkerWorld, wait_for: WaitFor) -> anyhow::Result<()> { let queue = world .queue .as_ref() @@ -156,29 +196,33 @@ async fn worker_runs(world: &mut WorkerWorld) -> anyhow::Result<()> { .server .as_ref() .context("server should be initialized")?; - let octocrab = - octocrab_for(server).context("octocrab client should build for the mock server")?; + let clients = Arc::new(token_clients(world, server)?); let (shutdown_tx, shutdown_rx) = watch::channel(()); - let idle = Arc::new(Notify::new()); - let idle_for_wait = Arc::clone(&idle); - let idle_notified = idle_for_wait.notified(); - let control = WorkerControl::new( - shutdown_rx, - WorkerHooks { + let milestone = Arc::new(Notify::new()); + let milestone_for_wait = Arc::clone(&milestone); + let milestone_notified = milestone_for_wait.notified(); + let hooks = match wait_for { + WaitFor::Idle => WorkerHooks { enqueued: None, - idle: Some(idle), + idle: Some(milestone), drained: None, }, - ); + WaitFor::Drained => WorkerHooks { + enqueued: None, + idle: None, + drained: Some(milestone), + }, + }; + let control = WorkerControl::new(shutdown_rx, hooks); let handle = tokio::spawn(async move { - let _ = run_worker(queue, octocrab, control).await; + let _ = run_worker(queue, clients, control).await; }); timeout( Duration::from_secs(30 * u64::from(coverage_timeout_multiplier())), - idle_notified, + milestone_notified, ) .await - .context("worker reached idle state within timeout")?; + .context("worker reached the awaited state within timeout")?; // Store handles in world for proper cleanup world.shutdown = Some(shutdown_tx); @@ -186,6 +230,16 @@ async fn worker_runs(world: &mut WorkerWorld) -> anyhow::Result<()> { Ok(()) } +#[when("the worker runs briefly")] +async fn worker_runs(world: &mut WorkerWorld) -> anyhow::Result<()> { + start_worker(world, WaitFor::Idle).await +} + +#[when("the worker drains the queue")] +async fn worker_drains(world: &mut WorkerWorld) -> anyhow::Result<()> { + start_worker(world, WaitFor::Drained).await +} + #[then("the comment is posted")] async fn comment_posted(world: &mut WorkerWorld) -> anyhow::Result<()> { let server = world diff --git a/tests/steps/worker_token_steps.rs b/tests/steps/worker_token_steps.rs new file mode 100644 index 0000000..277dec3 --- /dev/null +++ b/tests/steps/worker_token_steps.rs @@ -0,0 +1,114 @@ +//! Behavioural test steps for round-robin token rotation. +//! +//! Extends the worker scenarios with a two-token pool, asserting that +//! posts alternate tokens and that rotation resumes from the hash the +//! latest history record carries. + +use anyhow::Context as _; +use comenq_lib::CommentRequest; +use comenqd::daemon::token_hash; +use comenqd::store::{HistoryRecord, QueueStore, StoredEntry}; +use cucumber::{given, then}; + +use super::worker_steps::{WorkerWorld, listed_history}; + +/// Rotation pool used by the token scenarios, in configured order. +pub(crate) const ROTATION_TOKENS: [(&str, &str); 2] = [ + ("pandalump-token", "alpha-token"), + ("buzzybee-token", "bravo-token"), +]; + +#[given("two GitHub tokens are configured")] +fn two_tokens_configured(world: &mut WorkerWorld) { + world.tokens = ROTATION_TOKENS + .iter() + .map(|(name, value)| ((*name).to_owned(), (*value).to_owned())) + .collect(); +} + +#[given("the history already records a post with the second token")] +fn history_seeded_with_second_token(world: &mut WorkerWorld) -> anyhow::Result<()> { + let queue = world.queue.as_ref().context("queue initialized")?; + let store = QueueStore::open(&queue.config().queue_path).context("open store")?; + let (_, second_value) = ROTATION_TOKENS.get(1).context("second rotation token")?; + let posted = StoredEntry { + id: "deadbeef".into(), + order: 0, + flutter_seconds: 0, + enqueued_at: 0, + not_before: 0, + request: CommentRequest { + owner: "o".into(), + repo: "r".into(), + pr_number: 1, + body: "earlier".into(), + }, + }; + store + .append_history(&HistoryRecord::success( + &posted, + &token_hash(second_value), + 1, + )) + .context("seed history")?; + Ok(()) +} + +#[then("the posting history alternates between the two tokens")] +async fn history_alternates(world: &mut WorkerWorld) -> anyhow::Result<()> { + let queue = world.queue.as_ref().context("queue initialized")?.clone(); + let history = listed_history(&queue).await?; + let hashes: Vec<&str> = history + .iter() + .map(|record| record.token_hash.as_str()) + .collect(); + let expected: Vec = ROTATION_TOKENS + .iter() + .map(|(_, value)| token_hash(value)) + .collect(); + assert_eq!( + hashes, + expected.iter().map(String::as_str).collect::>(), + "the first post should use the first token, the second the next" + ); + let server = world + .server + .as_ref() + .context("server should be initialized")?; + let auth_headers: Vec = server + .received_requests() + .await + .context("inbound requests should be recorded")? + .iter() + .map(|request| { + request + .headers + .get("authorization") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_owned() + }) + .collect(); + assert_eq!(auth_headers.len(), 2, "both posts should reach GitHub"); + assert_ne!( + auth_headers.first(), + auth_headers.get(1), + "each post should authenticate with a different token" + ); + world.shutdown_and_join().await; + Ok(()) +} + +#[then("the newest history record uses the first token")] +async fn newest_record_uses_first_token(world: &mut WorkerWorld) -> anyhow::Result<()> { + let queue = world.queue.as_ref().context("queue initialized")?.clone(); + let history = listed_history(&queue).await?; + let newest = history.last().context("history should hold records")?; + let (_, first_value) = ROTATION_TOKENS.first().context("first rotation token")?; + assert_eq!( + newest.token_hash, + token_hash(first_value), + "rotation should wrap from the second token back to the first" + ); + Ok(()) +} From 17f066fa0df664bc02d9d7dd7b87a68b809fef7b Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 21 Jul 2026 19:57:59 +0200 Subject: [PATCH 2/3] Document round-robin token rotation Co-Authored-By: Claude Fable 5 --- README.md | 17 ++++++++++++++- docs/comenq-design.md | 41 +++++++++++++++++++++++++---------- packaging/config/comenqd.toml | 5 +++++ packaging/man/comenq.1 | 2 +- packaging/man/comenqd.1 | 14 ++++++++++++ 5 files changed, 65 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index 7d063f7..38427bb 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,8 @@ comenq del 1a2b3c4d # remove from the queue ``` The daemon also keeps a posting-history log, recording every successful and -failed posting attempt. `hist` prints it, oldest first: +failed posting attempt. `hist` prints it, oldest first, including a short +eight-character hash identifying which token made the attempt: ```bash comenq hist -n 20 # the twenty most recent posting attempts @@ -72,3 +73,17 @@ supplied through systemd's credential system (`LoadCredential=token:%h/pandalump-token` with `github_token_file = "${CREDENTIALS_DIRECTORY}/token"`), keeping the secret out of the unit file and process environment. + +### Rotating multiple GitHub tokens + +To spread posting across several personal access tokens, set +`github_token_files` instead of `github_token`/`github_token_file`: + +```toml +github_token_files = ["/home/user/pandalump-token", "/home/user/buzzybee-token"] +``` + +Each file must exist and hold a non-empty token; this is checked at startup. +The daemon then posts each comment with the successor of whichever token +posted last, wrapping back to the first token at the end of the list, so the +configured tokens are used in round-robin rotation. diff --git a/docs/comenq-design.md b/docs/comenq-design.md index 7fcd57a..76e1c2b 100644 --- a/docs/comenq-design.md +++ b/docs/comenq-design.md @@ -384,13 +384,14 @@ operation, all sharing a global `--socket` flag (also settable via the - `comenq hist [-n|--limit ]`: prints the posting-history log, oldest first, one line per record: the identifier, the age of the attempt (e.g. - `2m 30s ago`, or `just now`), the outcome (`ok` or `FAIL`), the target - `owner/repo#pr`, and the comment body collapsed to a single line of at most - 60 characters. Failed attempts append a one-line, em-dash-separated - description of the error. Records are always shown in chronological order - (oldest first); `--limit` restricts the output to the most recent `N` - records without changing that ordering. If no attempts have been recorded, - the client prints `No posting history recorded.` instead of a table. + `2m 30s ago`, or `just now`), the outcome (`ok` or `FAIL`), the first eight + characters of the posting token's hash, the target `owner/repo#pr`, and the + comment body collapsed to a single line of at most 60 characters. Failed + attempts append a one-line, em-dash-separated description of the error. + Records are always shown in chronological order (oldest first); `--limit` + restricts the output to the most recent `N` records without changing that + ordering. If no attempts have been recorded, the client prints `No posting + history recorded.` instead of a table. Each subcommand maps directly to one variant of the `Request` enum (see §3.2 for the store and §5 for the wire protocol) and prints a short @@ -562,8 +563,11 @@ appends one JSON line per posting attempt to `/history.jsonl` (JSON Lines: append-only, one record per line). Each `HistoryRecord` carries the entry's `id` (the identifier it had while queued), `posted_at` (Unix seconds), `success`, an `error` description (present only when `success` is -false), and the `request` that was posted or attempted. A successful post is -recorded once, when `complete` removes the entry from the queue and updates +false), `token_hash` (the hex SHA-256 of the value of the token that made the +attempt), and the `request` that was posted or attempted. Hashing rather than +naming the token means rotation state (see below) survives token files being +renamed, and the log never holds a secret. A successful post is recorded +once, when `complete` removes the entry from the queue and updates `last_post`. A failed post is recorded on every retry attempt, since the entry stays queued and is retried after a cooldown rather than being removed; each attempt therefore contributes its own failure record. Writing the @@ -574,6 +578,17 @@ back (to serve the `hist` protocol operation), a line that fails to deserialize is skipped with an error log rather than aborting the read, so one corrupt record cannot hide the rest of the history. +**Round-robin token selection.** When `github_token_files` configures more +than one token, the worker picks the token for the next post from the +history: it looks up the token whose hash matches the most recent history +record and uses its successor, in configured order, wrapping back to the +first token after the last. The first configured token is used when the +history is empty or the last record's hash matches no configured token (for +example, after the token files are reconfigured). With a single configured +token (via `github_token`, `github_token_file`, or a one-element +`github_token_files`), this always selects the same token, so rotation is a +strict generalization of the single-token case. + ### 3.3. The UDS Listener and Request Ingestion (`run_listener`) This task is responsible for handling all client communication. It will be @@ -717,8 +732,9 @@ at `/etc/comenqd/config.toml` is the conventional choice. | Parameter | Type | Description | Default Value | | ------------------------ | ------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------- | -| github_token | String | The GitHub Personal Access Token (PAT) used for authentication. Required unless `github_token_file` is set. | (none) | +| github_token | String | The GitHub Personal Access Token (PAT) used for authentication. Required unless `github_token_file` or `github_token_files` is set. | (none) | | github_token_file | PathBuf | Optional path to a file containing the PAT. Read at startup; its trimmed contents override `github_token`. A leading `${VAR}` placeholder is expanded from the environment, enabling systemd `LoadCredential` integration. | (none) | +| github_token_files | Vec\ | Optional array of paths, each containing one PAT. When non-empty, neither `github_token` nor `github_token_file` is required, and the daemon posts comments with these tokens in round-robin rotation (see §3.2). Each file is read and validated (must exist and hold a non-empty token) at startup. | (empty) | | socket_path | PathBuf | The filesystem path for the Unix Domain Socket. | `$XDG_RUNTIME_DIR/comenq/comenq.sock` when a user runtime directory is available, else /run/comenq/comenq.sock | | queue_path | PathBuf | The directory path for the persistent queue data (`entries/` and `last_post`, managed by `QueueStore`). | /var/lib/comenq/queue | | log_level | String | The minimum log level to record (e.g., "info", "debug", "trace"). | info | @@ -731,8 +747,9 @@ Configuration is loaded using the `ortho_config` crate. The daemon calls `COMENQD_*` environment variables, and any supplied CLI arguments. CLI arguments have the highest precedence, followed by environment variables, and finally the configuration file. Missing optional fields are replaced with -defaults, while an absent `github_token` or invalid TOML results in a -configuration error. +defaults, while a missing token (none of `github_token`, `github_token_file`, +or `github_token_files` set), an unreadable or empty token file, or invalid +TOML results in a configuration error. Robust logging is non-negotiable for a background process. The `tracing` crate with `tracing-subscriber` will be used to provide structured, asynchronous diff --git a/packaging/config/comenqd.toml b/packaging/config/comenqd.toml index e57af6a..bb19df9 100644 --- a/packaging/config/comenqd.toml +++ b/packaging/config/comenqd.toml @@ -9,6 +9,11 @@ # environment, enabling systemd LoadCredential integration: # github_token_file = "${CREDENTIALS_DIRECTORY}/token" +# Paths to files each containing one token. When non-empty, neither +# github_token nor github_token_file is required, and the daemon posts +# comments with these tokens in round-robin rotation: +# github_token_files = ["/home/user/pandalump-token", "/home/user/buzzybee-token"] + # Minimum log level to output # log_level = "info" diff --git a/packaging/man/comenq.1 b/packaging/man/comenq.1 index f154265..fcd8d49 100644 --- a/packaging/man/comenq.1 +++ b/packaging/man/comenq.1 @@ -63,7 +63,7 @@ Print the posting-history log, oldest first: identifier, age, status ( .I ok or .IR FAIL ), -target +the first eight characters of the posting token's hash, target .IR owner/repo#pr , and the comment text collapsed to a single line of at most 60 characters. Failed attempts also show a one-line description of the error. If nothing diff --git a/packaging/man/comenqd.1 b/packaging/man/comenqd.1 index 5332d9c..32f387d 100644 --- a/packaging/man/comenqd.1 +++ b/packaging/man/comenqd.1 @@ -48,6 +48,20 @@ environment, so a systemd unit using .B LoadCredential=token:... can reference \fB${CREDENTIALS_DIRECTORY}/token\fR. .PP +Alternatively, +.B github_token_files +names an array of files, each holding one token, e.g. +.BR github_token_files\ =\ ["/home/user/pandalump-token",\ "/home/user/buzzybee-token"] . +When non-empty, neither +.B github_token +nor +.B github_token_file +is required, and the daemon posts comments with the listed tokens in +round-robin rotation: each post uses the successor (in configured order, +wrapping) of the token that made the most recent posting attempt. Every +file is read and validated at startup; a missing file or an empty token +prevents the daemon from starting. +.PP The cooldown between comment posts is set by .B cooldown_period_seconds and may be lengthened by a random flutter of up to From 6fc6a02c914509b96f49a528dc3f580ef44808a3 Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 21 Jul 2026 19:58:45 +0200 Subject: [PATCH 3/3] Realign the configuration table after adding github_token_files Co-Authored-By: Claude Fable 5 --- docs/comenq-design.md | 219 +++++++++++++++++++++--------------------- 1 file changed, 109 insertions(+), 110 deletions(-) diff --git a/docs/comenq-design.md b/docs/comenq-design.md index 76e1c2b..de60581 100644 --- a/docs/comenq-design.md +++ b/docs/comenq-design.md @@ -92,10 +92,10 @@ configuration, and a `tokio::sync::Notify` used to wake the worker promptly when the queue changes. - The listener executes each client request directly against the shared - queue (via `SharedQueue::execute`) and writes the reply before the - connection closes. A request is durably persisted to disk before the client - receives its acknowledgement, so there is no in-memory backlog that could be - lost if the daemon exits unexpectedly. + queue (via `SharedQueue::execute`) and writes the reply before the connection + closes. A request is durably persisted to disk before the client receives its + acknowledgement, so there is no in-memory backlog that could be lost if the + daemon exits unexpectedly. - Mutating operations (`put`, `bump`, `bust`, `del`) call `notify_one` on the shared `Notify` after the change is written, so the worker interrupts any @@ -360,21 +360,22 @@ handling, so `run` operates on validated data without rechecking it. ### 2.4. Client Subcommands and ETA Semantics The illustrative blueprint above shows a client that only enqueues a comment. -The shipped client instead exposes six subcommands, one per protocol -operation, all sharing a global `--socket` flag (also settable via the -`COMENQ_SOCKET` environment variable): +The shipped client instead exposes six subcommands, one per protocol operation, +all sharing a global `--socket` flag (also settable via the `COMENQ_SOCKET` +environment variable): - `comenq put `: enqueues a comment and - prints its identifier and an approximate ETA, e.g. `Queued 1a2b3c4d for - octocat/hello-world#7 — posts in ~1h 01m`. By default the comment waits one - full cooldown (plus its flutter) from enqueue even when the queue is idle; - `--now` lifts that floor so the comment posts as soon as the queue allows. + prints its identifier and an approximate ETA, e.g. + `Queued 1a2b3c4d for octocat/hello-world#7 — posts in ~1h 01m`. By default + the comment waits one full cooldown (plus its flutter) from enqueue even when + the queue is idle; `--now` lifts that floor so the comment posts as soon as + the queue allows. - `comenq list`: prints one line per pending comment, in posting order, each - showing the identifier, the ETA, the target `owner/repo#pr`, and the - comment body collapsed to a single line of at most 60 characters (control - characters, including newlines, become spaces; longer bodies are truncated - with an ellipsis). + showing the identifier, the ETA, the target `owner/repo#pr`, and the comment + body collapsed to a single line of at most 60 characters (control characters, + including newlines, become spaces; longer bodies are truncated with an + ellipsis). - `comenq bump `: moves the identified comment to the head of the queue. @@ -390,13 +391,12 @@ operation, all sharing a global `--socket` flag (also settable via the attempts append a one-line, em-dash-separated description of the error. Records are always shown in chronological order (oldest first); `--limit` restricts the output to the most recent `N` records without changing that - ordering. If no attempts have been recorded, the client prints `No posting - history recorded.` instead of a table. + ordering. If no attempts have been recorded, the client prints + `No posting history recorded.` instead of a table. -Each subcommand maps directly to one variant of the `Request` enum (see -§3.2 for the store and §5 for the wire protocol) and prints a short -confirmation, or the daemon's error message, once the single reply is -received. +Each subcommand maps directly to one variant of the `Request` enum (see §3.2 +for the store and §5 for the wire protocol) and prints a short confirmation, or +the daemon's error message, once the single reply is received. **ETA semantics.** The estimated time until posting shown by `put` and `list` reflects three rules enforced by the daemon: @@ -404,13 +404,13 @@ reflects three rules enforced by the daemon: - **Flutter is fixed at enqueue time.** When a comment is enqueued, a random flutter duration (up to `cooldown_flutter_seconds`) is sampled once and stored with the entry. It does not change on subsequent `list` calls, so the - reported ETA for a given entry only decreases as time passes; it never - jumps around. + reported ETA for a given entry only decreases as time passes; it never jumps + around. - **The cooldown always runs in full.** The daemon never shortens - `cooldown_period_seconds` to catch up; each entry's projected posting time - is its predecessor's projected posting time plus a full cooldown plus its - own flutter (or, for the head entry, the last successful post plus a full + `cooldown_period_seconds` to catch up; each entry's projected posting time is + its predecessor's projected posting time plus a full cooldown plus its own + flutter (or, for the head entry, the last successful post plus a full cooldown plus its own flutter). This keeps the reported ETA consistent with what the worker will actually do. @@ -437,9 +437,9 @@ asynchronous tasks that run concurrently for the lifetime of the daemon: 1. **Listener** (`run_listener`): This task is the daemon's public-facing interface. It binds to the UDS and listens for incoming connections from - `comenq` clients. For each connection it reads one JSON `Request`, - executes it directly against the shared queue, and writes back one JSON - `Response` before closing the connection. + `comenq` clients. For each connection it reads one JSON `Request`, executes + it directly against the shared queue, and writes back one JSON `Response` + before closing the connection. 2. **Worker** (`run_worker`): This is the main worker task. It operates in a loop, recomputing the head entry's due time on every iteration, waiting @@ -447,24 +447,24 @@ asynchronous tasks that run concurrently for the lifetime of the daemon: signalled), posting the comment to GitHub, and recording the post before moving on to the next entry. -This concurrent design ensures that the daemon remains responsive to new -client requests even while the worker task is in its long sleep phase. A -request can be accepted, persisted, and acknowledged in milliseconds, while -the worker task independently processes the queue at its own deliberate pace. +This concurrent design ensures that the daemon remains responsive to new client +requests even while the worker task is in its long sleep phase. A request can +be accepted, persisted, and acknowledged in milliseconds, while the worker task +independently processes the queue at its own deliberate pace. -Both tasks share the queue through an `Arc` rather than a -channel: there is no intermediate queue-writer task. The listener performs -each mutation directly, and the worker is woken (via a `tokio::sync::Notify`) -whenever a mutation occurs. +Both tasks share the queue through an `Arc` rather than a channel: +there is no intermediate queue-writer task. The listener performs each mutation +directly, and the worker is woken (via a `tokio::sync::Notify`) whenever a +mutation occurs. Both the listener and the worker are supervised. If either task exits unexpectedly, the daemon logs the failure, waits using an exponential backoff with jitter (via the `backon` crate) to avoid a tight restart loop, and then respawns the task. The minimum delay between restarts is configurable via -`restart_min_delay_ms`. This keeps the service available without relying on -an external process supervisor. Because both tasks operate on the same -on-disk store rather than an in-memory buffer, restarting either task after a -panic does not discard any pending request. +`restart_min_delay_ms`. This keeps the service available without relying on an +external process supervisor. Because both tasks operate on the same on-disk +store rather than an in-memory buffer, restarting either task after a panic +does not discard any pending request. The supervision and restart behaviour is illustrated in the sequence diagram below. @@ -517,9 +517,9 @@ bespoke store, `comenqd::store::QueueStore`, built directly on plain files. `QueueStore` keeps one JSON file per pending comment under `/entries`, plus a `/last_post` file recording the Unix -time of the most recent successful post and a `/history.jsonl` -file recording every posting attempt, successful or failed (see "The posting -history log" below). Each entry (a `StoredEntry`) carries: +time of the most recent successful post and a `/history.jsonl` file +recording every posting attempt, successful or failed (see "The posting history +log" below). Each entry (a `StoredEntry`) carries: - **A deterministic eight-character identifier.** This is the first eight hex digits of a 64-bit FNV-1a hash computed over the request's owner, repo, PR @@ -529,10 +529,10 @@ history log" below). Each entry (a `StoredEntry`) carries: existing entry unchanged rather than creating a duplicate. - **An explicit integer ordering key.** New entries are appended after the - current tail. `bump` sets the key to one less than the current head's key - (or leaves it unchanged if the entry is already the head); `bust` sets it to - one more than the current tail's key. Repeated bumps or busts can drive the - key negative or arbitrarily large; entries are always read back sorted by + current tail. `bump` sets the key to one less than the current head's key (or + leaves it unchanged if the entry is already the head); `bust` sets it to one + more than the current tail's key. Repeated bumps or busts can drive the key + negative or arbitrarily large; entries are always read back sorted by `(order, enqueued_at, id)`. - **The flutter sampled at enqueue time.** A random duration up to @@ -544,10 +544,10 @@ history log" below). Each entry (a `StoredEntry`) carries: - **The enqueue time**, used both as a tiebreaker for ordering and as an input to the identifier hash. -Entries are written to a temporary sibling file and then renamed into place, -so a reader never observes a half-written entry. `bump`, `bust`, and `del` -mutate or remove a single entry file the same way. `complete` removes the -posted entry and atomically rewrites `last_post` with the current Unix time. +Entries are written to a temporary sibling file and then renamed into place, so +a reader never observes a half-written entry. `bump`, `bust`, and `del` mutate +or remove a single entry file the same way. `complete` removes the posted entry +and atomically rewrites `last_post` with the current Unix time. Because an entry is only removed from disk after a successful post (`complete`), and a failed post simply leaves the entry in place for the next @@ -559,35 +559,35 @@ and stores the `CommentRequest` struct defined in the shared library as part of each entry. **The posting history log.** Alongside `entries` and `last_post`, the store -appends one JSON line per posting attempt to `/history.jsonl` -(JSON Lines: append-only, one record per line). Each `HistoryRecord` carries -the entry's `id` (the identifier it had while queued), `posted_at` (Unix -seconds), `success`, an `error` description (present only when `success` is -false), `token_hash` (the hex SHA-256 of the value of the token that made the -attempt), and the `request` that was posted or attempted. Hashing rather than -naming the token means rotation state (see below) survives token files being -renamed, and the log never holds a secret. A successful post is recorded -once, when `complete` removes the entry from the queue and updates -`last_post`. A failed post is recorded on every retry attempt, since the -entry stays queued and is retried after a cooldown rather than being removed; -each attempt therefore contributes its own failure record. Writing the -history record on the failure path can itself fail (e.g. a full disk); the -worker logs that error but never lets it interrupt the retry loop, so a -history-logging fault cannot stall comment posting. When the log is read -back (to serve the `hist` protocol operation), a line that fails to -deserialize is skipped with an error log rather than aborting the read, so -one corrupt record cannot hide the rest of the history. - -**Round-robin token selection.** When `github_token_files` configures more -than one token, the worker picks the token for the next post from the -history: it looks up the token whose hash matches the most recent history -record and uses its successor, in configured order, wrapping back to the -first token after the last. The first configured token is used when the -history is empty or the last record's hash matches no configured token (for -example, after the token files are reconfigured). With a single configured -token (via `github_token`, `github_token_file`, or a one-element -`github_token_files`), this always selects the same token, so rotation is a -strict generalization of the single-token case. +appends one JSON line per posting attempt to `/history.jsonl` (JSON +Lines: append-only, one record per line). Each `HistoryRecord` carries the +entry's `id` (the identifier it had while queued), `posted_at` (Unix seconds), +`success`, an `error` description (present only when `success` is false), +`token_hash` (the hex SHA-256 of the value of the token that made the attempt), +and the `request` that was posted or attempted. Hashing rather than naming the +token means rotation state (see below) survives token files being renamed, and +the log never holds a secret. A successful post is recorded once, when +`complete` removes the entry from the queue and updates `last_post`. A failed +post is recorded on every retry attempt, since the entry stays queued and is +retried after a cooldown rather than being removed; each attempt therefore +contributes its own failure record. Writing the history record on the failure +path can itself fail (e.g. a full disk); the worker logs that error but never +lets it interrupt the retry loop, so a history-logging fault cannot stall +comment posting. When the log is read back (to serve the `hist` protocol +operation), a line that fails to deserialize is skipped with an error log +rather than aborting the read, so one corrupt record cannot hide the rest of +the history. + +**Round-robin token selection.** When `github_token_files` configures more than +one token, the worker picks the token for the next post from the history: it +looks up the token whose hash matches the most recent history record and uses +its successor, in configured order, wrapping back to the first token after the +last. The first configured token is used when the history is empty or the last +record's hash matches no configured token (for example, after the token files +are reconfigured). With a single configured token (via `github_token`, +`github_token_file`, or a one-element `github_token_files`), this always +selects the same token, so rotation is a strict generalization of the +single-token case. ### 3.3. The UDS Listener and Request Ingestion (`run_listener`) @@ -618,8 +618,8 @@ Its workflow is as follows: adjusted. It deserializes the received JSON into a `Request`, executes it directly against the shared queue (`SharedQueue::execute`), and writes the resulting `Response` back to the client before closing the connection. A - request that fails to deserialize receives an error `Response` rather than - a silently dropped connection. + request that fails to deserialize receives an error `Response` rather than a + silently dropped connection. This design makes the request ingestion process highly concurrent and robust, capable of handling multiple simultaneous client connections without impacting @@ -649,9 +649,8 @@ sequenceDiagram ### 3.4. The GitHub Comment-Posting Worker (`run_worker`) -This task implements the core business logic of the service. It runs in a -loop, ensuring that comments are processed one by one with the required -delay. +This task implements the core business logic of the service. It runs in a loop, +ensuring that comments are processed one by one with the required delay. #### 3.4.1. `octocrab` Initialization and API Usage @@ -686,9 +685,9 @@ The worker task's loop consists of the following steps: 2. **Wait until due:** If nothing is queued, the worker waits for the shared queue's change signal or a shutdown signal. If the head entry is not yet due, the worker sleeps for the remaining time, but the sleep is - interruptible: a queue change (e.g. a `bump` promoting a different entry) - or shutdown wakes it early, and it loops back to recompute the due entry - rather than blindly posting whichever entry it started waiting for. + interruptible: a queue change (e.g. a `bump` promoting a different entry) or + shutdown wakes it early, and it loops back to recompute the due entry rather + than blindly posting whichever entry it started waiting for. 3. **Post Comment:** Once an entry is due, the worker constructs and sends the API request to GitHub using the `octocrab` client and the data from the @@ -713,16 +712,16 @@ The worker task's loop consists of the following steps: **ETA projection.** The store computes, for each pending entry, an estimated number of seconds until it is posted (`schedule` in `QueueStore`). The head entry is due one full `cooldown_period_seconds` plus its own sampled flutter -after the last successful post, or immediately if nothing has been posted -yet. Each subsequent entry is due a further cooldown plus its own flutter -after its predecessor's projected posting time. Because flutter is sampled -once, at enqueue time, and the cooldown always runs in full, the ETA reported -to a client by `put` or `list` matches what the worker will actually do, and -does not drift as time passes. +after the last successful post, or immediately if nothing has been posted yet. +Each subsequent entry is due a further cooldown plus its own flutter after its +predecessor's projected posting time. Because flutter is sampled once, at +enqueue time, and the cooldown always runs in full, the ETA reported to a +client by `put` or `list` matches what the worker will actually do, and does +not drift as time passes. This workflow gives a highly resilient system that can tolerate both network -failures and process crashes without losing data: an entry is only ever -removed from the store once GitHub has confirmed the post. +failures and process crashes without losing data: an entry is only ever removed +from the store once GitHub has confirmed the post. ### 3.5. Daemon Configuration and Logging @@ -730,26 +729,26 @@ For operational flexibility and security, the daemon's behaviour must be controlled via a configuration file, not hard-coded values. A TOML file located at `/etc/comenqd/config.toml` is the conventional choice. -| Parameter | Type | Description | Default Value | -| ------------------------ | ------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------- | -| github_token | String | The GitHub Personal Access Token (PAT) used for authentication. Required unless `github_token_file` or `github_token_files` is set. | (none) | -| github_token_file | PathBuf | Optional path to a file containing the PAT. Read at startup; its trimmed contents override `github_token`. A leading `${VAR}` placeholder is expanded from the environment, enabling systemd `LoadCredential` integration. | (none) | +| Parameter | Type | Description | Default Value | +| ------------------------ | -------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------- | +| github_token | String | The GitHub Personal Access Token (PAT) used for authentication. Required unless `github_token_file` or `github_token_files` is set. | (none) | +| github_token_file | PathBuf | Optional path to a file containing the PAT. Read at startup; its trimmed contents override `github_token`. A leading `${VAR}` placeholder is expanded from the environment, enabling systemd `LoadCredential` integration. | (none) | | github_token_files | Vec\ | Optional array of paths, each containing one PAT. When non-empty, neither `github_token` nor `github_token_file` is required, and the daemon posts comments with these tokens in round-robin rotation (see §3.2). Each file is read and validated (must exist and hold a non-empty token) at startup. | (empty) | -| socket_path | PathBuf | The filesystem path for the Unix Domain Socket. | `$XDG_RUNTIME_DIR/comenq/comenq.sock` when a user runtime directory is available, else /run/comenq/comenq.sock | -| queue_path | PathBuf | The directory path for the persistent queue data (`entries/` and `last_post`, managed by `QueueStore`). | /var/lib/comenq/queue | -| log_level | String | The minimum log level to record (e.g., "info", "debug", "trace"). | info | -| cooldown_period_seconds | u64 | The cooling-off period in seconds after each comment post. | 960 | -| cooldown_flutter_seconds | u64 | Maximum random flutter in seconds added to each cooldown. The full cooldown always elapses; a fresh random duration up to this value is added on top. Zero disables flutter. | 0 | -| restart_min_delay_ms | u64 | The minimum delay (milliseconds) applied between supervised task restarts (backoff floor). | 100 | +| socket_path | PathBuf | The filesystem path for the Unix Domain Socket. | `$XDG_RUNTIME_DIR/comenq/comenq.sock` when a user runtime directory is available, else /run/comenq/comenq.sock | +| queue_path | PathBuf | The directory path for the persistent queue data (`entries/` and `last_post`, managed by `QueueStore`). | /var/lib/comenq/queue | +| log_level | String | The minimum log level to record (e.g., "info", "debug", "trace"). | info | +| cooldown_period_seconds | u64 | The cooling-off period in seconds after each comment post. | 960 | +| cooldown_flutter_seconds | u64 | Maximum random flutter in seconds added to each cooldown. The full cooldown always elapses; a fresh random duration up to this value is added on top. Zero disables flutter. | 0 | +| restart_min_delay_ms | u64 | The minimum delay (milliseconds) applied between supervised task restarts (backoff floor). | 100 | Configuration is loaded using the `ortho_config` crate. The daemon calls `Config::load()` which merges values from `/etc/comenqd/config.toml`, `COMENQD_*` environment variables, and any supplied CLI arguments. CLI arguments have the highest precedence, followed by environment variables, and finally the configuration file. Missing optional fields are replaced with -defaults, while a missing token (none of `github_token`, `github_token_file`, -or `github_token_files` set), an unreadable or empty token file, or invalid -TOML results in a configuration error. +defaults, while a missing token (none of `github_token`, `github_token_file`, or +`github_token_files` set), an unreadable or empty token file, or invalid TOML +results in a configuration error. Robust logging is non-negotiable for a background process. The `tracing` crate with `tracing-subscriber` will be used to provide structured, asynchronous