fix: fix "not a valid pdf error" in parallel mode (#186)

This was a subtle bug that came out in the retry logic. When we get a 500 during the requests.post,
we'll try again. However, the pdf was stored in a BytesIO, which had already been read the first
time we sent it. The next request sends an empty file, which results in a 400 response masking the
original error.

Steps to verify:

* First, checkout `main`
* Start up the api in parallel mode

```
export UNSTRUCTURED_PARALLEL_MODE_ENABLED=true
export UNSTRUCTURED_PARALLEL_MODE_URL=http://localhost:8000/general/v0/general
make run-web-app
```

* Insert a 500 error into `prepline_general/api/general.py:partition_pdf_splits()`
```
    # If it's small enough, just process locally
    if len(pdf_pages) <= pages_per_pdf:
        raise HTTPException(status_code=500)  # Throw an error here
        return partition(
            file=file, file_filename=file_filename, content_type=content_type, **partition_kwargs
        )

```

* Send a document and see that the 500 is hidden behind a 400 error
```
$ curl 'http://localhost:8000/general/v0/general' --header 'Accept: application/json' --form files=@sample-docs/layout-parser-paper-fast.pdf

{"detail":"layout-parser-paper-fast.pdf does not appear to be a valid PDF"}%
```

* Switch to this branch and do it again - you should now get a 500 `Internal server error` response

Author: awalker4

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.0.36-dev0
+
+* Fix a bug in parallel mode causing `not a valid pdf` errors
+
 ## 0.0.35
 
 * Bump unstructured library to 0.9.2

pipeline-notebooks/pipeline-general.ipynb

@@ -617,7 +617,7 @@
     "        new_pdf.write(pdf_buffer)\n",
     "        pdf_buffer.seek(0)\n",
     "\n",
-    "        split_pdfs.append((pdf_buffer, offset))\n",
+    "        split_pdfs.append((pdf_buffer.read(), offset))\n",
     "        offset += split_size\n",
     "\n",
     "    return split_pdfs\n",

prepline_general/api/general.py

@@ -90,7 +90,7 @@ def get_pdf_splits(pdf_pages, split_size=1):
         new_pdf.write(pdf_buffer)
         pdf_buffer.seek(0)
 
-        split_pdfs.append((pdf_buffer, offset))
+        split_pdfs.append((pdf_buffer.read(), offset))
         offset += split_size
 
     return split_pdfs
@@ -484,7 +484,7 @@ def return_content_type(filename):
 
 
 @router.post("/general/v0/general")
-@router.post("/general/v0.0.35/general")
+@router.post("/general/v0.0.36/general")
 def pipeline_1(
     request: Request,
     gz_uncompressed_content_type: Optional[str] = Form(default=None),

preprocessing-pipeline-family.yaml

@@ -1,2 +1,2 @@
 name: general
-version: 0.0.35
+version: 0.0.36

test_general/api/test_app.py

@@ -415,7 +415,6 @@ def test_parallel_mode_returns_errors(monkeypatch):
     response = client.post(
         MAIN_API_ROUTE,
         files=[("files", (str(test_file), open(test_file, "rb"), "application/pdf"))],
-        data={"pdf_processing_mode": "parallel"},
     )
 
     assert response.status_code == 500
@@ -432,13 +431,12 @@ def test_parallel_mode_returns_errors(monkeypatch):
     response = client.post(
         MAIN_API_ROUTE,
         files=[("files", (str(test_file), open(test_file, "rb"), "application/pdf"))],
-        data={"pdf_processing_mode": "parallel"},
     )
 
     assert response.status_code == 400
 
 
-def test_partition_file_via_api_retry(monkeypatch, mocker):
+def test_partition_file_via_api_will_retry(monkeypatch, mocker):
     """
     Verify number of retries with parallel mode
     """
@@ -449,23 +447,36 @@ def test_partition_file_via_api_retry(monkeypatch, mocker):
     monkeypatch.setenv("UNSTRUCTURED_PARALLEL_RETRY_ATTEMPTS", "2")
     monkeypatch.setenv("UNSTRUCTURED_PARALLEL_RETRY_BACKOFF_TIME", "0.1")
 
+    num_calls = 0
+
+    # Return a transient error the first time
+    def mock_response(*args, **kwargs):
+        nonlocal num_calls
+        num_calls += 1
+
+        if num_calls == 1:
+            return MockResponse(status_code=500)
+
+        return MockResponse(status_code=200)
+
     monkeypatch.setattr(
         requests,
         "post",
-        lambda *args, **kwargs: MockResponse(status_code=500),
+        mock_response,
     )
-    mock_sleep = mocker.patch("time.sleep")
+
+    # This needs to be mocked when we return 200
+    mocker.patch("prepline_general.api.general.elements_from_json")
+
     client = TestClient(app)
-    test_file = Path("sample-docs") / "layout-parser-paper.pdf"
+    test_file = Path("sample-docs") / "layout-parser-paper-fast.pdf"
 
     response = client.post(
         MAIN_API_ROUTE,
         files=[("files", (str(test_file), open(test_file, "rb"), "application/pdf"))],
-        data={"pdf_processing_mode": "parallel"},
     )
 
-    assert response.status_code == 500
-    assert mock_sleep.call_count == 2
+    assert response.status_code == 200
 
 
 def test_partition_file_via_api_no_retryable_error_code(monkeypatch, mocker):