several bug fixes

.gitignore

@@ -24,3 +24,6 @@ docs/
 .claude/
 .agents/
 skills-lock.json
+
+# logs
+*.log

CHANGELOG.md

@@ -5,6 +5,68 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.1.4] - 2026-08-17
+
+### Added
+
+- New `rate_limited` counter in `HandlerStats`: 429 `Retry-After` waits are
+  now counted even though they never consume the retry budget, so a
+  rate-limit storm shows up in the stats instead of reporting zero retries.
+
+### Fixed
+
+- Failed handler construction no longer leaves a half-built handler registered
+  with the logging module. Previously, exit-time `logging.shutdown()` called
+  `close()` on the half-built instance and printed a stray `AttributeError`
+  after the real construction error; the registration is now dropped on
+  failure. The cleanup itself is failure-proof (`contextlib.suppress`), so a
+  teardown error (e.g. logging internals already torn down at interpreter
+  shutdown) can never mask the original construction error.
+- Live smoke tests no longer race the worker: `close()` returns once the
+  `shutdown_timeout` elapses even while the daemon worker is still sending its
+  last batches, so the tests now poll the stats until the counters converge
+  before asserting. A new live test proves 10 records with `batch_size=5` go
+  out as exactly two 5-record messages (`batches_sent == 2`).
+- Worker shutdown is event-backed: a saturated queue or a `drop_oldest`
+  eviction can no longer lose the shutdown sentinel and leak the worker
+  thread (and its HTTP client) after `close()`.
+- **Shutdown no longer waits out a long `flush_interval`**: the accumulator
+  polls the shutdown event between capped waits, so `close()` returns promptly
+  (≤ ~1s) even with a partial batch pending and the sentinel lost; the pending
+  batch is still flushed first.
+- A close() that interleaves with an in-flight `emit()` no longer orphans the
+  record: `emit` and `close` are mutually exclusive, so nothing accepted
+  before `close()` is lost (and `queue_full_policy="block"` can no longer hang
+  an emitter past shutdown).
+- Post-close `emit()` drops and counts the record instead of potentially
+  hanging the caller under `queue_full_policy="block"`.
+- `truncate_text` clamps the marker when the limit is smaller than the marker
+  itself, so a truncated message can never exceed the cap (Telegram 400s
+  oversized payloads).
+- The one-time 429 heads-up notification now swallows any failure (not just
+  transport errors), so it can never surface through `send_with_retry`.
+- **Batches are no longer split by a pending shutdown.** The shutdown poll now
+  runs only after a capped wait elapsed with nothing arriving, so records that
+  keep flowing still group into one batch even when `close()` was called during
+  the drain (previously every remaining record flushed as its own message).
+- A non-200 `sendMessage` response (a 3xx — redirects are never followed — or
+  a non-200 2xx) is no longer reported as delivered; getMe validation
+  maps non-JSON 200 bodies to `TelegramConfigError` instead of a bare
+  `ValueError`; env `TG_CHAT_ID` trailing whitespace is stripped like the
+  token's; queue `task_done` accounting is now exact, so `queue.join()` can
+  no longer hang.
+- `close()` now enqueues the shutdown sentinel **before** signaling the
+  worker, so the worker can never exit while the sentinel is still about to be
+  queued and strand it in the queue (a user's `queue.join()` would hang).
+- getMe validation requires an object body with `ok == True` exactly: a
+  scalar/list JSON body or a merely truthy `ok` (e.g. `1`) maps to
+  `TelegramConfigError` instead of a raw `AttributeError` or a false accept.
+- The clamped truncation marker is nudged off a trailing backslash, so a tiny
+  `max_length` with MarkdownV2 can never leave a dangling escape pair.
+- HTTP redirects are disabled on the send client: a 302/308 with a `Location`
+  is never followed, so a redirected `sendMessage` request can never be
+  mistaken for a delivery.
+
 ## [0.1.3] - 2026-08-16
 
 ### Added

localtest.bat

@@ -0,0 +1,9 @@
+@echo off
+(
+  uv run mypy || exit /b 1
+  uv run ruff format --check . || exit /b 1
+  uv run ruff check . || exit /b 1
+  uv run pylint tests tg_logging_handler || exit /b 1
+  uv run flake8 tests tg_logging_handler || exit /b 1
+  uv run pytest --cov=tg_logging_handler || exit /b 1
+) > local.log 2>&1

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "tg-logging-handler"
-version = "0.1.3"
+version = "0.1.4"
 description = "Production-grade Telegram destination for Python's standard logging module"
 readme = "README.md"
 requires-python = ">=3.10"

tests/test_batching.py

@@ -109,3 +109,68 @@ def test_constructor_validates_args() -> None:
         BatchAccumulator(0, 1.0)
     with pytest.raises(ValueError):
         BatchAccumulator(1, -1.0)
+
+
+class _TimeoutQueue:
+    """Scripted queue whose get() always times out (raises Empty)."""
+
+    def __init__(self) -> None:
+        self._items: list[object] = []
+
+    def put_nowait(self, item: object) -> None:
+        self._items.append(item)
+
+    def get(self, timeout: float | None = None) -> object:
+        if not self._items:
+            raise queue.Empty
+        return self._items.pop(0)
+
+    def task_done(self) -> None:
+        pass
+
+
+def test_second_get_timeout_returns_partial_batch() -> None:
+    # The mid-batch get() can time out with nothing new arriving; the partial
+    # batch must be returned, not lost. Constant clock keeps remaining > 0 so
+    # the get() is actually attempted (and times out immediately, scripted).
+    acc, _ = _mk(batch_size=5, flush_interval=1.0, clock=ScriptedClock(0.0, 0.0))
+    q = _TimeoutQueue()
+    q.put_nowait(_record("only"))
+    batch, shutdown = acc.collect(q, Shutdown, timeout=0)  # type: ignore[arg-type]
+    assert [r.getMessage() for r in batch] == ["only"]
+    assert not shutdown
+
+
+def test_stop_flushes_partial_batch_promptly() -> None:
+    # A shutdown requested while a partial batch is mid-collection must flush
+    # it immediately, not wait out the flush interval (potentially minutes);
+    # collect polls the stop predicate after a capped wait with nothing
+    # arriving.
+    acc, q = _mk(batch_size=10, flush_interval=30.0)
+    _put(q, "only")
+    batch, shutdown = acc.collect(q, Shutdown, timeout=0, stop=lambda: True)
+    assert [r.getMessage() for r in batch] == ["only"]
+    assert shutdown is False
+
+
+def test_stop_does_not_split_a_flowing_batch() -> None:
+    # Live-smoke regression: close() sets the shutdown event while the worker
+    # is still draining a queue full of records. stop must only be consulted
+    # once a wait actually elapsed with nothing arriving, or every flowing
+    # record flushes as its own batch. stop=True here must not split the 5
+    # queued records (batch_size is exactly 5, so they must come back whole).
+    acc, q = _mk(batch_size=5, flush_interval=10.0)
+    _put(q, "a", "b", "c", "d", "e")
+    batch, shutdown = acc.collect(q, Shutdown, timeout=0, stop=lambda: True)
+    assert [r.getMessage() for r in batch] == ["a", "b", "c", "d", "e"]
+    assert not shutdown
+
+
+def test_stop_false_after_capped_wait_keeps_collecting() -> None:
+    # Stop present but still False when a capped wait elapses: the loop must
+    # keep going (re-check the interval) instead of aborting an active batch.
+    acc, q = _mk(batch_size=10, flush_interval=5.0, clock=ScriptedClock(0.0, 0.0, 0.0, 5.0))
+    _put(q, "a", "b")
+    batch, shutdown = acc.collect(q, Shutdown, timeout=0, stop=lambda: False)
+    assert [r.getMessage() for r in batch] == ["a", "b"]
+    assert not shutdown

tests/test_config_stats.py

@@ -52,6 +52,14 @@ def test_trailing_newline_token_is_stripped(self) -> None:
         # must not fail validation or leak into the request URL.
         assert resolve_credentials("111:abc\n", 1) == ("111:abc", "1")
 
+    def test_env_chat_id_trailing_newline_is_stripped(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        # Same story for chats: .env files routinely end values with a
+        # newline, and an unstripped one would 400 at send time.
+        monkeypatch.setenv("TG_CHAT_ID", "-42\n")
+        assert resolve_credentials("111:abc", None)[1] == "-42"
+
     def test_invalid_short_token_error_is_redacted(self) -> None:
         # A token too short for a safe partial reveal must not appear in the
         # error message at all.
@@ -163,6 +171,73 @@ def get_nowait(self) -> object:
             raise queue.Empty
         return "evicted"
 
+    def task_done(self) -> None:
+        pass
+
+
+class TestValidateToken:
+    def test_non_json_200_body_is_config_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # A proxy/gateway can answer 200 with an HTML error page; the docs
+        # promise TelegramConfigError, never a bare ValueError.
+        import httpx
+
+        from tg_logging_handler.config import validate_token
+
+        monkeypatch.setattr(
+            httpx,
+            "get",
+            lambda *args, **kwargs: httpx.Response(200, text="<html>gateway error</html>"),
+        )
+        with pytest.raises(TelegramConfigError, match="non-JSON"):
+            validate_token("111:abc", "https://api.telegram.org")
+
+    def test_network_error_is_config_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # An unreachable API (offline host, DNS failure) must surface as the
+        # documented TelegramConfigError, with the offline-escape hatch hint.
+        import httpx
+
+        from tg_logging_handler.config import validate_token
+
+        def offline(*args: object, **kwargs: object) -> httpx.Response:
+            raise httpx.ConnectError("offline")
+
+        monkeypatch.setattr(httpx, "get", offline)
+        with pytest.raises(TelegramConfigError, match="Could not reach Telegram"):
+            validate_token("111:abc", "https://api.telegram.org")
+
+    @pytest.mark.parametrize("payload", [[], "hello", 42])
+    def test_non_object_json_200_body_is_config_error(
+        self, monkeypatch: pytest.MonkeyPatch, payload: object
+    ) -> None:
+        # A 200 that decodes as a scalar/list (never an object) cannot carry
+        # ok; it must hit the TelegramConfigError path, not an AttributeError.
+        import httpx
+
+        from tg_logging_handler.config import validate_token
+
+        monkeypatch.setattr(
+            httpx,
+            "get",
+            lambda *args, **kwargs: httpx.Response(200, json=payload),
+        )
+        with pytest.raises(TelegramConfigError, match="rejected"):
+            validate_token("111:abc", "https://api.telegram.org")
+
+    def test_truthy_but_not_true_ok_is_config_error(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        # ok == 1 is truthy but not exactly True; only a boolean True means
+        # the token was accepted.
+        import httpx
+
+        from tg_logging_handler.config import validate_token
+
+        monkeypatch.setattr(
+            httpx,
+            "get",
+            lambda *args, **kwargs: httpx.Response(200, json={"ok": 1}),
+        )
+        with pytest.raises(TelegramConfigError, match="rejected"):
+            validate_token("111:abc", "https://api.telegram.org")
+
 
 class TestQueuePolicy:
     def test_put_drop_newest_succeeds_when_space(self) -> None:
@@ -190,6 +265,16 @@ def test_put_drop_oldest_evicts_to_make_room(self) -> None:
         assert result.dropped == 1  # evicted the oldest (1)
         assert list(q.queue) == [2, 3]
 
+    def test_put_drop_oldest_task_dones_evicted_items(self) -> None:
+        # Every eviction is a get(); without a matching task_done(),
+        # queue.join() would hang forever on a full-and-drained queue.
+        q: queue.Queue[object] = queue.Queue(maxsize=2)
+        q.put_nowait(1)
+        q.put_nowait(2)
+        queue_policy.put_drop_oldest(q, 3)
+        # Only items still in the queue may lack a task_done().
+        assert q.unfinished_tasks == q.qsize() == 2
+
     def test_put_drop_oldest_no_eviction_when_space(self) -> None:
         q: queue.Queue[object] = queue.Queue(maxsize=2)
         q.put_nowait(1)