refactor: reduce cross-file duplication (round 2)

tests/test_queue_kafka_dead_letter_deterministic.py

@@ -86,3 +86,43 @@ def test_reject_dead_letter_is_observable(self):
             assert job_id in ids, f"rejected job {job_id} not observable (got {ids})"
         finally:
             q.close()
+
+    def test_retry_job_clears_the_dead_letter_and_reports_true(self):
+        """retry(job_id) revives a dead-lettered job: it returns True and the
+        job leaves the dead-letter topic.
+
+        This exercises the shared TopicDeadLetterMixin.retry_job() on live
+        Kafka — the same method the RabbitMQ backend inherits. (Whether the
+        re-produced record is then re-consumed is a Kafka offset concern, not
+        part of this method's contract, so it is not asserted here.)
+        """
+        q = _make_queue()
+        try:
+            job_id = q.push({"task": "revive-me"})
+            job = q.pop()
+            assert job is not None
+            job.reject("dead for now")
+            assert job_id in [d.id for d in q.dead_letters()], "prime: not dead-lettered"
+
+            assert q.retry(job_id) is True, "retry() must report it revived the job"
+            assert job_id not in [d.id for d in q.dead_letters()], (
+                "revived job must no longer be in the dead-letter topic"
+            )
+        finally:
+            q.close()
+
+    def test_retry_job_unknown_id_returns_false(self):
+        """retry() on an id that is not dead-lettered reports False, without
+        disturbing the real dead letter."""
+        q = _make_queue()
+        try:
+            job_id = q.push({"task": "stays-dead"})
+            job = q.pop()
+            assert job is not None
+            job.reject("poison")
+            assert q.retry("no-such-id") is False
+            assert job_id in [d.id for d in q.dead_letters()], (
+                "a failed retry lookup must leave the real dead letter in place"
+            )
+        finally:
+            q.close()

tests/test_rate_limiter.py

@@ -147,3 +147,69 @@ def test_cleanup_keeps_active_ips(self, monkeypatch):
         now = time.monotonic()
         rl._cleanup(now)
         assert "10.0.0.1" in rl._requests
+
+
+class TestRateLimiterEnforcement:
+    """Lock the enforcement entry points, not just check()/apply_headers().
+
+    RateLimiter.apply() holds the enforcement path, and the two before_rate_limit
+    staticmethods (the class-based middleware entry points) delegate to it. These
+    entry points had no coverage before, so a refactor could quietly stop
+    returning the 429 with nothing going red. Real Response object, plain request
+    holder — no doubles.
+    """
+
+    def _request(self, ip="10.0.0.1"):
+        import types
+        return types.SimpleNamespace(ip=ip)
+
+    def test_apply_allows_under_limit(self, monkeypatch):
+        monkeypatch.setenv("TINA4_RATE_LIMIT", "3")
+        monkeypatch.setenv("TINA4_RATE_WINDOW", "60")
+        rl = RateLimiter()
+        resp = Response()
+        _, out = rl.apply(self._request(), resp)
+        assert out.status_code == 200
+        assert dict(resp._headers).get("x-ratelimit-limit") == "3"
+
+    def test_apply_refuses_over_limit_with_429_and_retry_after(self, monkeypatch):
+        monkeypatch.setenv("TINA4_RATE_LIMIT", "3")
+        monkeypatch.setenv("TINA4_RATE_WINDOW", "60")
+        rl = RateLimiter()
+        req = self._request()
+        statuses = []
+        for _ in range(4):
+            resp = Response()
+            _, out = rl.apply(req, resp)
+            statuses.append(out.status_code)
+        assert statuses == [200, 200, 200, 429]
+        # the refused response carries retry-after and the rate-limit headers
+        assert dict(resp._headers).get("retry-after") is not None
+        assert dict(resp._headers).get("x-ratelimit-remaining") == "0"
+
+    def test_shared_before_rate_limit_refuses_over_limit(self, monkeypatch):
+        monkeypatch.setenv("TINA4_RATE_LIMIT", "2")
+        monkeypatch.setenv("TINA4_RATE_WINDOW", "60")
+        RateLimiter._shared_instance = None
+        req = self._request("10.0.0.2")
+        statuses = []
+        for _ in range(3):
+            resp = Response()
+            _, out = RateLimiter.before_rate_limit(req, resp)
+            statuses.append(out.status_code)
+        RateLimiter._shared_instance = None
+        assert statuses == [200, 200, 429]
+
+    def test_middleware_before_rate_limit_refuses_over_limit(self, monkeypatch):
+        from tina4_python.core.middleware import RateLimiterMiddleware
+        monkeypatch.setenv("TINA4_RATE_LIMIT", "2")
+        monkeypatch.setenv("TINA4_RATE_WINDOW", "60")
+        RateLimiterMiddleware._limiter = None
+        req = self._request("10.0.0.3")
+        statuses = []
+        for _ in range(3):
+            resp = Response()
+            _, out = RateLimiterMiddleware.before_rate_limit(req, resp)
+            statuses.append(out.status_code)
+        RateLimiterMiddleware._limiter = None
+        assert statuses == [200, 200, 429]

tina4_python/core/middleware.py

@@ -528,18 +528,7 @@ def _get_limiter(cls):
     @staticmethod
     def before_rate_limit(request, response):
         """Middleware hook — enforces rate limiting before the route handler."""
-        limiter = RateLimiterMiddleware._get_limiter()
-        ip = getattr(request, "ip", None) or "unknown"
-        allowed, info = limiter.check(ip)
-        limiter.apply_headers(response, info)
-        if not allowed:
-            retry_after = max(1, int(info.get("reset", limiter.window)))
-            response.header("retry-after", str(retry_after))
-            if hasattr(response, "error"):
-                response.error("Too Many Requests", f"Rate limit exceeded. Retry in {retry_after}s.", 429)
-            else:
-                setattr(response, "status_code", 429)
-        return request, response
+        return RateLimiterMiddleware._get_limiter().apply(request, response)
 
     @staticmethod
     def check(ip: str):

tina4_python/core/rate_limiter.py

@@ -37,18 +37,7 @@ def _shared(cls) -> "RateLimiter":
     @staticmethod
     def before_rate_limit(request, response):
         """Class-based middleware entry point — enforces the shared rate limit."""
-        limiter = RateLimiter._shared()
-        ip = getattr(request, "ip", None) or "unknown"
-        allowed, info = limiter.check(ip)
-        limiter.apply_headers(response, info)
-        if not allowed:
-            retry_after = max(1, int(info.get("reset", limiter.window)))
-            response.header("retry-after", str(retry_after))
-            if hasattr(response, "error"):
-                response.error("Too Many Requests", f"Rate limit exceeded. Retry in {retry_after}s.", 429)
-            else:
-                setattr(response, "status_code", 429)
-        return request, response
+        return RateLimiter._shared().apply(request, response)
 
     def __init__(self):
         self._env_snapshot: tuple = ()

tina4_python/core/server.py

@@ -1181,7 +1181,7 @@ async def dash(req, res):
 
 
 # ── WebSocket support ──────────────────────────────────────────
-from tina4_python.websocket import CLOSE_GOING_AWAY, WebSocketConnection, WebSocketManager
+from tina4_python.websocket import CLOSE_GOING_AWAY, RoomMemberMixin, WebSocketConnection, WebSocketManager
 
 _ws_manager = WebSocketManager()
 
@@ -1309,7 +1309,7 @@ async def _handle_asgi_websocket(scope: dict, receive, send):
         _ws_manager.remove(conn)
 
 
-class _AsgiWebSocketConnection:
+class _AsgiWebSocketConnection(RoomMemberMixin):
     """WebSocket connection wrapper for ASGI servers (uvicorn, etc.).
 
     Supports both Router's (conn, event, data) style and WebSocketServer's
@@ -1344,10 +1344,8 @@ def __init__(self, scope, receive, send, path, params, manager):
     def closed(self) -> bool:
         return self._closed
 
-    @property
-    def rooms(self) -> set:
-        """Return the set of room names this connection has joined."""
-        return self._rooms
+    # rooms / join_room / leave_room / broadcast_to_room come from
+    # RoomMemberMixin — transport-agnostic, shared with WebSocketConnection.
 
     def on_message(self, handler):
         """Register a message handler (decorator style)."""
@@ -1361,25 +1359,6 @@ def on_error(self, handler):
         """Register an error handler (decorator style)."""
         self._on_error = handler
 
-    def join_room(self, room_name: str) -> None:
-        """Join a named room."""
-        self._rooms.add(room_name)
-        if self._manager:
-            self._manager._join_room(self.id, room_name)
-
-    def leave_room(self, room_name: str) -> None:
-        """Leave a named room."""
-        self._rooms.discard(room_name)
-        if self._manager:
-            self._manager._leave_room(self.id, room_name)
-
-    async def broadcast_to_room(self, room_name: str, message: str | bytes,
-                                 exclude_self: bool = False) -> None:
-        """Broadcast a message to all connections in a room."""
-        if self._manager:
-            exclude = self.id if exclude_self else None
-            await self._manager.broadcast_to_room(room_name, message, exclude=exclude)
-
     async def send(self, message: str | bytes):
         """Send a text or binary message."""
         if self._closed:

tina4_python/database/adapter.py

@@ -977,6 +977,22 @@ def _total_from_page(row_count: int, limit, offset, paginated: bool) -> int | No
             return 0
         return None
 
+    def _paginate_sql(self, sql: str, params, limit, offset, paginated: bool):
+        """Append this engine's LIMIT/OFFSET clause when ``paginated``.
+
+        Returns ``(sql, params)``. Each adapter decides ``paginated`` for
+        itself (SQLite/PostgreSQL also exclude writes), but the append is the
+        same everywhere: this engine's placeholder (``?`` or ``%s``) for the
+        two bound values. The clause goes on a NEW LINE — appended inline it
+        can land inside a trailing ``-- comment`` and be swallowed, the same
+        bug at the append site rather than the detector.
+        """
+        if not paginated:
+            return sql, params or []
+        marker = self.PARAM_MARKER
+        return (f"{sql}\nLIMIT {marker} OFFSET {marker}",
+                (params or []) + [limit, offset])
+
     @staticmethod
     def _strip_trailing_order_by(sql: str) -> str:
         """Strip a trailing top-level ``ORDER BY`` so the SQL can be safely

tina4_python/database/mysql.py

@@ -166,12 +166,7 @@ def fetch(self, sql: str, params: list = None,
         # a syntax error MEASURED on a live PostgreSQL. It worked on sqlite and
         # crashed on the server, which is the swap ADR-0024 exists to protect.
         paginated = not (limit is None or limit <= 0 or self._has_trailing_limit(sql))
-        if not paginated:
-            paginated_sql = sql
-            paginated_params = params or []
-        else:
-            paginated_sql = f"{sql}\nLIMIT %s OFFSET %s"
-            paginated_params = (params or []) + [limit, offset]
+        paginated_sql, paginated_params = self._paginate_sql(sql, params, limit, offset, paginated)
         cursor.execute(paginated_sql, paginated_params)  # FAILS LOUD
         rows = [dict(row) for row in cursor.fetchall()]
 

tina4_python/database/postgres.py

@@ -440,12 +440,7 @@ def fetch(self, sql: str, params: list = None,
         # a syntax error MEASURED on a live PostgreSQL. It worked on sqlite and
         # crashed on the server, which is the swap ADR-0024 exists to protect.
         paginated = not (is_write or limit is None or limit <= 0 or self._has_trailing_limit(sql))
-        if not paginated:
-            paginated_sql = sql
-            paginated_params = params or []
-        else:
-            paginated_sql = f"{sql}\nLIMIT %s OFFSET %s"
-            paginated_params = (params or []) + [limit, offset]
+        paginated_sql, paginated_params = self._paginate_sql(sql, params, limit, offset, paginated)
         self._exec_with_handling(cursor, paginated_sql, paginated_params)
         columns = [d[0] for d in cursor.description] if cursor.description else []
         indexes = range(len(columns))

tina4_python/database/sqlite.py

@@ -235,15 +235,7 @@ def fetch(self, sql: str, params: list = None,
         # never paginated and never COUNT-probed (a probe repeats the write).
         is_write = self._is_write_statement(sql)
         paginated = not (is_write or limit is None or limit <= 0 or self._has_trailing_limit(sql))
-        if not paginated:
-            paginated_sql = sql
-            paginated_params = params or []
-        else:
-            # The clause goes on a NEW LINE. Appended inline it lands INSIDE a
-            # trailing `-- comment` and is swallowed, which is the same bug at
-            # the append site rather than the detector.
-            paginated_sql = f"{sql}\nLIMIT ? OFFSET ?"
-            paginated_params = (params or []) + [limit, offset]
+        paginated_sql, paginated_params = self._paginate_sql(sql, params, limit, offset, paginated)
         # Hydrate via a tuple cursor + a column list computed ONCE, rather than
         # dict(sqlite3.Row) per row. The connection's row_factory is sqlite3.Row,
         # which builds a Row object per row that dict() then copies -- two

tina4_python/orm/fields.py

@@ -622,6 +622,27 @@ def __init__(self, to=None, related_name: str = None, **kwargs):
 _LAZY_PAGE_SIZE = 1000
 
 
+def fetch_all_pages(db, sql: str, params: list, start_offset: int = 0) -> list:
+    """Return every row of ``sql``, paging in blocks of ``_LAZY_PAGE_SIZE``.
+
+    Uncapped by design (REL-EAGER-UNBOUNDED): a relationship load that passed a
+    silent ``limit=1000`` truncated the tail with no signal, so this pages until
+    a short block instead. Shared by the descriptor lazy-load and the imperative
+    ``has_many`` so both revive the full set the same way. Returns raw rows; the
+    caller wraps them in the related model.
+    """
+    records = []
+    offset = start_offset
+    while True:
+        result = db.fetch(sql, params, limit=_LAZY_PAGE_SIZE, offset=offset)
+        batch = result.records
+        records.extend(batch)
+        if len(batch) < _LAZY_PAGE_SIZE:
+            break
+        offset += _LAZY_PAGE_SIZE
+    return records
+
+
 class RelationshipDescriptor:
     """Base descriptor for ORM relationships. Lazy-loads on first access."""
 
@@ -698,15 +719,7 @@ def _load(self, obj):
         sql = f"SELECT * FROM {table} WHERE {where} ORDER BY {order_col}"
         # REL-EAGER-UNBOUNDED: page through ALL children rather than silently
         # truncating at a fixed cap.
-        records = []
-        offset = 0
-        while True:
-            result = db.fetch(sql, [pk_value], limit=_LAZY_PAGE_SIZE, offset=offset)
-            batch = result.records
-            records.extend(batch)
-            if len(batch) < _LAZY_PAGE_SIZE:
-                break
-            offset += _LAZY_PAGE_SIZE
+        records = fetch_all_pages(db, sql, [pk_value])
         return [related_cls(row) for row in records]
 
 

tina4_python/orm/model.py

@@ -1506,7 +1506,7 @@ def has_many(self, related_class, foreign_key: str = None, limit: int = None, of
         row count whether accessed imperatively or lazily. An explicit ``limit``
         still pages (explicit, never silent).
         """
-        from tina4_python.orm.fields import _LAZY_PAGE_SIZE
+        from tina4_python.orm.fields import fetch_all_pages
         pk = self._get_pk()
         pk_value = getattr(self, pk)
         fk = foreign_key or f"{self.__class__.__name__.lower()}_id"
@@ -1522,15 +1522,7 @@ def has_many(self, related_class, foreign_key: str = None, limit: int = None, of
             return [related_class(row) for row in result.records]
 
         # No explicit limit -> page through ALL rows (uncapped, parity with lazy).
-        records = []
-        page_offset = offset
-        while True:
-            result = db.fetch(sql, [pk_value], limit=_LAZY_PAGE_SIZE, offset=page_offset)
-            batch = result.records
-            records.extend(batch)
-            if len(batch) < _LAZY_PAGE_SIZE:
-                break
-            page_offset += _LAZY_PAGE_SIZE
+        records = fetch_all_pages(db, sql, [pk_value], offset)
         return [related_class(row) for row in records]
 
     def belongs_to(self, related_class, foreign_key: str = None) -> Self | None:

tina4_python/queue/__init__.py

@@ -431,14 +431,7 @@ def pop_by_id(self, id: str) -> Job | None:
                         # claim the pending file — mirrors LiteBackend.pop().
                         self._backend._write_reserved(job_data)
                         os.unlink(filepath)
-                        return Job(
-                            queue=self, job_id=job_data["id"],
-                            topic=job_data.get("topic", self.topic),
-                            data=job_data["data"],
-                            priority=job_data.get("priority", 0),
-                            attempts=job_data.get("attempts", 0),
-                            error=job_data.get("error"),
-                        )
+                        return Job.from_data(self, job_data, self.topic)
                 except (json.JSONDecodeError, FileNotFoundError):
                     continue
         except FileNotFoundError:

tina4_python/queue/job.py

@@ -20,6 +20,26 @@ def __init__(self, queue, job_id, topic: str, data: dict,
         # without trawling logs.
         self.error: str | None = error
 
+    @classmethod
+    def from_data(cls, queue, job_data: dict, default_topic: str) -> "Job":
+        """Build a Job from a persisted job_data dict.
+
+        The file backends (LiteBackend and the FileQueue) store each job as a
+        JSON file and rebuild the Job from that dict on pop. The field mapping
+        is identical everywhere a stored job is revived — id, topic (falling
+        back to the queue's own topic), data, priority, attempts and error —
+        so it lives here once rather than being spelt out at each pop site.
+        """
+        return cls(
+            queue=queue,
+            job_id=job_data["id"],
+            topic=job_data.get("topic", default_topic),
+            data=job_data["data"],
+            priority=job_data.get("priority", 0),
+            attempts=job_data.get("attempts", 0),
+            error=job_data.get("error"),
+        )
+
     @property
     def data(self):
         """Alias for payload — deprecated, use .payload instead."""