YoursFunny/TelegramTwitterMediaBot
1
0
Fork 0
mirror of https://github.com/TheFunny/TelegramTwitterMediaBot.git synced 2026-09-23 23:32:05 +00:00
Code Releases Activity

feat(log): survive a bare deployment and name what each line is about

Browse Source 
P0 (foundation) + P1 (diagnostic depth) of the logging plan:

- main.rs initializes the timed builder with a default filter of
  `info,hyper_util=warn,reqwest=warn`. Without RUST_LOG nothing was logged at
  all (env_logger falls back to `error`), so `docker run --env-file .env` was
  silent, and the plain `init` had no timestamps.
- info-and-above lines stop printing user URLs (fetch/send failures, inline
  fetch, bsky's remux warnings). The full URL, the message text and the inline
  query move to `trace`, so a `debug` log can be handed to someone else.
- Lifecycle lines name the chat and the post: sent/failed/queued plus the
  total `ms`, the edit prompt, the channel forward, and every queue line
  (`chat=` + `[key=…]` + per-attempt `ms`, dead-letters included).
- Queue work is visible: `x-media`'s fetch line carries its duration (ugoira
  encode and HLS remux included), and the 300s sweep reports the pending count
  and how overdue the oldest task is — only when the queue is non-empty.
- URL workers are supervised like the queue workers: a panicking worker used
  to die silently and shrink the pool for the rest of the process.
- Degradations that still serve the user (cache/state write or read failures,
  a failed chat action) are `warn`, not `error`.

Verified against the scripted fake-API harness: unset RUST_LOG logs info with
timestamps, `debug` carries no user URL, `trace` does, a cache-hit send logs
`chat=111 in 5ms`, a failing send queues and dead-letters with chat+key, and
the sweep reports the pending retry.
This commit is contained in:
YoursFunny
2026-09-20 19:07:23 +08:00
parent 9e873131d4
commit 3f9821d475
13 changed files with 265 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/xmedia-bot/src/handlers/inline.rs
+5 -6
View File
@@ -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<bool, RequestError> {
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<bool, Reque
}
}
Ok(None) => {}
Err(e) => log::error!("inline fetch {}: {e}", query.query),
Err(e) => log::error!("inline fetch [key={}]: {e}", log_key(&query.query)),
}
Ok(false)
}
crates/xmedia-bot/src/handlers/mod.rs
+13 -3
View File
@@ -130,11 +130,14 @@ pub async fn message_handler(bot: Bot, message: Message) -> Result<(), RequestEr
&t[..end]
})
.unwrap_or("<no text>");
// 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("<empty>")
);
log::trace!("command text: {text_preview}");
execute_command(&bot, &message, command).await?;
return respond(());
}
crates/xmedia-bot/src/handlers/urls.rs
+56 -23
View File
@@ -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<F: Future<Output = ()>>(
// 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<F: Future<Output = ()>>(
() = 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<ActionHint>,
) {
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;
}
}
}
crates/xmedia-bot/src/link_cache.rs
+5 -5
View File
@@ -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
}
}
crates/xmedia-bot/src/main.rs
+28 -1
View File
@@ -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
crates/xmedia-bot/src/queue.rs
+135 -5
View File
@@ -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<f64>>(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;
crates/xmedia-bot/src/send/post_send.rs
+6 -4
View File
@@ -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,
crates/xmedia-bot/src/state.rs
+2 -2
View File
@@ -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}");
}
}