From 8a517d4065463b2e3cc3d0760cab0302a2cb2d65 Mon Sep 17 00:00:00 2001
From: Aaron Gibson <aagbsn@extc.org>
Date: Fri, 16 Oct 2026 15:15:07 +0000
Subject: [PATCH 1/3] fastpath uploader: never overwrite cans, wait for an
 hour's writes to stop

Each run numbered an hour's cans from 0 and PUT them unconditionally, and
took any hour that was over even if the fastpath was still writing to it.
When measurements reach the spool after their hour was uploaded (a fastpath
backlog, or the reuploader re-sending old measurements, which keep their
uid and so their hour), the next run wrote the same keys again and replaced
the earlier cans, losing their measurements from the current view and
leaving their jsonl rows pointing at other measurements. Scanning every
version under raw/ in ooni-data-eu-fra for 2026 found 35,131 measurements
that exist only in replaced versions (Feb 23, Mar 11, Apr 10, Apr 11,
Apr 23, Apr 25, Jul 14).

- Number a run's cans after the highest index already on S3 for the hour,
  cc, test and collector, and skip any index whose key exists.
- Upload an hour's directory only once nothing has been written to it for
  quiet_minutes (default 5, in api-uploader.conf).
- Delete the POSTs only after the cans are uploaded and their jsonl rows
  inserted, so a failed upload is retried by the next run instead of
  losing them.
- Keep measurements that arrive during a run for the next one, delete only
  stale .tmp files, and remove an hour's directory only after it has been
  empty for a day: removing it right after an upload could land between the
  fastpath's mkdir and its write.
- Lock the spool so a run started by hand cannot overlap the timer's.
---
 .../fastpath/templates/api-uploader.conf      |   4 +-
 .../fastpath/templates/ooni_api_uploader.py   | 150 +++++++++++++++---
 2 files changed, 130 insertions(+), 24 deletions(-)

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

From d885649e9984fca363d2c2d2046ac3483a20d41e Mon Sep 17 00:00:00 2001
From: Aaron Gibson <aagbsn@extc.org>
Date: Fri, 16 Oct 2026 15:15:07 +0000
Subject: [PATCH 2/3] Add an end-to-end test of the measurement uploader

Runs ooni_api_uploader.py against the fastpath image production runs, a
versioned S3 (MinIO) and ClickHouse 24.8, in the uploader's production
runtime (Ubuntu 22.04 packages). It submits real measurements the way the
ooniprobe service does and creates the situations that replaced cans in
ooni-data-eu-fra: measurements written after their hour was uploaded, a
full fastpath queue failing over to the failed-reports bucket, the
reuploader (master and production's version) re-sending into uploaded
hours, S3 unreachable during an upload, and overlapping runs. Every
measurement is then looked up as oonimeasurements does (jsonl -> s3path,
linenum) and its body compared; the run fails on any missing or wrong
measurement, replaced S3 object, or leftover in the spool or failed bucket.

The uploader before this branch fails it (120 of 270 measurements found,
4 objects replaced); this branch passes.
---
 .github/workflows/test_uploader_e2e.yml      | 73 ++++++++++++++++++++
 tests/uploader_e2e/.gitignore                |  2 +
 tests/uploader_e2e/Dockerfile.uploader       |  6 ++
 tests/uploader_e2e/README.md                 | 39 +++++++++++
 tests/uploader_e2e/buckets.py                | 16 +++++
 tests/uploader_e2e/check.py                  | 66 ++++++++++++++++++
 tests/uploader_e2e/conf/api-uploader.conf    |  8 +++
 tests/uploader_e2e/conf/fastpath.conf        |  7 ++
 tests/uploader_e2e/env.sh                    | 31 +++++++++
 tests/uploader_e2e/poster.py                 | 61 ++++++++++++++++
 tests/uploader_e2e/reuploader.sh             |  7 ++
 tests/uploader_e2e/run_uploader.py           | 21 ++++++
 tests/uploader_e2e/scenario.sh               | 63 +++++++++++++++++
 tests/uploader_e2e/sql/fastpath.sql          | 45 ++++++++++++
 tests/uploader_e2e/sql/fingerprints_dns.sql  | 17 +++++
 tests/uploader_e2e/sql/fingerprints_http.sql | 17 +++++
 tests/uploader_e2e/sql/jsonl.sql             | 15 ++++
 tests/uploader_e2e/uploader.sh               |  7 ++
 18 files changed, 501 insertions(+)
 create mode 100644 .github/workflows/test_uploader_e2e.yml
 create mode 100644 tests/uploader_e2e/.gitignore
 create mode 100644 tests/uploader_e2e/Dockerfile.uploader
 create mode 100644 tests/uploader_e2e/README.md
 create mode 100644 tests/uploader_e2e/buckets.py
 create mode 100644 tests/uploader_e2e/check.py
 create mode 100644 tests/uploader_e2e/conf/api-uploader.conf
 create mode 100644 tests/uploader_e2e/conf/fastpath.conf
 create mode 100755 tests/uploader_e2e/env.sh
 create mode 100644 tests/uploader_e2e/poster.py
 create mode 100755 tests/uploader_e2e/reuploader.sh
 create mode 100644 tests/uploader_e2e/run_uploader.py
 create mode 100755 tests/uploader_e2e/scenario.sh
 create mode 100644 tests/uploader_e2e/sql/fastpath.sql
 create mode 100644 tests/uploader_e2e/sql/fingerprints_dns.sql
 create mode 100644 tests/uploader_e2e/sql/fingerprints_http.sql
 create mode 100644 tests/uploader_e2e/sql/jsonl.sql
 create mode 100755 tests/uploader_e2e/uploader.sh

diff --git a/.github/workflows/test_uploader_e2e.yml b/.github/workflows/test_uploader_e2e.yml
new file mode 100644
index 00000000..f51b9a55
--- /dev/null
+++ b/.github/workflows/test_uploader_e2e.yml
@@ -0,0 +1,73 @@
+# 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)
+          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/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

From b1ec0ed6659ce55a5261eb70a9a0f699b308ff70 Mon Sep 17 00:00:00 2001
From: Aaron Gibson <aagbsn@extc.org>
Date: Mon, 19 Oct 2026 08:12:21 +0000
Subject: [PATCH 3/3] uploader e2e test: fetch ooni/backend's PR heads for the
 reuploader

The workflow reads production's reuploader from ooni/backend at the
commit in its image tag (8b35a38f). That commit is only on #1219's
branch, which was squash-merged, so the checkout did not have it
("fatal: invalid object name '8b35a38f'"). Fetch refs/pull/*/head too.
---
 .github/workflows/test_uploader_e2e.yml | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/.github/workflows/test_uploader_e2e.yml b/.github/workflows/test_uploader_e2e.yml
index f51b9a55..7c8acad5 100644
--- a/.github/workflows/test_uploader_e2e.yml
+++ b/.github/workflows/test_uploader_e2e.yml
@@ -54,6 +54,8 @@ jobs:
         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"