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

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 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 (lines 165–167):

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 (line 147), which wires S3 notifications to the document-upload-finalizer queue.

The queue itself is defined in 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) 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

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, it creates the document_text_extractor_queue:

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:

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.

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, which:

• Consumes messages from SearchQueue
• Delegates to 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 (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, lines 85–100):

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.

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). 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.

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