diff --git a/AGENTS.md b/AGENTS.md index b126273..a278b06 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -74,7 +74,7 @@ Docker: `docker build -t tgxmb .` then `docker run --rm -d --name tgxmb --env-fi - **Serde**: per-site `model.rs` are pure `Deserialize` DTOs mirroring API JSON; site structs in `interface.rs` have private fields, a `caption()` builder, and `impl From for Fetched`. Persisted payloads use internally-tagged enums (`#[serde(tag = "kind")]` / `type`). - **Naming**: module-per-concern, snake_case files, `CamelCase` types, `snake_case` fns. `//!` module docs and `///` docs on non-obvious logic (syndication token, ugoira encoding, `display_text_range`). - **Retries**: only `x-media::site::fetch` retries (3 attempts, `1 << attempt` backoff, HTTP errors only); `site::fetch_once` is the same code path with a single attempt, used by inline queries whose answer window is shorter than the backoff. Queue retries are explicit `QueueError::Retryable` with computed delay (`retry_delay_seconds`). -- Logging via `log` macros (`pretty_env_logger`, level from `RUST_LOG`). Level convention: `info` = lifecycle + per-post business results (`sent`/`forwarded`/`copied`), admin/operator actions and anomalies (fallback, retry enqueue, dead-letter is `error`); `debug` = per-request detail (message/command/URL extraction, `fetching`/`fetched`, batch sends, queue processing, photo processing, inline queries). Full user-submitted URLs and message text only appear at `debug`; at `info` and above links are printed via the normalized cache key (`handlers::log_key`, e.g. `[key=twitter:123...]`) so logs stay short and do not echo user data. +- Logging via `log` macros (`pretty_env_logger`, level from `RUST_LOG`). `main.rs` initializes the **timed** builder with a default filter of `info,hyper_util=warn,reqwest=warn` when `RUST_LOG` is unset: the plain `init` had no timestamps and fell back to `error`, so a deployment that forgot the variable logged nothing at all, and at `debug` the HTTP client's own lines outnumbered the bot's two to one. An explicit `RUST_LOG` overrides the default wholesale. Level convention: `info` = lifecycle + per-post business results (`sent`/`forwarded`/`copied`, with `chat=` and the total `ms`), admin/operator actions and anomalies (fallback, retry enqueue, dead-letter is `error`); `debug` = per-request detail (URL extraction, `fetching`/`fetched` with the fetch duration, batch sends, queue processing with the row's `chat=`/`key=` and per-attempt `ms`, photo processing, inline queries); `trace` = user data (the full URL, the message text, the inline query). At `debug` and above links are printed via the normalized cache key (`handlers::log_key`, e.g. `[key=twitter:123...]`), so a `debug` log can be shared without echoing what users pasted, and degradations that leave the user served (a failed cache read/write, a failed chat action) are `warn`, not `error`. The only queue/sweep aggregate is the 300 s sweep's queue line, and it speaks only when the queue is non-empty. ## Important Files diff --git a/README.en.md b/README.en.md index 3dd4dc8..9e57864 100644 --- a/README.en.md +++ b/README.en.md @@ -93,7 +93,7 @@ Telegram only accepts ports 443/80/88/8443. | `LINK_CACHE_TTL_SECONDS` | Link-result cache expiry in seconds, default 604800 (7 days) | | `CAPTION_QUOTE_TEXT_CHARS` | **The text part** of the caption (the joined `{title}` + `{content}`) is wrapped in a collapsible blockquote once it reaches this many characters, default 200; `0` disables | | `DATA_DIR` | Data directory (where the SQLite `task_queue.db` lives), default `data` (relative to the working directory, created automatically) | -| `RUST_LOG` | Log level | +| `RUST_LOG` | Log level, default `info,hyper_util=warn,reqwest=warn` (an unset variable no longer silences the log). Recipes: `info,xmedia_bot=debug,x_media=debug` (app detail, no dependency noise) / `debug,hyper_util=off` (everything) / `trace` (also prints full links and message text — **user data**) | | `TELOXIDE_PROXY` | HTTP proxy (e.g. `http://127.0.0.1:10808`); applies to both the Telegram Bot API and site fetches — required on restricted networks (e.g. behind the GFW) | | `LOCAL_USER_ID` | UID the container runs as, default 9001 | | `VIRTUAL_HOST` | Public domain or IP; nginx-proxy routes by this | diff --git a/README.md b/README.md index 9add957..eb5af36 100644 --- a/README.md +++ b/README.md @@ -93,7 +93,7 @@ Telegram 只接受 443/80/88/8443 端口。 | `LINK_CACHE_TTL_SECONDS` | 链接结果缓存过期秒数，默认 604800（7 天） | | `CAPTION_QUOTE_TEXT_CHARS` | 正文（`{title}` + `{content}` 合计）达到该长度（字符）时，caption 的**正文部分**用可折叠引用块包裹，默认 200；`0` 关闭 | | `DATA_DIR` | 数据目录（SQLite 数据库 `task_queue.db` 所在目录），默认 `data`（相对工作目录，会自动创建） | -| `RUST_LOG` | 日志级别 | +| `RUST_LOG` | 日志级别，默认 `info,hyper_util=warn,reqwest=warn`（未设置也**不会**哑掉）。排障配方：`info,xmedia_bot=debug,x_media=debug`（应用细节，无依赖噪音）/ `debug,hyper_util=off`（全量）/ `trace`（额外打印完整链接与消息原文，**含用户数据**） | | `TELOXIDE_PROXY` | HTTP 代理（如 `http://127.0.0.1:10808`）；同时作用于 Telegram Bot API 与站点抓取请求，网络受限环境（如 GFW）必需 | | `LOCAL_USER_ID` | 容器内运行用户 UID，默认 9001 | | `VIRTUAL_HOST` | 对外域名或 IP，nginx-proxy 按此路由 | diff --git a/crates/x-media/src/site/bsky/interface.rs b/crates/x-media/src/site/bsky/interface.rs index aa668d8..08ff764 100644 --- a/crates/x-media/src/site/bsky/interface.rs +++ b/crates/x-media/src/site/bsky/interface.rs @@ -51,6 +51,10 @@ pub async fn fetch_from_url(url: &str) -> Result { // encode path — the temp file stays alive via `_keep_alive`). On any // failure the video item is dropped and the post degrades to its text. let mut media = Vec::with_capacity(fetched.media.len()); + // The remux warnings below name the post, not the CDN URL they were + // working on: the media URL is derived from what the user pasted, and + // `warn` is a level operators share. + let key = cache_key(url).unwrap_or_else(|| "?".into()); for item in fetched.media { let is_hls = matches!(&item, Media::Video { url, .. } if url.contains("playlist") || url.ends_with(".m3u8")); @@ -72,8 +76,8 @@ pub async fn fetch_from_url(url: &str) -> Result { }); fetched._keep_alive = Some(keep_alive); } - Ok(None) => log::warn!("bsky video remux unavailable for {url}"), - Err(e) => log::warn!("bsky video remux failed for {url}: {e}"), + Ok(None) => log::warn!("bsky video remux unavailable for [key={key}]"), + Err(e) => log::warn!("bsky video remux failed for [key={key}]: {e}"), } } fetched.media = media; diff --git a/crates/x-media/src/site/mod.rs b/crates/x-media/src/site/mod.rs index bb9d745..4ab39f4 100644 --- a/crates/x-media/src/site/mod.rs +++ b/crates/x-media/src/site/mod.rs @@ -450,6 +450,9 @@ pub async fn fetch_once(url: &str) -> Result, FetchError> { const MAX_FETCH_ATTEMPTS: u32 = 3; async fn fetch_with_attempts(url: &str, attempts: u32) -> Result, FetchError> { + // Wall time of the whole fetch, retry backoff included: the ugoira encode + // and the HLS remux live inside it, so this is where a slow fetch shows. + let started = std::time::Instant::now(); let Some(site) = find_site(url) else { // A registered-but-disabled site (pixiv without a token) is not an // unsupported link: report it, so the bot answers the user instead of @@ -464,10 +467,11 @@ async fn fetch_with_attempts(url: &str, attempts: u32) -> Result Ok(fetched) => { // Per-request detail: debug only, keyed by the post id. log::debug!( - "fetched [key={}]: site {} returned {} media", + "fetched [key={}]: site {} returned {} media in {}ms", cache_key(url).unwrap_or_else(|| "?".into()), fetched.site_name(), - fetched.media.len() + fetched.media.len(), + started.elapsed().as_millis() ); return Ok(Some(fetched)); } diff --git a/crates/xmedia-bot/src/handlers/inline.rs b/crates/xmedia-bot/src/handlers/inline.rs index 7245fb6..268d37f 100644 --- a/crates/xmedia-bot/src/handlers/inline.rs +++ b/crates/xmedia-bot/src/handlers/inline.rs @@ -116,11 +116,10 @@ pub async fn inline_query_handler(bot: Bot, query: InlineQuery) -> Result<(), Re /// Fetches the post behind an inline query and answers it. The caller has /// already applied the debounce. Returns `true` when an answer was sent. async fn answer_inline_query(bot: Bot, query: InlineQuery) -> Result { - log::debug!( - "inline query: {} [key={}]", - query.query, - log_key(&query.query) - ); + // The query is user input: `debug` keeps only its normalized key, the + // text itself is `trace` (same split as the message handler). + log::debug!("inline query [key={}]", log_key(&query.query)); + log::trace!("inline query: {}", query.query); // No retries: the debounce plus a 1s/2s backoff would outlast the inline // query the answer belongs to. match x_media::site::fetch_once(&query.query).await { @@ -206,7 +205,7 @@ async fn answer_inline_query(bot: Bot, query: InlineQuery) -> Result {} - Err(e) => log::error!("inline fetch {}: {e}", query.query), + Err(e) => log::error!("inline fetch [key={}]: {e}", log_key(&query.query)), } Ok(false) } diff --git a/crates/xmedia-bot/src/handlers/mod.rs b/crates/xmedia-bot/src/handlers/mod.rs index 1ef5bff..398198e 100644 --- a/crates/xmedia-bot/src/handlers/mod.rs +++ b/crates/xmedia-bot/src/handlers/mod.rs @@ -130,11 +130,14 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr &t[..end] }) .unwrap_or(""); - // Per-request detail: debug only (message text is user data). + // Per-request detail: who and where at `debug`; the message text itself is + // user data and only ever appears at `trace`, so a `debug` log can be + // shared without leaking what people pasted. log::debug!( - "message from {sender} in {} (private={is_private}): {text_preview}", + "message from {sender} in {} (private={is_private})", message.chat.id ); + log::trace!("message text: {text_preview}"); // URL/edit flows only run in private chats; commands run in any chat. if is_private && let Some(reply) = message.reply_to_message() @@ -152,7 +155,14 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr if let Some(text) = message.text() && let Ok(command) = Command::parse(text, "") { - log::debug!("command from {}: {text_preview}", message.chat.id); + // The command name is what the operator needs at `debug`; its argument + // may be a user-supplied URL, which stays at `trace`. + log::debug!( + "command from {}: {}", + message.chat.id, + text.split_whitespace().next().unwrap_or("") + ); + log::trace!("command text: {text_preview}"); execute_command(&bot, &message, command).await?; return respond(()); } diff --git a/crates/xmedia-bot/src/handlers/urls.rs b/crates/xmedia-bot/src/handlers/urls.rs index 7bb683e..cd216dc 100644 --- a/crates/xmedia-bot/src/handlers/urls.rs +++ b/crates/xmedia-bot/src/handlers/urls.rs @@ -48,20 +48,34 @@ pub async fn start_url_workers() { for _ in 0..URL_WORKERS { let rx = std::sync::Arc::clone(&rx); handles.push(tokio::spawn(async move { + // Supervised like the queue workers: a panic inside a worker + // (a handler, a poisoned lock) used to kill it for good and + // silently shrink the pool — the remaining workers keep the + // channel drained, so nothing else surfaces the loss. The job the + // panicking worker held is lost; the panic is not. while !URL_STOP.load(std::sync::atomic::Ordering::Relaxed) { - let job = rx.lock().await.recv().await; - match job { - Some((message, url)) => { - url_media( - &CONTEXT, - message.chat.id.0, - message.id.0 as i64, - &url, - PostSend::FromChat, - ) - .await + let rx = std::sync::Arc::clone(&rx); + if let Err(e) = tokio::spawn(async move { + while !URL_STOP.load(std::sync::atomic::Ordering::Relaxed) { + let job = rx.lock().await.recv().await; + match job { + Some((message, url)) => { + url_media( + &CONTEXT, + message.chat.id.0, + message.id.0 as i64, + &url, + PostSend::FromChat, + ) + .await; + } + None => break, + } } - None => break, + }) + .await + { + log::error!("url worker panicked, restarting: {e}"); } } })); @@ -83,7 +97,9 @@ pub async fn stop_url_workers() { let handles = URL_WORKER_HANDLES.lock().take(); if let Some(handles) = handles { for handle in handles { - let _ = handle.await; + if let Err(e) = handle.await { + log::error!("url worker panicked at shutdown: {e}"); + } } } } @@ -167,16 +183,19 @@ async fn dispatch_send( reply_to: MessageId, task: &Task, url: &str, + started: std::time::Instant, ) { let result = match task { Task::SendAnimation { .. } => send::send_animation(ctx, task).await, Task::SendMediaSequence { .. } => send::send_media_sequence(ctx, task).await, Task::ForwardMessages { .. } => unreachable!(), }; + // Fetch + cache lookup + upload: the whole wait the user sat through. + let ms = started.elapsed().as_millis(); match result { Ok(message_ids) => { log::info!( - "sent {} message(s) for [key={}]", + "sent {} message(s) for [key={}] chat={chat_id} in {ms}ms", message_ids.len(), log_key(url) ); @@ -188,7 +207,7 @@ async fn dispatch_send( task, }) => { log::info!( - "send for [key={}] failed, queued for retry in {delay_seconds:.1}s", + "send for [key={}] chat={chat_id} failed after {ms}ms, queued for retry in {delay_seconds:.1}s", log_key(url) ); send::enqueue_retry(ctx.task_queue, *task, delay_seconds).await; @@ -210,7 +229,10 @@ async fn dispatch_send( task, }) => { send::settle_task(ctx, &task, send::Settled::Failed).await; - log::error!("send for {url} failed permanently: {err_message}"); + log::error!( + "send for [key={}] chat={chat_id} failed permanently after {ms}ms: {err_message}", + log_key(url) + ); let _ = reply( ctx.sender, chat_id, @@ -330,7 +352,9 @@ async fn run_with_chat_action>( // it makes the future !Send, and the URL workers spawn these. let action = hint.lock().action(); if let Err(e) = sender.send_chat_action(ChatId(chat_id), action).await { - log::error!("send_chat_action failed: {e}"); + // Cosmetic indicator: a failure degrades the experience, it does not + // break the send (a group where the bot cannot send actions). + log::warn!("send_chat_action failed for chat {chat_id}: {e}"); } tokio::pin!(pipeline); loop { @@ -342,7 +366,9 @@ async fn run_with_chat_action>( () = tokio::time::sleep(ACTION_REFRESH) => { let action = hint.lock().action(); if let Err(e) = sender.send_chat_action(ChatId(chat_id), action).await { - log::error!("send_chat_action failed: {e}"); + // Cosmetic indicator: a failure degrades the experience, it does not + // break the send (a group where the bot cannot send actions). + log::warn!("send_chat_action failed for chat {chat_id}: {e}"); } } } @@ -434,6 +460,9 @@ async fn url_media_inner( hint: &parking_lot::Mutex, ) { let reply_to = MessageId(reply_to_message_id as i32); + // Whole-link timer for the result lines: fetch (ugoira encode, HLS remux + // included) + cache lookup + upload — the wait the user actually had. + let started = std::time::Instant::now(); // Link cache: a post sent before is re-sent from Telegram file ids — // no source-site request, no download, no upload. Keyed by the @@ -502,19 +531,23 @@ async fn url_media_inner( Some(cached), post_send, ); - dispatch_send(ctx, chat_id, reply_to, &task, url).await; + dispatch_send(ctx, chat_id, reply_to, &task, url, started).await; return; } - log::debug!("fetching {url} [key={}]", log_key(url)); + log::debug!("fetching [key={}]", log_key(url)); + log::trace!("fetching {url}"); match x_media::site::fetch(url).await { // Unsupported links are ignored silently (Python parity). Ok(None) => { - log::debug!("no site pattern matches {url}; ignoring"); + // The URL itself is user data, so only `trace` names the link; + // `debug` just records that the message was looked at. + log::debug!("no site pattern matches the link; ignoring"); + log::trace!("no site pattern matches {url}"); } // Retries exhausted: notify the user (Rust-only requirement 3). Err(e) => { - log::error!("fetch {url}: {e}"); + log::error!("fetch [key={}]: {e}", log_key(url)); let _ = reply(ctx.sender, chat_id, reply_to, fetch_error_message(&e)).await; } Ok(Some(mut fetched)) => { @@ -577,7 +610,7 @@ async fn url_media_inner( if let Some(dir) = fetched.take_keep_alive() { send::KEEP_ALIVE.lock().push(dir); } - dispatch_send(ctx, chat_id, reply_to, &task, url).await; + dispatch_send(ctx, chat_id, reply_to, &task, url, started).await; } } } diff --git a/crates/xmedia-bot/src/link_cache.rs b/crates/xmedia-bot/src/link_cache.rs index a52186b..fe20af7 100644 --- a/crates/xmedia-bot/src/link_cache.rs +++ b/crates/xmedia-bot/src/link_cache.rs @@ -96,7 +96,7 @@ impl LinkCache { match result { Ok(v) => v, Err(e) => { - log::error!("link cache read failed: {e}"); + log::warn!("link cache read failed: {e}"); None } } @@ -116,7 +116,7 @@ impl LinkCache { }) .await; if let Err(e) = result { - log::error!("link cache write failed: {e}"); + log::warn!("link cache write failed: {e}"); } } @@ -131,7 +131,7 @@ impl LinkCache { }) .await; if let Err(e) = result { - log::error!("link cache delete failed: {e}"); + log::warn!("link cache delete failed: {e}"); } } @@ -150,7 +150,7 @@ impl LinkCache { match result { Ok(n) => n, Err(e) => { - log::error!("link cache prune failed: {e}"); + log::warn!("link cache prune failed: {e}"); 0 } } @@ -170,7 +170,7 @@ impl LinkCache { match result { Ok(n) => n, Err(e) => { - log::error!("link cache clear failed: {e}"); + log::warn!("link cache clear failed: {e}"); 0 } } diff --git a/crates/xmedia-bot/src/main.rs b/crates/xmedia-bot/src/main.rs index 5982d01..22ddc73 100644 --- a/crates/xmedia-bot/src/main.rs +++ b/crates/xmedia-bot/src/main.rs @@ -43,7 +43,18 @@ fn spawn_sigterm_handler(_stop_token: StopToken) {} #[tokio::main] async fn main() { dotenv().ok(); - pretty_env_logger::init(); + // Without RUST_LOG nothing at all was logged (env_logger falls back to + // `error`), so a deployment that forgot the variable looked like a bot + // with no logs; and at `debug` the HTTP client's own lines (hyper_util, + // reqwest) outnumbered the bot's by two to one. The timed builder adds + // the timestamp the plain `init` omitted, so a line can be compared with + // a user's report. An explicit RUST_LOG still wins outright. + pretty_env_logger::formatted_timed_builder() + .parse_filters( + &std::env::var("RUST_LOG") + .unwrap_or_else(|_| "info,hyper_util=warn,reqwest=warn".to_string()), + ) + .init(); log::info!("Starting bot"); let bot = Bot::from_env(); @@ -118,6 +129,22 @@ async fn main() { if idle_limiters > 0 { log::debug!("rate limiter: dropped {idle_limiters} idle bucket(s)"); } + // Only speaks up when the queue is not empty: a healthy bot + // has nothing to report, and a periodic "0 pending" line is + // noise that hides the lines that matter. + if let Some((pending, oldest_run_after)) = TASK_QUEUE.pending_backlog().await { + let overdue = crate::db::now_f64() - oldest_run_after; + if overdue >= 0.0 { + log::info!( + "queue: {pending} pending task(s), oldest {overdue:.0}s overdue" + ); + } else { + log::info!( + "queue: {pending} pending task(s), oldest retry in {:.0}s", + -overdue + ); + } + } for (chat_id, prompt_message_id) in removed { // Rewritten in place, not announced: the sweep is a // background timer, and a fresh message would wake the chat diff --git a/crates/xmedia-bot/src/queue.rs b/crates/xmedia-bot/src/queue.rs index 3c9688c..fe44ff6 100644 --- a/crates/xmedia-bot/src/queue.rs +++ b/crates/xmedia-bot/src/queue.rs @@ -192,6 +192,39 @@ impl PersistentTaskQueue { Ok(()) } + /// Pending task count and the oldest `run_after`, for the periodic sweep's + /// health line. Deliberately separate from the worker's own + /// `earliest_run_after`: that one runs on every idle worker cycle and must + /// stay a single indexed `MIN`, while the count is only asked for once per + /// sweep. + pub async fn pending_backlog(&self) -> Option<(i64, f64)> { + let result = self + .pool + .with_conn(|conn| { + let mut stmt = conn + .prepare("SELECT COUNT(*), MIN(run_after) FROM tasks WHERE status='pending'")?; + let mut rows = stmt.query([])?; + match rows.next()? { + Some(row) => { + let count = row.get::<_, i64>(0)?; + match row.get::<_, Option>(1)? { + Some(oldest) if count > 0 => Ok(Some((count, oldest))), + _ => Ok(None), + } + } + None => Ok(None), + } + }) + .await; + match result { + Ok(v) => v, + Err(e) => { + log::error!("queue backlog query failed: {e}"); + None + } + } + } + async fn recover_stale(&self) { self.recover_sweep().await; } @@ -204,6 +237,30 @@ impl PersistentTaskQueue { } } +/// Which task a lease/retry/dead-letter line is about: the chat from the +/// stored payload, plus the post's normalized cache key when the payload +/// carries one (`ForwardMessages` has no source URL). Without these a queue +/// line named only a row id, which is useless to whoever reads the log — the +/// row id is assigned at insert time and appears nowhere else. +/// +/// Built only when the line is actually logged (log arguments are lazy). +fn row_fields(payload: &Value) -> String { + let chat = payload + .get("chat_id") + .or_else(|| payload.get("from_chat_id")) + .and_then(Value::as_i64); + let key = payload + .get("source_url") + .and_then(Value::as_str) + .map(crate::handlers::log_key); + match (chat, key) { + (Some(chat), Some(key)) => format!("chat={chat} [key={key}]"), + (Some(chat), None) => format!("chat={chat}"), + (None, Some(key)) => format!("[key={key}]"), + (None, None) => String::new(), + } +} + impl QueueWorker { /// Supervised worker: the inner loop runs in its own task so a panic /// (e.g. inside a handler or a DB closure) kills only that task; the @@ -330,11 +387,18 @@ impl QueueWorker { return; } }; - log::debug!("processing {} (attempt {})", row.id, row.attempts + 1); + let fields = row_fields(&payload); + log::debug!( + "processing {} {fields} (attempt {})", + row.id, + row.attempts + 1 + ); + let attempt_started = std::time::Instant::now(); let outcome = self.run_with_lease(&row.id, payload).await; + let attempt_ms = attempt_started.elapsed().as_millis(); match outcome { Ok(()) => { - log::debug!("task {} completed", row.id); + log::debug!("task {} {fields} completed in {attempt_ms}ms", row.id); self.delete_row(&row.id).await; } Err(QueueError::Retryable { @@ -348,7 +412,7 @@ impl QueueWorker { // not restate its own wrapper — see `failure_text`.) let message = "retries exhausted".to_string(); log::error!( - "dead-lettering {}: {message} after {} attempt(s)", + "dead-lettering {} {fields}: {message} after {} attempt(s)", row.id, row.attempts + 1 ); @@ -357,7 +421,7 @@ impl QueueWorker { } else { let delay = scaled_retry_delay(delay_seconds, row.attempts); log::debug!( - "task {} rescheduled in {delay:.1}s (attempt {})", + "task {} {fields} attempt {} took {attempt_ms}ms, rescheduled in {delay:.1}s", row.id, row.attempts + 1 ); @@ -366,7 +430,7 @@ impl QueueWorker { } } Err(QueueError::Permanent { message, payload }) => { - log::error!("dead-lettering {}: {message}", row.id); + log::error!("dead-lettering {} {fields}: {message}", row.id); self.delete_row(&row.id).await; (self.dead_letter)(payload, message).await; } @@ -489,6 +553,72 @@ mod tests { queue.stop().await; } + #[test] + fn row_fields_name_the_chat_and_the_post() { + // The payload shapes the three task variants store. + assert_eq!( + row_fields(&serde_json::json!({ + "chat_id": 111, + "source_url": "https://x.com/u/status/1" + })), + "chat=111 [key=twitter:1]" + ); + // A forward has no source URL; a chat id alone must still name the line. + assert_eq!( + row_fields(&serde_json::json!({"from_chat_id": 111, "to_chat_id": 222})), + "chat=111" + ); + // Garbage in the payload must not panic a log line. + assert_eq!(row_fields(&serde_json::json!({"chat_id": "111"})), ""); + assert_eq!(row_fields(&serde_json::Value::Null), ""); + } + + #[tokio::test] + async fn pending_backlog_counts_only_unleased_rows() { + let (queue, _dir) = new_queue().await; + assert_eq!(queue.pending_backlog().await, None, "empty queue"); + + let due = now_f64(); + queue + .enqueue(serde_json::json!({"chat_id": 1}), due) + .await + .unwrap(); + queue + .enqueue(serde_json::json!({"chat_id": 2}), due + 600.0) + .await + .unwrap(); + // Hold the first row in the handler so it is leased, not pending: a + // health line that reported work already in flight as backlog would be + // lying about the queue. + let release = Arc::new(tokio::sync::Notify::new()); + let held = release.clone(); + queue + .start( + move |_payload| { + let held = held.clone(); + async move { + held.notified().await; + Ok(()) + } + }, + |_payload, _message| async {}, + ) + .await; + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + queue.pending_backlog().await.map(|(n, _)| n), + Some(1), + "the leased row is not pending" + ); + let (_, oldest) = queue.pending_backlog().await.unwrap(); + assert!( + (oldest - (due + 600.0)).abs() < 1.0, + "oldest is the earliest run_after: {oldest}" + ); + release.notify_one(); + queue.stop().await; + } + #[tokio::test] async fn retryable_reschedules_then_dead_letters() { let (queue, _dir) = new_queue().await; diff --git a/crates/xmedia-bot/src/send/post_send.rs b/crates/xmedia-bot/src/send/post_send.rs index 0955e9d..d6237de 100644 --- a/crates/xmedia-bot/src/send/post_send.rs +++ b/crates/xmedia-bot/src/send/post_send.rs @@ -257,8 +257,9 @@ pub(crate) async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message match prompt { Ok(prompt_id) => { log::info!( - "edit-before-forward prompt {prompt_id} opened for {} message(s)", - message_ids.len() + "edit-before-forward prompt {prompt_id} opened for {} message(s) [key={}] chat={chat_id}", + message_ids.len(), + log_key(&source_url) ); let source_url = source_url.clone(); ctx.chat_store @@ -283,8 +284,9 @@ pub(crate) async fn post_send_actions(ctx: &AppContext<'_>, task: &Task, message if let Some(channel_id) = forward_channel_id { log::info!( - "forwarding {} message(s) to channel {channel_id}", - message_ids.len() + "forwarding {} message(s) to channel {channel_id} from chat {chat_id} [key={}]", + message_ids.len(), + log_key(&source_url) ); let forward_task = Task::ForwardMessages { from_chat_id: chat_id, diff --git a/crates/xmedia-bot/src/state.rs b/crates/xmedia-bot/src/state.rs index 666c398..a44d095 100644 --- a/crates/xmedia-bot/src/state.rs +++ b/crates/xmedia-bot/src/state.rs @@ -73,7 +73,7 @@ impl ChatStore { }) .await .unwrap_or_else(|e| { - log::error!("chat_state read failed: {e}"); + log::warn!("chat_state read failed: {e}"); None }) .unwrap_or_default(); @@ -98,7 +98,7 @@ impl ChatStore { }) .await; if let Err(e) = result { - log::error!("chat_state write failed: {e}"); + log::warn!("chat_state write failed: {e}"); } }