fix(logger): Do not retry on 413s (#427)

fixes #415

supercedes
#416

---------

Co-authored-by: david.lustig <david.lustig@klaviyo.com>

Author: AbhiPrasad

py/src/braintrust/logger.py

@@ -1384,29 +1384,30 @@ def _submit_logs_request(self, items: Sequence[LogItemWithMeta], max_request_siz
                 if overflow_rows:
                     self._overflow_upload_count += 1
                 return
-            if error is None and resp is not None:
-                resp_errmsg = f"{resp.status_code}: {resp.text}"
-            else:
-                resp_errmsg = str(error)
+            has_response = error is None and resp is not None
+            is_413 = has_response and resp.status_code == 413
+            resp_errmsg = f"{resp.status_code}: {resp.text}" if has_response else str(error)
 
-            is_retrying = i + 1 < self.num_tries
-            retrying_text = "" if is_retrying else " Retrying"
-            errmsg = f"log request failed. Elapsed time: {time.time() - start_time} seconds. Payload size: {payload_bytes}.{retrying_text} Error: {resp_errmsg}"
+            should_retry = i + 1 < self.num_tries and not is_413
 
-            if not is_retrying and self.failed_publish_payloads_dir:
+            if not should_retry and self.failed_publish_payloads_dir:
                 _HTTPBackgroundLogger._write_payload_to_dir(
                     payload_dir=self.failed_publish_payloads_dir, payload=dataStr
                 )
                 self._log_failed_payloads_dir()
 
-            if not is_retrying and self.sync_flush:
+            retrying_text = " Retrying" if should_retry else ""
+            errmsg = f"log request failed. Elapsed time: {time.time() - start_time} seconds. Payload size: {payload_bytes}.{retrying_text} Error: {resp_errmsg}"
+            if not should_retry and self.sync_flush:
                 raise Exception(errmsg)
-            else:
-                print(errmsg, file=self.outfile)
-                if is_retrying:
-                    sleep_time_s = BACKGROUND_LOGGER_BASE_SLEEP_TIME_S * (2**i)
-                    print(f"Sleeping for {sleep_time_s}s", file=self.outfile)
-                    time.sleep(sleep_time_s)
+            print(errmsg, file=self.outfile)
+
+            if is_413:
+                return
+            if should_retry:
+                sleep_time_s = BACKGROUND_LOGGER_BASE_SLEEP_TIME_S * (2**i)
+                print(f"Sleeping for {sleep_time_s}s", file=self.outfile)
+                time.sleep(sleep_time_s)
 
         print(f"log request failed after {self.num_tries} retries. Dropping batch", file=self.outfile)
 

py/src/braintrust/test_logger.py

@@ -187,6 +187,60 @@ def test_init_with_saved_parameters_attaches_reference(self):
         assert payload["parameters_version"] == "v1"
 
 
+class TestHTTPBackgroundLoggerLogs3(TestCase):
+    def test_submit_logs_request_413_skips_retries(self) -> None:
+        """Any 413 while publishing ``/logs3`` cannot succeed on retry with the same payload.
+
+        ``sync_flush`` controls whether the terminal failure raises instead of printing.
+        """
+        from braintrust.logger import (
+            LogItemWithMeta,
+            Logs3OverflowInputRow,
+            _HTTPBackgroundLogger,
+        )
+
+        item = LogItemWithMeta(
+            str_value="{}",
+            overflow_meta=Logs3OverflowInputRow(
+                object_ids={},
+                has_comment=False,
+                is_delete=False,
+                byte_size=2,
+            ),
+        )
+        max_result = {"max_request_size": 10**9, "can_use_overflow": True}
+
+        for response_text in ("Request Too Long", "", "Payload Too Large"):
+            for sync_flush in (False, True):
+                with self.subTest(response_text=response_text, sync_flush=sync_flush):
+                    mock_resp = MagicMock()
+                    mock_resp.ok = False
+                    mock_resp.status_code = 413
+                    mock_resp.text = response_text
+
+                    mock_conn = MagicMock()
+                    mock_conn.post.return_value = mock_resp
+
+                    bg = _HTTPBackgroundLogger(LazyValue(lambda: mock_conn, use_mutex=False))
+                    bg.num_tries = 5
+                    bg.sync_flush = sync_flush
+                    bg.failed_publish_payloads_dir = "/tmp/failed-payloads"
+
+                    with patch.object(_HTTPBackgroundLogger, "_write_payload_to_dir") as mock_write_payload:
+                        if sync_flush:
+                            with self.assertRaises(Exception) as cm:
+                                bg._submit_logs_request([item], max_result)
+                            self.assertIn("413", str(cm.exception))
+                        else:
+                            bg._submit_logs_request([item], max_result)
+
+                    self.assertEqual(mock_conn.post.call_count, 1)
+                    mock_write_payload.assert_called_once()
+                    self.assertEqual(
+                        mock_write_payload.call_args.kwargs["payload_dir"], bg.failed_publish_payloads_dir
+                    )
+
+
 class TestLogger(TestCase):
     def test_load_prompt_prefers_version_over_environment_for_project_slug(self):
         mock_api_conn = MagicMock()