fix: deployment use prod env

.github/workflows/test.yaml

@@ -89,6 +89,10 @@ jobs:
             -n "$NAMESPACE" \
             --from-literal=OPENAI_API_KEY=ci-placeholder \
             --from-literal=ADMIN_API_KEY=ci-test-key \
+            --from-literal=TURSO_DATABASE_URL='http://markitdown-libsql:8080' \
+            --from-literal=S3_ENDPOINT='http://markitdown-minio:9000' \
+            --from-literal=S3_BUCKET=markitdown \
+            --from-literal=S3_REGION=us-east-1 \
             --from-literal=S3_ACCESS_KEY_ID=ci-minio-user \
             --from-literal=S3_SECRET_ACCESS_KEY=ci-minio-password \
             --from-literal=RABBITMQ_USER=ci-rabbit-user \
@@ -101,6 +105,9 @@ jobs:
           cd k8s
           kustomize edit set image \
             "ghcr.io/sirily11/markitdown-server=markitdown-server:${IMAGE_TAG}"
+          # These stateful dependencies belong only to this disposable CI
+          # cluster. The release kustomization intentionally excludes them.
+          kubectl apply -f rabbitmq.yaml -f libsql.yaml -f minio.yaml
           kubectl apply -k .
           # Drop the worker HPA and pin the worker to one replica. Conversion
           # pegs the CPU, so the HPA would scale up mid-run and hold there for
@@ -114,6 +121,7 @@ jobs:
         run: |
           # State first: the API and worker both fail readiness until these are
           # up, so rolling them out in order keeps the failure legible.
+          kubectl rollout status deployment/markitdown-rabbitmq -n "$NAMESPACE" --timeout=300s
           kubectl rollout status deployment/markitdown-libsql -n "$NAMESPACE" --timeout=300s
           kubectl rollout status deployment/markitdown-minio -n "$NAMESPACE" --timeout=300s
           kubectl wait --for=condition=complete job/markitdown-minio-setup \

README.md

@@ -121,11 +121,11 @@ your real secrets. `make reset` wipes the volumes.
 
 ## Deployment
 
-`k8s/` is a kustomize stack: an API deployment, a worker deployment (HPA'd on
-CPU), plus self-hosted libSQL and minio for state. To run fully managed, drop
-`libsql.yaml` and `minio.yaml` from `kustomization.yaml` and point the secret at
-hosted Turso and R2/S3 — note that the inline `env` entries in the deployments
-take precedence over `envFrom`, so those must be removed too.
+`k8s/` is a release kustomize stack containing the API, worker, HPAs and
+ingress. Their RabbitMQ, Turso and S3 connection settings come from
+`markitdown-server-secret`; release deployment does not provision those shared
+services. The Kubernetes benchmark explicitly deploys the bundled RabbitMQ,
+libSQL and MinIO manifests only inside its disposable CI minikube cluster.
 
 `scripts/k8s-benchmark.sh` converts four real Asimov PDFs against a cluster and
 enforces per-book time and memory budgets; it is the performance regression

k8s/api-deployment.yaml

@@ -33,32 +33,18 @@ spec:
               memory: "256Mi"
               cpu: "100m"
           env:
-            # NOTE: `env` takes precedence over `envFrom`, so these in-cluster
-            # defaults win over anything of the same name in the secret. To run
-            # against hosted Turso or R2, delete the relevant entries and supply
-            # them via the secret instead — changing only the secret will not
-            # take effect while they are here.
-            # RABBITMQ_URL is intentionally absent here: it arrives whole from
-            # the secret via envFrom, so the broker (in-cluster rabbitmq.yaml
-            # vs a shared one in another namespace) is a secret-only change.
-            # It used to be pinned here only because it was assembled from
-            # $(RABBITMQ_USER) — and $(VAR) expands solely from earlier entries
-            # in this same list, never from envFrom.
-            - name: TURSO_DATABASE_URL
-              value: "http://markitdown-libsql:8080"
-            - name: S3_ENDPOINT
-              value: "http://markitdown-minio:9000"
-            - name: S3_BUCKET
-              value: "markitdown"
+            # Database, object-storage and RabbitMQ connection settings all
+            # arrive from the Secret via envFrom. CI creates that Secret with
+            # disposable in-cluster endpoints; releases use real shared
+            # services without pod-spec overrides taking precedence.
             - name: OPENAI_MODEL
               value: "google/gemini-3.1-flash-lite-preview"
             - name: PDF_PAGES_PER_CHUNK
               value: "20"
             - name: PDF_MIN_PAGES_FOR_PARALLEL
               value: "40"
           envFrom:
-            # ADMIN_API_KEY, OPENAI_API_KEY, S3_ACCESS_KEY_ID,
-            # S3_SECRET_ACCESS_KEY (and TURSO_AUTH_TOKEN when managed).
+            # API keys plus database, object-storage and RabbitMQ settings.
             - secretRef:
                 name: markitdown-server-secret
           livenessProbe:

k8s/kustomization.yaml

@@ -5,12 +5,9 @@ namespace: markitdown-server
 
 resources:
   - namespace.yaml
-  # Self-hosted infrastructure. Drop libsql/minio and repoint the secret at
-  # hosted Turso and R2/S3 to run those managed; the application is identical
-  # either way.
-  - rabbitmq.yaml
-  - libsql.yaml
-  - minio.yaml
+  # Application resources only. The CI benchmark applies rabbitmq.yaml,
+  # libsql.yaml and minio.yaml explicitly to its disposable minikube cluster;
+  # release deployments use the shared services configured in the Secret.
   - api-deployment.yaml
   - worker-deployment.yaml
   - hpa.yaml

k8s/libsql.yaml

@@ -1,10 +1,9 @@
-# Self-hosted libSQL server (sqld).
+# CI-only self-hosted libSQL server (sqld).
 #
 # The API and worker pods must share one database — a SQLite file on a pod's
 # local disk cannot be, which is why this exists rather than a PVC. For a
-# managed setup, delete this file from kustomization.yaml and point
-# TURSO_DATABASE_URL / TURSO_AUTH_TOKEN at hosted Turso instead; nothing in the
-# application changes.
+# Release deployments use the hosted Turso database configured in their Secret.
+# test.yaml applies this manifest only to its disposable minikube.
 #
 # Unlike the Redis it replaces, this is a durable store, not a cache: it holds
 # the queue and every job's state. Hence a PVC and Recreate rather than an

k8s/minio.yaml

@@ -1,8 +1,8 @@
-# In-cluster S3-compatible object storage.
+# CI-only in-cluster S3-compatible object storage.
 #
 # Holds source files, per-chunk markdown and the final paginated pages. For a
-# managed setup, delete this from kustomization.yaml and point the S3_* secret
-# values at R2 or AWS instead — the application only speaks the S3 API.
+# Release deployments use the real S3-compatible service configured in their
+# Secret. test.yaml applies this manifest only to its disposable minikube.
 #
 # Object expiry is a bucket lifecycle rule (see the bucket-setup Job below),
 # not something the application enforces.

k8s/rabbitmq.yaml

@@ -1,4 +1,8 @@
-# RabbitMQ — job distribution.
+# CI-only RabbitMQ — job distribution.
+#
+# Release deployments use the shared cluster broker configured by RABBITMQ_URL
+# in their Secret. test.yaml applies this manifest only to its disposable
+# minikube.
 #
 # The API publishes one message per chunk; workers consume with prefetch=1, so
 # the broker hands each chunk to whichever worker is free. Adding a worker pod

k8s/secrets.example.yaml

@@ -5,9 +5,9 @@
 # Applied on its own rather than through kustomization.yaml, so that `kubectl
 # apply -k k8s` never carries credentials.
 #
-# Every manifest reads from this one secret. The keys under "required" are
-# referenced by secretKeyRef without `optional: true`, so a missing one leaves
-# the pod in CreateContainerConfigError instead of falling back to a default.
+# The API and worker import this whole Secret with envFrom. Fill every required
+# key so a release cannot silently fall back to a pod-local database or the AWS
+# default endpoint.
 apiVersion: v1
 kind: Secret
 metadata:
@@ -21,25 +21,20 @@ stringData:
   OPENAI_API_KEY: CHANGEME
   # The x-api-key callers present to the API.
   ADMIN_API_KEY: CHANGEME
-  # Consumed by rabbitmq.yaml (as RABBITMQ_DEFAULT_USER/PASS) and by the api
-  # and worker deployments, which interpolate them into RABBITMQ_URL.
-  RABBITMQ_USER: CHANGEME
-  RABBITMQ_PASSWORD: CHANGEME
-  # Both the credentials minio.yaml boots with and the ones the app
-  # authenticates using — they are the same pair on purpose.
+  # The application reads the complete shared-broker connection string.
+  RABBITMQ_URL: amqp://<user>:<password>@<host>:5672/%2F
+  # Credentials for the real S3-compatible object store.
   S3_ACCESS_KEY_ID: CHANGEME
   S3_SECRET_ACCESS_KEY: CHANGEME
+  S3_ENDPOINT: https://<account>.r2.cloudflarestorage.com
+  S3_BUCKET: markitdown
+  S3_REGION: auto
+  # Hosted Turso database shared by the API and worker deployments.
+  TURSO_DATABASE_URL: libsql://<db>-<org>.turso.io
+  TURSO_AUTH_TOKEN: CHANGEME
 
-  # ---- Managed Turso / R2 / S3 (optional) ----
-  # The default stack runs libsql and minio in-cluster, addressed by
-  # TURSO_DATABASE_URL and S3_ENDPOINT under `env:` in api-deployment.yaml and
-  # worker-deployment.yaml.
-  #
-  # `env:` takes precedence over `envFrom:`, so uncommenting these alone does
-  # NOTHING — delete the matching entries from both deployments first, and drop
-  # libsql.yaml / minio.yaml from kustomization.yaml.
-  # TURSO_DATABASE_URL: libsql://<db>-<org>.turso.io
-  # TURSO_AUTH_TOKEN: CHANGEME
-  # S3_ENDPOINT: https://<account>.r2.cloudflarestorage.com
-  # S3_BUCKET: markitdown
-  # S3_REGION: auto
+  # ---- CI-only infrastructure ----
+  # test.yaml creates its own Secret and uses these only to boot its disposable
+  # in-cluster RabbitMQ. Release deployments do not need these keys.
+  # RABBITMQ_USER: CHANGEME
+  # RABBITMQ_PASSWORD: CHANGEME

k8s/worker-deployment.yaml

@@ -34,23 +34,10 @@ spec:
               memory: "512Mi"
               cpu: "500m"
           env:
-            # NOTE: `env` takes precedence over `envFrom`, so these in-cluster
-            # defaults win over anything of the same name in the secret. To run
-            # against hosted Turso or R2, delete the relevant entries and supply
-            # them via the secret instead — changing only the secret will not
-            # take effect while they are here.
-            # RABBITMQ_URL is intentionally absent here: it arrives whole from
-            # the secret via envFrom, so the broker (in-cluster rabbitmq.yaml
-            # vs a shared one in another namespace) is a secret-only change.
-            # It used to be pinned here only because it was assembled from
-            # $(RABBITMQ_USER) — and $(VAR) expands solely from earlier entries
-            # in this same list, never from envFrom.
-            - name: TURSO_DATABASE_URL
-              value: "http://markitdown-libsql:8080"
-            - name: S3_ENDPOINT
-              value: "http://markitdown-minio:9000"
-            - name: S3_BUCKET
-              value: "markitdown"
+            # Database, object-storage and RabbitMQ connection settings all
+            # arrive from the Secret via envFrom. CI creates that Secret with
+            # disposable in-cluster endpoints; releases use real shared
+            # services without pod-spec overrides taking precedence.
             - name: OPENAI_MODEL
               value: "google/gemini-3.1-flash-lite-preview"
             - name: PDF_PAGES_PER_CHUNK

taskqueue.py

@@ -159,41 +159,50 @@ def declare(ch) -> None:
 )
 
 
+def _publish(exchange: str, routing_key: str, bodies: list[bytes]) -> None:
+    """
+    Publish bodies, reconnecting once if the cached connection has gone stale.
+
+    The producer only touches its connection when a request arrives, so a
+    BlockingConnection sitting idle between submissions never gets to send a
+    heartbeat and the broker eventually closes it. ``is_open`` still reports
+    True until the socket error surfaces, so the staleness can only be
+    discovered by publishing — hence retry rather than a pre-flight check.
+    """
+    try:
+        ch = channel()
+        for body in bodies:
+            ch.basic_publish(exchange=exchange, routing_key=routing_key,
+                             body=body, properties=_PERSISTENT)
+        return
+    except pika.exceptions.AMQPError:
+        logger.warning("publish failed, reconnecting to rabbitmq", exc_info=True)
+
+    close()
+    ch = channel()
+    for body in bodies:
+        ch.basic_publish(exchange=exchange, routing_key=routing_key,
+                         body=body, properties=_PERSISTENT)
+
+
 def publish(tasks: list[ChunkTask]) -> None:
     """Publish chunk tasks onto the work queue."""
-    ch = channel()
-    for task in tasks:
-        ch.basic_publish(
-            exchange=config.RABBITMQ_EXCHANGE,
-            routing_key=config.RABBITMQ_QUEUE,
-            body=task.to_bytes(),
-            properties=_PERSISTENT,
-        )
+    _publish(config.RABBITMQ_EXCHANGE, config.RABBITMQ_QUEUE,
+             [task.to_bytes() for task in tasks])
 
 
 def publish_retry(task: ChunkTask) -> None:
     """Schedule a failed chunk for another attempt after its backoff delay."""
-    ch = channel()
     queue = retry_queue_name(task.attempt)
-    ch.basic_publish(
-        exchange=RETRY_EXCHANGE,
-        routing_key=queue,
-        body=task.next_attempt().to_bytes(),
-        properties=_PERSISTENT,
-    )
+    _publish(RETRY_EXCHANGE, queue, [task.next_attempt().to_bytes()])
     logger.info("chunk %d of job %s retrying via %s",
                 task.chunk_index, task.job_id, queue)
 
 
 def publish_failed(task: ChunkTask, error: str) -> None:
     """Park a chunk that exhausted its attempts on the failed queue."""
-    ch = channel()
-    ch.basic_publish(
-        exchange=DLX_EXCHANGE,
-        routing_key=config.RABBITMQ_QUEUE,
-        body=json.dumps({**asdict(task), "error": error[:2000]}).encode(),
-        properties=_PERSISTENT,
-    )
+    _publish(DLX_EXCHANGE, config.RABBITMQ_QUEUE,
+             [json.dumps({**asdict(task), "error": error[:2000]}).encode()])
 
 
 def consume(handler: Callable[[ChunkTask], None], should_stop: Callable[[], bool]) -> None:

tests/test_taskqueue.py

@@ -5,6 +5,9 @@
 declaration arguments, since a wrong dead-letter setting silently sends retries
 to the wrong place and only shows up under load.
 """
+import pika
+import pytest
+
 import config
 import taskqueue
 from taskqueue import ChunkTask
@@ -143,3 +146,90 @@ def test_declaration_is_idempotent(self, monkeypatch):
         before = (len(ch.queues), len(ch.exchanges))
         taskqueue.declare(ch)
         assert (len(ch.queues), len(ch.exchanges)) == before
+
+
+class PublishingChannel:
+    """A channel that fails its first N publishes with a lost stream."""
+
+    def __init__(self, failures=0):
+        self.failures = failures
+        self.published = []
+
+    def basic_publish(self, exchange, routing_key, body, properties=None):
+        if self.failures > 0:
+            self.failures -= 1
+            raise pika.exceptions.StreamLostError("Stream connection lost")
+        self.published.append((exchange, routing_key, body))
+
+
+class TestStaleConnection:
+    """
+    The producer only touches its connection when a request arrives, so an idle
+    BlockingConnection misses heartbeats and the broker closes it. The next
+    publish then dies mid-request and the caller sees a 400 for a file that is
+    perfectly convertible — which is exactly what a long-running job between
+    two submissions caused in CI.
+    """
+
+    def channels(self, monkeypatch, first_failures):
+        """Hand out a broken channel first, then a healthy one."""
+        made = [PublishingChannel(failures=first_failures), PublishingChannel()]
+        handed = []
+
+        def fake_channel():
+            ch = made[min(len(handed), len(made) - 1)]
+            handed.append(ch)
+            return ch
+
+        monkeypatch.setattr(taskqueue, "channel", fake_channel)
+        monkeypatch.setattr(taskqueue, "close", lambda: handed.append(None))
+        return made
+
+    def test_publish_survives_a_dropped_connection(self, monkeypatch):
+        stale, fresh = self.channels(monkeypatch, first_failures=1)
+        taskqueue.publish([make_task()])
+        assert stale.published == []
+        assert len(fresh.published) == 1
+
+    def test_reconnect_drops_the_stale_connection_first(self, monkeypatch):
+        """Without close(), connect() returns the same dead connection and the
+        retry fails identically."""
+        closed = []
+        monkeypatch.setattr(taskqueue, "close", lambda: closed.append(True))
+        made = [PublishingChannel(failures=1), PublishingChannel()]
+        monkeypatch.setattr(taskqueue, "channel", lambda: made.pop(0) if len(made) > 1 else made[0])
+        taskqueue.publish([make_task()])
+        assert closed == [True]
+
+    def test_every_task_reaches_the_broker_after_a_reconnect(self, monkeypatch):
+        _, fresh = self.channels(monkeypatch, first_failures=1)
+        taskqueue.publish([make_task(chunk_index=i) for i in range(4)])
+        assert len(fresh.published) == 4
+
+    def test_a_second_failure_propagates(self, monkeypatch):
+        """One retry, not an infinite loop: a genuinely down broker must still
+        surface as an error rather than hang the request."""
+        ch = PublishingChannel(failures=99)
+        monkeypatch.setattr(taskqueue, "channel", lambda: ch)
+        monkeypatch.setattr(taskqueue, "close", lambda: None)
+        with pytest.raises(pika.exceptions.AMQPError):
+            taskqueue.publish([make_task()])
+
+    def test_retry_publish_reconnects_too(self, monkeypatch):
+        _, fresh = self.channels(monkeypatch, first_failures=1)
+        taskqueue.publish_retry(make_task())
+        assert len(fresh.published) == 1
+
+    def test_failed_publish_reconnects_too(self, monkeypatch):
+        _, fresh = self.channels(monkeypatch, first_failures=1)
+        taskqueue.publish_failed(make_task(), "boom")
+        assert len(fresh.published) == 1
+
+    def test_a_healthy_connection_is_not_reopened(self, monkeypatch):
+        closed = []
+        monkeypatch.setattr(taskqueue, "close", lambda: closed.append(True))
+        ch = PublishingChannel()
+        monkeypatch.setattr(taskqueue, "channel", lambda: ch)
+        taskqueue.publish([make_task()])
+        assert closed == []
+        assert len(ch.published) == 1