From 3fa011d95f31c6e8f45bd765ff32957e03e29df8 Mon Sep 17 00:00:00 2001
From: RishavTiwari25 <rishavt413@gmail.com>
Date: Mon, 26 Oct 2026 15:38:30 +0530
Subject: [PATCH] fix: retry events when reading the HTTP response fails

urlopen() wraps only errors raised while sending the request in URLError.
Errors raised while reading the response escaped HttpClient.post(): a
connection closed or reset by the server or a proxy (RemoteDisconnected,
ConnectionResetError), a cut-short body (IncompleteRead), or a body that
is not the expected JSON. The exception ended Workers.send() inside the
thread pool, where nothing observed it, so the batch was dropped without
a retry, a callback or a log line.

These are now reported as HttpStatus.UNKNOWN, which the response
processor retries (events carry an insert_id, so a retry of a request
the server did ingest is deduplicated).
---
 src/amplitude/http_client.py | 10 +++++++
 src/test/test_http_client.py | 52 ++++++++++++++++++++++++++++++++++++
 2 files changed, 62 insertions(+)
 create mode 100644 src/test/test_http_client.py

diff --git a/src/amplitude/http_client.py b/src/amplitude/http_client.py
index 6506216..babd8ad 100644
--- a/src/amplitude/http_client.py
+++ b/src/amplitude/http_client.py
@@ -1,5 +1,6 @@
 import json
 from enum import Enum
+from http.client import HTTPException
 from socket import timeout
 from typing import Optional
 from urllib import request, response, error
@@ -144,4 +145,13 @@ def post(url: str, payload: bytes, header=None) -> Response:
                 result.body = {'error': e.reason}
         except error.URLError as e:
             result.body = {'error': str(e.reason)}
+        except (OSError, HTTPException, ValueError, KeyError) as e:
+            # urlopen() only wraps errors raised while sending the request in
+            # URLError. Reading the response can still fail: the connection was
+            # closed or reset (RemoteDisconnected, ConnectionResetError), the
+            # body was cut short (IncompleteRead) or is not the expected JSON.
+            # Report these as UNKNOWN, which is retried, instead of raising
+            # into the worker thread, where the events were silently dropped.
+            result = Response()
+            result.body = {'error': str(e) or type(e).__name__}
         return result
diff --git a/src/test/test_http_client.py b/src/test/test_http_client.py
new file mode 100644
index 0000000..b797675
--- /dev/null
+++ b/src/test/test_http_client.py
@@ -0,0 +1,52 @@
+import http.client
+import unittest
+from unittest import mock
+
+from amplitude.http_client import HttpClient, HttpStatus
+
+# Other test modules replace HttpClient.post with a mock and do not restore it,
+# so keep a reference to the real implementation.
+http_post = HttpClient.post
+
+
+class FakeResponse:
+    def __init__(self, body):
+        self.body = body
+
+    def read(self):
+        return self.body
+
+
+class HttpClientTestCase(unittest.TestCase):
+    def post_with_urlopen(self, **urlopen_kwargs):
+        with mock.patch("urllib.request.urlopen", **urlopen_kwargs):
+            return http_post("https://api2.amplitude.com/2/httpapi", b"{}")
+
+    def test_post_success(self):
+        res = self.post_with_urlopen(return_value=FakeResponse(b'{"code": 200}'))
+        self.assertEqual(HttpStatus.SUCCESS, res.status)
+        self.assertEqual(200, res.code)
+
+    def test_post_connection_errors_after_request_sent_are_unknown(self):
+        errors = [
+            http.client.RemoteDisconnected("Remote end closed connection without response"),
+            ConnectionResetError(104, "Connection reset by peer"),
+            http.client.IncompleteRead(b"{\"co"),
+        ]
+        for exc in errors:
+            with self.subTest(exc=type(exc).__name__):
+                res = self.post_with_urlopen(side_effect=exc)
+                self.assertEqual(HttpStatus.UNKNOWN, res.status)
+                self.assertEqual(-1, res.code)
+                self.assertTrue(res.error)
+
+    def test_post_unexpected_response_body_is_unknown(self):
+        for body in (b"<html>502 Bad Gateway</html>", b'{"events_ingested": 1}'):
+            with self.subTest(body=body):
+                res = self.post_with_urlopen(return_value=FakeResponse(body))
+                self.assertEqual(HttpStatus.UNKNOWN, res.status)
+                self.assertTrue(res.error)
+
+
+if __name__ == '__main__':
+    unittest.main()