Fix uploader can overwrite

.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

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 }}
+db_uri = {{ clickhouse_url }}
+# minutes an hour's spool directory must go without writes before it is uploaded
+quiet_minutes = 5

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

tests/uploader_e2e/.gitignore

@@ -0,0 +1,2 @@
+work/
+__pycache__/

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/*

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`.

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/*"]}]}))