You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tooling.apache.org by sb...@apache.org on 2026/07/31 16:02:10 UTC
(tooling-trusted-releases) branch sbp updated (a0a5b64d -> 2e34dc01)
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a change to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
from a0a5b64d Add a database table to record PubSub event processing failures
add e2464008 #1479 - record vote thread ID and present in the catalog
new 2e34dc01 Save and use a PubSub events cursor for more accurate replay
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
atr/admin/__init__.py | 1 +
atr/ldap.py | 5 +-
atr/models/api.py | 2 +
atr/models/sql.py | 3 +
atr/pubsub.py | 122 +++++++++-
atr/shared/catalog.py | 1 +
atr/storage/writers/vote.py | 28 ++-
atr/svn/commits.py | 6 +-
atr/templates/_catalog_version.html | 4 +
atr/templates/catalog_site/release.html | 4 +
....15_91de479d.py => 0116_2026.07.31_f6b39be8.py} | 20 +-
tests/unit/test_ldap.py | 10 +-
tests/unit/test_pubsub.py | 267 ++++++++++++++++++++-
tests/unit/test_vote_resolution.py | 1 +
14 files changed, 429 insertions(+), 45 deletions(-)
copy migrations/versions/{0109_2026.07.15_91de479d.py => 0116_2026.07.31_f6b39be8.py} (65%)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org
(tooling-trusted-releases) 01/01: Save and use a PubSub events cursor for more accurate replay
Posted by sb...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sbp pushed a commit to branch sbp
in repository https://gitbox.apache.org/repos/asf/tooling-trusted-releases.git
commit 2e34dc014463ddda8c1472c7f8f003d680e10010
Author: Sean B. Palmer <se...@miscoranda.com>
AuthorDate: Fri Jul 31 17:01:35 2026 +0100
Save and use a PubSub events cursor for more accurate replay
---
atr/admin/__init__.py | 1 +
atr/ldap.py | 5 +-
atr/pubsub.py | 122 +++++++++++++++++++--
atr/svn/commits.py | 6 +-
tests/unit/test_ldap.py | 10 +-
tests/unit/test_pubsub.py | 267 +++++++++++++++++++++++++++++++++++++++++++++-
6 files changed, 385 insertions(+), 26 deletions(-)
diff --git a/atr/admin/__init__.py b/atr/admin/__init__.py
index f7b8975f..28333f7d 100644
--- a/atr/admin/__init__.py
+++ b/atr/admin/__init__.py
@@ -2255,6 +2255,7 @@ async def _data_browse(_session: web.Committer, model: str = "Committee") -> str
"CheckResultIgnore": data.check_result_ignore,
"Committee": data.committee,
"Project": data.project,
+ "PubSubFailure": data.pub_sub_failure,
"SigningCertificate": lambda: data.signing_certificate(deleted=db.NOT_SET),
"Release": data.release,
"ReleasePolicy": data.release_policy,
diff --git a/atr/ldap.py b/atr/ldap.py
index 1edae78b..23e4a246 100644
--- a/atr/ldap.py
+++ b/atr/ldap.py
@@ -254,15 +254,14 @@ async def github_to_apache(github_numeric_uid: int) -> str:
return ldap_params.results_list[0].uid[0]
-async def handle_update(payload: dict) -> None:
+async def handle_update(payload: dict) -> str | None:
import atr.cache as cache
import atr.log as log
try:
parsed = PubSubPayload.model_validate(payload)
except schema.pydantic.ValidationError:
- log.warning(f"Failed to parse LDAP pubsub payload with DN: {payload.get('dn', '<missing>')}")
- return
+ return f"Failed to parse LDAP pubsub payload with DN: {payload.get('dn', '<missing>')}"
uid = _extract_uid_from_pubsub(parsed)
if uid is None:
diff --git a/atr/pubsub.py b/atr/pubsub.py
index 67a7003b..95aabe0f 100644
--- a/atr/pubsub.py
+++ b/atr/pubsub.py
@@ -17,15 +17,25 @@
import asyncio
import json
+import pathlib
+import re
import urllib.parse
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Callable
from typing import Any
import aiohttp
+import atr.cache as cache
+import atr.db as db
import atr.ldap as ldap
import atr.log as log
+import atr.models.sql as sql
+import atr.paths as paths
+import atr.storage as storage
import atr.svn.commits as commits
+import atr.util as util
+
+_CURSOR_PATTERN = re.compile(r"[0-9A-Za-z-]{36}")
# The server sends keepalives every 5 seconds, so we should see
# activity well within this timeout period.
@@ -66,6 +76,7 @@ async def listen(
password: str | None = None,
sock_read: float | None = None,
buffersize: int | None = None,
+ cursor: Callable[[], str | None] | None = None,
) -> AsyncGenerator[dict[str, Any]]:
if username:
if password is None:
@@ -89,7 +100,7 @@ async def listen(
while True:
log.debug("Opening new connection...")
try:
- async for payload in _process_connection(session, pubsub_url):
+ async for payload in _process_connection(session, pubsub_url, cursor() if cursor else None):
# We got a payload, so reset the DELAY.
delay = 0.0
@@ -105,20 +116,75 @@ async def listen(
delay = min(30.0, (delay + 1.0) * 2)
-async def _handle_payload(payload: dict[str, Any]) -> None:
+async def _cursor_load(stream: str) -> str | None:
+ try:
+ text = await asyncio.to_thread(_cursor_path().read_text)
+ except FileNotFoundError:
+ return None
+ data = json.loads(text)
+ if data.get("stream") != stream:
+ return None
+ cursor = data.get("cursor")
+ if isinstance(cursor, str) and _CURSOR_PATTERN.fullmatch(cursor):
+ return cursor
+ raise ValueError(f"Invalid cursor in {_cursor_path()}")
+
+
+def _cursor_path() -> pathlib.Path:
+ return pathlib.Path(paths.get_runtime_dir()) / "pubsub-cursor.json"
+
+
+async def _cursor_save(stream: str, cursor: str) -> None:
+ await util.atomic_write_file(_cursor_path(), json.dumps({"stream": stream, "cursor": cursor}))
+
+
+async def _failure_record(payload: dict[str, Any], detail: str) -> None:
+ cursor = payload.get("pubsub_cursor")
+ failure = sql.PubSubFailure(
+ cursor=cursor if isinstance(cursor, str) else None,
+ detail=detail,
+ payload=payload,
+ )
+ try:
+ async with db.session() as data:
+ data.add(failure)
+ await data.commit()
+ except Exception as exc:
+ log.exception(f"PubSub failure record failed: {exc}")
+
+
+async def _halt_notify(detail: str) -> None:
+ message = (
+ f"PubSub is not running: the resume file {_cursor_path()} is damaged ({detail}). "
+ "Inspect it, then remove it and restart ATR to resume from the live stream."
+ )
+ try:
+ for asf_uid in sorted(cache.admins_get()):
+ async with storage.write_as_user_service(asf_uid) as waus:
+ await waus.notifications_create(message)
+ except Exception:
+ log.exception("Failed to record PubSub halt notifications")
+
+
+async def _handle_payload(payload: dict[str, Any]) -> str | None:
if is_commit_payload(payload):
- await commits.handle(payload)
- elif is_ldap_payload(payload):
- await ldap.handle_update(payload)
+ return await commits.handle(payload)
+ if is_ldap_payload(payload):
+ return await ldap.handle_update(payload)
+ return "No handler for payload topics"
-async def _process_connection(session, pubsub_url):
+async def _process_connection(session, pubsub_url, since):
# Connect to pubsub and listen for payloads.
- async with session.get(pubsub_url) as conn:
+ headers = {"X-Fetch-Since-Cursor": since} if since else {}
+ async with session.get(pubsub_url, headers=headers) as conn:
# print('LIMITS:', conn.content.get_read_buffer_limits())
conn.raise_for_status()
if conn.content_type != _STREAM_CONTENT_TYPE:
log.warning(f"Unexpected pubsub content type: {conn.content_type}")
+ if since:
+ # TODO: Modify the PubSub server to confirm replay
+ log.info(f"Requested replay since cursor {since}; the server does not confirm replay")
while True:
# The pubsub server defines stream payloads as:
@@ -155,6 +221,8 @@ class PubSubListener:
self.username = username
self.password = password
self.topics = topics
+ self.cursor: str | None = None
+ self.staged: str | None = None
async def start(self) -> None:
"""Run forever, processing PubSub payloads as they arrive."""
@@ -167,15 +235,36 @@ class PubSubListener:
log.info(f"PubSubListener starting with URL: {full_url}")
try:
- async for payload in listen(full_url, username=self.username, password=self.password):
+ self.cursor = await _cursor_load(full_url)
+ except Exception as exc:
+ log.exception(f"PubSub resume file is damaged, not starting: {exc}")
+ await _halt_notify(f"{type(exc).__name__}: {exc}")
+ return
+
+ try:
+ async for payload in listen(
+ full_url, username=self.username, password=self.password, cursor=lambda: self.cursor
+ ):
if "stillalive" in payload:
+ await self._checkpoint(full_url)
continue
# Isolate per-payload failures: one bad commit or LDAP event must not tear the
# whole listener down (and with it the other topic)
try:
- await _handle_payload(payload)
+ detail = await _handle_payload(payload)
except Exception as exc:
- log.exception(f"PubSub handler failed for one payload, skipping: {exc}")
+ log.exception(
+ f"PubSub handler failed for one payload, skipping: {exc}",
+ cursor=payload.get("pubsub_cursor"),
+ )
+ await _failure_record(payload, f"{type(exc).__name__}: {exc}")
+ continue
+ if detail is not None:
+ log.warning(f"PubSub handler reported a problem: {detail}", cursor=payload.get("pubsub_cursor"))
+ await _failure_record(payload, detail)
+ staged = payload.get("pubsub_cursor")
+ if isinstance(staged, str) and _CURSOR_PATTERN.fullmatch(staged):
+ self.staged = staged
except asyncio.CancelledError:
log.info("PubSubListener cancelled, shutting down gracefully")
raise
@@ -184,6 +273,17 @@ class PubSubListener:
finally:
log.info("PubSubListener.start() finished")
+ async def _checkpoint(self, stream: str) -> None:
+ cursor = self.staged
+ if (cursor is None) or (cursor == self.cursor):
+ return
+ try:
+ await _cursor_save(stream, cursor)
+ except Exception as exc:
+ log.exception(f"PubSub cursor save failed, will retry: {exc}")
+ return
+ self.cursor = cursor
+
def _configured(self) -> bool:
if not self.url:
log.error("PubSub URL is not configured")
diff --git a/atr/svn/commits.py b/atr/svn/commits.py
index a8618afb..67fcd494 100644
--- a/atr/svn/commits.py
+++ b/atr/svn/commits.py
@@ -27,13 +27,13 @@ import atr.svn.catalog as catalog
_DIST_PUBSUB_PREFIX: Final[str] = "/svn/dist/"
-async def handle(payload: dict) -> None:
+async def handle(payload: dict) -> str | None:
pubsub_path = str(payload.get("pubsub_path", ""))
if not pubsub_path.startswith(_DIST_PUBSUB_PREFIX):
- return
+ return None
commit = payload.get("commit", {})
if not isinstance(commit, dict):
- return
+ return "Commit in SVN pubsub payload is not an object"
# Log every dist commit we accept, so a deploy can confirm the prefix above
# actually matches dist commits even when a commit catalogues nothing.
changed = commit.get("changed", {})
diff --git a/tests/unit/test_ldap.py b/tests/unit/test_ldap.py
index 73fa55fd..cc35cd97 100644
--- a/tests/unit/test_ldap.py
+++ b/tests/unit/test_ldap.py
@@ -183,13 +183,11 @@ async def test_github_to_apache_raises_when_ldap_unavailable(monkeypatch: "Monke
@pytest.mark.asyncio
-async def test_handle_update_handles_invalid_payload(monkeypatch: "MonkeyPatch"):
- warnings: list[str] = []
- monkeypatch.setattr("atr.log.warning", lambda msg: warnings.append(msg))
+async def test_handle_update_handles_invalid_payload():
+ detail = await ldap.handle_update({"not": "valid"})
- await ldap.handle_update({"not": "valid"})
-
- assert any("Failed to parse" in msg for msg in warnings)
+ assert detail is not None
+ assert "Failed to parse" in detail
@pytest.mark.asyncio
diff --git a/tests/unit/test_pubsub.py b/tests/unit/test_pubsub.py
index dadc73d7..b9680ff8 100644
--- a/tests/unit/test_pubsub.py
+++ b/tests/unit/test_pubsub.py
@@ -17,6 +17,7 @@
import asyncio
import json
+import pathlib
from typing import Any
import aiohttp.web as web
@@ -31,12 +32,14 @@ _STILLALIVE = {"stillalive": 1234.5}
class Server:
def __init__(self) -> None:
self.connections = 0
+ self.cursors: list[str | None] = []
self.reject = 0
self.scripts: list[list[bytes]] = []
self.url = ""
async def handle(self, request: web.Request) -> web.StreamResponse:
self.connections += 1
+ self.cursors.append(request.headers.get("X-Fetch-Since-Cursor"))
if self.reject > 0:
self.reject -= 1
return web.Response(status=503)
@@ -64,6 +67,20 @@ async def server():
await runner.cleanup()
+async def test_cursor_file_round_trip(monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path) -> None:
+ monkeypatch.setattr(pubsub, "_cursor_path", lambda: tmp_path / "pubsub-cursor.json")
+ assert await pubsub._cursor_load("stream") is None
+ await pubsub._cursor_save("stream", "a" * 36)
+ assert await pubsub._cursor_load("stream") == "a" * 36
+ assert await pubsub._cursor_load("other") is None
+ await pubsub._cursor_save("stream", "bad")
+ with pytest.raises(ValueError):
+ await pubsub._cursor_load("stream")
+ (tmp_path / "pubsub-cursor.json").write_text("garbage")
+ with pytest.raises(json.JSONDecodeError):
+ await pubsub._cursor_load("stream")
+
+
async def test_listen_reconnects_after_eof(server: Server) -> None:
server.scripts = [[_line(_EVENT)], [_line(_EVENT)]]
payloads = await _collect(server.url, 2)
@@ -71,6 +88,13 @@ async def test_listen_reconnects_after_eof(server: Server) -> None:
assert server.connections == 2
+async def test_listen_requests_replay_from_cursor(server: Server) -> None:
+ server.scripts = [[_line(_EVENT)]]
+ payloads = await _collect(server.url, 1, cursor="c" * 36)
+ assert payloads[0]["commit"] == {"id": 1}
+ assert server.cursors == ["c" * 36]
+
+
async def test_listen_survives_bad_status(server: Server) -> None:
server.reject = 1
server.scripts = [[_line(_EVENT)]]
@@ -101,10 +125,158 @@ async def test_listen_yields_events_and_keepalives(server: Server) -> None:
assert server.connections == 1
+async def test_start_checkpoints_at_keepalives(monkeypatch: pytest.MonkeyPatch) -> None:
+ saves: list[tuple[str, str]] = []
+
+ async def fake_listen(url: str, username: str | None = None, password: str | None = None, cursor: Any = None):
+ assert cursor() == "resume-cursor"
+ yield {"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "a" * 36}
+ assert cursor() == "resume-cursor"
+ yield {"stillalive": 1.0}
+ assert cursor() == "a" * 36
+ yield {"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "b" * 36}
+
+ async def fake_handle(payload: dict[str, Any]) -> None:
+ return None
+
+ async def fake_load(stream: str) -> str | None:
+ return "resume-cursor"
+
+ async def fake_save(stream: str, cursor: str) -> None:
+ saves.append((stream, cursor))
+
+ monkeypatch.setattr(pubsub, "listen", fake_listen)
+ monkeypatch.setattr(pubsub, "_cursor_load", fake_load)
+ monkeypatch.setattr(pubsub, "_cursor_save", fake_save)
+ monkeypatch.setattr(pubsub.commits, "handle", fake_handle)
+ listener = pubsub.PubSubListener("https://pubsub.invalid/", "user", "password")
+ await listener.start()
+ assert saves == [("https://pubsub.invalid/commit/svn,ldap", "a" * 36)]
+ assert listener.cursor == "a" * 36
+ assert listener.staged == "b" * 36
+
+
+async def test_start_does_not_checkpoint_failed_events(monkeypatch: pytest.MonkeyPatch) -> None:
+ # Failed events are logged in full detail
+ # They must be handled by admins manually
+ saves: list[tuple[str, str]] = []
+
+ async def fake_listen(url: str, username: str | None = None, password: str | None = None, cursor: Any = None):
+ yield {"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "a" * 36}
+ yield {"stillalive": 1.0}
+ yield {"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "b" * 36}
+ yield {"stillalive": 2.0}
+
+ async def fake_handle(payload: dict[str, Any]) -> None:
+ if payload["pubsub_cursor"] == "a" * 36:
+ raise RuntimeError("boom")
+
+ async def fake_load(stream: str) -> str | None:
+ return None
+
+ async def fake_record(payload: dict[str, Any], detail: str) -> None:
+ return None
+
+ async def fake_save(stream: str, cursor: str) -> None:
+ saves.append((stream, cursor))
+
+ monkeypatch.setattr(pubsub, "listen", fake_listen)
+ monkeypatch.setattr(pubsub, "_cursor_load", fake_load)
+ monkeypatch.setattr(pubsub, "_cursor_save", fake_save)
+ monkeypatch.setattr(pubsub, "_failure_record", fake_record)
+ monkeypatch.setattr(pubsub.commits, "handle", fake_handle)
+ listener = pubsub.PubSubListener("https://pubsub.invalid/", "user", "password")
+ await listener.start()
+ assert [cursor for _, cursor in saves] == ["b" * 36]
+ assert listener.cursor == "b" * 36
+
+
+async def test_start_does_not_checkpoint_on_cancel(monkeypatch: pytest.MonkeyPatch) -> None:
+ saves: list[tuple[str, str]] = []
+
+ async def fake_listen(url: str, username: str | None = None, password: str | None = None, cursor: Any = None):
+ yield {"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "a" * 36}
+ raise asyncio.CancelledError
+
+ async def fake_handle(payload: dict[str, Any]) -> None:
+ return None
+
+ async def fake_load(stream: str) -> str | None:
+ return None
+
+ async def fake_save(stream: str, cursor: str) -> None:
+ saves.append((stream, cursor))
+
+ monkeypatch.setattr(pubsub, "listen", fake_listen)
+ monkeypatch.setattr(pubsub, "_cursor_load", fake_load)
+ monkeypatch.setattr(pubsub, "_cursor_save", fake_save)
+ monkeypatch.setattr(pubsub.commits, "handle", fake_handle)
+ listener = pubsub.PubSubListener("https://pubsub.invalid/", "user", "password")
+ with pytest.raises(asyncio.CancelledError):
+ await listener.start()
+ assert saves == []
+ assert listener.staged == "a" * 36
+ assert listener.cursor is None
+
+
+async def test_start_halts_when_cursor_load_fails(monkeypatch: pytest.MonkeyPatch) -> None:
+ halts: list[str] = []
+ handled: list[int] = []
+
+ async def fake_halt(detail: str) -> None:
+ halts.append(detail)
+
+ async def fake_handle(payload: dict[str, Any]) -> None:
+ handled.append(payload["n"])
+
+ async def fake_listen(url: str, username: str | None = None, password: str | None = None, cursor: Any = None):
+ yield {"pubsub_topics": ["commit", "svn"], "n": 1}
+
+ async def fake_load(stream: str) -> str | None:
+ raise ValueError("damaged")
+
+ monkeypatch.setattr(pubsub, "listen", fake_listen)
+ monkeypatch.setattr(pubsub, "_cursor_load", fake_load)
+ monkeypatch.setattr(pubsub, "_halt_notify", fake_halt)
+ monkeypatch.setattr(pubsub.commits, "handle", fake_handle)
+ await pubsub.PubSubListener("https://pubsub.invalid/", "user", "password").start()
+ assert halts == ["ValueError: damaged"]
+ assert handled == []
+
+
+async def test_start_ignores_invalid_cursors(monkeypatch: pytest.MonkeyPatch) -> None:
+ saves: list[tuple[str, str]] = []
+
+ async def fake_listen(url: str, username: str | None = None, password: str | None = None, cursor: Any = None):
+ yield {"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "bad\ncursor"}
+ yield {"stillalive": 1.0}
+ yield {"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "a" * 36}
+ yield {"stillalive": 2.0}
+
+ async def fake_handle(payload: dict[str, Any]) -> None:
+ return None
+
+ async def fake_load(stream: str) -> str | None:
+ return None
+
+ async def fake_save(stream: str, cursor: str) -> None:
+ saves.append((stream, cursor))
+
+ monkeypatch.setattr(pubsub, "listen", fake_listen)
+ monkeypatch.setattr(pubsub, "_cursor_load", fake_load)
+ monkeypatch.setattr(pubsub, "_cursor_save", fake_save)
+ monkeypatch.setattr(pubsub.commits, "handle", fake_handle)
+ listener = pubsub.PubSubListener("https://pubsub.invalid/", "user", "password")
+ await listener.start()
+ assert [cursor for _, cursor in saves] == ["a" * 36]
+ assert listener.cursor == "a" * 36
+
+
async def test_start_isolates_handler_failures(monkeypatch: pytest.MonkeyPatch) -> None:
handled = []
+ records: list[tuple[dict[str, Any], str]] = []
- async def fake_listen(url: str, username: str | None = None, password: str | None = None):
+ async def fake_listen(url: str, username: str | None = None, password: str | None = None, cursor: Any = None):
yield {"pubsub_topics": ["commit", "svn"], "n": 1}
yield {"stillalive": 1.0}
yield {"pubsub_topics": ["commit", "svn"], "n": 2}
@@ -114,15 +286,104 @@ async def test_start_isolates_handler_failures(monkeypatch: pytest.MonkeyPatch)
raise RuntimeError("boom")
handled.append(payload["n"])
+ async def fake_load(stream: str) -> str | None:
+ return None
+
+ async def fake_record(payload: dict[str, Any], detail: str) -> None:
+ records.append((payload, detail))
+
monkeypatch.setattr(pubsub, "listen", fake_listen)
+ monkeypatch.setattr(pubsub, "_cursor_load", fake_load)
+ monkeypatch.setattr(pubsub, "_failure_record", fake_record)
monkeypatch.setattr(pubsub.commits, "handle", fake_handle)
await pubsub.PubSubListener("https://pubsub.invalid/", "user", "password").start()
assert handled == [2]
+ assert len(records) == 1
+ assert records[0][0]["n"] == 1
+ assert "boom" in records[0][1]
+
+
+async def test_start_logs_handler_detail(monkeypatch: pytest.MonkeyPatch) -> None:
+ records: list[tuple[dict[str, Any], str]] = []
+ saves: list[tuple[str, str]] = []
+ warnings: list[tuple[str, dict[str, Any]]] = []
+
+ async def fake_listen(url: str, username: str | None = None, password: str | None = None, cursor: Any = None):
+ yield {"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "a" * 36}
+ yield {"stillalive": 1.0}
+
+ async def fake_handle(payload: dict[str, Any]) -> str | None:
+ return "problem detail"
+
+ async def fake_load(stream: str) -> str | None:
+ return None
+
+ async def fake_record(payload: dict[str, Any], detail: str) -> None:
+ records.append((payload, detail))
+
+ async def fake_save(stream: str, cursor: str) -> None:
+ saves.append((stream, cursor))
+
+ monkeypatch.setattr(pubsub, "listen", fake_listen)
+ monkeypatch.setattr(pubsub, "_cursor_load", fake_load)
+ monkeypatch.setattr(pubsub, "_cursor_save", fake_save)
+ monkeypatch.setattr(pubsub, "_failure_record", fake_record)
+ monkeypatch.setattr(pubsub.commits, "handle", fake_handle)
+ monkeypatch.setattr(pubsub.log, "warning", lambda msg, **kwargs: warnings.append((msg, kwargs)))
+ await pubsub.PubSubListener("https://pubsub.invalid/", "user", "password").start()
+ assert any("problem detail" in msg for msg, _ in warnings)
+ assert all("payload" not in kwargs for _, kwargs in warnings)
+ assert records == [({"pubsub_topics": ["commit", "svn"], "pubsub_cursor": "a" * 36}, "problem detail")]
+ assert [cursor for _, cursor in saves] == ["a" * 36]
+
+
+async def test_start_retries_cursor_save_without_stopping(monkeypatch: pytest.MonkeyPatch) -> None:
+ attempts: list[tuple[str, str]] = []
+ handled: list[int] = []
+
+ async def fake_listen(url: str, username: str | None = None, password: str | None = None, cursor: Any = None):
+ yield {
+ "pubsub_topics": ["commit", "svn"],
+ "pubsub_cursor": "a" * 36,
+ "n": 1,
+ }
+ yield {"stillalive": 1.0}
+ yield {"stillalive": 2.0}
+ yield {
+ "pubsub_topics": ["commit", "svn"],
+ "pubsub_cursor": "b" * 36,
+ "n": 2,
+ }
+
+ async def fake_handle(payload: dict[str, Any]) -> None:
+ handled.append(payload["n"])
+
+ async def fake_load(stream: str) -> str | None:
+ return None
+
+ async def fake_save(stream: str, cursor: str) -> None:
+ attempts.append((stream, cursor))
+ if len(attempts) == 1:
+ raise RuntimeError("database unavailable")
+
+ monkeypatch.setattr(pubsub, "listen", fake_listen)
+ monkeypatch.setattr(pubsub, "_cursor_load", fake_load)
+ monkeypatch.setattr(pubsub, "_cursor_save", fake_save)
+ monkeypatch.setattr(pubsub.commits, "handle", fake_handle)
+ listener = pubsub.PubSubListener("https://pubsub.invalid/", "user", "password")
+ await listener.start()
+ assert attempts == [
+ ("https://pubsub.invalid/commit/svn,ldap", "a" * 36),
+ ("https://pubsub.invalid/commit/svn,ldap", "a" * 36),
+ ]
+ assert handled == [1, 2]
+ assert listener.cursor == "a" * 36
+ assert listener.staged == "b" * 36
-async def _collect(url: str, count: int) -> list[dict[str, Any]]:
+async def _collect(url: str, count: int, cursor: str | None = None) -> list[dict[str, Any]]:
payloads: list[dict[str, Any]] = []
- gen = pubsub.listen(url)
+ gen = pubsub.listen(url, cursor=(lambda: cursor) if cursor else None)
async with asyncio.timeout(5):
async for payload in gen:
payloads.append(payload)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@tooling.apache.org
For additional commands, e-mail: commits-help@tooling.apache.org