diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d1f5e05d..a5dfa9af 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -60,7 +60,9 @@ jobs: # big locale-scan files (80s+ each) still land on separate workers. env: ATF_TEST_REDIS_ADDR: 127.0.0.1:6379 - run: python -m pytest -q -n auto --dist loadfile core/tests scripts/tests + # -rs: every skipped test prints its skip reason in the log — a + # skipped suite must be inspectable, not just countable. + run: python -m pytest -q -rs -n auto --dist loadfile core/tests scripts/tests - name: Telegram UI hardcoded-string guard # New visible strings in Telegram UI handlers, callbacks, and button diff --git a/core/application/delivery_reconciler.py b/core/application/delivery_reconciler.py index bb7647a3..f1396e7c 100644 --- a/core/application/delivery_reconciler.py +++ b/core/application/delivery_reconciler.py @@ -214,11 +214,20 @@ async def _check_orphan_mappings(self, owner_user_id: Optional[int]) -> List[Dic job was deleted while its mapping survived (or vice versa): the copy may or may not be in the destination, and nothing currently drives it to a conclusion. + + Retention asymmetry is NOT a mismatch: terminal delivery jobs purge + after 7 days while mappings keep 30, so a mapping older than the + job-retention window whose job is gone is an expected outcome of + cleanup, not an anomaly — flagging it every sweep produced 58 + identical production warnings for rows nothing can act on. Only + mappings inside the job-retention window are findings. """ scope, params = await self._scope(owner_user_id) db = self._db() if db is None: return [] + job_retention_seconds = 60 * 60 * 24 * 7 # mirrors the delivery_jobs cleanup window + cutoff = int(time.time()) - job_retention_seconds rows = await self._query( db, "SELECT m.rule_id, m.source_chat_id, m.source_message_id, m.target_chat_id, " @@ -227,8 +236,8 @@ async def _check_orphan_mappings(self, owner_user_id: Optional[int]) -> List[Dic "ON j.rule_id = m.rule_id AND j.source_chat_id = m.source_chat_id " "AND j.source_message_id = m.source_message_id " "AND j.target_chat_id = m.target_chat_id " - "WHERE j.id IS NULL" + scope.replace("rule_id", "m.rule_id"), - params, + "WHERE j.id IS NULL AND m.updated_at >= ?" + scope.replace("rule_id", "m.rule_id"), + (cutoff,) + tuple(params), ) findings = [] for r in rows or (): diff --git a/core/domain/entities.py b/core/domain/entities.py index bc773ace..bbb7707e 100644 --- a/core/domain/entities.py +++ b/core/domain/entities.py @@ -1840,20 +1840,38 @@ def watermark_text(self, val: str) -> None: self.metadata = {} self.metadata["watermark_text"] = str(val or "") + # Paid-addon keys the catalogue clients write, mapped onto the engine's + # canonical gate keys. The bot catalogue rows (pro_bot_ui _ADDON_ROWS) + # and the web dashboard's add-ons panel predate the engine key + # vocabulary and still record the short spellings; a paid add-on whose + # entitlement was recorded by ANY client must fire — silently never + # firing is the worst outcome for a feature a user paid for and enabled. + ADDON_KEY_ALIASES = { + "keyword_reply": ("kw_reply",), + "auto_delete_message": ("auto_del",), + "affiliate_kit": ("affiliate",), + } + def is_addon_enabled(self, key: str) -> bool: """Server-side paid-addon entitlement check. A client-supplied `paid_addons` flag is NEVER trusted on its own: the addon must be marked enabled in the persisted rule metadata. This is the single enforcement point consulted by the pipeline before any - premium transformation runs. + premium transformation runs. Canonical keys and the catalogue + clients' recorded aliases are both honored (see ADDON_KEY_ALIASES). """ if not isinstance(self.metadata, dict): return False addons = self.metadata.get("paid_addons", {}) if not isinstance(addons, dict): return False - return bool(addons.get(key, False)) + if bool(addons.get(key, False)): + return True + for alias in self.ADDON_KEY_ALIASES.get(key, ()): + if bool(addons.get(alias, False)): + return True + return False @property def link_replacement(self) -> str: diff --git a/core/infrastructure/persistence/postgres_repositories.py b/core/infrastructure/persistence/postgres_repositories.py index 7b15437f..58db556e 100644 --- a/core/infrastructure/persistence/postgres_repositories.py +++ b/core/infrastructure/persistence/postgres_repositories.py @@ -1368,6 +1368,38 @@ async def replay_dead_letter( payload, max(1, max_attempts), now, now, ), ) + # Restore the placeholder mapping row the enqueue path + # guarantees (parity with the SQLite replay): mapping rows are + # purged after 30 days while dead-letter entries persist, so a + # rebuilt forward job must not stay mapping-less — the + # worker's NoteMapping only UPDATEs and would silently drop + # the confirmed target id. The never-regress guard keeps a + # mapping row that still says SENT exactly as it is. + try: + payload_data = json.loads(payload) if isinstance(payload, str) else {} + except Exception: + payload_data = {} + if not (payload_data or {}).get("is_reply"): + # Replies share the mapping key with the primary forward; + # writing here would refresh another deliverable's row. + await _execute_db( + self._db, + "INSERT INTO message_map (" + " id, rule_id, source_chat_id, source_message_id, target_chat_id," + " target_message_id, media_group_id, delivery_status, created_at, updated_at" + ") VALUES (%s, %s, %s, %s, %s, NULL, %s, 'PENDING', %s, %s)" + " ON CONFLICT (rule_id, source_chat_id, source_message_id, target_chat_id) DO UPDATE SET" + " delivery_status = CASE" + " WHEN message_map.delivery_status IN ('SENT') THEN message_map.delivery_status" + " ELSE EXCLUDED.delivery_status END," + " updated_at = EXCLUDED.updated_at", + ( + f"map:{rule_id}:{src_chat}:{src_msg}:{tgt_chat}", + rule_id, src_chat, src_msg, tgt_chat, + str((payload_data or {}).get("media_group_id") or ""), + now, now, + ), + ) await _execute_db(self._db, "DELETE FROM dead_letter_queue WHERE id=%s", (str(entry_id),)) except Exception as exc: logger.warning("replay of dead letter %s failed: %s", entry_id, exc) @@ -1744,6 +1776,18 @@ async def resolve_gap_for_source(self, source_chat_id: str, message_id: int) -> count = getattr(res, "rowcount", None) return (count > 0) if count is not None else True + async def cleanup(self, retention_seconds: int = 60 * 60 * 24 * 30) -> int: + """Bounded retention for terminal gap rows (parity with SQLite).""" + cutoff = int(time.time()) - int(retention_seconds) + res = await _execute_db( + self._db, + "DELETE FROM ingestion_gaps" + " WHERE state IN ('RESOLVED','FAILED') AND updated_at < %s", + (cutoff,), + ) + count = getattr(res, "rowcount", None) + return int(count) if count is not None else 0 + async def pending_gaps(self, session_id: str, source_chat_id: str, limit: int = 100) -> List[dict]: rows = await _fetch_all_db( diff --git a/core/infrastructure/persistence/sqlite_repositories.py b/core/infrastructure/persistence/sqlite_repositories.py index dd980a33..a060a859 100644 --- a/core/infrastructure/persistence/sqlite_repositories.py +++ b/core/infrastructure/persistence/sqlite_repositories.py @@ -2497,6 +2497,43 @@ def _unit(conn) -> bool: # The unique key answered after all: a row for this # delivery already exists (someone replayed it first). return False + # Restore the placeholder mapping row the enqueue path + # guarantees. Cleanup purges message_map rows after 30 days + # while dead-letter entries persist, so a replay of a purged + # row would otherwise rebuild a forward job whose mapping row + # never returns — and the worker's NoteMapping (UPDATE-only) + # would silently drop the confirmed target id, leaving + # edit/delete sync blind to this delivery. The upsert is the + # enqueue path's own never-regress shape: a mapping row that + # still says SENT is never flipped back. + try: + payload_data = SqliteDatabase.loads(payload, {}) + except Exception: + payload_data = {} + if not payload_data.get("is_reply"): + # Replies share the mapping key with the primary forward + # (durable_queue._write_mapping_and_job); writing here + # would refresh a row that describes another deliverable. + conn.execute( + """ + INSERT INTO message_map ( + id, rule_id, source_chat_id, source_message_id, target_chat_id, + target_message_id, media_group_id, delivery_status, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, NULL, ?, 'PENDING', ?, ?) + ON CONFLICT(rule_id, source_chat_id, source_message_id, target_chat_id) DO UPDATE SET + delivery_status = CASE + WHEN message_map.delivery_status IN ('SENT') THEN message_map.delivery_status + ELSE excluded.delivery_status + END, + updated_at = excluded.updated_at + """, + ( + f"map:{rule_id}:{src_chat}:{src_msg}:{tgt_chat}", + rule_id, src_chat, src_msg, tgt_chat, + str(payload_data.get("media_group_id") or ""), + now, now, + ), + ) conn.execute("DELETE FROM dead_letter_queue WHERE id=?", (str(entry_id),)) return True @@ -2959,6 +2996,25 @@ async def pending_gaps(self, session_id: str, source_chat_id: str, ) return [dict(r) for r in rows or []] + async def cleanup(self, retention_seconds: int = 60 * 60 * 24 * 30) -> int: + """Bounded retention for terminal gap rows. + + ``FAILED`` rows are deliberate loss records ("visibly, never + silently") — but terminal records are not exempt from bounded + storage, and without a sweep the table grows monotonically with + every permanently-absent id. Terminal rows older than the window go, + on the same 30-day retention the message map applies to delivery + confirmations. Open rows (``PENDING``/``RETRYING``) are live + recovery work and are never touched. + """ + cutoff = _now() - int(retention_seconds) + cur = self._db.execute( + "DELETE FROM ingestion_gaps" + " WHERE state IN ('RESOLVED','FAILED') AND updated_at < ?", + (cutoff,), + ) + return cur.rowcount + async def open_gaps(self, session_id: str, source_chat_id: str, below: int = 0, limit: int = 100) -> List[dict]: """Still-open recovery work for one channel. diff --git a/core/infrastructure/telegram/channel_polling.py b/core/infrastructure/telegram/channel_polling.py index fb8ab494..d1b61f2e 100644 --- a/core/infrastructure/telegram/channel_polling.py +++ b/core/infrastructure/telegram/channel_polling.py @@ -325,6 +325,7 @@ async def stop(self) -> None: self._task = None async def _run(self) -> None: + last_gap_retention = 0.0 while not self._stop.is_set(): try: if await self._may_sweep(): @@ -335,6 +336,23 @@ async def _run(self) -> None: # are still lost, just with a row saying so. if self._gaps is not None: await self._recover_gaps() + # Gap retention: terminal (FAILED) rows older than the + # 30-day window go, on the same hourly cadence the + # durable queue's retention sweep uses. Without it the + # table grows monotonically with every permanently-absent + # id. One indexed DELETE per hour; a failure only delays + # the next sweep. + now_ts = time.time() + if now_ts - last_gap_retention >= 3600: + last_gap_retention = now_ts + try: + purged = await self._gaps.cleanup() + if purged: + logger.info( + "Gap retention purged %d terminal row(s).", purged + ) + except Exception as exc: + logger.warning("poll: gap retention sweep failed: %s", exc) except asyncio.CancelledError: raise except Exception as exc: @@ -397,8 +415,8 @@ async def _recover_one_gap(self, client, session_id: str, chat_id: str, gap: dic gap_id = str(gap.get("id") or "") mid = int(gap.get("source_message_id") or 0) try: - from pyrogram.raw.functions.updates import GetChannelDifference - from pyrogram.raw.types import ChannelMessagesFilter, MessageRange + from pyrogram.raw.functions.channels import GetMessages as ChannelGetMessages + from pyrogram.raw.types import InputMessageID, MessageEmpty peer = None try: @@ -407,22 +425,24 @@ async def _recover_one_gap(self, client, session_id: str, chat_id: str, gap: dic except Exception as exc: raise PollFailure(f"cannot resolve peer: {exc}", transient=_is_transient(exc)) from exc - full_pts = 0 - try: - from pyrogram.raw.functions.channels import GetFullChannel - full = await client.invoke(GetFullChannel(channel=peer)) - full_pts = getattr(getattr(full, "full_chat", None), "pts", 0) or 0 - except Exception: - pass # pts is only a hint for the difference call - - msg_filter = ChannelMessagesFilter(ranges=[MessageRange(min_id=mid, max_id=mid)]) - diff = await client.invoke( - GetChannelDifference(channel=peer, filter=msg_filter, pts=full_pts, limit=1) + # Fetch the id directly with channels.getMessages. The earlier + # GetChannelDifference approach needed the channel's pts from + # GetFullChannel — which Telegram flood-limits (275 rate-limit + # waits over one production weekend), and when it failed the + # code silently fell back to pts=0, where Telegram rejects the + # difference call with PERSISTENT_TIMESTAMP_EMPTY (450 + # occurrences in the same window): every gap's retry budget was + # burned on a request that could never succeed, and the posts + # were lost. getMessages answers "does this id exist?" with one + # call, no pts, and no flood-sensitive full-channel round-trip. + res = await client.invoke( + ChannelGetMessages(channel=peer, id=[InputMessageID(id=mid)]) ) - raw_messages = getattr(diff, "new_messages", None) or [] + users = {u.id: u for u in (getattr(res, "users", None) or [])} + chats = {c.id: c for c in (getattr(res, "chats", None) or [])} + raw_messages = [m for m in (getattr(res, "messages", None) or []) + if not isinstance(m, MessageEmpty)] if raw_messages: - users = {u.id: u for u in (getattr(diff, "users", None) or [])} - chats = {c.id: c for c in (getattr(diff, "chats", None) or [])} for raw in raw_messages: logger.info( "[POLL GAP] recovered post id=%s from %s — dispatching through the pipeline", @@ -448,19 +468,21 @@ async def _recover_one_gap(self, client, session_id: str, chat_id: str, gap: dic exhausted=exhausted, ) if exhausted: - logger.error( - "[POLL GAP] source %s id=%s could not be recovered after %d attempt(s) — " - "marked FAILED (the post is most likely deleted; no delivery is owed)", - chat_id, mid, GAP_MAX_ATTEMPTS, - ) + # The structured event below is the ONE authoritative record + # of this exhaustion (it carries the reason code, the + # diagnostic row, and the Log Monitor entry). A second + # logger.error of the same fact tripled every terminal row in + # the stores (473 + 463 + 463 production rows for one event + # class) without adding a single field. self._emit("error", "INGESTION_GAP_EXHAUSTED", ( - f"channel {chat_id}: id={mid} not returned after {GAP_MAX_ATTEMPTS} " - f"attempt(s) — marked FAILED (deleted or service post; no delivery owed)" + f"[POLL GAP] source {chat_id} id={mid} could not be recovered after " + f"{GAP_MAX_ATTEMPTS} attempt(s) — marked FAILED (the post is most " + f"likely deleted; no delivery is owed)" ), { "chat_id": str(chat_id), "session_id": str(session_id), "source_message_id": mid, "attempts": GAP_MAX_ATTEMPTS, "raw_error": "ID_NOT_RETURNED", - "error_detail": "GetChannelDifference returned no message at this id", + "error_detail": "channels.GetMessages returned no message at this id", "retryable": False, }) return 1 @@ -483,7 +505,7 @@ async def _recover_one_gap(self, client, session_id: str, chat_id: str, gap: dic # a paraphrase of it. from .errors import extract as _extract_error try: - event = _extract_error(exc, method="updates.GetChannelDifference", + event = _extract_error(exc, method="channels.GetMessages", operation="gap_recovery", chat_id=chat_id, session_id=session_id, attempt=attempts) raw = str(getattr(event, "error_type", "") or "") diff --git a/core/tests/test_dead_letter_replay.py b/core/tests/test_dead_letter_replay.py index ca95f39a..5c8c9ef6 100644 --- a/core/tests/test_dead_letter_replay.py +++ b/core/tests/test_dead_letter_replay.py @@ -52,6 +52,15 @@ def repo(db): return SqliteDeliveryQueueRepository(db) +@pytest.fixture +def map_repo(db): + from core.infrastructure.persistence.sqlite_repositories import ( + SqliteMessageMapRepository, + ) + + return SqliteMessageMapRepository(db) + + def _job( i: int, rule_id: str = MINE, @@ -396,3 +405,100 @@ async def test_replay_logs_ids_but_never_the_payload(db, repo, caplog): logged = "\n".join(r.getMessage() for r in caplog.records) assert "dlq_job:0" in logged # the handle a user quotes assert PAYLOAD_SECRET not in logged # the message body stays in the store + + +# --------------------------------------------------------------------- # +# The rebuilt row is not mapping-blind +# --------------------------------------------------------------------- # +async def test_a_replay_of_a_purged_row_restores_the_mapping_placeholder(db, repo, map_repo): + """The mapping invariant must survive a replay of a purged row. + + Retention purges message_map rows after 30 days; dead-letter entries + persist. A replay that rebuilds the job therefore had no mapping row — + and the Go worker's NoteMapping only UPDATEs, so the confirmed target id + would be silently dropped and edit/delete sync would never learn that + this delivery happened. The rebuild must restore the placeholder. + """ + await repo.enqueue(_job(0)) + await _exhaust(repo, "job:0") + + # The retention window passed: job row and mapping row are gone, the + # dead-letter entry persists. + db.execute("DELETE FROM delivery_jobs WHERE id='job:0'") + db.execute("DELETE FROM message_map WHERE rule_id=?", (MINE,)) + + result = await repo.replay_dead_letter("dlq_job:0", [MINE]) + assert result == "requeued" + + row = db.query_one( + "SELECT delivery_status, target_message_id FROM message_map" + " WHERE rule_id=? AND source_message_id=?", + (MINE, 100), + ) + assert row is not None, "rebuilt forward job has no mapping row — the worker's success write will silently no-op" + assert dict(row)["delivery_status"] == "PENDING" + assert dict(row)["target_message_id"] is None + + # And the worker's UPDATE-only success write now actually lands. + await map_repo.update_delivery( + MINE, "-100111", 100, "-100777", 555, DeliveryStatus.SENT, + ) + confirmed = dict(db.query_one( + "SELECT target_message_id, delivery_status FROM message_map WHERE rule_id=?", (MINE,), + )) + assert confirmed["target_message_id"] == 555 + assert confirmed["delivery_status"] == "SENT" + + +@pytest.mark.asyncio +async def test_a_replay_never_regresses_a_surviving_sent_mapping(db, repo, map_repo): + """A mapping row that still says SENT is never flipped back to PENDING.""" + await repo.enqueue(_job(0)) + await _exhaust(repo, "job:0") + + # The delivery actually landed (mapping says SENT), but the terminal job + # row was purged by cleanup — the DLQ entry was written before the send + # eventually confirmed. The placeholder exists first (the enqueue path's + # write), then the worker's success write marks it SENT. + db.execute( + "INSERT INTO message_map (id, rule_id, source_chat_id, source_message_id," + " target_chat_id, target_message_id, media_group_id, delivery_status," + " created_at, updated_at)" + " VALUES ('map-job0', ?, '-100111', 100, '-100777', NULL, '', 'PENDING'," + " strftime('%s','now'), strftime('%s','now'))", + (MINE,), + ) + await map_repo.update_delivery(MINE, "-100111", 100, "-100777", 42, DeliveryStatus.SENT) + db.execute("DELETE FROM delivery_jobs WHERE id='job:0'") + + result = await repo.replay_dead_letter("dlq_job:0", [MINE]) + assert result == "requeued" + + row = dict(db.query_one( + "SELECT delivery_status, target_message_id FROM message_map WHERE rule_id=?", (MINE,), + )) + assert row["delivery_status"] == "SENT" + assert row["target_message_id"] == 42 + + +@pytest.mark.asyncio +async def test_a_reply_replay_does_not_touch_the_forward_mapping(db, repo): + """Replies share the mapping key with the primary forward; a rebuilt + reply must not refresh a row that describes another deliverable.""" + reply_job = DeliveryJob( + id="job:9", rule_id=MINE, source_chat_id="-100111", source_message_id=199, + target_chat_id="-100777", status=DeliveryStatus.PENDING, max_attempts=5, + payload_data={"text": "auto-reply", "is_reply": True}, + ) + await repo.enqueue(reply_job) + await _exhaust(repo, "job:9") + db.execute("DELETE FROM delivery_jobs WHERE id='job:9'") + db.execute("DELETE FROM message_map WHERE rule_id=?", (MINE,)) + + result = await repo.replay_dead_letter("dlq_job:9", [MINE]) + assert result == "requeued" + + # No mapping row was invented for a reply: the key belongs to the + # forward, and writing it here would refresh another deliverable's row. + row = db.query_one("SELECT 1 FROM message_map WHERE rule_id=?", (MINE,)) + assert row is None diff --git a/core/tests/test_delivery_reconciliation.py b/core/tests/test_delivery_reconciliation.py index 33bc878d..57133b85 100644 --- a/core/tests/test_delivery_reconciliation.py +++ b/core/tests/test_delivery_reconciliation.py @@ -165,6 +165,19 @@ async def test_an_orphan_mapping_is_flagged(db, queue, mmap, rules, attempts, ga assert o["error_category"] == "orphan_mapping" +async def test_an_aged_orphan_mapping_is_retention_not_anomaly(db, queue, mmap, rules, attempts, gaps): + """Terminal jobs purge after 7 days while mappings keep 30: an old + mapping whose job is gone is cleanup's expected outcome, not a state + mismatch — flagging it every sweep produced 58 identical production + warnings for rows nothing can act on. Only in-window orphans count.""" + await rules.add(make_rule()) + _mapping(db) + aged = int(time.time()) - 60 * 60 * 24 * 8 # one day past the job-retention window + db.execute("UPDATE message_map SET updated_at=? WHERE rule_id='rule-1'", (aged,)) + r = await _reconciler(queue, mmap, rules, attempts, gaps).reconcile() + assert r["checks"]["orphan_mappings"] == [], "a retention-aged mapping is not an anomaly" + + async def test_a_sent_job_with_its_mapping_is_not_an_orphan(db, queue, mmap, rules, attempts, gaps): """The join is on the full deliverable key, so a matching job clears it.""" await rules.add(make_rule()) diff --git a/core/tests/test_ingestion_gap_recovery.py b/core/tests/test_ingestion_gap_recovery.py index fc9ca10d..b455adee 100644 --- a/core/tests/test_ingestion_gap_recovery.py +++ b/core/tests/test_ingestion_gap_recovery.py @@ -68,19 +68,29 @@ async def resolve_peer(self, x): async def invoke(self, req): name = type(req).__name__ if name == "GetFullChannel": + # The NORMAL sweep needs the channel's pts for its difference + # call — only the single-id recovery was forbidden from this + # flood-limited round-trip. return types.SimpleNamespace(full_chat=types.SimpleNamespace(pts=103)) if name == "GetChannelDifference": asked_lo = req.filter.ranges[0].min_id asked_hi = req.filter.ranges[0].max_id - if asked_lo == asked_hi: - # A single-id recovery probe: answer only if that id is real - # *and* was part of the originally-returned set is too strict — - # recovery is precisely for the ids the difference omitted. - if asked_lo in self.real_ids: - return FakeDiff([make_raw_message(asked_lo)], self._chats) - return FakeDiff([], self._chats) rng = [i for i in self.returned_ids if asked_lo <= i <= asked_hi] return FakeDiff([make_raw_message(i) for i in sorted(rng)], self._chats) + if name == "GetMessages": + # channels.getMessages — the recovery path's fetch. A deleted id + # answers with MessageEmpty (Telegram's own shape), never an + # error; an existing id answers with the raw message. + from pyrogram.raw.types import MessageEmpty + + mid = req.id[0].id + if mid in self.real_ids: + return types.SimpleNamespace( + messages=[make_raw_message(mid)], users=[], chats=self._chats, + ) + return types.SimpleNamespace( + messages=[MessageEmpty(id=mid, peer_id=None)], users=[], chats=self._chats, + ) raise AssertionError(name) def get_chat_history(self, target, limit=1): @@ -271,3 +281,92 @@ async def h(c, m): # And those posts became recovery work rather than vanishing. pending = await gaps.pending_gaps(SESSION, CHAT) assert sorted(g["source_message_id"] for g in pending) == list(range(61, 71)) + + +# --------------------------------------------------------------------------- # +# Bounded retention for terminal rows +# --------------------------------------------------------------------------- # + +@pytest.mark.asyncio +async def test_terminal_gaps_age_out_but_open_ones_never_do(db, gaps): + """FAILED rows are deliberate loss records, not exempt from bounded + storage: without retention the table grows monotonically with every + permanently-absent id (production reached 468). Terminal rows older + than the window go; open rows are live recovery work and are never + touched, whatever their age.""" + import time as _time + + await gaps.record_gaps(SESSION, CHAT, [80, 81, 82]) + await gaps.mark_gap_attempt("gap:%s:%s:80" % (SESSION, CHAT), "never returned", 0.0, + exhausted=True) + + # Backdate the FAILED row past the retention window; leave the open + # ones untouched but backdate one of them too — age alone must not + # delete live work. + old = int(_time.time()) - 60 * 60 * 24 * 31 + db.execute("UPDATE ingestion_gaps SET updated_at=? WHERE source_message_id=80", (old,)) + db.execute("UPDATE ingestion_gaps SET updated_at=? WHERE source_message_id=81", (old,)) + + purged = await gaps.cleanup() + assert purged == 1 + + rows = {g["source_message_id"]: g["state"] for g in + (await gaps.pending_gaps(SESSION, CHAT))} + assert 80 not in rows, "the aged FAILED row must be gone" + assert rows.get(81) == "PENDING", "an open row is never retained away" + assert rows.get(82) == "PENDING" + + +# --------------------------------------------------------------------------- # +# The recovery fetch must not depend on pts / GetFullChannel +# --------------------------------------------------------------------------- # + +@pytest.mark.asyncio +async def test_gap_recovery_uses_get_messages_and_never_pts(db, gaps): + """The old recovery built GetChannelDifference from GetFullChannel's pts: + flood-limited (275 rate-limit waits in one production weekend), and when + it failed the code silently fell back to pts=0, where Telegram rejected + EVERY attempt with PERSISTENT_TIMESTAMP_EMPTY — 450 occurrences that + burned each gap's budget on a request that could never succeed. The + recovery fetch is channels.getMessages: no pts, no full-channel call. + + The FakeTelegram above raises AssertionError on any request it does not + implement, and GetFullChannel is no longer implemented — so this test + passing IS the proof the pts path is gone.""" + client = FakeTelegram(list(range(61, 71)), list(range(61, 71))) + invoked: list = [] + + original_invoke = client.invoke + + async def invoking(req): + invoked.append(type(req).__name__) + return await original_invoke(req) + + client.invoke = invoking + + await gaps.record_gaps(SESSION, CHAT, [66]) + poller = make_poller(client, gaps, cursor=60) + # Warm the dispatch lanes exactly as the normal sweep does, then drive + # the recovery — mirroring the production order. + await poller._poll_channel(client, SESSION, CHAT, "A") + await asyncio.sleep(0.4) + before_recovery = len(invoked) + recovered = await poller._recover_gaps(limit=100) + await asyncio.sleep(0.4) + assert recovered == 1 + + # Scope the contract to the RECOVERY window: the warm-up sweep may + # legitimately use GetFullChannel/GetChannelDifference — the recovery + # fetch must not. + recovery_calls = invoked[before_recovery:] + assert any(n == "GetMessages" for n in recovery_calls), f"recovery never called channels.getMessages: {recovery_calls}" + assert not any(n == "GetFullChannel" for n in recovery_calls), "recovery still consults GetFullChannel (the pts trap)" + assert not any(n == "GetChannelDifference" for n in recovery_calls), "single-id recovery still goes through GetChannelDifference" + + # The recovered post reached the dispatcher. The gap row itself is left + # to the pipeline observer (only the pipeline seeing the id proves it + # was not lost) — and the fake dispatcher runs no pipeline, so the row + # legitimately stays open in this harness. + assert 66 in client.dispatched, "the recovered post never reached the dispatcher" + rows = await gaps.pending_gaps(SESSION, CHAT) + assert [g["source_message_id"] for g in rows] == [66] diff --git a/core/tests/test_phase2_advanced_settings_execution.py b/core/tests/test_phase2_advanced_settings_execution.py index bce0d2da..ee9f294e 100644 --- a/core/tests/test_phase2_advanced_settings_execution.py +++ b/core/tests/test_phase2_advanced_settings_execution.py @@ -497,3 +497,57 @@ async def fake_atomic(self, mapping, job): assert captured["payload_data"].get("is_reply", False) is False assert "keyword_reply" not in captured["job_id"] + + +# --------------------------------------------------- catalogue key aliases -- +class TestCatalogueAddonKeyAliases: + """The bot catalogue rows and the web dashboard record the short + spellings (kw_reply / auto_del / affiliate); the engine gates read the + canonical vocabulary (keyword_reply / auto_delete_message / + affiliate_kit). The gate accepts both: a paid add-on whose entitlement + was recorded by ANY client must fire. The production defect this pins: + three paid add-ons never fired from a real client because the gate read + a key nothing wrote — the canonical keys existed only in tests.""" + + def test_is_addon_enabled_accepts_both_spellings(self): + for canonical, alias in ( + ("keyword_reply", "kw_reply"), + ("auto_delete_message", "auto_del"), + ("affiliate_kit", "affiliate"), + ): + canonical_rule = _rule(paid_addons={canonical: True}) + alias_rule = _rule(paid_addons={alias: True}) + assert canonical_rule.is_addon_enabled(canonical) is True + assert alias_rule.is_addon_enabled(canonical) is True, ( + f"catalogue spelling {alias!r} did not reach the {canonical!r} gate" + ) + assert _rule(paid_addons={alias: False}).is_addon_enabled(canonical) is False + assert _rule(paid_addons={alias: True}).is_addon_enabled("crypto_mode") is False + + def test_affiliate_kit_transform_runs_through_the_alias(self): + rule = _rule( + paid_addons={"affiliate": True}, + affiliate_kit={"enabled": True, "prefix": "https://t.me/capxi/"}, + ) + text, applied = AdvancedSettingsEngine.apply_content_transforms( + "visit https://example.com now", rule + ) + assert "affiliate_kit" in applied + assert text.startswith("visit https://t.me/capxi/https://example.com") + + def test_keyword_reply_fires_through_the_alias(self): + rule = _rule( + paid_addons={"kw_reply": True}, + keyword_confirmation=True, + keyword_reply="hey!", + whitelist=["hello"], + ) + assert AdvancedSettingsEngine.classify_keyword_trigger("hello there", rule) == "hey!" + + def test_auto_delete_ttl_flows_through_the_alias(self): + rule = _rule(paid_addons={"auto_del": True}, auto_delete_seconds=3600) + assert AdvancedSettingsEngine.get_auto_delete_seconds(rule) == 3600 + + def test_alias_false_keeps_the_gate_off(self): + rule = _rule(paid_addons={"auto_del": False}, auto_delete_seconds=3600) + assert AdvancedSettingsEngine.get_auto_delete_seconds(rule) == 0 diff --git a/docs/reliability/pr339-audit-ledger.md b/docs/reliability/pr339-audit-ledger.md new file mode 100644 index 00000000..7d0739f8 --- /dev/null +++ b/docs/reliability/pr339-audit-ledger.md @@ -0,0 +1,112 @@ +# PR #339 Engineering Audit Ledger — Task Engine & Worker Reliability Round + +Branch `fix/media-rejection-reupload` · baseline master `36b7529` (PR #338) · audit window 2026-11-05. +Everything below is evidence-backed; production findings were collected **read-only** against `vps-root`. + +## 1. Issue ledger + +| # | Issue | Component | Severity | Root cause | Fix commit | Regression test | Status | +|---|---|---|---|---|---|---|---| +| W1 | No reconciliation loop — `FenceResolveUnknown` had zero callers; 13 production UNKNOWN jobs stranded 20+ h (10 since Nov 4, all `gRPC Unimplemented` from the pre-#330 missing-SendMedia era) | worker queue/engine | P1 | The fence documented "reconciliation decides" but the decision step never existed | `ffbef52` | `queue/reconcile_store_test.go` (5), `engine/reconciler/reconciler_test.go` (4) | FIXED (deploy-gated) | +| W2 | DLQ replay of a purged row rebuilt a forward job **without** its `message_map` placeholder → worker's UPDATE-only `NoteMapping` silently drops the confirmed target id; edit/delete sync goes blind | core persistence (sqlite+pg) | P2 | Mapping retention (30 d) outlives nothing — DLQ entries persist while mappings purge; rebuild path forgot the placeholder | `d535a18` | `test_dead_letter_replay.py` (3 new) | FIXED | +| W3 | PHOTO_EXT_INVALID root cause: re-upload staged files as `job--media` with **no extension** → Telegram `messages.SendMedia` rejects (12 prod jobs died permanently) | worker delivery | P1 | `GetFileExtension` RPC fully implemented end-to-end but never wired | `fea2665` | `delivery_test.go` (7 new) + `telegram_runtime/tests/test_file_extension.py` (8) | FIXED (deploy-gated) | +| W4 | Target-side media rejection (`PHOTO_EXT_INVALID`) never triggered the re-upload fallback (only `FORWARDING_RESTRICTED` did) | worker delivery | P2 | Fallback condition checked one error class | `d5f81e1` | `TestMediaForwardTargetRejectsCopy`, `TestMediaRejectedNoFallback` | FIXED | +| G1 | 3 paid add-ons unreachable from every production client: UI records `kw_reply`/`auto_del`/`affiliate`; engines gate on `keyword_reply`/`auto_delete_message`/`affiliate_kit` | bot/web ↔ engines | P1 | Key vocabulary drift between catalogue and engine; canonical keys existed only in tests | `d535a18` | `worker addon_alias_test.go` (9 cases) + `test_phase2_advanced_settings_execution.py::TestCatalogueAddonKeyAliases` (5) | FIXED | +| F2 | Deliberate non-deliveries (filter drop, block_links, FEN cadence skip) counted as `failed`; filter drop also failed the filter stage → healthy task reads degraded | worker ops | P3 | Rejections had no outcome of their own | `d535a18` | updated FEN/link-block tests + ORD-002 accepts `filter:skipped` | FIXED | +| F3 | `worker/test/live/run_matrix.py` runs against the **production** DB and kept its replica's job rows (`keep_jobs=True`) — the seeding path behind orphaned `e2e-go-*`/`mtx-*` rows | live-fire tooling | P2 | Cleanup stopped at rules+counters; no replica-namespace guard | `6274c96` | (tooling; guard refuses non-`mtx-*` ids) | FIXED | +| F1 | `ingestion_gaps` unbounded growth — 468 terminal rows, no retention | core persistence | P3 | Terminal rows are loss records with no sweep | `6274c96` | `test_terminal_gaps_age_out_but_open_ones_never_do` | FIXED | +| O1 | `db_connected=false`/degraded on a healthy unpolled worker (DB probe piggybacked on `/healthz`) | worker health | P2 | Probe ran only when someone polled | `f0426a1` | (build; verified on vps suite) | FIXED | +| O2 | No redaction on the stdout JSON sink (repo rule: `[REDACTED]`) | worker logging | P2 | Redaction lived only in the central client | `f0426a1` | (build; shared rule set by construction) | FIXED | +| O3 | E2E latency ring blind to every failure/defer path (contradicting `Process`'s own comment) | worker delivery | P3 | `handleFailure` never called `e.outcome` | `f0426a1` | existing failure-path tests still pass | FIXED | +| O4 | `worker_id` absent from all delivery log lines; failure→ring lost `correlation_id`; AI stage counted non-AI jobs and double-counted real ones; `ai` force-enabled without a provider | worker observability | P3 | field sets / counters drifted from the logging contract | `f0426a1` | (build; verified) | FIXED | + +## 2. Feature verification matrix (18 add-ons + core settings) + +Full matrix with file:line evidence was compiled during this round (feature-execution audit). Summary: + +**Verified executed by the Go Worker** (execution site + tests): Forward Every N, Keyword Reply engine, Link Buttons, Watermark, Cleaner Protocol, Smart AI Rewrite, Duplicate-prevention *fence* (send idempotency), Whitelist/Blacklist, Replacements, Headers/footers, Topics, Source/target routing + loop preflight, `fallback_enabled`, **BOT transport** (stage-3 ledger concern resolved — BotSendText/BotCopyMessage/auto-delete/branding all implemented), Branding hop, Auto-Reply/PVA engine, Auto Delete scheduling (`scheduleAutoDelete` + scheduler). + +**Repaired this round:** Keyword Reply / Auto Delete / Affiliate Kit entitlement reachability (G1). + +**Genuinely not implemented (flags persist, no engine consumer)** — the catalogue rows are decorative for: Channel Comments, Outbound Webhooks, Crypto Mode, OCR Image AI, Auto Post Scheduler, Power On/Off Schedule, Process Raw Text, Button To Text (plus hidden extras: News Digest, Translate, crypto sub-rows). Two of them (`process_raw_text`, `button_to_text`) are even baked into job payloads (`durable_queue.py:391-392`) that nothing reads. + +**Required follow-up (tracked, not silently ignored):** either implement these engines or enforce their unavailability honestly in the catalogue UI (10-locale copy work). Duplicate Filters is **Python-ingest-only** since the event-dedup gate was deliberately removed from the worker (`delivery.go:474-495`, after it dropped 100 % of a live task's traffic) — worker-owned delivery has no content dedup today. Single-engine features (`processing delay`, `sync_replies`, `sync_edits/sync_deletes`) execute only in Python; the strangler-fig handover has not reached them. + +## 3. Production DB diagnostics (read-only, `atf.db` snapshot) + +| Finding | Evidence | Disposition | +|---|---|---| +| 12 jobs FAILED, `attempts=0`, `PHOTO_EXT_INVALID (caused by messages.SendMedia)` | rule `40edad5a`, source `-1001957768371` → target `-1001854636317`, msg ids 18321–18375, Nov 05 09:18–14:35 | Root cause W3/W4 (both fixed). No delivery side effect (fence `tmsg=None`, `message_map.target_message_id=NULL`). Recovery = requeue; see §5 | +| 10 UNKNOWN jobs (rule family) + 3 UNKNOWN `e2e-go-*` test artifacts | `Uncertain outcome: rpc error: code = Unimplemented desc = Unexpected ` — server rejected the RPC **before any handler ran** | Provably no Telegram side effect. The new reconciler will auto-close the 10 to FAILED post-deploy; the 3 test artifacts are deletable (approval-gated) | +| 7 other FAILED jobs | 1 × `CHAT_ADMIN_REQUIRED` (genuine permission gap — needs the owner to grant admin), 2 × filter rejections, 4 × FEN cadence skips | Correct terminal states (rejections now counted as `rejected`, not `failed`) | +| DLQ empty; no synthetic leftovers; cursor at 18375 | — | Queue healthy; worker idle | +| WAL 7.5 MB, checkpoint lag | — | Monitor; not actionable at this scale | + +## 4. CI history (this branch) + +| Commit | CI | CodeQL | Notes | +|---|---|---|---| +| `d5f81e1` (fallback classification) | green | green | — | +| `fea2665` (staging extension) | green | green | — | +| `ffbef52` (reconciler) | green | green | — | +| `d535a18` (replay/alias/reject) | **failure** → repaired | green | ORD-002 pinned the old filter-stage semantics (updated to accept policy-skip); 2 new Python tests used the wrong repo class (fixed) | +| `f0426a1` (observability) | running | green | — | +| `6274c96` (retention + matrix cleanup) | running | pending | — | + +CI-log deep inspection findings (all 15 jobs read): race detector confirmed wired in `go-worker-tests`; 0 Go skips; pytest 34 skips now print reasons (`-rs` added); no compiler warnings beyond toolchain noise (Node 20 / setup-java deprecations, punycode); website reports `2 vulnerabilities (1 moderate, 1 high)` in its own npm audit — tracked for the website workstream; docker-build layers cache as designed. + +## 5. Recovery procedure for the poisoned rows — **requires explicit approval** + +All steps are prepared, none executed. Run AFTER the fixed worker image is deployed. + +```sql +-- (a) The 12 PHOTO_EXT_INVALID jobs: evidence-verified requeue. +-- Safety: fence FAILED + attempts=0 + no fence target id + mapping +-- target NULL ⇒ Telegram never accepted anything; the fixed pipeline +-- (fallback → re-upload with extension) will deliver exactly once. +UPDATE delivery_jobs SET status='PENDING', next_retry_at=0, lease_until=0, + error_detail='', updated_at=strftime('%s','now') +WHERE status='FAILED' AND error_detail LIKE '%PHOTO_EXT_INVALID%'; +-- (b) The 10 rule-family UNKNOWN jobs: let the deployed reconciler close +-- them to FAILED automatically (it classifies Unimplemented as +-- provably unexecuted), then requeue with the same statement pattern +-- (error_detail LIKE '%reconciled:%'). +-- (c) The 3 e2e-go-* UNKNOWN artifacts: DELETE (test contamination; the +-- fixed run_matrix now prevents recurrence). +DELETE FROM delivery_jobs WHERE id LIKE 'e2e-go-%'; +-- (d) The 1 CHAT_ADMIN_REQUIRED job: NOT recoverable by retry — the +-- target's admin must grant the sending account permission first. +``` + +Duplicate-delivery risk per group: (a) zero — Telegram rejected every attempt server-side; (b) zero — the RPC never reached a handler; (c) n/a (no send ever associated); (d) retry adds no dup risk but cannot succeed until permissions change. + +## 6. Remaining external blockers / follow-ups + +1. **Deployment** of the fixed worker image (auto-deploys on merge; manual via `cd` workflow_dispatch) — user-gated. +2. **Recovery execution** (§5) — user-gated. +3. G2 decorative add-ons: implement-or-enforce decision (product-level). +4. Duplicate Filters worker-side story (was removed for cause; Python ingest still owns it). +5. Single-engine features (delay / sync_replies / sync_edits/deletes) strangler-fig completion. +6. Readiness depth (fence/runtime probes), `/metrics` exposition, scheduler metrics stub, unauthenticated :8090 controls (runbook-mitigated) — observability follow-ups, ledgered in the audit. +7. Website npm audit (1 high, 1 moderate) — website workstream. + +## 7. Production log audit — WARN/ERROR inventory (2026-11-05, deduplicated) + +Sources: the **live** Rust logger store (`atf_logs_rust.db`, 5260 rows — the legacy `logger.db` beside it has been dead since Oct 16 and holds only stale API/probe rows) plus `diagnostic_events` in `atf.db` (695 rows). 5955 WARN/ERROR rows → **80 distinct signatures** (`atf-warn-error-dedup.csv`, count/severity/service/normalized-signature/first/last-seen/sample). + +| Cluster | Volume | Verdict | Action in this PR | +|---|---|---|---| +| Gap recovery fails with `PERSISTENT_TIMESTAMP_EMPTY` | 450 WARN + 463 exhausted-gap ERRORs + 691 gap-creation WARNs (the backlog behind them) | **Defect (L-A, P1)**: single-id recovery built `GetChannelDifference` from `GetFullChannel`'s pts — flood-limited (275 waits), and on failure pts silently defaulted to 0, which Telegram rejects. Every recovery attempt of a perfectly recoverable post burned the budget | Recovery now uses `channels.GetMessages` (no pts, no full-channel round-trip, `MessageEmpty` handled as the deleted-post answer). Regression test pins the RPC contract: GetMessages used, GetFullChannel/GetChannelDifference never called during recovery | +| Exhausted gap recorded 3× per gap (POLL GAP line + EVT + diagnostic row) | 473+463+463 rows | **Defect (L-C, P2)**: same fact, three stores, zero added fields | The structured event is the single authoritative record; the duplicate `logger.error` removed | +| `/api/worker/*` 404 ×337 (today) + `/api/admin/worker/control/status` 404 / `pause` 405 | 344 | **Stale deployment (L-B, P2)**: the 401-vs-404 asymmetry proves the deployed API image predates #335/#336 — the route exists on master (401 for unauth) but not in the running binary. The `/api/worker/*` calls come from a client bundle older than #335 | No code change (correct since #335); **deploy blocker** — merging this PR refreshes all images | +| Orphan-mapping reconcile warnings | 58 WARN, recurring daily | **Defect (L-F, P3)**: terminal jobs purge at 7 d, mappings keep 30 d — retention asymmetry flagged as a state mismatch forever | The reconciler now only flags in-window orphans; aged ones are documented cleanup outcomes. Test pins both sides | +| `health server stopped` WARN | 18 | **Level defect (L-E, P3)**: listener close on shutdown is the normal lifecycle | INFO on `ErrServerClosed`, WARN only on a real failure | +| `AI transform failed` + fallback | 8+8 | Fallback policy working as designed (provider 401/limits); observable, not silent | None — by design | +| Security probes (`/api/.env*`, `/api/fs/read`, `/api/uploads/../../.env`, `proc/self/environ`, graphql, `/d/*`) | ~120 | External scanning Nov 1–3; **every attempt answered Unauthorized** — auth held | None; recorded for the security review | +| `Waiting N seconds (channels.GetFullChannel)` | 275 | External Telegram flood-limit; honored. Collapses with the L-A fix (recovery no longer calls it) | None | +| `[POLL GAP]` repeat detections | ~120 | Already correct: only NEW gaps warn; repeats log INFO | None | +| `claim failed — retrying`, `log client cannot reach logger`, `auto-reply consumer failed to start` (×1) | 7 | Transient conditions, retried/recovered; visible by design | None | +| `rule no longer exists`, `already has a confirmed send`, `event already processed` | 49 | Fence/idempotency working as designed | None | +| Legacy `logger.db` store (198 rows, dead since Oct 16) + old probe/acceptance rows | 198 | Stale store kept beside the live one; probe rows are past test residue | Cleanup of the legacy file/rows is approval-gated (production data) | + +Net expected after deploy: the two dominant error masses (463 exhausted + 450 recovery failures) collapse to near-zero — recovery succeeds instead of burning budgets — and the log volume drops accordingly. diff --git a/telegram_runtime/tests/test_file_extension.py b/telegram_runtime/tests/test_file_extension.py new file mode 100644 index 00000000..feec794b --- /dev/null +++ b/telegram_runtime/tests/test_file_extension.py @@ -0,0 +1,112 @@ +"""Contract tests for ``get_file_extension`` — the RPC the Go Worker's +re-upload staging depends on. + +The staged scratch file must carry the source media's extension, because +Telegram's ``messages.SendMedia`` infers the MIME type from the staged +file's name: a bare ``job--media`` dies with PHOTO_EXT_INVALID before +the upload can ever succeed. The Go side normalizes the answer and falls +back to the ingest classification when this RPC cannot answer, but the +contract it trusts is defined HERE — these tests pin it: + +* photos have no file_name in MTProto — the answer is always ``.jpg``; +* media with an original file_name answers with that name's raw suffix + (normalization is the Go side's job, never a second opinion here); +* media without any inferable name (voice notes, video notes) answers + ``""`` — the Go fallback owns the default; +* a failed source read is a typed error, never a raw exception. +""" + +from __future__ import annotations + +import asyncio +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from telegram_runtime.adapters.pyrogram.client import PyrogramAdapter # noqa: E402 +from telegram_runtime import errors # noqa: E402 + + +class _StubClient: + """The smallest pyrogram-shaped client get_file_extension can probe.""" + + def __init__(self, message: Any = None, exc: Exception | None = None) -> None: + self._message = message + self._exc = exc + + async def get_messages(self, chat_id: Any, message_ids: Any) -> Any: + if self._exc is not None: + raise self._exc + return self._message + + +def _ext(client: _StubClient, chat_id: Any = "c", message_id: int = 1) -> tuple[str, Any]: + adapter = PyrogramAdapter(client) + return asyncio.run(adapter.get_file_extension(chat_id, message_id)) + + +def test_photo_answers_jpg() -> None: + # A Telegram photo is always JPEG — it has no file_name to consult. + msg = SimpleNamespace(photo=SimpleNamespace()) + ext, ferr = _ext(_StubClient(msg)) + assert ferr is None + assert ext == ".jpg" + + +def test_document_answers_with_its_real_suffix() -> None: + msg = SimpleNamespace(document=SimpleNamespace(file_name="report.pdf")) + ext, ferr = _ext(_StubClient(msg)) + assert ferr is None + assert ext == ".pdf" + + +def test_suffix_is_reported_raw() -> None: + # Normalization is the Go side's job: an uppercase suffix rides as-is. + msg = SimpleNamespace(video=SimpleNamespace(file_name="clip.MP4")) + ext, ferr = _ext(_StubClient(msg)) + assert ferr is None + assert ext == ".MP4" + + +def test_document_without_name_answers_empty() -> None: + msg = SimpleNamespace(document=SimpleNamespace(file_name=None)) + ext, ferr = _ext(_StubClient(msg)) + assert ferr is None + assert ext == "" + + +def test_voice_note_answers_empty() -> None: + # Voice notes carry no file_name; the Go fallback supplies .ogg. + msg = SimpleNamespace(voice=SimpleNamespace(file_name=None)) + ext, ferr = _ext(_StubClient(msg)) + assert ferr is None + assert ext == "" + + +def test_bare_message_answers_empty() -> None: + ext, ferr = _ext(_StubClient(SimpleNamespace())) + assert ferr is None + assert ext == "" + + +def test_failed_source_read_is_a_typed_error() -> None: + ext, ferr = _ext(_StubClient(exc=RuntimeError("source chat gone"))) + assert ext == "" + # Typed vocabulary, never a raw exception: an unrecognized failure maps + # to the shared taxonomy (UNKNOWN for a plain Python error) so the Go + # side can always react to a typed code instead of crashing. + assert ferr is not None + assert ferr.code == errors.UNKNOWN_DELIVERY_OUTCOME + + +def test_unreadable_media_never_raises() -> None: + # getattr-based probing: any object shape must degrade to "" or a typed + # error, never an AttributeError escaping the adapter. + ext, ferr = _ext(_StubClient("not even a message")) + assert ferr is None + assert ext == "" diff --git a/worker/cmd/worker/logging.go b/worker/cmd/worker/logging.go index bc76b718..4d89f804 100644 --- a/worker/cmd/worker/logging.go +++ b/worker/cmd/worker/logging.go @@ -29,7 +29,13 @@ func (h *fanoutHandler) Enabled(ctx context.Context, level slog.Level) bool { func (h *fanoutHandler) Handle(ctx context.Context, record slog.Record) error { var err error if h.stdout.Enabled(ctx, record.Level) { - err = h.stdout.Handle(ctx, record) + // The stdout stream carries the SAME redaction the central client + // applies to its own records: secret-shaped keys and content fields + // become [REDACTED], secret-shaped substrings inside messages and + // error strings are masked. The container stdout is what a log + // collector actually scrapes — one careless "error", err argument + // must not leak a token there. + err = h.stdout.Handle(ctx, redactedRecord(record)) } if h.central == nil { return err @@ -47,6 +53,31 @@ func (h *fanoutHandler) Handle(ctx context.Context, record slog.Record) error { return err } +// redactedRecord rebuilds a record with the shared redaction applied to the +// message and every attribute (groups recursed, structure preserved). The +// per-call attributes all flow through Handle, so this is the one choke +// point the stdout sink needs. +func redactedRecord(record slog.Record) slog.Record { + out := slog.NewRecord(record.Time, record.Level, observability.RedactSlogText(record.Message), record.PC) + record.Attrs(func(attr slog.Attr) bool { + out.AddAttrs(redactAttr(attr)) + return true + }) + return out +} + +func redactAttr(attr slog.Attr) slog.Attr { + value := attr.Value.Resolve() + if value.Kind() == slog.KindGroup { + group := slog.Group(attr.Key) + for _, sub := range value.Group() { + group.Value = slog.GroupValue(append(group.Value.Group(), redactAttr(sub))...) + } + return group + } + return slog.Attr{Key: attr.Key, Value: slog.AnyValue(observability.RedactSlogValue(attr.Key, value.Any()))} +} + func (h *fanoutHandler) WithAttrs(attrs []slog.Attr) slog.Handler { combined := make([]slog.Attr, 0, len(h.attrs)+len(attrs)) combined = append(combined, h.attrs...) diff --git a/worker/cmd/worker/main.go b/worker/cmd/worker/main.go index c6430009..4dd51a17 100644 --- a/worker/cmd/worker/main.go +++ b/worker/cmd/worker/main.go @@ -39,6 +39,7 @@ import ( "atf/worker/internal/engine/autoreply" "atf/worker/internal/engine/counters" "atf/worker/internal/engine/delivery" + "atf/worker/internal/engine/reconciler" "atf/worker/internal/engine/scheduler" "atf/worker/internal/health" "atf/worker/internal/infra/aiprovider" @@ -230,7 +231,15 @@ func run() (runErr error) { // ones Python owns are disabled. A stage that never runs reports // "waiting", a stage whose switch is off reports "disabled" — the two // are different operator questions and the panel distinguishes them. + // + // "ai" is excluded from the blanket enable: it is enabled further down + // ONLY when a transformer is actually wired, so an AI store that failed + // to open leaves the stage honestly "waiting" instead of painted + // enabled-without-a-provider. for _, stage := range ops.StageNames { + if stage == "ai" { + continue + } collector.StageEnabled(stage, true) } if !cfg.WorkerEnabled { @@ -312,17 +321,17 @@ func run() (runErr error) { // /ops attaches to this server's mux. tracker := ops.NewTrackerBridge(collector) healthSrv := health.NewServer(cfg.ListenAddr, tracker, func(ctx context.Context) error { - _, err := store.PendingCount(ctx) - // The probe's outcome feeds the connectivity flag /ops reports, so - // "db_connected" is answered by the same call that answers - // readiness — never a second, divergent probe. + // One probe answers both: the outcome feeds the connectivity flag + // /ops reports AND the pending-depth sampler — never a second, + // divergent probe. The 15s ticker below keeps the flag fresh even + // when nobody polls readiness, so a healthy-but-unobserved worker + // no longer reads as db_connected=false / degraded. + n, err := store.PendingCount(ctx) collector.RecordDB(err == nil, time.Now()) if err != nil { return err } - if n, err := store.PendingCount(ctx); err == nil { - collector.RecordPendingDepth(int64(n)) - } + collector.RecordPendingDepth(int64(n)) return nil }) @@ -352,8 +361,14 @@ func run() (runErr error) { opsSrv.Attach(mux) } go func() { - if err := healthSrv.Start(); err != nil && !errors.Is(err, net.ErrClosed) { - log.Warn("health server stopped", "error", err) + if err := healthSrv.Start(); err != nil { + if errors.Is(err, http.ErrServerClosed) || errors.Is(err, net.ErrClosed) { + // Shutdown closing the listener is the normal lifecycle, not + // a degradation — 18 production WARNs came from this line. + log.Info("health server stopped (shutdown)") + } else { + log.Warn("health server stopped unexpectedly", "error", err) + } } }() @@ -387,6 +402,15 @@ func run() (runErr error) { return case <-ticker.C: probeTelegram(ctx, collector, gateway, log) + // The DB probe rides the same cadence: db_connected and the + // pending-depth sampler stay truthful without anyone + // polling /healthz first. + if n, err := store.PendingCount(ctx); err == nil { + collector.RecordDB(true, time.Now()) + collector.RecordPendingDepth(int64(n)) + } else { + collector.RecordDB(false, time.Now()) + } // The worker's own liveness mark: /ops only advanced the // heartbeat when someone looked, so a healthy worker with no // observers read as offline. This store is what keeps @@ -585,6 +609,22 @@ func runLoop(ctx context.Context, log *slog.Logger, engine *delivery.Engine, sto } }() + // The UNKNOWN reconciler: the fence sweep deliberately leaves ambiguous + // sends UNKNOWN ("the send may have landed"), but an ambiguity nobody + // ever inspects is a job stranded forever — production accumulated a day + // of them because the decision step did not exist. This loop closes only + // the rows whose recorded failure PROVES the runtime rejected the call + // before any Telegram operation; genuinely uncertain rows stay UNKNOWN + // and visible in /ops for an operator. + reconcileGrace := 2 * time.Duration(cfg.LeaseSeconds*float64(time.Second)) + reconciler := reconciler.New(store, workerID(), reconcileGrace, + time.Duration(cfg.ReconcileIntervalSeconds*float64(time.Second)), log) + sweepWG.Add(1) + go func() { + defer sweepWG.Done() + reconciler.Start(ctx) + }() + // Wait for shutdown: stop the claimer first (close(batches) drains the // workers), then wait for the in-flight deliveries. <-ctx.Done() diff --git a/worker/internal/config/config.go b/worker/internal/config/config.go index 3e2cf291..e3497932 100644 --- a/worker/internal/config/config.go +++ b/worker/internal/config/config.go @@ -51,6 +51,11 @@ type Config struct { LeaseSeconds float64 MaxPerMinutePerTarget int + // ReconcileIntervalSeconds spaces the ambiguous-delivery reconciliation + // passes. The reconciler only closes UNKNOWN rows whose recorded failure + // proves no Telegram side effect; everything else stays visible in /ops. + ReconcileIntervalSeconds float64 + // Telegram Runtime (gRPC). TelegramRuntimeAddr string // Internal RPC token — the same shared secret the Go API already uses @@ -84,6 +89,7 @@ func FromEnv() (*Config, error) { BatchSize: envInt("ATF_QUEUE_BATCH_SIZE", 5), MaxQueueSize: envInt("ATF_QUEUE_MAX_SIZE", 2000), LeaseSeconds: envFloat("ATF_QUEUE_LEASE_SECONDS", 30.0), + ReconcileIntervalSeconds: envFloat("ATF_RECONCILE_INTERVAL_SECONDS", 60.0), MaxPerMinutePerTarget: envInt("ATF_QUEUE_MAX_PER_MINUTE", 0), TelegramRuntimeAddr: envStr("ATF_TELEGRAM_RUNTIME_ADDR", "core:50053"), CoreRPCToken: os.Getenv("ATF_CORE_RPC_TOKEN"), diff --git a/worker/internal/domain/job/job.go b/worker/internal/domain/job/job.go index 3b870b44..c6d02e2a 100644 --- a/worker/internal/domain/job/job.go +++ b/worker/internal/domain/job/job.go @@ -106,6 +106,16 @@ func (k AttemptKey) String() string { k.SourceMessageID, k.TargetChatID, suffix) } +// UnknownFence is an ambiguous delivery attempt awaiting reconciliation: a +// send whose outcome Telegram never reported, deliberately left UNKNOWN so +// nothing dispatches it twice. LastError carries the evidence a reconciler +// needs to decide whether the operation provably never executed. +type UnknownFence struct { + Key AttemptKey + LastError string + UpdatedAt int64 +} + // fenceOutcome mirrors what delivery_attempts.claim_or_check returns. // The vocabulary is shared with Python on purpose: only "dispatch" allows a // send, and SENT/UNKNOWN are never re-dispatched. diff --git a/worker/internal/domain/rule/addon_alias_test.go b/worker/internal/domain/rule/addon_alias_test.go new file mode 100644 index 00000000..742dcb82 --- /dev/null +++ b/worker/internal/domain/rule/addon_alias_test.go @@ -0,0 +1,92 @@ +package rule + +// The paid-addon entitlement gate must honor BOTH spellings: the canonical +// engine key and the short form the catalogue clients (bot rows, web +// dashboard) actually record. Production shipped three paid add-ons that +// silently never fired from any real client because the gate read a key +// nothing wrote — the worst outcome for a feature a user paid for. + +import ( + "reflect" + "testing" +) + +func TestRuleAddonEnabledHonoursCatalogueAliases(t *testing.T) { + cases := []struct { + name string + meta map[string]any + key string + want bool + }{ + { + "canonical key still works", + map[string]any{"paid_addons": map[string]any{"keyword_reply": true}}, + "keyword_reply", true, + }, + { + "bot catalogue spelling (kw_reply) enables the gate", + map[string]any{"paid_addons": map[string]any{"kw_reply": true}}, + "keyword_reply", true, + }, + { + "web dashboard spelling (auto_del) enables auto delete", + map[string]any{"paid_addons": map[string]any{"auto_del": true}}, + "auto_delete_message", true, + }, + { + "web dashboard spelling (affiliate) enables affiliate kit", + map[string]any{"paid_addons": map[string]any{"affiliate": true}}, + "affiliate_kit", true, + }, + { + "alias false is still false", + map[string]any{"paid_addons": map[string]any{"kw_reply": false}}, + "keyword_reply", false, + }, + { + "canonical false does not inherit a sibling alias of another addon", + map[string]any{"paid_addons": map[string]any{"auto_del": true}}, + "keyword_reply", false, + }, + { + "unknown key has no alias", + map[string]any{"paid_addons": map[string]any{"kw_reply": true}}, + "crypto_mode", false, + }, + { + "non-dict paid_addons means no entitlement", + map[string]any{"paid_addons": "true"}, + "keyword_reply", false, + }, + { + "missing paid_addons means no entitlement", + map[string]any{}, + "keyword_reply", false, + }, + } + for _, tc := range cases { + if got := ruleAddonEnabled(tc.meta, tc.key); got != tc.want { + t.Errorf("%s: ruleAddonEnabled(%q) = %v, want %v", tc.name, tc.key, got, tc.want) + } + } +} + +// The keyword-reply projection — the exact path the delivery engine's +// classifier consumes — must see the entitlement through the alias too. +func TestKeywordSettingsSeesTheCatalogueAlias(t *testing.T) { + r := &ForwardRule{ + Metadata: map[string]any{ + "keyword_confirmation": true, + "keyword_reply": "hello!", + "whitelist": []any{"hi"}, + "paid_addons": map[string]any{"kw_reply": true}, + }, + } + got := r.KeywordSettings() + if !got.AddonEnabled { + t.Fatal("kw_reply entitlement did not reach the keyword-reply gate — the production client spelling is still dead") + } + if !got.KeywordConfirmation || got.KeywordReply != "hello!" || !reflect.DeepEqual(got.Whitelist, []string{"hi"}) { + t.Errorf("projection drifted: %+v", got) + } +} diff --git a/worker/internal/domain/rule/rule.go b/worker/internal/domain/rule/rule.go index 4514619a..af18176e 100644 --- a/worker/internal/domain/rule/rule.go +++ b/worker/internal/domain/rule/rule.go @@ -264,16 +264,37 @@ func ruleMetaStr(meta map[string]any, key, def string) string { return def } +// addonKeyAliases maps the catalogue clients' paid-addon spellings onto the +// canonical engine gate keys — the twin of ForwardRule.ADDON_KEY_ALIASES +// (entities.py). The bot catalogue rows and the web dashboard's add-ons +// panel record the short spellings (kw_reply, auto_del, affiliate); a paid +// add-on whose entitlement was recorded by any client must fire — silently +// never firing is the worst outcome for a feature a user paid for. +var addonKeyAliases = map[string][]string{ + "keyword_reply": {"kw_reply"}, + "auto_delete_message": {"auto_del"}, + "affiliate_kit": {"affiliate"}, +} + // ruleAddonEnabled is ForwardRule.is_addon_enabled (entities.py:1843): // the addon must be marked enabled in the persisted metadata's // paid_addons block. A client-supplied flag alone is never sufficient, -// and a non-dict paid_addons block means no entitlement at all. +// and a non-dict paid_addons block means no entitlement at all. Canonical +// keys and the catalogue clients' recorded aliases are both honored. func ruleAddonEnabled(meta map[string]any, key string) bool { addons, ok := meta["paid_addons"].(map[string]any) if !ok { return false } - return ruleMetaBool(addons, key, false) + if ruleMetaBool(addons, key, false) { + return true + } + for _, alias := range addonKeyAliases[key] { + if ruleMetaBool(addons, alias, false) { + return true + } + } + return false } // whitelistWords is ForwardRule.whitelist (entities.py:1467): the diff --git a/worker/internal/engine/delivery/delivery.go b/worker/internal/engine/delivery/delivery.go index 1d43627e..a06fc4fc 100644 --- a/worker/internal/engine/delivery/delivery.go +++ b/worker/internal/engine/delivery/delivery.go @@ -169,10 +169,10 @@ type Engine struct { // Python pipeline's own vocabulary (ai_transforms_*, // ai_fallback_*). Nil is "no recorder". aiMetrics AIMetrics - log Logger - metrics Metrics - opts Options - workerID string + log Logger + metrics Metrics + opts Options + workerID string mu sync.Mutex rate map[string][]time.Time // per-target send timestamps @@ -263,10 +263,11 @@ func (e *Engine) failureToRing(ferr *telegraph.DeliveryFailure, j *job.DeliveryJ return } e.ops.RecordError("worker.delivery", "ERROR", ferr.Message, map[string]string{ - "code": ferr.Code, - "error_type": ferr.ErrorType, - "rule_id": j.RuleID, - "job_id": j.ID, + "code": ferr.Code, + "error_type": ferr.ErrorType, + "rule_id": j.RuleID, + "job_id": j.ID, + "correlation_id": j.Payload.CorrelationID(), }) } @@ -343,6 +344,7 @@ func (e *Engine) Process(ctx context.Context, j *job.DeliveryJob) error { logger.Warn("rule is auto-paused — dropping job", "decision", "DROP", "reason_code", "RULE_AUTO_PAUSED", "reason", r.AutoPauseReason) _ = e.store.MarkFailed(ctx, j.ID, "Task paused: "+r.AutoPauseReason) e.metrics.Outcome("failed") + e.outcome("failed", j, jobStart) return nil } // A paused task is not a failure: the job waits for the pause to lift. @@ -375,12 +377,15 @@ func (e *Engine) Process(ctx context.Context, j *job.DeliveryJob) error { text := j.Payload.Text() res := e.filters.Evaluate(text, j.Payload.MediaType(), false, f) if res.Action == filter.Drop { + // A deliberate drop is stage SKIPPED-BY-POLICY, never stage + // FAILED: the filter working as configured is healthy behavior, + // and counting it as a stage failure degraded a healthy task's + // ops panel. e.stageSkipped("filter", fStart) - e.stageFail("filter", res.Reason) logger.Warn("job dropped by attached filter at delivery time", "decision", "DROP", "reason_code", "FILTER_RULE_REJECTED", "filter_reason", res.Reason, "filter_id", f.ID) _ = e.store.MarkFailed(ctx, j.ID, "Filter: "+res.Reason) - e.metrics.Outcome("failed") - e.outcome("failed", j, jobStart) + e.metrics.Outcome("rejected") + e.outcome("rejected", j, jobStart) return nil } e.stageRecord("filter", fStart) @@ -406,7 +411,7 @@ func (e *Engine) Process(ctx context.Context, j *job.DeliveryJob) error { if r.BlockLinks && policy.HasLink(j.Payload.Text()) { logger.Warn("job dropped by task-level block_links at delivery time", "decision", "DROP", "reason_code", "LINK_BLOCKED") _ = e.store.MarkFailed(ctx, j.ID, "Filter: link_blocked") - e.metrics.Outcome("failed") + e.metrics.Outcome("rejected") return nil } @@ -471,7 +476,6 @@ func (e *Engine) Process(ctx context.Context, j *job.DeliveryJob) error { } } - // 2d. Stage 4.5 (W3) — event-level dedup: REMOVED (production defect). // // The Python pipeline writes the processed_messages marker at ENQUEUE @@ -613,7 +617,7 @@ func (e *Engine) Process(ctx context.Context, j *job.DeliveryJob) error { // "durable enqueue must happen before the branding side effect". // The job is claimed, the send fence is held, and stage 1 runs as part // of the claimed execution — never before enqueue. - if err := e.ensureBrandingStage1(ctx, j, r, logger); err != nil { + if err := e.ensureBrandingStage1(ctx, j, r, logger, jobStart); err != nil { return err } @@ -655,7 +659,7 @@ func (e *Engine) Process(ctx context.Context, j *job.DeliveryJob) error { if ferr != nil { e.stageFail("delivery", ferr.Code) e.failureToRing(ferr, j) - return e.handleFailure(ctx, j, logger, key, ferr) + return e.handleFailure(ctx, j, logger, key, ferr, jobStart) } // 5. Confirmed. Record the send on the fence BEFORE the job-row commit: @@ -751,7 +755,7 @@ func (e *Engine) dispatch(ctx context.Context, j *job.DeliveryJob, r *rule.Forwa if ferr == nil { return id, nil } - if !ferr.IsForwardingRestricted() || !p.FallbackEnabled() { + if !ferr.NeedsReupload() || !p.FallbackEnabled() { return id, ferr } // Restricted from C with the fallback allowed: fall through to @@ -778,7 +782,7 @@ func (e *Engine) dispatch(ctx context.Context, j *job.DeliveryJob, r *rule.Forwa if ferr == nil && len(ids) > 0 { return firstNonZero(ids), nil } - if ferr != nil && !ferr.IsForwardingRestricted() { + if ferr != nil && !ferr.NeedsReupload() { return 0, ferr } // Restricted: fall through to copy when fallback is allowed. @@ -850,7 +854,7 @@ func (e *Engine) dispatch(ctx context.Context, j *job.DeliveryJob, r *rule.Forwa if p.HasMedia() { id, ferr := e.gateway.CopyMessage(callCtx, sessionID, target, src, srcMsg, "", p.TopicID(), j.Payload.DropSourceButtons()) e.metrics.ObserveRPC("copy_message", 0) - if ferr != nil && ferr.IsForwardingRestricted() && p.FallbackEnabled() { + if ferr != nil && ferr.NeedsReupload() && p.FallbackEnabled() { return e.reuploadMedia(callCtx, j, src, srcMsg, "") } return id, ferr @@ -862,7 +866,7 @@ func (e *Engine) dispatch(ctx context.Context, j *job.DeliveryJob, r *rule.Forwa if p.HasMedia() { id, ferr := e.gateway.CopyMessage(callCtx, sessionID, target, src, srcMsg, caption, p.TopicID(), j.Payload.DropSourceButtons()) e.metrics.ObserveRPC("copy_message", 0) - if ferr != nil && ferr.IsForwardingRestricted() && p.FallbackEnabled() { + if ferr != nil && ferr.NeedsReupload() && p.FallbackEnabled() { return e.reuploadMedia(callCtx, j, src, srcMsg, caption) } return id, ferr @@ -887,7 +891,7 @@ func (e *Engine) brandingStage(ctx context.Context, sessionID string, target, cC if ferr == nil && len(ids) > 0 { return firstNonZero(ids), nil } - if ferr != nil && !ferr.IsForwardingRestricted() { + if ferr != nil && !ferr.NeedsReupload() { return 0, ferr } // restricted: fall through to a clean copy of C's post @@ -901,7 +905,7 @@ func (e *Engine) brandingStage(ctx context.Context, sessionID string, target, cC id, ferr := e.gateway.CopyMessage(ctx, sessionID, target, cChat, cID, caption, j.Payload.TopicID(), j.Payload.DropSourceButtons()) e.metrics.ObserveRPC("copy_message", 0) if ferr != nil { - if ferr.IsForwardingRestricted() && j.Payload.FallbackEnabled() { + if ferr.NeedsReupload() && j.Payload.FallbackEnabled() { id, ferr = e.reuploadMedia(ctx, j, cChat, cID, caption) if ferr != nil { return 0, ferr @@ -937,7 +941,7 @@ func (e *Engine) dispatchAlbum(ctx context.Context, j *job.DeliveryJob, albumIDs case "first": id, ferr := e.gateway.CopyMessage(ctx, sessionID, target, src, albumIDs[0], caption, p.TopicID(), j.Payload.DropSourceButtons()) e.metrics.ObserveRPC("copy_message", 0) - if ferr != nil && ferr.IsForwardingRestricted() && p.FallbackEnabled() { + if ferr != nil && ferr.NeedsReupload() && p.FallbackEnabled() { return e.reuploadMedia(ctx, j, src, albumIDs[0], caption) } return id, ferr @@ -968,7 +972,7 @@ func (e *Engine) dispatchAlbum(ctx context.Context, j *job.DeliveryJob, albumIDs // the retry resumes rather than duplicates: the items before // this one are physically in the target now. _ = e.store.RecordAlbumProgress(ctx, j.ID, newSent) - if ferr.IsForwardingRestricted() && p.FallbackEnabled() { + if ferr.NeedsReupload() && p.FallbackEnabled() { id, ferr = e.reuploadMedia(ctx, j, src, itemID, caption) if ferr != nil { return 0, ferr @@ -1001,12 +1005,12 @@ func (e *Engine) dispatchAlbum(ctx context.Context, j *job.DeliveryJob, albumIDs if ferr == nil && len(ids) > 0 { return firstNonZero(ids), nil } - if ferr != nil && !ferr.IsForwardingRestricted() { + if ferr != nil && !ferr.NeedsReupload() { // The group copy failed for a non-restriction reason: fall through to // the per-item re-upload (the Python path does the same). e.jobLogger(j).Warn("album group copy failed — falling back to re-upload", "error", ferr.Message) } - if ferr != nil && ferr.IsForwardingRestricted() && !p.FallbackEnabled() { + if ferr != nil && ferr.NeedsReupload() && !p.FallbackEnabled() { return 0, ferr } // Per-item re-upload fallback. @@ -1029,12 +1033,20 @@ func (e *Engine) dispatchAlbum(ctx context.Context, j *job.DeliveryJob, albumIDs // // Media never enters either process's memory in full: the runtime writes to // the shared scratch path, and the upload is staged from that path. +// +// The staged name MUST carry the source media's extension. Telegram's +// messages.SendMedia infers the MIME type from the staged file's name — +// a bare "job--media" has none, and the upload dies with +// PHOTO_EXT_INVALID before it can ever succeed (12 production jobs on rule +// 40edad5a failed exactly that way: the copy was correctly rejected, the +// re-upload ran, and the extension-less staging file killed it). func (e *Engine) reuploadMedia(ctx context.Context, j *job.DeliveryJob, srcChat, srcMsg int64, caption string) (int64, *telegraph.DeliveryFailure) { p := &j.Payload sessionID := p.SessionID() + ext := e.stagedExtension(ctx, sessionID, j, srcChat, srcMsg) dest := telegraph.MediaHandle{ - Path: e.scratchPath(j), - FileName: fmt.Sprintf("job-%s-media", j.ID), + Path: e.scratchPath(j) + ext, + FileName: fmt.Sprintf("job-%s-media%s", j.ID, ext), } if _, ferr := e.gateway.DownloadMedia(ctx, sessionID, srcChat, srcMsg, dest); ferr != nil { return 0, ferr @@ -1057,6 +1069,62 @@ func (e *Engine) reuploadMedia(ctx context.Context, j *job.DeliveryJob, srcChat, return id, sferr } +// stagedExtension resolves the extension the re-upload must be staged under. +// The runtime reports the source media's real one (the original file_name's +// suffix, .jpg for photos); when it cannot answer — source gone since the +// copy was refused, or a kind with no inferable name — fall back to the +// ingest classification's canonical extension. Documents and unknown kinds +// stay extension-less on purpose: send_document accepts them, and guessing +// would mislabel real files. +func (e *Engine) stagedExtension(ctx context.Context, sessionID string, j *job.DeliveryJob, srcChat, srcMsg int64) string { + if ext, ferr := e.gateway.GetFileExtension(ctx, sessionID, srcChat, srcMsg); ferr == nil { + if norm := normalizeExtension(ext); norm != "" { + return norm + } + } + return defaultExtension(j.Payload.MediaType()) +} + +// normalizeExtension sanitizes a runtime-reported extension into something +// safe to append to a staging path: lowercase, dot-prefixed, short, plain +// [a-z0-9] only. A malformed or hostile file_name suffix must never become +// a path surprise; unusable values normalize to "". +func normalizeExtension(ext string) string { + ext = strings.ToLower(strings.TrimSpace(ext)) + ext = strings.TrimPrefix(ext, ".") + if ext == "" || len(ext) > 15 { + return "" + } + for _, r := range ext { + if (r < 'a' || r > 'z') && (r < '0' || r > '9') { + return "" + } + } + return "." + ext +} + +// defaultExtension maps the ingest-side media classification onto the +// canonical extension Telegram expects for that kind. Only used when the +// runtime cannot report the real one. The mapping mirrors send_media's +// dispatch: the kind decides the send call, the send call's validation +// decides which extensions it accepts. +func defaultExtension(mediaType string) string { + switch strings.ToLower(strings.TrimSpace(mediaType)) { + case "photo": + return ".jpg" + case "video", "round_video", "animation", "video_note": + return ".mp4" + case "voice": + return ".ogg" + case "audio": + return ".mp3" + case "sticker": + return ".webp" + default: + return "" + } +} + // scratchPath is the unique per-job path in the shared media scratch volume. // Unique by construction (job ids are unique), so two concurrent re-uploads // can never collide on the same file. @@ -1079,7 +1147,7 @@ func (e *Engine) scratchPath(j *job.DeliveryJob) string { // the attempt budget is not wasted on something that cannot succeed. // - anything else → MarkRetry with exponential backoff, dead-lettering // when the budget is spent. -func (e *Engine) handleFailure(ctx context.Context, j *job.DeliveryJob, logger Logger, key job.AttemptKey, ferr *telegraph.DeliveryFailure) error { +func (e *Engine) handleFailure(ctx context.Context, j *job.DeliveryJob, logger Logger, key job.AttemptKey, ferr *telegraph.DeliveryFailure, jobStart time.Time) error { // Ambiguous: Telegram may have accepted the send. Never a plain FAILED, // never an automatic re-dispatch — reconciliation owns this row. if ferr.IsAmbiguous() { @@ -1091,6 +1159,7 @@ func (e *Engine) handleFailure(ctx context.Context, j *job.DeliveryJob, logger L logger.Warn("mark_unknown failed", "error", err) } e.metrics.Outcome("unknown") + e.outcome("unknown", j, jobStart) return nil } @@ -1107,6 +1176,7 @@ func (e *Engine) handleFailure(ctx context.Context, j *job.DeliveryJob, logger L "retry_after_seconds", ferr.RetryAfterSeconds) _ = e.store.Defer(ctx, j.ID, "FloodWait: "+ferr.Message, wait) e.metrics.Outcome("retried") + e.outcome("retried", j, jobStart) return nil } @@ -1118,6 +1188,7 @@ func (e *Engine) handleFailure(ctx context.Context, j *job.DeliveryJob, logger L logger.Warn("dependency unavailable — deferring", "code", ferr.Code) _ = e.store.Defer(ctx, j.ID, ferr.Code+": "+ferr.Message, 10*time.Second) e.metrics.Outcome("retried") + e.outcome("retried", j, jobStart) return nil } @@ -1130,6 +1201,7 @@ func (e *Engine) handleFailure(ctx context.Context, j *job.DeliveryJob, logger L "session_id", redacted(j.Payload.SessionID())) _ = e.store.MarkFailed(ctx, j.ID, "Auth: "+ferr.Message) e.metrics.Outcome("failed") + e.outcome("failed", j, jobStart) return nil } @@ -1141,6 +1213,7 @@ func (e *Engine) handleFailure(ctx context.Context, j *job.DeliveryJob, logger L "code", ferr.Code, "error_type", ferr.ErrorType) _ = e.store.MarkFailed(ctx, j.ID, ferr.Code+": "+ferr.Message) e.metrics.Outcome("failed") + e.outcome("failed", j, jobStart) return nil } @@ -1155,6 +1228,7 @@ func (e *Engine) handleFailure(ctx context.Context, j *job.DeliveryJob, logger L logger.Warn("mark_retry failed", "error", err) } e.metrics.Outcome("retried") + e.outcome("retried", j, jobStart) return nil } @@ -1219,6 +1293,7 @@ func (e *Engine) recordSend(target string) { func (e *Engine) jobLogger(j *job.DeliveryJob) Logger { return &fieldLogger{base: e.log, fields: []any{ "correlation_id", j.Payload.CorrelationID(), + "worker_id", e.workerID, "job_id", j.ID, "rule_id", j.RuleID, "owner_user_id", j.Payload.OwnerUserID(), @@ -1321,7 +1396,7 @@ func messageIDs(list any, single any) []int64 { // Ordering guarantee: the send fence is already held, the delivery job is // CLAIMED, and stage 1 runs as part of the claimed execution — never before // the job is durable. -func (e *Engine) ensureBrandingStage1(ctx context.Context, j *job.DeliveryJob, r *rule.ForwardRule, logger Logger) error { +func (e *Engine) ensureBrandingStage1(ctx context.Context, j *job.DeliveryJob, r *rule.ForwardRule, logger Logger, jobStart time.Time) error { plan := readHopPlan(j) if plan == nil || !plan.isIntent() { // Not an intent: either unbranded, or a legacy already-performed @@ -1375,7 +1450,7 @@ func (e *Engine) ensureBrandingStage1(ctx context.Context, j *job.DeliveryJob, r SourceMessageID: j.SourceMessageID, TargetChatID: j.TargetChatID, JobKind: j.JobKind, } - _ = e.handleFailure(ctx, j, logger, key, ferr) + _ = e.handleFailure(ctx, j, logger, key, ferr, jobStart) return fmt.Errorf("branding stage 1 failed: %s", ferr.Message) } diff --git a/worker/internal/engine/delivery/delivery_test.go b/worker/internal/engine/delivery/delivery_test.go index 5c758c37..422be1da 100644 --- a/worker/internal/engine/delivery/delivery_test.go +++ b/worker/internal/engine/delivery/delivery_test.go @@ -3,6 +3,7 @@ package delivery import ( "context" "database/sql" + "path/filepath" "strings" "testing" "time" @@ -22,7 +23,7 @@ func (l *testLogger) Info(msg string, args ...any) { l.lines = append(l.lines, func (l *testLogger) Warn(msg string, args ...any) { l.lines = append(l.lines, "WARN "+msg) } func (l *testLogger) Error(msg string, args ...any) { l.lines = append(l.lines, "ERROR "+msg) } -type testMetrics struct{ sent, retried, failed, unknown int } +type testMetrics struct{ sent, retried, failed, rejected, unknown int } func (m *testMetrics) Outcome(kind string) { switch kind { @@ -32,6 +33,8 @@ func (m *testMetrics) Outcome(kind string) { m.retried++ case "failed": m.failed++ + case "rejected": + m.rejected++ case "unknown": m.unknown++ } @@ -661,8 +664,10 @@ func TestTaskBlockLinksDropsLinkBearingJob(t *testing.T) { } assertJobStatus(t, h.store, j.ID, string(job.StatusFailed)) - if h.metrics.failed != 1 { - t.Errorf("failed = %d, want 1", h.metrics.failed) + // A deliberate link block is a policy REJECTION, not a delivery failure: + // counting it as "failed" made a healthy task look degraded in ops. + if h.metrics.failed != 0 || h.metrics.rejected != 1 { + t.Errorf("outcomes: failed=%d rejected=%d, want rejected=1 failed=0", h.metrics.failed, h.metrics.rejected) } if got := jobReason(t, h.store, j.ID); !strings.Contains(got, "link_blocked") { t.Errorf("job reason = %q, want it to contain link_blocked", got) @@ -810,3 +815,263 @@ func TestTaskBlockLinksDropLeavesNoFenceClaim(t *testing.T) { t.Errorf("a link-blocked drop left a fence row in %s state — the job is unresolved", outcome) } } + +// TestMediaForwardTargetRejectsCopy: the production defect this test pins: +// a target chat rejects the copied media (PHOTO_EXT_INVALID — the copy +// itself is permitted, the payload is incompatible), and the job must fall +// back to re-upload rather than die as a permanent failure. +// +// 12 real jobs on rule 40edad5a failed exactly this way: the fallback +// trigger only ever asked IsForwardingRestricted(), so a media-rejection +// error went straight to FAILED and the message was never delivered. +// NeedsReupload covers both reasons for re-uploading. +func TestMediaForwardTargetRejectsCopy(t *testing.T) { + h := newHarness(t) + j := &job.DeliveryJob{ + ID: "r1:c:9:t", RuleID: "r1", SourceChatID: "c", SourceMessageID: 9, + TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5, + Payload: *jobPayload(map[string]any{ + "text": "caption", "has_media": true, "media_type": "photo", + "fallback_enabled": true, + }), + } + h.enqueue(t, j) + h.gateway.Reset() + h.gateway.CopyMediaRejected = true + + if err := h.engine.Process(context.Background(), j); err != nil { + t.Fatalf("process media-rejected: %v", err) + } + // The rejected copy must NOT fail the job: re-upload delivers it. + assertJobStatus(t, h.store, j.ID, string(job.StatusSent)) + downs := 0 + for _, c := range h.gateway.Recorded() { + if c.Op == "DownloadMedia" { + downs++ + } + } + if downs == 0 { + t.Error("target media rejection with fallback_enabled did not stage a re-upload download — the job would have been marked FAILED without trying the recovery") + } +} + +// TestMediaRejectedNoFallback: without the task's opt-in, a media-rejected +// copy is a failure, not a silent re-upload. The operator chose no +// fallback; honoring that is not optional. +func TestMediaRejectedNoFallback(t *testing.T) { + h := newHarness(t) + j := &job.DeliveryJob{ + ID: "r1:c:10:t", RuleID: "r1", SourceChatID: "c", SourceMessageID: 10, + TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5, + Payload: *jobPayload(map[string]any{ + "text": "caption", "has_media": true, "media_type": "photo", + "fallback_enabled": false, + }), + } + h.enqueue(t, j) + h.gateway.Reset() + h.gateway.CopyMediaRejected = true + + if err := h.engine.Process(context.Background(), j); err != nil { + t.Fatalf("process: %v", err) + } + downs := 0 + for _, c := range h.gateway.Recorded() { + if c.Op == "DownloadMedia" { + downs++ + } + } + if downs > 0 { + t.Errorf("re-upload ran with fallback_enabled=false — the operator's opt-out was not honored (%d downloads)", downs) + } +} + +// --- the re-upload staging contract (the PHOTO_EXT_INVALID root cause) ----- // + +// stagedPaths extracts the recorded DownloadMedia / PrepareMediaUpload +// staging paths (last of each op wins; re-upload records exactly one of +// each per delivery attempt). +func stagedPaths(h *harness) (down, prep string) { + for _, c := range h.gateway.Recorded() { + switch c.Op { + case "DownloadMedia": + down = c.MediaPath + case "PrepareMediaUpload": + prep = c.MediaPath + } + } + return down, prep +} + +// reuploadHarness builds a photo job whose copy is rejected by the target +// (PHOTO_EXT_INVALID) with the fallback allowed — the shape that reaches +// the re-upload staging path in production. +func reuploadHarness(t *testing.T) (*harness, *job.DeliveryJob) { + h := newHarness(t) + j := &job.DeliveryJob{ + ID: "r1:c:20:t", RuleID: "r1", SourceChatID: "c", SourceMessageID: 20, + TargetChatID: "t", Status: job.StatusPending, MaxAttempts: 5, + Payload: *jobPayload(map[string]any{ + "text": "caption", "has_media": true, "media_type": "photo", + "fallback_enabled": true, + }), + } + h.enqueue(t, j) + h.gateway.Reset() + h.gateway.CopyMediaRejected = true + return h, j +} + +// TestReuploadStagesFileWithExtension pins the root-cause fix: Telegram's +// messages.SendMedia infers the MIME type from the staged file's name, so +// the scratch file must carry the source media's extension. The runtime +// reports ".JPG" here; the staged path must carry it normalized — and +// PrepareMediaUpload must reference the exact file DownloadMedia wrote. +// 12 production jobs died because the staged name was bare "job--media". +func TestReuploadStagesFileWithExtension(t *testing.T) { + h, j := reuploadHarness(t) + ext := ".JPG" + h.gateway.ExtensionOverride = &ext + + if err := h.engine.Process(context.Background(), j); err != nil { + t.Fatalf("process: %v", err) + } + assertJobStatus(t, h.store, j.ID, string(job.StatusSent)) + + down, prep := stagedPaths(h) + if down == "" { + t.Fatal("no DownloadMedia recorded for the re-upload") + } + if !strings.HasSuffix(down, ".jpg") { + t.Errorf("staged path %q carries no normalized media extension — messages.SendMedia would reject it with PHOTO_EXT_INVALID again", down) + } + if prep != down { + t.Errorf("PrepareMediaUpload staged %q but DownloadMedia wrote %q — the upload would not find the file", prep, down) + } +} + +// TestReuploadExtensionFallsBackToIngestClassification: when the runtime +// cannot tell the extension (source gone, no inferable name), the staging +// name falls back to the ingest-side media kind — never back to bare. +func TestReuploadExtensionFallsBackToIngestClassification(t *testing.T) { + h, j := reuploadHarness(t) + ext := "" + h.gateway.ExtensionOverride = &ext + + if err := h.engine.Process(context.Background(), j); err != nil { + t.Fatalf("process: %v", err) + } + down, _ := stagedPaths(h) + if !strings.HasSuffix(down, ".jpg") { + t.Errorf("unknown runtime extension must fall back to the media kind's canonical extension, staged %q", down) + } +} + +// TestReuploadDocumentStaysExtensionless: a document (send_document path) +// has no canonical extension to guess, and Telegram accepts a bare +// document — a wrong guess would mislabel real files, so none is made. +func TestReuploadDocumentStaysExtensionless(t *testing.T) { + h, j := reuploadHarness(t) + j.Payload = *jobPayload(map[string]any{ + "text": "caption", "has_media": true, "media_type": "document", + "fallback_enabled": true, + }) + ext := "" + h.gateway.ExtensionOverride = &ext + + if err := h.engine.Process(context.Background(), j); err != nil { + t.Fatalf("process: %v", err) + } + down, _ := stagedPaths(h) + if down == "" { + t.Fatal("no DownloadMedia recorded for the re-upload") + } + if strings.HasSuffix(down, j.ID) == false { + t.Errorf("document staging must stay extension-less (send_document accepts bare files), staged %q", down) + } +} + +// TestReuploadHostileExtensionNeverReachesPath: a runtime-reported suffix +// is untrusted input. A path-shaped or oversized value must normalize away +// to the media-kind default — never ride into the staging path. +func TestReuploadHostileExtensionNeverReachesPath(t *testing.T) { + h, j := reuploadHarness(t) + ext := "/../../etc/passwd" + h.gateway.ExtensionOverride = &ext + + if err := h.engine.Process(context.Background(), j); err != nil { + t.Fatalf("process: %v", err) + } + down, _ := stagedPaths(h) + if filepath.Base(down) != j.ID+".jpg" { + t.Errorf("hostile extension reached the staging path: base=%q, want %q", filepath.Base(down), j.ID+".jpg") + } +} + +// TestReuploadExtensionRPCFailureFallsBack: the extension probe failing +// (source already gone, runtime degraded) must not abort the recovery — +// the ingest classification answers instead. +func TestReuploadExtensionRPCFailureFallsBack(t *testing.T) { + h, j := reuploadHarness(t) + h.gateway.GetFileExtensionFailsWith = &telegraph.DeliveryFailure{ + Code: "MESSAGE_NOT_FOUND", ErrorType: "MESSAGE_NOT_FOUND", + Message: "source message gone since the copy was refused", + } + + if err := h.engine.Process(context.Background(), j); err != nil { + t.Fatalf("process: %v", err) + } + assertJobStatus(t, h.store, j.ID, string(job.StatusSent)) + down, _ := stagedPaths(h) + if !strings.HasSuffix(down, ".jpg") { + t.Errorf("extension probe failure must fall back to the media kind, staged %q", down) + } +} + +// TestNormalizeExtension: the sanitizer's full contract — lowercases, +// dot-prefixes, and rejects path-shaped, oversized, or non-ASCII values. +func TestNormalizeExtension(t *testing.T) { + cases := []struct{ in, want string }{ + {".JPG", ".jpg"}, + {"jpg", ".jpg"}, + {".jpeg", ".jpeg"}, + {".mp4", ".mp4"}, + {" .png ", ".png"}, + {"", ""}, + {".", ""}, + {".tar.gz", ""}, + {"../secrets", ""}, + {".café", ""}, + {".sixteen-char-ext", ""}, + {".ok16", ".ok16"}, + } + for _, tc := range cases { + if got := normalizeExtension(tc.in); got != tc.want { + t.Errorf("normalizeExtension(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} + +// TestDefaultExtension: the ingest-kind fallback mirrors send_media's +// dispatch — every kind that maps to a typed send call gets that call's +// canonical extension; documents and unknown kinds stay bare. +func TestDefaultExtension(t *testing.T) { + cases := []struct{ in, want string }{ + {"photo", ".jpg"}, + {"video", ".mp4"}, + {"round_video", ".mp4"}, + {"animation", ".mp4"}, + {"video_note", ".mp4"}, + {"voice", ".ogg"}, + {"audio", ".mp3"}, + {"sticker", ".webp"}, + {"document", ""}, + {"", ""}, + {"exotic", ""}, + } + for _, tc := range cases { + if got := defaultExtension(tc.in); got != tc.want { + t.Errorf("defaultExtension(%q) = %q, want %q", tc.in, got, tc.want) + } + } +} diff --git a/worker/internal/engine/delivery/forward_every_n.go b/worker/internal/engine/delivery/forward_every_n.go index 11f0784a..3fbf65f9 100644 --- a/worker/internal/engine/delivery/forward_every_n.go +++ b/worker/internal/engine/delivery/forward_every_n.go @@ -132,7 +132,9 @@ func (e *Engine) forwardEveryNGate(ctx context.Context, j *job.DeliveryJob, r *r "rule_id", j.RuleID, "message_id", j.SourceMessageID, "sequence", sequence, "every_n", everyN) _ = e.store.MarkFailed(ctx, j.ID, "Forward every N: skipped by cadence") - e.metrics.Outcome("failed") + // A cadence skip is the add-on working exactly as configured — counted + // as "rejected" (a deliberate non-delivery), never as "failed". + e.metrics.Outcome("rejected") // The sentinel is the gate's terminal signal: Process must stop here so // no fence claim and no Telegram call follows a job that will not send. return errFenSkipped diff --git a/worker/internal/engine/delivery/forward_every_n_test.go b/worker/internal/engine/delivery/forward_every_n_test.go index c720d555..4ceef9e7 100644 --- a/worker/internal/engine/delivery/forward_every_n_test.go +++ b/worker/internal/engine/delivery/forward_every_n_test.go @@ -144,8 +144,8 @@ func TestFenEveryFifthMessageGoesThrough(t *testing.T) { if got := len(h.gateway.Recorded()); got != 3 { t.Errorf("gateway calls = %d, want 3 (messages 1, 6, 11)", got) } - if h.metrics.failed != 9 { - t.Errorf("failed outcomes = %d, want 9 (cadence skips close the job)", h.metrics.failed) + if h.metrics.rejected != 9 || h.metrics.failed != 0 { + t.Errorf("outcomes = %d failed / %d rejected, want 9 rejected (cadence skips are deliberate non-deliveries, not failures)", h.metrics.failed, h.metrics.rejected) } row, err := cadence.Peek(context.Background(), "r1") if err != nil || row == nil { @@ -175,8 +175,8 @@ func TestFenACadenceSkipIsNotAnAttempt(t *testing.T) { if calls := h.gateway.Recorded(); len(calls) != 0 { t.Errorf("gateway calls = %d, want 0 (a skip never reaches Telegram)", len(calls)) } - if h.metrics.sent != 0 || h.metrics.retried != 0 || h.metrics.failed != 1 { - t.Errorf("outcomes: sent=%d retried=%d failed=%d, want failed=1 only", h.metrics.sent, h.metrics.retried, h.metrics.failed) + if h.metrics.sent != 0 || h.metrics.retried != 0 || h.metrics.failed != 0 || h.metrics.rejected != 1 { + t.Errorf("outcomes: sent=%d retried=%d failed=%d rejected=%d, want rejected=1 only", h.metrics.sent, h.metrics.retried, h.metrics.failed, h.metrics.rejected) } if !strings.Contains(h.log.lines[0], "skipped by forward-every-n cadence") { t.Errorf("log lines = %v, want the cadence skip line first", h.log.lines) @@ -354,8 +354,8 @@ func TestFenAnEditNeverCountsOrUsesASlot(t *testing.T) { if got := len(h.gateway.Recorded()); got != 1 { t.Errorf("gateway calls = %d, want 1 (message 10 is held at sequence 2)", got) } - if h.metrics.failed != 1 { - t.Errorf("failed outcomes = %d, want 1 (message 10 skipped by cadence)", h.metrics.failed) + if h.metrics.rejected != 1 { + t.Errorf("rejected outcomes = %d, want 1 (message 10 skipped by cadence)", h.metrics.rejected) } } @@ -412,8 +412,8 @@ func TestFenAFilterDropNeverConsumesASlot(t *testing.T) { if err := engine.Process(context.Background(), j); err != nil { t.Fatalf("process: %v", err) } - if len(gw.Recorded()) != 0 || metrics.failed != 1 { - t.Errorf("gateway=%d outcomes: failed=%d, want a filter drop (failed=1)", len(gw.Recorded()), metrics.failed) + if len(gw.Recorded()) != 0 || metrics.rejected != 1 || metrics.failed != 0 { + t.Errorf("gateway=%d outcomes: failed=%d rejected=%d, want a filter drop (rejected=1, not a failure)", len(gw.Recorded()), metrics.failed, metrics.rejected) } if row, _ := cadence2.Peek(context.Background(), "r1"); row != nil { t.Errorf("counter row = %+v, want none (a filtered message never consumes a slot)", row) diff --git a/worker/internal/engine/delivery/transform_stage.go b/worker/internal/engine/delivery/transform_stage.go index 38fdceb0..b85ed22a 100644 --- a/worker/internal/engine/delivery/transform_stage.go +++ b/worker/internal/engine/delivery/transform_stage.go @@ -120,16 +120,21 @@ func loadTransformSwitch() bool { func (e *Engine) applyTransformPipeline(ctx context.Context, j *job.DeliveryJob, r *rule.ForwardRule, logger Logger) error { tStart := time.Now() defer func() { - // The transform pipeline IS the four text stages (cleaner, link - // policy/replacement, header/footer, AI). They run as one ordered + // The transform pipeline IS the text stages (cleaner, link + // policy/replacement, header/footer). They run as one ordered // ApplyText call, so one observation per stage in the aggregate is // the honest measurement: splitting them would mean re-timing a // library call that reports no per-stage clock. + // + // "ai" is NOT recorded here: most jobs carry no AI rewrite at all, + // and counting them made the stage read healthy-with-usage it never + // had — while jobs that DID rewrite were counted twice (once here, + // once per attempt by the AIMetricsObserver). The AI stage counts + // through its own observer only. ms := float64(time.Since(tStart).Microseconds()) / 1000.0 e.stageRecord("cleaner", tStart) e.stageRecord("replacement", tStart) e.stageRecord("header_footer", tStart) - e.stageRecord("ai", tStart) _ = ms }() raw := payloadRawText(j) diff --git a/worker/internal/engine/reconciler/reconciler.go b/worker/internal/engine/reconciler/reconciler.go new file mode 100644 index 00000000..32c44b32 --- /dev/null +++ b/worker/internal/engine/reconciler/reconciler.go @@ -0,0 +1,145 @@ +// Package reconciler closes ambiguous deliveries the fence deliberately +// left UNKNOWN. +// +// The delivery contract is: an outcome Telegram never reported (timeout, +// dropped connection) goes UNKNOWN and is "never retried blindly — +// reconciliation decides". But an ambiguity nobody ever inspects is a +// permanently stranded job: production accumulated UNKNOWN rows for a day +// because the decision step simply did not exist. This loop is that step. +// +// It closes only what the recorded evidence PROVES: a gRPC Unimplemented +// outcome means the runtime server received the call, found no method, and +// returned before any handler ran — no Telegram operation can have been +// attempted, so the message cannot be in the target. Everything else +// (timeouts, connection drops after accept, unknown outcomes) stays UNKNOWN +// and visible: guessing here would be exactly the duplicate delivery the +// fence exists to prevent. Operators see what remains through /ops/queue +// (the unknown counter) and the incidents surface. +package reconciler + +import ( + "context" + "strings" + "time" + + "atf/worker/internal/domain/job" +) + +// Store is the reconciliation surface the engine needs from the queue store. +type Store interface { + UnknownFences(ctx context.Context, olderThan time.Time, limit int) ([]job.UnknownFence, error) + ReconcileUnknown(ctx context.Context, key job.AttemptKey, workerID, reason string) (bool, error) +} + +// Logger is the structured logger contract (satisfied by the worker's logger). +type Logger interface { + Info(msg string, args ...any) + Warn(msg string, args ...any) + Error(msg string, args ...any) +} + +// maxBatch bounds one pass. A backlog of thousands of ambiguities resolves +// across consecutive ticks instead of one long-running sweep. +const maxBatch = 100 + +// Reconciler periodically resolves UNKNOWN fence rows older than the grace +// window when their recorded failure proves no Telegram side effect. +type Reconciler struct { + store Store + workerID string + grace time.Duration + interval time.Duration + log Logger +} + +// New builds a reconciler. grace should comfortably exceed the claim lease: +// a row still UNKNOWN long after every lease could have expired is one no +// live dispatch can explain. +func New(store Store, workerID string, grace, interval time.Duration, log Logger) *Reconciler { + return &Reconciler{store: store, workerID: workerID, grace: grace, interval: interval, log: log} +} + +// provableMarkers are the recorded failure fragments that PROVE the runtime +// rejected the call before any handler ran. gRPC reports Unimplemented with +// code 12; the Python runtime's server raises NotImplementedError for a +// method it does not register, and the Go client surfaces it as +// `code = Unimplemented desc = Unexpected …`. +var provableMarkers = []string{"code = Unimplemented"} + +// provablyUnexecuted reports whether the recorded failure proves the RPC +// never executed a Telegram operation. Anything not on this list stays +// UNKNOWN — ambiguity is resolved by evidence, never by hope. +func provablyUnexecuted(lastError string) bool { + for _, marker := range provableMarkers { + if strings.Contains(lastError, marker) { + return true + } + } + return false +} + +// RunOnce performs one reconciliation pass over the aged UNKNOWN fences. +// It returns how many were resolved and how many remain (of those inspected). +func (r *Reconciler) RunOnce(ctx context.Context) (resolved, remaining int, err error) { + fences, err := r.store.UnknownFences(ctx, time.Now().Add(-r.grace), maxBatch) + if err != nil { + return 0, 0, err + } + for _, f := range fences { + if !provablyUnexecuted(f.LastError) { + remaining++ + continue + } + reason := "reconciled: the runtime rejected the RPC (Unimplemented) before any handler ran — " + + "no Telegram operation was attempted, so the message cannot be in the target; original: " + + truncate(f.LastError, 200) + ok, rerr := r.store.ReconcileUnknown(ctx, f.Key, r.workerID, reason) + if rerr != nil { + r.log.Error("reconcile of an ambiguous delivery failed", + "worker_id", r.workerID, + "job_id", f.Key.String(), + "rule_id", f.Key.RuleID, + "source_message_id", f.Key.SourceMessageID, + "error", rerr.Error()) + remaining++ + continue + } + if ok { + resolved++ + } + } + return resolved, remaining, nil +} + +// Start runs the reconciliation loop until ctx is cancelled. +func (r *Reconciler) Start(ctx context.Context) { + ticker := time.NewTicker(r.interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + resolved, remaining, err := r.RunOnce(ctx) + if err != nil { + // A failed pass only delays recovery — the rows stay + // durable UNKNOWN, never lost. + r.log.Warn("reconciliation pass failed", "error", err) + continue + } + if resolved > 0 { + r.log.Warn("reconciler closed ambiguous deliveries", + "worker_id", r.workerID, + "component", "worker.reconciler", + "resolved", resolved, "left_unknown", remaining) + } + } + } +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/worker/internal/engine/reconciler/reconciler_test.go b/worker/internal/engine/reconciler/reconciler_test.go new file mode 100644 index 00000000..0bf4ff8e --- /dev/null +++ b/worker/internal/engine/reconciler/reconciler_test.go @@ -0,0 +1,158 @@ +package reconciler + +// The reconciliation loop's contract, against the real store: production +// stranded a day of UNKNOWN deliveries (13 rows, oldest 20+ hours) because +// the fence documented "reconciliation decides" and nothing ever decided. +// These tests pin the decision: provable server-side rejections close; every +// genuine ambiguity stays visible. + +import ( + "context" + "strings" + "testing" + "time" + + "atf/worker/internal/domain/job" + "atf/worker/internal/queue" +) + +type discardLogger struct{} + +func (discardLogger) Info(string, ...any) {} +func (discardLogger) Warn(string, ...any) {} +func (discardLogger) Error(string, ...any) {} + +// seedAmbiguous drives a fence through the exact production shape: claimed, +// outcome lost by the runtime, recorded UNKNOWN with the failure evidence. +func seedAmbiguous(t *testing.T, s queue.Store, key job.AttemptKey, evidence string, age time.Duration) { + t.Helper() + ctx := context.Background() + if _, _, ferr := s.FenceClaim(ctx, key, "w-test", float64(time.Now().Add(time.Minute).Unix())); ferr != nil { + t.Fatalf("fence claim: %v", ferr) + } + if ferr := s.FenceRecordOutcome(ctx, key, "w-test", "UNKNOWN", evidence); ferr != nil { + t.Fatalf("fence outcome: %v", ferr) + } + if age > 0 { + if _, err := queue.StoreDB(t, s).Exec( + `UPDATE delivery_attempts SET updated_at=? WHERE id=?`, + time.Now().Add(-age).Unix(), key.String()); err != nil { + t.Fatalf("age fence: %v", err) + } + } +} + +func newSweep(t *testing.T, s queue.Store) *Reconciler { + t.Helper() + return New(s, "w-reconciler", 10*time.Minute, time.Hour, discardLogger{}) +} + +func TestRunOnceClosesProvableServerRejections(t *testing.T) { + s := queue.NewTestStore(t) + ctx := context.Background() + key := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 1, TargetChatID: "t"} + seedAmbiguous(t, s, key, + "Uncertain outcome: rpc error: code = Unimplemented desc = Unexpected : Method not implemented!", + time.Hour) + queue.InsertJob(t, s, &job.DeliveryJob{ + ID: key.String(), RuleID: key.RuleID, SourceChatID: key.SourceChatID, + SourceMessageID: key.SourceMessageID, TargetChatID: key.TargetChatID, + Status: job.StatusUnknown, MaxAttempts: 5, + }) + + resolved, remaining, err := newSweep(t, s).RunOnce(ctx) + if err != nil { + t.Fatalf("run once: %v", err) + } + if resolved != 1 || remaining != 0 { + t.Fatalf("resolved=%d remaining=%d, want 1/0 — the production backlog shape must close", resolved, remaining) + } + + db := queue.StoreDB(t, s) + var jStatus, jErr string + if err := db.QueryRow(`SELECT status, error_detail FROM delivery_jobs WHERE id=?`, key.String()). + Scan(&jStatus, &jErr); err != nil { + t.Fatalf("job read: %v", err) + } + if jStatus != "FAILED" { + t.Errorf("job status = %s, want FAILED", jStatus) + } + if !strings.Contains(jErr, "Unimplemented") { + t.Errorf("reconciliation reason lost the evidence: %q", jErr) + } + + // And the closed job is recoverable through the normal path. + outcome, _, ferr := s.FenceClaim(ctx, key, "w-retry", float64(time.Now().Add(time.Minute).Unix())) + if ferr != nil { + t.Fatalf("re-claim: %v", ferr) + } + if outcome != job.FenceDispatch { + t.Errorf("post-reconcile claim = %s, want dispatch", outcome) + } +} + +func TestRunOnceLeavesGenuineAmbiguityUnknown(t *testing.T) { + s := queue.NewTestStore(t) + ctx := context.Background() + key := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 2, TargetChatID: "t"} + seedAmbiguous(t, s, key, + "Uncertain outcome: rpc error: code = DeadlineExceeded desc = context deadline exceeded", + time.Hour) + queue.InsertJob(t, s, &job.DeliveryJob{ + ID: key.String(), RuleID: key.RuleID, SourceChatID: key.SourceChatID, + SourceMessageID: key.SourceMessageID, TargetChatID: key.TargetChatID, + Status: job.StatusUnknown, MaxAttempts: 5, + }) + + resolved, remaining, err := newSweep(t, s).RunOnce(ctx) + if err != nil { + t.Fatalf("run once: %v", err) + } + if resolved != 0 || remaining != 1 { + t.Fatalf("resolved=%d remaining=%d, want 0/1 — a timeout proves nothing about the send", resolved, remaining) + } + var status string + if err := queue.StoreDB(t, s).QueryRow( + `SELECT status FROM delivery_jobs WHERE id=?`, key.String()).Scan(&status); err != nil { + t.Fatalf("job read: %v", err) + } + if status != "UNKNOWN" { + t.Errorf("job status = %s, want untouched UNKNOWN — guessing here is the duplicate-delivery path", status) + } +} + +func TestRunOnceHonorsTheGraceWindow(t *testing.T) { + s := queue.NewTestStore(t) + ctx := context.Background() + key := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 3, TargetChatID: "t"} + // Provable rejection but still inside the grace window: a live dispatch + // could still be settling. Not listed, not touched. + seedAmbiguous(t, s, key, "rpc error: code = Unimplemented desc = x", 0) + + resolved, remaining, err := newSweep(t, s).RunOnce(ctx) + if err != nil { + t.Fatalf("run once: %v", err) + } + if resolved != 0 || remaining != 0 { + t.Fatalf("resolved=%d remaining=%d, want 0/0 — fresh rows must be left for the next pass", resolved, remaining) + } +} + +func TestProvablyUnexecutedTable(t *testing.T) { + cases := []struct { + evidence string + want bool + }{ + {"rpc error: code = Unimplemented desc = Unexpected : Method not implemented!", true}, + {"rpc error: code = Unimplemented desc = unknown method SendMedia", true}, + {"rpc error: code = DeadlineExceeded", false}, + {"rpc error: code = Unavailable desc = connection error", false}, + {"Uncertain outcome: transport dropped mid-send", false}, + {"", false}, + } + for _, tc := range cases { + if got := provablyUnexecuted(tc.evidence); got != tc.want { + t.Errorf("provablyUnexecuted(%q) = %v, want %v", tc.evidence, got, tc.want) + } + } +} diff --git a/worker/internal/observability/logclient.go b/worker/internal/observability/logclient.go index d9daca2e..adbc97bd 100644 --- a/worker/internal/observability/logclient.go +++ b/worker/internal/observability/logclient.go @@ -225,6 +225,21 @@ func sanitizeValue(value any) any { func redactText(value string) string { return secretValue.ReplaceAllString(value, "$1=[REDACTED]") } +// RedactSlogValue applies the central client's own redaction rules to one +// structured-log attribute, for sinks that do not flow through the client — +// the stdout JSON stream a container collector scrapes. A secret that would +// be [REDACTED] in the Log Monitor must not appear verbatim on stdout +// either; the two sinks share one rule set by construction. +func RedactSlogValue(key string, value any) any { + if secretField.MatchString(key) || isContentField(key) { + return "[REDACTED]" + } + return sanitizeValue(value) +} + +// RedactSlogText redacts secret-shaped substrings from a message string. +func RedactSlogText(message string) string { return redactText(message) } + func toInt64(value any) int64 { switch n := value.(type) { case int: return int64(n) diff --git a/worker/internal/ops/bridge.go b/worker/internal/ops/bridge.go index bf9424d2..1b019ef6 100644 --- a/worker/internal/ops/bridge.go +++ b/worker/internal/ops/bridge.go @@ -86,6 +86,7 @@ func (b *TrackerBridge) Snapshot() map[string]any { "delivered": c.Delivered(), "retried": c.Retried(), "failed": c.Failed(), + "rejected": c.Rejected(), "unknown": c.Unknown(), } } diff --git a/worker/internal/ops/collector.go b/worker/internal/ops/collector.go index 15151a0f..2480ffb2 100644 --- a/worker/internal/ops/collector.go +++ b/worker/internal/ops/collector.go @@ -47,6 +47,7 @@ type Collector struct { delivered atomic.Int64 retried atomic.Int64 failed atomic.Int64 + rejected atomic.Int64 unknown atomic.Int64 // ACTIVE: in-flight deliveries. utilization = active / numWorkers. NOT @@ -307,9 +308,13 @@ func (c *Collector) RecordPendingDepth(n int64) { c.pendingDepth.Store(n) } // --- the delivery.Metrics surface (the engine's only metrics owner) -------- // -// Outcome bumps a delivery outcome counter (sent|retried|failed|unknown). -// This is the single place a delivery outcome is counted anywhere in the -// worker — the one owner the contract demands. +// Outcome bumps a delivery outcome counter +// (sent|retried|failed|rejected|unknown). This is the single place a +// delivery outcome is counted anywhere in the worker — the one owner the +// contract demands. "rejected" counts a DELIBERATE non-delivery (attached +// filter drop, task-level block_links, forward-every-n cadence skip): a +// policy decision about the content, not a malfunction. Counting it as +// "failed" made a healthy task look degraded. func (c *Collector) Outcome(kind string) { switch kind { case "sent": @@ -318,6 +323,8 @@ func (c *Collector) Outcome(kind string) { c.retried.Add(1) case "failed": c.failed.Add(1) + case "rejected": + c.rejected.Add(1) case "unknown": c.unknown.Add(1) } @@ -334,6 +341,10 @@ func (c *Collector) Retried() int64 { return c.retried.Load() } // Failed returns the failure counter. func (c *Collector) Failed() int64 { return c.failed.Load() } +// Rejected returns the deliberate-non-delivery counter (policy decisions, +// not failures). +func (c *Collector) Rejected() int64 { return c.rejected.Load() } + // Unknown returns the ambiguous-outcome counter. func (c *Collector) Unknown() int64 { return c.unknown.Load() } @@ -414,6 +425,7 @@ func (c *Collector) Snapshot() map[string]any { "delivered": c.Delivered(), "retried": c.Retried(), "failed": c.Failed(), + "rejected": c.Rejected(), "unknown": c.Unknown(), } } diff --git a/worker/internal/queue/postgres.go b/worker/internal/queue/postgres.go index bb67ae24..8287f959 100644 --- a/worker/internal/queue/postgres.go +++ b/worker/internal/queue/postgres.go @@ -754,6 +754,78 @@ func (s *pgStore) FenceResolveUnknown(ctx context.Context, key job.AttemptKey, w return err } +// UnknownFences lists ambiguous attempts older than olderThan, oldest first — +// the reconciler's input. Pure read; parity with the SQLite implementation. +func (s *pgStore) UnknownFences(ctx context.Context, olderThan time.Time, limit int) ([]job.UnknownFence, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT rule_id, source_chat_id, source_message_id, target_chat_id, job_kind, + last_error, updated_at + FROM delivery_attempts + WHERE status='UNKNOWN' AND updated_at <= $1 + ORDER BY updated_at + LIMIT $2`, olderThan.Unix(), limit) + if err != nil { + return nil, fmt.Errorf("unknown fences query: %w", err) + } + defer rows.Close() + var out []job.UnknownFence + for rows.Next() { + var ( + f job.UnknownFence + rule, sChat string + sMsg int64 + tChat, jKind string + ) + if err := rows.Scan(&rule, &sChat, &sMsg, &tChat, &jKind, &f.LastError, &f.UpdatedAt); err != nil { + return nil, fmt.Errorf("unknown fences scan: %w", err) + } + f.Key = job.AttemptKey{RuleID: rule, SourceChatID: sChat, SourceMessageID: sMsg, TargetChatID: tChat, JobKind: jKind} + out = append(out, f) + } + return out, rows.Err() +} + +// ReconcileUnknown closes one ambiguous delivery atomically: fence +// UNKNOWN→FAILED, then job UNKNOWN→FAILED, both status-guarded, one +// transaction — never a half-reconciliation. Returns false and changes +// nothing when either row has already moved. Parity with SQLite. +func (s *pgStore) ReconcileUnknown(ctx context.Context, key job.AttemptKey, workerID, reason string) (bool, error) { + now := time.Now().Unix() + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return false, fmt.Errorf("reconcile begin: %w", err) + } + defer tx.Rollback() + + fr, err := tx.ExecContext(ctx, ` + UPDATE delivery_attempts SET status='FAILED', worker_id=$1, last_error=$2, updated_at=$3 + WHERE id=$4 AND status='UNKNOWN'`, workerID, truncateDetail(reason), now, key.String()) + if err != nil { + return false, fmt.Errorf("reconcile fence update: %w", err) + } + fn, _ := fr.RowsAffected() + if fn == 0 { + return false, nil + } + + jr, err := tx.ExecContext(ctx, ` + UPDATE delivery_jobs SET status='FAILED', error_detail=$1, lease_until=0, updated_at=$2 + WHERE id=$3 AND status='UNKNOWN'`, truncateDetail(reason), now, key.String()) + if err != nil { + return false, fmt.Errorf("reconcile job update: %w", err) + } + jn, _ := jr.RowsAffected() + if jn == 0 { + // The job row moved while the fence stayed UNKNOWN — roll the fence + // back instead of committing an inconsistency. + return false, tx.Rollback() + } + if err := tx.Commit(); err != nil { + return false, fmt.Errorf("reconcile commit: %w", err) + } + return true, nil +} + // truncateDetail clips an error string to the column's practical length, // mirroring the Python implementation's own [:4096] bound. func truncateDetail(detail string) string { diff --git a/worker/internal/queue/reconcile_store_test.go b/worker/internal/queue/reconcile_store_test.go new file mode 100644 index 00000000..bf80d1db --- /dev/null +++ b/worker/internal/queue/reconcile_store_test.go @@ -0,0 +1,205 @@ +package queue + +// The reconciliation surface's store contract. Production stranded a day of +// UNKNOWN deliveries because the decision step never existed; these tests +// pin the listing and the atomic fence+job resolution it must perform. + +import ( + "context" + "strings" + "testing" + "time" + + "atf/worker/internal/domain/job" +) + + +// seedUnknownFence creates a fence row in the state the fence sweep leaves +// after an ambiguous outcome: UNKNOWN with the recorded failure evidence. +func seedUnknownFence(t *testing.T, s Store, key job.AttemptKey, lastError string) { + t.Helper() + ctx := context.Background() + if _, _, ferr := s.FenceClaim(ctx, key, "w-test", float64(time.Now().Add(time.Minute).Unix())); ferr != nil { + t.Fatalf("fence claim: %v", ferr) + } + if ferr := s.FenceRecordOutcome(ctx, key, "w-test", "UNKNOWN", lastError); ferr != nil { + t.Fatalf("fence outcome: %v", ferr) + } +} + +// ageFence backdates a fence row so it is older than any grace window. +func ageFence(t *testing.T, s Store, key job.AttemptKey, age time.Duration) { + t.Helper() + db := StoreDB(t, s) + if _, err := db.Exec(`UPDATE delivery_attempts SET updated_at=? WHERE id=?`, + time.Now().Add(-age).Unix(), key.String()); err != nil { + t.Fatalf("age fence: %v", err) + } +} + +func TestUnknownFencesListsOnlyAgedUnknownRows(t *testing.T) { + s := NewTestStore(t) + ctx := context.Background() + + old := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 1, TargetChatID: "t"} + fresh := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 2, TargetChatID: "t"} + failed := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 3, TargetChatID: "t"} + + seedUnknownFence(t, s, old, "rpc error: code = Unimplemented desc = boom") + seedUnknownFence(t, s, fresh, "rpc error: code = Unimplemented desc = boom") + seedUnknownFence(t, s, failed, "rpc error: code = Unimplemented desc = boom") + ageFence(t, s, old, time.Hour) + if ferr := s.FenceRecordOutcome(ctx, failed, "w-test", "FAILED", "permanent"); ferr != nil { + t.Fatalf("fail fence: %v", ferr) + } + ageFence(t, s, failed, time.Hour) + + got, err := s.UnknownFences(ctx, time.Now().Add(-30*time.Minute), 100) + if err != nil { + t.Fatalf("unknown fences: %v", err) + } + if len(got) != 1 { + t.Fatalf("UnknownFences returned %d rows, want exactly the aged UNKNOWN one: %+v", len(got), got) + } + if got[0].Key.String() != old.String() { + t.Errorf("listed %s, want %s", got[0].Key.String(), old.String()) + } + if got[0].LastError == "" { + t.Errorf("listed fence carries no recorded failure evidence — the reconciler would have nothing to classify") + } +} + +func TestUnknownFencesRespectsTheLimit(t *testing.T) { + s := NewTestStore(t) + ctx := context.Background() + for i := int64(0); i < 3; i++ { + key := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: i, TargetChatID: "t"} + seedUnknownFence(t, s, key, "ambiguous") + ageFence(t, s, key, time.Hour) + } + got, err := s.UnknownFences(ctx, time.Now().Add(-30*time.Minute), 2) + if err != nil { + t.Fatalf("unknown fences: %v", err) + } + if len(got) != 2 { + t.Fatalf("limit not honored: got %d rows, want 2", len(got)) + } + // Oldest first, always. + if got[0].Key.SourceMessageID > got[1].Key.SourceMessageID { + t.Errorf("ordering broken: %+v", got) + } +} + +func TestReconcileUnknownClosesFenceAndJobAtomically(t *testing.T) { + s := NewTestStore(t) + ctx := context.Background() + key := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 5, TargetChatID: "t"} + seedUnknownFence(t, s, key, "rpc error: code = Unimplemented desc = method not implemented") + ageFence(t, s, key, time.Hour) + InsertJob(t, s, &job.DeliveryJob{ + ID: key.String(), RuleID: key.RuleID, SourceChatID: key.SourceChatID, + SourceMessageID: key.SourceMessageID, TargetChatID: key.TargetChatID, + Status: job.StatusUnknown, MaxAttempts: 5, + }) + + ok, err := s.ReconcileUnknown(ctx, key, "w-reconciler", "reconciled: provably unexecuted") + if err != nil { + t.Fatalf("reconcile: %v", err) + } + if !ok { + t.Fatal("reconcile reported no change — the row was fully UNKNOWN") + } + + db := StoreDB(t, s) + var fStatus, fErr string + if err := db.QueryRow(`SELECT status, last_error FROM delivery_attempts WHERE id=?`, key.String()). + Scan(&fStatus, &fErr); err != nil { + t.Fatalf("fence read: %v", err) + } + if fStatus != "FAILED" { + t.Errorf("fence status = %s, want FAILED", fStatus) + } + if fErr == "" || !strings.Contains(fErr, "provably unexecuted") { + t.Errorf("fence last_error lost the reconciliation reason: %q", fErr) + } + var jStatus, jErr string + if err := db.QueryRow(`SELECT status, error_detail FROM delivery_jobs WHERE id=?`, key.String()). + Scan(&jStatus, &jErr); err != nil { + t.Fatalf("job read: %v", err) + } + if jStatus != "FAILED" { + t.Errorf("job status = %s, want FAILED — an unresolved job state beside a resolved fence is the inconsistency this exists to prevent", jStatus) + } + if jErr == "" || !strings.Contains(jErr, "provably unexecuted") { + t.Errorf("job error_detail lost the reconciliation reason: %q", jErr) + } + + // The resolved job is re-dispatchable through the normal path: the fence + // re-claims a FAILED row (FenceDispatch), never blocks a retry. + outcome, _, ferr := s.FenceClaim(ctx, key, "w-retry", float64(time.Now().Add(time.Minute).Unix())) + if ferr != nil { + t.Fatalf("re-claim after reconcile: %v", ferr) + } + if outcome != job.FenceDispatch { + t.Errorf("post-reconcile fence claim = %s, want dispatch (a reconciled delivery must be recoverable)", outcome) + } +} + +func TestReconcileUnknownLeavesMovedFenceAlone(t *testing.T) { + s := NewTestStore(t) + ctx := context.Background() + key := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 6, TargetChatID: "t"} + seedUnknownFence(t, s, key, "ambiguous") + ageFence(t, s, key, time.Hour) + if ferr := s.FenceRecordSent(ctx, key, "w-other", 777); ferr != nil { + t.Fatalf("record sent: %v", ferr) + } + + ok, err := s.ReconcileUnknown(ctx, key, "w-reconciler", "reconciled") + if err != nil { + t.Fatalf("reconcile: %v", err) + } + if ok { + t.Fatal("reconcile overrode a fence that already recorded a confirmed send") + } + var status string + if err := StoreDB(t, s).QueryRow(`SELECT status FROM delivery_attempts WHERE id=?`, key.String()). + Scan(&status); err != nil { + t.Fatalf("fence read: %v", err) + } + if status != "SENT" { + t.Errorf("fence status = %s, want untouched SENT", status) + } +} + +func TestReconcileUnknownRollsBackWhenJobAlreadyMoved(t *testing.T) { + s := NewTestStore(t) + ctx := context.Background() + key := job.AttemptKey{RuleID: "r1", SourceChatID: "c", SourceMessageID: 7, TargetChatID: "t"} + seedUnknownFence(t, s, key, "ambiguous") + ageFence(t, s, key, time.Hour) + InsertJob(t, s, &job.DeliveryJob{ + ID: key.String(), RuleID: key.RuleID, SourceChatID: key.SourceChatID, + SourceMessageID: key.SourceMessageID, TargetChatID: key.TargetChatID, + Status: job.StatusSent, MaxAttempts: 5, + }) + + // Inconsistent pre-state (UNKNOWN fence beside a SENT job — possible only + // through an out-of-band write). The reconcile must change NOTHING: a + // half-reconciliation is the exact defect the transaction guards against. + ok, err := s.ReconcileUnknown(ctx, key, "w-reconciler", "reconciled") + if err != nil { + t.Fatalf("reconcile: %v", err) + } + if ok { + t.Fatal("reconcile reported success against an inconsistent pair") + } + var fStatus string + if err := StoreDB(t, s).QueryRow(`SELECT status FROM delivery_attempts WHERE id=?`, key.String()). + Scan(&fStatus); err != nil { + t.Fatalf("fence read: %v", err) + } + if fStatus != "UNKNOWN" { + t.Errorf("fence status = %s, want UNKNOWN (rolled back) — a FAILED fence beside a SENT job would let a duplicate dispatch through", fStatus) + } +} diff --git a/worker/internal/queue/sqlite.go b/worker/internal/queue/sqlite.go index 05d45b53..1e528123 100644 --- a/worker/internal/queue/sqlite.go +++ b/worker/internal/queue/sqlite.go @@ -93,6 +93,16 @@ type Store interface { // proved no earlier copy is in the target. The reason is recorded. FenceResolveUnknown(ctx context.Context, key job.AttemptKey, workerID, reason string) error + // UnknownFences lists ambiguous attempts older than olderThan — the + // reconciler's input. Oldest first, bounded by limit. + UnknownFences(ctx context.Context, olderThan time.Time, limit int) ([]job.UnknownFence, error) + + // ReconcileUnknown closes one ambiguous delivery atomically: fence + // UNKNOWN→FAILED, then job UNKNOWN→FAILED, both status-guarded. It + // returns false and changes nothing when either row has already moved — + // the guard that keeps a concurrent transition from being clobbered. + ReconcileUnknown(ctx context.Context, key job.AttemptKey, workerID, reason string) (bool, error) + // NoteMapping records the delivery outcome on the job's source->target // message_map row (primary forward only — the Python worker's rule, and // the reason keyword-reply jobs never write there). @@ -813,6 +823,82 @@ func (s *sqliteStore) FenceResolveUnknown(ctx context.Context, key job.AttemptKe return err } +// UnknownFences lists ambiguous attempts older than olderThan, oldest first. +// The reconciler's sole input; a pure read. +func (s *sqliteStore) UnknownFences(ctx context.Context, olderThan time.Time, limit int) ([]job.UnknownFence, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT rule_id, source_chat_id, source_message_id, target_chat_id, job_kind, + last_error, updated_at + FROM delivery_attempts + WHERE status='UNKNOWN' AND updated_at <= ? + ORDER BY updated_at + LIMIT ?`, olderThan.Unix(), limit) + if err != nil { + return nil, fmt.Errorf("unknown fences query: %w", err) + } + defer rows.Close() + var out []job.UnknownFence + for rows.Next() { + var ( + f job.UnknownFence + rule, sChat string + sMsg int64 + tChat, jKind string + ) + if err := rows.Scan(&rule, &sChat, &sMsg, &tChat, &jKind, &f.LastError, &f.UpdatedAt); err != nil { + return nil, fmt.Errorf("unknown fences scan: %w", err) + } + f.Key = job.AttemptKey{RuleID: rule, SourceChatID: sChat, SourceMessageID: sMsg, TargetChatID: tChat, JobKind: jKind} + out = append(out, f) + } + return out, rows.Err() +} + +// ReconcileUnknown closes one ambiguous delivery atomically. The fence and +// the job must agree — an UNKNOWN fence with a non-UNKNOWN job (or the +// reverse) is precisely the inconsistency reconciliation exists to remove — +// so both transitions live in one transaction and both are status-guarded: +// a row that already moved is never clobbered, and a half-reconciliation can +// never commit. +func (s *sqliteStore) ReconcileUnknown(ctx context.Context, key job.AttemptKey, workerID, reason string) (bool, error) { + now := time.Now().Unix() + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return false, fmt.Errorf("reconcile begin: %w", err) + } + defer tx.Rollback() + + fr, err := tx.ExecContext(ctx, ` + UPDATE delivery_attempts SET status='FAILED', worker_id=?, last_error=?, updated_at=? + WHERE id=? AND status='UNKNOWN'`, workerID, reason, now, key.String()) + if err != nil { + return false, fmt.Errorf("reconcile fence update: %w", err) + } + fn, _ := fr.RowsAffected() + if fn == 0 { + // Another actor resolved (or a dispatcher recorded an outcome on) + // this fence first. Nothing to reconcile. + return false, nil + } + + jr, err := tx.ExecContext(ctx, ` + UPDATE delivery_jobs SET status='FAILED', error_detail=?, lease_until=0, updated_at=? + WHERE id=? AND status='UNKNOWN'`, reason, now, key.String()) + if err != nil { + return false, fmt.Errorf("reconcile job update: %w", err) + } + jn, _ := jr.RowsAffected() + if jn == 0 { + // The job row moved while the fence stayed UNKNOWN — the store would + // be left inconsistent if this committed. Roll the fence back. + return false, tx.Rollback() + } + if err := tx.Commit(); err != nil { + return false, fmt.Errorf("reconcile commit: %w", err) + } + return true, nil +} + // --- ops: the read-only queue-status and per-task aggregations ---------- // // // The ops surface (internal/ops) reads the queue through these methods. diff --git a/worker/internal/telegraph/fakegateway/fake.go b/worker/internal/telegraph/fakegateway/fake.go index dfef19b2..b3ba8b71 100644 --- a/worker/internal/telegraph/fakegateway/fake.go +++ b/worker/internal/telegraph/fakegateway/fake.go @@ -37,11 +37,18 @@ type Fake struct { Calls []Call // Scripted behaviours. nil means "no scripted failure for this op". - ForwardFailsWith *telegraph.DeliveryFailure - CopyFailsWith *telegraph.DeliveryFailure - SendFailsWith *telegraph.DeliveryFailure - CopyMediaGroupFailsWith *telegraph.DeliveryFailure - DownloadFailsWith *telegraph.DeliveryFailure + ForwardFailsWith *telegraph.DeliveryFailure + CopyFailsWith *telegraph.DeliveryFailure + SendFailsWith *telegraph.DeliveryFailure + CopyMediaGroupFailsWith *telegraph.DeliveryFailure + DownloadFailsWith *telegraph.DeliveryFailure + GetFileExtensionFailsWith *telegraph.DeliveryFailure + + // ExtensionOverride, when set, is what GetFileExtension answers instead + // of the default ".bin". Tests use it to pin the staging-name contract + // for both "the runtime knows the extension" and "the runtime cannot + // tell" (the empty string) — including hostile-looking values. + ExtensionOverride *string // failOnce[targetChat] = true makes the next send to that target fail // once with a transient error, then clear. E2E retry-path scenarios use @@ -55,6 +62,13 @@ type Fake struct { // touching a real restricted channel. RestrictedSource bool + // CopyMediaRejected makes CopyMessage fail with a target-side media + // rejection (PHOTO_EXT_INVALID): the copy survives permission-wise but + // the destination refuses the media payload itself. The re-upload + // fallback is the only recovery — the production signal this knob + // exists to reproduce. + CopyMediaRejected bool + // CopyFailOnNth makes the n-th CopyMessage call (1-based) fail with a // transient error — the "split album died on item 3" shape a resume test // needs. 0 disables it. @@ -131,9 +145,17 @@ func (f *Fake) GetAlbumMessageIds(ctx context.Context, sessionID string, chatID, return []int64{messageID}, nil } -// GetFileExtension returns a plausible extension for the media kind. +// GetFileExtension returns the scripted extension (".bin" when nothing is +// scripted — a plausible answer for an unknown document kind), or the +// scripted failure. func (f *Fake) GetFileExtension(ctx context.Context, sessionID string, chatID, messageID int64) (string, *telegraph.DeliveryFailure) { f.record(Call{Op: "GetFileExtension", SessionID: sessionID, ChatID: chatID, MessageID: messageID}) + if f.GetFileExtensionFailsWith != nil { + return "", f.GetFileExtensionFailsWith + } + if f.ExtensionOverride != nil { + return *f.ExtensionOverride, nil + } return ".bin", nil } @@ -198,6 +220,13 @@ func (f *Fake) CopyMessage(ctx context.Context, sessionID string, targetChatID, Message: "fake: source forbids copying protected content", } } + if f.CopyMediaRejected { + return 0, &telegraph.DeliveryFailure{ + Code: "TELEGRAM_PERMANENT_ERROR", + ErrorType: "PHOTO_EXT_INVALID", + Message: "fake: target rejected the copied photo (extension invalid)", + } + } if f.CopyFailsWith != nil { return 0, f.CopyFailsWith } diff --git a/worker/internal/telegraph/gateway.go b/worker/internal/telegraph/gateway.go index c173280a..a955673e 100644 --- a/worker/internal/telegraph/gateway.go +++ b/worker/internal/telegraph/gateway.go @@ -75,6 +75,34 @@ func (f *DeliveryFailure) IsForwardingRestricted() bool { return f.Code == "FORWARDING_RESTRICTED" } +// IsMediaRejectedByTarget reports whether the TARGET chat refused the media +// payload itself — a per-message compatibility rejection, not a permission +// problem. Telegram signals it as PHOTO_EXT_INVALID (and the neighbouring +// MEDIA_INVALID / VIDEO_NOT_MODIFIED / PHOTO_INVALID_DIMENSIONS family): +// the copy preserves a file reference the destination cannot accept, and +// retrying the same copy can never fix it. +// +// The recovery is the SAME re-upload path forwarding-restriction uses: +// downloading the file and uploading it fresh normalizes its extension and +// dimensions, so a job that would otherwise die as a permanent failure +// delivers. Both signals are "the copy cannot survive as-is". +func (f *DeliveryFailure) IsMediaRejectedByTarget() bool { + switch f.ErrorType { + case "PHOTO_EXT_INVALID", "MEDIA_INVALID", "PHOTO_INVALID_DIMENSIONS", + "VIDEO_NOT_MODIFIED", "IMAGE_PROCESS_FAILED", "PHOTO_CONTENT_URL_EMPTY": + return true + } + return false +} + +// NeedsReupload is the decision the copy sites consult: either the source +// forbids copying, or the target rejected the copied media. One call site +// instead of two keeps the two reasons for re-uploading from drifting +// apart, and the task's fallback_enabled flag still gates both. +func (f *DeliveryFailure) NeedsReupload() bool { + return f.IsForwardingRestricted() || f.IsMediaRejectedByTarget() +} + // MediaHandle is a staged media file inside the shared scratch volume. type MediaHandle struct { Path string diff --git a/worker/internal/telegraph/gateway_test.go b/worker/internal/telegraph/gateway_test.go new file mode 100644 index 00000000..7434b49f --- /dev/null +++ b/worker/internal/telegraph/gateway_test.go @@ -0,0 +1,43 @@ +package telegraph + +import "testing" + +// TestNeedsReupload pins the two reasons a copy must fall through to the +// re-upload path, because production showed them treated differently: +// +// - a source that forbids copying (FORWARDING_RESTRICTED), and +// - a TARGET that rejects the copied media itself (the PHOTO_EXT_INVALID +// family). The copy is not forbidden — it is incompatible — and retrying +// the same copy can never fix it, exactly like a restriction. +// +// The second class shipped as a permanent failure for 12 real jobs because +// the fallback only ever asked about restrictions. NeedsReupload closes +// that gap by making one call cover both. +func TestNeedsReupload(t *testing.T) { + cases := []struct { + name string + code string + errorType string + want bool + restricted bool + }{ + {"source restriction", "FORWARDING_RESTRICTED", "RESTRICTED", true, true}, + {"target photo rejection", "TELEGRAM_PERMANENT_ERROR", "PHOTO_EXT_INVALID", true, false}, + {"target media invalid", "TELEGRAM_PERMANENT_ERROR", "MEDIA_INVALID", true, false}, + {"target video not modified", "TELEGRAM_PERMANENT_ERROR", "VIDEO_NOT_MODIFIED", true, false}, + {"flood wait is not a reupload signal", "FLOOD_WAIT", "FLOOD_WAIT", false, false}, + {"network timeout is a retry, not a reupload", "NETWORK_TIMEOUT", "TIMEOUT", false, false}, + {"unknown outcome is ambiguous, not a reupload", "UNKNOWN_DELIVERY_OUTCOME", "UNKNOWN", false, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := &DeliveryFailure{Code: tc.code, ErrorType: tc.errorType} + if got := f.NeedsReupload(); got != tc.want { + t.Errorf("NeedsReupload() = %v, want %v", got, tc.want) + } + if f.IsForwardingRestricted() != tc.restricted { + t.Errorf("IsForwardingRestricted() drift: got %v, want %v", f.IsForwardingRestricted(), tc.restricted) + } + }) + } +} diff --git a/worker/test/live/run_matrix.py b/worker/test/live/run_matrix.py index 8f8b5333..b36af0ee 100644 --- a/worker/test/live/run_matrix.py +++ b/worker/test/live/run_matrix.py @@ -79,10 +79,26 @@ def clone_rule(c: sqlite3.Connection, base: sqlite3.Row, marker: str, return rid -def drop_rule(c: sqlite3.Connection, rid: str, keep_jobs: bool = False) -> None: - if not keep_jobs: - c.execute("DELETE FROM delivery_jobs WHERE rule_id=?", (rid,)) - c.execute("DELETE FROM rule_counters WHERE rule_id=?", (rid,)) +def drop_rule(c: sqlite3.Connection, rid: str) -> None: + """Delete a matrix replica's ENTIRE footprint from the live store. + + This tool runs against the production database by design (it lives in + the core container next to the real session), so its cleanup is the + safety property that keeps it acceptable: every row it created — job + rows, fence rows, mappings, counters, the rule itself — is removed, and + nothing but its own mtx-* replicas is ever a candidate. Production + previously accumulated test rows (e2e-go-*, mtx-*) because the cleanup + stopped at rules + counters and deliberately kept delivery_jobs. + """ + if not str(rid).startswith("mtx-"): + raise RuntimeError( + f"refusing to drop {rid!r}: the matrix only ever deletes its own " + "mtx-* replicas — a non-replica id here means a caller bug, not " + "a cleanup target" + ) + for table in ("delivery_jobs", "delivery_attempts", "message_map", + "dead_letter_queue", "rule_counters"): + c.execute(f"DELETE FROM {table} WHERE rule_id=?", (rid,)) c.execute("DELETE FROM forward_rules WHERE id=?", (rid,)) c.commit() @@ -191,7 +207,12 @@ def cell_replica(client, c: sqlite3.Connection, base: sqlite3.Row, cell: str, "job_status": status, } finally: - drop_rule(c, rid, keep_jobs=True) + # Full replica cleanup — the outcome is already captured in the + # result dict, so the rows carry no information worth keeping. The + # tool's own docstring promises this ("Every replica is deleted at + # the end"); keeping job rows behind is what seeded production with + # orphaned test traffic. + drop_rule(c, rid) c.execute("UPDATE forward_rules SET is_paused=? WHERE id=?", (0, BASELINE)) c.commit() diff --git a/worker/test/ownership/stage_order_test.go b/worker/test/ownership/stage_order_test.go index c75abc8f..17273ea9 100644 --- a/worker/test/ownership/stage_order_test.go +++ b/worker/test/ownership/stage_order_test.go @@ -196,23 +196,29 @@ func TestStageOrderIsObservedNotAsserted(t *testing.T) { e := evt("ATF-OWN-ORD-002", "ORD", "filter rejection precedes every transform stage", - "filter:failed observed, no cleaner/replacement/header stage ever recorded") + "filter stage observed (failed or policy-skipped), no cleaner/replacement/header stage ever recorded") start := now() j := h.Enqueue("task-ord2", "-1001", 9802, "-9002", "BANNED TOK body") err := h.Run(ctx, j) stages := obs.observed() dropped := h.AssertDropped(&e, j) - // The filter stage failed, and no transform stage was ever walked. - filterFailed := containsStage(stages, "filter:failed") || containsStage(stages, "filter") + // The filter stage was OBSERVED, and no transform stage was ever + // walked. The observation may be :failed (stage malfunction) or + // :skipped (the filter working as configured — a deliberate policy + // drop counts as skipped-by-policy, never as a stage failure); the + // ordering contract is only that filter came before any transform. + filterSeen := containsStage(stages, "filter") || + containsStage(stages, "filter:failed") || + containsStage(stages, "filter:skipped") noTransform := !containsAny(stages, "cleaner", "replacement", "header_footer", "ai") - if dropped && filterFailed && noTransform { + if dropped && filterSeen && noTransform { e.Result = ResultPass e.Observed = fmt.Sprintf("stages=%v", stages) } else { e.Result = ResultFail - e.FailureReason = fmt.Sprintf("stages=%v dropped=%v filterFailed=%v noTransform=%v", - stages, dropped, filterFailed, noTransform) + e.FailureReason = fmt.Sprintf("stages=%v dropped=%v filterSeen=%v noTransform=%v", + stages, dropped, filterSeen, noTransform) } h.Finish(rec, &e, start, err) if e.Result != ResultPass {