fix(jobs): initialize the queue table inside worker processes

Every worker process died immediately on startup with KeyError: 'jobs',
so no job was ever leased. The blocking process_file endpoint then polled
a job nobody would run until its 1800s timeout, which is what hung the
docling integration job for the full 300s client timeout.

SqlStore learns a table's schema only by declaring it via create_table();
the server does that at startup, but a spawned worker is a fresh process
with empty SQLAlchemy metadata, so its first fetch_all() raised on the
table name. The worker now calls queue.initialize() before entering the
lease loop. create_table() is idempotent, so this attaches to the
server's existing table rather than recreating it.

This was invisible because the crash handler logged exc_info=True, which
the structlog logger renders as a literal field rather than a traceback,
reducing a dead worker pool to one context-free line. The handler now
formats the traceback into the message, since the parent process only
ever sees an exit code.

Verified end-to-end against a real spawned worker with
file_processors=inline::pypdf: the deprecated blocking endpoint returns
chunks, and the async submit/poll/list endpoints reach completed.

Signed-off-by: Charlie Doern <cdoern@redhat.com>

Author: cdoern

src/ogx/core/jobs/worker.py

@@ -20,6 +20,7 @@
 import multiprocessing as mp
 import os
 import socket
+import traceback
 from multiprocessing.synchronize import Event as EventType
 
 from pydantic import BaseModel, Field
@@ -103,6 +104,10 @@ async def _run_worker(config: WorkerConfig, stop_event: EventType) -> None:
     register_sqlstore_backends(config.backends)
     store = await get_system_sqlstore(SqlStoreReference(backend=config.jobs_backend, table_name=config.jobs_table))
     queue = JobQueue(store, config.jobs_table, config.lease_ttl_seconds)
+    # The table already exists (the server created it), but SqlStore only learns a
+    # table's schema by declaring it, and this is a fresh process with empty
+    # metadata. Without this every query raises KeyError on the table name.
+    await queue.initialize()
 
     impls: dict[str, object] = {}
     for descriptor in config.descriptors:
@@ -129,8 +134,10 @@ def _worker_main(config: WorkerConfig, stop_event: EventType) -> None:
     """Process entrypoint (must be module-level for the spawn start method)."""
     try:
         asyncio.run(_run_worker(config, stop_event))
-    except Exception:
-        logger.error("Worker process crashed", exc_info=True)
+    except Exception as e:
+        # The traceback is formatted into the message because this is the only
+        # record of a child's death: the parent sees an exit code, not the error.
+        logger.error("Worker process crashed", error=str(e), traceback=traceback.format_exc())
 
 
 class WorkerPool:

tests/unit/core/jobs/test_queue.py

@@ -8,7 +8,11 @@
 
 import time
 
+import pytest
+
 from ogx.core.jobs.queue import JobQueue
+from ogx.core.storage.datatypes import SqliteSqlStoreConfig
+from ogx.core.storage.sqlstore.sqlalchemy_sqlstore import SqlAlchemySqlStoreImpl
 from ogx_api.common.job_types import JobStatus
 
 
@@ -150,3 +154,32 @@ async def test_list_filters_by_api(queue: JobQueue):
 
     listed = await queue.list(api="file_processors")
     assert {r.job_id for r in listed} == {first.job_id, second.job_id}
+
+
+async def test_second_process_must_initialize_before_querying(tmp_path):
+    """A worker attaches to the server's existing table with its own empty metadata.
+
+    SqlStore learns a table's schema only by declaring it, so a fresh process must
+    call initialize() before any query — otherwise every operation raises KeyError
+    on the table name. This is what the worker loop relies on at startup.
+    """
+    db_path = str(tmp_path / "shared-jobs.db")
+
+    server_store = SqlAlchemySqlStoreImpl(SqliteSqlStoreConfig(db_path=db_path))
+    server_queue = JobQueue(server_store, table_name="jobs", lease_ttl_seconds=60)
+    await server_queue.initialize()
+    record = await _enqueue(server_queue)
+
+    # A distinct store over the same DB stands in for the worker process.
+    worker_store = SqlAlchemySqlStoreImpl(SqliteSqlStoreConfig(db_path=db_path))
+    worker_queue = JobQueue(worker_store, table_name="jobs", lease_ttl_seconds=60)
+    with pytest.raises(KeyError):
+        await worker_queue.lease("worker-A")
+
+    await worker_queue.initialize()
+    leased = await worker_queue.lease("worker-A")
+    assert leased is not None
+    assert leased.job_id == record.job_id
+
+    await server_store.shutdown()
+    await worker_store.shutdown()