diff --git a/README.md b/README.md index fe1b8f2..7d063f7 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,13 @@ comenq bust 1a2b3c4d # move to the tail of the queue 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: + +```bash +comenq hist -n 20 # the twenty most recent posting attempts +``` + ## Running as a user service The daemon also runs unprivileged under `systemd --user`. Install the binaries diff --git a/crates/comenq/src/client.rs b/crates/comenq/src/client.rs index 60d3cc3..d16a2d6 100644 --- a/crates/comenq/src/client.rs +++ b/crates/comenq/src/client.rs @@ -14,7 +14,7 @@ use tokio::{ }; use tracing::warn; -use crate::output::{render_entry, render_put}; +use crate::output::{render_entry, render_history, render_put}; use crate::{Args, Command}; /// Errors that can occur when interacting with the daemon. @@ -102,9 +102,13 @@ async fn transact(candidates: &[PathBuf], request: &Request) -> Result Result<(), ClientError> { let request = args.command.to_request(); let response = transact(&args.socket_candidates(), &request).await?; - let (entry, entries) = match response { + let (entry, entries, history) = match response { Response::Error { message } => return Err(ClientError::Daemon(message)), - Response::Ok { entry, entries } => (entry, entries), + Response::Ok { + entry, + entries, + history, + } => (entry, entries, history), }; match &args.command { Command::Put { .. } => { @@ -124,6 +128,19 @@ pub async fn run(args: Args) -> Result<(), ClientError> { Command::Bump { id } => println!("Moved {id} to the head of the queue."), Command::Bust { id } => println!("Moved {id} to the tail of the queue."), Command::Del { id } => println!("Removed {id} from the queue."), + Command::Hist { .. } => { + let history = history.ok_or(ClientError::UnexpectedResponse)?; + if history.is_empty() { + println!("No posting history recorded."); + } else { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_secs()); + for record in &history { + println!("{}", render_history(record, now)); + } + } + } } Ok(()) } @@ -190,6 +207,32 @@ mod tests { assert!(!immediate, "put must default to deferred posting"); } + #[tokio::test] + async fn run_sends_hist_request_and_accepts_reply() { + let dir = tempdir().expect("temp dir"); + let socket = dir.path().join("sock"); + let listener = UnixListener::bind(&socket).expect("bind socket"); + let reply = Response::history(vec![comenq_lib::protocol::HistoryEntry { + id: "1a2b3c4d".into(), + posted_at: 1_000, + success: true, + error: None, + owner: "octocat".into(), + repo: "hello-world".into(), + pr_number: 1, + body: "Hi".into(), + }]); + let accept = spawn_daemon(listener, reply); + + let args = Args { + socket: Some(socket), + command: Command::Hist { limit: Some(2) }, + }; + run(args).await.expect("run succeeds"); + let request = accept.await.expect("join"); + assert_eq!(request, Request::Hist { limit: Some(2) }); + } + #[tokio::test] async fn run_surfaces_daemon_errors() { let dir = tempdir().expect("temp dir"); diff --git a/crates/comenq/src/lib.rs b/crates/comenq/src/lib.rs index d9fb079..4dfa0fe 100644 --- a/crates/comenq/src/lib.rs +++ b/crates/comenq/src/lib.rs @@ -8,7 +8,7 @@ mod client; mod output; pub use client::{ClientError, run}; -pub use output::{format_eta, one_line_summary}; +pub use output::{format_age, format_eta, one_line_summary}; /// A GitHub repository slug in `owner/repo` format. #[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)] @@ -155,6 +155,12 @@ pub enum Command { /// Identifier printed by `put` and `list`. id: String, }, + /// Show posted comments and failed attempts, oldest first. + Hist { + /// Show only the most recent LIMIT records. + #[arg(long, short = 'n', value_name = "LIMIT")] + limit: Option, + }, } impl Command { @@ -181,6 +187,7 @@ impl Command { Self::Bump { id } => Request::Bump { id: id.clone() }, Self::Bust { id } => Request::Bust { id: id.clone() }, Self::Del { id } => Request::Del { id: id.clone() }, + Self::Hist { limit } => Request::Hist { limit: *limit }, } } } @@ -283,6 +290,22 @@ mod tests { } } + #[rstest] + #[case::bare(&["comenq", "hist"], None)] + #[case::long(&["comenq", "hist", "--limit", "5"], Some(5))] + #[case::short(&["comenq", "hist", "-n", "5"], Some(5))] + fn parses_hist_subcommand(#[case] argv: &[&str], #[case] expected: Option) { + let args = Args::try_parse_from(argv).expect("valid arguments should parse"); + assert_eq!( + args.command.to_request(), + comenq_lib::protocol::Request::Hist { limit: expected } + ); + let Command::Hist { limit } = args.command else { + panic!("expected hist command"); + }; + assert_eq!(limit, expected); + } + #[test] fn missing_subcommand_is_rejected() { assert!(Args::try_parse_from(["comenq"]).is_err()); diff --git a/crates/comenq/src/output.rs b/crates/comenq/src/output.rs index a21be89..4b7f24c 100644 --- a/crates/comenq/src/output.rs +++ b/crates/comenq/src/output.rs @@ -1,9 +1,9 @@ //! Human-readable rendering of daemon replies. //! -//! Formats estimated posting times and one-line comment summaries for the -//! `put` and `list` subcommands. +//! Formats estimated posting times, past-attempt ages, and one-line comment +//! summaries for the `put`, `list`, and `hist` subcommands. -use comenq_lib::protocol::PendingEntry; +use comenq_lib::protocol::{HistoryEntry, PendingEntry}; /// Maximum characters of comment text shown by `list`. const SUMMARY_LIMIT: usize = 60; @@ -67,6 +67,47 @@ pub fn one_line_summary(body: &str) -> String { truncated } +/// Render how long ago something happened as a compact human duration. +/// +/// # Examples +/// +/// ```rust +/// assert_eq!(comenq::format_age(0), "just now"); +/// assert_eq!(comenq::format_age(45), "45s ago"); +/// assert_eq!(comenq::format_age(3_720), "1h 02m ago"); +/// ``` +#[must_use] +pub fn format_age(seconds: u64) -> String { + if seconds == 0 { + return "just now".to_owned(); + } + format!("{} ago", format_eta(seconds)) +} + +/// Render one `hist` line for a past posting attempt. +/// +/// `now` is the current Unix time, used to show the attempt's age. 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 mut line = format!( + "{} {:>11} {:<4} {}/{}#{} {}", + entry.id, + age, + status, + entry.owner, + entry.repo, + entry.pr_number, + one_line_summary(&entry.body) + ); + if let Some(error) = &entry.error { + line.push_str(" — "); + line.push_str(&one_line_summary(error)); + } + line +} + /// Render the `put` confirmation line. pub(crate) fn render_put(entry: &PendingEntry) -> String { format!( @@ -94,9 +135,11 @@ pub(crate) fn render_entry(entry: &PendingEntry) -> String { #[cfg(test)] mod tests { - //! Unit tests for ETA and summary rendering. - use super::{format_eta, one_line_summary, render_entry, render_put}; - use comenq_lib::protocol::PendingEntry; + //! Unit tests for ETA, age, and summary rendering. + use super::{ + format_age, format_eta, one_line_summary, render_entry, render_history, render_put, + }; + use comenq_lib::protocol::{HistoryEntry, PendingEntry}; use rstest::rstest; fn entry(body: &str, eta: u64) -> PendingEntry { @@ -154,6 +197,46 @@ mod tests { ); } + fn history(success: bool, error: Option<&str>) -> HistoryEntry { + HistoryEntry { + id: "1a2b3c4d".into(), + posted_at: 1_000, + success, + error: error.map(str::to_owned), + owner: "octocat".into(), + repo: "hello-world".into(), + pr_number: 7, + body: "Hi there".into(), + } + } + + #[rstest] + #[case(0, "just now")] + #[case(45, "45s ago")] + #[case(150, "2m 30s ago")] + #[case(3_720, "1h 02m ago")] + fn formats_age(#[case] seconds: u64, #[case] expected: &str) { + assert_eq!(format_age(seconds), expected); + } + + #[rstest] + fn renders_successful_history_line() { + let line = render_history(&history(true, None), 1_150); + assert_eq!( + line, + "1a2b3c4d 2m 30s ago ok octocat/hello-world#7 Hi there" + ); + } + + #[rstest] + fn renders_failed_history_line_with_error() { + 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" + ); + } + #[rstest] fn renders_list_line_with_truncated_body() { let body = "b".repeat(100); diff --git a/crates/comenqd/src/queue.rs b/crates/comenqd/src/queue.rs index 22764d5..434d97f 100644 --- a/crates/comenqd/src/queue.rs +++ b/crates/comenqd/src/queue.rs @@ -12,7 +12,7 @@ use comenq_lib::protocol::{Request, Response}; use tokio::sync::{Mutex, Notify}; use crate::config::Config; -use crate::store::{PutOptions, QueueStore, Result as StoreResult, StoredEntry}; +use crate::store::{HistoryRecord, PutOptions, QueueStore, Result as StoreResult, StoredEntry}; /// Current Unix time in whole seconds. /// @@ -62,9 +62,24 @@ impl SharedQueue { .next_due(self.cfg.cooldown_period_seconds, unix_now()) } - /// Remove the posted entry and record the posting time. - pub async fn complete(&self, id: &str) -> StoreResult<()> { - self.store.lock().await.complete(id, unix_now()) + /// Remove the posted entry, recording the posting time and a history + /// record of the success. + pub async fn complete(&self, entry: &StoredEntry) -> StoreResult<()> { + let store = self.store.lock().await; + let now = unix_now(); + store.complete(&entry.id, now)?; + store.append_history(&HistoryRecord::success(entry, now)) + } + + /// Record a failed posting attempt in the history log. + /// + /// The entry itself stays queued; the worker retries it after a + /// cooldown. + pub async fn record_failure(&self, entry: &StoredEntry, error: &str) -> StoreResult<()> { + self.store + .lock() + .await + .append_history(&HistoryRecord::failure(entry, unix_now(), error)) } /// Execute a protocol request and produce the reply. @@ -106,6 +121,19 @@ impl SharedQueue { Request::Bump { id } => (store.bump(&id).map(|()| Response::ok()), true), Request::Bust { id } => (store.bust(&id).map(|()| Response::ok()), true), Request::Del { id } => (store.del(&id).map(|()| Response::ok()), true), + Request::Hist { limit } => { + let outcome = store.history().map(|records| { + let skip = limit.map_or(0, |n| records.len().saturating_sub(n)); + Response::history( + records + .into_iter() + .skip(skip) + .map(|record| record.to_entry()) + .collect(), + ) + }); + (outcome, false) + } }; drop(store); match response { diff --git a/crates/comenqd/src/store.rs b/crates/comenqd/src/store.rs index 5362b15..50a27be 100644 --- a/crates/comenqd/src/store.rs +++ b/crates/comenqd/src/store.rs @@ -5,6 +5,8 @@ //! be moved to the head (`bump`) or tail (`bust`) of the queue, or removed //! (`del`). The Unix timestamp of the most recent successful post is kept in //! `/last_post` so estimated posting times survive restarts. +//! Every posting attempt, successful or not, is appended as one JSON line to +//! `/history.jsonl` so past activity can be reported. //! //! The store itself holds no in-memory state; callers serialize mutations //! (the daemon wraps it in a `tokio::sync::Mutex`). Files are written to a @@ -23,6 +25,11 @@ use std::path::{Path, PathBuf}; const ENTRIES_DIR: &str = "entries"; /// File recording the Unix time of the most recent successful post. const LAST_POST_FILE: &str = "last_post"; +/// Append-only JSON Lines log of posting attempts. +const HISTORY_FILE: &str = "history.jsonl"; + +mod history; +pub use history::HistoryRecord; /// A queued comment with its scheduling metadata. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] @@ -97,6 +104,7 @@ pub struct PutOptions { pub struct QueueStore { entries_dir: PathBuf, last_post_path: PathBuf, + history_path: PathBuf, } /// Compute the deterministic eight-character identifier for an entry. @@ -134,6 +142,7 @@ impl QueueStore { Ok(Self { entries_dir, last_post_path: queue_path.join(LAST_POST_FILE), + history_path: queue_path.join(HISTORY_FILE), }) } @@ -256,6 +265,16 @@ impl QueueStore { Ok(()) } + /// Append `record` to the posting history log. + pub fn append_history(&self, record: &HistoryRecord) -> Result<()> { + history::append(&self.history_path, record) + } + + /// All recorded posting attempts in chronological order. + pub fn history(&self) -> Result> { + history::read(&self.history_path) + } + /// Pending entries paired with their estimated seconds-until-post. /// /// The head is due one full cooldown plus its own flutter after the most diff --git a/crates/comenqd/src/store/history.rs b/crates/comenqd/src/store/history.rs new file mode 100644 index 0000000..f28c4de --- /dev/null +++ b/crates/comenqd/src/store/history.rs @@ -0,0 +1,112 @@ +//! Posting-history records and their append-only JSON Lines log. +//! +//! Every posting attempt, successful or not, becomes one [`HistoryRecord`] +//! appended as a single JSON line to `/history.jsonl`. + +use comenq_lib::CommentRequest; +use comenq_lib::protocol::HistoryEntry; +use serde::{Deserialize, Serialize}; +use std::fs; +use std::io::{self, Write as _}; +use std::path::Path; + +use super::{Result, StoredEntry}; + +/// One posting attempt, as recorded in the history log. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct HistoryRecord { + /// Identifier the entry carried while queued. + pub id: String, + /// Unix time the posting attempt finished, in seconds. + pub posted_at: u64, + /// Whether GitHub accepted the comment. + pub success: bool, + /// Failure description when `success` is false. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + /// The comment that was posted (or attempted). + pub request: CommentRequest, +} + +impl HistoryRecord { + /// Record a successful post of `entry` at `posted_at`. + #[must_use] + pub fn success(entry: &StoredEntry, posted_at: u64) -> Self { + Self { + id: entry.id.clone(), + posted_at, + success: true, + error: None, + request: entry.request.clone(), + } + } + + /// Record a failed posting attempt of `entry` at `posted_at`. + #[must_use] + pub fn failure(entry: &StoredEntry, posted_at: u64, error: impl Into) -> Self { + Self { + id: entry.id.clone(), + posted_at, + success: false, + error: Some(error.into()), + request: entry.request.clone(), + } + } + + /// Convert to the wire representation. + #[must_use] + pub fn to_entry(&self) -> HistoryEntry { + HistoryEntry { + id: self.id.clone(), + posted_at: self.posted_at, + success: self.success, + error: self.error.clone(), + owner: self.request.owner.clone(), + repo: self.request.repo.clone(), + pr_number: self.request.pr_number, + body: self.request.body.clone(), + } + } +} + +/// Append `record` as one line to the log at `path`. +/// +/// The log is JSON Lines: one record per line, appended atomically enough +/// for a single writer (the daemon serializes store access). +pub(super) fn append(path: &Path, record: &HistoryRecord) -> Result<()> { + let mut line = serde_json::to_vec(record)?; + line.push(b'\n'); + let mut file = fs::OpenOptions::new() + .create(true) + .append(true) + .open(path)?; + file.write_all(&line)?; + Ok(()) +} + +/// All recorded posting attempts in chronological order. +/// +/// A missing log means no attempts have been recorded. Malformed lines are +/// skipped with an error log so one corrupt record cannot hide the rest of +/// the history. +pub(super) fn read(path: &Path) -> Result> { + let text = match fs::read_to_string(path) { + Ok(text) => text, + Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(e) => return Err(e.into()), + }; + let mut records = Vec::new(); + for line in text.lines().filter(|line| !line.trim().is_empty()) { + match serde_json::from_str::(line) { + Ok(record) => records.push(record), + Err(e) => { + tracing::error!( + path = %path.display(), + error = %e, + "Skipping unreadable history record" + ); + } + } + } + Ok(records) +} diff --git a/crates/comenqd/src/store/tests.rs b/crates/comenqd/src/store/tests.rs index c6ac13f..40d8e93 100644 --- a/crates/comenqd/src/store/tests.rs +++ b/crates/comenqd/src/store/tests.rs @@ -1,6 +1,6 @@ //! Tests for the reorderable persistent queue store. -use super::{PutOptions, QueueStore, StoreError, entry_id}; +use super::{HistoryRecord, PutOptions, QueueStore, StoreError, entry_id}; use comenq_lib::CommentRequest; use rstest::rstest; use tempfile::TempDir; @@ -280,3 +280,53 @@ fn deferred_floor_includes_the_entry_flutter() { "floor must be enqueue + cooldown + sampled flutter" ); } + +#[rstest] +fn history_starts_empty() { + let dir = TempDir::new().expect("tempdir"); + let store = open_store(&dir); + assert!(store.history().expect("read history").is_empty()); +} + +#[rstest] +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"); + store.append_history(&success).expect("append success"); + store.append_history(&failure).expect("append failure"); + let records = store.history().expect("read history"); + assert_eq!(records, vec![success, failure.clone()]); + let wire = failure.to_entry(); + assert_eq!(wire.id, entry.id); + assert_eq!(wire.posted_at, 2200); + assert!(!wire.success); + assert_eq!(wire.error.as_deref(), Some("timeout")); + assert_eq!(wire.owner, "octocat"); + assert_eq!(wire.repo, "hello-world"); + assert_eq!(wire.pr_number, 7); + assert_eq!(wire.body, "a"); +} + +#[rstest] +fn history_skips_malformed_lines() { + let dir = TempDir::new().expect("tempdir"); + let store = open_store(&dir); + let entry = store.put(request("a"), &immediate(0), 1000).expect("put"); + store + .append_history(&HistoryRecord::success(&entry, 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")) + .expect("append failure"); + let records = store.history().expect("read history"); + assert_eq!(records.len(), 2, "the corrupt line must not hide records"); + assert!(records[0].success); + assert!(!records[1].success); +} diff --git a/crates/comenqd/src/worker.rs b/crates/comenqd/src/worker.rs index 36081ca..9d464d4 100644 --- a/crates/comenqd/src/worker.rs +++ b/crates/comenqd/src/worker.rs @@ -190,7 +190,7 @@ pub async fn run_worker( hooks.notify_enqueued(); match post_comment(&octocrab, &entry.request, &config).await { Ok(()) => { - queue.complete(&entry.id).await?; + queue.complete(&entry).await?; } Err(e) => { tracing::error!( @@ -201,6 +201,14 @@ pub async fn run_worker( 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 { + tracing::error!( + error = %log_error, + id = %entry.id, + "Failed to record the posting failure in the history log", + ); + } hooks.notify_idle(); // Pace retries so a persistently failing API is not hammered. if WorkerHooks::wait_or_shutdown(config.cooldown_period_seconds, shutdown).await { diff --git a/docs/comenq-design.md b/docs/comenq-design.md index 506f4ea..7fcd57a 100644 --- a/docs/comenq-design.md +++ b/docs/comenq-design.md @@ -360,7 +360,7 @@ 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 five subcommands, one per protocol +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): @@ -382,6 +382,16 @@ operation, all sharing a global `--socket` flag (also settable via the - `comenq del `: removes the identified comment from the queue. +- `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. + 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 @@ -506,7 +516,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. 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 @@ -545,6 +557,23 @@ The queue is initialized at a configurable path (e.g., `/var/lib/comenq/queue`) 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), and the `request` that was posted or attempted. 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. + ### 3.3. The UDS Listener and Request Ingestion (`run_listener`) This task is responsible for handling all client communication. It will be diff --git a/packaging/man/comenq.1 b/packaging/man/comenq.1 index f9e27b0..f154265 100644 --- a/packaging/man/comenq.1 +++ b/packaging/man/comenq.1 @@ -20,6 +20,12 @@ comenq \- enqueue a GitHub pull request comment for the comenqd daemon .br .B comenq del .I id +.br +.B comenq hist +[ +.B -n +.I limit +] .SH DESCRIPTION The .B comenq @@ -51,6 +57,24 @@ Move the identified comment to the tail of the queue. .TP .B del Remove the identified comment from the queue. +.TP +.B hist +Print the posting-history log, oldest first: identifier, age, status ( +.I ok +or +.IR FAIL ), +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 +has been recorded, prints a message saying so. +.RS +.TP +.BR \-n ", " \-\-limit " " \fIlimit\fP +Show only the most recent +.I limit +records, still printed oldest first. By default the whole log is printed. +.RE .PP Use .B --socket diff --git a/src/protocol.rs b/src/protocol.rs index a116fe8..f40f280 100644 --- a/src/protocol.rs +++ b/src/protocol.rs @@ -38,6 +38,12 @@ pub enum Request { /// Identifier printed by `list` and `put`. id: String, }, + /// Report past posting attempts in chronological order. + Hist { + /// Most recent records to return; `None` returns them all. + #[serde(default)] + limit: Option, + }, } /// A pending queue entry as reported by the daemon. @@ -57,6 +63,28 @@ pub struct PendingEntry { pub body: String, } +/// A past posting attempt as reported by the daemon. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct HistoryEntry { + /// Identifier the entry carried while queued. + pub id: String, + /// Unix time the posting attempt finished, in seconds. + pub posted_at: u64, + /// Whether GitHub accepted the comment. + pub success: bool, + /// Failure description when `success` is false. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + /// Repository owner. + pub owner: String, + /// Repository name. + pub repo: String, + /// Pull request number. + pub pr_number: u64, + /// Full comment body; consumers truncate for display. + pub body: String, +} + /// Daemon reply to a [`Request`]. #[derive(Debug, Serialize, Deserialize, PartialEq, Eq)] #[serde(tag = "outcome", rename_all = "snake_case")] @@ -69,6 +97,9 @@ pub enum Response { /// Pending entries, returned by `list`. #[serde(default, skip_serializing_if = "Option::is_none")] entries: Option>, + /// Past posting attempts, returned by `hist`. + #[serde(default, skip_serializing_if = "Option::is_none")] + history: Option>, }, /// The request failed; `message` explains why. Error { @@ -84,6 +115,7 @@ impl Response { Self::Ok { entry: None, entries: None, + history: None, } } @@ -93,6 +125,7 @@ impl Response { Self::Ok { entry: Some(entry), entries: None, + history: None, } } @@ -102,6 +135,17 @@ impl Response { Self::Ok { entry: None, entries: Some(entries), + history: None, + } + } + + /// Successful reply for `hist`. + #[must_use] + pub fn history(history: Vec) -> Self { + Self::Ok { + entry: None, + entries: None, + history: Some(history), } } @@ -117,7 +161,7 @@ impl Response { #[cfg(test)] mod tests { //! Serialization round-trip tests for the client-daemon protocol. - use super::{PendingEntry, Request, Response}; + use super::{HistoryEntry, PendingEntry, Request, Response}; use crate::CommentRequest; fn sample_entry() -> PendingEntry { @@ -131,6 +175,19 @@ mod tests { } } + fn sample_history() -> HistoryEntry { + HistoryEntry { + id: "0011aabb".into(), + posted_at: 1_700_000_000, + success: false, + error: Some("timeout".into()), + owner: "octocat".into(), + repo: "hello-world".into(), + pr_number: 7, + body: "Hi".into(), + } + } + #[test] fn put_round_trips_through_json() { let req = Request::Put { @@ -162,6 +219,8 @@ mod tests { Request::Del { id: "0011aabb".into(), }, + Request::Hist { limit: None }, + Request::Hist { limit: Some(20) }, ] { let json = serde_json::to_string(&req).unwrap_or_else(|e| panic!("serialize: {e}")); let back: Request = @@ -176,6 +235,7 @@ mod tests { Response::ok(), Response::entry(sample_entry()), Response::entries(vec![sample_entry()]), + Response::history(vec![sample_history()]), Response::error("nope"), ] { let json = serde_json::to_string(&resp).unwrap_or_else(|e| panic!("serialize: {e}")); @@ -201,6 +261,13 @@ mod tests { )); } + #[test] + fn hist_defaults_to_no_limit() { + let req: Request = + serde_json::from_str(r#"{"op":"hist"}"#).unwrap_or_else(|e| panic!("parse: {e}")); + assert_eq!(req, Request::Hist { limit: None }); + } + #[test] fn unknown_operation_fails_to_parse() { let result: Result = serde_json::from_str(r#"{"op":"zap"}"#); diff --git a/tests/cucumber.rs b/tests/cucumber.rs index 265554f..636cd66 100644 --- a/tests/cucumber.rs +++ b/tests/cucumber.rs @@ -6,8 +6,8 @@ mod steps; use cucumber::World as _; use steps::{ - CliWorld, ClientWorld, CommentWorld, ConfigWorld, ListenerWorld, PackagingWorld, QueueWorld, - ReleaseWorld, WorkerWorld, + CliWorld, ClientWorld, CommentWorld, ConfigWorld, HistoryWorld, ListenerWorld, PackagingWorld, + QueueWorld, ReleaseWorld, WorkerWorld, }; fn main() -> anyhow::Result<()> { @@ -26,6 +26,7 @@ fn main() -> anyhow::Result<()> { ClientWorld::run("tests/features/client_main.feature"), CommentWorld::run("tests/features/comment_request.feature"), ConfigWorld::run("tests/features/config.feature"), + HistoryWorld::run("tests/features/history.feature"), ListenerWorld::run("tests/features/listener.feature"), PackagingWorld::run("tests/features/packaging.feature"), QueueWorld::run("tests/features/queue.feature"), diff --git a/tests/features/history.feature b/tests/features/history.feature new file mode 100644 index 0000000..e1d10c6 --- /dev/null +++ b/tests/features/history.feature @@ -0,0 +1,22 @@ +Feature: Posting history + + Scenario: an empty history lists nothing + Given an empty posting queue + When the posting history is requested + Then the history lists nothing + + Scenario: posting attempts are recorded in order + Given an empty posting queue + And the comments "First" and "Second" are queued for posting + When the head comment is posted successfully + And the head comment fails to post with "boom" + And the posting history is requested + Then the history lists 2 records + And the history records a success then a failure with "boom" + + Scenario: the history honours a limit + Given an empty posting queue + And the comments "First", "Second" and "Third" are queued for posting + When each queued comment is posted successfully + And the posting history is requested with a limit of 2 + Then the history lists the 2 most recent records diff --git a/tests/features/worker.feature b/tests/features/worker.feature index b27b488..17134d0 100644 --- a/tests/features/worker.feature +++ b/tests/features/worker.feature @@ -5,9 +5,11 @@ Feature: Worker task And GitHub returns success When the worker runs briefly Then the comment is posted + And the posting history records the success Scenario: API failure requeues job Given a queued comment request And GitHub returns an error When the worker runs briefly Then the queue retains the job + And the posting history records the failure diff --git a/tests/steps/history_steps.rs b/tests/steps/history_steps.rs new file mode 100644 index 0000000..577d34b --- /dev/null +++ b/tests/steps/history_steps.rs @@ -0,0 +1,214 @@ +//! Behavioural test steps for the posting history. +//! +//! Exercises the daemon's `hist` protocol operation against a real +//! `SharedQueue`, simulating posting outcomes through the queue's +//! completion and failure-recording API. + +use std::sync::Arc; + +use anyhow::Context as _; +use comenq_lib::CommentRequest; +use comenq_lib::protocol::{HistoryEntry, Request, Response}; +use comenqd::config::Config; +use comenqd::daemon::SharedQueue; +use comenqd::store::StoredEntry; +use cucumber::{World, given, then, when}; +use tempfile::TempDir; +use test_support::temp_config; + +#[derive(Default, World)] +pub struct HistoryWorld { + dir: Option, + queue: Option>, + /// Identifiers in the order their posting attempts were recorded. + recorded: Vec, + last_history: Option>, +} + +impl std::fmt::Debug for HistoryWorld { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("HistoryWorld").finish() + } +} + +impl HistoryWorld { + fn queue(&self) -> anyhow::Result<&Arc> { + self.queue.as_ref().context("queue not initialized") + } + + async fn put(&mut self, body: &str) -> anyhow::Result<()> { + let response = self + .queue()? + .execute(Request::Put { + request: CommentRequest { + owner: "octocat".into(), + repo: "hello-world".into(), + pr_number: 7, + body: body.into(), + }, + immediate: true, + }) + .await; + anyhow::ensure!( + matches!(response, Response::Ok { .. }), + "put should succeed, got {response:?}" + ); + Ok(()) + } + + async fn head(&self) -> anyhow::Result { + let due = self.queue()?.next_due().await.context("next due")?; + let (entry, _) = due.context("queue should hold a head entry")?; + Ok(entry) + } + + async fn request_history(&mut self, limit: Option) -> anyhow::Result<()> { + let response = self.queue()?.execute(Request::Hist { limit }).await; + match response { + Response::Ok { + history: Some(history), + .. + } => { + self.last_history = Some(history); + Ok(()) + } + other => anyhow::bail!("expected hist reply, got {other:?}"), + } + } + + fn history(&self) -> anyhow::Result<&[HistoryEntry]> { + self.last_history + .as_deref() + .context("no hist reply recorded") + } +} + +#[given("an empty posting queue")] +fn empty_queue(world: &mut HistoryWorld) -> anyhow::Result<()> { + let dir = TempDir::new().context("tempdir")?; + let cfg = Arc::new(Config::from(temp_config(&dir).with_cooldown(600))); + world.queue = Some(SharedQueue::open(cfg).context("open queue")?); + world.dir = Some(dir); + Ok(()) +} + +#[given(regex = r#"^the comments \"([^\"]+)\" and \"([^\"]+)\" are queued for posting$"#)] +async fn two_comments_queued( + world: &mut HistoryWorld, + first: String, + second: String, +) -> anyhow::Result<()> { + for body in [first, second] { + world.put(&body).await?; + } + Ok(()) +} + +#[given( + regex = r#"^the comments \"([^\"]+)\", \"([^\"]+)\" and \"([^\"]+)\" are queued for posting$"# +)] +async fn three_comments_queued( + world: &mut HistoryWorld, + first: String, + second: String, + third: String, +) -> anyhow::Result<()> { + for body in [first, second, third] { + world.put(&body).await?; + } + Ok(()) +} + +#[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.recorded.push(entry.id); + Ok(()) +} + +#[when("each queued comment is posted successfully")] +async fn all_posted(world: &mut HistoryWorld) -> anyhow::Result<()> { + while world + .queue()? + .next_due() + .await + .context("next due")? + .is_some() + { + head_posted(world).await?; + } + Ok(()) +} + +#[when(regex = r#"^the head comment fails to post with \"([^\"]+)\"$"#)] +async fn head_fails(world: &mut HistoryWorld, error: String) -> anyhow::Result<()> { + let entry = world.head().await?; + world + .queue()? + .record_failure(&entry, &error) + .await + .context("record failure")?; + world.recorded.push(entry.id); + Ok(()) +} + +#[when("the posting history is requested")] +async fn history_requested(world: &mut HistoryWorld) -> anyhow::Result<()> { + world.request_history(None).await +} + +#[when(regex = r"^the posting history is requested with a limit of ([0-9]+)$")] +async fn history_requested_with_limit( + world: &mut HistoryWorld, + limit: usize, +) -> anyhow::Result<()> { + world.request_history(Some(limit)).await +} + +#[then("the history lists nothing")] +fn history_is_empty(world: &mut HistoryWorld) -> anyhow::Result<()> { + anyhow::ensure!( + world.history()?.is_empty(), + "expected an empty history, got {:?}", + world.history()? + ); + Ok(()) +} + +#[then(regex = r"^the history lists ([0-9]+) records$")] +fn history_has_len(world: &mut HistoryWorld, expected: usize) -> anyhow::Result<()> { + assert_eq!(world.history()?.len(), expected); + Ok(()) +} + +#[then(regex = r#"^the history records a success then a failure with \"([^\"]+)\"$"#)] +fn history_outcomes(world: &mut HistoryWorld, error: String) -> anyhow::Result<()> { + let history = world.history()?; + let success = history.first().context("history should hold a record")?; + assert!(success.success, "the first record should be a success"); + assert_eq!(success.error, None); + let failure = history.get(1).context("history should hold two records")?; + assert!(!failure.success, "the second record should be a failure"); + assert_eq!(failure.error, Some(error)); + for (record, id) in history.iter().zip(&world.recorded) { + assert_eq!(&record.id, id, "records should appear in posting order"); + } + Ok(()) +} + +#[then(regex = r"^the history lists the ([0-9]+) most recent records$")] +fn history_is_most_recent(world: &mut HistoryWorld, expected: usize) -> anyhow::Result<()> { + let history = world.history()?; + assert_eq!(history.len(), expected); + let listed: Vec<&str> = history.iter().map(|record| record.id.as_str()).collect(); + let skip = world.recorded.len().saturating_sub(expected); + let recent: Vec<&str> = world + .recorded + .iter() + .skip(skip) + .map(String::as_str) + .collect(); + assert_eq!(listed, recent, "the oldest records should be dropped"); + Ok(()) +} diff --git a/tests/steps/mod.rs b/tests/steps/mod.rs index 2d5ab05..1941ae9 100644 --- a/tests/steps/mod.rs +++ b/tests/steps/mod.rs @@ -7,6 +7,8 @@ pub mod comment_steps; pub use comment_steps::CommentWorld; pub mod config_steps; pub use config_steps::ConfigWorld; +pub mod history_steps; +pub use history_steps::HistoryWorld; pub mod listener_steps; pub use listener_steps::ListenerWorld; pub mod packaging_steps; diff --git a/tests/steps/worker_steps.rs b/tests/steps/worker_steps.rs index 636cf3e..3ce808e 100644 --- a/tests/steps/worker_steps.rs +++ b/tests/steps/worker_steps.rs @@ -11,7 +11,7 @@ use std::time::Duration; use anyhow::Context as _; use comenq_lib::CommentRequest; -use comenq_lib::protocol::{PendingEntry, Request, Response}; +use comenq_lib::protocol::{HistoryEntry, PendingEntry, Request, Response}; use comenqd::config::Config; use comenqd::daemon::{SharedQueue, WorkerControl, WorkerHooks, run_worker}; use cucumber::{World, given, then, when}; @@ -80,6 +80,17 @@ async fn listed_entries(queue: &Arc) -> anyhow::Result) -> anyhow::Result> { + match queue.execute(Request::Hist { limit: None }).await { + Response::Ok { + history: Some(history), + .. + } => Ok(history), + other => anyhow::bail!("expected hist reply, got {other:?}"), + } +} + #[given("a queued comment request")] async fn queued_request(world: &mut WorkerWorld) -> anyhow::Result<()> { let dir = TempDir::new().context("tempdir")?; @@ -197,6 +208,42 @@ async fn comment_posted(world: &mut WorkerWorld) -> anyhow::Result<()> { Ok(()) } +#[then("the posting history records the success")] +async fn history_records_success(world: &mut WorkerWorld) -> anyhow::Result<()> { + let queue = world.queue.as_ref().context("queue initialized")?.clone(); + let history = listed_history(&queue).await?; + assert_eq!(history.len(), 1, "one post should yield one record"); + let record = history.first().context("history should hold a record")?; + assert!(record.success, "the record should mark a success"); + assert_eq!(record.error, None); + assert_eq!(record.owner, "o"); + assert_eq!(record.repo, "r"); + assert_eq!(record.pr_number, 1); + Ok(()) +} + +#[then("the posting history records the failure")] +async fn history_records_failure(world: &mut WorkerWorld) -> anyhow::Result<()> { + let queue = world.queue.as_ref().context("queue initialized")?.clone(); + let history = listed_history(&queue).await?; + anyhow::ensure!( + !history.is_empty(), + "a failed attempt should be recorded in the history" + ); + for record in &history { + assert!(!record.success, "every record should mark a failure"); + let error = record + .error + .as_deref() + .context("failure carries an error")?; + anyhow::ensure!( + !error.is_empty(), + "the error description should not be empty" + ); + } + Ok(()) +} + #[then("the queue retains the job")] async fn queue_retains(world: &mut WorkerWorld) -> anyhow::Result<()> { let queue = world.queue.as_ref().context("queue initialized")?.clone();