feat: improve the performance and add tests

.github/workflows/test.yaml

@@ -0,0 +1,182 @@
+name: Tests
+
+on:
+  push:
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  unit-tests:
+    name: Unit Tests
+    runs-on: blacksmith-4vcpu-ubuntu-2404
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v6
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+
+      - name: Install system dependencies
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y --no-install-recommends \
+            ffmpeg libimage-exiftool-perl libmagic1
+
+      - name: Install Python dependencies
+        run: |
+          pip install -r requirements.txt
+          pip install pytest reportlab
+
+      - name: Run unit tests
+        env:
+          PYTHONPATH: ${{ github.workspace }}
+        run: pytest tests/ -v
+
+  k8s-benchmark:
+    name: PDF Benchmark (Kubernetes)
+    runs-on: ubuntu-latest
+    timeout-minutes: 45
+    permissions:
+      contents: read
+      pull-requests: write
+    env:
+      IMAGE_TAG: ci-${{ github.sha }}
+      NAMESPACE: markitdown-server
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Build image
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          platforms: linux/amd64
+          push: false
+          load: true
+          tags: markitdown-server:${{ env.IMAGE_TAG }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max
+
+      - name: Start minikube
+        uses: medyagh/setup-minikube@v0.0.21
+        with:
+          driver: docker
+          cpus: 4
+          memory: 6000m
+          addons: metrics-server
+
+      - name: Load image into minikube
+        run: minikube image load "markitdown-server:${IMAGE_TAG}"
+
+      - name: Create secret
+        run: |
+          kubectl create namespace "$NAMESPACE" --dry-run=client -o yaml | kubectl apply -f -
+          # PDF conversion never calls the LLM (that path is images only), so a
+          # placeholder key is enough to satisfy the import-time lookup.
+          kubectl create secret generic markitdown-server-secret \
+            -n "$NAMESPACE" \
+            --from-literal=OPENAI_API_KEY=ci-placeholder \
+            --from-literal=ADMIN_API_KEY=ci-test-key \
+            --dry-run=client -o yaml | kubectl apply -f -
+
+      - name: Deploy
+        run: |
+          cd k8s
+          kustomize edit set image \
+            "ghcr.io/sirily11/markitdown-server=markitdown-server:${IMAGE_TAG}"
+          kubectl apply -k .
+          # Drop the HPA for the benchmark. Conversion pegs the CPU, so the HPA
+          # scales to maxReplicas mid-run and then holds there for its 300s
+          # scale-down window. Two replicas behind the service make per-book
+          # timing and peak-memory attribution meaningless, since requests and
+          # the cgroup reads can land on different pods.
+          kubectl delete hpa markitdown-server-hpa -n "$NAMESPACE" --ignore-not-found
+          kubectl scale deployment/markitdown-server -n "$NAMESPACE" --replicas=1
+
+      - name: Wait for rollout
+        run: |
+          kubectl rollout status deployment/markitdown-redis -n "$NAMESPACE" --timeout=300s || true
+          kubectl rollout status deployment/markitdown-server -n "$NAMESPACE" --timeout=300s
+
+      # No port-forward here on purpose: the benchmark restarts the pod between
+      # books, which tears a forward down, so the script manages its own.
+      - name: Benchmark Foundation books
+        env:
+          ADMIN_API_KEY: ci-test-key
+          LOCAL_PORT: "8080"
+        run: |
+          set -o pipefail
+          bash scripts/k8s-benchmark.sh | tee /tmp/benchmark.txt
+
+      - name: Build benchmark report
+        if: always()
+        run: |
+          # On failure the script appends per-book pod logs, so cap the size:
+          # a GitHub comment is rejected outright past ~65k characters.
+          if [ -f /tmp/benchmark.txt ]; then
+            head -c 55000 /tmp/benchmark.txt > /tmp/benchmark-trimmed.txt
+            if [ "$(wc -c < /tmp/benchmark.txt)" -gt 55000 ]; then
+              echo '... output truncated, see the k8s-logs artifact ...' \
+                >> /tmp/benchmark-trimmed.txt
+            fi
+          else
+            echo 'benchmark did not produce output' > /tmp/benchmark-trimmed.txt
+          fi
+          {
+            echo '<!-- markitdown-pdf-benchmark -->'
+            echo '## PDF conversion benchmark'
+            echo
+            echo '```'
+            cat /tmp/benchmark-trimmed.txt
+            echo '```'
+            echo
+            echo "Commit \`${GITHUB_SHA:0:7}\` · [workflow run](${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID})"
+          } > /tmp/benchmark-report.md
+          cat /tmp/benchmark-report.md >> "$GITHUB_STEP_SUMMARY"
+
+      # Posted even when the benchmark fails -- a budget violation is exactly
+      # the result worth surfacing on the PR.
+      - name: Comment benchmark on PR
+        if: always()
+        env:
+          GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        run: |
+          # This workflow runs on push, not pull_request, so the PR number has
+          # to be looked up from the branch.
+          PR=$(gh pr list --head "$GITHUB_REF_NAME" --state open \
+                 --json number --jq '.[0].number // empty')
+          if [ -z "$PR" ]; then
+            echo "No open PR for branch $GITHUB_REF_NAME; skipping comment."
+            exit 0
+          fi
+          # Edit the previous report rather than stacking a new comment on
+          # every push; fall back to creating one on the first run.
+          gh pr comment "$PR" --body-file /tmp/benchmark-report.md --edit-last \
+            || gh pr comment "$PR" --body-file /tmp/benchmark-report.md
+
+      - name: Collect pod diagnostics
+        if: always()
+        run: |
+          mkdir -p /tmp/k8s-logs
+          # Per-book logs captured during the run. The pod is restarted between
+          # books, so these are the only copy of an earlier book's output.
+          cp -r /tmp/benchmark-pod-logs /tmp/k8s-logs/per-book 2>/dev/null || true
+          kubectl logs -n "$NAMESPACE" -l app=markitdown-server --tail=-1 \
+            > /tmp/k8s-logs/server-final.log 2>&1 || true
+          kubectl get pods -n "$NAMESPACE" -o wide > /tmp/k8s-logs/pods.txt 2>&1 || true
+          kubectl describe pods -n "$NAMESPACE" -l app=markitdown-server \
+            > /tmp/k8s-logs/describe.txt 2>&1 || true
+
+      - name: Upload diagnostics
+        uses: actions/upload-artifact@v7
+        if: always()
+        with:
+          name: k8s-logs
+          path: /tmp/k8s-logs/
+          retention-days: 7

converter.py

@@ -9,6 +9,8 @@
 import requests
 import tempfile
 
+from pdf_chunk import convert_pdf_chunked, should_chunk
+
 logger = logging.getLogger(__name__)
 
 client = OpenAI(
@@ -93,9 +95,17 @@ def convert(url: str) -> DocumentConverterResult:
     downloaded_at = time.monotonic()
     logger.info("downloaded %s in %.3fs", url, downloaded_at - started)
 
-    md = MarkItDown(llm_client=client, llm_model=model)
     try:
-        converted = md.convert(temp_file)
+        # Large PDFs are converted chunk-by-chunk across worker processes.
+        # Done whole-document, markitdown holds the parsed object graph for
+        # every page at once, which pins the container memory limit and
+        # serialises all the parsing onto one core.
+        if should_chunk(temp_file):
+            markdown = convert_pdf_chunked(temp_file)
+            converted = DocumentConverterResult(markdown=markdown)
+        else:
+            md = MarkItDown(llm_client=client, llm_model=model)
+            converted = md.convert(temp_file)
     finally:
         if os.path.exists(temp_file):
             os.remove(temp_file)

k8s/deployment.yaml

@@ -24,18 +24,29 @@ spec:
           image: ghcr.io/sirily11/markitdown-server:latest
           resources:
             limits:
-              memory: "1024Mi"
-              cpu: "500m"
+              # PDF conversion is CPU-bound and now fans chunks out across
+              # worker processes; the limit has to exceed PDF_MAX_WORKERS
+              # cores or the workers just contend for the same slice.
+              memory: "2048Mi"
+              cpu: "2000m"
             requests:
               # Baseline RSS is ~120-140Mi; a 128Mi request pinned memory
               # utilization at ~100% and kept the HPA at max replicas.
-              memory: "384Mi"
-              cpu: "250m"
+              memory: "512Mi"
+              cpu: "500m"
           env:
             - name: REDIS_URL
               value: "redis://markitdown-redis:6379/0"
             - name: OPENAI_MODEL
               value: "google/gemini-2.5-flash-preview-05-20"
+            # Keep in step with the CPU limit above: each worker saturates
+            # roughly one core, and each costs ~150-200Mi of resident memory
+            # for its own markitdown import, so this trades against the
+            # memory limit too.
+            - name: PDF_MAX_WORKERS
+              value: "2"
+            - name: PDF_PAGES_PER_CHUNK
+              value: "20"
           envFrom:
             - secretRef:
                 name: markitdown-server-secret

k8s/ingress.yaml

@@ -6,6 +6,11 @@ metadata:
   annotations:
     cert-manager.io/cluster-issuer: letsencrypt-prod
     nginx.ingress.kubernetes.io/proxy-body-size: "1000m"
+    # Large PDFs take several minutes to parse; nginx's 60s default cut the
+    # connection mid-conversion and returned a 504 to the caller.
+    nginx.ingress.kubernetes.io/proxy-connect-timeout: "60"
+    nginx.ingress.kubernetes.io/proxy-send-timeout: "600"
+    nginx.ingress.kubernetes.io/proxy-read-timeout: "600"
     nginx.ingress.kubernetes.io/affinity: "cookie"
     nginx.ingress.kubernetes.io/affinity-mode: "persistent"
     nginx.ingress.kubernetes.io/session-cookie-name: "mcp-router-affinity"

pdf_chunk.py

@@ -0,0 +1,134 @@
+"""
+Page-chunked, multi-process PDF conversion.
+
+markitdown's ``PdfConverter`` holds the parsed pdfplumber object graph for the
+*whole* document at once (and, for prose PDFs, then re-parses everything with
+pdfminer), so both peak memory and wall-clock grow with document size in a
+single process.
+
+Splitting the PDF into page ranges and converting each range in its own process
+bounds both: a worker only ever holds N pages of parsed objects, and it returns
+that memory to the OS when it exits.
+
+Both pdfplumber and pdfminer.six are pure Python, so this has to use processes —
+threads would serialise on the GIL and buy nothing.
+"""
+import logging
+import os
+import shutil
+import tempfile
+from concurrent.futures import ProcessPoolExecutor
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+# Pages per chunk. Smaller chunks lower peak memory per worker but add
+# per-chunk parser startup overhead.
+PAGES_PER_CHUNK = int(os.environ.get("PDF_PAGES_PER_CHUNK", "20"))
+# Worker processes. Each holds ~140-190 MiB just for the markitdown import, so
+# this must be sized against the container memory limit, not just its CPU.
+MAX_WORKERS = int(os.environ.get("PDF_MAX_WORKERS", "2"))
+# Below this page count the process startup cost outweighs the parallelism.
+MIN_PAGES_FOR_PARALLEL = int(os.environ.get("PDF_MIN_PAGES_FOR_PARALLEL", "40"))
+
+
+def page_count(path: str) -> Optional[int]:
+    """
+    Return the number of pages in ``path``, or None if it is not a readable PDF.
+
+    Used both to decide whether chunking is worthwhile and as the PDF sniff, so
+    a corrupt or non-PDF file transparently falls back to the normal path.
+    """
+    try:
+        from pypdf import PdfReader
+
+        return len(PdfReader(path).pages)
+    except Exception:
+        return None
+
+
+def split_pdf(path: str, pages_per_chunk: int, out_dir: str) -> list[str]:
+    """
+    Split ``path`` into single-file PDFs of ``pages_per_chunk`` pages each.
+
+    Chunks are written to ``out_dir`` and passed to workers by path rather than
+    by value — pickling the bytes through the process pool would duplicate the
+    document in memory, which is the thing this module exists to avoid.
+    """
+    from pypdf import PdfReader, PdfWriter
+
+    if pages_per_chunk < 1:
+        raise ValueError("pages_per_chunk must be >= 1")
+
+    reader = PdfReader(path)
+    total = len(reader.pages)
+    chunk_paths: list[str] = []
+
+    for index, start in enumerate(range(0, total, pages_per_chunk)):
+        writer = PdfWriter()
+        for page in reader.pages[start:start + pages_per_chunk]:
+            writer.add_page(page)
+        chunk_path = os.path.join(out_dir, f"chunk_{index:05d}.pdf")
+        with open(chunk_path, "wb") as fh:
+            writer.write(fh)
+        chunk_paths.append(chunk_path)
+
+    return chunk_paths
+
+
+def convert_chunk(chunk_path: str) -> str:
+    """
+    Convert one chunk file to markdown. Runs in a worker process.
+
+    Must stay importable at module level so the pool can pickle it. markitdown
+    is imported lazily here so the parent does not pay for it when the chunked
+    path is never taken.
+    """
+    from markitdown._stream_info import StreamInfo
+    from markitdown.converters._pdf_converter import PdfConverter
+
+    stream_info = StreamInfo(extension=".pdf", mimetype="application/pdf")
+    with open(chunk_path, "rb") as fh:
+        return PdfConverter().convert(fh, stream_info).markdown
+
+
+def convert_pdf_chunked(
+    path: str,
+    pages_per_chunk: int = PAGES_PER_CHUNK,
+    max_workers: int = MAX_WORKERS,
+) -> str:
+    """
+    Convert a PDF by splitting it into page ranges converted in parallel.
+
+    Chunk results are concatenated in page order, so output ordering matches a
+    whole-document conversion.
+    """
+    work_dir = tempfile.mkdtemp(prefix="pdf_chunks_")
+    try:
+        chunk_paths = split_pdf(path, pages_per_chunk, work_dir)
+        logger.info("split %s into %d chunks of <=%d pages",
+                    path, len(chunk_paths), pages_per_chunk)
+
+        if len(chunk_paths) == 1:
+            return convert_chunk(chunk_paths[0])
+
+        # "spawn" rather than the Linux default "fork": this runs inside a
+        # FastAPI threadpool worker, and forking a process that holds threads
+        # can deadlock if a lock is held at fork time.
+        import multiprocessing
+
+        context = multiprocessing.get_context("spawn")
+        workers = max(1, min(max_workers, len(chunk_paths)))
+        with ProcessPoolExecutor(max_workers=workers, mp_context=context) as pool:
+            # map() preserves input order, so pages stay sequential.
+            parts = list(pool.map(convert_chunk, chunk_paths))
+
+        return "\n\n".join(part for part in parts if part and part.strip())
+    finally:
+        shutil.rmtree(work_dir, ignore_errors=True)
+
+
+def should_chunk(path: str, min_pages: int = MIN_PAGES_FOR_PARALLEL) -> bool:
+    """True if ``path`` is a PDF large enough to be worth chunking."""
+    pages = page_count(path)
+    return pages is not None and pages >= min_pages