Moving Our Document Upload Pipeline to Background Workers

At work, uploading a document was never just uploading a file.

A user picked a PDF or DOCX and clicked upload. That one click kicked off a chain of steps: check the organization's page quota, create a MongoDB record, get a signed upload URL, send the file to S3, and start ingestion in our Python backend. The backend gave us back a job ID, and we polled it to see how processing was going.

Then we wanted a new way in: connect OneDrive, pick a folder, and let documents flow in on their own.

The same steps still had to happen. But now there might be no button click, no open tab, and nobody around to push the document to the next step.

That's what led me to background workers with Redis and BullMQ.

I also built a small demo repository that makes the queue hand-offs easy to see, and I'll use it throughout this post. The real feature used OneDrive; the demo fakes a "new file in drive" event instead. It shows the worker setup, not a full cloud-drive integration. S3 upload and ingestion are mocked, but pipeline progress and quota counts are real MongoDB writes.

What happened after clicking upload

Before talking about queues, let's unpack everything hiding behind that one button.

1. Check the page quota. Make sure the document won't push the organization over its remaining pages.
2. Create the MongoDB record. Give the app something to attach the document and its progress to.
3. Prepare the upload. Generate a signed URL for where the file will live in S3.
4. Upload the file to S3. The database record and the actual file are separate things. Creating one doesn't create the other.
5. Start ingestion. Call the Python backend and get a job ID back.
6. Track processing. Poll that job ID until it succeeds or fails.
7. Update page usage. The backend file processor does this after processing.

The quota check at the start and the usage update at the end are different jobs. The first one blocks documents that clearly won't fit. The second one records the pages that were actually processed. The early check doesn't reserve anything, so it isn't safe when several uploads happen at once. More on that later.

The Python side was already asynchronous: calling it returned before ingestion finished. But that only covered the waiting inside that one service. Someone still had to own the bigger workflow around it.

That's the key point: one async backend call doesn't make the whole document lifecycle independent of the browser.

OneDrive changed who owns the workflow

With a manual upload, the user brings the file and starts the steps. With a connected folder, the system has to find new files and start the steps by itself.

Say someone connects a folder and later drops twenty PDFs into it. For each file, we need to link it to the right organization, check whether it can be processed, create its records, copy it into our storage, and track ingestion. The user should be able to come back later and see how it went.

The frontend can show that progress. It shouldn't be the thing keeping the work alive.

This is also why firing off an async function without awaiting it in an API route isn't enough. That work still lives inside the web server. A deploy, a crash, or a request timeout can kill it, and a loose promise leaves no record of what was left to do.

We needed to hand the work to something else that could pick it up.

The three pieces: producer, queue, worker

It gets easier once each piece has one clear job:

Redis stores the queue here. It's not where I'd put PDF data or full file contents. Jobs should carry IDs, and each worker fetches what it needs using those IDs.

BullMQ gives you "jobs" on top of Redis. A worker picks up a job, runs it, and either finishes or throws an error. BullMQ records which one happened.

Once the server has accepted the work, the browser can go away. Workers keep going as long as the worker process and the services it depends on are running.

A chain of queues in the demo

The demo splits the work into four stages:

object-create → s3-upload → ingest → quota

Each stage has its own queue and processor. When a processor finishes, it adds a job for the next stage.

This diagram shows the demo chain, not the full OneDrive feature. In particular, the last quota stage only adds to the usage count. It doesn't do the upfront quota check from the manual flow. In a real folder import, that check should still happen before any expensive work starts.

The queue definitions make the stages explicit:

export const QUEUE_NAMES = {
  objectCreate: "object-create",
  s3Upload: "s3-upload",
  ingest: "ingest",
  quota: "quota",
} as const;

You could run the whole pipeline in one worker, and for a small workflow that's a fine place to start. Splitting it into stages makes it easy to see where a document is stuck, and lets each stage have its own concurrency limit.

The catch: every hand-off is another place where things can break halfway. Four queues don't make the pipeline all-or-nothing.

The producer starts the chain

In the demo, POST /api/drive/webhook stands in for "a new file showed up". It creates a jobChainId, saves a pending run in MongoDB, and adds the first job.

In the public demo, the producer is a small Express route:

const now = new Date();
await runs.insertOne({
  _id: jobChainId, jobChainId, orgId, driveFileId, fileName,
  status: "pending", createdAt: now, updatedAt: now,
});
 
await getQueue(QUEUE_NAMES.objectCreate).add("object-create", jobData, {
  jobId: `${jobChainId}-object`,
});

Note that adding the job is awaited. We only reply once Redis has actually accepted the job, instead of claiming background work exists before it does.

The reply means accepted, not done. The producer sends back the chain ID so the UI can check progress later.

The demo endpoint uses a fixed demo organization and fake input on purpose. A real connector has to verify the incoming event, look up the connected account and folder on the server, and check that the organization actually owns it. An org ID sent by the caller doesn't prove anything.

Also note there are two writes: MongoDB first, then Redis. If the database write works but adding the job fails, you're left with a pending run and no job. A reliable import system needs a way to recover from that. Returning an error doesn't undo the MongoDB write.

Pass IDs forward, not the whole file

The demo payload starts with the file and organization IDs. Each stage adds what it produces:

export type PipelineJobData = {
  jobChainId: string;
  orgId: string;
  driveFileId: string;
  fileName: string;
  pages: number;
  simulateFailure: boolean;
  caseId?: string;
  documentId?: string;
  s3Key?: string;
  signedUrl?: string;
};

jobChainId ties the whole import together. It's not the same as any single stage's BullMQ job ID, or the Python ingestion job ID.

Each ID answers a different question:

• Chain ID: which document are we following?
• BullMQ job ID: which stage (or attempt) are we looking at?
• Python job ID: which ingestion run should we ask about?

Keeping them separate makes logs and debugging much easier.

The demo passes a mock signedUrl along. In a real import, I'd store something permanent like the S3 key and create a fresh short-lived URL when needed. A job can sit in the queue longer than a signed URL stays valid. Also, signed URLs and OAuth tokens grant access, so treat them more carefully than normal data.

What each worker does

1. Object creation

The object-create processor sets the run status to creating, makes stable mock caseId and documentId values from the chain ID, and saves them on the run. Then it queues s3-upload with those IDs.

This stands in for creating app records. It's not a full production case/document model.

The hand-off itself is short:

await getQueue(QUEUE_NAMES.s3Upload).add("s3-upload", next, {
  jobId: `${data.jobChainId}-s3`,
});

The upload stage now has everything it needs, with no dependency on the original HTTP request.

2. S3 upload

The s3-upload processor checks that the case and document IDs exist, builds an S3 key, and fakes the signing and upload. Then it queues ingestion.

In the OneDrive version, this stage also needs the actual file. There's no browser File anymore. The server has to download it from OneDrive and send it to S3, dealing with auth, failed transfers, size limits, and timeouts along the way.

The worker can use a fresh presigned URL or upload with the AWS SDK. Either way, the rule is the same: don't start ingestion until the file is safely in S3.

3. Ingestion

The ingest processor takes the fake page count from the demo form and queues the quota stage. The worker runner adds a delay before each stage so you can watch the hand-offs.

The real backend has one more loop:

submit document → receive Python job ID
                         ↓
                  query job status
                         ↓
               pending / done / failed

Getting a job ID back only means the backend accepted the request. It doesn't mean ingestion is done. A real worker has to save that ID and keep checking until it finishes or fails, before marking the document ready. The backend file processor handles the page-usage update.

We don't need to know what Python does inside. From the worker's point of view, it's just a service where you submit work, check its status, and get a result.

4. Quota accounting

The quota processor adds to the organization's pagesUsed only if this chain hasn't been counted yet, then marks the run done.

This stage exists only in the demo, to make the hand-off visible. In the actual application, no worker or API route touches page usage: the backend file processor updates it, and the frontend only displays progress and the result.

A plain MongoDB $inc would count twice if the increment worked but the worker died before marking the job done. The demo guards against this: one conditional update both increments usage and adds the chain ID to a chargedRuns array on the organization. Running the job again won't count it twice. That growing array is a shortcut for the demo, not a real long-term usage ledger.

Polling moves to the server too

Moving the upload into a worker is only half of it. Someone still has to keep checking the Python job when there's no browser.

There are two simple ways to do that:

For a small workload, polling inside the worker is fine. You still need request timeouts, a polling interval, an overall deadline, and clear handling for failed or unknown states. An async sleep doesn't block Node's event loop, but the job still counts as active.

For longer processing, delayed check-up jobs work well: check once, save the result, and schedule another check if needed. Just make sure a Python job can't get forgotten if the worker dies between checking and scheduling the next check.

Either way, after a restart the worker should pick up the saved ingestion job, not submit the document again. And if the submit worked but the response got lost, you need the backend to support retries safely, or a way to look up the job by a stable document ID.

Two kinds of quota work

This is easy to mix up because both involve page counts.

The upfront check asks: can this document go in? The final step records how many pages were actually used. In our system, the first happens before submission; the second belongs to the backend file processor.

Say an organization has 100 pages left, and two workers each pick up a 70-page document:

Both checks pass. Together, they go over the limit.

Moving the check into a worker doesn't fix this race. You need an atomic rule, like reserving the pages against the limit up front and adjusting once the real count is known. Failed or cancelled imports need to release their reservation. The exact approach depends on when you can get a reliable page count.

DOCX files are tricky here: the real page count might only be known after conversion. If the early check is only an estimate, don't treat it as a hard guarantee.

Whichever service owns final accounting (for us, the backend file processor), a durable usage entry keyed by organization and document version can tell whether that work was already counted. If saving the entry and updating the counter are two separate writes, they still need a transaction or a way to fix things up later. A unique key alone doesn't make two writes atomic.

These are requirements for the real import path. The demo keeps the accounting stage small on purpose so you can watch it.

Retries need safe side effects

BullMQ can retry failed jobs with a set number of attempts and a backoff. This example is something you could add, not a setting the demo uses today:

await queue.add("s3-upload", data, {
  jobId: `${data.jobChainId}-s3`,
  attempts: 3,
  backoff: { type: "exponential", delay: 1000 },
});

attempts: 3 means three tries in total, including the first one. Backoff waits longer between retries so you don't keep hammering a service that's already down. To mark a job as failed, the processor has to throw an Error. See BullMQ's retry docs.

The demo runs each job once, with no automatic retries. Its failure switch throws during object creation. The Clear simulated failure & retry button clears that flag and retries the same job. You can also retry from Bull Board, but if simulateFailure is still true, it'll just fail again.

Retries help with temporary problems like a network blip. They don't help if the account connection was revoked or the file type isn't supported. Those need a different outcome, like asking the user to reconnect or rejecting the file. Retrying forever won't make them work.

The bigger issue: retrying a stage can repeat whatever it already did. BullMQ's idempotent jobs guide puts the goal simply: the end result should be the same whether a job worked the first time or after a retry.

For this pipeline, that means asking:

• If object creation runs twice, does it find the existing document instead of making a new ID?
• If the upload runs twice, does it write to the same, stable storage key?
• If submitting to Python times out, can we tell whether it already started?
• If the same document is submitted twice, will the backend charge its pages twice?

For S3, reusing a key only helps if it can't overwrite a different version of the file by mistake.

The demo's stage job IDs, like chain-id-s3, stop the same job from being added twice while the original is still kept in that queue. They don't undo repeated MongoDB writes, and once a finished job is removed, its ID can be used again. See BullMQ's job ID guide.

There's one more gap before the chain even starts. The demo makes a new chain ID for every event, so getting the same OneDrive event twice creates two chains. A real connector needs a stable import ID, usually built from the organization, connection, file, and file version, and enforced in the database.

Queue hand-offs are not a transaction

Each stage writes some data and then queues the next job. Either of those can fail on its own.

If the MongoDB update works but adding the next Redis job fails, the next stage never starts. If the next job gets queued but the worker dies before BullMQ marks the current one done, the current stage may run again while the next one is already waiting.

So a status field helps, but it isn't proof. A run that says uploading doesn't mean an upload worker has picked it up. The demo sets some statuses right before queuing the next job.

For a production pipeline, one fix is the outbox pattern: save the status change and a "queue the next stage" note in the same MongoDB transaction, and have a separate process keep pushing those notes to Redis until they land. Stable job IDs and idempotent processors take care of duplicates. Another option is a cleanup job that finds stuck runs and safely re-queues whatever stage is missing.

Either way, you need clear rules. BullMQ manages jobs in Redis. It can't make MongoDB, S3, and an HTTP service succeed or fail together.

Set concurrency based on what's downstream

The worker runner sets up the stages like this:

These are demo values, not tuned production numbers. Uploads, database writes, and ingestion can each handle very different loads.

BullMQ lets you set concurrency per worker and also run multiple worker processes. The limit is per worker: two ingestion workers with concurrency 1 can still run two jobs at once. Higher concurrency helps with I/O-heavy work like network calls, but it won't make CPU-heavy JavaScript run in parallel. See the concurrency guide.

The demo runs all four Workers in a single Node.js process. They're separate consumers, not four separate services. You can split them out later if one stage needs more capacity or isolation.

If ingestion is the bottleneck, adding more upload workers just makes the ingestion line longer. Before adding workers, I'd look at how long the oldest document has been waiting, how long each stage takes, and whether downstream services are throttling.

Making progress visible

Once the browser isn't driving things, the app needs a way to answer "what happened to my document?"

The demo saves a pipeline run in MongoDB with the jobChainId, status, document IDs, S3 key, page count, and any error. The /api/jobs/:id endpoint reads that run, and the UI uses /api/runs to show recent runs and usage.

A real status endpoint has to check that the caller can see that organization's runs. The demo endpoints are for local use only and don't do that.

There are two kinds of progress:

• App progress: which document is importing, what stage it's at, and whether the user needs to do anything.
• Queue progress: which jobs are waiting, active, completed, or failed.

Bull Board shows the queue side in the demo. It helps explain why a document is waiting, while MongoDB holds the history users see. When object-create completes, only that stage is done; the document is ready only after every stage succeeds.

In production, I'd log the chain ID in every stage and watch how long jobs wait, failures, retries, stalled jobs, and time per stage. And an admin dashboard like Bull Board needs auth before it's exposed anywhere beyond your machine.

What the OneDrive connector adds

The queue chain starts once we know there's work. A connected folder also needs something that finds that work.

The connector has to handle the files already in the folder, changes after that, pagination, and remembering where it left off. Microsoft Graph's delta API returns file changes page by page, plus a link to fetch the next batch of changes later. Notifications can tell you to sync, but you still have to fetch the changes and work out what's new.

There are product questions too: does editing a file create a new import? Does deleting a file in OneDrive delete anything on our side? Which subfolders and file types count? Those answers decide the import ID and what the producer queues.

If you save the sync position before the imports are safely recorded, a crash can make you skip files. If you save it after, you might process some changes twice. That's why handling duplicates is part of the connector's job.

The connector also handles token refresh, revoked access, and making sure only the chosen folder's allowed files get imported. The demo skips this whole layer with a fake event so the worker hand-offs stay easy to follow.

Running the demo locally

With Node.js 22.9 or later and Docker installed:

git clone https://github.com/faisal004/redis-bullmq-demo.git
cd redis-bullmq-demo
npm ci
cp .env.example .env
npm run demo

npm run demo starts Redis and MongoDB with Docker Compose, then runs the web app, the workers, and Bull Board. Open the app at http://localhost:3100 and the board at http://localhost:3100/queues.

Click Simulate new document and watch it move through the queues. Delays are added on purpose so you can see each stage. The final quota update is a real MongoDB write; the file transfer and page count are fake.

To really see the separation, stop only the workers and leave the web app and Redis running. Submit another fake file, then start the workers again. The waiting job gets picked up, even though the original request is long gone. To control each service on its own, use the separate-terminal commands in the README.

For a real deployment, how Redis stores data matters. A queue in Redis isn't safe from data loss by default. BullMQ recommends turning on persistence and setting the noeviction memory policy. Also keep in mind that job data is stored in plain text. See the production guide.

The demo Redis helper uses maxRetriesPerRequest: null for workers, while producer connections give up after a few retries and don't queue commands while offline. That way an HTTP request fails fast instead of hanging, while a worker can just wait for Redis to come back. BullMQ's connection guide explains why.

What changed for the user

The goal for the user was simple: connect a folder and let documents flow in without uploading each one by hand.

To make that work, the folder flow had to handle quota checks, record creation, file transfer, and ingestion tracking on its own. The backend file processor kept doing what it already did: processing and usage accounting.

Redis and BullMQ gave me a way to pass work between processes and see each stage. The demo makes those hand-offs concrete. But the reliability still comes from decisions in the app: stable IDs, saved progress, safe retries, and a plan for when things fail halfway.

The big change was that importing a document became work the system owned from start to finish, even after the user closed the tab.

Bye for now!