diff --git a/.env.example b/.env.example index f3cba2f..42b1860 100644 --- a/.env.example +++ b/.env.example @@ -22,3 +22,10 @@ VALID_X_GITLAB_TOKEN=your-webhook-secret-token-1,your-webhook-secret-token-2 # GitLab API tokens for fetching MR discussion stats (requires read_api scope) GITLAB_API_TOKENS='{"gitlab-saas":{"url":"https://gitlab.com","token":"glpat-xxxxxxxx"}}' + +# Outbound HTTP timeouts, in seconds +# ACTIVITY_API_TIMEOUT_SECONDS: activity-api calls (default 10.0). Must exceed observed Teams +# delivery latency, which has been measured well above the default. +ACTIVITY_API_TIMEOUT_SECONDS=10.0 +# GITLAB_API_TIMEOUT_SECONDS: GitLab API calls (default 5.0) +GITLAB_API_TIMEOUT_SECONDS=5.0 diff --git a/config.py b/config.py index 5e8300c..5e6889f 100644 --- a/config.py +++ b/config.py @@ -5,6 +5,7 @@ from dataclasses import dataclass import dotenv +import httpx dotenv.load_dotenv() @@ -28,6 +29,8 @@ class DefaultConfig: LOG_QUERIES = os.environ.get("LOG_QUERIES", "") VALID_X_GITLAB_TOKEN = os.environ.get("VALID_X_GITLAB_TOKEN", "") MESSAGE_DELETE_DELAY_SECONDS = int(os.environ.get("MESSAGE_DELETE_DELAY_SECONDS", "30")) + ACTIVITY_API_TIMEOUT_SECONDS = float(os.environ.get("ACTIVITY_API_TIMEOUT_SECONDS", "10.0")) + GITLAB_API_TIMEOUT_SECONDS = float(os.environ.get("GITLAB_API_TIMEOUT_SECONDS", "5.0")) NOTE_DEBOUNCE_SECONDS = float(os.environ.get("NOTE_DEBOUNCE_SECONDS", "5.0")) EMOJI_DEBOUNCE_SECONDS = float(os.environ.get("EMOJI_DEBOUNCE_SECONDS", "5.0")) _valid_tokens: list[str] @@ -51,6 +54,12 @@ def __init__(self): except (json.JSONDecodeError, KeyError, TypeError): pass + def activity_api_timeout(self) -> httpx.Timeout: + return httpx.Timeout(self.ACTIVITY_API_TIMEOUT_SECONDS, connect=self.ACTIVITY_API_TIMEOUT_SECONDS / 2) + + def gitlab_api_timeout(self) -> httpx.Timeout: + return httpx.Timeout(self.GITLAB_API_TIMEOUT_SECONDS, connect=self.GITLAB_API_TIMEOUT_SECONDS / 2) + def is_valid_token(self, token: str) -> bool: return token.lower() in self._valid_tokens diff --git a/gitlab_api.py b/gitlab_api.py index 38d5780..5095cde 100644 --- a/gitlab_api.py +++ b/gitlab_api.py @@ -80,7 +80,7 @@ async def _fetch_mr_discussion_stats( ) try: - timeout = httpx.Timeout(5.0, connect=2.0) + timeout = config.gitlab_api_timeout() async with httpx.AsyncClient(timeout=timeout) as client: discussions: list[dict[str, Any]] = [] url: str | None = base_url @@ -220,7 +220,7 @@ async def _fetch_mr_status( url = f"{api_token.url.rstrip('/')}/api/v4/projects/{encoded_project_id}/merge_requests/{mr_iid}" try: - timeout = httpx.Timeout(5.0, connect=2.0) + timeout = config.gitlab_api_timeout() async with httpx.AsyncClient(timeout=timeout) as client: response = await client.get( url, diff --git a/periodic_cleanup.py b/periodic_cleanup.py index 490bb9e..da3d2a2 100644 --- a/periodic_cleanup.py +++ b/periodic_cleanup.py @@ -17,6 +17,7 @@ from db import make_mr_summary from gitlab_api import fetch_and_persist_discussion_stats from gitlab_api import fetch_and_refresh_mr_status +from webhook.messaging import is_unknown_message_id from webhook.messaging import update_all_messages_transactional @@ -191,7 +192,7 @@ async def _process_pending_refreshes() -> int: async def _cleanup_task(config: DefaultConfig, database: DatabaseLifecycleHandler): - timeout = httpx.Timeout(10.0, connect=5.0) + timeout = config.activity_api_timeout() client = httpx.AsyncClient(timeout=timeout) while True: wait_sec: float = MAX_WAIT @@ -218,7 +219,9 @@ async def _cleanup_task(config: DefaultConfig, database: DatabaseLifecycleHandle "message_id": str(record["message_id"]), }, ) - if res.status_code not in (410, 200): + # An id activity-api cannot resolve was claimed but never delivered: + # there is nothing to delete, so drop the row instead of retrying it. + if res.status_code not in (200, 410) and not is_unknown_message_id(res): res.raise_for_status() await connection.execute( "DELETE FROM msg_to_delete WHERE msg_to_delete_id = $1", diff --git a/requirements-dev.txt b/requirements-dev.txt index 0db9554..11e49b2 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -4,5 +4,7 @@ pytest-asyncio>=0.23.0 pytest-mock>=3.12.0 pytest-cov>=4.1.0 respx>=0.21.0 +# starlette 1.x TestClient imports httpx2, not httpx, and raises on the fallback +httpx2>=2.12.0 testcontainers>=3.7.0 psycopg2-binary>=2.9.0 diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..1a719aa --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,42 @@ +"""Guards for the env-driven outbound timeouts. + +A typo in an env var name would leave the shipped defaults in place with nothing to show for it, +so the wiring itself is what these assert, not just the helper arithmetic. +""" + +import importlib + +import config as config_module + + +def _reloaded_config(): + importlib.reload(config_module) + return config_module.DefaultConfig() + + +def test_timeouts_fall_back_to_shipped_defaults(monkeypatch): + monkeypatch.delenv("ACTIVITY_API_TIMEOUT_SECONDS", raising=False) + monkeypatch.delenv("GITLAB_API_TIMEOUT_SECONDS", raising=False) + + try: + cfg = _reloaded_config() + assert cfg.activity_api_timeout().read == 10.0 + assert cfg.activity_api_timeout().connect == 5.0 + assert cfg.gitlab_api_timeout().read == 5.0 + assert cfg.gitlab_api_timeout().connect == 2.5 + finally: + importlib.reload(config_module) + + +def test_timeouts_read_their_env_vars(monkeypatch): + monkeypatch.setenv("ACTIVITY_API_TIMEOUT_SECONDS", "42.0") + monkeypatch.setenv("GITLAB_API_TIMEOUT_SECONDS", "7.0") + + try: + cfg = _reloaded_config() + assert cfg.activity_api_timeout().read == 42.0 + assert cfg.activity_api_timeout().connect == 21.0 + assert cfg.gitlab_api_timeout().read == 7.0 + assert cfg.gitlab_api_timeout().connect == 3.5 + finally: + importlib.reload(config_module) diff --git a/tests/test_e2e_race_conditions.py b/tests/test_e2e_race_conditions.py index 3eda828..d35fd92 100644 --- a/tests/test_e2e_race_conditions.py +++ b/tests/test_e2e_race_conditions.py @@ -1033,8 +1033,8 @@ async def test_e2e_race_update_arrives_before_open_with_older_timestamp( 7. Result: DB points to deleted M2, actual M1 orphaned After fix: - - create_or_update_message() returns None when duplicate detected - - DB keeps M1, no orphaned messages + - the id is claimed in the database before the create call, so the loser never posts + - DB keeps M1, no orphaned messages, no compensating delete Validates: - Duplicate message detection works correctly @@ -1110,16 +1110,13 @@ async def test_e2e_race_update_arrives_before_open_with_older_timestamp( message_id = msg_refs[0]["message_id"] assert message_id is not None, "Message ref should have a valid message_id" - # Count API requests: POST creates, DELETE removes duplicates requests = mock_activity_api["requests"] post_requests = [r for r in requests if r["method"] == "POST"] delete_requests = [r for r in requests if r["method"] == "DELETE"] - # With the fix: if duplicate detected, message is deleted and create_or_update_message returns None - # So we should have at most 1 successful create (no orphaned messages) - # Under race: 1 or 2 POSTs depending on timing, but at most 1 DELETE - assert len(post_requests) >= 1, "At least one message should be created" - assert len(delete_requests) <= 1, "At most one duplicate should be deleted" + # Claiming the id first means only one request can post, and nothing needs undoing. + assert len(post_requests) == 1, "Exactly one message should be created" + assert len(delete_requests) == 0, "No compensating delete should be needed" # Verify timestamp is from the newer event (Open with T1) final_updated_at = msg_refs[0]["last_processed_updated_at"] diff --git a/tests/test_merge_request_deletion.py b/tests/test_merge_request_deletion.py index 9782c7e..ac1d6d7 100644 --- a/tests/test_merge_request_deletion.py +++ b/tests/test_merge_request_deletion.py @@ -112,18 +112,16 @@ async def test_merge_close_transactional_rollback_on_failure(mock_database, samp @pytest.mark.asyncio -async def test_race_condition_duplicate_message_deletion_logs_failure(mock_database): +async def test_race_condition_loser_sends_nothing(mock_database): """ - POSITIVE TEST: Race condition handling gracefully logs DELETE failures. + POSITIVE TEST: concurrent creates cannot produce two messages. - Scenario: Two webhooks create messages simultaneously, second one's DELETE fails - Result: DELETE failure is logged but doesn't crash; returns None to signal duplicate - Impact: Graceful degradation - duplicate may persist in Teams but caller knows not to use it + Scenario: two webhooks race to create the message for the same ref + Result: the request that loses the id claim never calls activity-api and returns None + Impact: no duplicate in Teams, and no compensating delete to get wrong Location: webhook/messaging.py:create_or_update_message """ - import httpx - from webhook.messaging import MRMessRef from webhook.messaging import create_or_update_message @@ -131,18 +129,6 @@ async def test_race_condition_duplicate_message_deletion_logs_failure(mock_datab connection.fetchrow.return_value = None client = AsyncMock() - create_response = MagicMock() - create_response.status_code = 200 - new_message_id = str(uuid.uuid4()) - create_response.json.return_value = {"message_id": new_message_id} - - delete_response = MagicMock() - delete_response.status_code = 500 - delete_response.raise_for_status.side_effect = httpx.HTTPStatusError( - "Server error", request=MagicMock(), response=delete_response - ) - - client.request.side_effect = [create_response, delete_response] mrmsgref = MRMessRef( merge_request_message_ref_id=1, @@ -159,7 +145,7 @@ async def test_race_condition_duplicate_message_deletion_logs_failure(mock_datab ) assert result is None - assert client.request.call_count == 2 + client.request.assert_not_called() assert connection.fetchrow.call_count == 1 diff --git a/tests/test_message_operations.py b/tests/test_message_operations.py index cee8d83..5e3737f 100644 --- a/tests/test_message_operations.py +++ b/tests/test_message_operations.py @@ -149,22 +149,15 @@ async def test_get_or_create_returns_existing_and_new_refs(mock_database): @pytest.mark.asyncio -async def test_duplicate_message_cleanup_on_db_conflict(mock_database): - """Test that duplicate message is deleted when DB update returns NULL (conflict).""" +async def test_no_duplicate_sent_when_id_claim_lost(mock_database): + """A ref whose id another request already claimed must not send a second message.""" from webhook.messaging import create_or_update_message connection = mock_database.connection connection.fetchrow.return_value = None - create_response = MagicMock() - create_response.status_code = 200 - create_response.json.return_value = {"message_id": str(uuid.uuid4())} - - delete_response = MagicMock() - delete_response.status_code = 200 - client = AsyncMock() - client.request = AsyncMock(side_effect=[create_response, delete_response]) + client.request = AsyncMock() mrmsgref = MRMessRef( merge_request_message_ref_id=1, @@ -173,20 +166,15 @@ async def test_duplicate_message_cleanup_on_db_conflict(mock_database): ) with patch("webhook.messaging.database", mock_database): - await create_or_update_message( + result = await create_or_update_message( client=client, mrmsgref=mrmsgref, card={"type": "AdaptiveCard"}, summary="Test", ) - assert client.request.call_count == 2 - - first_call = client.request.call_args_list[0] - assert first_call[0][0] == "POST" - - second_call = client.request.call_args_list[1] - assert second_call[0][0] == "DELETE" + assert result is None + client.request.assert_not_called() @pytest.mark.asyncio diff --git a/tests/test_messaging.py b/tests/test_messaging.py index 874e5de..5adb65e 100644 --- a/tests/test_messaging.py +++ b/tests/test_messaging.py @@ -12,6 +12,7 @@ from webhook.messaging import MRMessRef from webhook.messaging import create_or_update_message from webhook.messaging import get_or_create_message_refs +from webhook.messaging import is_unknown_message_id from webhook.messaging import update_all_messages_transactional from webhook.messaging import update_message_with_fingerprint @@ -191,33 +192,192 @@ async def test_update_only_skips_creation(self, mock_http_client, sample_mr_mess assert result is None mock_http_client.request.assert_not_called() - async def test_deletes_duplicate_message_on_db_conflict( - self, mock_http_client, mock_database, sample_mr_mess_ref - ): + async def test_claims_id_before_posting(self, mock_http_client, mock_database, sample_mr_mess_ref): mock_db, mock_conn = mock_database sample_mr_mess_ref.message_id = None - new_message_id = str(uuid.uuid4()) + + order = [] response_mock = MagicMock() response_mock.status_code = 200 - response_mock.json.return_value = {"message_id": new_message_id} - mock_http_client.request.return_value = response_mock + response_mock.json.side_effect = lambda: {"message_id": str(mock_conn.fetchrow.call_args[0][1])} + + async def record_claim(*args, **kwargs): + order.append("claim") + return {"merge_request_message_ref_id": 123} + + async def record_post(*args, **kwargs): + order.append("post") + return response_mock + + mock_conn.fetchrow.side_effect = record_claim + mock_http_client.request.side_effect = record_post + + result = await create_or_update_message( + mock_http_client, + sample_mr_mess_ref, + card={"body": []}, + ) + assert order == ["claim", "post"] + claim_sql, claimed_id, _ = mock_conn.fetchrow.call_args[0] + assert "SET message_id = $1" in claim_sql + assert "AND message_id IS NULL" in claim_sql + # The id reaches the database before it reaches activity-api, and it is the one posted. + assert mock_http_client.request.call_args[1]["json"]["message_id"] == str(claimed_id) + assert result == claimed_id + + async def test_no_post_when_id_already_claimed(self, mock_http_client, mock_database, sample_mr_mess_ref): + mock_db, mock_conn = mock_database + sample_mr_mess_ref.message_id = None mock_conn.fetchrow.return_value = None - await create_or_update_message( + result = await create_or_update_message( mock_http_client, sample_mr_mess_ref, card={"body": []}, ) - assert mock_http_client.request.call_count == 2 - delete_call = mock_http_client.request.call_args_list[1] - assert delete_call[0][0] == "DELETE" - assert delete_call[1]["json"]["message_id"] == new_message_id + assert result is None + mock_http_client.request.assert_not_called() + + @pytest.mark.parametrize( + "failure", + [ + httpx.ReadTimeout("timed out"), + httpx.HTTPError("Connection failed"), + ], + ) + async def test_claimed_id_persists_when_create_call_fails( + self, mock_http_client, mock_database, sample_mr_mess_ref, failure + ): + """A create call that fails after Teams accepted the card must still leave a usable id. + + Losing it is what stranded MR !367's message: activity-api answered 201 after 17s, the + client had already timed out at 10s, and nothing could delete the message afterwards. + """ + mock_db, mock_conn = mock_database + sample_mr_mess_ref.message_id = None + mock_conn.fetchrow.return_value = {"merge_request_message_ref_id": 123} + mock_http_client.request.side_effect = failure + + with pytest.raises(type(failure)): + await create_or_update_message( + mock_http_client, + sample_mr_mess_ref, + card={"body": []}, + ) + + claim_sql, claimed_id, ref_id = mock_conn.fetchrow.call_args[0] + assert "UPDATE merge_request_message_ref" in claim_sql + assert isinstance(claimed_id, uuid.UUID) + assert ref_id == sample_mr_mess_ref.merge_request_message_ref_id + assert mock_http_client.request.call_args[1]["json"]["message_id"] == str(claimed_id) + + @pytest.mark.parametrize("refused_status", [400, 409]) + async def test_refused_create_gives_the_claimed_id_back( + self, mock_http_client, mock_database, sample_mr_mess_ref, refused_status + ): + """activity-api answers 400/409 without touching Teams, so the ref must not keep the id. + + Keeping it is worse than the bug it fixes: every later event patches an id activity-api + does not know, and the merge request never gets a card at all. + """ + mock_db, mock_conn = mock_database + sample_mr_mess_ref.message_id = None + mock_conn.fetchrow.return_value = {"merge_request_message_ref_id": 123} + + refused = MagicMock() + refused.status_code = refused_status + refused.raise_for_status.side_effect = httpx.HTTPStatusError( + "refused", request=MagicMock(), response=refused + ) + mock_http_client.request.return_value = refused + + with pytest.raises(httpx.HTTPStatusError): + await create_or_update_message( + mock_http_client, + sample_mr_mess_ref, + card={"body": []}, + ) + + claimed_id = mock_conn.fetchrow.call_args[0][1] + releases = [c for c in mock_conn.execute.call_args_list if "SET message_id = NULL" in str(c[0][0])] + assert len(releases) == 1 + assert releases[0][0][2] == claimed_id + + async def test_ambiguous_create_failure_keeps_the_claimed_id( + self, mock_http_client, mock_database, sample_mr_mess_ref + ): + """A timeout may mean Teams got the card, so the id stays: it is the only handle left.""" + mock_db, mock_conn = mock_database + sample_mr_mess_ref.message_id = None + mock_conn.fetchrow.return_value = {"merge_request_message_ref_id": 123} + mock_http_client.request.side_effect = httpx.ReadTimeout("timed out") + + with pytest.raises(httpx.ReadTimeout): + await create_or_update_message( + mock_http_client, + sample_mr_mess_ref, + card={"body": []}, + ) + + releases = [c for c in mock_conn.execute.call_args_list if "SET message_id = NULL" in str(c[0][0])] + assert releases == [] + + async def test_patch_on_unknown_id_gives_it_back( + self, mock_http_client, mock_database, sample_mr_mess_ref + ): + mock_db, mock_conn = mock_database + stale_id = sample_mr_mess_ref.message_id + + unknown = MagicMock() + unknown.status_code = 400 + unknown.json.return_value = {"detail": "invalid message_id"} + unknown.raise_for_status.side_effect = httpx.HTTPStatusError( + "bad", request=MagicMock(), response=unknown + ) + mock_http_client.request.return_value = unknown + + with pytest.raises(httpx.HTTPStatusError): + await create_or_update_message( + mock_http_client, + sample_mr_mess_ref, + card={"body": []}, + ) + + releases = [c for c in mock_conn.execute.call_args_list if "SET message_id = NULL" in str(c[0][0])] + assert len(releases) == 1 + assert releases[0][0][2] == stale_id + + async def test_patch_failure_that_is_not_an_unknown_id_keeps_it( + self, mock_http_client, mock_database, sample_mr_mess_ref + ): + mock_db, mock_conn = mock_database + + server_error = MagicMock() + server_error.status_code = 500 + server_error.raise_for_status.side_effect = httpx.HTTPStatusError( + "boom", request=MagicMock(), response=server_error + ) + mock_http_client.request.return_value = server_error + + with pytest.raises(httpx.HTTPStatusError): + await create_or_update_message( + mock_http_client, + sample_mr_mess_ref, + card={"body": []}, + ) + + releases = [c for c in mock_conn.execute.call_args_list if "SET message_id = NULL" in str(c[0][0])] + assert releases == [] - async def test_error_on_create_raises_exception(self, mock_http_client, sample_mr_mess_ref): + async def test_error_on_create_raises_exception( + self, mock_http_client, mock_database, sample_mr_mess_ref + ): + mock_db, mock_conn = mock_database sample_mr_mess_ref.message_id = None + mock_conn.fetchrow.return_value = {"merge_request_message_ref_id": 123} mock_http_client.request.side_effect = httpx.HTTPError("Connection failed") @@ -397,6 +557,45 @@ async def test_schedules_message_deletion_when_requested(self, mock_database, sa assert any("msg_to_delete" in call for call in execute_calls) assert any("DELETE FROM merge_request_message_ref" in call for call in execute_calls) + async def test_warns_instead_of_dropping_a_ref_without_message_id(self, mock_database, sample_mri): + """A ref whose creation call never returned an id must be reported, not silently dropped. + + This is the state MR !367's ref was in when it closed: no msg_to_delete row was written + and the ref was deleted anyway, leaving the Teams message with nothing pointing at it. + """ + mock_db, mock_conn = mock_database + + mock_conn.fetch.return_value = [ + { + "merge_request_message_ref_id": 1, + "conversation_token": uuid.uuid4(), + "message_id": None, + } + ] + + with ( + patch("webhook.messaging.httpx.AsyncClient"), + patch("webhook.messaging.logger") as mock_logger, + ): + await update_all_messages_transactional( + sample_mri, + {"body": []}, + "summary", + "fingerprint", + datetime.datetime(2025, 1, 1, tzinfo=datetime.UTC), + "close/merge", + schedule_deletion=True, + deletion_delay=datetime.timedelta(seconds=0), + ) + + execute_calls = [str(call[0][0]) for call in mock_conn.execute.call_args_list] + assert not any("msg_to_delete" in call for call in execute_calls) + assert any("DELETE FROM merge_request_message_ref" in call for call in execute_calls) + + warnings = [call for call in mock_logger.warning.call_args_list if "orphan" in call[0][0]] + assert len(warnings) == 1 + assert warnings[0][1]["merge_request_message_ref_id"] == 1 + async def test_continues_on_http_error_per_message(self, mock_database, sample_mri): mock_db, mock_conn = mock_database message_id_1 = uuid.uuid4() @@ -493,37 +692,6 @@ async def test_create_with_text_message(self, mock_http_client, mock_database, s call_args = mock_http_client.request.call_args assert call_args[1]["json"]["text"] == "Hello World" - async def test_handles_delete_failure_gracefully( - self, mock_http_client, mock_database, sample_mr_mess_ref - ): - mock_db, mock_conn = mock_database - sample_mr_mess_ref.message_id = None - new_message_id = str(uuid.uuid4()) - - response_mock = MagicMock() - response_mock.status_code = 200 - response_mock.json.return_value = {"message_id": new_message_id} - - call_count = 0 - - async def side_effect(*args, **kwargs): - nonlocal call_count - call_count += 1 - if call_count == 1: - return response_mock - raise httpx.HTTPError("Delete failed") - - mock_http_client.request.side_effect = side_effect - mock_conn.fetchrow.return_value = None - - await create_or_update_message( - mock_http_client, - sample_mr_mess_ref, - card={"body": []}, - ) - - assert mock_http_client.request.call_count == 2 - class TestUpdateAllMessagesTransactionalOrdering: """Tests for out-of-order and duplicate event handling in transactional updates.""" @@ -845,3 +1013,66 @@ def test_allows_none_message_id(self): assert ref.message_id is None assert ref.last_processed_fingerprint is None + + +class TestUnknownMessageIdDetection: + """`is_unknown_message_id` decides whether a claimed id gets given back, so it must not + mistake an unrelated 400 for activity-api's `invalid message_id`.""" + + def test_recognises_the_unknown_id_answer(self): + res = MagicMock() + res.status_code = 400 + res.json.return_value = {"detail": "invalid message_id"} + assert is_unknown_message_id(res) is True + + @pytest.mark.parametrize( + "status_code, json_value", + [ + (400, {"detail": "invalid conversation_token"}), + (400, {"detail": "message deleted, can't be updated"}), + (409, {"detail": "invalid message_id"}), + (500, {"detail": "invalid message_id"}), + ], + ) + def test_rejects_everything_else(self, status_code, json_value): + res = MagicMock() + res.status_code = status_code + res.json.return_value = json_value + assert is_unknown_message_id(res) is False + + def test_survives_a_body_that_is_not_json(self): + res = MagicMock() + res.status_code = 400 + res.json.side_effect = ValueError("not json") + assert is_unknown_message_id(res) is False + + def test_handles_no_response_at_all(self): + assert is_unknown_message_id(None) is False + + +class TestUpdateMessageWithFingerprintRelease: + async def test_unknown_id_is_given_back(self, mock_http_client, mock_database, sample_mr_mess_ref): + mock_db, mock_conn = mock_database + stale_id = sample_mr_mess_ref.message_id + + unknown = MagicMock() + unknown.status_code = 400 + unknown.json.return_value = {"detail": "invalid message_id"} + unknown.raise_for_status.side_effect = httpx.HTTPStatusError( + "bad", request=MagicMock(), response=unknown + ) + mock_http_client.request.return_value = unknown + + with pytest.raises(httpx.HTTPStatusError): + await update_message_with_fingerprint( + mock_http_client, + sample_mr_mess_ref, + {"body": []}, + "summary", + "fingerprint", + datetime.datetime(2025, 1, 1, tzinfo=datetime.UTC), + ) + + releases = [c for c in mock_conn.execute.call_args_list if "SET message_id = NULL" in str(c[0][0])] + assert len(releases) == 1 + assert releases[0][0][2] == stale_id diff --git a/tests/test_periodic_cleanup.py b/tests/test_periodic_cleanup.py index 02bbbf3..0bbef3a 100644 --- a/tests/test_periodic_cleanup.py +++ b/tests/test_periodic_cleanup.py @@ -218,6 +218,99 @@ async def test_successful_deletion_with_410_gone(mock_database, fresh_signal): assert delete_call[0][1] == msg_to_delete_id +@pytest.mark.asyncio +async def test_deletion_with_400_invalid_message_id_drops_the_record(mock_database, fresh_signal): + """ + POSITIVE TEST: 400 invalid message_id is terminal. + + Scenario: the id was claimed but the card never reached Teams, so activity-api has no row + Result: record removed from msg_to_delete instead of retried forever + + Location: periodic_cleanup.py:_cleanup_task + """ + from config import DefaultConfig + from periodic_cleanup import _cleanup_task + + connection = mock_database.connection + + msg_to_delete_id = 1 + records = [{"msg_to_delete_id": msg_to_delete_id, "message_id": uuid.uuid4()}] + + prepared_stmt = MockPreparedStatement(records) + connection.prepare.return_value = prepared_stmt + connection.fetchval.return_value = None + + client = AsyncMock() + invalid_response = MagicMock() + invalid_response.status_code = 400 + invalid_response.json.return_value = {"detail": "invalid message_id"} + invalid_response.raise_for_status.side_effect = AssertionError("400 must not be raised") + client.request.return_value = invalid_response + + config = DefaultConfig() + + with ( + patch("periodic_cleanup.httpx.AsyncClient", return_value=client), + patch("periodic_cleanup.signal", fresh_signal), + ): + task = _cleanup_task(config, mock_database) + try: + await asyncio.wait_for(task, timeout=0.1) + except TimeoutError: + pass + + assert connection.execute.call_count == 1 + delete_call = connection.execute.call_args_list[0] + assert "DELETE FROM msg_to_delete" in delete_call[0][0] + assert delete_call[0][1] == msg_to_delete_id + + +@pytest.mark.asyncio +async def test_deletion_with_an_unrelated_400_keeps_the_record(mock_database, fresh_signal): + """ + NEGATIVE TEST: a 400 that is not "invalid message_id" must not be treated as terminal. + + Scenario: a proxy or a future validation error answers 400 + Result: the record stays in msg_to_delete instead of being dropped with the message still up + + Location: periodic_cleanup.py:_cleanup_task + """ + from config import DefaultConfig + from periodic_cleanup import _cleanup_task + + connection = mock_database.connection + + records = [{"msg_to_delete_id": 1, "message_id": uuid.uuid4()}] + + prepared_stmt = MockPreparedStatement(records) + connection.prepare.return_value = prepared_stmt + connection.fetchval.return_value = None + + client = AsyncMock() + other_400 = MagicMock() + other_400.status_code = 400 + other_400.json.return_value = {"detail": "invalid conversation_token"} + other_400.raise_for_status.side_effect = httpx.HTTPStatusError( + "bad request", request=MagicMock(), response=other_400 + ) + client.request.return_value = other_400 + + config = DefaultConfig() + + with ( + patch("periodic_cleanup.httpx.AsyncClient", return_value=client), + patch("periodic_cleanup.signal", fresh_signal), + ): + task = _cleanup_task(config, mock_database) + try: + await asyncio.wait_for(task, timeout=0.1) + except TimeoutError: + pass + + deletes = [c for c in connection.execute.call_args_list if "DELETE FROM msg_to_delete" in c[0][0]] + assert deletes == [] + + @pytest.mark.asyncio async def test_successful_deletion_with_200_ok(mock_database, fresh_signal): """ diff --git a/webhook/merge_request.py b/webhook/merge_request.py index e3b9df4..f904b94 100644 --- a/webhook/merge_request.py +++ b/webhook/merge_request.py @@ -227,7 +227,7 @@ async def merge_request( messages_processed = 0 messages_failed = 0 - timeout = httpx.Timeout(10.0, connect=5.0) + timeout = config.activity_api_timeout() async with httpx.AsyncClient(timeout=timeout) as client: for mrmsgref in all_message_refs: if mrmsgref.last_processed_updated_at is not None: diff --git a/webhook/messaging.py b/webhook/messaging.py index c418502..7e0c871 100644 --- a/webhook/messaging.py +++ b/webhook/messaging.py @@ -26,6 +26,35 @@ class MRMessRef(BaseModel): last_processed_updated_at: datetime.datetime | None = None +def is_unknown_message_id(res: httpx.Response | None) -> bool: + """activity-api answers 400 `invalid message_id` when it holds no row for that id.""" + if res is None or res.status_code != 400: + return False + try: + return bool(res.json().get("detail") == "invalid message_id") + except Exception: + return False + + +async def release_stale_message_id(merge_request_message_ref_id: int, message_id: uuid.UUID) -> None: + """Give a claimed id back so the next event creates the message instead of patching a ghost.""" + connection: asyncpg.Connection + async with await database.acquire() as connection: + await connection.execute( + """UPDATE merge_request_message_ref + SET message_id = NULL + WHERE merge_request_message_ref_id = $1 + AND message_id = $2""", + merge_request_message_ref_id, + message_id, + ) + logger.warning( + "released stale message id - next event recreates the message", + merge_request_message_ref_id=merge_request_message_ref_id, + message_id=str(message_id), + ) + + async def get_or_create_message_refs( merge_request_ref_id: int, conv_tokens: list[str], @@ -143,7 +172,31 @@ async def create_or_update_message( if update_only is True: return None + # Claim the id before calling activity-api. A POST that times out after Teams already + # accepted the card used to leave a message no later patch or delete could ever reach. + # Claiming first also settles the concurrent-create race: the loser never POSTs. + reserved_id = uuid.uuid4() + connection: asyncpg.Connection + async with await database.acquire() as connection: + result = await connection.fetchrow( + """UPDATE merge_request_message_ref + SET message_id = $1 + WHERE merge_request_message_ref_id = $2 + AND message_id IS NULL + RETURNING merge_request_message_ref_id + """, + reserved_id, + mrmsgref.merge_request_message_ref_id, + ) + if result is None: + logger.warning( + "message id already claimed by a concurrent request - deferring to it", + merge_request_message_ref_id=mrmsgref.merge_request_message_ref_id, + ) + return None + payload["conversation_token"] = str(mrmsgref.conversation_token) + payload["message_id"] = str(reserved_id) try: res = await client.request( "POST", @@ -158,42 +211,17 @@ async def create_or_update_message( method="POST", url=config.ACTIVITY_API + "api/v1/message", conversation_token=str(mrmsgref.conversation_token), + message_id=str(reserved_id), status_code=res.status_code if "res" in locals() else None, project_id=project_id, mr_iid=mr_iid, exc_info=True, ) + # 400 and 409 are both answered without anything reaching Teams, so the claimed id + # names nothing and must go back: keeping it locks the ref onto a ghost forever. + if "res" in locals() and res.status_code in (400, 409): + await release_stale_message_id(mrmsgref.merge_request_message_ref_id, reserved_id) raise - - connection: asyncpg.Connection - async with await database.acquire() as connection: - result = await connection.fetchrow( - """UPDATE merge_request_message_ref - SET message_id = $1 - WHERE merge_request_message_ref_id = $2 - AND message_id IS NULL - RETURNING merge_request_message_ref_id - """, - response.get("message_id"), - mrmsgref.merge_request_message_ref_id, - ) - if result is None or len(result) == 0: - logger.warning( - "duplicate message detected - another request already set message_id", - merge_request_message_ref_id=mrmsgref.merge_request_message_ref_id, - duplicate_message_id=response.get("message_id"), - ) - try: - await client.request( - "DELETE", - config.ACTIVITY_API + "api/v1/message", - json={ - "message_id": str(response.get("message_id")), - }, - ) - except Exception: - logger.exception("Failed to delete duplicate message %s", response.get("message_id")) - return None else: payload["message_id"] = str(mrmsgref.message_id) try: @@ -215,6 +243,8 @@ async def create_or_update_message( mr_iid=mr_iid, exc_info=True, ) + if is_unknown_message_id(res if "res" in locals() else None): + await release_stale_message_id(mrmsgref.merge_request_message_ref_id, mrmsgref.message_id) raise return uuid.UUID(response.get("message_id")) @@ -296,6 +326,8 @@ async def update_message_with_fingerprint( mr_iid=mr_iid, exc_info=True, ) + if is_unknown_message_id(res if "res" in locals() else None): + await release_stale_message_id(mrmsgref.merge_request_message_ref_id, mrmsgref.message_id) raise @@ -324,7 +356,7 @@ async def update_all_messages_transactional( ) connection: asyncpg.Connection - timeout = httpx.Timeout(10.0, connect=5.0) + timeout = config.activity_api_timeout() message_count = 0 async with await database.acquire() as connection: @@ -433,6 +465,14 @@ async def update_all_messages_transactional( str(message_id), deletion_delay, ) + else: + logger.warning( + "orphan message at deletion - ref has no message_id", + action=action_name, + merge_request_message_ref_id=row.get("merge_request_message_ref_id"), + conversation_token=str(row.get("conversation_token")), + mr_ref_id=mri.merge_request_ref_id, + ) await connection.execute( """DELETE FROM merge_request_message_ref WHERE merge_request_message_ref_id = $1""",