How the Worker Lock Mechanism Prevents Duplicate PDF Processing in OLMocr

The OLMocr worker lock mechanism creates deterministic file-based locks in a workspace or S3 backend to ensure only one worker owns a specific PDF group at any time, using modification timestamps to detect stale locks and prevent duplicate processing.

When processing large-scale PDF collections across distributed workers, race conditions can cause the same document to be processed multiple times, wasting compute resources. The allenai/olmocr repository solves this through a robust file-based coordination system that implements distributed locking with automatic timeout handling. This mechanism guarantees exclusive access to work items while ensuring crashed workers do not permanently block progress.

File-Based Lock Architecture

The lock system relies on deterministic hash-based filenames and filesystem metadata to track work ownership across parallel workers. Each component of the mechanism is designed to be atomic and resilient to network failures or process crashes.

Lock File Naming Convention

Every work item in the queue receives a deterministic hash (WorkItem.hash) that uniquely identifies a group of PDF paths. When a worker claims a work item, it creates a lock file named worker_<hash>.lock in the worker_locks directory. The path construction is handled by _get_worker_lock_path in olmocr/work_queue.py:

def _get_worker_lock_path(self, work_hash: str) -> str:
    return os.path.join(self._locks_dir, f"worker_{work_hash}.lock")

This deterministic naming ensures all workers agree on the canonical lock location for any given work item, preventing split-brain scenarios where different workers create conflicting locks.

Stale Lock Detection with Timeouts

Before pulling work from the queue, workers call Backend.is_worker_lock_taken to verify availability. The system uses the lock file's modification time (mtime) to determine if work is actively claimed or abandoned:

• If the lock file does not exist, the work is available
• If the file exists and its age is ≤ worker_lock_timeout_secs (default 30 minutes), the work is locked and skipped
• If the file is older than the timeout, it is treated as stale and can be reclaimed

async def is_worker_lock_taken(self, work_hash: str, worker_lock_timeout_secs: int = 1800) -> bool:
    lock_path = self._get_worker_lock_path(work_hash)
    lock_mtime = await self._get_object_mtime(lock_path)
    if not lock_mtime:
        return False
    now = datetime.datetime.now(datetime.timezone.utc)
    return (now - lock_mtime).total_seconds() <= worker_lock_timeout_secs

This timeout-based approach ensures that if a worker crashes mid-processing, its lock eventually expires, allowing other workers to reclaim and complete the work without manual intervention.

Lock Lifecycle Implementation

The core coordination logic resides in olmocr/work_queue.py, which manages the complete lifecycle from lock acquisition through release.

Acquiring Work Ownership

When WorkQueue.get_work identifies an available work item, the worker atomically claims it by creating an empty lock file. The mere presence of the file denotes ownership, and the operation includes timeout handling to prevent queue blocking:

async def create_worker_lock(self, work_hash: str) -> None:
    lock_path = self._get_worker_lock_path(work_hash)
    with open(lock_path, "wb"):
        pass

If lock creation fails due to unexpected timeouts or filesystem errors, the item is returned to the queue and the worker proceeds to the next candidate, maintaining system availability.

Integration with Work Queue Processing

The get_work method implements the complete locking protocol. It first checks the completed cache, verifies lock status, and conditionally acquires ownership:

if await self.backend.is_worker_lock_taken(work_item.hash, worker_lock_timeout_secs):
    logger.debug(f"Work item {work_item.hash} is locked …")
    self._queue.task_done()
    continue

# Acquire lock

await self.backend.create_worker_lock(work_item.hash)

This sequence ensures that the check-and-set operation is atomic from the perspective of the individual worker, preventing race conditions where two workers might simultaneously claim the same PDF group.

Releasing Locks After Completion

Upon successful processing, WorkQueue.mark_done performs cleanup by creating a completion flag and removing the lock:

await self.backend.create_done_flag(work_item.hash)
await self.backend.delete_worker_lock(work_item.hash)

The done flag (done_<hash>.flag) serves as a permanent marker that the work is complete, while deleting the lock file frees the slot for any future retry scenarios.

Distributed Worker Implementation

The locking mechanism supports both local filesystem and S3 backends, enabling deployment across single machines or distributed cloud environments. The LocalBackend and S3 backend implementations in olmocr/work_queue.py and olmocr/s3_utils.py provide consistent semantics across storage systems.

Here is a complete example of running parallel workers that respect the lock boundaries:

import asyncio
from olmocr.work_queue import WorkQueue, LocalBackend

async def worker_loop(q: WorkQueue):
    while True:
        item = await q.get_work()
        if item is None:                     # queue empty

            break
        # Process the PDFs listed in item.work_paths

        print(f"Processing {len(item.work_paths)} PDFs...")
        await q.mark_done(item)              # releases the lock

backend = LocalBackend(workspace_path="/tmp/olmocr_workspace")
queue = WorkQueue(backend)

# Populate the queue before starting workers

await queue.populate_queue(work_paths=["doc1.pdf", "doc2.pdf"], items_per_group=5)
await queue.initialize_queue()

# Start 4 parallel workers with duplicate protection

await asyncio.gather(*(worker_loop(queue) for _ in range(4)))

Handling Worker Failures and Recovery

The 30-minute default timeout (worker_lock_timeout_secs) acts as a dead-worker detector. If a worker crashes after acquiring a lock but before calling mark_done, the lock file persists but ages beyond the threshold. Subsequent workers detect the stale timestamp and safely reclaim the work, ensuring eventual progress without duplicate effort during the initial locked period.

The test suite in tests/test_s3_work_queue.py validates this behavior:

async def test_get_work_locked():
    # Simulate an active lock (mtime < timeout)

    fake_s3 = FakeS3Backend()
    await fake_s3.create_worker_lock("abc123")
    wq = WorkQueue(fake_s3)
    # Populate queue with work item having hash "abc123"

    result = await wq.get_work()
    assert result is None        # worker correctly skips locked work

Summary

• Deterministic hashing generates consistent lock filenames (worker_<hash>.lock) across all workers via _get_worker_lock_path in olmocr/work_queue.py
• Timestamp-based detection uses file modification times to distinguish active locks from stale ones, with a default 30-minute timeout threshold
• Atomic acquisition through create_worker_lock ensures only one worker owns a PDF group at any time, while is_worker_lock_taken prevents duplicate processing attempts
• Automatic recovery allows new workers to reclaim work from crashed processes once locks expire, ensuring pipeline progress without manual cleanup
• Dual backend support implements identical locking semantics for both local filesystem and S3 storage through the Backend abstraction layer

Frequently Asked Questions

How does OLMocr handle a worker crash during PDF processing?

When a worker crashes after acquiring a lock but before completion, the lock file remains in the workspace. However, because is_worker_lock_taken compares the file's modification time against worker_lock_timeout_secs (default 30 minutes), the stale lock is eventually ignored. A subsequent worker detects the expired timestamp, claims the work, and processes the PDF group, ensuring no permanent work loss occurs.

What is the default worker lock timeout and can it be configured?

The default worker lock timeout is 30 minutes (1800 seconds), defined by the worker_lock_timeout_secs parameter in get_work. This value can be adjusted based on expected PDF processing times; shorter timeouts enable faster recovery from crashed workers, while longer timeouts prevent premature work reclamation during slow processing of large documents.

Where are lock files physically stored in the OLMocr architecture?

Lock files are stored in the worker_locks subdirectory of the workspace. For local deployments using LocalBackend, this is a filesystem path. For distributed deployments using the S3 backend, locks are objects within the S3 bucket prefix. The path construction in _get_worker_lock_path ensures consistent locations regardless of backend type.

How does the lock mechanism prevent race conditions between simultaneous workers?

The mechanism prevents race conditions through check-and-set semantics in get_work. Workers first verify lock status via is_worker_lock_taken, then immediately attempt to create the lock file. While the underlying filesystem or S3 provides the atomicity guarantee for the file creation itself, the timestamp-based validation ensures that even if two workers check simultaneously, only the first successful creator retains valid ownership, and subsequent attempts detect the fresh lock and skip the work.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}