# How the Document Processing Pipeline Handles Uploads, Text Extraction, and Search Indexing in Macro

> Explore the Macro document processing pipeline. Learn how uploads, text extraction, and search indexing work seamlessly via S3, queues, and asynchronous workers.

- Repository: [Macro/macro](https://github.com/macro-inc/macro)
- Tags: internals
- Published: 2026-08-14

---

**The Macro platform uses an event-driven, three-stage pipeline that moves documents from client upload to searchable OpenSearch index through S3 storage, queue-based text extraction, and asynchronous indexing workers.**

The document processing pipeline in the [macro-inc/macro](https://github.com/macro-inc/macro) repository orchestrates file ingestion through a decoupled, queue-based architecture. This system handles **PDF and DOCX uploads**, extracts plain-text content using native parsers, and builds searchable indices with durability and horizontal scalability guarantees. This article breaks down each stage with source code references from the Rust-based implementation.

## Stage 1: Upload Handling and S3 Storage

### Generating Presigned Upload URLs

Clients initiate uploads by requesting a **presigned S3 URL** through the `S3UploadUrlAdapter`. This adapter is instantiated in [`document_cognition_service/src/main.rs`](https://github.com/macro-inc/macro/blob/main/document_cognition_service/src/main.rs) (lines 165–167):

```rust
let s3_upload_adapter = S3UploadUrlAdapter::new(
    s3_client,
    config.document_storage_bucket.to_string(),
    config.docx_document_upload_bucket.to_string(),
);

```

The `presigned_put_url()` method returns a temporary URL that clients use to upload directly to the `document_storage_bucket`.

### S3 Event Routing to Queues

Once the upload completes, S3 emits an **ObjectCreated** event. In local development, this routing is configured in [`tooling/xtask_local/src/local/resources.rs`](https://github.com/macro-inc/macro/blob/main/tooling/xtask_local/src/local/resources.rs) (line 147), which wires S3 notifications to the `document-upload-finalizer` queue.

The queue itself is defined in [`crates/macro_queues/src/lib.rs`](https://github.com/macro-inc/macro/blob/main/crates/macro_queues/src/lib.rs) as `DocumentUploadFinalizerQueue` — one of several queue types managed by the `macro_queues` crate.

## Stage 2: Text Extraction Pipeline

### Lambda Trigger and Message Enqueuing

The `upload_extractor_lambda_trigger` service (at [`services/upload_extractor_lambda_trigger/src/main.rs`](https://github.com/macro-inc/macro/blob/main/services/upload_extractor_lambda_trigger/src/main.rs)) listens for finalization events. Lines 71–78 show how it:

1. Reads the `BulkUploadRequest` record from the database
2. Validates the S3 key exists
3. Enqueues a `DocumentTextExtractor` message

```rust
let extractor_msg = DocumentTextExtractorMessage {
    bucket: config.document_storage_bucket.clone(),
    key,
    request_id: upload_request_id.to_string(),
};
sqs_client
    .document_text_extractor_queue()
    .send_message(&extractor_msg)
    .await?;

```

### Worker-Based Text Extraction

The **DocumentTextExtractor** service runs in `document_cognition_service`. At line 102 of [`services/document_cognition_service/src/main.rs`](https://github.com/macro-inc/macro/blob/main/services/document_cognition_service/src/main.rs), it creates the `document_text_extractor_queue`:

```rust
let document_text_extractor_queue = queues.document_text_extractor_queue().clone();

```

Workers pull messages and extract text using `pdfium` for PDFs and `docx` parsers for Word documents. The extracted plain-text is then written back to S3:

```rust
let msg = document_text_extractor_queue.receive_message().await?;
let plain_text = pdfium::extract_text(&msg.bucket, &msg.key).await?;
s3_upload_adapter.upload_text(&msg.key, &plain_text).await?;

```

## Stage 3: Search Indexing

### Queue Handoff to Search Processing

After successful extraction, the worker posts a **SearchExtractor** message to the `SearchQueue`. This decouples extraction from indexing, allowing each stage to scale independently.

```rust
let search_msg = SearchQueueMessage::Document {
    document_id,
    version: latest_version,
};
search_processing_queue.send_message(&search_msg).await?;

```

### OpenSearch Document Construction

The `search_processing_service` handles indexing. Its entry point is [`services/search_processing_service/src/process/worker.rs`](https://github.com/macro-inc/macro/blob/main/services/search_processing_service/src/process/worker.rs), which:

- Consumes messages from `SearchQueue`
- Delegates to [`services/search_processing_service/src/process/document.rs`](https://github.com/macro-inc/macro/blob/main/services/search_processing_service/src/process/document.rs) for document shape normalization
- Upserts records via the `opensearch_client` crate

The OpenSearch client wrapper at `crates/opensearch_client` provides the low-level HTTP interface for index operations.

## Reliability and Observability Guarantees

### Idempotency Through Database Tracking

Each pipeline stage records its status in the `bulk_upload` table. The helper methods in [`tooling/seed_cli/src/entity/document/mod.rs`](https://github.com/macro-inc/macro/blob/main/tooling/seed_cli/src/entity/document/mod.rs) (lines 251–259) demonstrate how uploads track state transitions, preventing duplicate processing.

### Stateless Workers and At-Least-Once Delivery

- **SQS queues** guarantee at-least-once message delivery
- Workers are **stateless** and horizontally scalable
- Failed messages return to the queue for retry with exponential backoff

### Distributed Tracing

Every major step uses `tracing::info!` and `tracing::error!` macros. The `worker-rs-otel` crate provides OpenTelemetry integration. Before shutdown, the `MacroEventBroker` flushes pending events (see [`services/document_cognition_service/src/main.rs`](https://github.com/macro-inc/macro/blob/main/services/document_cognition_service/src/main.rs), lines 85–100):

```rust
event_broker.flush().await?;

```

## Summary

- **Upload stage**: Presigned S3 URLs and ObjectCreated events trigger `DocumentUploadFinalizerQueue`
- **Extraction stage**: `upload_extractor_lambda_trigger` enqueues jobs; `document_cognition_service` workers extract text using `pdfium`/`docx` parsers
- **Indexing stage**: `search_processing_service` builds normalized documents and upserts to OpenSearch via `opensearch_client`
- **Resilience**: Idempotent database tracking, SQS durability, and distributed tracing ensure production reliability

## Frequently Asked Questions

### What file formats does the Macro document pipeline support?

The pipeline primarily handles **PDF** and **DOCX** files. The `document_cognition_service` uses `pdfium` for PDF text extraction and native `docx` parsers for Word documents, as implemented in the extraction worker at [`services/document_cognition_service/src/main.rs`](https://github.com/macro-inc/macro/blob/main/services/document_cognition_service/src/main.rs).

### How does the pipeline prevent duplicate document processing?

Each stage writes status updates to the `bulk_upload` table (see [`tooling/seed_cli/src/entity/document/mod.rs`](https://github.com/macro-inc/macro/blob/main/tooling/seed_cli/src/entity/document/mod.rs)). Workers check this state before processing, making operations **idempotent**. Additionally, SQS message deduplication and S3 object versioning provide extra safety layers.

### Can the text extraction and search indexing stages scale independently?

Yes. The architecture uses **separate SQS queues** (`DocumentTextExtractorQueue` and `SearchQueue`) with stateless workers. You can scale `document_cognition_service` workers for heavy extraction workloads while keeping `search_processing_service` workers minimal, or vice versa.

### What happens if text extraction fails midway?

Failed extractions return messages to `DocumentTextExtractorQueue` for retry. The `BulkUploadRequest` record retains error state, allowing operators to inspect and re-trigger processing. All errors are emitted via `tracing::error!` and captured by the OpenTelemetry pipeline in `worker-rs-otel`.