Retry a busy server in a loop, not recursively; time out citation lists (#134)

On a 503 the document was re-posted by calling _send_pdf (or process_txt)
again from inside the handler, in the try block whose finally closes the
stream: every attempt nested a frame holding an open file. A server that
stayed busy ended in exhausted descriptors (reported as a 400 "failed to
open file" marker) or a RecursionError (a 500 marker), and --skip_errors
then skipped the document for good. Attempts now run in a loop, each one
opening, posting and closing its own stream before the wait.

process_txt also posted without a timeout, so a hung server blocked a
citation-list run indefinitely; it now uses the configured timeout and
reports a read timeout as 408 like PDFs do.

Author: lfoppiano

grobid_client/grobid_client.py

@@ -215,11 +215,10 @@ def _warn_on_consolidation_timeout(self, consolidate_citations: bool) -> None:
                 f"(2-3 minutes is recommended)."
             )
 
-    def _handle_server_busy_retry(self, file_path: str, retry_func: Any, *args: Any, **kwargs: Any) -> Any:
-        """Handle server busy (503) retry logic."""
+    def _wait_for_busy_server(self, file_path: str) -> None:
+        """Pause before re-posting a document the server answered 503 to."""
         self.logger.warning(f"Server busy (503), retrying {file_path} after {self.config['sleep_time']} seconds")
         time.sleep(self.config["sleep_time"])
-        return retry_func(*args, **kwargs)
 
     def _handle_request_error(
             self, file_path: str, error: Exception, error_type: str = "Request"
@@ -1824,89 +1823,77 @@ def _send_pdf(
             start: int = -1,
             end: int = -1
     ) -> Tuple[str, int, Optional[str]]:
-        """Post one document, retrying it from a fresh stream on a 503."""
-        pdf_handle: Optional[BinaryIO] = None
-        try:
-            pdf_handle = open_handle()
+        """Post one document, re-posting it from a fresh stream while the server answers 503.
 
-            files = {
-                "input": (
-                    name,
-                    pdf_handle,
-                    "application/pdf",
-                    {"Expires": "0"},
-                )
-            }
-
-            the_url = self.get_server_url(service)
-
-            # set the GROBID parameters
-            the_data = {}
-            if generate_ids:
-                the_data["generateIDs"] = "1"
-            if consolidate_header:
-                the_data["consolidateHeader"] = "1"
-            if consolidate_citations:
-                the_data["consolidateCitations"] = "1"
-            if include_raw_citations:
-                the_data["includeRawCitations"] = "1"
-            if include_raw_affiliations:
-                the_data["includeRawAffiliations"] = "1"
-            if tei_coordinates:
-                the_data["teiCoordinates"] = self.config["coordinates"]
-            if segment_sentences:
-                the_data["segmentSentences"] = "1"
-            if flavor:
-                the_data["flavor"] = flavor
-            if start and start > 0:
-                the_data["start"] = str(start)
-            if end and end > 0:
-                the_data["end"] = str(end)
-
-            res, status = self.post(
-                url=the_url, files=files, data=the_data, headers={"Accept": "text/plain"},
-                timeout=self.config['timeout']
-            )
+        A loop rather than a recursive call: every attempt opens, posts and
+        closes its own stream before the wait, so a server that stays busy
+        costs nothing but time. (Recursing kept one open file per attempt
+        and ended in exhausted descriptors or a RecursionError, recorded as
+        a bogus error marker.)
+        """
+        the_url = self.get_server_url(service)
 
-            if status == 503:
-                return self._handle_server_busy_retry(
-                    name,
-                    self._send_pdf,
-                    service,
-                    name,
-                    open_handle,
-                    generate_ids,
-                    consolidate_header,
-                    consolidate_citations,
-                    include_raw_citations,
-                    include_raw_affiliations,
-                    tei_coordinates,
-                    segment_sentences,
-                    flavor,
-                    start,
-                    end
+        # set the GROBID parameters
+        the_data = {}
+        if generate_ids:
+            the_data["generateIDs"] = "1"
+        if consolidate_header:
+            the_data["consolidateHeader"] = "1"
+        if consolidate_citations:
+            the_data["consolidateCitations"] = "1"
+        if include_raw_citations:
+            the_data["includeRawCitations"] = "1"
+        if include_raw_affiliations:
+            the_data["includeRawAffiliations"] = "1"
+        if tei_coordinates:
+            the_data["teiCoordinates"] = self.config["coordinates"]
+        if segment_sentences:
+            the_data["segmentSentences"] = "1"
+        if flavor:
+            the_data["flavor"] = flavor
+        if start and start > 0:
+            the_data["start"] = str(start)
+        if end and end > 0:
+            the_data["end"] = str(end)
+
+        while True:
+            pdf_handle: Optional[BinaryIO] = None
+            try:
+                pdf_handle = open_handle()
+                files = {
+                    "input": (
+                        name,
+                        pdf_handle,
+                        "application/pdf",
+                        {"Expires": "0"},
+                    )
+                }
+                res, status = self.post(
+                    url=the_url, files=files, data=the_data, headers={"Accept": "text/plain"},
+                    timeout=self.config['timeout']
                 )
+            # requests' exceptions derive from IOError, so they have to be
+            # caught before it or every timeout/connection error would be
+            # reported as a failure to open the file (status 400).
+            except requests.exceptions.ReadTimeout as e:
+                self.logger.error(f"Request timeout for {name}: {str(e)}")
+                return (name, 408, f"Request timeout: {str(e)}")
+            except requests.exceptions.RequestException as e:
+                return self._handle_request_error(name, e)
+            except IOError as e:
+                self.logger.error(f"Failed to open PDF file {name}: {str(e)}")
+                return (name, 400, f"Failed to open file: {str(e)}")
+            except Exception as e:
+                return self._handle_unexpected_error(name, e)
+            finally:
+                # Always ours to close: the caller's own stream is never kept,
+                # only the bytes read out of it.
+                if pdf_handle:
+                    pdf_handle.close()
 
-            return (name, status, res.text)
-
-        # requests' exceptions derive from IOError, so they have to be caught
-        # before it or every timeout/connection error would be reported as a
-        # failure to open the file (status 400).
-        except requests.exceptions.ReadTimeout as e:
-            self.logger.error(f"Request timeout for {name}: {str(e)}")
-            return (name, 408, f"Request timeout: {str(e)}")
-        except requests.exceptions.RequestException as e:
-            return self._handle_request_error(name, e)
-        except IOError as e:
-            self.logger.error(f"Failed to open PDF file {name}: {str(e)}")
-            return (name, 400, f"Failed to open file: {str(e)}")
-        except Exception as e:
-            return self._handle_unexpected_error(name, e)
-        finally:
-            # Always ours to close: the caller's own stream is never kept, only
-            # the bytes read out of it.
-            if pdf_handle:
-                pdf_handle.close()
+            if status != 503:
+                return (name, status, res.text)
+            self._wait_for_busy_server(name)
 
     def process_documents(
             self,
@@ -2039,31 +2026,23 @@ def process_txt(
             the_data["includeRawCitations"] = "1"
         the_data["citations"] = references
 
-        try:
-            res, status = self.post(
-                url=the_url, data=the_data, headers={"Accept": "application/xml"}
-            )
-
-            if status == 503:
-                return self._handle_server_busy_retry(
-                    txt_file,
-                    self.process_txt,
-                    service,
-                    txt_file,
-                    generate_ids,
-                    consolidate_header,
-                    consolidate_citations,
-                    include_raw_citations,
-                    include_raw_affiliations,
-                    tei_coordinates,
-                    segment_sentences
+        while True:
+            try:
+                res, status = self.post(
+                    url=the_url, data=the_data, headers={"Accept": "application/xml"},
+                    timeout=self.config['timeout']
                 )
-        except requests.exceptions.RequestException as e:
-            return self._handle_request_error(txt_file, e)
-        except Exception as e:
-            return self._handle_unexpected_error(txt_file, e)
+            except requests.exceptions.ReadTimeout as e:
+                self.logger.error(f"Request timeout for {txt_file}: {str(e)}")
+                return (txt_file, 408, f"Request timeout: {str(e)}")
+            except requests.exceptions.RequestException as e:
+                return self._handle_request_error(txt_file, e)
+            except Exception as e:
+                return self._handle_unexpected_error(txt_file, e)
 
-        return (txt_file, status, res.text)
+            if status != 503:
+                return (txt_file, status, res.text)
+            self._wait_for_busy_server(txt_file)
 
 
 def main() -> None:

tests/test_busy_retry.py

@@ -0,0 +1,132 @@
+"""
+A 503 is retried in a loop: each attempt closes its stream before the wait,
+and a server that stays busy for a long time cannot exhaust descriptors or
+the recursion limit.
+"""
+import io
+import os
+import tempfile
+from unittest.mock import Mock, patch
+
+import requests
+
+from grobid_client.grobid_client import GrobidClient
+
+
+def _client():
+    with patch('grobid_client.grobid_client.GrobidClient._test_server_connection'):
+        with patch('grobid_client.grobid_client.GrobidClient._configure_logging'):
+            client = GrobidClient(check_server=False)
+    client.logger = Mock()
+    client.config['sleep_time'] = 0
+    return client
+
+
+def _busy_then_ok(times):
+    """A post() that answers 503 `times` times, then 200."""
+    calls = []
+
+    def post(url, files=None, data=None, headers=None, timeout=None):
+        calls.append(files['input'][1] if files else None)
+        resp = Mock()
+        if len(calls) <= times:
+            resp.text = 'busy'
+            return (resp, 503)
+        resp.text = '<TEI>ok</TEI>'
+        return (resp, 200)
+
+    return post, calls
+
+
+class TestPdfRetry:
+
+    def test_each_attempt_uses_a_fresh_closed_stream(self):
+        client = _client()
+        post, calls = _busy_then_ok(3)
+        with patch.object(GrobidClient, 'post', side_effect=post):
+            with patch('grobid_client.grobid_client.time.sleep') as sleep:
+                name, status, text = client.process_pdf('processFulltextDocument', b'%PDF-bytes')
+        assert (status, text) == (200, '<TEI>ok</TEI>')
+        assert sleep.call_count == 3
+        assert len(calls) == 4
+        assert len({id(h) for h in calls}) == 4          # a new stream per attempt
+        assert all(h.closed for h in calls)              # none left open
+
+    def test_a_file_on_disk_is_reopened_each_time(self):
+        client = _client()
+        post, calls = _busy_then_ok(2)
+        with tempfile.TemporaryDirectory() as d:
+            pdf = os.path.join(d, 'p.pdf')
+            with open(pdf, 'wb') as fh:
+                fh.write(b'%PDF-disk')
+            with patch.object(GrobidClient, 'post', side_effect=post):
+                with patch('grobid_client.grobid_client.time.sleep'):
+                    _, status, _ = client.process_pdf('processFulltextDocument', pdf)
+        assert status == 200
+        assert all(h.closed for h in calls)
+
+    def test_a_long_busy_spell_does_not_recurse(self):
+        client = _client()
+        post, calls = _busy_then_ok(5000)
+        with patch.object(GrobidClient, 'post', side_effect=post):
+            with patch('grobid_client.grobid_client.time.sleep'):
+                _, status, _ = client.process_pdf('processFulltextDocument', b'%PDF')
+        assert status == 200
+        assert len(calls) == 5001
+
+    def test_timeout_during_a_retry_is_reported_as_408(self):
+        client = _client()
+        answers = iter([(Mock(text='busy'), 503), requests.exceptions.ReadTimeout('slow')])
+
+        def post(url, files=None, data=None, headers=None, timeout=None):
+            answer = next(answers)
+            if isinstance(answer, Exception):
+                raise answer
+            return answer
+
+        with patch.object(GrobidClient, 'post', side_effect=post):
+            with patch('grobid_client.grobid_client.time.sleep'):
+                _, status, text = client.process_pdf('processFulltextDocument', b'%PDF')
+        assert status == 408 and 'timeout' in text.lower()
+
+
+class TestTxtRetry:
+
+    def _refs(self, d):
+        path = os.path.join(d, 'refs.txt')
+        with open(path, 'w', encoding='utf8') as fh:
+            fh.write('one\ntwo\n')
+        return path
+
+    def test_503_then_ok(self):
+        client = _client()
+        post, calls = _busy_then_ok(2)
+        with tempfile.TemporaryDirectory() as d:
+            with patch.object(GrobidClient, 'post', side_effect=post):
+                with patch('grobid_client.grobid_client.time.sleep') as sleep:
+                    name, status, text = client.process_txt(
+                        'processCitationList', self._refs(d), False, False, False, False, False, False, False)
+        assert (status, text) == (200, '<TEI>ok</TEI>')
+        assert sleep.call_count == 2 and len(calls) == 3
+
+    def test_post_carries_the_configured_timeout(self):
+        client = _client()
+        client.config['timeout'] = 42
+        seen = {}
+
+        def post(url, files=None, data=None, headers=None, timeout=None):
+            seen['timeout'] = timeout
+            return (Mock(text='<TEI/>'), 200)
+
+        with tempfile.TemporaryDirectory() as d:
+            with patch.object(GrobidClient, 'post', side_effect=post):
+                client.process_txt('processCitationList', self._refs(d), False, False, False, False, False, False, False)
+        assert seen['timeout'] == 42
+
+    def test_timeout_is_reported_as_408(self):
+        client = _client()
+        with tempfile.TemporaryDirectory() as d:
+            with patch.object(GrobidClient, 'post', side_effect=requests.exceptions.ReadTimeout('slow')):
+                _, status, text = client.process_txt(
+                    'processCitationList', self._refs(d), False, False, False, False, False, False, False)
+        assert status == 408 and 'timeout' in text.lower()