Empty and delete S3-compatible buckets fast. Sharded parallel listing streams straight into parallel deletes, so wall clock is max(list, delete) instead of list + delete. Streaming bulk passes run to exhaustion BEFORE any per-key repair, because repair costs a HEAD plus a PUT and a DELETE each. Clears what `aws s3 rm --recursive` leaves behind: …

s3-purge.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["boto3>=1.34"]
# ///
"""Empty (and optionally delete) S3-compatible buckets, fast.

Runs against AWS S3 and S3-compatible stores such as Ceph RGW, which backs
Hetzner Object Storage, Scaleway and OVH.

Why this exists, when ``aws s3 rm --recursive`` already does most of it:

1. Listing is the bottleneck, not deleting. ListObjects returns at most 1000
   keys per round trip and paginates serially, so a 200k-object bucket costs
   ~200 sequential requests before the first delete is even issued. This script
   splits the keyspace on common prefixes and runs two concurrent stages joined
   by a bounded queue, so deleting starts while listing is still going. Wall
   clock becomes max(list, delete) instead of list + delete.

2. Object versions and delete markers keep a versioned bucket non-empty.

3. Incomplete multipart uploads keep billing storage and block DeleteBucket,
   and no object-level delete removes them.

4. Ceph RGW can leave the bucket index disagreeing with the data behind it. It
   shows up in two shapes, and only the first one is fixable from the outside.

   a. The listing returns a key whose object is gone. HeadObject answers 404
      while the web console still shows the file, and DeleteObject on it is a
      no-op. Re-PUT a zero-byte object to rebuild the entry, then DELETE it —
      that is what the repair stage does.

   b. Every listing call returns nothing, yet DeleteBucket still answers
      BucketNotEmpty. No object-level request reaches whatever the index is
      counting, so this script cannot clear it. Wait and retry first: the
      index count lags a large deletion, and Hetzner puts that lag at 15-20
      minutes. If it persists, only a cluster-side `radosgw-admin bucket check
      --fix` clears it, which a managed-hosting customer cannot run — open a
      support ticket with the bucket id.

Order of work, and why it is that way: streaming bulk passes repeat until one
removes nothing, and only then does per-key repair run on the remainder. A bulk
pass moves thousands of keys per second; repair costs a HEAD plus a PUT and a
DELETE each. Repairing before the bulk passes have drained the bucket pays the
per-key price for keys the next bulk pass would have taken for free.

Bulk passes also stop when two in a row remove the SAME count, which is what a
residue of stale index entries looks like: each pass deletes them successfully
and the next lists them again, so the figure never reaches zero and the loop
would otherwise run to its cap. That residue is repair's job, not streaming's.

Repair needs HeadObject to tell a live object from a stale entry. On an SSE-C
bucket that call requires the customer key and answers 400 without it, so repair
is detected as inoperative up front and skipped with one message, rather than
failing once per key. Pass --sse-customer-key to enable it there.

Credentials come from the environment. This script never reads or writes a
credentials file, and it prints no secret: the only values it echoes are the
endpoint, the concurrency and the bucket names.

    AWS_ACCESS_KEY_ID       required
    AWS_SECRET_ACCESS_KEY   required
    AWS_SESSION_TOKEN       optional
    S3_ENDPOINT_URL         optional, same as --endpoint
    AWS_DEFAULT_REGION      optional, defaults to us-east-1
    S3_SSE_CUSTOMER_KEY     optional, same as --sse-customer-key

Prefer S3_SSE_CUSTOMER_KEY to the flag. An argument is visible in the process
table to every user on the host and lands in shell history; an environment
variable does neither.

Most S3-compatible stores ignore the region for reads and deletes, but some
validate it: Hetzner rejects a mismatch with the endpoint's location, so pair
https://fsn1.your-objectstorage.com with AWS_DEFAULT_REGION=fsn1.

Usage:

    export AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=...
    export S3_ENDPOINT_URL=https://fsn1.your-objectstorage.com

    uv run s3-purge.py --dry-run my-bucket
    uv run s3-purge.py --delete-bucket --yes old-a old-b
    uv run s3-purge.py -p 64 big-bucket

    # SSE-C bucket: without the key the repair stage is skipped, not silently
    # broken. Supply it to let repair clear stale index entries.
    S3_SSE_CUSTOMER_KEY=$(cat key.b64) uv run s3-purge.py --delete-bucket --yes enc-bucket

    # Millions of keys: bulk passes do the work, so cap what repair holds.
    uv run s3-purge.py -p 64 --repair-limit 50000 huge-bucket

License: MIT
"""

from __future__ import annotations

import argparse
import os
import queue
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from typing import Any, Iterator

import boto3
from botocore.config import Config
from botocore.exceptions import BotoCoreError, ClientError

VERSION = "1.1.1"

# DeleteObjects accepts at most 1000 keys per call.
BATCH_SIZE = 1000

# How long a delete worker holds a partial batch before sending it anyway. It
# only matters at the tail of a pass; during the pass batches fill on their own.
IDLE_FLUSH_SECONDS = 2.0

# Caps on each stage. The module docstring explains why bulk runs before repair.
# Both are ceilings, not targets: each stage also stops as soon as a pass makes
# no progress, so a healthy bucket never reaches them.
MAX_BULK_PASSES = 20
MAX_REPAIR_PASSES = 5

# Entries held for one repair round. Repair needs a worklist, not the whole
# bucket, and an unbounded collect on a multi-million-key bucket is gigabytes of
# dicts. Successive rounds pick up whatever this leaves behind.
REPAIR_LIMIT = 100_000

# Prefix levels to descend while splitting the keyspace for parallel listing.
# Six covers the layouts these stores produce in practice — ClickHouse writes
# data/<3>/<29>, pgBackRest archive/<stanza>/<major>/<segment> — while keeping
# discovery itself from turning into the serial listing it exists to avoid.
MAX_SHARD_DEPTH = 6

# Sentinel pushed onto the key queue to retire one delete worker.
_DONE = object()

# Memoised keyspace split, per (bucket, versioned). See `discover_shards`.
_SHARD_CACHE: "dict[tuple[str, bool], list[Shard]]" = {}


# ---------------------------------------------------------------------------
# model
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Shard:
    """A slice of the keyspace one list worker owns.

    ``deep`` lists everything under ``prefix``. A shallow shard lists only the
    keys sitting directly at ``prefix``, which is what makes an expanded prefix
    safe to split: the union of one shallow shard and the deep shards for each
    child prefix is exactly the deep listing it replaced.
    """

    prefix: str
    deep: bool = True


@dataclass
class Stats:
    listed: int = 0
    deleted: int = 0
    repaired: int = 0
    errors: list[str] = field(default_factory=list)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def add_listed(self, n: int) -> None:
        with self._lock:
            self.listed += n

    def add_deleted(self, n: int) -> None:
        with self._lock:
            self.deleted += n

    def add_repaired(self, n: int) -> None:
        with self._lock:
            self.repaired += n

    def error(self, message: str) -> None:
        with self._lock:
            self.errors.append(message)


def log(message: str = "") -> None:
    print(message, file=sys.stderr, flush=True)


# ---------------------------------------------------------------------------
# client
# ---------------------------------------------------------------------------
def make_client(args: argparse.Namespace, pool_size: int) -> Any:
    """Build one S3 client shared by every worker.

    botocore's low-level clients are thread safe, so a single client with a
    connection pool at least as large as the worker count beats one client per
    thread: connections are reused instead of renegotiated per worker.
    """
    cfg: dict[str, Any] = {
        "max_pool_connections": max(pool_size, 10),
        "retries": {"max_attempts": args.max_attempts, "mode": "adaptive"},
        "connect_timeout": 15,
        "read_timeout": 60,
    }
    if args.addressing_style != "auto":
        cfg["s3"] = {"addressing_style": args.addressing_style}

    # botocore >= 1.36 sends CRC32 checksums on every upload by default, which
    # several S3-compatible stores reject. Ask for them only when required.
    # Older botocore does not accept these keys at all.
    try:
        config = Config(
            request_checksum_calculation="when_required",
            response_checksum_validation="when_required",
            **cfg,
        )
    except TypeError:
        config = Config(**cfg)

    return boto3.client(
        "s3",
        endpoint_url=args.endpoint or None,
        region_name=os.environ.get("AWS_DEFAULT_REGION") or "us-east-1",
        config=config,
    )


def is_missing(exc: ClientError) -> bool:
    """True when the store said the object is not there, and only then.

    Repair keys its whole decision off this: a false positive makes it PUT a
    zero-byte object over live data. So it matches 404 alone and treats every
    other failure — 400 from a missing SSE-C key, 403, a throttle — as unknown
    rather than as absent.
    """
    code = str(exc.response.get("Error", {}).get("Code", ""))
    status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
    return code in {"404", "NoSuchKey", "NotFound"} or status == 404


# ---------------------------------------------------------------------------
# discovery
# ---------------------------------------------------------------------------
def is_versioned(client: Any, bucket: str) -> bool:
    try:
        status = client.get_bucket_versioning(Bucket=bucket).get("Status")
    except (ClientError, BotoCoreError):
        return False
    return status in {"Enabled", "Suspended"}


def sse_headers(args: argparse.Namespace) -> dict:
    """SSE-C headers for HeadObject, or an empty dict when none was supplied."""
    if not args.sse_customer_key:
        return {}
    return {
        "SSECustomerAlgorithm": args.sse_customer_algorithm,
        "SSECustomerKey": args.sse_customer_key,
    }


def preflight_sse(
    client: Any, bucket: str, sample: dict | None, sse: dict
) -> tuple[bool, str]:
    """Decide whether repair can classify keys on this bucket, before it starts.

    Repair rests entirely on HeadObject telling a live object from a stale index
    entry. On an SSE-C bucket that call needs the customer key and returns 400
    without it, so every key reads as an error: repair does nothing, but only
    after logging one failure per key. One probe up front turns thousands of
    useless errors into a single accurate line.
    """
    if sample is None:
        return True, ""
    key = sample["Key"]
    try:
        client.head_object(Bucket=bucket, Key=key, **sse)
        return True, ""
    except ClientError as exc:
        if is_missing(exc):
            # A stale entry: HeadObject answered correctly, which is the signal
            # repair needs. Nothing to warn about.
            return True, ""
        status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode")
        if status == 400 and not sse:
            return False, (
                "HeadObject answers 400, which means the objects are SSE-C encrypted. "
                "Repair cannot tell a live object from a stale index entry without the "
                "customer key; pass --sse-customer-key to enable it."
            )
        if status in {400, 403} and sse:
            return False, f"HeadObject rejected the supplied SSE-C key ({status})"
        return True, ""
    except BotoCoreError:
        return True, ""


def probe(
    client: Any, bucket: str, shard: Shard, versioned: bool, stats: "Stats | None" = None
) -> tuple[list[str], bool]:
    """Return one level of child prefixes under a shard, and whether it has
    keys of its own sitting directly at that level.

    A versioned bucket must be probed with ListObjectVersions. ListObjectsV2
    reports only current objects, so a prefix holding nothing but non-current
    versions or delete markers never appears as a common prefix, no shard
    covers it, and those entries are silently never listed.
    """
    params: dict[str, Any] = {
        "Bucket": bucket,
        "Prefix": shard.prefix,
        "Delimiter": "/",
        "MaxKeys": 1000,
    }
    try:
        if versioned:
            resp = client.list_object_versions(**params)
            own = bool(resp.get("Versions") or resp.get("DeleteMarkers"))
        else:
            resp = client.list_objects_v2(**params)
            own = bool(resp.get("Contents"))
    except (ClientError, BotoCoreError) as exc:
        # Report it. Silently returning "no children" collapses the keyspace to a
        # single flat shard, so the run loses all its parallelism and nothing says
        # why — the failure looks like a small bucket rather than a denied probe.
        if stats is not None:
            stats.error(f"probe {shard.prefix!r}: {exc}")
        return [], False
    return [c["Prefix"] for c in resp.get("CommonPrefixes") or []], own


def discover_shards(
    client: Any,
    bucket: str,
    target: int,
    versioned: bool,
    stats: "Stats | None" = None,
    cache: bool = True,
) -> list[Shard]:
    """Split the keyspace until there are enough shards to fill the pool.

    Two bounds keep discovery from becoming the bottleneck it is meant to
    remove. Shards are capped, because a level holding thousands of prefixes
    would otherwise turn one listing into thousands of them. And each level's
    probes run concurrently, since they are independent I/O.

    The result is memoised per bucket. A bucket is walked many times over one
    run — every bulk pass, then every repair round — and re-probing the prefix
    tree each time costs a request per node for a split that barely moves. The
    prefixes stay valid as the bucket drains: a shard whose keys are gone simply
    lists nothing, so a stale split is a no-op rather than a miss.
    """
    cache_key = (bucket, versioned)
    if cache and cache_key in _SHARD_CACHE:
        return _SHARD_CACHE[cache_key]

    max_shards = max(target * 4, 8)
    shards = [Shard("", deep=True)]

    for _ in range(MAX_SHARD_DEPTH):
        if len(shards) >= target:
            break

        deep = [s for s in shards if s.deep]
        if not deep:
            break

        with ThreadPoolExecutor(max_workers=min(len(deep), max(target, 1))) as pool:
            probes = list(
                pool.map(lambda s: (s, probe(client, bucket, s, versioned, stats)), deep)
            )

        nxt: list[Shard] = [s for s in shards if not s.deep]
        expanded = False
        for shard, (children, own) in probes:
            # A flat leaf has nothing to split on, and past the cap the extra
            # requests cost more than the parallelism buys.
            if not children or len(nxt) >= max_shards:
                nxt.append(shard)
                continue

            # A single child yields no parallelism by itself, but the level
            # below it often does: archive/ -> platform-postgres/ -> 18-1/ ->
            # hundreds of segment directories. Keep descending through the
            # narrow part instead of giving up at the first one.
            expanded = True
            if own:
                nxt.append(Shard(shard.prefix, deep=False))
            nxt.extend(Shard(p, deep=True) for p in children)

        if not expanded:
            break
        shards = nxt

    if cache:
        _SHARD_CACHE[cache_key] = shards
    return shards


# ---------------------------------------------------------------------------
# stage 1: listing
# ---------------------------------------------------------------------------
def iter_shard(client: Any, bucket: str, shard: Shard, versioned: bool) -> Iterator[dict]:
    """Yield every delete-ready record in one shard.

    A versioned bucket needs ListObjectVersions so that non-current versions
    and delete markers are removed too; ListObjectsV2 only sees current
    objects and would leave the bucket non-empty.
    """
    params: dict[str, Any] = {"Bucket": bucket}
    if shard.prefix:
        params["Prefix"] = shard.prefix
    if not shard.deep:
        params["Delimiter"] = "/"

    op = "list_object_versions" if versioned else "list_objects_v2"
    groups = ("Versions", "DeleteMarkers") if versioned else ("Contents",)

    for page in client.get_paginator(op).paginate(**params):
        for group in groups:
            for obj in page.get(group) or []:
                record = {"Key": obj["Key"]}
                version = obj.get("VersionId")
                # "null" is a real version id, not a missing one: it is what every
                # object carries in a bucket whose versioning was enabled and then
                # suspended. Dropping it makes the delete omit VersionId, which on
                # a versioned bucket writes a DELETE MARKER instead of removing the
                # version — so the bucket never empties and DeleteBucket keeps
                # failing. Send whatever the listing returned.
                if version:
                    record["VersionId"] = version
                yield record


def list_worker(
    client: Any,
    bucket: str,
    versioned: bool,
    shards: queue.Queue,
    keys: queue.Queue,
    stats: Stats,
    stop: threading.Event,
) -> None:
    while not stop.is_set():
        try:
            shard = shards.get_nowait()
        except queue.Empty:
            return
        try:
            count = 0
            for record in iter_shard(client, bucket, shard, versioned):
                if stop.is_set():
                    return
                # Blocks when the delete stage falls behind, which bounds
                # memory on a bucket with millions of keys.
                keys.put(record)
                count += 1
            stats.add_listed(count)
        except (ClientError, BotoCoreError) as exc:
            stats.error(f"list {shard.prefix!r}: {exc}")


# ---------------------------------------------------------------------------
# stage 2: deleting
# ---------------------------------------------------------------------------
def flush_batch(client: Any, bucket: str, batch: list[dict], stats: Stats) -> None:
    """Delete one batch and count what the store accepted.

    Quiet mode returns only failures, so the count is the batch minus its
    errors. That is a count of ACCEPTED DELETE REQUESTS, not of objects that
    existed: S3 reports success for a key that was never there, so a stale index
    entry scores here exactly like a real object. Never report it as space
    reclaimed.
    """
    if not batch:
        return
    try:
        resp = client.delete_objects(
            Bucket=bucket, Delete={"Objects": batch, "Quiet": True}
        )
    except (ClientError, BotoCoreError) as exc:
        stats.error(f"delete batch of {len(batch)}: {exc}")
        return

    failures = resp.get("Errors") or []
    for err in failures[:10]:
        stats.error(f"delete {err.get('Key')!r}: {err.get('Code')} {err.get('Message')}")
    stats.add_deleted(len(batch) - len(failures))


def delete_worker(
    client: Any,
    bucket: str,
    keys: queue.Queue,
    stats: Stats,
    stop: threading.Event,
) -> None:
    batch: list[dict] = []
    last_add = time.monotonic()
    while True:
        try:
            item = keys.get(timeout=0.25)
        except queue.Empty:
            # Flush a partial batch only once it has gone cold, NOT on every poll.
            # Listing is the slow stage, so a worker routinely holds a fraction of
            # a batch: flushing each time the poll expires splits one 1000-key
            # request into dozens of tiny ones, and the extra request RATE is what
            # tips a throttling store into 503s. The wait costs latency on the
            # tail of a pass only, where there is nothing left to overlap with.
            if batch and time.monotonic() - last_add >= IDLE_FLUSH_SECONDS:
                flush_batch(client, bucket, batch, stats)
                batch = []
            if stop.is_set():
                flush_batch(client, bucket, batch, stats)
                return
            continue

        if item is _DONE:
            # Safe to leave immediately: the sentinels are queued only after every
            # lister has finished, and the queue is FIFO, so all remaining keys sit
            # ahead of them. Draining past the sentinel here would swallow another
            # worker's copy and hang it — `stop` is not set until after the joins.
            flush_batch(client, bucket, batch, stats)
            return

        batch.append(item)
        last_add = time.monotonic()
        if len(batch) >= BATCH_SIZE:
            flush_batch(client, bucket, batch, stats)
            batch = []


def purge_pass(
    client: Any,
    bucket: str,
    versioned: bool,
    args: argparse.Namespace,
    stats: Stats,
) -> int:
    """Run one full list-and-delete pass. Returns the keys listed BY THIS PASS.

    ``stats`` accumulates across the passes of one bucket, so the per-pass figure
    has to be a difference, not the running total.
    """
    listed_before = stats.listed
    shards = discover_shards(client, bucket, args.parallel, versioned, stats)

    shard_q: queue.Queue = queue.Queue()
    for shard in shards:
        shard_q.put(shard)

    # Backpressure: cap in-flight keys so a huge bucket cannot exhaust memory.
    key_q: queue.Queue = queue.Queue(maxsize=BATCH_SIZE * args.parallel * 2)
    stop = threading.Event()

    n_list = min(args.parallel, max(len(shards), 1))
    # Size the delete stage to what the list stage can actually feed it. One
    # request moves 1000 keys either way, so deleters much beyond the lister
    # count cannot go faster — they only split the same stream into smaller
    # batches, costing a round trip per handful. A single flat shard is the case
    # that matters: it yields one lister, and 32 deleters behind it would send
    # tiny requests all pass long. Double for headroom against uneven shards.
    n_delete = max(2, min(args.parallel, n_list * 2))

    log(f"    {len(shards)} shard(s), {n_list} list + {n_delete} delete worker(s)")

    listers = [
        threading.Thread(
            target=list_worker,
            args=(client, bucket, versioned, shard_q, key_q, stats, stop),
            daemon=True,
        )
        for _ in range(n_list)
    ]
    deleters = [
        threading.Thread(
            target=delete_worker,
            args=(client, bucket, key_q, stats, stop),
            daemon=True,
        )
        for _ in range(n_delete)
    ]

    before = stats.deleted
    for t in listers + deleters:
        t.start()

    reporter = threading.Thread(
        target=progress, args=(stats, stop, before), daemon=True
    )
    reporter.start()

    try:
        for t in listers:
            t.join()
        # Listing is done, so retire the delete stage once the queue drains.
        for _ in deleters:
            key_q.put(_DONE)
        for t in deleters:
            t.join()
    except KeyboardInterrupt:
        stop.set()
        raise
    finally:
        stop.set()
        reporter.join(timeout=1)

    return stats.listed - listed_before


def progress(stats: Stats, stop: threading.Event, before: int) -> None:
    start = time.monotonic()
    last = -1
    while not stop.wait(2.0):
        done = stats.deleted - before
        if done == last:
            continue
        last = done
        rate = done / max(time.monotonic() - start, 0.001)
        log(f"      {done:,} deleted ({rate:,.0f}/s)")


# ---------------------------------------------------------------------------
# repair
# ---------------------------------------------------------------------------
def repair_key(
    client: Any, bucket: str, record: dict, stats: Stats, sse: dict | None = None
) -> bool:
    """Clear one entry the bulk pass left behind.

    HeadObject separates a live object that resisted deletion from a stale index
    entry, so the zero-byte rebuild can never overwrite real data. That check is
    load bearing: without it the rebuild is a blind PUT over whatever is there.

    On an SSE-C bucket HeadObject REQUIRES the customer key and answers 400
    without it — not 404 — so every key looks like an error and repair cannot run
    at all. Pass ``sse`` to make it work; ``preflight_sse`` detects the case.
    """
    key = record["Key"]
    try:
        client.head_object(Bucket=bucket, Key=key, **(sse or {}))
        live = True
    except ClientError as exc:
        if not is_missing(exc):
            stats.error(f"head {key!r}: {exc}")
            return False
        live = False
    except BotoCoreError as exc:
        stats.error(f"head {key!r}: {exc}")
        return False

    try:
        if not live:
            # Rebuild the orphaned index entry so the delete below can bind
            # to a real object and remove both.
            client.put_object(Bucket=bucket, Key=key, Body=b"")
        client.delete_object(Bucket=bucket, Key=key)
    except (ClientError, BotoCoreError) as exc:
        stats.error(f"repair {key!r}: {exc}")
        return False

    stats.add_repaired(1)
    return True


def collect_all(
    client: Any,
    bucket: str,
    versioned: bool,
    parallel: int,
    stats: "Stats | None" = None,
    limit: int = 0,
) -> list[dict]:
    """Collect the entries a bulk pass left behind, bounded and fault tolerant.

    ``limit`` caps how many records are held at once. The repair stage only ever
    needs a worklist, never the whole bucket, and an unbounded collect on a
    multi-million-key bucket is gigabytes of dicts in memory — the streaming path
    is bounded by the key queue, so this one has to be bounded too.

    A listing error here is recorded and skipped rather than raised. Letting it
    propagate aborts the process mid-run and takes every remaining bucket with
    it, which is exactly what a transient 403 from the store used to do.
    """
    shards = discover_shards(client, bucket, parallel, versioned, stats)
    records: list[dict] = []
    lock = threading.Lock()
    full = threading.Event()

    def run(shard: Shard) -> None:
        if full.is_set():
            return
        try:
            found: list[dict] = []
            for record in iter_shard(client, bucket, shard, versioned):
                found.append(record)
                if limit and len(found) >= limit:
                    break
            with lock:
                records.extend(found)
                if limit and len(records) >= limit:
                    del records[limit:]
                    full.set()
        except (ClientError, BotoCoreError) as exc:
            if stats is not None:
                stats.error(f"collect {shard.prefix!r}: {exc}")

    with ThreadPoolExecutor(max_workers=max(parallel, 1)) as pool:
        list(pool.map(run, shards))
    return records


# ---------------------------------------------------------------------------
# multipart
# ---------------------------------------------------------------------------
def abort_multipart_uploads(client: Any, bucket: str, parallel: int, stats: Stats) -> int:
    uploads: list[tuple[str, str]] = []
    try:
        for page in client.get_paginator("list_multipart_uploads").paginate(Bucket=bucket):
            for upload in page.get("Uploads") or []:
                uploads.append((upload["Key"], upload["UploadId"]))
    except (ClientError, BotoCoreError) as exc:
        stats.error(f"list multipart uploads: {exc}")
        return 0

    if not uploads:
        return 0

    def abort(item: tuple[str, str]) -> bool:
        key, upload_id = item
        try:
            client.abort_multipart_upload(Bucket=bucket, Key=key, UploadId=upload_id)
            return True
        except (ClientError, BotoCoreError) as exc:
            stats.error(f"abort mpu {key!r}: {exc}")
            return False

    with ThreadPoolExecutor(max_workers=max(parallel, 1)) as pool:
        return sum(pool.map(abort, uploads))


# ---------------------------------------------------------------------------
# per-bucket driver
# ---------------------------------------------------------------------------
def purge_bucket(client: Any, bucket: str, args: argparse.Namespace) -> bool:
    log(f"\n== {bucket}")
    started = time.monotonic()
    stats = Stats()

    try:
        versioned = is_versioned(client, bucket)
    except (ClientError, BotoCoreError) as exc:
        log(f"  cannot inspect: {exc}")
        return False

    if versioned:
        log("  versioning is on; removing versions and delete markers")

    ok = True

    # Drain with bulk passes until one removes nothing. A single bulk pass does
    # not empty a large bucket: deletes race the listing that feeds them, and a
    # store that hands back stale index entries re-serves keys mid-scan. Each
    # extra pass is cheap because it streams, so repeat before paying for repair.
    last_removed = -1
    for bulk in range(1, args.max_bulk_passes + 1):
        log(f"  bulk pass {bulk}: streaming list -> delete")
        before = stats.deleted
        purge_pass(client, bucket, versioned, args, stats)
        removed = stats.deleted - before
        # Delete requests accepted, NOT bytes reclaimed: S3 reports success for a
        # key that does not exist, so a stale index entry counts here exactly like
        # a real object. Read it as progress, never as space recovered.
        log(f"    {removed:,} delete(s) accepted")
        if removed == 0:
            break
        # A count that repeats exactly is not progress. A stale index entry is
        # deleted successfully on every pass and listed again on the next, so a
        # residue of them holds `removed` at a small constant and would otherwise
        # burn every remaining pass. Two identical rounds is the signal to stop
        # streaming and hand what is left to repair, which can actually clear it.
        if removed == last_removed:
            log("    same count twice; bulk passes are not making progress")
            break
        last_removed = removed

    left = collect_all(
        client, bucket, versioned, args.parallel, stats, limit=args.repair_limit
    )

    sse = sse_headers(args)
    can_repair, why_not = preflight_sse(client, bucket, left[0] if left else None, sse)
    if left and not can_repair:
        log(f"  skipping repair: {why_not}")
        repair_passes = 0
    else:
        repair_passes = args.max_repair_passes

    # Whatever survives the bulk pass is either a live object that resisted
    # deletion or a stale index entry. Repair those one by one, and stop as
    # soon as a pass stops making progress.
    for attempt in range(1, repair_passes + 1):
        if not left:
            break
        log(f"  repair pass {attempt}: {len(left):,} entries survived bulk deletion")
        before = stats.repaired
        with ThreadPoolExecutor(max_workers=max(args.parallel, 1)) as pool:
            list(pool.map(lambda r: repair_key(client, bucket, r, stats, sse), left))
        progressed = stats.repaired - before
        log(f"    repaired {progressed:,} of {len(left):,}")
        if progressed == 0:
            break
        left = collect_all(
            client, bucket, versioned, args.parallel, stats, limit=args.repair_limit
        )

    if left:
        log(f"  STILL NOT EMPTY: {len(left):,} entries remain")
        for record in left[:5]:
            log(f"    {record['Key']}")
        ok = False
    else:
        log(f"  empty ({stats.deleted:,} deleted, {stats.repaired:,} repaired)")

    aborted = abort_multipart_uploads(client, bucket, args.parallel, stats)
    if aborted:
        log(f"  aborted {aborted} incomplete multipart upload(s)")

    if args.delete_bucket and ok:
        try:
            client.delete_bucket(Bucket=bucket)
            log("  bucket deleted")
        except ClientError as exc:
            if str(exc.response.get("Error", {}).get("Code")) == "NoSuchBucket":
                log("  bucket already gone")
            else:
                log(f"  bucket NOT deleted: {exc}")
                ok = False
        except BotoCoreError as exc:
            log(f"  bucket NOT deleted: {exc}")
            ok = False

    if stats.errors:
        log(f"  {len(stats.errors)} error(s); first few:")
        for message in stats.errors[:10]:
            log(f"    {message}")
        ok = False

    log(f"  took {time.monotonic() - started:.1f}s")
    return ok


def report_bucket(client: Any, bucket: str, args: argparse.Namespace) -> None:
    log(f"\n== {bucket} (dry run)")
    stats = Stats()
    versioned = is_versioned(client, bucket)
    shards = discover_shards(client, bucket, args.parallel, versioned, stats)
    records = collect_all(
        client, bucket, versioned, args.parallel, stats, limit=args.repair_limit
    )

    uploads = 0
    try:
        for page in client.get_paginator("list_multipart_uploads").paginate(Bucket=bucket):
            uploads += len(page.get("Uploads") or [])
    except (ClientError, BotoCoreError) as exc:
        stats.error(str(exc))

    capped = args.repair_limit and len(records) >= args.repair_limit
    log(f"  versioned:                    {versioned}")
    log(f"  shards for parallel listing:  {len(shards)}")
    log(
        f"  objects/versions listed:      {len(records):,}"
        + (f" (capped at --repair-limit; the bucket holds more)" if capped else "")
    )
    log(f"  incomplete multipart uploads: {uploads}")
    if stats.errors:
        log(f"  listing errors:               {len(stats.errors)} (first: {stats.errors[0][:80]})")
    if args.delete_bucket:
        log("  would delete the bucket itself")


# ---------------------------------------------------------------------------
# cli
# ---------------------------------------------------------------------------
def parse_args(argv: list[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        prog="s3-purge.py",
        description="Empty (and optionally delete) S3-compatible buckets.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Credentials come from AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY\n"
            "(and optionally AWS_SESSION_TOKEN). The endpoint may come from\n"
            "S3_ENDPOINT_URL instead of --endpoint. Some stores validate the\n"
            "region: pair a Hetzner fsn1 endpoint with AWS_DEFAULT_REGION=fsn1.\n\n"
            "Pass an SSE-C key through S3_SSE_CUSTOMER_KEY rather than the flag:\n"
            "an argument is readable in the process table and in shell history.\n\n"
            "Work runs as streaming bulk passes until one deletes nothing, then\n"
            "per-key repair on the remainder. Repair needs HeadObject, which an\n"
            "SSE-C bucket refuses without the key, so it is skipped there.\n\n"
            "Examples:\n"
            "  uv run s3-purge.py --dry-run my-bucket\n"
            "  uv run s3-purge.py --delete-bucket --yes old-a old-b\n"
            "  uv run s3-purge.py -e https://fsn1.your-objectstorage.com -p 64 big\n"
            "  S3_SSE_CUSTOMER_KEY=$(cat key.b64) uv run s3-purge.py --yes enc-bucket\n"
        ),
    )
    parser.add_argument("buckets", nargs="+", metavar="BUCKET")
    parser.add_argument(
        "-e", "--endpoint",
        default=os.environ.get("S3_ENDPOINT_URL", ""),
        help="S3 endpoint URL (default: $S3_ENDPOINT_URL; omit for AWS S3)",
    )
    parser.add_argument(
        "-p", "--parallel", type=int, default=32,
        help="workers per stage (default: 32)",
    )
    parser.add_argument(
        "-D", "--delete-bucket", action="store_true",
        help="delete each bucket once it is empty",
    )
    parser.add_argument(
        "-n", "--dry-run", action="store_true",
        help="report what would be deleted; change nothing",
    )
    parser.add_argument(
        "-y", "--yes", action="store_true",
        help="do not prompt for confirmation",
    )
    parser.add_argument(
        "--addressing-style", choices=("auto", "path", "virtual"), default="auto",
        help="S3 addressing style; try 'path' if the endpoint rejects requests",
    )
    parser.add_argument(
        "--max-attempts", type=int, default=10,
        help="retry attempts per request (default: 10)",
    )
    parser.add_argument(
        "--max-bulk-passes", type=int, default=MAX_BULK_PASSES,
        help=f"streaming list+delete passes before repair (default: {MAX_BULK_PASSES})",
    )
    parser.add_argument(
        "--max-repair-passes", type=int, default=MAX_REPAIR_PASSES,
        help=f"per-key repair passes after the bulk passes (default: {MAX_REPAIR_PASSES})",
    )
    parser.add_argument(
        "--repair-limit", type=int, default=REPAIR_LIMIT,
        help=(
            "cap on entries collected for repair at once; 0 means no cap. Bounds "
            f"memory on a multi-million-key bucket (default: {REPAIR_LIMIT})"
        ),
    )
    parser.add_argument(
        "--sse-customer-key",
        default=os.environ.get("S3_SSE_CUSTOMER_KEY", ""),
        help=(
            "SSE-C key (base64) for HeadObject during repair; without it repair "
            "cannot classify keys on an SSE-C bucket. Default: $S3_SSE_CUSTOMER_KEY"
        ),
    )
    parser.add_argument(
        "--sse-customer-algorithm", default="AES256",
        help="SSE-C algorithm (default: AES256)",
    )
    parser.add_argument("-V", "--version", action="version", version=VERSION)

    args = parser.parse_args(argv)
    if args.parallel < 1:
        parser.error("--parallel must be at least 1")
    return args


def confirm(buckets: list[str], delete_bucket: bool) -> bool:
    what = "permanently delete every object in"
    if delete_bucket:
        what = "permanently delete every object in, then remove,"
    log(f"\nAbout to {what} {len(buckets)} bucket(s):")
    for bucket in buckets:
        log(f"  - {bucket}")
    log("This cannot be undone.")
    try:
        reply = input(f"Type the number of buckets ({len(buckets)}) to continue: ")
    except (EOFError, KeyboardInterrupt):
        return False
    return reply.strip() == str(len(buckets))


def main(argv: list[str]) -> int:
    args = parse_args(argv)

    for name in ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"):
        if not os.environ.get(name):
            log(f"s3-purge: {name} is not set")
            return 1

    client = make_client(args, pool_size=args.parallel * 2 + 8)

    log(f"s3-purge {VERSION}")
    log(f"endpoint:    {args.endpoint or '<AWS S3 default>'}")
    log(f"concurrency: {args.parallel} per stage")
    log(f"buckets:     {' '.join(args.buckets)}")

    if args.dry_run:
        for bucket in args.buckets:
            report_bucket(client, bucket, args)
        log("\ndry run: nothing was changed")
        return 0

    if not args.yes and not confirm(args.buckets, args.delete_bucket):
        log("aborted")
        return 1

    failed = False
    for bucket in args.buckets:
        if not purge_bucket(client, bucket, args):
            failed = True

    log("\nfinished with errors" if failed else "\ndone")
    return 2 if failed else 0


if __name__ == "__main__":
    try:
        sys.exit(main(sys.argv[1:]))
    except KeyboardInterrupt:
        log("\ninterrupted")
        sys.exit(130)