Resumable Chunked Uploads for Large GeoPackages

← Back to Async Bulk Uploads with Celery

This page covers accepting multi-gigabyte spatial files over connections that cannot be relied on to stay up, so a failure at 80 % costs one chunk rather than the whole transfer.

# Context & When to Use

Field data arrives as large files over poor links: a 4 GB GeoPackage from a survey laptop on a hotel connection, a multi-gigabyte shapefile bundle from a partner over a VPN. A single POST of that file is a bet that nothing interrupts it for twenty minutes. When the bet loses — and over a mobile link it loses regularly — the client starts again from zero, and the third attempt is no more likely to succeed than the first.

Chunked, resumable upload changes the unit of failure from the file to the chunk. The client asks the server what it already holds, sends the next piece, and repeats. A dropped connection costs the chunk in flight. The pattern is what object storage providers implement for multipart uploads, and it is worth building into the API when the bytes cannot go straight to a bucket.

It also composes cleanly with the asynchronous import path. Once the last chunk lands and the whole-file checksum matches, the request queues a task and returns immediately — the arrangement described in Async Bulk Uploads with Celery. Upload and import stay separate concerns with separate failure modes.

# Runnable Implementation

import hashlib
from pathlib import Path
from typing import Annotated, Any

from fastapi import APIRouter, Header, HTTPException, Request, Response, status

router = APIRouter(prefix="/v1/uploads", tags=["uploads"])

CHUNK_MAX = 32 * 1024 * 1024          # reject anything larger in one request
STAGING = Path("/var/spool/uploads")


@router.post("", status_code=status.HTTP_201_CREATED)
async def create_session(total_bytes: int, sha256: str, filename: str) -> dict[str, Any]:
    """Register the upload before any bytes move."""
    if total_bytes <= 0 or total_bytes > 32 * 1024**3:
        raise HTTPException(422, detail={"error": "total_bytes_out_of_range"})
    upload_id = await register_upload(total_bytes, sha256, filename)
    (STAGING / upload_id).touch()
    return {"upload_id": upload_id, "chunk_max_bytes": CHUNK_MAX}


@router.head("/{upload_id}")
async def upload_status(upload_id: str, response: Response) -> Response:
    """Tell the client where to resume: the highest CONTIGUOUS byte held."""
    session = await load_upload(upload_id)
    if session is None:
        raise HTTPException(404, detail={"error": "unknown_upload"})
    response.headers["Upload-Offset"] = str(session.contiguous_bytes)
    response.headers["Upload-Length"] = str(session.total_bytes)
    return response


@router.put("/{upload_id}")
async def put_chunk(
    upload_id: str,
    request: Request,
    content_range: Annotated[str, Header(alias="Content-Range")],
    chunk_sha256: Annotated[str, Header(alias="X-Chunk-SHA256")],
) -> dict[str, Any]:
    session = await load_upload(upload_id)
    if session is None:
        raise HTTPException(404, detail={"error": "unknown_upload"})

    # "bytes 33554432-67108863/4294967296"
    try:
        span, total = content_range.removeprefix("bytes ").split("/")
        start, end = (int(v) for v in span.split("-"))
    except ValueError:
        raise HTTPException(422, detail={"error": "malformed_content_range"})

    if int(total) != session.total_bytes:
        raise HTTPException(409, detail={"error": "total_size_mismatch"})
    if end - start + 1 > CHUNK_MAX:
        raise HTTPException(413, detail={"error": "chunk_too_large",
                                         "max_bytes": CHUNK_MAX})

    body = await request.body()
    if hashlib.sha256(body).hexdigest() != chunk_sha256:
        # Reject rather than store: a corrupt chunk found now costs one chunk
        raise HTTPException(422, detail={"error": "chunk_checksum_mismatch",
                                         "resume_at": session.contiguous_bytes})

    with open(STAGING / upload_id, "r+b") as fh:
        fh.seek(start)
        fh.write(body)
    session = await record_chunk(upload_id, start, len(body))

    if session.contiguous_bytes < session.total_bytes:
        return {"received": session.contiguous_bytes, "total": session.total_bytes}

    # Complete: verify the whole file, then hand off to the import worker
    digest = await sha256_file(STAGING / upload_id)
    if digest != session.sha256:
        await discard_upload(upload_id)
        raise HTTPException(422, detail={"error": "file_checksum_mismatch"})

    task_id = enqueue_import.delay(upload_id, session.filename).id
    return {"received": session.total_bytes, "total": session.total_bytes,
            "task_id": task_id, "status_url": f"/v1/jobs/{task_id}"}

# Key Parameters & Options

The distinction between “highest contiguous byte” and “highest byte received” matters when chunks arrive out of order. If chunks 1, 2 and 4 are held, the resumption point is the end of chunk 2 — reporting the end of chunk 4 would leave a hole that only the final checksum catches, after the whole file has been transferred.

# Where the time and the risk go

# Cleaning up after abandoned uploads

Every upload session reserves disk for a file that may never arrive. A field laptop that closes its lid mid-transfer leaves a sparse multi-gigabyte file in staging with no client left to finish it, and unless something removes it the staging volume fills within weeks.

Expiry has to consider two clocks. A session that has received no chunk for several hours is abandoned regardless of how recently it was created; a session created days ago is abandoned even if a chunk trickled in this morning. Sweeping on both catches the stalled uploader and the pathologically slow one without cutting off a legitimately slow transfer that is still making progress.

# Gotchas & Failure Modes

• Reporting the highest byte received rather than the highest contiguous one. Out-of-order chunks leave a hole the client never re-sends, and it is only discovered by the final checksum after the whole file has moved.
• No session expiry. Abandoned uploads accumulate full-size sparse files in staging. Expire sessions and sweep the directory on a schedule.
• Parsing the file in the request. Opening a 4 GB GeoPackage inside the final PUT blocks a worker for minutes. Queue the import, as in Handling Async File Uploads for Shapefile Processing.
• A proxy body-size limit below the chunk size. nginx defaults to 1 MB; a 32 MB chunk gets a 413 from the proxy that never reaches the application. Set client_max_body_size to match CHUNK_MAX.
• Trusting the declared total. A client that lies about total_bytes can reserve arbitrary disk. Cap it, and check free space before creating the session.
• Chunk checksums skipped for speed. Detecting corruption only at the end means re-sending gigabytes. A SHA-256 over 32 MB takes about 90 ms — cheap next to the transfer it protects.

# Verification Snippet

UP=$(curl -s -X POST "localhost:8000/v1/uploads?total_bytes=4294967296&sha256=$FULL&filename=survey.gpkg" | jq -r .upload_id)

# Where should we resume?
curl -sI "localhost:8000/v1/uploads/$UP" | grep -i upload-offset
# Upload-Offset: 167772160

# Send the next chunk from that offset
dd if=survey.gpkg bs=1M skip=160 count=32 2>/dev/null > /tmp/chunk
curl -s -X PUT "localhost:8000/v1/uploads/$UP" \
  -H "Content-Range: bytes 167772160-201326591/4294967296" \
  -H "X-Chunk-SHA256: $(sha256sum /tmp/chunk | cut -d' ' -f1)" \
  --data-binary @/tmp/chunk | jq
# {"received":201326592,"total":4294967296}

async def test_resume_after_interruption(client, big_file):
    upload_id = await create_session(client, big_file)
    await send_chunks(client, upload_id, big_file, stop_after=5)

    head = await client.head(f"/v1/uploads/{upload_id}")
    offset = int(head.headers["Upload-Offset"])
    assert offset == 5 * CHUNK_SIZE          # contiguous, not merely received

    result = await send_chunks(client, upload_id, big_file, start_at=offset)
    assert result["task_id"]                  # completed and handed off

# Related

• Async Bulk Uploads with Celery — what happens after the last chunk lands
• Handling Async File Uploads for Shapefile Processing — the import task itself
• Rejecting Invalid Polygons with ST_IsValid — the validation that runs once the file is assembled

← Back to Async Bulk Uploads with Celery