[aagbsn]

The problem

ooni_api_uploader.py numbered each run's cans from 0 (<hour>_<cc>_<test>.n<collector>.0.tar.gz / .jsonl.gz) and PUT them without checking. It also took any hour that was over, even while the fastpath was still writing to it. When measurements reach the spool after their hour was uploaded, the next run writes the same keys again and replaces the earlier cans. That happens with a fastpath backlog, and with the reuploader re-sending old measurements, which keep their uid and so their hour. The replaced cans' measurements disappear from the current view of the bucket, and their jsonl rows point at other measurements' bodies.

The change

• 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. The file name format is unchanged. The jsonl rows record the key actually used, so Explorer, the fastpath S3 feeder and ooni/data read the new cans as before.
• 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 both cans are uploaded and the jsonl rows inserted.
• Keep measurements that arrive during a run for the next one, and delete only stale .tmp files. 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.
• flock on the spool, so a run started by hand cannot overlap the timer's (the timer itself never overlaps, Type=oneshot).
• New metrics: late_hour_count, can_index_taken, hourdir_busy, hourdir_kept.

The test

.github/workflows/test_uploader_e2e.yml runs tests/uploader_e2e/ (README there). It uses:

• the fastpath image production runs (from ansible/host_vars/fastpath.prod.ooni.io/vars.yml);
• MinIO with versioning on, standing in for ooni-data-eu-fra and the failed-reports bucket;
• ClickHouse 24.8.6.70;
• the uploader in its production runtime (Ubuntu 22.04 apt packages);
• the reuploader at master and at the version production runs (from tf/environments/prod/main.tf).

It submits real measurements the way the ooniprobe service does, then exercises:

1. a normal hour;
2. measurements written after their hour was uploaded (fastpath workers suspended, then resumed);
3. a full fastpath queue failing over to the failed-reports bucket;
4. the reuploader re-sending into uploaded hours;
5. S3 unreachable during an upload, and two uploader runs at once.