fix: claim the message id before creating the message

.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

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
 

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,

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",

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

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)

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"]

tests/test_merge_request_deletion.py

@@ -112,37 +112,23 @@ 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
 
     connection = mock_database.connection
     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
 
 

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