Automated sync from GitHub

.gitattributes

@@ -1,8 +1,8 @@
-*.otf filter=lfs diff=lfs merge=lfs -text
-*.webp filter=lfs diff=lfs merge=lfs -text
 *.mp4 filter=lfs diff=lfs merge=lfs -text
 *.mp3 filter=lfs diff=lfs merge=lfs -text
 *.jpg filter=lfs diff=lfs merge=lfs -text
 *.jpeg filter=lfs diff=lfs merge=lfs -text
 *.png filter=lfs diff=lfs merge=lfs -text
 *.ttf filter=lfs diff=lfs merge=lfs -text
+*.otf filter=lfs diff=lfs merge=lfs -text
+*.webp filter=lfs diff=lfs merge=lfs -text

bot/app.py

@@ -21,7 +21,7 @@ from telethon import errors, events, functions, types
 from telethon.tl.custom import Button
 
 from . import fasttransfer as transfer
-from . import media, mediainfo, panels, speed, sysinfo
+from . import errorlog, media, mediainfo, panels, speed, sysinfo
 from .config import Config
 from .db import MAX_FORMATS, Database
 from .errors import FormatError
@@ -195,7 +195,7 @@ class App:
         self._speed_lock = asyncio.Lock()  # one speed test at a time
         self.telegraph = Telegraph(self._load_telegraph_token, self._save_telegraph_token)
         self._tasks: Set["asyncio.Future[Any]"] = set()
-        self.renamer = Renamer(client, cfg, db, self._log_send)
+        self.renamer = Renamer(client, cfg, db)
         self.queue = JobQueue(self.renamer.process, workers=cfg.max_jobs, on_cancelled=self._on_job_cancelled, settle=QUEUE_SETTLE)
         self._warned: Set[int] = set()
 
@@ -257,6 +257,7 @@ class App:
                 else:
                     await self._stranger(uid)
                 return
+            context = errorlog.error_context.set(lambda: errorlog.describe_update(event))  # for the log channel report
             try:
                 await fn(event)
             except events.StopPropagation:
@@ -264,7 +265,7 @@ class App:
             except errors.MessageNotModifiedError:
                 pass
             except Exception:  # noqa: BLE001
-                log.exception("handler %s failed", getattr(fn, "__name__", fn))
+                log.exception("Error handling update:")
                 try:
                     if hasattr(event, "data"):
                         await event.answer("⚠️ Something went wrong.", alert=True)
@@ -272,6 +273,8 @@ class App:
                         await event.respond("⚠️ Something went wrong. Please try again.")
                 except Exception:  # noqa: BLE001
                     pass
+            finally:
+                errorlog.error_context.reset(context)
 
         return wrapper
 

bot/errorlog.py

@@ -0,0 +1,242 @@
+"""Error reports for the log channel.
+
+Every ERROR-level log record becomes one message: what the bot was doing, then the full traceback, inside a Python code
+block (Telegram shows it with a "Python" label and a copy button).  An identical error is reported once per couple of
+minutes, no more than ``PER_MINUTE`` reports go out in a minute, and secrets are masked before anything is sent.
+"""
+from __future__ import annotations
+
+import asyncio
+import contextvars
+import logging
+import os
+import re
+import time
+from collections import deque
+from dataclasses import dataclass
+from typing import Any, Awaitable, Callable, Deque, Dict, Iterable, List, Optional, Set, Tuple
+from urllib.parse import urlsplit
+
+from .utils import clip, esc
+
+log = logging.getLogger(__name__)
+
+Send = Callable[[str], Awaitable[None]]
+
+MAX_TEXT = 3800  # UTF-16 units of the code block: Telegram takes 4096 per message
+HEAD_LIMIT = 700  # characters of "what was happening" kept when a traceback shares the message
+REPEAT_WINDOW = 120.0  # seconds during which an identical error is reported only once
+PER_MINUTE = 10  # reports sent per minute at most
+
+# What the bot is working on right now: text, or a function that returns it.  Errors logged anywhere below it mention
+# it, so a failure deep inside a helper still says which message or file it belongs to.
+error_context: "contextvars.ContextVar[Any]" = contextvars.ContextVar("error_context", default=None)
+
+
+def describe_update(event: Any) -> str:
+    """One line about a Telethon event: who sent what, where.  Never raises."""
+    try:
+        kind = type(event).__qualname__.removesuffix(".Event")  # NewMessage / CallbackQuery
+        parts = [f"sender_id={getattr(event, 'sender_id', None)}", f"chat_id={getattr(event, 'chat_id', None)}"]
+        data = getattr(event, "data", None)
+        if data is not None:
+            parts.append(f"message_id={getattr(event, 'message_id', None)}")
+            parts.append("data=" + repr(bytes(data).decode("utf-8", "replace")))
+        else:
+            parts.append(f"message_id={getattr(event, 'id', None)}")
+            text = getattr(event, "raw_text", None)
+            if text:
+                parts.append("text=" + repr(clip(text, 200)))
+            file = getattr(getattr(event, "message", None), "file", None)
+            if file is not None:
+                parts.append(f"file={getattr(file, 'name', None)!r} ({getattr(file, 'mime_type', None)}, {getattr(file, 'size', None)} bytes)")
+        return f"{kind}({', '.join(parts)})"
+    except Exception:  # noqa: BLE001
+        return type(event).__name__
+
+
+_SQL_PARAMS = re.compile(r"\[parameters: [^\n]*\]")  # SQLAlchemy prints the values it bound, e.g. a saved session
+_TOKENISH = re.compile(r"[A-Za-z0-9_\-+/=]{60,}")  # session strings, access tokens, long hashes
+
+
+def _mask_token(match: "re.Match[str]") -> str:
+    text = match.group(0)
+    return "***" if re.search(r"[A-Za-z0-9]", text) else text  # a line of dashes is not a secret
+
+
+def redact(text: str, secrets: Iterable[str] = ()) -> str:
+    """Hide the configured secrets and anything that looks like a session string or access token."""
+    for secret in secrets:
+        if len(secret) >= 6:
+            text = text.replace(secret, "***")
+    return _TOKENISH.sub(_mask_token, _SQL_PARAMS.sub("[parameters: …]", text))
+
+
+def secrets_from(cfg: Any) -> List[str]:
+    """The values of the configuration that must never appear in a report."""
+    values = [cfg.bot_token, cfg.api_hash, cfg.session_string, cfg.database_url]
+    try:
+        values.append(urlsplit(cfg.database_url).password or "")
+    except ValueError:
+        pass
+    return [v for v in values if v]
+
+
+def _u16(text: str) -> int:
+    return len(text.encode("utf-16-le")) // 2
+
+
+def _tail(text: str, units: int) -> str:
+    """The end of ``text`` that fits in ``units`` UTF-16 units, starting at the beginning of a line."""
+    used, start = 0, len(text)
+    for i in range(len(text) - 1, -1, -1):
+        used += 2 if ord(text[i]) > 0xFFFF else 1
+        if used > units:
+            break
+        start = i
+    tail = text[start:]
+    if start > 0:
+        newline = tail.find("\n")
+        tail = tail[newline + 1:] if newline != -1 else tail
+    return tail
+
+
+def _fit(head: str, trace: str) -> str:
+    """``head`` + the traceback in at most ``MAX_TEXT``: when it is too long the end is kept, as that holds the error."""
+    if not trace:
+        return clip(head, MAX_TEXT // 2)
+    head = clip(head, HEAD_LIMIT)
+    room = MAX_TEXT - _u16(head) - 2
+    if _u16(trace) > room:
+        trace = "…\n" + _tail(trace, room - 2)
+    return head + "\n\n" + trace
+
+
+def _signature(exc_info: Any) -> str:
+    """Same exception type raised on the same line = the same error."""
+    etype, _value, tb = exc_info
+    while tb is not None and tb.tb_next is not None:
+        tb = tb.tb_next
+    where = f"{os.path.basename(tb.tb_frame.f_code.co_filename)}:{tb.tb_lineno}" if tb is not None else ""
+    return f"{getattr(etype, '__name__', etype)}@{where}"
+
+
+def build_report(record: logging.LogRecord, secrets: Iterable[str] = ()) -> Tuple[str, str]:
+    """``(signature, html)``: the record's message, what the bot was working on, and the traceback in a Python block."""
+    try:
+        head = record.getMessage()
+    except Exception:  # noqa: BLE001
+        head = str(record.msg)
+    context = error_context.get()
+    if callable(context):
+        context = context()
+    if context and str(context) not in head:
+        head += "\n" + str(context)
+    trace, signature = "", f"{record.name}|{head[:100]}"
+    if record.exc_info and record.exc_info[0] is not None:
+        trace = logging.Formatter().formatException(record.exc_info)
+        signature = _signature(record.exc_info)
+    text = _fit(redact(head, secrets), redact(trace, secrets))
+    return signature, f'<pre><code class="language-python">{esc(text)}</code></pre>'
+
+
+@dataclass
+class _Seen:
+    at: float
+    repeats: int = 0
+
+
+class ErrorReporter:
+    """Sends the reports to the log channel one after another and keeps repeats and floods out of it."""
+
+    def __init__(self, send: Send, *, repeat_window: float = REPEAT_WINDOW, per_minute: int = PER_MINUTE,
+                 clock: Callable[[], float] = time.monotonic):
+        self._send = send
+        self._repeat_window = repeat_window
+        self._per_minute = per_minute
+        self._clock = clock
+        self._seen: Dict[str, _Seen] = {}
+        self._sent: Deque[float] = deque()  # when the reports of the last minute went out
+        self._skipped = 0
+        self._lock = asyncio.Lock()
+        self._tasks: Set["asyncio.Future[None]"] = set()
+
+    def submit(self, signature: str, body: str) -> None:
+        """Queue one report (inside the event loop).  A repeat or a flood is counted and mentioned in the next report."""
+        now = self._clock()
+        seen = self._seen.get(signature)
+        if seen is not None and now - seen.at < self._repeat_window:
+            seen.repeats += 1
+            return
+        while self._sent and now - self._sent[0] >= 60:
+            self._sent.popleft()
+        if len(self._sent) >= self._per_minute:
+            self._skipped += 1
+            return
+        notes = []
+        if seen is not None and seen.repeats:
+            notes.append(f"🔁 {seen.repeats} more of this error since the last report.")
+        if self._skipped:
+            notes.append(f"⚠️ {self._skipped} other error report(s) were left out to keep this chat readable.")
+        self._skipped = 0
+        self._seen[signature] = _Seen(now)
+        if len(self._seen) > 200:
+            self._seen = {k: v for k, v in self._seen.items() if now - v.at < self._repeat_window}
+        self._sent.append(now)
+        task = asyncio.ensure_future(self._deliver(body + "".join("\n" + note for note in notes)))
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.discard)
+
+    async def _deliver(self, text: str) -> None:
+        async with self._lock:  # in the order they happened
+            try:
+                await self._send(text)
+            except Exception as exc:  # noqa: BLE001
+                log.warning("could not send an error report to the log channel: %s", exc)
+
+    async def flush(self, timeout: float = 5.0) -> None:
+        """Wait (a little) for the reports that are still on their way."""
+        if self._tasks:
+            await asyncio.wait(set(self._tasks), timeout=timeout)
+
+
+class ChannelLogHandler(logging.Handler):
+    """Turns every ERROR record, from any thread, into a report for the log channel."""
+
+    def __init__(self, reporter: ErrorReporter, loop: asyncio.AbstractEventLoop, secrets: Iterable[str] = ()):
+        super().__init__(level=logging.ERROR)
+        self._reporter = reporter
+        self._loop = loop
+        self._secrets = [s for s in secrets if s]
+        self._logger: Optional[logging.Logger] = None
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if record.name == __name__:  # a problem with the reports themselves must not become a report
+            return
+        try:
+            signature, body = build_report(record, self._secrets)
+            self._loop.call_soon_threadsafe(self._reporter.submit, signature, body)
+        except Exception:  # noqa: BLE001 - logging must never break the code that logs
+            pass
+
+    def attach(self, logger: logging.Logger) -> None:
+        logger.addHandler(self)
+        self._logger = logger
+
+    def detach(self) -> None:
+        if self._logger is not None:
+            self._logger.removeHandler(self)
+            self._logger = None
+
+    async def aclose(self, timeout: float = 5.0) -> None:
+        """Stop listening, then give the reports that are on their way a moment to arrive."""
+        self.detach()
+        await asyncio.sleep(0)  # reports handed over from another thread land in the loop first
+        await self._reporter.flush(timeout)
+
+
+def install(send: Send, secrets: Iterable[str] = (), *, logger: Optional[logging.Logger] = None) -> ChannelLogHandler:
+    """Send the errors logged on ``logger`` (the root logger by default) through ``send``.  Call it inside the event loop."""
+    handler = ChannelLogHandler(ErrorReporter(send), asyncio.get_running_loop(), secrets)
+    handler.attach(logger or logging.getLogger())
+    return handler

bot/main.py

@@ -9,7 +9,7 @@ import sys
 from pathlib import Path
 from typing import Awaitable, Callable
 
-from . import __version__
+from . import __version__, errorlog
 from .config import Config, ConfigError, load_config
 from .db import Database
 
@@ -68,6 +68,7 @@ async def amain() -> int:
     username = getattr(me, "username", "") or ""
     app.telegraph.set_author(getattr(me, "first_name", "") or username, f"https://t.me/{username}" if username else "")
     await app.setup()
+    channel_log = errorlog.install(app._log_send, errorlog.secrets_from(cfg)) if cfg.log_channel else None  # errors -> LOG_CHANNEL
     log.info("@%s is online (DC %s, %d connection(s) per transfer, fast transfer %s)",
              me.username, client.session.dc_id, cfg.connections, "on" if cfg.fast_transfer else "off")
     if cfg.log_channel:
@@ -90,6 +91,8 @@ async def amain() -> int:
 
     log.info("shutting down")
     await app.shutdown()
+    if channel_log is not None:
+        await channel_log.aclose()  # sends the reports still on their way, while the client is connected
     await runner.cleanup()
     try:
         await client.disconnect()

bot/pipeline.py

@@ -5,12 +5,12 @@ import logging
 import mimetypes
 import shutil
 import time
-import traceback
 from pathlib import Path
-from typing import Any, Awaitable, Callable, Dict, Optional
+from typing import Any, Dict, Optional
 
 from telethon import errors, types
 
+from . import errorlog
 from . import fasttransfer as transfer
 from . import media
 from .config import Config
@@ -66,20 +66,12 @@ def original_name_of(message: Any) -> str:
 
 
 class Renamer:
-    def __init__(self, client: Any, cfg: Config, db: Database, log_channel_send: Optional[Callable[[str], Awaitable[None]]] = None):
+    def __init__(self, client: Any, cfg: Config, db: Database):
         self.client = client
         self.cfg = cfg
         self.db = db
-        self._log_send = log_channel_send
 
     # ------------------------------------------------------------------
-    async def _notify_log(self, text: str) -> None:
-        if self._log_send is not None:
-            try:
-                await self._log_send(text)
-            except Exception as exc:  # noqa: BLE001
-                log.warning("could not write to the log channel: %s", exc)
-
     async def _fresh_message(self, job: Job) -> Any:
         """Re-fetch the message so its file reference is valid even after a long queue wait."""
         msg = await self.client.get_messages(job.chat_id, ids=job.msg_id)
@@ -102,6 +94,9 @@ class Renamer:
         work.mkdir(parents=True, exist_ok=True)
         reporter = ProgressReporter(self.client, job.status_msg, job.id, job.new_name, self.cfg.progress_interval)
         started = time.monotonic()
+        context = errorlog.error_context.set(  # the log channel report says which file this was
+            f"user {job.user_id} · job #{job.id}\n{job.file_name} → {job.new_name} ({human_size(job.size)})"
+        )
         try:
             await self._run(job, s, work, reporter, started)
         except Cancelled:
@@ -109,12 +104,11 @@ class Renamer:
             await reporter.finish("❌ <b>Cancelled</b>\n<code>" + esc(clip(job.new_name, 120)) + "</code>")
         except Exception as exc:  # noqa: BLE001
             job.state = "failed"
-            log.exception("job %s failed", job.id)
+            log.exception("Error processing file:")
             reason = clip(f"{type(exc).__name__}: {exc}", 300)
             await reporter.finish(f"❌ <b>Failed</b>\n<code>{esc(clip(job.new_name, 120))}</code>\n\n{esc(reason)}")
-            tb = "".join(traceback.format_exception_only(type(exc), exc)).strip()
-            await self._notify_log(f"❌ <b>Job failed</b> (user <code>{job.user_id}</code>)\n<code>{esc(clip(job.file_name, 100))}</code>\n<code>{esc(clip(tb, 600))}</code>")
         finally:
+            errorlog.error_context.reset(context)
             shutil.rmtree(work, ignore_errors=True)
 
     async def _run(self, job: Job, s: Dict[str, Any], work: Path, reporter: ProgressReporter, started: float) -> None:
@@ -229,7 +223,3 @@ class Renamer:
         await reporter.finish(
             f"✅ <b>Done</b> in {human_duration(took)}\n<code>{esc(clip(job.new_name, 120))}</code>\n{human_size(final_size)}{where}{note}"
         )
-        await self._notify_log(
-            f"✅ <code>{job.user_id}</code>: <code>{esc(clip(job.file_name, 80))}</code> → "
-            f"<code>{esc(clip(job.new_name, 80))}</code> ({human_size(final_size)}, {human_duration(took)})"
-        )

tests/conftest.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import os
 import shutil
 import subprocess
@@ -9,6 +10,8 @@ import pytest
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 
+from bot.errorlog import ChannelLogHandler  # noqa: E402
+
 from .appkit import ctx  # noqa: E402  the handler-level fixture, shared by every test module
 
 __all__ = ["ctx", "needs_ffmpeg"]
@@ -16,6 +19,14 @@ __all__ = ["ctx", "needs_ffmpeg"]
 HAVE_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
 needs_ffmpeg = pytest.mark.skipif(not HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed")
 
+
+@pytest.fixture(autouse=True)
+def _no_stray_error_reporters():
+    """A test that stops halfway must not leave an error reporter on the root logger for the tests after it."""
+    yield
+    for handler in [h for h in logging.getLogger().handlers if isinstance(h, ChannelLogHandler)]:
+        handler.detach()
+
 SRT_A = "1\n00:00:00,000 --> 00:00:01,000\nkonnichiwa\n\n"
 SRT_B = "1\n00:00:00,000 --> 00:00:01,000\nhello\n\n"
 FFMETA = """;FFMETADATA1

tests/test_app.py

@@ -8,7 +8,7 @@ import pytest
 from telethon import functions
 
 from bot import app as appmod
-from bot import panels
+from bot import errorlog, panels
 from bot.app import COMMANDS, OWNER_COMMANDS, PANEL_NAMES, App, Pending, cmd_pattern, is_renameable
 from bot.config import load_config
 from bot.db import Database
@@ -788,6 +788,40 @@ async def test_handler_failures_are_reported_to_the_user(ctx, monkeypatch):
     assert "Auto Rename Bot" in (await say(ctx, OWNER, "/start")).texts[0]
 
 
+async def test_handler_failures_reach_the_log_channel_with_the_update_and_the_traceback(ctx, monkeypatch):
+    channel_log = errorlog.install(ctx.app._log_send, errorlog.secrets_from(ctx.cfg))  # the root logger, as in main.py
+
+    async def boom(*a, **k):
+        raise RuntimeError("database is down")
+
+    monkeypatch.setattr(ctx.db, "get_settings", boom)
+    ev = await say(ctx, OWNER, "/start")
+    await channel_log.aclose()
+    assert ev.texts == ["⚠️ Something went wrong. Please try again."]
+    (_chat, text, kw), = [s for s in ctx.client.sent if s[0] == LOG]
+    assert kw["parse_mode"] == "html"
+    assert text.startswith(
+        '<pre><code class="language-python">Error handling update:\n'
+        f"FakeEvent(sender_id={OWNER}, chat_id={OWNER}, message_id=1, text='/start')\n\nTraceback (most recent call last):"
+    )
+    assert text.endswith("RuntimeError: database is down</code></pre>")
+
+
+async def test_an_error_inside_a_helper_still_names_the_update_that_started_it(ctx, monkeypatch):
+    channel_log = errorlog.install(ctx.app._log_send)
+
+    async def boom(*a, **k):
+        raise RuntimeError("telegraph exploded")
+
+    monkeypatch.setattr(ctx.app, "_mediainfo", boom)
+    reply = document_message("Movie.mkv", 5 * 1024 * 1024)
+    await say(ctx, OWNER, "/mediainfo", reply=reply)
+    await channel_log.aclose()
+    (_chat, text, _kw), = [s for s in ctx.client.sent if s[0] == LOG]
+    assert text.startswith('<pre><code class="language-python">/mediainfo failed\nFakeEvent(sender_id=')
+    assert "text='/mediainfo'" in text and text.endswith("RuntimeError: telegraph exploded</code></pre>")
+
+
 async def test_a_file_is_handled_by_exactly_one_handler(ctx):
     ctx.app.pending[OWNER] = Pending("template", OWNER, 1, "rename", False, time.time() + 100)
     ev = await send_doc(ctx, OWNER)

tests/test_errorlog.py

@@ -0,0 +1,402 @@
+"""Error reports for the log channel: what they look like, what is hidden or held back, and how they are delivered."""
+import asyncio
+import datetime
+import html
+import logging
+import re
+from types import SimpleNamespace
+
+import pytest
+from telethon import events, types
+
+from bot import errorlog
+from bot.config import load_config
+from bot.errorlog import ChannelLogHandler, ErrorReporter, build_report, describe_update, redact
+
+from .tgfakes import FakeCallback, FakeEvent, document_message
+
+CODE = '<pre><code class="language-python">'
+
+
+class Catch(logging.Handler):
+    def __init__(self):
+        super().__init__()
+        self.records = []
+
+    def emit(self, record):
+        self.records.append(record)
+
+
+def make_record(message="Error handling update:", raiser=None, *args):
+    """A real log record, made the way the bot makes them: ``log.exception`` inside an ``except``."""
+    logger = logging.getLogger("errorlog.test.records")
+    logger.propagate = False
+    catch = Catch()
+    logger.addHandler(catch)
+    try:
+        try:
+            if raiser is not None:
+                raiser()
+        except Exception:  # noqa: BLE001
+            logger.exception(message, *args)
+        else:
+            logger.error(message, *args)
+    finally:
+        logger.removeHandler(catch)
+    return catch.records[0]
+
+
+def visible(report):
+    """What Telegram shows: the report without its HTML markup."""
+    return html.unescape(re.sub(r"</?(?:pre|code)[^>]*>", "", report))
+
+
+def fail(text="bad"):
+    raise ValueError(text)
+
+
+def fail_elsewhere():
+    raise ValueError("bad")
+
+
+def chained():
+    try:
+        {}["missing"]
+    except KeyError:
+        raise RuntimeError("could not read <b>this</b> & that")
+
+
+def very_deep(depth=200):
+    """A traceback with ``depth`` different frames (a recursion would be shortened by Python)."""
+    scope = {}
+    source = "\n".join(f"def f{i}():\n    return f{i + 1}()" for i in range(depth)) + f"\ndef f{depth}():\n    raise ValueError('bottom')"
+    exec(source, scope)  # noqa: S102
+    scope["f0"]()
+
+
+# ---------------------------------------------------------------------------
+# describing the update
+# ---------------------------------------------------------------------------
+
+
+def test_a_message_with_a_file_is_described_with_its_text_and_file():
+    message = document_message("1146 - One Piece [Dual] 720p @Anime_X_Universe.mkv", 316_000_000, caption="hello")
+    line = describe_update(FakeEvent(1000001, message=message))
+    assert line == (
+        "FakeEvent(sender_id=1000001, chat_id=1000001, message_id=10, text='hello', "
+        "file='1146 - One Piece [Dual] 720p @Anime_X_Universe.mkv' (video/x-matroska, 316000000 bytes))"
+    )
+
+
+def test_a_button_press_is_described_with_its_data():
+    assert describe_update(FakeCallback(5, "fmt:v:3", msg_id=77)) == "FakeCallback(sender_id=5, chat_id=5, message_id=77, data='fmt:v:3')"
+
+
+def test_real_telethon_events_are_described_too():
+    now = datetime.datetime.now(datetime.timezone.utc)
+    message = types.Message(id=5, peer_id=types.PeerUser(77), message="/mediainfo", date=now, from_id=types.PeerUser(77), out=False)
+    assert describe_update(events.NewMessage.Event(message)) == "NewMessage(sender_id=77, chat_id=77, message_id=5, text='/mediainfo')"
+    query = types.UpdateBotCallbackQuery(query_id=1, user_id=77, peer=types.PeerUser(77), msg_id=9, chat_instance=1, data=b"fmt:v:3")
+    assert describe_update(events.CallbackQuery.Event(query, query.peer, query.msg_id)) == (
+        "CallbackQuery(sender_id=77, chat_id=77, message_id=9, data='fmt:v:3')"
+    )
+
+
+def test_a_long_text_is_cut_and_an_odd_object_never_raises():
+    line = describe_update(FakeEvent(1, "x" * 500))
+    assert "…" in line and len(line) < 400
+
+    class Odd:
+        @property
+        def sender_id(self):
+            raise RuntimeError("no")
+
+    assert describe_update(Odd()) == "Odd"
+
+
+# ---------------------------------------------------------------------------
+# hiding secrets
+# ---------------------------------------------------------------------------
+
+
+def test_configured_secrets_are_hidden_but_short_values_are_left_alone():
+    assert redact("token abc123XYZ789 and abc", ["abc123XYZ789", "abc"]) == "token *** and abc"
+
+
+def test_session_strings_and_access_tokens_are_hidden():
+    session = "1" + "A1b2C3_-" * 40
+    telegraph = "b968da509bb76866c35425099bc0989a5ec3b32997d55286c657e6994bbb"
+    text = f"bound {session} and {telegraph} here"
+    assert redact(text) == "bound *** and *** here"
+
+
+def test_ordinary_text_survives():
+    for text in (
+        "File \"/home/user/app/bot/app.py\", line 1021, in _mediainfo",
+        "1147 - One Piece [Dual] 720p @Anime_X_Universe.mkv → One Piece - E1147.mkv (276.61 MB)",
+        "-" * 80,
+    ):
+        assert redact(text) == text
+
+
+def test_the_values_an_sql_error_prints_are_hidden():
+    text = "[SQL: INSERT INTO kv VALUES (?, ?)]\n[parameters: ('session', 'short')]\n(Background on this error)"
+    assert "short" not in redact(text) and "[SQL: INSERT INTO kv VALUES (?, ?)]" in redact(text)
+
+
+def test_secrets_come_from_the_configuration():
+    cfg = load_config({
+        "API_ID": "1", "API_HASH": "0123456789abcdef", "BOT_TOKEN": "123:botsecret", "OWNER_ID": "1",
+        "DATABASE_URL": "postgresql://user:pw123456@host/db",
+    })
+    secrets = errorlog.secrets_from(cfg)
+    assert {"0123456789abcdef", "123:botsecret", "pw123456", "postgresql://user:pw123456@host/db"} <= set(secrets)
+    assert "" not in secrets  # no SESSION_STRING set
+
+
+# ---------------------------------------------------------------------------
+# the report itself
+# ---------------------------------------------------------------------------
+
+
+def test_the_report_has_the_message_then_the_whole_traceback_in_a_python_block():
+    _signature, body = build_report(make_record("Error handling update:", chained))
+    assert body.startswith(CODE + "Error handling update:\n\nTraceback (most recent call last):") and body.endswith("</code></pre>")
+    text = visible(body)
+    assert "KeyError: 'missing'" in text and "During handling of the above exception, another exception occurred:" in text
+    assert text.endswith("RuntimeError: could not read <b>this</b> & that")
+    assert "&lt;b&gt;this&lt;/b&gt; &amp; that" in body  # escaped for Telegram's HTML
+
+
+def test_a_message_without_an_exception_is_just_the_message():
+    _signature, body = build_report(make_record("something odd: %s", None, 42))
+    assert body == f"{CODE}something odd: 42</code></pre>"
+
+
+def test_what_the_bot_was_working_on_comes_under_the_message():
+    token = errorlog.error_context.set("user 42 · job #7\nold.mkv → new.mkv (1.00 MB)")
+    try:
+        _signature, body = build_report(make_record("Error processing file:", fail))
+    finally:
+        errorlog.error_context.reset(token)
+    assert body.startswith(CODE + "Error processing file:\nuser 42 · job #7\nold.mkv → new.mkv (1.00 MB)\n\nTraceback")
+    assert "user 42" not in build_report(make_record("later", fail))[1]  # only while it is set
+
+
+def test_the_working_context_can_be_a_function_and_is_not_repeated():
+    token = errorlog.error_context.set(lambda: "FakeEvent(sender_id=1)")
+    try:
+        body = build_report(make_record("Error handling update:", fail))[1]
+        assert body.count("FakeEvent(sender_id=1)") == 1
+        assert build_report(make_record("FakeEvent(sender_id=1) failed", fail))[1].count("FakeEvent(sender_id=1)") == 1
+    finally:
+        errorlog.error_context.reset(token)
+
+
+def test_secrets_are_hidden_in_the_message_and_in_the_traceback():
+    body = build_report(make_record("login with TOP-SECRET-VALUE failed", lambda: fail("TOP-SECRET-VALUE")), ["TOP-SECRET-VALUE"])[1]
+    assert "TOP-SECRET-VALUE" not in body and body.count("***") >= 2
+
+
+def test_a_long_traceback_keeps_its_end_and_fits_in_one_message():
+    _signature, body = build_report(make_record("Error handling update:", very_deep))
+    text = visible(body)
+    assert errorlog._u16(text) <= errorlog.MAX_TEXT
+    assert text.startswith("Error handling update:\n\n…\n  File ")  # cut at the beginning of a line
+    assert text.endswith("ValueError: bottom")
+
+
+def test_emoji_count_double_towards_the_limit():
+    text = visible(build_report(make_record("Error handling update:", lambda: fail("🔥" * 3000)))[1])
+    assert errorlog._u16(text) <= errorlog.MAX_TEXT and text.endswith("🔥")
+
+
+def test_a_log_call_with_wrong_arguments_still_gives_a_report():
+    record = logging.LogRecord("bot.test", logging.ERROR, __file__, 1, "%d items", ("x",), None)  # what such a call makes
+    assert build_report(record)[1] == f"{CODE}%d items</code></pre>"
+
+
+def test_the_same_exception_on_the_same_line_is_the_same_error():
+    same = build_report(make_record("one", fail))[0], build_report(make_record("another message", fail))[0]
+    assert same[0] == same[1]
+    assert build_report(make_record("one", fail_elsewhere))[0] != same[0]
+    assert build_report(make_record("one", None))[0] != same[0]
+
+
+# ---------------------------------------------------------------------------
+# keeping repeats and floods out
+# ---------------------------------------------------------------------------
+
+
+class Clock:
+    def __init__(self):
+        self.now = 1000.0
+
+    def __call__(self):
+        return self.now
+
+
+@pytest.fixture
+def sink():
+    sent = []
+
+    async def send(text):
+        sent.append(text)
+
+    return SimpleNamespace(sent=sent, send=send, clock=Clock())
+
+
+async def test_an_identical_error_is_reported_once_and_counted(sink):
+    reporter = ErrorReporter(sink.send, clock=sink.clock)
+    for n in (1, 2, 3):
+        reporter.submit("same", f"<pre>{n}</pre>")
+    await reporter.flush()
+    assert sink.sent == ["<pre>1</pre>"]
+    sink.clock.now += 121
+    reporter.submit("same", "<pre>4</pre>")
+    await reporter.flush()
+    assert sink.sent[1] == "<pre>4</pre>\n🔁 2 more of this error since the last report."
+    sink.clock.now += 121
+    reporter.submit("same", "<pre>5</pre>")
+    await reporter.flush()
+    assert sink.sent[2] == "<pre>5</pre>"  # nothing came back in between: no note
+
+
+async def test_different_errors_all_go_out_in_the_order_they_happened(sink):
+    async def slow_first(text):
+        await asyncio.sleep(0.05 if text == "<pre>a</pre>" else 0)
+        await sink.send(text)
+
+    reporter = ErrorReporter(slow_first, clock=sink.clock)
+    for name in "abc":
+        reporter.submit(name, f"<pre>{name}</pre>")
+    await reporter.flush()
+    assert sink.sent == ["<pre>a</pre>", "<pre>b</pre>", "<pre>c</pre>"]
+
+
+async def test_a_flood_is_cut_off_and_mentioned_in_the_next_report(sink):
+    reporter = ErrorReporter(sink.send, per_minute=3, clock=sink.clock)
+    for n in range(5):
+        reporter.submit(f"e{n}", f"<pre>{n}</pre>")
+    await reporter.flush()
+    assert sink.sent == ["<pre>0</pre>", "<pre>1</pre>", "<pre>2</pre>"]
+    sink.clock.now += 61
+    reporter.submit("later", "<pre>later</pre>")
+    await reporter.flush()
+    assert sink.sent[3] == "<pre>later</pre>\n⚠️ 2 other error report(s) were left out to keep this chat readable."
+
+
+async def test_a_failing_send_does_not_stop_the_next_reports(sink, caplog):
+    calls = []
+
+    async def flaky(text):
+        calls.append(text)
+        if len(calls) == 1:
+            raise RuntimeError("CHAT_WRITE_FORBIDDEN")
+        await sink.send(text)
+
+    reporter = ErrorReporter(flaky, clock=sink.clock)
+    reporter.submit("a", "<pre>a</pre>")
+    reporter.submit("b", "<pre>b</pre>")
+    await reporter.flush()
+    assert sink.sent == ["<pre>b</pre>"]
+    assert "could not send an error report" in caplog.text and "CHAT_WRITE_FORBIDDEN" in caplog.text
+
+
+async def test_old_entries_are_forgotten(sink):
+    reporter = ErrorReporter(sink.send, per_minute=10_000, clock=sink.clock)
+    for n in range(150):
+        reporter.submit(f"old{n}", "<pre>x</pre>")
+    sink.clock.now += 500
+    for n in range(100):
+        reporter.submit(f"new{n}", "<pre>x</pre>")
+    assert len(reporter._seen) <= 101
+    await reporter.flush()
+
+
+# ---------------------------------------------------------------------------
+# the logging handler
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+async def channel(sink):
+    logger = logging.getLogger("errorlog.test.channel")
+    logger.propagate = False
+    logger.setLevel(logging.DEBUG)
+    handler = errorlog.install(sink.send, ["TOP-SECRET-VALUE"], logger=logger)
+    yield SimpleNamespace(log=logger, sent=sink.sent, handler=handler)
+    handler.detach()
+
+
+async def test_an_error_with_a_traceback_arrives_as_one_python_block(channel):
+    try:
+        fail("disk is full")
+    except ValueError:
+        channel.log.exception("Error processing file:")
+    await channel.handler.aclose()
+    (report,) = channel.sent
+    assert report.startswith(CODE + "Error processing file:\n\nTraceback (most recent call last):")
+    assert report.endswith("ValueError: disk is full</code></pre>")
+
+
+async def test_only_errors_are_reported(channel):
+    channel.log.info("fine")
+    channel.log.warning("careful")
+    channel.log.error("broken")
+    channel.log.critical("very broken")
+    await channel.handler.aclose()
+    assert channel.sent == [f"{CODE}broken</code></pre>", f"{CODE}very broken</code></pre>"]
+
+
+async def test_errors_logged_from_another_thread_arrive_too(channel):
+    await asyncio.to_thread(channel.log.error, "from a worker thread")
+    await channel.handler.aclose()
+    assert channel.sent == [f"{CODE}from a worker thread</code></pre>"]
+
+
+async def test_the_working_context_is_added_to_errors_logged_below_it(channel):
+    token = errorlog.error_context.set("user 42 · job #7")
+    try:
+        channel.log.error("helper failed")
+    finally:
+        errorlog.error_context.reset(token)
+    await channel.handler.aclose()
+    assert channel.sent == [f"{CODE}helper failed\nuser 42 · job #7</code></pre>"]
+
+
+async def test_secrets_never_leave_the_bot(channel):
+    channel.log.error("the key TOP-SECRET-VALUE was refused")
+    await channel.handler.aclose()
+    assert channel.sent == [f"{CODE}the key *** was refused</code></pre>"]
+
+
+async def test_closing_stops_the_reports(channel):
+    await channel.handler.aclose()
+    channel.log.error("too late")
+    await asyncio.sleep(0.01)
+    assert channel.sent == []
+
+
+async def test_the_reporters_own_trouble_is_not_reported(sink):
+    handler = errorlog.install(sink.send)  # the root logger, as the bot does
+    try:
+        logging.getLogger("bot.errorlog").error("this would loop")
+        logging.getLogger("bot.other").error("this is reported")
+        await handler.aclose()
+    finally:
+        handler.detach()
+    assert sink.sent == [f"{CODE}this is reported</code></pre>"]
+
+
+def test_a_closed_event_loop_does_not_break_logging(sink):
+    loop = asyncio.new_event_loop()
+    loop.close()
+    logger = logging.getLogger("errorlog.test.closed")
+    logger.propagate = False
+    handler = ChannelLogHandler(ErrorReporter(sink.send), loop)
+    logger.addHandler(handler)
+    try:
+        logger.error("nobody is listening")  # must not raise
+    finally:
+        logger.removeHandler(handler)

tests/test_pipeline.py

@@ -5,6 +5,7 @@ from types import SimpleNamespace
 import pytest
 from telethon import errors, types
 
+from bot import errorlog
 from bot.config import load_config
 from bot.db import Database
 from bot.media import probe
@@ -69,12 +70,7 @@ async def env(tmp_path, media_dir):
     })
     db = Database(f"sqlite+aiosqlite:///{tmp_path}/p.db")
     await db.init()
-    logs = []
-
-    async def log_send(text):
-        logs.append(text)
-
-    ns = SimpleNamespace(cfg=cfg, db=db, logs=logs, log_send=log_send, media_dir=media_dir, tmp=tmp_path)
+    ns = SimpleNamespace(cfg=cfg, db=db, media_dir=media_dir, tmp=tmp_path)
     yield ns
     await db.close()
 
@@ -83,7 +79,7 @@ def make(env, name="[SubsPlease] Tokyo Revengers - 05 (1080p) [ABCD1234].mkv", s
     source = env.media_dir / src
     message = FakeMessage(name, source.stat().st_size, **msg_kwargs)
     client = FakeClient(source, message)
-    renamer = Renamer(client, env.cfg, env.db, env.log_send)
+    renamer = Renamer(client, env.cfg, env.db)
     new_name, _ = compute_new_name(template=DEFAULT_TEMPLATE, mode="filename", original_name=name, original_ext=Path(name).suffix)
     job = Job(id=1, user_id=42, chat_id=42, msg_id=7, file_name=name, size=source.stat().st_size, new_name=new_name,
               status_msg=SimpleNamespace(id=99))
@@ -111,7 +107,6 @@ async def test_plain_rename_sends_video_with_thumbnail(env):
     assert job.state == "done"
     assert (await env.db.totals())[0] == 1
     assert not (Path(env.cfg.temp_dir) / "job1").exists()  # temp files cleaned
-    assert any("Tokyo Revengers" in t for t in env.logs)
 
 
 @needs_ffmpeg
@@ -164,7 +159,7 @@ async def test_forced_video_for_non_video_falls_back_to_document(env, tmp_path):
     blob.write_bytes(b"PK" + b"0" * 5000)
     message = FakeMessage("stuff.zip", 5002, video=False, mime="application/zip")
     client = FakeClient(blob, message)
-    renamer = Renamer(client, env.cfg, env.db, env.log_send)
+    renamer = Renamer(client, env.cfg, env.db)
     await env.db.update_settings(42, file_type="video")
     job = Job(id=2, user_id=42, chat_id=42, msg_id=7, file_name="stuff.zip", size=5002, new_name="Stuff.zip", status_msg=object())
     await renamer.process(job)
@@ -215,7 +210,7 @@ async def test_unsupported_container_with_metadata_warns_but_still_renames(env,
     client = FakeClient(avi, message)
     await env.db.update_settings(42, metadata_on=True, meta={"title": "T"})
     job = Job(id=3, user_id=42, chat_id=42, msg_id=7, file_name="x.avi", size=avi.stat().st_size, new_name="Y.avi", status_msg=object())
-    await Renamer(client, env.cfg, env.db, env.log_send).process(job)
+    await Renamer(client, env.cfg, env.db).process(job)
     assert client.sent and "not supported" in client.edits[-1][0]
 
 
@@ -239,17 +234,41 @@ async def test_cancel_before_start(env):
     assert not (Path(env.cfg.temp_dir) / "job1").exists()
 
 
+@pytest.fixture
+async def log_channel():
+    """What reaches the log channel while a test runs (the error reporter is attached to the root logger, as in main.py)."""
+    reports = []
+
+    async def send(text):
+        reports.append(text)
+
+    handler = errorlog.install(send)
+    yield SimpleNamespace(reports=reports, flush=handler.aclose)
+    handler.detach()
+
+
 @needs_ffmpeg
-async def test_failure_is_reported_and_logged(env):
+async def test_failure_is_reported_and_logged(env, log_channel):
     client, renamer, job = make(env)
     client.download_error = RuntimeError("network down")
     await renamer.process(job)
+    await log_channel.flush()
     assert job.state == "failed" and client.sent == []
     assert "Failed" in client.edits[-1][0] and "network down" in client.edits[-1][0]
-    assert any("Job failed" in t and "network down" in t for t in env.logs)
+    (report,) = log_channel.reports  # the log channel gets the whole traceback, as a Python code block
+    assert report.startswith(f'<pre><code class="language-python">Error processing file:\nuser 42 · job #1\n{job.file_name} → {job.new_name} (')
+    assert "Traceback (most recent call last):" in report and report.endswith("RuntimeError: network down</code></pre>")
     assert not (Path(env.cfg.temp_dir) / "job1").exists()
 
 
+@needs_ffmpeg
+async def test_a_finished_job_leaves_nothing_in_the_log_channel(env, log_channel):
+    client, renamer, job = make(env)
+    await renamer.process(job)
+    await log_channel.flush()
+    assert job.state == "done" and log_channel.reports == []
+
+
 @needs_ffmpeg
 async def test_not_enough_disk_space(env, monkeypatch):
     client, renamer, job = make(env)

tests/test_web_main.py

@@ -1,5 +1,6 @@
 import asyncio
 import dataclasses
+import logging
 import socket
 from types import SimpleNamespace
 
@@ -13,6 +14,7 @@ import bot.main as bot_main
 from bot import __version__
 from bot.config import load_config
 from bot.db import Database
+from bot.errorlog import ChannelLogHandler
 from bot.web import make_app, start_web
 
 from .tgfakes import FakeClient
@@ -187,6 +189,14 @@ async def test_amain_wires_everything_and_exits_nonzero_if_telegram_disconnects(
         assert any(isinstance(r, functions.bots.SetBotCommandsRequest) for r in client.requests)
         assert not (tmp_path / "work" / "job9").exists()
         assert client.sent[0][0] == -1001111111111
+        assert any(isinstance(h, ChannelLogHandler) for h in logging.getLogger().handlers)
+        logging.getLogger("bot.something").error("an error somewhere in the bot")  # errors go to the log channel as well
+        for _ in range(200):
+            if any("an error somewhere in the bot" in text for _chat, text, _kw in client.sent):
+                break
+            await asyncio.sleep(0.05)
+        else:
+            pytest.fail("the error never reached the log channel")
         gate.set()  # Telegram connection lost -> exit non-zero so the container restarts
         assert await asyncio.wait_for(task, 10) == 1
     finally:
@@ -194,6 +204,7 @@ async def test_amain_wires_everything_and_exits_nonzero_if_telegram_disconnects(
         if not task.done():
             task.cancel()
     assert client.disconnected
+    assert not any(isinstance(h, ChannelLogHandler) for h in logging.getLogger().handlers)  # detached again on the way out
     with pytest.raises(aiohttp.ClientError):
         async with aiohttp.ClientSession() as http:
             async with http.get(f"http://127.0.0.1:{port}/health"):