fix(message_queue): prevent a 502 from stopping the bot

The bot stopped for more than 5 hours. The process continued to run, and
systemd showed NRestarts=0, but the bot wrote no more lines to the journal.
The progress message of a job stayed at "Step 19/25", but the job was
complete and the push to GitLab was successful.

Cause: the read timeout and the write timeout apply to one socket operation
only. They do not apply to the full request. The Telegram API server sent a
502 page through a slow proxy. Each part of that page started the socket
timeout again, thus the request continued for an unlimited time. The consumer
sends one message at a time. Thus this one request stopped all messages. The
"except Exception" block in the consumer loop did not operate, because the
code raised no error.

This commit makes three changes:

1. Time limit. Each request to Telegram now has a limit on the total time.
   If the limit passes, the code raises NetworkError. The usual retry code
   then does the retry. The limit applies to the queue, to the immediate
   messages, to the automatic deletions, and to the check that the worker
   does before a dump. Without a limit, that check can keep a worker for as
   much as job_timeout, which is 2 hours.

2. Shorter errors. The function sanitize_telegram_error() keeps the status
   code and a short part of the body. The failure wrote more than 700 lines
   of HTML to the journal. It now writes one line. The bot also has an error
   handler for the poll loop, thus that loop writes short lines too.

3. Correct order of edits. Each edit of a Telegram message gets an order
   number from Redis. Redis keeps the number of the last edit that Telegram
   accepted. The bot discards an edit whose number is less than that number.
   Example: step 20 of 25 fails, then step 21 of 25 succeeds. The bot then
   discards step 20 and does not show an earlier step. A retry of an edit
   that has a number keeps that number, thus an old edit cannot move in
   front of a later edit.

Two conditions need more care:

- If Redis loses the counter key but keeps the watermark key, the counter
  must not start again at 1. All new numbers would then be less than the
  watermark, and the bot would discard all subsequent edits. Thus the script
  starts the counter at the watermark.

- A message that a version before this one put in the queue has no order
  number. If Telegram accepted an edit for the same message, that edit is
  more recent. Thus the bot discards the message with no number. If Telegram
  accepted no edit, the bot sends the message.

The Lua scripts are in the new file dumpyarabot/lua_scripts.py. Each script
does a compare and a write. Redis runs a script from the start to the end and
does no other command at the same time. Thus two workers cannot change a key
between the compare and the write. The full cycle of test, edit, and mark is
not one operation, but only the bot process reads the queue.

Author: akhilnarang

dumpyarabot/__main__.py

@@ -1,11 +1,14 @@
 import os
 import sys
+import logging
 
+from telegram.error import TelegramError
 from telegram.ext import (ApplicationBuilder, CallbackQueryHandler,
-                          CommandHandler, MessageHandler, filters, JobQueue)
+                          CommandHandler, ContextTypes, MessageHandler,
+                          filters, JobQueue)
 
 from dumpyarabot.handlers import cancel_dump, clear_queue, dump, help_command, restart, status
-from dumpyarabot.message_queue import message_queue
+from dumpyarabot.message_queue import message_queue, sanitize_telegram_error
 from dumpyarabot.mockup_handlers import (handle_enhanced_callback_query,
                                          mockup_command)
 from dumpyarabot.moderated_handlers import (accept_command,
@@ -105,6 +108,39 @@ async def _shutdown_runtime(application):
         pass
 
 
+async def _handle_application_error(
+    update: object, context: ContextTypes.DEFAULT_TYPE
+) -> None:
+    """Write a short line for a failure in the poll loop or in a handler.
+
+    Without this handler, the library writes the full body of the error. A
+    proxy can send an HTML error page that is more than 700 lines long.
+
+    Args:
+        update: The update that caused the failure. The update can be None if
+            the failure occurred in the poll loop.
+        context: The context that holds the error in its error field.
+    """
+    from rich.console import Console
+
+    if isinstance(context.error, TelegramError):
+        Console().print(
+            f"Telegram application error: {sanitize_telegram_error(context.error)}",
+            style="red",
+            markup=False,
+        )
+        return
+
+    logging.getLogger(__name__).error(
+        "Unhandled application error",
+        exc_info=(
+            type(context.error),
+            context.error,
+            context.error.__traceback__,
+        ),
+    )
+
+
 async def register_bot_commands(application):
     """Register bot commands with Telegram for the menu interface."""
     from dumpyarabot.config import USER_COMMANDS
@@ -163,6 +199,7 @@ async def register_bot_commands_job(context):
     application.add_handler(callback_handler)
     application.add_handler(restart_handler)
     application.add_handler(clearqueue_handler)
+    application.add_error_handler(_handle_application_error)
 
     application.post_init = _startup_init
     application.post_shutdown = _shutdown_runtime

dumpyarabot/config.py

@@ -28,8 +28,10 @@ class Settings(BaseSettings):
     DEFAULT_PARSE_MODE: str = "Markdown"
     TELEGRAM_TEXT_READ_TIMEOUT: float = 60.0
     TELEGRAM_TEXT_WRITE_TIMEOUT: float = 60.0
+    TELEGRAM_TEXT_REQUEST_TIMEOUT: float = 135.0
     TELEGRAM_DOCUMENT_READ_TIMEOUT: float = 120.0
     TELEGRAM_DOCUMENT_WRITE_TIMEOUT: float = 120.0
+    TELEGRAM_DOCUMENT_REQUEST_TIMEOUT: float = 255.0
 
     # Optional custom base URL for Telegram Bot API (e.g. nginx reverse proxy)
     # Default: https://api.telegram.org/bot

dumpyarabot/lua_scripts.py

@@ -0,0 +1,159 @@
+"""Lua scripts that keep the order of Telegram message edits.
+
+The bot shows the progress of a job in one Telegram message. The bot edits
+that message for each step. Two workers and the bot can send edits at the
+same time, thus the edits can arrive in the wrong order.
+
+Two Redis keys control the order. Each key applies to one Telegram message,
+and the key name contains the chat and the message:
+
+- The counter key gives an order number to each edit.
+- The watermark key holds the order number of the last edit that Telegram
+  accepted.
+
+An edit whose order number is less than the watermark is too old. The bot
+discards such an edit. Example: step 20 of 25 fails and waits for a retry.
+Step 21 of 25 then succeeds and moves the watermark. Step 20 is now too old.
+If the bot sent step 20, the display would show an earlier step.
+
+Each script does a compare and a write. Redis runs a script from the start to
+the end and does no other command at the same time. Thus no worker can change
+a key between the compare and the write. Python code that did a GET and then
+a SET would not be safe.
+
+Note: one script is safe, but the full cycle is not one operation. The bot
+does the stale test, then sends the edit to Telegram, then moves the
+watermark. Only the bot process reads the queue. The arq workers only write
+to the queue. If two bot processes read the queue at the same time, the two
+processes can send edits in the wrong order.
+"""
+
+# Give the next edit of one Telegram message its order number.
+#
+# If Redis loses the counter key but keeps the watermark key, the counter
+# starts again at 1. All new edits would then be less than the watermark, and
+# the bot would discard all of them. The message would stay at the same step
+# for all time. To prevent this, start the counter at the watermark.
+#
+# KEYS[1]: the counter key.
+# KEYS[2]: the watermark key.
+# ARGV[1]: the time-to-live of the key, in seconds.
+#
+# Returns the new order number.
+STAMP_EDIT_SEQUENCE: str = """
+if redis.call('EXISTS', KEYS[1]) == 0 then
+    local applied = tonumber(redis.call('GET', KEYS[2]))
+    if applied then
+        redis.call('SET', KEYS[1], applied)
+    end
+end
+local sequence = redis.call('INCR', KEYS[1])
+redis.call('EXPIRE', KEYS[1], ARGV[1])
+return sequence
+"""
+
+# Tell if a later edit replaced this edit.
+#
+# KEYS[1]: the watermark key.
+# ARGV[1]: the order number of the edit.
+#
+# Returns 1 if the edit is too old, or 0 if the edit is current.
+IS_STALE_EDIT: str = """
+local applied = tonumber(redis.call('GET', KEYS[1]))
+local incoming = tonumber(ARGV[1])
+if applied and incoming < applied then
+    return 1
+end
+return 0
+"""
+
+# Move the watermark forward after Telegram accepts an edit.
+#
+# The script writes only if the order number is more than the watermark. Thus
+# a slow edit that arrives late cannot move the watermark to the rear.
+#
+# KEYS[1]: the watermark key.
+# ARGV[1]: the order number of the edit.
+# ARGV[2]: the time-to-live of the key, in seconds.
+#
+# Returns 1 if the script moved the watermark, or 0 if it did not.
+MARK_EDIT_APPLIED: str = """
+local applied = tonumber(redis.call('GET', KEYS[1]))
+local incoming = tonumber(ARGV[1])
+local ttl = tonumber(ARGV[2])
+if not applied or incoming > applied then
+    redis.call('SET', KEYS[1], incoming, 'EX', ttl)
+    return 1
+end
+if incoming == applied then
+    redis.call('EXPIRE', KEYS[1], ttl)
+end
+return 0
+"""
+
+# Put an edit in the queue again, but only if a later edit did not replace it.
+#
+# The compare and the write occur in one script. Two Redis commands would let
+# another worker move the watermark between the two commands. The bot would
+# then put an edit in the queue that it must discard.
+#
+# KEYS[1]: the watermark key.
+# KEYS[2]: the destination queue.
+# ARGV[1]: the order number of the edit.
+# ARGV[2]: 'zset' for a delayed retry, or 'list' for an immediate retry.
+# ARGV[3]: the score for the delayed set, as a UNIX time.
+# ARGV[4]: the message, as JSON.
+#
+# Returns 1 if the script put the message in the queue, or 0 if the message
+# is too old.
+REQUEUE_EDIT_IF_CURRENT: str = """
+local applied = tonumber(redis.call('GET', KEYS[1]))
+local incoming = tonumber(ARGV[1])
+if applied and incoming < applied then
+    return 0
+end
+if ARGV[2] == 'zset' then
+    redis.call('ZADD', KEYS[2], ARGV[3], ARGV[4])
+else
+    redis.call('LPUSH', KEYS[2], ARGV[4])
+end
+return 1
+"""
+
+# Keep the status text of a job, but only if the text is not older than the
+# text in Redis. The value has this format: "v1\n<sequence>\n<time>\n<text>".
+#
+# KEYS[1]: the status-text key.
+# ARGV[1]: the order number of the text.
+# ARGV[2]: the time of the text, as a UNIX time.
+# ARGV[3]: the time-to-live of the key, in seconds.
+# ARGV[4]: the full value to write.
+#
+# Returns 1 if the script wrote the text, or 0 if the text is too old.
+STORE_LATEST_STATUS_TEXT: str = """
+local current = redis.call('GET', KEYS[1])
+local next_seq = tonumber(ARGV[1])
+local next_ts = tonumber(ARGV[2])
+local ttl = tonumber(ARGV[3])
+
+local current_seq = nil
+local current_ts = nil
+if current and string.sub(current, 1, 3) == "v1\\n" then
+    local rest = string.sub(current, 4)
+    local first_newline = string.find(rest, "\\n", 1, true)
+    if first_newline then
+        current_seq = tonumber(string.sub(rest, 1, first_newline - 1))
+        local rest_after_seq = string.sub(rest, first_newline + 1)
+        local second_newline = string.find(rest_after_seq, "\\n", 1, true)
+        if second_newline then
+            current_ts = tonumber(string.sub(rest_after_seq, 1, second_newline - 1))
+        end
+    end
+end
+
+if not current_seq or next_seq > current_seq or (next_seq == current_seq and (not current_ts or next_ts >= current_ts)) then
+    redis.call('SET', KEYS[1], ARGV[4], 'EX', ttl)
+    return 1
+end
+return 0
+"""