Implementing an Asynchronous Upload & Processing System with Golang

File uploads processed synchronously right away often become a source of problems when file sizes grow or processing logic gets heavier. HTTP requests waiting for Excel parsing, validating thousands of rows, or transforming data can time out, frustrate users, and burden the server for no clear reason. This article discusses a practical implementation of the upload-and-process-asynchronously pattern using Golang, with two worker approaches that have different trade-offs: a CLI Worker polling the database, and an AWS SQS Worker that’s event-driven. You’ll see the project structure, data model, code examples for each approach, and a guide for when to choose which based on traffic characteristics and available infrastructure.

Why Upload Needs to Be Separated from Processing

When a user uploads a file, there are actually two different things happening: receiving and storing the file, then processing it. Many naive implementations combine both in one HTTP request — the file is received, immediately parsed, validated row by row, saved to the database, and only then is the response sent. This approach works for small files, but collapses as soon as data size grows.

There are three main problems. First, timeouts. Load balancers, reverse proxies, or even browsers have response wait limits. A process taking 30 seconds or more gets cut off before finishing, even though the file was already uploaded and resources were already used. Second, resource contention. While that HTTP request runs, one worker from the HTTP server pool is held just to process one file. If there’s a surge of simultaneous uploads, the entire API can slow down because all workers are occupied with processing rather than serving other requests. Third, a poor user experience — users must wait in front of the screen without knowing progress, and if the connection drops mid-way, the processing status becomes unclear.

The solution is separating upload from processing into two independent stages:

STAGE 1 (Synchronous, fast):
  ✓ Receive the file
  ✓ Save to object storage
  ✓ Record the task in the database with status PENDING
  ✓ Respond to the user immediately

STAGE 2 (Asynchronous, in the background):
  ✓ Worker picks up PENDING tasks
  ✓ Process the file (parsing, validation, transformation)
  ✓ Update the status to DONE or FAILED

With this separation, the upload request finishes in milliseconds to seconds, while the heavy processing runs in a completely different process. The user gets a task_id as a reference and can check its status anytime without waiting on the same page.

sequenceDiagram
    participant User
    participant API
    participant Storage
    participant DB
    participant Worker

    User->>API: POST /upload (multipart file)
    API->>Storage: Save(file)
    Storage-->>API: file path
    API->>DB: INSERT upload_task (status=PENDING)
    DB-->>API: task_id
    API-->>User: { task_id, status: PENDING }

    Note over Worker: Runs independently
    Worker->>DB: FetchPendingTask()
    DB-->>Worker: task
    Worker->>DB: MarkProcessing(task_id)
    Worker->>Storage: Download(file path)
    Worker->>Worker: Process(file)
    Worker->>DB: MarkDone(task_id)

    User->>API: GET /uploads/{id}
    API->>DB: Query status
    DB-->>API: status=DONE
    API-->>User: { status: DONE }

High-Level Project Structure

The separation of upload and processing should also be reflected in the code structure, not just the execution flow. Two different entry points — one for the HTTP API, one for the worker — allow both to be deployed, scaled, and restarted independently without interfering with each other.

/cmd
  /api            -> HTTP API (upload, status)
  /worker         -> CLI worker (polling)
/internal
  /handler        -> HTTP handlers
  /service        -> Business logic
  /repository     -> DB access
  /model          -> DB structs
  /processor      -> File processing logic
  /storage        -> S3 / object storage abstraction

The /cmd folder contains two separate binaries. /cmd/api runs the HTTP server handling uploads and status queries — this process is lightweight and must always be responsive. /cmd/worker runs the background process doing polling and file processing — this process can be heavy and may be a bit slow, because no user is waiting for a direct response from here.

The /internal structure follows common layering: handler receives HTTP requests and parses input, service contains business logic (validation, orchestration between repositories), repository is purely for database access, model contains structs representing tables, processor is dedicated to the heavy file processing logic, and storage is an abstraction over S3 or other object storage so the code isn’t directly tied to a specific SDK.

The /cmd/api and /cmd/worker separation also means you can scale both differently. For example running 5 API replicas but only 2 worker replicas, or vice versa, according to each one’s load needs.

Data Model

A single upload_tasks table is enough to track a task’s entire lifecycle, from being received to finished processing or failed.

type UploadTask struct {
    ID           int64
    UserID       int64
    FilePath     string
    Status       string // PENDING, PROCESSING, DONE, FAILED
    ErrorMessage *string
    CreatedAt    time.Time
    UpdatedAt    time.Time
}

The Status field is the heart of this system. The four possible values represent a simple state machine:

stateDiagram-v2
    [*] --> PENDING: task created
    PENDING --> PROCESSING: worker picks up the task
    PROCESSING --> DONE: process succeeded
    PROCESSING --> FAILED: process failed
    FAILED --> PENDING: retry (optional)
    DONE --> [*]
    FAILED --> [*]

ErrorMessage is a pointer type (*string) because its value is optional — only filled when the status is FAILED. Using a pointer here is more appropriate than an empty string, because it explicitly distinguishes between “no error” and “an error with an empty message”. CreatedAt and UpdatedAt are important for observability — you can calculate how long a task waited in PENDING status, or how long the average process runs.

API: File Upload

The upload endpoint runs the flow described at the start: receive the file, save to storage, record it as a task, then respond. There’s no heavy processing logic here at all.

Flow

1. Receive the multipart upload
2. Save the file to object storage
3. Insert a record into the DB (status = PENDING)
4. (Optional) Push a message to SQS
5. Respond to the user

Example Handler

func UploadHandler(w http.ResponseWriter, r *http.Request) {
    file, _, err := r.FormFile("file")
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadRequest)
        return
    }
    defer file.Close()

    // ANTI-PATTERN: processing the file directly in the handler
    // records := parseExcel(file)
    // for _, rec := range records { validateAndSave(rec) }
    // -- this blocks the HTTP request until the entire file is processed

    // CORRECT: save the file, record the task, respond immediately
    path, err := storage.Save(file)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

    taskID, err := repo.CreateUploadTask(r.Context(), path)
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }

    // If using SQS, publish the event here
    // sqs.Publish(taskID)

    w.Header().Set("Content-Type", "application/json")
    json.NewEncoder(w).Encode(map[string]interface{}{
        "task_id": taskID,
        "status":  "PENDING",
    })
}

The anti-pattern comment above is deliberately shown as a direct contrast. Calling parseExcel and validation inside the handler would make the request wait until the entire file is processed — exactly the problem we want to avoid. The correct version stops after the file is saved and the task is recorded; no heavy logic runs here.

Always validate file size and type before saving to storage. Without initial validation, users could send files far larger than should be supported, or formats the processor can’t handle, which only gets discovered once the worker has already started.

API: Status & History

Once a task is created, users need a way to monitor its progress. Two simple endpoints are enough for this need.

Both query the upload_tasks table directly, no complex joins or extra caching needed for the basic use case. Because this table is relatively small per row and the queries are simple (filter by user_id, optional filter by status), indexes on the user_id and status columns are usually enough for good performance.

-- CORRECT: indexes for frequently used queries
CREATE INDEX idx_upload_tasks_user_id ON upload_tasks (user_id);
CREATE INDEX idx_upload_tasks_status ON upload_tasks (status);

-- ANTI-PATTERN: query without an index on an ever-growing table
-- SELECT * FROM upload_tasks WHERE user_id = ? ORDER BY created_at DESC;
-- -- without an index, this query does a full table scan once the table grows

Worker Option 1: CLI Worker with DB Polling

The first approach is a simple worker running as a long-lived process, periodically checking the database for tasks waiting to be processed.

Concept

The worker runs in an infinite loop. On each iteration, it tries to fetch one task with PENDING status. If one exists, the task is immediately changed to PROCESSING so other workers (if more than one instance runs) don’t pick up the same task. After processing completes, the status is updated to DONE or FAILED. If no task is waiting, the worker sleeps briefly before trying again.

flowchart TD
    A[Start loop] --> B{Any PENDING task?}
    B -- No --> C[Sleep 5 seconds]
    C --> A
    B -- Yes --> D[Mark PROCESSING]
    D --> E[Process file]
    E --> F{Success?}
    F -- Yes --> G[Mark DONE]
    F -- No --> H[Mark FAILED + error message]
    G --> A
    H --> A

Example Worker Loop

for {
    task, err := repo.FetchPendingTask(ctx)
    if err == sql.ErrNoRows {
        time.Sleep(5 * time.Second)
        continue
    }
    if err != nil {
        log.Printf("error fetching task: %v", err)
        time.Sleep(5 * time.Second)
        continue
    }

    if err := repo.MarkProcessing(ctx, task.ID); err != nil {
        log.Printf("error marking task %d as processing: %v", task.ID, err)
        continue
    }

    if err := processor.Process(task); err != nil {
        repo.MarkFailed(ctx, task.ID, err.Error())
        continue
    }

    repo.MarkDone(ctx, task.ID)
}

This loop is simple, but has one hidden weakness to watch out for: if more than one worker instance runs simultaneously (for example for redundancy or scaling), both could pick up the exact same task at nearly the same time, before either finishes marking it as PROCESSING.

Preventing Double Processing

The most common solution is using row-level locking at the database level when fetching a task, so only one worker successfully “claims” a given task.

-- ANTI-PATTERN: a plain SELECT without locking
-- SELECT * FROM upload_tasks WHERE status = 'PENDING' LIMIT 1;
-- -- two workers could read the same row before either one updates the status

-- CORRECT: SELECT FOR UPDATE SKIP LOCKED inside a transaction
BEGIN;
SELECT * FROM upload_tasks
WHERE status = 'PENDING'
ORDER BY created_at ASC
LIMIT 1
FOR UPDATE SKIP LOCKED;

UPDATE upload_tasks SET status = 'PROCESSING' WHERE id = $1;
COMMIT;

FOR UPDATE SKIP LOCKED (supported by PostgreSQL and newer MySQL versions) ensures one row can only be “held” by one transaction at a time, and other workers running similar queries will automatically skip locked rows — instead of waiting or failing. This lets multiple worker instances run simultaneously without the risk of picking up the same task.

Important Notes

BEFORE DEPLOYING A CLI WORKER, MAKE SURE:
  □ The task fetch query uses a transaction + locking (FOR UPDATE SKIP LOCKED)
  □ The polling interval matches the required SLA (shorter = more responsive, but more queries)
  □ There's a mechanism for tasks "stuck" in PROCESSING too long (worker crashed mid-process)
  □ Logging records the task_id at every step to make debugging easier

If a worker crashes right after marking a task as PROCESSING but before completing the processing, that task will be “stuck” forever unless there’s an additional mechanism — for example a separate job checking tasks with PROCESSING status older than a certain threshold, then returning them to PENDING for reprocessing.

The CLI Worker with polling suits small-to-medium workloads well, especially when the infrastructure doesn’t yet need message queue complexity. There’s no external dependency besides the existing database, so it’s easier to run on-premise or in a simple environment.

Worker Option 2: AWS SQS Worker (Event-Driven)

The second approach replaces polling with an event-driven model using AWS SQS (Simple Queue Service). Instead of the worker constantly asking “any new tasks?”, the API directly sends a notification the moment a task is created, and the worker reacts to that notification.

Flow

1. Upload -> push a message to SQS
2. SQS triggers a Lambda / container worker
3. Worker takes the task_id from the message
4. Process the file
5. Update the DB

sequenceDiagram
    participant API
    participant SQS
    participant Worker
    participant DB
    participant Storage

    API->>SQS: Publish({ task_id })
    SQS->>Worker: Trigger (poll or event source mapping)
    Worker->>DB: GetTask(task_id)
    Worker->>DB: MarkProcessing(task_id)
    Worker->>Storage: Download(file)
    Worker->>Worker: Process(file)
    alt Success
        Worker->>DB: MarkDone(task_id)
        Worker->>SQS: Delete message (ack)
    else Failure
        Worker->>DB: MarkFailed(task_id, error)
        Worker->>SQS: Message becomes visible again (implicit nack)
    end

The fundamental difference from the CLI Worker is who initiates the action. In the CLI Worker, the worker actively asks the database. In the SQS Worker, the API actively tells the worker there’s new work. This model eliminates the polling interval delay — a task can be processed immediately once the message arrives in the queue, not waiting for the next polling cycle.

Message Payload

The SQS payload is deliberately minimal — just task_id. Full task details are still fetched from the database during processing, not tucked into the message payload.

{
  "task_id": 123
}

// ANTI-PATTERN: putting the entire task data in the message payload
// {
//   "task_id": 123,
//   "user_id": 456,
//   "file_path": "s3://bucket/file.xlsx",
//   "metadata": { ... }
// }
// -- if the DB data changes after the message is sent, the worker processes stale data

// CORRECT: minimal payload, the worker fetches the latest data from the DB
// { "task_id": 123 }

This minimal payload approach matters because messages in SQS can be delayed in processing (for example during traffic surges), and you don’t want workers processing data that’s no longer relevant to the latest database state.

Example SQS Handler

func HandleMessage(ctx context.Context, taskID int64) error {
    task, err := repo.GetTask(ctx, taskID)
    if err != nil {
        return fmt.Errorf("get task %d: %w", taskID, err)
    }

    if err := repo.MarkProcessing(ctx, task.ID); err != nil {
        return fmt.Errorf("mark processing %d: %w", taskID, err)
    }

    if err := processor.Process(task); err != nil {
        repo.MarkFailed(ctx, task.ID, err.Error())
        return err
    }

    return repo.MarkDone(ctx, task.ID)
}

Returning an error from this function matters because SQS uses a result-based acknowledgment mechanism. If the handler returns an error, the message isn’t deleted from the queue, so SQS will make that message visible again after the visibility timeout ends — effectively triggering automatic retry without extra logic on your side.

Important Configurations

If the visibility timeout is too short compared to the actual processing time, the same message could be picked up by another worker before the first one finishes — causing the same task to be processed twice simultaneously. Always set the visibility timeout with a safe margin above the realistic maximum processing time, and make sure the processor is idempotent as a second line of defense.

The DLQ (Dead Letter Queue) acts as a “holding area” for messages that repeatedly fail processing, instead of being retried endlessly or lost entirely. Tasks entering the DLQ can be investigated manually, and this prevents one consistently failing task from repeatedly consuming worker resources.

The File Processor

File processing logic should be truly separate from the task retrieval mechanism — whether the task comes from DB polling or from SQS, the core logic must be exactly the same. This allows both worker approaches to call the identical Process function.

func Process(task UploadTask) error {
    file, err := storage.Download(task.FilePath)
    if err != nil {
        return fmt.Errorf("download file: %w", err)
    }
    defer file.Close()

    records, err := parseExcel(file)
    if err != nil {
        return fmt.Errorf("parse excel: %w", err)
    }

    for _, r := range records {
        if err := validate(r); err != nil {
            // ANTI-PATTERN: stopping the entire process because of one invalid row
            // return fmt.Errorf("validation failed: %w", err)

            // CORRECT: record the failed row, continue processing other rows
            log.Printf("task %d: invalid row, skipped: %v", task.ID, err)
            continue
        }

        if err := repo.SaveRecord(r); err != nil {
            return fmt.Errorf("save record: %w", err)
        }
    }

    return nil
}

The decision to skip invalid rows rather than stopping the entire process is an important design choice to consciously consider. In a file with thousands of rows, one row with a wrong date format shouldn’t cause the whole file to fail processing — better to record that row as individually failed while the other valid rows still get saved. The trade-off: you need an additional mechanism to report which rows failed, so users still have visibility into this partial result.

If one file can produce a mix of successful and failed rows, consider adding fields like success_count and failed_count to the upload_tasks table, or a separate upload_task_errors table recording which rows failed and why.

Observability & Monitoring

Asynchronous systems by design separate execution time from request time, which means debugging can no longer rely only on HTTP responses. Without good observability, you’ll only know something’s wrong when users complain their task never finishes.

Four things should always be present:

Logging per task_id. Every log related to task processing must include task_id as an identifier, so you can trace one task’s entire lifecycle — from being received, entering the queue, starting processing, to finishing or failing — just by filtering on that ID.

Success vs failed metrics. Count the ratio of successful to failed tasks over a time period. A sudden spike in the failure ratio usually indicates a systemic problem — not just an individual corrupt file — like a data format change from an upstream source, or a newly deployed bug.

Timeout guard. A process that should finish in seconds but runs far longer is most likely having a problem (deadlock, infinite loop, or an unresponsive external dependency). Use context.WithTimeout so stuck processes can be forcibly stopped rather than hanging forever.

Context cancellation. Make sure every I/O operation (database queries, storage downloads, external API calls) receives and honors the propagated context.Context, so the process can be gracefully cancelled when a timeout is reached or the application shuts down.

func Process(ctx context.Context, task UploadTask) error {
    ctx, cancel := context.WithTimeout(ctx, 2*time.Minute)
    defer cancel()

    file, err := storage.Download(ctx, task.FilePath)
    if err != nil {
        return fmt.Errorf("task %d: download file: %w", task.ID, err)
    }
    defer file.Close()

    // ... continue processing using the same ctx
    return nil
}

CLI Worker vs SQS Worker Comparison

When to Choose Which Option

flowchart TD
    A{Already on AWS infrastructure?} -- No --> B[CLI Worker]
    A -- Yes --> C{Upload traffic is bursty / unpredictable?}
    C -- No, stable and small --> B
    C -- Yes --> D{Per-file processing very long, e.g. > 15 minutes?}
    D -- Yes --> E[AWS Batch]
    D -- No --> F{Sensitive to idle cost?}
    F -- Yes --> G[SQS + Lambda]
    F -- No --> H[SQS + Container Worker]

There’s no universal answer between these two approaches — both are valid for different contexts. The CLI Worker excels in simplicity and independence from any specific cloud vendor, while the SQS Worker excels in scalability and naturally handling traffic surges without extra logic.

These two approaches aren’t mutually exclusive. Some teams start with a CLI Worker for an MVP or early product stage, then migrate to an SQS Worker after traffic grows and scalability needs become more real. Because the Process logic is separated from the task retrieval mechanism, this migration is relatively non-disruptive to the core processing code.

Implementation Notes

The code examples in this article are deliberately simplified and don’t strictly implement the full service-repository pattern, so the discussion stays focused on the asynchronous upload-process flow itself. In real production implementations, you’ll likely add a service layer for orchestrating more complex business logic, dependency injection for testability, and explicit interfaces on repository and storage so they’re easier to mock during unit testing.

This asynchronous upload-process pattern is very common in enterprise systems — from mass data imports, report generation, to media file transformation. Its core characteristics are always the same: uploads must be fast and responsive, while heavy processing runs in the background without burdening HTTP requests, and users retain full visibility into their task status through task_id.

Summary

• Separate upload from processing — the HTTP request only receives the file and records the task; heavy processing runs in a separate process.
• The project structure separates entry points — /cmd/api for the HTTP server, /cmd/worker for the background process, each independently scalable.
• A simple state machine with PENDING → PROCESSING → DONE/FAILED statuses is enough to track a task’s lifecycle.
• The CLI Worker (DB polling) suits small-to-medium traffic and on-premise infrastructure; use FOR UPDATE SKIP LOCKED to prevent double processing across worker instances.
• The SQS Worker (event-driven) suits traffic bursts and cloud environments; leverage SQS’s built-in visibility timeout, DLQ, and retry.
• Message payloads should be minimal — just task_id is enough; full details are refetched from the database during processing.
• File processing logic (Process) must be separated from the task retrieval mechanism, so it can be called from both worker approaches without duplication.
• Observability is mandatory — per-task_id logging, success/failed metrics, timeout guards, and context cancellation prevent tasks from “disappearing” without a trace.
• There’s no universal choice between the CLI Worker and SQS Worker — the decision depends on traffic characteristics, available infrastructure, and cost sensitivity.