diff --git a/.github/workflows/test_uploader_e2e.yml b/.github/workflows/test_uploader_e2e.yml
new file mode 100644
index 00000000..7c8acad5
--- /dev/null
+++ b/.github/workflows/test_uploader_e2e.yml
@@ -0,0 +1,75 @@
+# End-to-end test of the measurement uploader (ansible/roles/fastpath/templates/ooni_api_uploader.py):
+# the fastpath image production runs, a versioned S3 (MinIO) and ClickHouse 24.8, with backlogs, failover
+# to the failed-reports bucket, the reuploader, an S3 outage and overlapping runs. See tests/uploader_e2e/README.md
+name: test uploader end to end
+on:
+  pull_request:
+    paths:
+      - "ansible/roles/fastpath/templates/ooni_api_uploader.py"
+      - "ansible/roles/fastpath/templates/api-uploader.conf"
+      - "ansible/host_vars/fastpath.prod.ooni.io/vars.yml"
+      - "tests/uploader_e2e/**"
+      - ".github/workflows/test_uploader_e2e.yml"
+  push:
+    branches: [main]
+    paths:
+      - "ansible/roles/fastpath/templates/ooni_api_uploader.py"
+      - "ansible/roles/fastpath/templates/api-uploader.conf"
+      - "ansible/host_vars/fastpath.prod.ooni.io/vars.yml"
+      - "tests/uploader_e2e/**"
+      - ".github/workflows/test_uploader_e2e.yml"
+  workflow_dispatch:
+
+jobs:
+  uploader-e2e:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 45
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Check out ooni/backend for the reuploader
+        uses: actions/checkout@v4
+        with:
+          repository: ooni/backend
+          path: backend
+          fetch-depth: 0
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
+      - name: Install the test's Python dependencies
+        run: pip install boto3 requests clickhouse-driver
+
+      - name: Build the uploader's runtime (Ubuntu 22.04 packages, as on the fastpath hosts)
+        run: docker build -t ooni-uploader-e2e -f tests/uploader_e2e/Dockerfile.uploader tests/uploader_e2e
+
+      - name: Pull the fastpath image production runs
+        run: |
+          image=$(sed -n 's/^fastpath_image: *"\(.*\)"/\1/p' ansible/host_vars/fastpath.prod.ooni.io/vars.yml)
+          docker pull "$image"
+          echo "FASTPATH_IMAGE=$image" >> "$GITHUB_ENV"
+
+      - name: Get the reuploader at master and at the version production runs
+        run: |
+          cp backend/reuploader/src/reuploader/main.py "$RUNNER_TEMP/reuploader_new.py"
+          prod_tag=$(sed -n 's/.*default_docker_image_url *= *"ooni\/reuploader:\([^"]*\)".*/\1/p' tf/environments/prod/main.tf)
+          # the image tag's commit can be on a squash-merged PR's branch only: fetch the PR heads too
+          git -C backend fetch --quiet --no-tags origin "+refs/pull/*/head:refs/remotes/origin/pr/*"
+          git -C backend show "${prod_tag##*-}:reuploader/src/reuploader/main.py" > "$RUNNER_TEMP/reuploader_prod.py"
+          echo "REUPLOADER_NEW=$RUNNER_TEMP/reuploader_new.py" >> "$GITHUB_ENV"
+          echo "REUPLOADER_OLD=$RUNNER_TEMP/reuploader_prod.py" >> "$GITHUB_ENV"
+          echo "reuploader in production: $prod_tag"
+
+      - name: Run the end-to-end test
+        run: tests/uploader_e2e/scenario.sh ci
+
+      - name: Keep the logs
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: uploader-e2e-logs
+          path: |
+            tests/uploader_e2e/work/logs
+            tests/uploader_e2e/work/result-*.json
+            tests/uploader_e2e/work/ledger.tsv
diff --git a/ansible/roles/fastpath/templates/api-uploader.conf b/ansible/roles/fastpath/templates/api-uploader.conf
index 82b8c26e..02defffe 100644
--- a/ansible/roles/fastpath/templates/api-uploader.conf
+++ b/ansible/roles/fastpath/templates/api-uploader.conf
@@ -7,4 +7,6 @@ aws_secret_access_key = {{ s3_ooni_open_data_access_key }}
 bucket_name = {{ bucket_name }}
 msmt_spool_dir = /var/lib/ooniapi/measurements
 collector_id = {{ collector_id }}
-db_uri = {{ clickhouse_url }}
\ No newline at end of file
+db_uri = {{ clickhouse_url }}
+# minutes an hour's spool directory must go without writes before it is uploaded
+quiet_minutes = 5
\ No newline at end of file
diff --git a/ansible/roles/fastpath/templates/ooni_api_uploader.py b/ansible/roles/fastpath/templates/ooni_api_uploader.py
index 8c9a5255..039c30fe 100644
--- a/ansible/roles/fastpath/templates/ooni_api_uploader.py
+++ b/ansible/roles/fastpath/templates/ooni_api_uploader.py
@@ -9,9 +9,12 @@
 from pathlib import PosixPath as PP
 from datetime import datetime, timedelta
 from typing import List, Dict
+import fcntl
 import gzip
 import logging
+import re
 import tarfile
+import time
 import yaml
 
 import ujson
@@ -76,6 +79,72 @@ def upload_to_s3(s3, bucket_name: str, tarf: PP, s3path: str) -> None:
     obj.put(Body=tarf.read_bytes())
 
 
+def first_free_can_index(s3, bucket_name: str, s3dir: str, basename: str) -> int:
+    """Index of the first can of this run for an hour, cc and test name.
+
+    Measurements for an hour can reach the spool after the hour has been
+    uploaded (a fastpath backlog, or the reuploader re-sending old
+    measurements, which keep their measurement_uid and so their hour). The
+    next run then finds the hour's directory again. Numbering its cans from 0
+    would overwrite the cans already on S3 for that hour, and the jsonl rows
+    pointing into them, so continue after the highest index already there.
+
+    s3dir is "raw/<YYYYMMDD>/<HH>/<cc>/<testname>/", basename
+    "<YYYYMMDDHH>_<cc>_<testname>.<identity>"
+    """
+    pattern = re.compile(re.escape(basename) + r"\.(\d+)\.(tar|jsonl)\.gz$")
+    highest = -1
+    paginator = s3.meta.client.get_paginator("list_objects_v2")
+    for page in paginator.paginate(Bucket=bucket_name, Prefix=s3dir):
+        for obj in page.get("Contents", []):
+            m = pattern.match(obj["Key"][len(s3dir) :])
+            if m:
+                highest = max(highest, int(m.group(1)))
+    return highest + 1
+
+
+def s3_key_exists(s3, bucket_name: str, s3path: str) -> bool:
+    try:
+        s3.meta.client.head_object(Bucket=bucket_name, Key=s3path)
+        return True
+    except s3.meta.client.exceptions.ClientError as e:
+        if e.response["Error"]["Code"] in ("404", "NoSuchKey", "NotFound"):
+            return False
+        raise
+
+
+def remove_hourdir_if_done(
+    hourdir: PP, tmp_max_age_s: int = 3600, empty_max_age_s: int = 86400
+) -> None:
+    """Clean up an hour's spool directory after its upload.
+
+    The fastpath can still write to it: a .post that arrived after the last
+    can was filled, or a .post.tmp being written, belongs to the next run.
+    Only stale temporary files, left by an interrupted write, are deleted.
+
+    The directory itself is removed only once it has been empty and without
+    writes for empty_max_age_s. The fastpath creates it (mkdir exist_ok) and
+    then writes into it: removing it right after an upload, when late writes
+    for the hour are most likely, could land between the two and make the
+    fastpath's write fail.
+    """
+    now = time.time()
+    for f in hourdir.iterdir():
+        if f.name.endswith(".tmp") and now - f.stat().st_mtime > tmp_max_age_s:
+            log.info(f"Deleting stale {f}")
+            f.unlink()
+    if any(hourdir.iterdir()):
+        log.info(f"Keeping {hourdir}: measurements arrived during the upload")
+        metrics.incr("hourdir_kept")
+        return
+    if now - hourdir.stat().st_mtime < empty_max_age_s:
+        return
+    try:
+        hourdir.rmdir()
+    except OSError:
+        pass  # written to since the check: the next run handles it
+
+
 @metrics.timer("fill_postcan")
 def fill_postcan(hourdir: PP, postcanf: PP) -> List[PP]:
     msmt_files = sorted(f for f in hourdir.iterdir() if f.suffix == ".post")
@@ -160,6 +229,17 @@ def main():
     log.info(f"Uploader {collector_id} starting")
     assert collector_id, "collector_id is not set"
     log.info(f"Using bucket {bucket_name} and spool {spooldir}")
+    # minutes an hour's spool directory must go without writes before upload
+    quiet_s = 60 * int(conf.get("quiet_minutes", "5"))
+
+    # A run started by hand while the timer's run is in progress would pick
+    # the same can indexes. The timer itself never overlaps (Type=oneshot).
+    lockf = open(spooldir / "uploader.lock", "w")
+    try:
+        fcntl.flock(lockf, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except BlockingIOError:
+        log.info("Another uploader is running, exiting")
+        return
 
     s3 = create_s3_client(conf)
 
@@ -181,43 +261,67 @@ def main():
         if hourdir_time > threshold:
             log.info(f"Stopping before {hourdir_time}")
             break
+        # Uploaded hours stay as empty directories for a day (see
+        # remove_hourdir_if_done): clean those up without calling S3
+        if not any(f.suffix == ".post" for f in hourdir.iterdir()):
+            remove_hourdir_if_done(hourdir)
+            continue
+        # The hour being over does not mean its measurements have all been
+        # written: a fastpath backlog writes them later, and the reuploader
+        # writes into hours long past. Each write (.tmp, then rename to .post)
+        # updates the directory's mtime: wait until it has been quiet.
+        since_write = time.time() - hourdir.stat().st_mtime
+        if since_write < quiet_s:
+            log.info(f"Skipping {hourdir.name}: written {int(since_write)} s ago, waiting for {quiet_s} s without writes")
+            metrics.incr("hourdir_busy")
+            continue
 
         log.info(f"Processing {hourdir}")
-        # Split msmts across multiple postcans and jsonl files
-        can_cnt = 0
+        s3dir = f"raw/{tstamp[:8]}/{tstamp[8:10]}/{cc}/{testname}/"
+        # Split msmts across multiple postcans and jsonl files, numbered after
+        # any cans already uploaded for this hour
+        can_cnt = first_free_can_index(s3, bucket_name, s3dir, f"{hourdir.name}.{identity}")
+        if can_cnt:
+            log.warning(f"{hourdir.name} was already uploaded: new cans start at {can_cnt}")
+            metrics.incr("late_hour_count")
         while True:
-            # Compress raw POSTs into a tar.gz postcan
             postcanf = hourdir.with_suffix(f".{identity}.{can_cnt}.tar.gz")
             jsonlf = hourdir.with_suffix(f".{identity}.{can_cnt}.jsonl.gz")
+            postcan_s3path = s3dir + postcanf.name
+            jsonl_s3path = s3dir + jsonlf.name
+            # Never replace an object: a replaced can loses its measurements
+            # and leaves their jsonl rows pointing at other measurements
+            if s3_key_exists(s3, bucket_name, postcan_s3path) or s3_key_exists(
+                s3, bucket_name, jsonl_s3path
+            ):
+                log.error(f"{postcan_s3path} or its jsonl already exists, skipping index {can_cnt}")
+                metrics.incr("can_index_taken")
+                can_cnt += 1
+                continue
+            # Compress raw POSTs into a tar.gz postcan
             msmfiles = fill_postcan(hourdir, postcanf)
             if len(msmfiles) == 0:
                 break
-            # Also create jsonl file and delete msmt POSTs
+            # Also create the jsonl file
             lookup_list = fill_jsonl(msmfiles, jsonlf)
-            delete_msmt_posts(msmfiles)
-
-            # Upload current postcan to S3
-            postcan_s3path = (
-                f"raw/{tstamp[:8]}/{tstamp[8:10]}/{cc}/{testname}/{postcanf.name}"
-            )
-            jsonl_s3path = (
-                f"raw/{tstamp[:8]}/{tstamp[8:10]}/{cc}/{testname}/{jsonlf.name}"
-            )
-
-            upload_to_s3(s3, bucket_name, postcanf, postcan_s3path)
-            upload_to_s3(s3, bucket_name, jsonlf, jsonl_s3path)
-            update_db_table(db_conn, lookup_list, jsonl_s3path)
 
-            postcanf.unlink()
-            jsonlf.unlink()
+            # Upload the postcan and jsonl to S3 and record where each
+            # measurement is. Delete the POSTs only once all of that worked:
+            # if it fails they stay in the spool and the next run uploads them
+            # under the next free index
+            try:
+                upload_to_s3(s3, bucket_name, postcanf, postcan_s3path)
+                upload_to_s3(s3, bucket_name, jsonlf, jsonl_s3path)
+                update_db_table(db_conn, lookup_list, jsonl_s3path)
+            finally:
+                postcanf.unlink()
+                jsonlf.unlink()
+            delete_msmt_posts(msmfiles)
 
             can_cnt += 1
             metrics.incr("postcan_count")
 
-        # Delete whole hourly directory
-        for f in sorted(hourdir.iterdir()):
-            f.unlink()
-        hourdir.rmdir()
+        remove_hourdir_if_done(hourdir)
 
     log.info("Exiting")
 
diff --git a/tests/uploader_e2e/.gitignore b/tests/uploader_e2e/.gitignore
new file mode 100644
index 00000000..e958bd44
--- /dev/null
+++ b/tests/uploader_e2e/.gitignore
@@ -0,0 +1,2 @@
+work/
+__pycache__/
diff --git a/tests/uploader_e2e/Dockerfile.uploader b/tests/uploader_e2e/Dockerfile.uploader
new file mode 100644
index 00000000..b9f47061
--- /dev/null
+++ b/tests/uploader_e2e/Dockerfile.uploader
@@ -0,0 +1,6 @@
+# The uploader's runtime on the fastpath hosts: Ubuntu 22.04 (tf/modules/ec2) with the apt packages
+# ansible/roles/fastpath/tasks/main.yml installs for it
+FROM docker.io/library/ubuntu:22.04
+RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
+      python3 python3-statsd python3-boto3 python3-clickhouse-driver python3-ujson python3-systemd python3-yaml \
+    && rm -rf /var/lib/apt/lists/*
diff --git a/tests/uploader_e2e/README.md b/tests/uploader_e2e/README.md
new file mode 100644
index 00000000..c0e13c9e
--- /dev/null
+++ b/tests/uploader_e2e/README.md
@@ -0,0 +1,39 @@
+# Measurement uploader end-to-end test
+
+Runs `ansible/roles/fastpath/templates/ooni_api_uploader.py` against the real pieces it works with and checks
+that every measurement can still be found the way Explorer finds it:
+
+- the fastpath image production runs (`fastpath_image` in `ansible/host_vars/fastpath.prod.ooni.io/vars.yml`),
+  receiving measurements over HTTP and writing them to its spool (`--write-to-disk`);
+- MinIO with versioning on, standing in for `ooni-data-eu-fra` (raw cans) and the failed-reports bucket;
+- ClickHouse 24.8.6.70 with the `jsonl` and `fastpath` tables;
+- the uploader in its production runtime (Ubuntu 22.04 and its apt packages, `Dockerfile.uploader`);
+- the reuploader from ooni/backend, at master and at the version production runs.
+
+`poster.py` submits real measurements (one public can from `ooni-data-eu-fra`, downloaded on first run) the
+way the ooniprobe service does: it mints the measurement_uid from the submission time, POSTs to the fastpath,
+and writes to the failed-reports bucket when that fails. `check.py` then looks every measurement up with the
+same query and fetch as oonimeasurements (`jsonl` row -> `s3path`, `linenum` -> body) and compares the body.
+
+## Scenarios
+1. A normal hour.
+2. Measurements written to the spool after their hour was uploaded (fastpath workers suspended, then
+   resumed), as on 2026-07-14.
+3. A full fastpath queue: requests time out, measurements go to the failed-reports bucket.
+4. The reuploader re-sending them into hours already uploaded: production's version (never deletes) twice,
+   then master's.
+5. S3 unreachable during an upload, and two uploader runs at once.
+
+The run fails if any measurement is not found or comes back with another measurement's body, if any S3
+object was replaced, or if anything is left in the spool or the failed-reports bucket.
+
+## Running it
+```
+cd tests/uploader_e2e
+docker build -t ooni-uploader-e2e -f Dockerfile.uploader .
+pip install boto3 requests clickhouse-driver
+REUPLOADER_NEW=../../../backend/reuploader/src/reuploader/main.py ./scenario.sh local
+```
+About 20 minutes. `DOCKER=podman` works too; `UPLOADER=<file>` tests another version of the uploader
+(the version before 2026-10 fails it: replaced cans, measurements found with another's body, measurements
+lost when S3 is unreachable). Logs and results are in `work/`. CI: `.github/workflows/test_uploader_e2e.yml`.
diff --git a/tests/uploader_e2e/buckets.py b/tests/uploader_e2e/buckets.py
new file mode 100644
index 00000000..e62b5603
--- /dev/null
+++ b/tests/uploader_e2e/buckets.py
@@ -0,0 +1,16 @@
+"""Create the raw bucket (versioned, publicly readable, like ooni-data-eu-fra) and the failed-reports bucket."""
+import json, time
+import boto3
+s3 = boto3.client("s3", endpoint_url="http://127.0.0.1:19900", aws_access_key_id="e2eadmin",
+                  aws_secret_access_key="e2esecret123", region_name="us-east-1")
+for i in range(60):
+    try:
+        s3.list_buckets(); break
+    except Exception:
+        time.sleep(1)
+for b in ("e2e-raw", "e2e-failed"):
+    s3.create_bucket(Bucket=b)
+    s3.put_bucket_versioning(Bucket=b, VersioningConfiguration={"Status": "Enabled"})
+s3.put_bucket_policy(Bucket="e2e-raw", Policy=json.dumps({"Version": "2012-10-17", "Statement": [{
+    "Effect": "Allow", "Principal": "*", "Action": ["s3:GetObject", "s3:ListBucket"],
+    "Resource": ["arn:aws:s3:::e2e-raw", "arn:aws:s3:::e2e-raw/*"]}]}))
diff --git a/tests/uploader_e2e/check.py b/tests/uploader_e2e/check.py
new file mode 100644
index 00000000..e907e5aa
--- /dev/null
+++ b/tests/uploader_e2e/check.py
@@ -0,0 +1,66 @@
+"""Check that every submitted measurement can be found the way Explorer finds it, and that no S3 object
+was ever replaced.
+
+The lookup copies ooni/backend oonimeasurements (routers/v1/measurements.py): measurement_uid_to_s3path_linenum
+(the same SQL on the jsonl table) and _fetch_jsonl_measurement_body_from_s3 (line `linenum` of the gzipped
+jsonl at s3path). Exit status 1 on any failure.  usage: check.py LABEL"""
+import collections, gzip, io, json, os, sys
+import boto3, requests
+from clickhouse_driver import Client
+
+work = os.environ.get("WORK") or os.path.join(os.path.dirname(os.path.abspath(__file__)), "work")
+label = sys.argv[1] if len(sys.argv) > 1 else "run"
+db = Client.from_url("clickhouse://test:test@127.0.0.1:19901/default")
+s3 = boto3.client("s3", endpoint_url="http://127.0.0.1:19900", aws_access_key_id="e2eadmin",
+                  aws_secret_access_key="e2esecret123", region_name="us-east-1")
+LOOKUP = """SELECT s3path, linenum FROM jsonl
+    PREWHERE (report_id, input) IN (
+        SELECT report_id, input FROM fastpath WHERE measurement_uid = %(uid)s
+    )
+    LIMIT 1"""
+
+def body(s3path, linenum):
+    r = requests.get(f"http://127.0.0.1:19900/e2e-raw/{s3path}", timeout=30)
+    r.raise_for_status()
+    for n, line in enumerate(gzip.GzipFile(fileobj=io.BytesIO(r.content))):
+        if n == linenum:
+            return json.loads(line)
+    raise LookupError(f"{s3path} has no line {linenum}")
+
+uids = {}
+for line in open(os.path.join(work, "ledger.tsv")):
+    uid, outcome = line.split()
+    uids.setdefault(uid, outcome)
+res, bad = collections.Counter(), collections.defaultdict(list)
+for uid in sorted(uids):
+    rows = db.execute(LOOKUP, {"uid": uid})
+    if not rows:
+        res["not found (no jsonl row)"] += 1; bad["not_found"].append(uid); continue
+    s3path, linenum = rows[0]
+    try:
+        got = body(s3path, linenum).get("e2e_uid")
+    except Exception as e:
+        res["jsonl row points at a missing file or line"] += 1; bad["missing"].append((uid, s3path, linenum, repr(e)[:80])); continue
+    if got == uid:
+        res["ok"] += 1
+    else:
+        res["WRONG BODY (another measurement)"] += 1; bad["wrong_body"].append((uid, s3path, linenum, got))
+
+replaced = sorted({v["Key"] for p in s3.get_paginator("list_object_versions").paginate(Bucket="e2e-raw")
+                   for v in p.get("Versions", []) if not v["IsLatest"]})
+spool = os.path.join(work, "spool", "incoming")
+spool_left = [os.path.join(d, f) for d, _, fs in os.walk(spool) for f in fs if f.endswith(".post")]
+failed_left = sum(len(p.get("Contents", [])) for p in s3.get_paginator("list_objects_v2").paginate(Bucket="e2e-failed"))
+cans = sum(1 for p in s3.get_paginator("list_objects_v2").paginate(Bucket="e2e-raw")
+           for o in p.get("Contents", []) if o["Key"].endswith(".jsonl.gz"))
+
+print(f"== {label}: {len(uids)} measurements submitted ({dict(collections.Counter(uids.values()))}), {cans} jsonl cans")
+for k, v in res.most_common():
+    print(f"   {k}: {v}")
+print(f"   S3 objects replaced (have older versions): {len(replaced)} {replaced[:4]}")
+print(f"   .post left in the spool: {len(spool_left)}   objects left in the failed bucket: {failed_left}")
+json.dump({"label": label, "results": res, "replaced": replaced, "spool_left": spool_left, "failed_left": failed_left,
+           "bad": bad}, open(os.path.join(work, f"result-{label}.json"), "w"), indent=1, default=str)
+passed = res["ok"] == len(uids) and not replaced and not spool_left and failed_left == 0
+print("   RESULT:", "PASS" if passed else "FAIL")
+sys.exit(0 if passed else 1)
diff --git a/tests/uploader_e2e/conf/api-uploader.conf b/tests/uploader_e2e/conf/api-uploader.conf
new file mode 100644
index 00000000..a3a891cc
--- /dev/null
+++ b/tests/uploader_e2e/conf/api-uploader.conf
@@ -0,0 +1,8 @@
+[DEFAULT]
+aws_access_key_id = e2eadmin
+aws_secret_access_key = e2esecret123
+bucket_name = e2e-raw
+msmt_spool_dir = /var/lib/ooniapi/measurements
+collector_id = 1
+db_uri = clickhouse://test:test@uploader-e2e-ch:9000/default
+quiet_minutes = 1
diff --git a/tests/uploader_e2e/conf/fastpath.conf b/tests/uploader_e2e/conf/fastpath.conf
new file mode 100644
index 00000000..aa242c7c
--- /dev/null
+++ b/tests/uploader_e2e/conf/fastpath.conf
@@ -0,0 +1,7 @@
+[DEFAULT]
+collectors = localhost
+db_uri = postgresql://readonly@localhost/metadb
+s3_access_key =
+s3_secret_key =
+clickhouse_url = clickhouse://test:test@uploader-e2e-ch:9000/default
+msmt_spool_dir = /var/lib/ooniapi/measurements
diff --git a/tests/uploader_e2e/env.sh b/tests/uploader_e2e/env.sh
new file mode 100755
index 00000000..00567912
--- /dev/null
+++ b/tests/uploader_e2e/env.sh
@@ -0,0 +1,31 @@
+#!/bin/bash
+# Test environment for the uploader: MinIO (versioned buckets, like ooni-data-eu-fra), ClickHouse 24.8.6.70
+# (production's version) and the fastpath image production runs.
+# usage: env.sh up|down     needs FASTPATH_IMAGE; DOCKER (docker|podman), WORKERS, QUEUE optional
+set -eu
+T=$(cd "$(dirname "$0")" && pwd); WORK=${WORK:-$T/work}; DOCKER=${DOCKER:-docker}; NET=uploader-e2e
+MINIO_IMAGE=${MINIO_IMAGE:-docker.io/pgsty/minio:RELEASE.2026-08-04T00-00-00Z}
+CH_IMAGE=${CH_IMAGE:-docker.io/clickhouse/clickhouse-server:24.8.6.70}
+ch() { $DOCKER exec -i uploader-e2e-ch clickhouse-client -u test --password test "$@"; }
+up() {
+  : "${FASTPATH_IMAGE:?set FASTPATH_IMAGE, e.g. ooni/fastpath:<tag> from ansible/host_vars}"
+  mkdir -p $WORK/spool && chmod 0777 $WORK/spool
+  $DOCKER network inspect $NET >/dev/null 2>&1 || $DOCKER network create $NET >/dev/null
+  $DOCKER run -d --name uploader-e2e-minio --network $NET -p 127.0.0.1:19900:9000 \
+    -e MINIO_ROOT_USER=e2eadmin -e MINIO_ROOT_PASSWORD=e2esecret123 $MINIO_IMAGE server /data >/dev/null
+  $DOCKER run -d --name uploader-e2e-ch --network $NET -p 127.0.0.1:19901:9000 \
+    -e CLICKHOUSE_USER=test -e CLICKHOUSE_PASSWORD=test $CH_IMAGE >/dev/null
+  for i in $(seq 90); do ch -q "SELECT 1" >/dev/null 2>&1 && break; sleep 1; done
+  for f in $T/sql/*.sql; do ch --multiquery < $f; done
+  python3 $T/buckets.py
+  $DOCKER run -d --name uploader-e2e-fastpath --network $NET -p 127.0.0.1:19903:8472 --user root \
+    -e FASTPATH_QUEUE_PUT_TIMEOUT=-1 -e FASTPATH_NUM_WORKERS=${WORKERS:-2} -e FASTPATH_QUEUE_SIZE=${QUEUE:-100} \
+    -v $T/conf/fastpath.conf:/etc/ooni/fastpath.conf:ro,z -v $WORK/spool:/var/lib/ooniapi/measurements:z \
+    $FASTPATH_IMAGE python /home/fastpath/app/run_fastpath --write-to-disk --stdout >/dev/null
+  for i in $(seq 60); do curl -s -o /dev/null http://127.0.0.1:19903/ && break; sleep 1; done
+}
+down() {
+  $DOCKER rm -f uploader-e2e-fastpath uploader-e2e-minio uploader-e2e-ch >/dev/null 2>&1 || true
+  $DOCKER network rm $NET >/dev/null 2>&1 || true
+}
+case ${1:-} in up) up;; down) down;; *) echo "usage: $0 up|down"; exit 1;; esac
diff --git a/tests/uploader_e2e/poster.py b/tests/uploader_e2e/poster.py
new file mode 100644
index 00000000..383e32ff
--- /dev/null
+++ b/tests/uploader_e2e/poster.py
@@ -0,0 +1,61 @@
+"""Submit measurements the way the ooniprobe service does: mint a measurement_uid from the submission time,
+POST to the fastpath, and on failure put the measurement in the failed-reports bucket. Uses real
+measurements from the cans in $WORK/cans, each tagged with its uid so the check can tell them apart.
+usage: poster.py HOURS_AGO COUNT [--timeout S] [--threads N]     appends uid<TAB>outcome to $WORK/ledger.tsv"""
+import argparse, concurrent.futures as cf, datetime as dt, glob, hashlib, json, os, random, tarfile, threading
+import boto3, requests
+
+ap = argparse.ArgumentParser()
+ap.add_argument("hours_ago", type=int)
+ap.add_argument("count", type=int)
+ap.add_argument("--timeout", type=float, default=10)
+ap.add_argument("--threads", type=int, default=4)
+a = ap.parse_args()
+work = os.environ.get("WORK") or os.path.join(os.path.dirname(os.path.abspath(__file__)), "work")
+
+# take the next COUNT measurements from the cans, so every submission is a different measurement
+offset_f = os.path.join(work, "poster.offset")
+offset = int(open(offset_f).read()) if os.path.exists(offset_f) else 0
+posts = []
+for fn in sorted(glob.glob(os.path.join(work, "cans", "*.tar.gz"))):
+    with tarfile.open(fn) as t:
+        posts += [t.extractfile(m).read() for m in t.getmembers() if m.name.endswith(".post")]
+batch = posts[offset:offset + a.count]
+assert len(batch) == a.count, f"not enough measurements in {work}/cans"
+open(offset_f, "w").write(str(offset + len(batch)))
+
+s3 = boto3.client("s3", endpoint_url="http://127.0.0.1:19900", aws_access_key_id="e2eadmin",
+                  aws_secret_access_key="e2esecret123", region_name="us-east-1")
+hour = (dt.datetime.now(dt.timezone.utc) - dt.timedelta(hours=a.hours_ago)).replace(minute=0, second=0, microsecond=0)
+lock = threading.Lock()
+ledger = open(os.path.join(work, "ledger.tsv"), "a")
+
+def submit(raw):
+    post = json.loads(raw)
+    c = post["content"]
+    t = hour + dt.timedelta(seconds=random.randrange(3600), microseconds=random.randrange(10**6))
+    ts = t.strftime("%Y%m%d%H%M%S.%f")
+    cc = c.get("probe_cc", "ZZ")
+    test = (c.get("test_name") or "x").replace("_", "")
+    h = hashlib.sha512(raw + ts.encode()).hexdigest()[:16]
+    uid = f"{ts}_{cc}_{test}_{h}"
+    c["e2e_uid"] = uid
+    c["report_id"] = f"e2e-{uid}"  # unique (report_id, input): the API looks measurements up by those
+    data = json.dumps(post).encode()
+    try:
+        r = requests.post(f"http://127.0.0.1:19903/{uid}", data=data, timeout=a.timeout)
+        r.raise_for_status()
+        outcome = "fastpath"
+    except Exception:
+        key = f"postcans/{ts[:10]}/{ts[:10]}_{cc}_{test}/{uid}.post"
+        s3.put_object(Bucket="e2e-failed", Key=key, Body=data)
+        outcome = "failed_bucket"
+    with lock:
+        ledger.write(f"{uid}\t{outcome}\n")
+        ledger.flush()
+    return outcome
+
+with cf.ThreadPoolExecutor(a.threads) as ex:
+    res = list(ex.map(submit, batch))
+print(f"posted {len(res)} for hour {hour:%Y%m%d%H}: {res.count('fastpath')} to fastpath, "
+      f"{res.count('failed_bucket')} to the failed bucket")
diff --git a/tests/uploader_e2e/reuploader.sh b/tests/uploader_e2e/reuploader.sh
new file mode 100755
index 00000000..3e1557cb
--- /dev/null
+++ b/tests/uploader_e2e/reuploader.sh
@@ -0,0 +1,7 @@
+#!/bin/bash
+# Run a reuploader version (ooni/backend reuploader/src/reuploader/main.py) once against the failed bucket.
+T=$(cd "$(dirname "$0")" && pwd); DOCKER=${DOCKER:-docker}
+$DOCKER run --rm --network uploader-e2e -v $1:/r.py:ro,z \
+  -e S3_BUCKET_NAME=e2e-failed -e FASTPATH_API=http://uploader-e2e-fastpath:8472 -e BATCH_SIZE=1000 -e LOG_LEVEL=INFO \
+  -e AWS_ENDPOINT_URL=http://uploader-e2e-minio:9000 -e AWS_ACCESS_KEY_ID=e2eadmin -e AWS_SECRET_ACCESS_KEY=e2esecret123 \
+  -e AWS_REGION=us-east-1 $FASTPATH_IMAGE python /r.py
diff --git a/tests/uploader_e2e/run_uploader.py b/tests/uploader_e2e/run_uploader.py
new file mode 100644
index 00000000..7c502ab2
--- /dev/null
+++ b/tests/uploader_e2e/run_uploader.py
@@ -0,0 +1,21 @@
+"""Run an ooni_api_uploader.py once against the test environment, without changing it: log to stdout
+(no journald in the container) and point its S3 client at the test endpoint (the hosts' boto3 1.20
+ignores AWS_ENDPOINT_URL). usage: run_uploader.py UPLOADER_FILE"""
+import importlib.util, logging, os, sys
+import boto3
+
+spec = importlib.util.spec_from_file_location("uploader", sys.argv[1])
+up = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(up)
+for h in list(up.log.handlers):
+    up.log.removeHandler(h)
+h = logging.StreamHandler(sys.stdout)
+h.setFormatter(logging.Formatter("%(asctime)s uploader %(levelname)s %(message)s"))
+up.log.addHandler(h)
+
+def create_s3_client(conf):
+    session = boto3.Session(aws_access_key_id=conf.get("aws_access_key_id"), aws_secret_access_key=conf.get("aws_secret_access_key"))
+    return session.resource("s3", endpoint_url=os.environ["S3_ENDPOINT"], region_name="us-east-1")
+
+up.create_s3_client = create_s3_client
+up.main()
diff --git a/tests/uploader_e2e/scenario.sh b/tests/uploader_e2e/scenario.sh
new file mode 100755
index 00000000..ff967397
--- /dev/null
+++ b/tests/uploader_e2e/scenario.sh
@@ -0,0 +1,63 @@
+#!/bin/bash
+# End-to-end test of the measurement uploader against the fastpath, a versioned S3 and ClickHouse.
+# usage: scenario.sh LABEL
+#   UPLOADER        the ooni_api_uploader.py to test (default: ansible/roles/fastpath/templates/ooni_api_uploader.py)
+#   FASTPATH_IMAGE  e.g. ooni/fastpath:<tag> (default: the tag in ansible/host_vars/fastpath.prod.ooni.io/vars.yml)
+#   REUPLOADER_NEW  ooni/backend reuploader/src/reuploader/main.py at master (scenario 4; skipped if unset)
+#   REUPLOADER_OLD  the same file at the version production runs (optional)
+#   DOCKER          docker (default) or podman
+# The uploader runs every 20 s (the hourly timer, sped up) with quiet_minutes = 1. Fastpath workers are
+# suspended (SIGSTOP) to make backlogs. Exits 1 if any measurement is not found the way Explorer finds it,
+# if any S3 object was replaced, or if anything is left in the spool or the failed bucket.
+set -u
+python3 -c "import boto3, requests, clickhouse_driver" 2>/dev/null || { echo "needs: pip install boto3 requests clickhouse-driver"; exit 2; }
+T=$(cd "$(dirname "$0")" && pwd); REPO=$(cd $T/../.. && pwd); LABEL=${1:-run}
+export WORK=${WORK:-$T/work}; export DOCKER=${DOCKER:-docker}
+export UPLOADER=${UPLOADER:-$REPO/ansible/roles/fastpath/templates/ooni_api_uploader.py}
+export FASTPATH_IMAGE=${FASTPATH_IMAGE:-$(sed -n 's/^fastpath_image: *"\(.*\)"/\1/p' $REPO/ansible/host_vars/fastpath.prod.ooni.io/vars.yml)}
+L=$WORK/logs/$LABEL; mkdir -p $L $WORK/cans; WORKERS=2; export WORKERS
+CAN=raw/20261008/00/IT/webconnectivity/2026100800_IT_webconnectivity.n1.2.tar.gz   # public, 663 measurements
+[ -s $WORK/cans/test.tar.gz ] || curl -sf -o $WORK/cans/test.tar.gz "https://ooni-data-eu-fra.s3.eu-central-1.amazonaws.com/$CAN"
+workers() { $DOCKER exec uploader-e2e-fastpath sh -c "for p in /proc/[0-9]*/status; do awk '/^Pid/{p=\$2} /^PPid/{if (\$2==1) print p}' \$p; done" | sort -n | head -$WORKERS; }
+stop_workers() { for p in $(workers); do $DOCKER exec uploader-e2e-fastpath sh -c "kill -STOP $p"; done; echo "   fastpath workers suspended"; }
+cont_workers() { for p in $(workers); do $DOCKER exec uploader-e2e-fastpath sh -c "kill -CONT $p"; done; echo "   fastpath workers resumed"; }
+step() { echo "== $(date -u +%T) $*"; }
+post() { python3 $T/poster.py "$@"; }
+
+echo "uploader: $UPLOADER"; echo "fastpath: $FASTPATH_IMAGE"
+$T/env.sh down; rm -rf $WORK/spool $WORK/ledger.tsv $WORK/poster.offset $WORK/STOP_LOOP $WORK/PAUSE_LOOP; $T/env.sh up
+(while [ ! -e $WORK/STOP_LOOP ]; do [ -e $WORK/PAUSE_LOOP ] || $T/uploader.sh >> $L/uploader.log 2>&1; sleep 20; done) &
+LOOP=$!
+
+step "S1 normal hour (5 h ago)"; post 5 40; sleep 100
+
+step "S2 measurements written after their hour was uploaded (4 h ago), as on 2026-07-14"
+post 4 30; sleep 100
+stop_workers; post 4 40; sleep 80
+cont_workers; sleep 120
+
+step "S3 full fastpath queue: timeouts, failover to the failed bucket (3 h ago)"
+stop_workers; post 3 140 --timeout 3 --threads 8; sleep 10; cont_workers; sleep 120
+
+step "S4 reuploader re-sending into hours already uploaded"
+if [ -n "${REUPLOADER_OLD:-}" ]; then
+  for i in 1 2; do $T/reuploader.sh $REUPLOADER_OLD > $L/reuploader-old-$i.log 2>&1; echo "   old reuploader run $i: $(grep -c 'Successfully submitted' $L/reuploader-old-$i.log) submitted"; sleep 100; done
+fi
+if [ -n "${REUPLOADER_NEW:-}" ]; then
+  $T/reuploader.sh $REUPLOADER_NEW > $L/reuploader-new.log 2>&1; echo "   reuploader: $(grep -c 'Submitted' $L/reuploader-new.log) submitted, $(grep -c 'Failed to process' $L/reuploader-new.log) failed"; sleep 100
+fi
+
+step "S5 interruptions (2 h ago): S3 unreachable during an upload, two uploaders at once"
+touch $WORK/PAUSE_LOOP; sleep 25; post 2 20; sleep 70
+S3_ENDPOINT=http://uploader-e2e-minio:9999 $T/uploader.sh > $L/uploader-s3down.log 2>&1; echo "   uploader with S3 unreachable: exit $?"
+$T/uploader.sh > $L/uploader-concurrent-a.log 2>&1 & A=$!; $T/uploader.sh > $L/uploader-concurrent-b.log 2>&1 & B=$!; wait $A $B
+echo "   concurrent runs: $(grep -l 'Another uploader is running' $L/uploader-concurrent-*.log 2>/dev/null | wc -l) exited on the lock"
+rm -f $WORK/PAUSE_LOOP
+
+step "final: wait until quiet, last upload, check"
+sleep 90; touch $WORK/STOP_LOOP; wait $LOOP; $T/uploader.sh >> $L/uploader.log 2>&1
+$DOCKER logs uploader-e2e-fastpath > $L/fastpath.log 2>&1
+echo "   fastpath 'Failed to upload measurement' errors: $(grep -c 'Failed to upload measurement' $L/fastpath.log)"
+python3 $T/check.py $LABEL; RC=$?
+[ -n "${KEEP_ENV:-}" ] || $T/env.sh down
+exit $RC
diff --git a/tests/uploader_e2e/sql/fastpath.sql b/tests/uploader_e2e/sql/fastpath.sql
new file mode 100644
index 00000000..a12e712f
--- /dev/null
+++ b/tests/uploader_e2e/sql/fastpath.sql
@@ -0,0 +1,45 @@
+-- production-shaped definition (as in ooni/backend ooniapi/common/fixtures), used by the uploader end-to-end test
+CREATE TABLE default.fastpath
+(
+    `measurement_uid` String,
+    `report_id` String,
+    `input` String,
+    `probe_cc` LowCardinality(String),
+    `probe_asn` Int32,
+    `test_name` LowCardinality(String),
+    `test_start_time` DateTime,
+    `measurement_start_time` DateTime,
+    `filename` String,
+    `scores` String,
+    `platform` String,
+    `anomaly` String,
+    `confirmed` String,
+    `msm_failure` String,
+    `domain` String,
+    `software_name` String,
+    `software_version` String,
+    `control_failure` String,
+    `blocking_general` Float32,
+    `is_ssl_expected` Int8,
+    `page_len` Int32,
+    `page_len_ratio` Float32,
+    `server_cc` String,
+    `server_asn` Int8,
+    `server_as_name` String,
+    `update_time` DateTime64(3) MATERIALIZED now64(),
+    `test_version` String,
+    `architecture` String,
+    `engine_name` LowCardinality(String),
+    `engine_version` String,
+    `test_runtime` Float32,
+    `blocking_type` String,
+    `test_helper_address` LowCardinality(String),
+    `test_helper_type` LowCardinality(String),
+    `ooni_run_link_id` Nullable(UInt64),
+    `is_verified` LowCardinality(String) DEFAULT 'u',
+    INDEX fastpath_rid_idx report_id TYPE minmax GRANULARITY 1,
+    INDEX measurement_uid_idx measurement_uid TYPE minmax GRANULARITY 8
+)
+ENGINE = ReplacingMergeTree(update_time)
+ORDER BY (measurement_start_time, report_id, input, measurement_uid)
+SETTINGS index_granularity = 8192
diff --git a/tests/uploader_e2e/sql/fingerprints_dns.sql b/tests/uploader_e2e/sql/fingerprints_dns.sql
new file mode 100644
index 00000000..ac1516ef
--- /dev/null
+++ b/tests/uploader_e2e/sql/fingerprints_dns.sql
@@ -0,0 +1,17 @@
+-- production-shaped definition (as in ooni/backend ooniapi/common/fixtures), used by the uploader end-to-end test
+CREATE TABLE default.fingerprints_dns
+(
+    `name` String,
+    `scope` Enum8('nat' = 1, 'isp' = 2, 'prod' = 3, 'inst' = 4, 'vbw' = 5, 'fp' = 6),
+    `other_names` String,
+    `location_found` String,
+    `pattern_type` Enum8('full' = 1, 'prefix' = 2, 'contains' = 3, 'regexp' = 4),
+    `pattern` String,
+    `confidence_no_fp` UInt8,
+    `expected_countries` String,
+    `source` String,
+    `exp_url` String,
+    `notes` String
+)
+ENGINE = EmbeddedRocksDB
+PRIMARY KEY name
diff --git a/tests/uploader_e2e/sql/fingerprints_http.sql b/tests/uploader_e2e/sql/fingerprints_http.sql
new file mode 100644
index 00000000..ee4fb1cb
--- /dev/null
+++ b/tests/uploader_e2e/sql/fingerprints_http.sql
@@ -0,0 +1,17 @@
+-- production-shaped definition (as in ooni/backend ooniapi/common/fixtures), used by the uploader end-to-end test
+CREATE TABLE default.fingerprints_http
+(
+    `name` String,
+    `scope` Enum8('nat' = 1, 'isp' = 2, 'prod' = 3, 'inst' = 4, 'vbw' = 5, 'fp' = 6, 'injb' = 7, 'prov' = 8),
+    `other_names` String,
+    `location_found` String,
+    `pattern_type` Enum8('full' = 1, 'prefix' = 2, 'contains' = 3, 'regexp' = 4),
+    `pattern` String,
+    `confidence_no_fp` UInt8,
+    `expected_countries` String,
+    `source` String,
+    `exp_url` String,
+    `notes` String
+)
+ENGINE = EmbeddedRocksDB
+PRIMARY KEY name
diff --git a/tests/uploader_e2e/sql/jsonl.sql b/tests/uploader_e2e/sql/jsonl.sql
new file mode 100644
index 00000000..2e590515
--- /dev/null
+++ b/tests/uploader_e2e/sql/jsonl.sql
@@ -0,0 +1,15 @@
+-- production-shaped definition (as in ooni/backend ooniapi/common/fixtures), used by the uploader end-to-end test
+CREATE TABLE default.jsonl
+(
+    `report_id` String,
+    `input` String,
+    `s3path` String,
+    `linenum` Int32,
+    `measurement_uid` String,
+    `date` Date,
+    `source` String,
+    `update_time` DateTime64(3) MATERIALIZED now64()
+)
+ENGINE = ReplacingMergeTree(update_time)
+ORDER BY (report_id, input, measurement_uid)
+SETTINGS index_granularity = 8192
diff --git a/tests/uploader_e2e/uploader.sh b/tests/uploader_e2e/uploader.sh
new file mode 100755
index 00000000..caee437e
--- /dev/null
+++ b/tests/uploader_e2e/uploader.sh
@@ -0,0 +1,7 @@
+#!/bin/bash
+# Run the uploader under test once. UPLOADER: path to ooni_api_uploader.py; S3_ENDPOINT overrides the endpoint
+T=$(cd "$(dirname "$0")" && pwd); WORK=${WORK:-$T/work}; DOCKER=${DOCKER:-docker}
+$DOCKER run --rm --network uploader-e2e --user root -v $WORK/spool:/var/lib/ooniapi/measurements:z \
+  -v $T/conf/api-uploader.conf:/etc/ooni/api-uploader.conf:ro,z -v $T/run_uploader.py:/w/run_uploader.py:ro,z \
+  -v $UPLOADER:/u.py:ro,z -e S3_ENDPOINT=${S3_ENDPOINT:-http://uploader-e2e-minio:9000} \
+  ${UPLOADER_IMAGE:-ooni-uploader-e2e} python3 /w/run_uploader.py /u.py