diff --git a/.github/workflows/test_clickhouse_backup.yml b/.github/workflows/test_clickhouse_backup.yml
new file mode 100644
index 00000000..f660b978
--- /dev/null
+++ b/.github/workflows/test_clickhouse_backup.yml
@@ -0,0 +1,26 @@
+# Tests the ClickHouse backup role, its alerts and their Alertmanager routes
+# (ansible/roles/clickhouse_backup/tests), only when one of them changes
+name: Test ClickHouse backups
+on:
+  pull_request:
+    paths:
+      - "ansible/roles/clickhouse_backup/**"
+      - "ansible/roles/prometheus/files/alert_rules.yml"
+      - "ansible/roles/prometheus_alertmanager/**"
+      - ".github/workflows/test_clickhouse_backup.yml"
+  push:
+    branches: [main]
+    paths:
+      - "ansible/roles/clickhouse_backup/**"
+      - "ansible/roles/prometheus/files/alert_rules.yml"
+      - "ansible/roles/prometheus_alertmanager/**"
+      - ".github/workflows/test_clickhouse_backup.yml"
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    steps:
+      - uses: actions/checkout@v4
+      - run: sudo apt-get update && sudo apt-get install -y --no-install-recommends ansible-core python3-boto3
+      - run: ansible/roles/clickhouse_backup/tests/run.sh
diff --git a/ansible/deploy-clickhouse.yml b/ansible/deploy-clickhouse.yml
index 630d1736..b40c3ec7 100644
--- a/ansible/deploy-clickhouse.yml
+++ b/ansible/deploy-clickhouse.yml
@@ -11,3 +11,16 @@
       tags: [node_exporter]
     - role: oonidata_clickhouse
       tags: [clickhouse]
+
+# data3 backs up the replicated tables: they hold the same data on every
+# node, and data3 isn't behind clickhouseproxy and doesn't run Airflow.
+# data1 backs up the plain MergeTree tables whose data only it holds (see
+# clickhouse_backup_data1_tables). The jobs are set in host_vars.
+- name: Deploy ClickHouse backups
+  hosts:
+    - data1.htz-fsn.prod.ooni.nu
+    - data3.htz-fsn.prod.ooni.nu
+  become: true
+  roles:
+    - role: clickhouse_backup
+      tags: [clickhouse_backup]
diff --git a/ansible/group_vars/clickhouse/vars.yml b/ansible/group_vars/clickhouse/vars.yml
index 73275001..9b5ad668 100644
--- a/ansible/group_vars/clickhouse/vars.yml
+++ b/ansible/group_vars/clickhouse/vars.yml
@@ -411,6 +411,27 @@ clickhouse_prometheus:
   asynchronous_metrics: true
   status_info: true
 
+# For textfile metrics, e.g. from clickhouse_backup
+node_exporter_textfile_dir: /var/lib/node_exporter/textfile_collector
+
+# clickhouse_backup: Cloudflare R2. The endpoint names the account, so it's
+# kept in SSM with the token's keys.
+clickhouse_backup_s3_endpoint: "{{ lookup('amazon.aws.aws_ssm', '/oonidevops/secrets/clickhouse_backup_r2_endpoint', profile='oonidevops_user_prod') }}"
+clickhouse_backup_s3_access_key_id: "{{ lookup('amazon.aws.aws_ssm', '/oonidevops/secrets/clickhouse_backup_r2_access_key_id', profile='oonidevops_user_prod') }}"
+clickhouse_backup_s3_secret_access_key: "{{ lookup('amazon.aws.aws_ssm', '/oonidevops/secrets/clickhouse_backup_r2_secret_access_key', profile='oonidevops_user_prod') }}"
+# Tables with personal or unpublished data, backed up to the private bucket
+clickhouse_backup_private_tables:
+  - msmt_feedback        # account ids and free-text comments
+  - faulty_measurements  # until its contents are reviewed
+  - url_priorities       # until it's decided whether it's published
+# Plain MergeTree tables whose data only data1 holds (2026-10-20). Remove a
+# table from here once it's replicated, and data3 picks it up.
+clickhouse_backup_data1_tables:
+  - labeling_frames
+  - event_detector_cusums
+  - event_detector_v2_changepoints
+  - changepoint_label
+
 prometheus_nginx_proxy_config:
   - location: /metrics/node_exporter
     proxy_pass: http://127.0.0.1:8100/metrics
diff --git a/ansible/host_vars/data1.htz-fsn.prod.ooni.nu b/ansible/host_vars/data1.htz-fsn.prod.ooni.nu
index 74f6d22f..a0b90b64 100644
--- a/ansible/host_vars/data1.htz-fsn.prod.ooni.nu
+++ b/ansible/host_vars/data1.htz-fsn.prod.ooni.nu
@@ -2,3 +2,10 @@ tls_cert_dir: /var/lib/dehydrated/certs
 ssl_domains:
    - "data1.htz-fsn.prod.ooni.nu"
    - "airflow.prod.ooni.io"
+
+# Only the tables whose data only data1 holds; data3 backs up the rest
+clickhouse_backup_jobs:
+  - name: data1
+    s3_bucket: ooni-clickhouse-backup
+    include_tables: "{{ clickhouse_backup_data1_tables }}"
+    exclude_tables: []
diff --git a/ansible/host_vars/data3.htz-fsn.prod.ooni.nu b/ansible/host_vars/data3.htz-fsn.prod.ooni.nu
index bf1b24a0..34959203 100644
--- a/ansible/host_vars/data3.htz-fsn.prod.ooni.nu
+++ b/ansible/host_vars/data3.htz-fsn.prod.ooni.nu
@@ -1,2 +1,20 @@
 non_admin_usernames: []
 clickhouse_base_path: /data/clickhouse
+
+# The replicated tables, minus the ones only data1 holds data for. Private
+# tables go to their own bucket: R2 tokens grant access per bucket, and
+# researchers may be given read access to the measurement tables.
+clickhouse_backup_jobs:
+  - name: measurements
+    s3_bucket: ooni-clickhouse-backup
+    include_tables: ["*"]
+    exclude_tables: "{{ clickhouse_backup_private_tables + clickhouse_backup_data1_tables }}"
+  - name: private
+    s3_bucket: ooni-clickhouse-backup-private
+    include_tables: "{{ clickhouse_backup_private_tables }}"
+    exclude_tables: []
+
+# After each run, report the tables on data1-3 that no backup covers;
+# data1's own tables are backed up by data1
+clickhouse_backup_check_cluster: oonidata_cluster
+clickhouse_backup_backed_up_elsewhere: "{{ clickhouse_backup_data1_tables }}"
diff --git a/ansible/roles/clickhouse_backup/defaults/main.yml b/ansible/roles/clickhouse_backup/defaults/main.yml
new file mode 100644
index 00000000..6ae23389
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/defaults/main.yml
@@ -0,0 +1,78 @@
+---
+# Back up the ooni ClickHouse database to S3-compatible object storage
+# (Cloudflare R2) with ClickHouse's native BACKUP. Deploy where the data is:
+# data3 for the replicated tables, data1 for the tables only data1 holds
+# (see deploy-clickhouse.yml and host_vars).
+# Restore and upgrade procedure: docs/Runbooks.md, "ClickHouse backup".
+
+# false stops and disables the timer, e.g. while upgrading the cluster
+clickhouse_backup_enabled: true
+# Weekly, on Sundays at a quiet time (UTC): a full backup of everything takes
+# more than a day. A run backs up every job's tables, one job after the other;
+# a run still going delays the next one instead of overlapping it.
+clickhouse_backup_on_calendar: "Sun *-*-* 02:30:00"
+
+clickhouse_backup_database: ooni
+clickhouse_backup_clickhouse_url: "http://127.0.0.1:8123/"
+clickhouse_backup_clickhouse_user: "{{ clickhouse_admin_user }}"
+clickhouse_backup_clickhouse_password: "{{ clickhouse_admin_password }}"
+
+# Cloudflare R2: the endpoint names the account,
+# https://<account id>.r2.cloudflarestorage.com. One API token with Object
+# Read & Write on every bucket the jobs below use.
+clickhouse_backup_s3_endpoint: ""
+clickhouse_backup_s3_region: auto
+clickhouse_backup_s3_prefix: prod
+clickhouse_backup_s3_access_key_id: ""
+clickhouse_backup_s3_secret_access_key: ""
+
+# Query settings for ClickHouse's uploads. R2 rejects a multipart upload
+# whose parts (all but the last) differ in size, and ClickHouse grows the
+# part size as an upload goes on unless the factor is 1. R2 also allows at
+# most 10,000 parts, so with 128 MiB parts one file can reach 1.25 TiB.
+clickhouse_backup_s3_settings:
+  s3_upload_part_size_multiply_factor: 1
+  s3_min_upload_part_size: 134217728
+  s3_max_upload_part_size: 134217728
+
+# One job per bucket. A job backs up the tables matching include_tables ("*"
+# for every table; an empty list backs up none) minus the excludes below and
+# its own.
+clickhouse_backup_jobs:
+  - name: measurements
+    s3_bucket: ""
+    include_tables: ["*"]
+    exclude_tables: []
+
+# Complete chains kept per table, besides chains pinned with --keep: the
+# current one and the one before, so a month or two of restore points
+clickhouse_backup_keep_chains: 2
+# Bytes/s passed to ClickHouse as max_backup_bandwidth: at most 50 MB/s, 40%
+# of the 1 Gbit/s link, leaving the rest for replication and queries.
+# Against MinIO in testing, 24.8 uploaded at about a third of this value; check
+# the real rate on the first backups (ooni_clickhouse_backup_last_size_bytes /
+# ooni_clickhouse_backup_last_duration_seconds) and raise it if they're too slow.
+clickhouse_backup_max_bandwidth: 50000000
+
+# fnmatch patterns, for every job. Materialized views' .inner tables are
+# always skipped: they're backed up with their view.
+# Tables that are never backed up, for every job, and why
+clickhouse_backup_exclude_tables:
+  - "*_tmp"               # updater staging tables
+  - msmt_feedback_old
+  - fingerprints_dns      # the hourly fingerprints updater rebuilds them
+  - fingerprints_http
+  - session_expunge       # empty, no longer used (backend #972)
+
+# The cluster whose tables are checked after each run: any table on any of its
+# nodes that no job here backs up, that isn't excluded above and that isn't in
+# clickhouse_backup_backed_up_elsewhere is reported (ClickHouseTableNotBackedUp).
+# Set on one host, the one taking most backups; empty checks nothing.
+clickhouse_backup_check_cluster: ""
+# fnmatch patterns of the tables another host backs up
+clickhouse_backup_backed_up_elsewhere: []
+
+# Results for Prometheus, read by node_exporter's textfile collector
+# (node_exporter_textfile_dir in the prometheus_node_exporter role)
+clickhouse_backup_metrics_file: "{{ node_exporter_textfile_dir | default('/var/lib/node_exporter/textfile_collector') }}/ooni_clickhouse_backup.prom"
+clickhouse_backup_state_file: /var/lib/ooni-clickhouse-backup/state.json
diff --git a/ansible/roles/clickhouse_backup/files/ooni_clickhouse_backup.py b/ansible/roles/clickhouse_backup/files/ooni_clickhouse_backup.py
new file mode 100644
index 00000000..e743fb5c
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/files/ooni_clickhouse_backup.py
@@ -0,0 +1,828 @@
+#!/usr/bin/env python3
+"""
+Back up a ClickHouse database to S3-compatible object storage using
+ClickHouse's native BACKUP, with one incremental chain per table.
+
+Configured by /etc/ooni/clickhouse-backup.json (see the clickhouse_backup
+Ansible role). Runs from a systemd timer. The config lists one or more jobs,
+run one after the other; each job backs up a set of tables to its own bucket
+(R2 tokens grant access per bucket, so private tables get a bucket of their
+own).
+
+Layout in each bucket:
+
+    <prefix>/<database>/<table>/<chain>/<backup>/          a native backup
+    <prefix>/<database>/<table>/<chain>/<backup>/data.native.zst
+                                                           data export, for
+                                                           EmbeddedRocksDB and
+                                                           Join tables only
+    <prefix>/<database>/<table>/<chain>/KEEP-<label>       pin, never pruned
+
+<chain> and <backup> are UTC timestamps (20261013T023000Z). The first backup
+of a chain is a full backup; every later one is incremental on top of the
+previous one. A table starts a new chain at its first backup in each
+calendar month (UTC): the 1st with daily runs, the first Sunday with weekly
+ones. A backup is complete once ClickHouse has written its .backup file,
+which it does last.
+
+Why per table: a new full backup of obs_web takes more than a day at our
+bandwidth cap, so one failure shouldn't throw away the other tables' work,
+and chains can be restarted one table at a time.
+
+Why the data export: BACKUP writes only the CREATE statement for
+EmbeddedRocksDB and Join tables (checked on 24.8.6.70), so their rows are
+exported separately in Native format.
+
+ClickHouse reads the bucket credentials from an <s3> endpoint in its config
+(config.d/ooni_backup_s3.xml); this script only uses them itself, through
+boto3, to list and delete backups.
+
+Commands:
+    run [--job J ...] [--full] [--table T ...] [--keep LABEL]
+    list [--job J ...] [--table T ...]
+    release [--job J ...] LABEL
+
+After each table, the result goes to a Prometheus textfile (metrics_file in
+the config) for node_exporter: last success and attempt per table, whether
+the last attempt failed, its duration and size, and whether a job is running.
+At the end of a run, each job's result: when it finished, how long it took,
+and how many tables it backed up, as full backups and in all, and failed.
+
+Monitor with:  sudo journalctl -f --identifier ooni-clickhouse-backup
+"""
+
+import argparse
+import fcntl
+import fnmatch
+import json
+import logging
+import os
+import re
+import signal
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import datetime, timezone
+
+import boto3  # debdeps: python3-boto3
+from botocore.config import Config as BotoConfig
+
+log = logging.getLogger("ooni-clickhouse-backup")
+
+DEFAULT_CONFIG = "/etc/ooni/clickhouse-backup.json"
+TS_FMT = "%Y%m%dT%H%M%SZ"
+SAFE_NAME = re.compile(r"^[A-Za-z0-9_]+$")
+# BACKUP stores only the schema of these engines, so their rows are exported
+EXPORT_ENGINES = {"EmbeddedRocksDB", "Join"}
+
+
+def utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def ts(dt: datetime) -> str:
+    return dt.strftime(TS_FMT)
+
+
+def parse_ts(s: str) -> datetime:
+    return datetime.strptime(s, TS_FMT).replace(tzinfo=timezone.utc)
+
+
+def size(n: int) -> str:
+    for unit in ("B", "KiB", "MiB", "GiB"):
+        if n < 1024:
+            return f"{n:.0f} {unit}" if unit == "B" else f"{n:.1f} {unit}"
+        n /= 1024
+    return f"{n:.2f} TiB"
+
+
+def sql_str(s: str) -> str:
+    return "'" + s.replace("\\", "\\\\").replace("'", "\\'") + "'"
+
+
+class ClickHouse:
+    """Minimal client for the HTTP interface"""
+
+    def __init__(self, url: str, user: str, password: str):
+        self.url = url.rstrip("/") + "/"
+        self.user = user
+        self.password = password
+
+    def query(self, sql: str, settings: dict = None, timeout: float = 600) -> str:
+        url = self.url
+        if settings:
+            url += "?" + urllib.parse.urlencode(settings)
+        req = urllib.request.Request(url, data=sql.encode(), method="POST")
+        req.add_header("X-ClickHouse-User", self.user)
+        if self.password:
+            req.add_header("X-ClickHouse-Key", self.password)
+        try:
+            with urllib.request.urlopen(req, timeout=timeout) as resp:
+                return resp.read().decode()
+        except urllib.error.HTTPError as e:
+            body = e.read().decode(errors="replace").strip()
+            raise RuntimeError(f"ClickHouse error: {body}") from None
+
+    def rows(self, sql: str, **kw) -> list:
+        out = self.query(sql + "\nFORMAT JSONEachRow", **kw)
+        return [json.loads(line) for line in out.splitlines() if line.strip()]
+
+
+class Store:
+    """The backup bucket, as seen by ClickHouse and by boto3"""
+
+    def __init__(self, conf: dict):
+        self.bucket = conf["s3_bucket"]
+        self.prefix = conf["s3_prefix"].strip("/")
+        self.key_id = conf["s3_access_key_id"]
+        self.secret = conf["s3_secret_access_key"]
+        endpoint = conf["s3_endpoint"].rstrip("/")
+        # ClickHouse may reach the store under another name (e.g. in tests)
+        ch_endpoint = conf.get("clickhouse_s3_endpoint", endpoint).rstrip("/")
+        self.ch_base = f"{ch_endpoint}/{self.bucket}"
+        botoconf = dict(s3={"addressing_style": "path"})
+        try:
+            # boto3 >= 1.36 adds checksums some S3-compatible stores reject
+            cfg = BotoConfig(
+                request_checksum_calculation="when_required",
+                response_checksum_validation="when_required",
+                **botoconf,
+            )
+        except TypeError:
+            cfg = BotoConfig(**botoconf)
+        self.s3 = boto3.client(
+            "s3",
+            endpoint_url=endpoint,
+            region_name=conf.get("s3_region", "us-east-1"),
+            aws_access_key_id=self.key_id,
+            aws_secret_access_key=self.secret,
+            config=cfg,
+        )
+
+    def table_prefix(self, database: str, table: str) -> str:
+        return f"{self.prefix}/{database}/{table}/"
+
+    def ch_s3(self, key: str) -> str:
+        """S3(...) argument for BACKUP/RESTORE and the s3 function.
+
+        No credentials: ClickHouse takes them from the <s3> endpoint in its
+        own config. An incremental backup records its base_backup argument
+        verbatim in its .backup file, so keys passed here would end up in
+        the bucket in plain text."""
+        return sql_str(f"{self.ch_base}/{key}")
+
+    def subdirs(self, prefix: str) -> list:
+        """Names of the "directories" directly below prefix"""
+        names = []
+        pager = self.s3.get_paginator("list_objects_v2")
+        for page in pager.paginate(Bucket=self.bucket, Prefix=prefix, Delimiter="/"):
+            for p in page.get("CommonPrefixes", []):
+                names.append(p["Prefix"][len(prefix) :].rstrip("/"))
+            for o in page.get("Contents", []):
+                names.append(o["Key"][len(prefix) :])
+        return names
+
+    def exists(self, key: str) -> bool:
+        try:
+            self.s3.head_object(Bucket=self.bucket, Key=key)
+            return True
+        except self.s3.exceptions.ClientError as e:
+            if e.response.get("Error", {}).get("Code") in ("404", "NoSuchKey", "NotFound"):
+                return False
+            raise
+
+    def put_empty(self, key: str) -> None:
+        self.s3.put_object(Bucket=self.bucket, Key=key, Body=b"")
+
+    def delete_key(self, key: str) -> None:
+        self.s3.delete_object(Bucket=self.bucket, Key=key)
+
+    def delete_prefix(self, prefix: str) -> int:
+        assert prefix.endswith("/") and prefix.startswith(self.prefix + "/")
+        n = 0
+        pager = self.s3.get_paginator("list_objects_v2")
+        for page in pager.paginate(Bucket=self.bucket, Prefix=prefix):
+            keys = [{"Key": o["Key"]} for o in page.get("Contents", [])]
+            if keys:
+                resp = self.s3.delete_objects(Bucket=self.bucket, Delete={"Objects": keys, "Quiet": True})
+                # a 200 response can still list keys it didn't delete
+                errors = resp.get("Errors", [])
+                if errors:
+                    e = errors[0]
+                    raise RuntimeError(f"{len(errors)} objects not deleted under {prefix}, e.g. {e.get('Key')}: {e.get('Code')} {e.get('Message')}")
+                n += len(keys)
+        return n
+
+    def chains(self, database: str, table: str) -> dict:
+        """{chain: {"backups": {backup: complete}, "keep": [labels]}}"""
+        tprefix = self.table_prefix(database, table)
+        out = {}
+        for chain in self.subdirs(tprefix):
+            if not chain:
+                continue
+            entry = {"backups": {}, "keep": []}
+            for name in self.subdirs(f"{tprefix}{chain}/"):
+                if name.startswith("KEEP-"):
+                    entry["keep"].append(name[len("KEEP-") :])
+                elif name:
+                    marker = f"{tprefix}{chain}/{name}/.backup"
+                    entry["backups"][name] = self.exists(marker)
+            out[chain] = entry
+        return out
+
+
+def write_atomically(path: str, text: str) -> None:
+    """node_exporter must never read a half-written file"""
+    tmp = f"{path}.{os.getpid()}.tmp"
+    with open(tmp, "w") as f:
+        f.write(text)
+        f.flush()
+        os.fsync(f.fileno())  # or a power loss can leave an empty file behind
+    os.chmod(tmp, 0o644)
+    os.replace(tmp, path)
+
+
+def label_value(s: str) -> str:
+    return s.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
+
+
+class Metrics:
+    """Backup results for Prometheus, as a node_exporter textfile. The job's
+    name goes in a "backup" label: the scrape already sets "job".
+
+    The latest result of every table is kept in a JSON state file, so a run
+    that backs up some tables doesn't drop the others' metrics."""
+
+    GAUGES = [
+        ("last_success_timestamp_seconds", "last_success",
+         "When the table's last successful backup finished"),
+        ("last_attempt_timestamp_seconds", "last_attempt",
+         "When the table's last backup attempt finished"),
+        ("last_attempt_failed", "last_failed",
+         "1 if the table's last backup attempt failed"),
+        ("last_duration_seconds", "duration",
+         "How long the table's last successful backup took"),
+        ("last_size_bytes", "total_size",
+         "Size of the table's last successful backup, as system.backups reports it"),
+        ("last_files", "num_files",
+         "Files in the table's last successful backup"),
+    ]
+
+    RUN_GAUGES = [
+        ("last_run_finished_timestamp_seconds", "finished", "When the job's last run finished"),
+        ("last_run_duration_seconds", "duration", "How long the job's last run took"),
+        ("last_run_tables", "tables", "Tables the job's last run backed up"),
+        ("last_run_full_backups", "fulls", "Tables the job's last run backed up in full"),
+        ("last_run_failed_tables", "failed", "Tables whose backup failed in the job's last run"),
+    ]
+
+    def __init__(self, metrics_file: str = "", state_file: str = ""):
+        self.metrics_file = metrics_file
+        self.state_file = state_file
+        self.state = {"tables": {}, "running": {}}
+        if state_file and os.path.exists(state_file):
+            try:
+                with open(state_file) as f:
+                    self.state = json.load(f)
+            except ValueError as e:
+                # the results are rebuilt as tables are backed up again
+                log.warning(f"{state_file} is unreadable ({e}), starting from no results")
+        self.state.setdefault("runs", {})
+        self.state.setdefault("running_since", {})
+        self.state.setdefault("not_backed_up", {})
+
+    def table_result(self, job: str, table: str, ok: bool, st: dict = None, duration_s: float = 0) -> None:
+        entry = self.state["tables"].setdefault(f"{job}/{table}", {"job": job, "table": table})
+        now = time.time()
+        entry["last_attempt"] = now
+        entry["last_failed"] = 0 if ok else 1
+        if ok:
+            entry["last_success"] = now
+            entry["duration"] = round(duration_s, 1)
+            entry["total_size"] = int(st.get("total_size", 0))
+            entry["num_files"] = int(st.get("num_files", 0))
+        self.save()
+
+    def job_running(self, job: str, running: bool) -> None:
+        self.state["running"][job] = 1 if running else 0
+        self.state["running_since"][job] = time.time() if running else 0
+        self.save()
+
+    def job_failed(self, job: str) -> None:
+        """The job couldn't start: its tables' last attempts failed"""
+        for entry in self.state["tables"].values():
+            if entry["job"] == job:
+                entry["last_attempt"] = time.time()
+                entry["last_failed"] = 1
+        self.save()
+
+    def forget_tables(self, job: str, tables: set) -> None:
+        """Drop the job's other tables, e.g. dropped or moved to another job,
+        so that their old results don't alert forever"""
+        for key, entry in list(self.state["tables"].items()):
+            if entry["job"] == job and entry["table"] not in tables:
+                log.info(f"{job}: {entry['table']}: no longer backed up by this job, forgetting its results")
+                del self.state["tables"][key]
+        self.save()
+
+    def forget_jobs(self, jobs: set) -> None:
+        """Drop the results of jobs that are no longer configured"""
+        for key, entry in list(self.state["tables"].items()):
+            if entry["job"] not in jobs:
+                del self.state["tables"][key]
+        for part in ("runs", "running", "running_since"):
+            for job in set(self.state[part]) - jobs:
+                log.info(f"{job}: no longer configured, forgetting its results")
+                del self.state[part][job]
+        self.save()
+
+    def not_backed_up(self, tables: dict) -> None:
+        """{table: [hosts]} of the cluster's tables that no backup covers"""
+        self.state["not_backed_up"] = tables
+        self.save()
+
+    def run_result(self, job: str, duration_s: float, tables: int, fulls: int, failed: int) -> None:
+        self.state["runs"][job] = {
+            "finished": time.time(), "duration": round(duration_s, 1),
+            "tables": tables, "fulls": fulls, "failed": failed,
+        }
+        self.save()
+
+    def render(self) -> str:
+        p = "ooni_clickhouse_backup_"
+        lines = [
+            f"# HELP {p}running 1 while the job is backing up tables",
+            f"# TYPE {p}running gauge",
+        ]
+        for job, v in sorted(self.state["running"].items()):
+            lines.append(f'{p}running{{backup="{label_value(job)}"}} {v}')
+        lines += [
+            f"# HELP {p}running_since_timestamp_seconds When the job's current run started, 0 if none",
+            f"# TYPE {p}running_since_timestamp_seconds gauge",
+        ]
+        for job, v in sorted(self.state["running_since"].items()):
+            lines.append(f'{p}running_since_timestamp_seconds{{backup="{label_value(job)}"}} {v}')
+        for name, field, help_ in self.RUN_GAUGES:
+            lines += [f"# HELP {p}{name} {help_}", f"# TYPE {p}{name} gauge"]
+            for job, run in sorted(self.state["runs"].items()):
+                lines.append(f'{p}{name}{{backup="{label_value(job)}"}} {run[field]}')
+        lines += [
+            f"# HELP {p}table_not_backed_up 1 for a table on the cluster that no backup job covers or excludes",
+            f"# TYPE {p}table_not_backed_up gauge",
+        ]
+        for table, hosts in sorted(self.state["not_backed_up"].items()):
+            labels = f'table="{label_value(table)}",hosts="{label_value(",".join(sorted(hosts)))}"'
+            lines.append(f"{p}table_not_backed_up{{{labels}}} 1")
+        for name, field, help_ in self.GAUGES:
+            lines += [f"# HELP {p}{name} {help_}", f"# TYPE {p}{name} gauge"]
+            for entry in sorted(self.state["tables"].values(), key=lambda e: (e["job"], e["table"])):
+                if field in entry:
+                    labels = f'backup="{label_value(entry["job"])}",table="{label_value(entry["table"])}"'
+                    lines.append(f"{p}{name}{{{labels}}} {entry[field]}")
+        return "\n".join(lines) + "\n"
+
+    def save(self) -> None:
+        if self.state_file:
+            os.makedirs(os.path.dirname(self.state_file), exist_ok=True)
+            write_atomically(self.state_file, json.dumps(self.state, indent=1))
+        if self.metrics_file:
+            write_atomically(self.metrics_file, self.render())
+
+
+def complete_chains(chains: dict) -> list:
+    """Chains holding at least one complete backup, oldest first"""
+    return sorted(c for c, e in chains.items() if any(e["backups"].values()))
+
+
+def latest_backup(chains: dict):
+    """(chain, backup) of the newest complete backup, or (None, None)"""
+    done = complete_chains(chains)
+    if not done:
+        return None, None
+    chain = done[-1]
+    backups = sorted(b for b, ok in chains[chain]["backups"].items() if ok)
+    return chain, backups[-1]
+
+
+class BackupJob:
+    def __init__(self, conf: dict, metrics: Metrics = None):
+        self.conf = conf
+        self.name = conf.get("name", "default")
+        self.metrics = metrics or Metrics()
+        self.db = conf.get("database", "ooni")
+        self.ch = ClickHouse(
+            conf.get("clickhouse_url", "http://127.0.0.1:8123/"),
+            conf.get("clickhouse_user", "default"),
+            conf.get("clickhouse_password", ""),
+        )
+        self.store = Store(conf)
+        self.keep_chains = max(1, int(conf.get("keep_chains", 1)))
+        # fnmatch patterns: "*" for every table; an empty list selects none, so
+        # emptying a host's list can't make it back up every table
+        self.include = conf.get("include_tables", [])
+        self.exclude = conf.get("exclude_tables", [])
+        self.poll_s = float(conf.get("poll_interval_seconds", 30))
+        self.clickhouse_wait_s = float(conf.get("clickhouse_wait_seconds", 600))
+        # Query settings for the uploads, e.g. R2 needs every part of a
+        # multipart upload but the last to be the same size
+        self.settings = {k: str(v) for k, v in conf.get("s3_settings", {}).items()}
+        bw = int(conf.get("max_backup_bandwidth", 0))
+        if bw:
+            # A query setting, not a BACKUP setting: 24.8 rejects it inside
+            # BACKUP ... SETTINGS. Against MinIO in testing, the throughput
+            # was about a third of this value.
+            self.settings["max_backup_bandwidth"] = str(bw)
+
+    def tables(self, only: list) -> tuple:
+        """([(name, engine)] to back up, smallest first; [missing names])"""
+        rows = self.ch.rows(
+            f"""
+            SELECT t.name AS name, t.engine AS engine,
+                   coalesce(p.bytes, 0) AS bytes
+            FROM system.tables AS t
+            LEFT JOIN (
+                SELECT table, sum(bytes_on_disk) AS bytes
+                FROM system.parts
+                WHERE active AND database = {sql_str(self.db)}
+                GROUP BY table
+            ) AS p ON p.table = t.name
+            WHERE t.database = {sql_str(self.db)} AND NOT t.is_temporary
+            ORDER BY bytes, name
+            """
+        )
+        out = []
+        for r in rows:
+            name, engine = r["name"], r["engine"]
+            if only and name not in only:
+                continue
+            if name.startswith(".inner"):
+                continue  # backed up together with its materialized view
+            if not self.includes(name):
+                continue  # another job's table
+            if self.excludes(name):
+                log.info(f"{self.name}: {name}: excluded")
+                continue
+            if not SAFE_NAME.match(name):
+                log.error(f"{name}: unsupported characters in table name, skipping")
+                continue
+            out.append((name, engine))
+        # Tables asked for with --table that are this job's but don't exist
+        missing = sorted(
+            n for n in set(only) - {n for n, _ in out} if self.includes(n) and not self.excludes(n)
+        )
+        for name in missing:
+            log.error(f"{self.name}: {name}: not found")
+        return out, missing
+
+    def includes(self, table: str) -> bool:
+        return any(fnmatch.fnmatchcase(table, p) for p in self.include)
+
+    def excludes(self, table: str) -> bool:
+        return any(fnmatch.fnmatchcase(table, p) for p in self.exclude)
+
+    def running_backups(self) -> list:
+        """Backups still running server-side, e.g. from a killed run"""
+        rows = self.ch.rows(
+            "SELECT id, name FROM system.backups WHERE status = 'CREATING_BACKUP'"
+        )
+        # this job's bucket and prefix only: one bucket's name can begin another's
+        ours = f"{self.store.ch_base}/{self.store.prefix}/"
+        return [r for r in rows if ours in r["name"]]
+
+    def wait_for_clickhouse(self) -> None:
+        """At boot a missed run starts before ClickHouse accepts queries"""
+        t0 = time.monotonic()
+        while True:
+            try:
+                self.ch.query("SELECT 1", timeout=30)
+                return
+            except (OSError, RuntimeError) as e:
+                if time.monotonic() - t0 >= self.clickhouse_wait_s:
+                    raise RuntimeError(f"ClickHouse not answering after {self.clickhouse_wait_s:.0f}s: {e}")
+                log.info(f"waiting for ClickHouse: {e}")
+                time.sleep(10)
+
+    def wait(self, op_id: str, what: str) -> dict:
+        t0 = time.monotonic()
+        last_log = t0
+        failing_since = None
+        while True:
+            # A backup can run for days: retry a failed poll for a while
+            # rather than give up on a BACKUP that is still running
+            try:
+                rows = self.ch.rows(
+                    f"""
+                    SELECT status, error, num_files, total_size, uncompressed_size
+                    FROM system.backups WHERE id = {sql_str(op_id)}
+                    """
+                )
+                failing_since = None
+            except (OSError, RuntimeError) as e:
+                failing_since = failing_since or time.monotonic()
+                if time.monotonic() - failing_since > 600:
+                    raise
+                log.warning(f"{what}: polling failed, retrying: {e}")
+                time.sleep(self.poll_s)
+                continue
+            if not rows:
+                raise RuntimeError(f"{what}: gone from system.backups (server restart?)")
+            st = rows[0]
+            if st["status"] == "BACKUP_CREATED":
+                return st
+            if st["status"] != "CREATING_BACKUP":
+                raise RuntimeError(f"{what}: {st['status']}: {st['error']}")
+            now = time.monotonic()
+            if now - last_log >= 1800:
+                log.info(f"{what}: still running after {(now - t0) / 3600:.1f}h")
+                last_log = now
+            time.sleep(self.poll_s)
+
+    def cleanup_incomplete(self, table: str, chains: dict) -> None:
+        tprefix = self.store.table_prefix(self.db, table)
+        for chain, entry in chains.items():
+            if not any(entry["backups"].values()) and not entry["keep"]:
+                n = self.store.delete_prefix(f"{tprefix}{chain}/")
+                log.info(f"{table}: removed incomplete chain {chain} ({n} objects)")
+                continue
+            for backup, ok in entry["backups"].items():
+                if not ok:
+                    n = self.store.delete_prefix(f"{tprefix}{chain}/{backup}/")
+                    log.info(f"{table}: removed incomplete backup {chain}/{backup} ({n} objects)")
+
+    def prune(self, table: str) -> None:
+        """Keep the newest keep_chains complete chains plus pinned ones"""
+        chains = self.store.chains(self.db, table)
+        done = complete_chains(chains)
+        tprefix = self.store.table_prefix(self.db, table)
+        for chain in done[: -self.keep_chains]:
+            if chains[chain]["keep"]:
+                continue
+            n = self.store.delete_prefix(f"{tprefix}{chain}/")
+            log.info(f"{table}: pruned chain {chain} ({n} objects)")
+
+    def backup_table(self, table: str, engine: str, force_full: bool, keep: str) -> bool:
+        """Back up one table; True if that was a full backup"""
+        chains = self.store.chains(self.db, table)
+        self.cleanup_incomplete(table, chains)
+        chains = self.store.chains(self.db, table)
+        base_chain, base = latest_backup(chains)
+
+        now = utcnow()
+        export = engine in EXPORT_ENGINES
+        full = (
+            force_full
+            or base is None
+            or export  # tiny tables: a fresh full backup every run
+            or parse_ts(base_chain).strftime("%Y%m") != now.strftime("%Y%m")
+        )
+        chain = ts(now) if full else base_chain
+        name = ts(now)
+        key = f"{self.store.table_prefix(self.db, table)}{chain}/{name}"
+        kind = "full" if full else f"incremental on {base_chain}/{base}"
+        what = f"{table}: {chain}/{name} ({kind})"
+
+        if export:
+            log.info(f"{table}: exporting {engine} rows")
+            self.ch.query(
+                f"""
+                INSERT INTO FUNCTION s3({self.store.ch_s3(key + "/data.native.zst")}, 'Native')
+                SELECT * FROM `{self.db}`.`{table}`
+                """,
+                settings=self.settings,  # R2 needs equal upload parts here too
+                timeout=3600,
+            )
+
+        sql = f"BACKUP TABLE `{self.db}`.`{table}` TO S3({self.store.ch_s3(key)})"
+        if not full:
+            base_key = f"{self.store.table_prefix(self.db, table)}{base_chain}/{base}"
+            sql += f" SETTINGS base_backup = S3({self.store.ch_s3(base_key)})"
+        sql += " ASYNC"
+
+        log.info(f"{what}: starting")
+        t0 = time.monotonic()
+        out = self.ch.query(sql + "\nFORMAT JSONEachRow", settings=self.settings)
+        op_id = json.loads(out.splitlines()[0])["id"]
+        st = self.wait(op_id, what)
+        secs = time.monotonic() - t0
+        log.info(
+            f"{what}: done in {secs / 60:.1f} min, {st['num_files']} files, "
+            f"{size(int(st['uncompressed_size']))} new data"
+        )
+        self.metrics.table_result(self.name, table, True, st, secs)
+
+        if keep:
+            self.store.put_empty(f"{self.store.table_prefix(self.db, table)}{chain}/KEEP-{keep}")
+            log.info(f"{table}: chain {chain} pinned as {keep}")
+        self.prune(table)
+        return full
+
+    def run(self, only: list, force_full: bool, keep: str) -> int:
+        try:
+            self.wait_for_clickhouse()
+            running = self.running_backups()
+            if running:
+                for r in running:
+                    log.error(f"backup {r['id']} is still running on the server")
+                raise RuntimeError("not starting: wait for it to finish, or for the server to drop it")
+            tables, failed = self.tables(only)
+            if not tables and not only:
+                log.warning(f"{self.name}: no tables to back up (include_tables: {self.include})")
+        except Exception as e:
+            log.error(f"{self.name}: could not start: {e}")
+            self.metrics.job_failed(self.name)
+            return 1
+
+        t0, done, fulls = time.monotonic(), 0, 0
+        self.metrics.job_running(self.name, True)
+        try:
+            for table, engine in tables:
+                try:
+                    fulls += self.backup_table(table, engine, force_full, keep)
+                    done += 1
+                except Exception as e:
+                    log.error(f"{self.name}: {table}: backup failed: {e}")
+                    self.metrics.table_result(self.name, table, False)
+                    failed.append(table)
+        finally:
+            self.metrics.job_running(self.name, False)
+        self.metrics.run_result(self.name, time.monotonic() - t0, done, fulls, len(failed))
+        if not only:
+            self.metrics.forget_tables(self.name, {table for table, _ in tables})
+
+        if failed:
+            log.error(f"{self.name}: failed tables: {', '.join(failed)}")
+            return 1
+        log.info(f"{self.name}: all tables backed up")
+        return 0
+
+    def list(self, only: list) -> int:
+        # The listing is the command's output, so it goes to stdout with
+        # print(), to be read or piped (e.g. the latest: URL into a RESTORE);
+        # log is for what run and release report, to stderr and the journal.
+        tables = only or sorted(
+            n for n in self.store.subdirs(f"{self.store.prefix}/{self.db}/") if n
+        )
+        for table in tables:
+            chains = self.store.chains(self.db, table)
+            for chain in sorted(chains):
+                entry = chains[chain]
+                keep = f"  KEEP: {', '.join(entry['keep'])}" if entry["keep"] else ""
+                print(f"{table}/{chain}{keep}")
+                for backup, ok in sorted(entry["backups"].items()):
+                    status = "" if ok else "  (incomplete)"
+                    print(f"    {backup}{status}")
+            chain, backup = latest_backup(chains)
+            if backup:
+                key = f"{self.store.table_prefix(self.db, table)}{chain}/{backup}"
+                print(f"  latest: {self.store.ch_base}/{key}")
+        return 0
+
+    def release(self, label: str) -> int:
+        n = 0
+        for table in self.store.subdirs(f"{self.store.prefix}/{self.db}/"):
+            if not table:
+                continue
+            for chain, entry in self.store.chains(self.db, table).items():
+                if label in entry["keep"]:
+                    tprefix = self.store.table_prefix(self.db, table)
+                    self.store.delete_key(f"{tprefix}{chain}/KEEP-{label}")
+                    n += 1
+        log.info(f"released {n} chain(s) pinned as {label}; the next run prunes them")
+        return 0
+
+
+def not_backed_up(tables: dict, here: str, jobs: list, excluded: list, elsewhere: list) -> dict:
+    """The tables of {table: [hosts]} that no job here backs up and that
+    aren't excluded (never backed up) or backed up elsewhere (by another
+    host). A job only backs up the tables that exist on this host."""
+    match = lambda t, patterns: any(fnmatch.fnmatchcase(t, p) for p in patterns)
+    return {
+        t: hosts for t, hosts in tables.items()
+        if not (here in hosts and any(j.includes(t) and not j.excludes(t) for j in jobs))
+        and not match(t, excluded) and not match(t, elsewhere)
+    }
+
+
+def check_cluster(conf: dict, metrics: "Metrics") -> None:
+    """Report the tables on any node of the cluster that no backup covers,
+    e.g. a new table created on one node only"""
+    cluster = conf.get("check_cluster", "")
+    if not cluster:
+        return
+    jobs = [BackupJob(j, metrics) for j in job_confs(conf, [])]
+    try:
+        here = jobs[0].ch.query("SELECT hostName()").strip()
+        rows = jobs[0].ch.rows(
+            f"""
+            SELECT name, groupUniqArray(hostName()) AS hosts
+            FROM clusterAllReplicas({sql_str(cluster)}, system.tables)
+            WHERE database = {sql_str(jobs[0].db)} AND NOT is_temporary AND NOT startsWith(name, '.inner')
+            GROUP BY name
+            """
+        )
+    except Exception as e:
+        log.error(f"could not list the tables of cluster {cluster}: {e}")
+        return
+    missing = not_backed_up(
+        {r["name"]: r["hosts"] for r in rows}, here, jobs,
+        conf.get("exclude_tables", []), conf.get("backed_up_elsewhere", []),
+    )
+    for table, hosts in sorted(missing.items()):
+        log.error(f"{table} (on {', '.join(sorted(hosts))}) is in no backup job and not excluded")
+    metrics.not_backed_up(missing)
+
+
+def jobs_for_tables(jobs: list, tables: list) -> tuple:
+    """(the jobs that back up one of tables, the tables no job backs up)"""
+    owns = lambda j, t: j.includes(t) and not j.excludes(t)
+    orphans = [t for t in tables if not any(owns(j, t) for j in jobs)]
+    return [j for j in jobs if any(owns(j, t) for t in tables)], orphans
+
+
+def job_confs(conf: dict, names: list) -> list:
+    """Each job's config: the top-level keys, overridden by the job's own.
+
+    A config without "jobs" is one job named "default", for every table. A
+    job's exclude_tables add to the top-level ones."""
+    common = {k: v for k, v in conf.items() if k != "jobs"}
+    out = []
+    for job in conf.get("jobs") or [{"name": "default", "include_tables": ["*"]}]:
+        merged = {**common, **job}
+        merged["exclude_tables"] = common.get("exclude_tables", []) + job.get("exclude_tables", [])
+        if not names or merged["name"] in names:
+            out.append(merged)
+    unknown = set(names) - {j["name"] for j in out}
+    if unknown:
+        raise SystemExit(f"unknown job(s): {', '.join(sorted(unknown))}")
+    return out
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--config", default=DEFAULT_CONFIG)
+    sub = ap.add_subparsers(dest="command", required=True)
+    p = sub.add_parser("run", help="back up every table")
+    p.add_argument("--full", action="store_true", help="start new chains")
+    p.add_argument("--table", action="append", default=[], help="only this table (repeatable)")
+    p.add_argument("--keep", help="pin the chains used by this run under LABEL")
+    p = sub.add_parser("list", help="show backups in the bucket")
+    p.add_argument("--table", action="append", default=[])
+    p = sub.add_parser("release", help="unpin chains pinned under LABEL")
+    p.add_argument("label")
+    for p in sub.choices.values():
+        p.add_argument("--job", action="append", default=[], help="only this job (repeatable)")
+    args = ap.parse_args()
+
+    logging.basicConfig(
+        stream=sys.stderr, level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s"
+    )
+    log.setLevel(logging.INFO)
+
+    for label in [getattr(args, "keep", None), getattr(args, "label", None)]:
+        if label and not SAFE_NAME.match(label.replace(".", "_").replace("-", "_")):
+            ap.error(f"invalid label: {label}")
+
+    with open(args.config) as f:
+        conf = json.load(f)
+    metrics = Metrics(conf.get("metrics_file", ""), conf.get("state_file", ""))
+    jobs = [BackupJob(j, metrics) for j in job_confs(conf, args.job)]
+    configured = {j["name"] for j in job_confs(conf, [])}
+
+    if args.command == "list":
+        for job in jobs:
+            job.list(args.table)
+        return 0
+
+    if args.command == "run" and args.table:
+        # only these tables' jobs: the others would run with no tables, and report that
+        jobs, orphans = jobs_for_tables(jobs, args.table)
+        if orphans:
+            ap.error(f"no job backs up: {', '.join(orphans)}")
+
+    # Exit cleanly on SIGTERM; a BACKUP already sent keeps running
+    # server-side, and the next run waits for it (see running_backups).
+    signal.signal(signal.SIGTERM, lambda *_: sys.exit(1))
+    lock_path = conf.get("lock_file", "/run/lock/ooni-clickhouse-backup.lock")
+    with open(lock_path, "w") as lock:
+        try:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            log.error(f"another instance holds {lock_path}")
+            return 1
+        rc = 0
+        for job in jobs:
+            if args.command == "run":
+                # A failed job doesn't stop the next one
+                rc |= job.run(args.table, args.full, args.keep)
+            else:
+                rc |= job.release(args.label)
+        if args.command == "run":
+            metrics.forget_jobs(configured)
+            check_cluster(conf, metrics)
+        return rc
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/ansible/roles/clickhouse_backup/tasks/main.yml b/ansible/roles/clickhouse_backup/tasks/main.yml
new file mode 100644
index 00000000..77afea76
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tasks/main.yml
@@ -0,0 +1,75 @@
+---
+- name: Check backup bucket settings
+  ansible.builtin.assert:
+    that:
+      - clickhouse_backup_s3_endpoint | length > 0
+      - clickhouse_backup_s3_access_key_id | length > 0
+      - clickhouse_backup_s3_secret_access_key | length > 0
+      - clickhouse_backup_jobs | length > 0
+      - clickhouse_backup_jobs | rejectattr('s3_bucket', 'truthy') | list | length == 0
+      - clickhouse_backup_jobs | map(attribute='name') | unique | list | length == clickhouse_backup_jobs | length
+    fail_msg: set clickhouse_backup_s3_endpoint, the bucket credentials, and an s3_bucket and a unique name for every job
+
+- name: Create the directories for the backup state and the node_exporter textfile
+  ansible.builtin.file:
+    path: "{{ item }}"
+    state: directory
+    owner: root
+    group: root
+    mode: "0755"
+  loop:
+    - "{{ clickhouse_backup_state_file | dirname }}"
+    - "{{ clickhouse_backup_metrics_file | dirname }}"
+
+- name: Install dependencies for the backup script
+  ansible.builtin.apt:
+    pkg:
+      - python3-boto3
+
+- name: Give ClickHouse the backup bucket credentials
+  ansible.builtin.template:
+    src: ooni_backup_s3.xml.j2
+    dest: /etc/clickhouse-server/config.d/ooni_backup_s3.xml
+    owner: clickhouse
+    group: clickhouse
+    mode: "0600"
+
+- name: Ensure /etc/ooni exists
+  ansible.builtin.file:
+    path: /etc/ooni
+    state: directory
+    mode: "0755"
+
+- name: Configure the backup script
+  ansible.builtin.template:
+    src: clickhouse-backup.json.j2
+    dest: /etc/ooni/clickhouse-backup.json
+    owner: root
+    group: root
+    mode: "0600"
+
+- name: Install the backup script
+  ansible.builtin.copy:
+    src: ooni_clickhouse_backup.py
+    dest: /usr/local/bin/ooni-clickhouse-backup
+    owner: root
+    group: root
+    mode: "0755"
+
+- name: Install the backup service and timer
+  ansible.builtin.template:
+    src: "{{ item }}"
+    dest: "/etc/systemd/system/{{ item }}"
+    owner: root
+    group: root
+    mode: "0644"
+  loop:
+    - ooni-clickhouse-backup.service
+    - ooni-clickhouse-backup.timer
+
+- name: Set the backup timer state
+  ansible.builtin.systemd:
+    name: ooni-clickhouse-backup.timer
+    daemon_reload: true
+    state: "{{ 'started' if clickhouse_backup_enabled else 'stopped' }}"
+    enabled: "{{ clickhouse_backup_enabled }}"
diff --git a/ansible/roles/clickhouse_backup/templates/clickhouse-backup.json.j2 b/ansible/roles/clickhouse_backup/templates/clickhouse-backup.json.j2
new file mode 100644
index 00000000..01c7b3ca
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/templates/clickhouse-backup.json.j2
@@ -0,0 +1,20 @@
+{{ {
+  "clickhouse_url": clickhouse_backup_clickhouse_url,
+  "clickhouse_user": clickhouse_backup_clickhouse_user,
+  "clickhouse_password": clickhouse_backup_clickhouse_password,
+  "database": clickhouse_backup_database,
+  "s3_endpoint": clickhouse_backup_s3_endpoint,
+  "s3_region": clickhouse_backup_s3_region,
+  "s3_prefix": clickhouse_backup_s3_prefix,
+  "s3_access_key_id": clickhouse_backup_s3_access_key_id,
+  "s3_secret_access_key": clickhouse_backup_s3_secret_access_key,
+  "s3_settings": clickhouse_backup_s3_settings,
+  "keep_chains": clickhouse_backup_keep_chains,
+  "max_backup_bandwidth": clickhouse_backup_max_bandwidth,
+  "exclude_tables": clickhouse_backup_exclude_tables,
+  "check_cluster": clickhouse_backup_check_cluster,
+  "backed_up_elsewhere": clickhouse_backup_backed_up_elsewhere,
+  "metrics_file": clickhouse_backup_metrics_file,
+  "state_file": clickhouse_backup_state_file,
+  "jobs": clickhouse_backup_jobs,
+} | to_nice_json }}
diff --git a/ansible/roles/clickhouse_backup/templates/ooni-clickhouse-backup.service b/ansible/roles/clickhouse_backup/templates/ooni-clickhouse-backup.service
new file mode 100644
index 00000000..d414389d
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/templates/ooni-clickhouse-backup.service
@@ -0,0 +1,14 @@
+[Unit]
+Description=Back up ClickHouse tables to object storage
+Wants=ooni-clickhouse-backup.timer
+After=clickhouse-server.service
+
+[Service]
+Type=oneshot
+ExecStart=/usr/local/bin/ooni-clickhouse-backup run
+SyslogIdentifier=ooni-clickhouse-backup
+# A new full backup of the largest tables takes more than a day
+TimeoutStartSec=infinity
+
+[Install]
+WantedBy=multi-user.target
diff --git a/ansible/roles/clickhouse_backup/templates/ooni-clickhouse-backup.timer b/ansible/roles/clickhouse_backup/templates/ooni-clickhouse-backup.timer
new file mode 100644
index 00000000..5b599270
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/templates/ooni-clickhouse-backup.timer
@@ -0,0 +1,10 @@
+[Unit]
+Description=Back up ClickHouse tables to object storage
+
+[Timer]
+Unit=ooni-clickhouse-backup.service
+OnCalendar={{ clickhouse_backup_on_calendar }}
+Persistent=true
+
+[Install]
+WantedBy=timers.target
diff --git a/ansible/roles/clickhouse_backup/templates/ooni_backup_s3.xml.j2 b/ansible/roles/clickhouse_backup/templates/ooni_backup_s3.xml.j2
new file mode 100644
index 00000000..808c000b
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/templates/ooni_backup_s3.xml.j2
@@ -0,0 +1,19 @@
+<?xml version="1.0"?>
+<!-- {{ ansible_managed }}
+     Credentials for the backup buckets, one endpoint per bucket. BACKUP/RESTORE
+     and the s3 function use them for any URL under the endpoint, so queries
+     never carry keys: an incremental backup stores its base_backup argument
+     in plain text in the bucket. ClickHouse reloads this file without a
+     restart. -->
+<clickhouse>
+    <s3>
+{% for bucket in clickhouse_backup_jobs | map(attribute='s3_bucket') | unique %}
+        <ooni_backup_{{ loop.index }}>
+            <endpoint>{{ clickhouse_backup_s3_endpoint | regex_replace('/$', '') }}/{{ bucket }}/</endpoint>
+            <region>{{ clickhouse_backup_s3_region }}</region>
+            <access_key_id>{{ clickhouse_backup_s3_access_key_id | e }}</access_key_id>
+            <secret_access_key>{{ clickhouse_backup_s3_secret_access_key | e }}</secret_access_key>
+        </ooni_backup_{{ loop.index }}>
+{% endfor %}
+    </s3>
+</clickhouse>
diff --git a/ansible/roles/clickhouse_backup/tests/.gitignore b/ansible/roles/clickhouse_backup/tests/.gitignore
new file mode 100644
index 00000000..9d931c43
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/.gitignore
@@ -0,0 +1 @@
+work/
diff --git a/ansible/roles/clickhouse_backup/tests/README.md b/ansible/roles/clickhouse_backup/tests/README.md
new file mode 100644
index 00000000..cf2e9b50
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/README.md
@@ -0,0 +1,40 @@
+# Tests for the clickhouse_backup role
+
+`./run.sh` runs them all, in about nine minutes:
+
+- `alert_rules_test.yml`: promtool unit tests of the backup alerts in
+  `roles/prometheus/files/alert_rules.yml`, then amtool checks that
+  Alertmanager's config is valid and routes them to the right receivers.
+- `test_schedule.py`: which runs start a full backup and what gets pruned,
+  over months of daily and weekly runs with failures, ClickHouse and the
+  bucket stubbed.
+- `test_job.py`: how a job runs and what it reports, ClickHouse stubbed.
+- `test_e2e.py`: the real thing, in the containers of `compose.yml`:
+  two ClickHouse nodes at production's version (`data3`, where the backups
+  run, and `data1`, in one test cluster), MinIO standing in for R2, the script
+  on Debian bookworm with its python3-boto3, as on the hosts, node_exporter,
+  Prometheus with the repo's alert rules, Alertmanager with the repo's
+  routes, and a stand-in for Slack. Full and incremental backups, and
+  restoring the full and each incremental of a chain, and an
+  EmbeddedRocksDB table with its exported rows; a table created on `data1`
+  only, which ClickHouseTableNotBackedUp must name; a failed run; an
+  interrupted full, the month's new full, pruning and restoring what's
+  left; and the Slack messages and emails each one causes.
+
+The configs are rendered from the role's templates into `work/` by
+`render.yml`, with the variables in `vars.yml`.
+
+Needs docker compose, python3 with boto3, and ansible-core. With rootless
+podman:
+
+```bash
+DOCKER_HOST=unix:///run/user/$(id -u)/podman/podman.sock ./run.sh
+```
+
+`KEEP=1 ./run.sh` leaves the containers up afterwards (Prometheus on
+127.0.0.1:19990, Alertmanager on 19993); remove them with
+`docker compose -f compose.yml down -v`.
+
+GitHub runs it (`.github/workflows/test_clickhouse_backup.yml`) when a pull
+request or a push to main changes this role, the alert rules or
+Alertmanager's role.
diff --git a/ansible/roles/clickhouse_backup/tests/alert_rules_test.yml b/ansible/roles/clickhouse_backup/tests/alert_rules_test.yml
new file mode 100644
index 00000000..72d1759a
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/alert_rules_test.yml
@@ -0,0 +1,130 @@
+# promtool unit tests for the ClickHouse backup alerts in alert_rules.yml, and for the backup file's
+# exclusion from TextfileJam. Test time starts at 0 (1970-01-01), so a timestamp N days before the test
+# starts is -N * 86400.
+rule_files: [alert_rules.yml]
+evaluation_interval: 1m
+tests:
+  # Completed and Failed
+  - interval: 1m
+    input_series:
+      # measurements finished at t=0 after 30 min, all 14 tables backed up, 1 in full
+      - {series: 'ooni_clickhouse_backup_last_run_finished_timestamp_seconds{instance="data3", backup="measurements"}', values: '0x180'}
+      - {series: 'ooni_clickhouse_backup_last_run_duration_seconds{instance="data3", backup="measurements"}', values: '1800x180'}
+      - {series: 'ooni_clickhouse_backup_last_run_failed_tables{instance="data3", backup="measurements"}', values: '0x180'}
+      - {series: 'ooni_clickhouse_backup_last_run_tables{instance="data3", backup="measurements"}', values: '14x180'}
+      - {series: 'ooni_clickhouse_backup_last_run_full_backups{instance="data3", backup="measurements"}', values: '1x180'}
+      # private finished at the same time, with fastpath failed
+      - {series: 'ooni_clickhouse_backup_last_run_finished_timestamp_seconds{instance="data3", backup="private"}', values: '0x180'}
+      - {series: 'ooni_clickhouse_backup_last_run_duration_seconds{instance="data3", backup="private"}', values: '60x180'}
+      - {series: 'ooni_clickhouse_backup_last_run_failed_tables{instance="data3", backup="private"}', values: '1x180'}
+      - {series: 'ooni_clickhouse_backup_last_attempt_failed{instance="data3", backup="private", table="fastpath"}', values: '1x180'}
+    alert_rule_test:
+      - eval_time: 10m
+        alertname: ClickHouseBackupCompleted
+        exp_alerts:
+          - exp_labels: {instance: data3, backup: measurements}
+            exp_annotations:
+              summary: 'ClickHouse backup measurements on data3 completed in 30m 0s'
+              description: 'Tables backed up: 14, in full: 1. Nothing to do.'
+      # an hour after the run it clears
+      - eval_time: 2h
+        alertname: ClickHouseBackupCompleted
+        exp_alerts: []
+      - eval_time: 10m
+        alertname: ClickHouseBackupFailed
+        exp_alerts:
+          - exp_labels: {instance: data3, backup: private, table: fastpath}
+            exp_annotations:
+              summary: 'ClickHouse backup of fastpath (private) failed on data3'
+              description: 'Nothing was deleted: the older backups of fastpath are kept. The next run retries it. See `journalctl --identifier ooni-clickhouse-backup` on data3.'
+
+  # Stale: after 8 days, unless a job on the same host is in a run started in the last 5 days
+  - interval: 1m
+    input_series:
+      # data1: nothing running; one table backed up 9 days ago, one 7 days ago
+      - {series: 'ooni_clickhouse_backup_running_since_timestamp_seconds{instance="data1", backup="data1"}', values: '0x30'}
+      - {series: 'ooni_clickhouse_backup_last_success_timestamp_seconds{instance="data1", backup="data1", table="nine_days"}', values: '-777600x30'}
+      - {series: 'ooni_clickhouse_backup_last_success_timestamp_seconds{instance="data1", backup="data1", table="seven_days"}', values: '-604800x30'}
+      # data3: measurements' monthly full started 2 days ago; private, waiting behind it, last ran 9 days ago
+      - {series: 'ooni_clickhouse_backup_running_since_timestamp_seconds{instance="data3", backup="measurements"}', values: '-172800x30'}
+      - {series: 'ooni_clickhouse_backup_running_since_timestamp_seconds{instance="data3", backup="private"}', values: '0x30'}
+      - {series: 'ooni_clickhouse_backup_last_success_timestamp_seconds{instance="data3", backup="private", table="msmt_feedback"}', values: '-777600x30'}
+      # stuck: a run that "started" 6 days ago, after the script was killed; its table is 9 days old
+      - {series: 'ooni_clickhouse_backup_running_since_timestamp_seconds{instance="stuck", backup="measurements"}', values: '-518400x30'}
+      - {series: 'ooni_clickhouse_backup_last_success_timestamp_seconds{instance="stuck", backup="measurements", table="fastpath"}', values: '-777600x30'}
+    alert_rule_test:
+      - eval_time: 10m
+        alertname: ClickHouseBackupStale
+        exp_alerts:
+          - exp_labels: {instance: data1, backup: data1, table: nine_days}
+            exp_annotations:
+              summary: 'No ClickHouse backup of nine_days (data1) for 9d 0h 10m 0s'
+              description: 'Check `systemctl status ooni-clickhouse-backup.timer` and the journal on data1.'
+          - exp_labels: {instance: stuck, backup: measurements, table: fastpath}
+            exp_annotations:
+              summary: 'No ClickHouse backup of fastpath (measurements) for 9d 0h 10m 0s'
+              description: 'Check `systemctl status ooni-clickhouse-backup.timer` and the journal on stuck.'
+
+  # Missing: data1 has results, data3 never wrote any; it fires after 8 days
+  - interval: 1h
+    input_series:
+      - {series: 'ooni_clickhouse_backup_running{instance="data1.htz-fsn.prod.ooni.nu", backup="data1"}', values: '0x240'}
+    alert_rule_test:
+      - {eval_time: 7d, alertname: ClickHouseBackupMissing, exp_alerts: []}
+      - eval_time: 9d
+        alertname: ClickHouseBackupMissing
+        exp_alerts:
+          - exp_labels: {instance: data3.htz-fsn.prod.ooni.nu}
+            exp_annotations:
+              summary: 'No ClickHouse backup results from data3.htz-fsn.prod.ooni.nu for over a week'
+
+  # HostLoadHigh: data3's 8 CPUs with a load of 5
+  - interval: 1m
+    input_series:
+      - {series: 'node_load5{instance="data3.htz-fsn.prod.ooni.nu"}', values: '5x60'}
+      - {series: 'node_cpu_seconds_total{instance="data3.htz-fsn.prod.ooni.nu", cpu="0", mode="idle"}', values: '0+1x60'}
+      - {series: 'node_cpu_seconds_total{instance="data3.htz-fsn.prod.ooni.nu", cpu="1", mode="idle"}', values: '0+1x60'}
+      - {series: 'node_cpu_seconds_total{instance="data3.htz-fsn.prod.ooni.nu", cpu="2", mode="idle"}', values: '0+1x60'}
+      - {series: 'node_cpu_seconds_total{instance="data3.htz-fsn.prod.ooni.nu", cpu="3", mode="idle"}', values: '0+1x60'}
+      - {series: 'node_cpu_seconds_total{instance="data3.htz-fsn.prod.ooni.nu", cpu="4", mode="idle"}', values: '0+1x60'}
+      - {series: 'node_cpu_seconds_total{instance="data3.htz-fsn.prod.ooni.nu", cpu="5", mode="idle"}', values: '0+1x60'}
+      - {series: 'node_cpu_seconds_total{instance="data3.htz-fsn.prod.ooni.nu", cpu="6", mode="idle"}', values: '0+1x60'}
+      - {series: 'node_cpu_seconds_total{instance="data3.htz-fsn.prod.ooni.nu", cpu="7", mode="idle"}', values: '0+1x60'}
+    alert_rule_test:
+      - {eval_time: 10m, alertname: ClickHouseBackupHostLoadHigh, exp_alerts: []}
+      - eval_time: 20m
+        alertname: ClickHouseBackupHostLoadHigh
+        exp_alerts:
+          - exp_labels: {instance: data3.htz-fsn.prod.ooni.nu}
+            exp_annotations:
+              summary: 'data3 load is 62.5% of its CPUs'
+              description: 'If a backup is running (ooni_clickhouse_backup_running on data3), lower clickhouse_backup_max_bandwidth; otherwise check merges and queries.'
+
+  # TextfileJam leaves out the backup's file, which changes only once a week. node_exporter's file label
+  # is the file's full path.
+  - interval: 1m
+    input_series:
+      - {series: 'node_time{instance="data3", job="node"}', values: '0+60x240'}
+      - {series: 'node_textfile_mtime{instance="data3", job="node", file="/var/lib/node_exporter/textfile_collector/ooni_clickhouse_backup.prom"}', values: '0x240'}
+      - {series: 'node_textfile_mtime{instance="data3", job="node", file="/var/lib/node_exporter/textfile_collector/other.prom"}', values: '0x240'}
+    alert_rule_test:
+      - eval_time: 3h
+        alertname: TextfileJam
+        exp_alerts:
+          - exp_labels: {instance: data3, job: node, file: /var/lib/node_exporter/textfile_collector/other.prom}
+            exp_annotations:
+              summary: 'Stale `textfile` at data3'
+              description: '`/var/lib/node_exporter/textfile_collector/other.prom` is 3h old.'
+
+  # TableNotBackedUp: a table no backup job covers
+  - interval: 1m
+    input_series:
+      - {series: 'ooni_clickhouse_backup_table_not_backed_up{instance="data3", table="made_on_data1", hosts="data1.htz-fsn.prod.ooni.nu"}', values: '1x30'}
+    alert_rule_test:
+      - eval_time: 5m
+        alertname: ClickHouseTableNotBackedUp
+        exp_alerts:
+          - exp_labels: {instance: data3, table: made_on_data1, hosts: data1.htz-fsn.prod.ooni.nu}
+            exp_annotations:
+              summary: 'ClickHouse table made_on_data1 is not backed up (on data1.htz-fsn.prod.ooni.nu)'
+              description: 'No backup job covers it and it is not excluded. Add it to a job (clickhouse_backup_data1_tables if only data1 holds its data, clickhouse_backup_private_tables if private), or to clickhouse_backup_exclude_tables if it needs no backup. See the runbook, "ClickHouse backup".'
diff --git a/ansible/roles/clickhouse_backup/tests/clickhouse.xml b/ansible/roles/clickhouse_backup/tests/clickhouse.xml
new file mode 100644
index 00000000..4410ded5
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/clickhouse.xml
@@ -0,0 +1,16 @@
+<clickhouse>
+    <!-- Keep the test server small -->
+    <metric_log remove="1"/>
+    <asynchronous_metric_log remove="1"/>
+    <trace_log remove="1"/>
+    <max_server_memory_usage_to_ram_ratio>0.5</max_server_memory_usage_to_ram_ratio>
+    <!-- The two test nodes, for the check of tables no backup covers -->
+    <remote_servers>
+        <test_cluster>
+            <shard>
+                <replica><host>clickhouse</host><port>9000</port><user>admin</user><password>admin</password></replica>
+                <replica><host>clickhouse2</host><port>9000</port><user>admin</user><password>admin</password></replica>
+            </shard>
+        </test_cluster>
+    </remote_servers>
+</clickhouse>
diff --git a/ansible/roles/clickhouse_backup/tests/compose.yml b/ansible/roles/clickhouse_backup/tests/compose.yml
new file mode 100644
index 00000000..0f8eea78
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/compose.yml
@@ -0,0 +1,82 @@
+# The backup's environment: ClickHouse at production's version, MinIO in place
+# of R2, the backup script in "runner", node_exporter reading its textfile,
+# Prometheus with the repo's alert rules, Alertmanager with the repo's routes,
+# and a stand-in for Slack. run.sh renders work/ first.
+name: chbackup-test
+services:
+  minio:
+    image: docker.io/pgsty/minio:RELEASE.2026-08-04T00-00-00Z
+    command: server /data
+    environment:
+      MINIO_ROOT_USER: testkey
+      MINIO_ROOT_PASSWORD: testsecret123
+    ports: ["127.0.0.1:19910:9000"]
+
+  # data3, the node the backups run against
+  clickhouse:
+    image: docker.io/clickhouse/clickhouse-server:24.8.6.70
+    hostname: data3
+    environment:
+      CLICKHOUSE_USER: admin
+      CLICKHOUSE_PASSWORD: admin
+    volumes:
+      - ./work/ooni_backup_s3.xml:/etc/clickhouse-server/config.d/ooni_backup_s3.xml:ro,z
+      - ./clickhouse.xml:/etc/clickhouse-server/config.d/test.xml:ro,z
+    ulimits:
+      nofile: 262144
+    mem_limit: 2g
+    ports: ["127.0.0.1:19911:8123"]
+
+  # another node of the cluster, where a table nobody backs up gets created
+  clickhouse2:
+    image: docker.io/clickhouse/clickhouse-server:24.8.6.70
+    hostname: data1
+    environment:
+      CLICKHOUSE_USER: admin
+      CLICKHOUSE_PASSWORD: admin
+    volumes:
+      - ./clickhouse.xml:/etc/clickhouse-server/config.d/test.xml:ro,z
+    ulimits:
+      nofile: 262144
+    mem_limit: 1g
+    ports: ["127.0.0.1:19912:8123"]
+
+  runner:
+    build: {context: ., dockerfile: runner.Dockerfile}
+    command: sleep infinity
+    volumes:
+      - ../files:/role:ro,z
+      - ./work:/work:ro,z
+      - ./work/textfile:/var/lib/node_exporter/textfile_collector:z
+
+  node_exporter:
+    image: quay.io/prometheus/node-exporter:v1.8.2
+    command: [--collector.disable-defaults, --collector.textfile, --collector.textfile.directory=/textfile]
+    volumes:
+      - ./work/textfile:/textfile:ro,z
+
+  # monitoring.ooni.org runs Debian's prometheus 2.42.0
+  prometheus:
+    image: docker.io/prom/prometheus:v2.42.0
+    command: [--config.file=/etc/prometheus/prometheus.yml]
+    volumes:
+      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro,z
+      - ../../prometheus/files/alert_rules.yml:/etc/prometheus/alert_rules.yml:ro,z
+      - ./alert_rules_test.yml:/etc/prometheus/alert_rules_test.yml:ro,z
+    ports: ["127.0.0.1:19990:9090"]
+
+  # and Debian's alertmanager 0.25.0
+  alertmanager:
+    image: docker.io/prom/alertmanager:v0.25.0
+    command: [--config.file=/etc/alertmanager/alertmanager.yml]
+    volumes:
+      - ./work/alertmanager.yml:/etc/alertmanager/alertmanager.yml:ro,z
+      - ../../prometheus_alertmanager/files/templates:/etc/prometheus/alertmanager_templates:ro,z
+    ports: ["127.0.0.1:19993:9093"]
+
+  slack:
+    image: docker.io/library/python:3.11-slim-bookworm
+    command: [python3, /slack.py]
+    volumes:
+      - ./slack.py:/slack.py:ro,z
+    ports: ["127.0.0.1:19980:8080"]
diff --git a/ansible/roles/clickhouse_backup/tests/prometheus.yml b/ansible/roles/clickhouse_backup/tests/prometheus.yml
new file mode 100644
index 00000000..6e86326b
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/prometheus.yml
@@ -0,0 +1,14 @@
+global:
+  scrape_interval: 5s
+  evaluation_interval: 5s
+rule_files: [alert_rules.yml]
+alerting:
+  alertmanagers:
+    - static_configs: [{targets: ["alertmanager:9093"]}]
+scrape_configs:
+  # The backup's textfile, labelled as data3 like the production scrape
+  - job_name: node
+    static_configs: [{targets: ["node_exporter:9100"]}]
+    relabel_configs:
+      - target_label: instance
+        replacement: data3.htz-fsn.prod.ooni.nu
diff --git a/ansible/roles/clickhouse_backup/tests/render.yml b/ansible/roles/clickhouse_backup/tests/render.yml
new file mode 100644
index 00000000..62c84e3a
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/render.yml
@@ -0,0 +1,35 @@
+# Renders the role's templates and Alertmanager's into work/, with vars.yml
+- hosts: localhost
+  connection: local
+  gather_facts: false
+  vars_files:
+    - ../defaults/main.yml
+    - vars.yml
+  tasks:
+    - name: Render the backup script's configs
+      ansible.builtin.template:
+        src: "{{ playbook_dir }}/../templates/clickhouse-backup.json.j2"
+        dest: "{{ playbook_dir }}/work/{{ item.name }}.json"
+        mode: "0644"
+      vars:
+        clickhouse_backup_jobs: "{{ item.jobs }}"
+        clickhouse_backup_max_bandwidth: "{{ item.bandwidth }}"
+        clickhouse_backup_check_cluster: "{{ item.check_cluster | default('') }}"
+      loop: "{{ test_configs }}"
+
+    - name: Render ClickHouse's bucket credentials, for every bucket of every config
+      ansible.builtin.template:
+        src: "{{ playbook_dir }}/../templates/ooni_backup_s3.xml.j2"
+        dest: "{{ playbook_dir }}/work/ooni_backup_s3.xml"
+        mode: "0644"
+      vars:
+        clickhouse_backup_jobs: "{{ test_configs | map(attribute='jobs') | flatten }}"
+
+    - name: Render Alertmanager's config, with Slack and the mail server stood in for
+      ansible.builtin.template:
+        src: "{{ playbook_dir }}/work/alertmanager.yml.j2"
+        dest: "{{ playbook_dir }}/work/alertmanager.yml"
+        mode: "0644"
+      vars:
+        alertmanager_slack_api_url: http://slack:8080/
+        am_mx_openobservatory: mx.invalid
diff --git a/ansible/roles/clickhouse_backup/tests/run.sh b/ansible/roles/clickhouse_backup/tests/run.sh
new file mode 100755
index 00000000..c45abe8f
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/run.sh
@@ -0,0 +1,36 @@
+#!/bin/bash
+# Tests the clickhouse_backup role: renders its configs from the role's templates, checks the backup alerts
+# (promtool) and their Alertmanager routes (amtool), runs the schedule tests, then test_e2e.py against the
+# containers in compose.yml. Needs docker compose, python3-boto3 and ansible-core. With rootless podman,
+# point DOCKER_HOST at its socket. KEEP=1 leaves the containers running.
+set -euo pipefail
+cd "$(dirname "$0")"
+export COMPOSE=${COMPOSE:-docker compose}
+export PYTHONDONTWRITEBYTECODE=1  # no __pycache__ left in the role's files/
+compose() { $COMPOSE -f compose.yml "$@"; }
+
+echo "== rendering the configs"
+rm -rf work && mkdir -p work/textfile
+# The template looks up openobservatory.org's MX record for the mail server; render.yml sets one instead
+sed '/lookup(.dig./d' ../../prometheus_alertmanager/templates/alertmanager.yml > work/alertmanager.yml.j2
+ansible-playbook -i localhost, render.yml > work/render.log || { cat work/render.log; exit 1; }
+
+echo "== alert rules and routes"
+compose run --rm --no-deps --entrypoint promtool prometheus test rules /etc/prometheus/alert_rules_test.yml
+compose run --rm --no-deps --entrypoint amtool alertmanager check-config /etc/alertmanager/alertmanager.yml
+route() {  # alert labels, receiver it must reach
+  compose run --rm --no-deps --entrypoint amtool alertmanager config routes test \
+    --config.file=/etc/alertmanager/alertmanager.yml --verify.receivers="$2" $1
+}
+route "alertname=ClickHouseBackupCompleted instance=data3 backup=measurements" slack-notice
+route "alertname=ClickHouseBackupFailed instance=data3 backup=measurements table=fastpath" team-all
+
+echo "== schedule"
+python3 -m unittest test_schedule test_job
+
+echo "== end to end"
+# the containers' logs if it fails, then remove them, unless KEEP is set
+trap 'rc=$?; [ $rc = 0 ] || compose logs --no-color --tail=200; [ -n "${KEEP:-}" ] || compose down -v --remove-orphans >/dev/null 2>&1' EXIT
+compose down -v --remove-orphans >/dev/null 2>&1  # left from a run with KEEP=1
+compose up -d --build --quiet-pull
+python3 test_e2e.py
diff --git a/ansible/roles/clickhouse_backup/tests/runner.Dockerfile b/ansible/roles/clickhouse_backup/tests/runner.Dockerfile
new file mode 100644
index 00000000..3abc14ab
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/runner.Dockerfile
@@ -0,0 +1,6 @@
+# Debian bookworm with its python3 and python3-boto3, which the role installs on the ClickHouse hosts
+FROM docker.io/library/debian:bookworm-slim
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends python3 python3-boto3 \
+    && rm -rf /var/lib/apt/lists/* \
+    && mkdir -p /run/lock
diff --git a/ansible/roles/clickhouse_backup/tests/slack.py b/ansible/roles/clickhouse_backup/tests/slack.py
new file mode 100644
index 00000000..85ff0e93
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/slack.py
@@ -0,0 +1,27 @@
+"""Stands in for Slack's incoming webhook: keeps what Alertmanager posts, and
+returns it all, as a JSON list, to GET."""
+import json
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+messages = []
+
+
+class Handler(BaseHTTPRequestHandler):
+    def do_POST(self):
+        messages.append(json.loads(self.rfile.read(int(self.headers["Content-Length"]))))
+        self.reply(b"ok")
+
+    def do_GET(self):
+        self.reply(json.dumps(messages).encode())
+
+    def reply(self, body):
+        self.send_response(200)
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def log_message(self, *args):
+        pass
+
+
+ThreadingHTTPServer(("", 8080), Handler).serve_forever()
diff --git a/ansible/roles/clickhouse_backup/tests/test_e2e.py b/ansible/roles/clickhouse_backup/tests/test_e2e.py
new file mode 100644
index 00000000..eb70e3ec
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/test_e2e.py
@@ -0,0 +1,282 @@
+"""End-to-end test of the backup in compose.yml's environment, after run.sh has rendered work/ and started it.
+In order:
+  1. a first run takes a full backup of every table, each job to its own bucket
+  2. ClickHouseBackupCompleted reaches Slack once per job, as a notice, without email
+  3. two more runs take incrementals; restoring the full and each incremental returns the rows of
+     its moment, and an EmbeddedRocksDB table comes back with its exported rows
+  4. a table created on the cluster's other node only: ClickHouseTableNotBackedUp names it
+  5. a run whose bucket is missing fails, and ClickHouseBackupFailed reaches Slack and email
+  6. a new month's full interrupted by killing ClickHouse deletes nothing; the next run
+     removes what it left, takes the full and prunes the oldest chain, and the full restores
+Prints each check; exits 1 at the first failure."""
+import json
+import os
+import shlex
+import subprocess
+import sys
+import time
+import urllib.request
+from datetime import datetime, timezone
+
+import boto3
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+COMPOSE = shlex.split(os.environ.get("COMPOSE", "docker compose")) + ["-f", os.path.join(HERE, "compose.yml")]
+MEASUREMENTS, PRIVATE = "ooni-clickhouse-backup", "ooni-clickhouse-backup-private"
+DATA3 = "data3.htz-fsn.prod.ooni.nu"  # the instance label prometheus.yml gives the textfile
+s3 = boto3.client(
+    "s3", endpoint_url="http://127.0.0.1:19910", region_name="us-east-1",
+    aws_access_key_id="testkey", aws_secret_access_key="testsecret123",
+)
+
+
+def check(ok, what, output=""):
+    print(f"  {'ok  ' if ok else 'FAIL'} {what}")
+    if not ok:
+        if output:
+            print("  the script said:\n" + output[-3000:])
+        sys.exit(1)
+
+
+def http(url, data=None):
+    with urllib.request.urlopen(url, data=data, timeout=30) as r:
+        return r.read().decode()
+
+
+def ch(sql, port=19911):
+    """a query on data3, or with port=19912 on the cluster's other node, data1"""
+    return http(f"http://127.0.0.1:{port}/?user=admin&password=admin", sql.encode()).strip()
+
+
+def backup(config, *args):
+    """Run the backup script in the runner container, as the timer runs it on data3"""
+    cmd = COMPOSE + ["exec", "-T", "runner", "python3", "/role/ooni_clickhouse_backup.py",
+                     "--config", f"/work/{config}.json", *args]
+    return subprocess.run(cmd, capture_output=True, text=True)
+
+
+def keys(bucket, prefix):
+    pages = s3.get_paginator("list_objects_v2").paginate(Bucket=bucket, Prefix=prefix)
+    return [o["Key"] for page in pages for o in page.get("Contents", [])]
+
+
+def chains(bucket, table):
+    """{chain: {backup: complete}} under prod/ooni/<table>/"""
+    out = {}
+    for key in keys(bucket, f"prod/ooni/{table}/"):
+        chain, name = key.split("/")[3:5]
+        backups = out.setdefault(chain, {})
+        backups[name] = backups.get(name, False) or key.endswith(f"/{name}/.backup")
+    return out
+
+
+def copy_prefix(bucket, src, dst):
+    for key in keys(bucket, src):
+        s3.copy_object(Bucket=bucket, Key=dst + key[len(src):], CopySource={"Bucket": bucket, "Key": key})
+
+
+def delete_prefix(bucket, prefix):
+    for key in keys(bucket, prefix):
+        s3.delete_object(Bucket=bucket, Key=key)
+
+
+def wait_for(what, get, timeout=150):
+    """Poll get() until it returns something truthy"""
+    t0 = time.monotonic()
+    while time.monotonic() - t0 < timeout:
+        try:
+            value = get()
+        except OSError:
+            value = None
+        if value:
+            return value
+        time.sleep(2)
+    check(False, f"{what}, within {timeout}s")
+
+
+def slack():
+    """(title, text) of every message the Slack stand-in has received"""
+    return [(a["title"].strip(), a.get("text", "").strip()) for m in json.loads(http("http://127.0.0.1:19980/"))
+            for a in m.get("attachments", [])]
+
+
+def email_attempts():
+    lines = http("http://127.0.0.1:19993/metrics").splitlines()
+    return sum(float(l.split()[-1]) for l in lines
+               if l.startswith('alertmanager_notifications_total{integration="email"'))
+
+
+def months_ago(n):
+    now = datetime.now(timezone.utc)
+    y, m = divmod(now.year * 12 + now.month - 1 - n, 12)
+    return f"{y}{m + 1:02}01T023000Z"
+
+
+def setup():
+    print("== waiting for the services")
+    wait_for("ClickHouse", lambda: ch("SELECT 1") == "1")
+    wait_for("ClickHouse's other node", lambda: ch("SELECT 1", port=19912) == "1")
+    wait_for("MinIO", lambda: http("http://127.0.0.1:19910/minio/health/live") == "")
+    wait_for("Prometheus", lambda: "Ready" in http("http://127.0.0.1:19990/-/ready"))
+    wait_for("Alertmanager", lambda: "OK" in http("http://127.0.0.1:19993/-/ready"))
+    wait_for("Slack stand-in", lambda: http("http://127.0.0.1:19980/") == "[]")
+    for bucket in (MEASUREMENTS, PRIVATE):
+        s3.create_bucket(Bucket=bucket)
+    for sql in [
+        "CREATE DATABASE ooni",
+        "CREATE TABLE ooni.fastpath (measurement_uid String, input String, anomaly LowCardinality(String)) "
+        "ENGINE = MergeTree ORDER BY measurement_uid",
+        "INSERT INTO ooni.fastpath SELECT toString(number), 'https://example.org/' || toString(number % 100), "
+        "if(number % 7 = 0, 't', 'f') FROM numbers(200000)",
+        # ~20 MB that doesn't compress: takes ~20 s to back up with work/slow.json
+        "CREATE TABLE ooni.obs_web (measurement_uid String, blob String) ENGINE = MergeTree ORDER BY measurement_uid",
+        "INSERT INTO ooni.obs_web SELECT toString(number), randomString(4096) FROM numbers(5000)",
+        "CREATE TABLE ooni.test_groups (name String, pattern String) ENGINE = EmbeddedRocksDB PRIMARY KEY name",
+        "INSERT INTO ooni.test_groups VALUES ('a', '1.2.3.4'), ('b', '5.6.7.8')",
+        "CREATE TABLE ooni.msmt_feedback (measurement_uid String, comment String) ENGINE = MergeTree ORDER BY measurement_uid",
+        "INSERT INTO ooni.msmt_feedback VALUES ('m1', 'blocked?')",
+        "CREATE TABLE ooni.labeling_frames (measurement_uid String) ENGINE = MergeTree ORDER BY measurement_uid",
+        "INSERT INTO ooni.labeling_frames VALUES ('x')",
+        "CREATE TABLE ooni.asnmeta_tmp (asn UInt32) ENGINE = MergeTree ORDER BY asn",  # excluded: *_tmp
+    ]:
+        ch(sql)
+    ch("CREATE DATABASE ooni", port=19912)
+
+
+def test_first_run():
+    print("== 1. first run: a full backup of every table")
+    r = backup("backup", "run")
+    check(r.returncode == 0, "the run succeeds", r.stderr)
+    for bucket, table in [(MEASUREMENTS, "fastpath"), (MEASUREMENTS, "obs_web"),
+                          (MEASUREMENTS, "test_groups"), (PRIVATE, "msmt_feedback")]:
+        c = chains(bucket, table)
+        check(len(c) == 1 and all(all(b.values()) for b in c.values()), f"{table}: one complete chain in {bucket}")
+    check(any(k.endswith("/data.native.zst") for k in keys(MEASUREMENTS, "prod/ooni/test_groups/")),
+          "test_groups (EmbeddedRocksDB): its rows are exported too")
+    check(not keys(MEASUREMENTS, "prod/ooni/msmt_feedback/"), "msmt_feedback isn't in the measurements bucket")
+    check(not keys(MEASUREMENTS, "prod/ooni/asnmeta_tmp/"), "asnmeta_tmp (excluded) isn't backed up")
+    check(not keys(MEASUREMENTS, "prod/ooni/labeling_frames/"), "labeling_frames (data1's) isn't backed up")
+
+
+def test_completed_alert(emails_before):
+    print("== 2. ClickHouseBackupCompleted")
+    for job, tables in (("measurements", 3), ("private", 1)):
+        title = f"ClickHouse backup {job} on {DATA3} completed"
+        msgs = wait_for(f"a Slack message: {title}", lambda: [m for m in slack() if m[0].startswith(title)])
+        check(len(msgs) == 1, f"one Slack message for {job}: {msgs[0][0]!r}")
+        check(f"Tables backed up: {tables}, in full: {tables}" in msgs[0][1], f"it says: {msgs[0][1]!r}")
+    check(not any(t.startswith("[FIRING]") for t, _ in slack()), "no [FIRING] in the titles")
+    check(email_attempts() == emails_before, "no email sent")
+
+
+
+def digest(table):
+    """row count and a checksum of every row"""
+    return ch(f"SELECT count(), sum(cityHash64(*)) FROM {table}")
+
+
+def backup_url(bucket, table, chain, name):
+    return f"http://minio:9000/{bucket}/prod/ooni/{table}/{chain}/{name}"
+
+
+def test_incremental_and_restore():
+    print("== 3. incrementals, and restoring each backup of a chain")
+    # for step 4: a table created on data1 alone, which no job backs up
+    ch("CREATE TABLE ooni.made_on_data1 (x UInt8) ENGINE = MergeTree ORDER BY x", port=19912)
+    # fastpath as at the first run's full, then after each of two more runs
+    states = [digest("ooni.fastpath")]
+    for first in (200000, 201000):
+        ch(f"INSERT INTO ooni.fastpath SELECT toString(number), 'https://example.org/new', 'f' FROM numbers({first}, 1000)")
+        r = backup("backup", "run")
+        check(r.returncode == 0, "the run succeeds", r.stderr)
+        states.append(digest("ooni.fastpath"))
+    (chain, backups), = chains(MEASUREMENTS, "fastpath").items()
+    names = sorted(backups)
+    check(len(names) == 3 and all(backups.values()), f"fastpath: a full and two incrementals in chain {chain}")
+    ch("CREATE DATABASE restored")
+    for i, (name, state) in enumerate(zip(names, states)):
+        what = "the full" if i == 0 else f"incremental {i}"
+        ch(f"RESTORE TABLE ooni.fastpath AS restored.fastpath_{i} FROM S3('{backup_url(MEASUREMENTS, 'fastpath', chain, name)}')")
+        check(digest(f"restored.fastpath_{i}") == state, f"{what} restores fastpath as it was then ({state.split()[0]} rows)")
+
+    # EmbeddedRocksDB: BACKUP keeps only the schema, the rows are in data.native.zst (the runbook's steps 2 and 3)
+    c = chains(MEASUREMENTS, "test_groups")
+    chain = max(ch_ for ch_, b in c.items() if any(b.values()))
+    url = backup_url(MEASUREMENTS, "test_groups", chain, max(n for n, ok in c[chain].items() if ok))
+    ch(f"RESTORE TABLE ooni.test_groups AS restored.test_groups FROM S3('{url}')")
+    ch(f"INSERT INTO restored.test_groups SELECT * FROM s3('{url}/data.native.zst', 'Native')")
+    check(digest("restored.test_groups") == digest("ooni.test_groups"),
+          "test_groups (EmbeddedRocksDB) restores with its exported rows")
+
+
+def test_not_backed_up_alert():
+    print("== 4. a table no backup covers: ClickHouseTableNotBackedUp")
+    title = "[FIRING] ClickHouse table made_on_data1 is not backed up (on data1)"
+    msgs = wait_for(f"a Slack message: {title}", lambda: [m for m in slack() if m[0] == title])
+    check("Add it to a job" in msgs[0][1], f"it says what to do: {msgs[0][1][:90]!r}...")
+    others = [t for t, _ in slack() if "is not backed up" in t and t != title]
+    check(not others, f"no other table is reported (labeling_frames is data1's, asnmeta_tmp excluded): {others}")
+
+
+def test_failure_alert(emails_before):
+    print("== 5. a run that fails: ClickHouseBackupFailed")
+    r = backup("broken", "run")
+    check(r.returncode == 1, "the run exits 1", r.stderr)
+    title = f"[FIRING] ClickHouse backup of labeling_frames (broken) failed on {DATA3}"
+    msgs = wait_for(f"a Slack message: {title}", lambda: [m for m in slack() if m[0] == title])
+    check("Nothing was deleted" in msgs[0][1], f"it says what was done: {msgs[0][1]!r}")
+    wait_for("an email attempt for it", lambda: email_attempts() > emails_before)
+    check(not any(t.startswith("ClickHouse backup broken") for t, _ in slack()), "no 'completed' message for it")
+
+
+def test_interrupted_full():
+    print("== 6. a new month's full, interrupted, then completed")
+    (chain, backups), = chains(MEASUREMENTS, "obs_web").items()
+    full = f"prod/ooni/obs_web/{chain}/{min(backups)}/"
+    old, previous = months_ago(2), months_ago(1)
+    # obs_web's chains as at the start of a month: the full of two months ago and of last month
+    for c in (old, previous):
+        copy_prefix(MEASUREMENTS, full, f"prod/ooni/obs_web/{c}/{c}/")
+    delete_prefix(MEASUREMENTS, f"prod/ooni/obs_web/{chain}/")
+
+    run = subprocess.Popen(COMPOSE + ["exec", "-T", "runner", "python3", "/role/ooni_clickhouse_backup.py",
+                                      "--config", "/work/slow.json", "run", "--table", "obs_web"],
+                           stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+    wait_for("the new full's first files", lambda: set(chains(MEASUREMENTS, "obs_web")) - {old, previous}, 60)
+    # killed, not stopped: ClickHouse waits for running backups when it stops
+    # (shutdown_wait_backups_and_restores), so a clean stop would finish this one
+    subprocess.run(COMPOSE + ["kill", "clickhouse"], check=True, capture_output=True)
+    subprocess.run(COMPOSE + ["start", "clickhouse"], check=True, capture_output=True)
+    out, _ = run.communicate(timeout=900)
+    check(run.returncode == 1, "the interrupted run exits 1", out)
+    c = chains(MEASUREMENTS, "obs_web")
+    new = sorted(set(c) - {old, previous})
+    check(new and not any(c[new[0]].values()), f"the new chain {new[0]} has no .backup")
+    check(all(c[x][x] for x in (old, previous)), f"both older chains, {old} and {previous}, are still there")
+
+    wait_for("ClickHouse after it was killed", lambda: ch("SELECT 1") == "1")
+    r = backup("backup", "run", "--table", "obs_web")
+    check(r.returncode == 0, "the next run succeeds", r.stderr)
+    check("removed incomplete chain" in r.stderr, "it removed the interrupted full")
+    c = chains(MEASUREMENTS, "obs_web")
+    check(previous in c and old not in c and len(c) == 2 and all(all(b.values()) for b in c.values()),
+          f"it took the full and pruned {old}: chains {sorted(c)}")
+    new = max(c)
+    ch(f"RESTORE TABLE ooni.obs_web AS restored.obs_web FROM S3('{backup_url(MEASUREMENTS, 'obs_web', new, new)}')")
+    check(digest("restored.obs_web") == digest("ooni.obs_web"), f"the new full {new} restores obs_web")
+
+
+def main():
+    setup()
+    emails_before = email_attempts()
+    test_first_run()
+    test_completed_alert(emails_before)
+    test_incremental_and_restore()
+    test_not_backed_up_alert()
+    test_failure_alert(emails_before)
+    test_interrupted_full()
+    print("PASS")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/ansible/roles/clickhouse_backup/tests/test_job.py b/ansible/roles/clickhouse_backup/tests/test_job.py
new file mode 100644
index 00000000..51417fc1
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/test_job.py
@@ -0,0 +1,211 @@
+"""How a job runs and what it reports, with ClickHouse stubbed; test_e2e.py runs the real thing.
+    python3 -m unittest test_job"""
+import importlib.util
+import json
+import os
+import tempfile
+import types
+import unittest
+import urllib.error
+from unittest import mock
+
+spec = importlib.util.spec_from_file_location(
+    "backup", os.path.join(os.path.dirname(__file__), "..", "files", "ooni_clickhouse_backup.py")
+)
+backup = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(backup)
+backup.log.disabled = True  # the tests check results, not log lines
+
+
+class FlakyClickHouse:
+    """Fails the first `failures` calls with a connection error, then answers"""
+
+    def __init__(self, failures=0, status="BACKUP_CREATED"):
+        self.failures, self.status, self.calls = failures, status, 0
+
+    def query(self, sql, settings=None, timeout=600):
+        self.calls += 1
+        if self.calls <= self.failures:
+            raise urllib.error.URLError("Connection refused")
+        return ""
+
+    def rows(self, sql, **kw):
+        self.query(sql)
+        if "system.backups WHERE id" in sql:
+            return [{"status": self.status, "error": "", "num_files": 1, "total_size": 1, "uncompressed_size": 1}]
+        return []
+
+
+def job(name="measurements", include=(), exclude=(), metrics=None, ch=None):
+    j = backup.BackupJob.__new__(backup.BackupJob)
+    j.name, j.include, j.exclude = name, list(include), list(exclude)
+    j.metrics = metrics or backup.Metrics()
+    j.ch = ch or FlakyClickHouse()
+    j.poll_s, j.clickhouse_wait_s = 0, 0
+    return j
+
+
+class TestRunningSince(unittest.TestCase):
+    def setUp(self):
+        d = tempfile.mkdtemp()
+        self.metrics = backup.Metrics(os.path.join(d, "out.prom"), os.path.join(d, "state.json"))
+
+    def test_running_since_is_set_while_a_job_runs(self):
+        self.metrics.job_running("measurements", True)
+        self.assertGreater(self.metrics.state["running_since"]["measurements"], 0)
+        with open(self.metrics.metrics_file) as f:
+            self.assertIn('ooni_clickhouse_backup_running_since_timestamp_seconds{backup="measurements"}', f.read())
+        self.metrics.job_running("measurements", False)
+        self.assertEqual(self.metrics.state["running_since"]["measurements"], 0)
+
+    def test_old_state_files_still_load(self):
+        with open(self.metrics.state_file, "w") as f:
+            json.dump({"tables": {}, "running": {"measurements": 0}}, f)
+        m = backup.Metrics(self.metrics.metrics_file, self.metrics.state_file)
+        m.job_running("measurements", True)
+        self.assertIn("measurements", m.state["running_since"])
+
+
+class TestRunTable(unittest.TestCase):
+    def test_only_the_jobs_that_back_up_the_tables(self):
+        jobs = [job("measurements", include=["*"], exclude=["msmt_feedback"]), job("private", include=["msmt_feedback"])]
+        selected, orphans = backup.jobs_for_tables(jobs, ["msmt_feedback"])
+        self.assertEqual([j.name for j in selected], ["private"])
+        self.assertEqual(orphans, [])
+
+    def test_tables_no_job_backs_up(self):
+        jobs = [job("private", include=["msmt_feedback"])]
+        self.assertEqual(backup.jobs_for_tables(jobs, ["fastpath"]), ([], ["fastpath"]))
+
+
+class TestIncludeTables(unittest.TestCase):
+    def test_star_is_every_table(self):
+        self.assertTrue(job(include=["*"]).includes("fastpath"))
+
+    def test_an_empty_list_is_no_table(self):
+        # once data1's list of its own tables is emptied, data1 mustn't back up everything
+        self.assertFalse(job(include=[]).includes("msmt_feedback"))
+
+    def test_a_config_without_jobs_backs_up_every_table(self):
+        [conf] = backup.job_confs({"exclude_tables": ["*_tmp"]}, [])
+        self.assertEqual(conf["include_tables"], ["*"])
+
+
+class TestForget(unittest.TestCase):
+    def setUp(self):
+        d = tempfile.mkdtemp()
+        self.metrics = backup.Metrics(os.path.join(d, "out.prom"), os.path.join(d, "state.json"))
+        for job_, table in [("data1", "labeling_frames"), ("data1", "changepoint_label"), ("private", "msmt_feedback")]:
+            self.metrics.table_result(job_, table, True, {"total_size": 1, "num_files": 1})
+        self.metrics.job_running("old_job", False)
+
+    def tables(self):
+        return sorted(self.metrics.state["tables"])
+
+    def test_a_table_the_job_no_longer_selects(self):
+        # labeling_frames was replicated, and moved from data1's job to data3's
+        self.metrics.forget_tables("data1", {"changepoint_label"})
+        self.assertEqual(self.tables(), ["data1/changepoint_label", "private/msmt_feedback"])
+
+    def test_a_job_no_longer_configured(self):
+        self.metrics.forget_jobs({"data1"})
+        self.assertEqual(self.tables(), ["data1/changepoint_label", "data1/labeling_frames"])
+        self.assertNotIn("old_job", self.metrics.state["running"])
+
+    def test_a_run_with_table_forgets_nothing(self):
+        j = job("data1", include=["*"], metrics=self.metrics)
+        j.running_backups = lambda: []
+        j.tables = lambda only: ([("changepoint_label", "MergeTree")], [])
+        j.backup_table = lambda *a: False
+        j.run(["changepoint_label"], False, "")
+        self.assertIn("data1/labeling_frames", self.tables())
+        j.run([], False, "")
+        self.assertNotIn("data1/labeling_frames", self.tables())
+
+
+class TestStoreAndState(unittest.TestCase):
+    def test_only_this_buckets_running_backups(self):
+        j = job()
+        j.store = types.SimpleNamespace(ch_base="https://r2.example/ooni-clickhouse-backup", prefix="prod")
+        j.ch = types.SimpleNamespace(rows=lambda sql: [
+            {"id": "1", "name": "S3('https://r2.example/ooni-clickhouse-backup-private/prod/ooni/msmt_feedback/a/b')"},
+            {"id": "2", "name": "S3('https://r2.example/ooni-clickhouse-backup/prod/ooni/fastpath/a/b')"},
+        ])
+        self.assertEqual([r["id"] for r in j.running_backups()], ["2"])
+
+    def test_objects_not_deleted_are_an_error(self):
+        store = backup.Store.__new__(backup.Store)
+        store.bucket, store.prefix = "b", "prod"
+        store.s3 = types.SimpleNamespace(
+            get_paginator=lambda name: types.SimpleNamespace(paginate=lambda **kw: [{"Contents": [{"Key": "prod/ooni/t/c/x"}]}]),
+            delete_objects=lambda **kw: {"Errors": [{"Key": "prod/ooni/t/c/x", "Code": "AccessDenied", "Message": "no"}]},
+        )
+        with self.assertRaisesRegex(RuntimeError, "1 objects not deleted"):
+            store.delete_prefix("prod/ooni/t/c/")
+
+    def test_an_unreadable_state_file_starts_empty(self):
+        d = tempfile.mkdtemp()
+        state = os.path.join(d, "state.json")
+        with open(state, "w") as f:
+            f.write("")  # e.g. truncated by a power loss
+        m = backup.Metrics(os.path.join(d, "out.prom"), state)
+        m.job_running("measurements", True)
+        self.assertEqual(m.state["running"], {"measurements": 1})
+
+
+class TestNotBackedUp(unittest.TestCase):
+    def test_tables_no_job_covers(self):
+        jobs = [job("measurements", include=["*"], exclude=["msmt_feedback", "labeling_frames", "*_tmp"]),
+                job("private", include=["msmt_feedback"])]
+        tables = {
+            "fastpath": ["data1", "data2", "data3"],         # replicated: measurements
+            "msmt_feedback": ["data1", "data2", "data3"],    # private
+            "labeling_frames": ["data1", "data2", "data3"],  # data1 backs it up
+            "asnmeta_tmp": ["data1"],                         # excluded
+            "fingerprints_dns": ["data1", "data2"],           # excluded: rebuilt hourly
+            "made_on_data1": ["data1"],                       # nothing: data3 doesn't have it
+        }
+        missing = backup.not_backed_up(tables, "data3", jobs, ["*_tmp", "fingerprints_*"], ["labeling_frames"])
+        self.assertEqual(missing, {"made_on_data1": ["data1"]})
+
+    def test_the_metric_names_the_table_and_its_hosts(self):
+        d = tempfile.mkdtemp()
+        m = backup.Metrics(os.path.join(d, "out.prom"), os.path.join(d, "state.json"))
+        m.not_backed_up({"made_on_data1": ["data2", "data1"]})
+        with open(m.metrics_file) as f:
+            self.assertIn('ooni_clickhouse_backup_table_not_backed_up{table="made_on_data1",hosts="data1,data2"} 1', f.read())
+        m.not_backed_up({})
+        with open(m.metrics_file) as f:
+            self.assertNotIn("made_on_data1", f.read())
+
+
+class TestClickHouseErrors(unittest.TestCase):
+    def setUp(self):
+        d = tempfile.mkdtemp()
+        self.metrics = backup.Metrics(os.path.join(d, "out.prom"), os.path.join(d, "state.json"))
+
+    def test_waits_for_clickhouse_at_start(self):
+        j = job(metrics=self.metrics, ch=FlakyClickHouse(failures=2))
+        j.clickhouse_wait_s = 60
+        with mock.patch.object(backup.time, "sleep"):
+            j.wait_for_clickhouse()
+        self.assertEqual(j.ch.calls, 3)
+
+    def test_a_job_that_cant_start_marks_its_tables_failed(self):
+        self.metrics.table_result("measurements", "fastpath", True, {"total_size": 1, "num_files": 1})
+        self.metrics.table_result("private", "msmt_feedback", True, {"total_size": 1, "num_files": 1})
+        j = job(metrics=self.metrics, ch=FlakyClickHouse(failures=1000))
+        self.assertEqual(j.run([], False, ""), 1)
+        tables = self.metrics.state["tables"]
+        self.assertEqual(tables["measurements/fastpath"]["last_failed"], 1)
+        self.assertEqual(tables["private/msmt_feedback"]["last_failed"], 0, "another job's table")
+        self.assertNotIn("measurements", self.metrics.state["runs"], "no completed run to report")
+
+    def test_a_failed_poll_is_retried(self):
+        j = job(ch=FlakyClickHouse(failures=2))
+        self.assertEqual(j.wait("op", "obs_web")["status"], "BACKUP_CREATED")
+        self.assertEqual(j.ch.calls, 3)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/ansible/roles/clickhouse_backup/tests/test_schedule.py b/ansible/roles/clickhouse_backup/tests/test_schedule.py
new file mode 100644
index 00000000..b7ea3146
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/test_schedule.py
@@ -0,0 +1,120 @@
+"""When ooni_clickhouse_backup.py starts a full backup and what it prunes, over months of daily or weekly runs,
+some of them failing. ClickHouse and the bucket are stubbed; test_e2e.py runs the real thing.
+    python3 -m unittest test_schedule"""
+import importlib.util
+import json
+import os
+import unittest
+from datetime import datetime, timedelta, timezone
+
+spec = importlib.util.spec_from_file_location(
+    "backup", os.path.join(os.path.dirname(__file__), "..", "files", "ooni_clickhouse_backup.py")
+)
+backup = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(backup)
+
+START = datetime(2026, 10, 23, 2, 30, tzinfo=timezone.utc)  # a Friday
+
+
+class FakeStore:
+    """The bucket: {chain: {"backups": {backup: complete}, "keep": [labels]}}"""
+
+    def __init__(self):
+        self.c = {}
+
+    def table_prefix(self, db, table):
+        return f"{db}/{table}/"
+
+    def ch_s3(self, key):
+        return f"'{key}'"
+
+    def chains(self, db, table):
+        return {k: {"backups": dict(e["backups"]), "keep": list(e["keep"])} for k, e in self.c.items()}
+
+    def delete_prefix(self, prefix):
+        parts = prefix.rstrip("/").split("/")
+        if parts[-2] in self.c:  # one backup of a chain
+            self.c[parts[-2]]["backups"].pop(parts[-1], None)
+        else:
+            self.c.pop(parts[-1], None)
+        return 1
+
+    def put_empty(self, key):
+        pass
+
+
+class FakeClickHouse:
+    """Takes a backup, or fails one part-way: files uploaded, no .backup"""
+
+    def __init__(self, store):
+        self.store, self.fail = store, False
+
+    def query(self, sql, settings=None, timeout=600):
+        chain, name = sql.split("TO S3('")[1].split("'")[0].split("/")[-2:]
+        self.store.c.setdefault(chain, {"backups": {}, "keep": []})["backups"][name] = not self.fail
+        if self.fail:
+            raise RuntimeError("BACKUP_FAILED")
+        return json.dumps({"id": "op"})
+
+
+def runs(every_days, until, failing=(), jitter=True):
+    """Back up a table at 02:30 UTC daily or on Sundays from START. Yields (day, full, complete chains)."""
+    job = backup.BackupJob.__new__(backup.BackupJob)
+    job.db, job.name, job.keep_chains, job.settings = "ooni", "t", 2, {}
+    job.store = FakeStore()
+    job.ch = FakeClickHouse(job.store)
+    job.metrics = backup.Metrics()
+    job.wait = lambda op, what: {"num_files": 1, "uncompressed_size": 1}
+    t, n = START, 0
+    while t < until:
+        # a table's backup starts a few minutes earlier or later each time
+        now = t + timedelta(minutes=9 if n % 2 else -7)
+        backup.utcnow = lambda now=now: now
+        job.ch.fail = now.date().isoformat() in failing
+        try:
+            full = job.backup_table("obs_web", "ReplicatedMergeTree", False, "")
+        except RuntimeError:
+            full = None
+        complete = sorted(c for c, e in job.store.c.items() if any(e["backups"].values()))
+        yield now.date().isoformat(), full, complete
+        n += 1
+        if every_days == 7:
+            t += timedelta(days=(6 - t.weekday()) % 7 or 7)  # the next Sunday
+        else:
+            t += timedelta(days=1)
+
+
+class TestSchedule(unittest.TestCase):
+    def fulls(self, every_days):
+        return [day for day, full, _ in runs(every_days, datetime(2027, 4, 1, tzinfo=timezone.utc)) if full]
+
+    def test_weekly_fulls_on_the_first_sunday_of_each_month(self):
+        self.assertEqual(
+            self.fulls(7), ["2026-10-23", "2026-11-01", "2026-12-06", "2027-01-03", "2027-02-07", "2027-03-07"]
+        )
+
+    def test_daily_fulls_on_the_first_of_each_month(self):
+        self.assertEqual(
+            self.fulls(1), ["2026-10-23", "2026-11-01", "2026-12-01", "2027-01-01", "2027-02-01", "2027-03-01"]
+        )
+
+    def test_two_complete_chains_kept(self):
+        *_, (day, _, complete) = runs(7, datetime(2027, 4, 1, tzinfo=timezone.utc))
+        self.assertEqual([c[:8] for c in complete], ["20270207", "20270307"])
+
+    def test_failed_fulls_prune_nothing_and_are_retried(self):
+        failing = {"2026-12-06", "2026-12-13", "2027-02-07", "2027-02-14", "2027-02-21", "2027-02-28", "2027-03-07"}
+        before = []
+        for day, full, complete in runs(7, datetime(2027, 5, 1, tzinfo=timezone.utc), failing):
+            month = day[:7].replace("-", "")
+            if not any(c.startswith(month) for c in complete):
+                # no complete full this month yet: every complete chain from before is still there
+                self.assertTrue(set(before) <= set(complete), f"{day}: deleted {set(before) - set(complete)}")
+            if day in failing:
+                self.assertIsNone(full, day)
+            before = complete
+        self.assertEqual([c[:8] for c in complete], ["20270314", "20270404"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/ansible/roles/clickhouse_backup/tests/vars.yml b/ansible/roles/clickhouse_backup/tests/vars.yml
new file mode 100644
index 00000000..33957689
--- /dev/null
+++ b/ansible/roles/clickhouse_backup/tests/vars.yml
@@ -0,0 +1,38 @@
+# Role variables for the test environment in compose.yml, with data3's two
+# jobs: the measurement tables and the private tables, each in its own bucket.
+clickhouse_admin_user: admin
+clickhouse_admin_password: admin
+clickhouse_backup_clickhouse_url: http://clickhouse:8123/
+clickhouse_backup_s3_endpoint: http://minio:9000
+clickhouse_backup_s3_access_key_id: testkey
+clickhouse_backup_s3_secret_access_key: testsecret123
+
+test_jobs:
+  - name: measurements
+    s3_bucket: ooni-clickhouse-backup
+    include_tables: ["*"]
+    exclude_tables: [msmt_feedback, labeling_frames]
+  - name: private
+    s3_bucket: ooni-clickhouse-backup-private
+    include_tables: [msmt_feedback]
+    exclude_tables: []
+
+# data1 backs up its own labeling_frames
+clickhouse_backup_backed_up_elsewhere: [labeling_frames]
+
+# One script config per entry, in work/<name>.json
+test_configs:
+  - name: backup  # as deployed on data3
+    bandwidth: 0
+    jobs: "{{ test_jobs }}"
+    check_cluster: test_cluster
+  - name: slow  # ~1 MB/s, so a full backup can be interrupted
+    bandwidth: 3000000
+    jobs: "{{ test_jobs }}"
+  - name: broken  # its bucket doesn't exist, so every backup fails
+    bandwidth: 0
+    jobs:
+      - name: broken
+        s3_bucket: no-such-bucket
+        include_tables: [labeling_frames]
+        exclude_tables: []
diff --git a/ansible/roles/prometheus/files/alert_rules.yml b/ansible/roles/prometheus/files/alert_rules.yml
index 1e704a93..a2f33105 100644
--- a/ansible/roles/prometheus/files/alert_rules.yml
+++ b/ansible/roles/prometheus/files/alert_rules.yml
@@ -99,9 +99,10 @@ groups:
     annotations:
       summary: 'Broken `textfile` at {{ $labels.instance }}'
 
-  # all the textfile are updated at least hourly
+  # all the textfile are updated at least hourly, except the ClickHouse
+  # backup results, which change after each backup (see ClickHouseBackupStale)
   - alert: TextfileJam
-    expr: (node_textfile_mtime - ON(instance,job) GROUP_LEFT() node_time) / -3600 > 2
+    expr: (node_textfile_mtime{file!~".*/ooni_clickhouse_backup\\.prom"} - ON(instance,job) GROUP_LEFT() node_time) / -3600 > 2
     annotations:
       summary: 'Stale `textfile` at {{ $labels.instance }}'
       description: '`{{ $labels.file }}` is {{ printf "%.0f" $value }}h old.'
@@ -144,4 +145,65 @@ groups:
     expr: rate(alertmanager_notifications_failed_total[1m]) > 0
     annotations:
       summary: 'Alertmanager {{ $labels.instance }} fails {{ $labels.integration }} notifications.'
+
+# Results written by ooni-clickhouse-backup (ansible role clickhouse_backup)
+# to node_exporter's textfile directory on data1 and data3
+- name: ClickHouse backups
+  rules:
+  - alert: ClickHouseBackupFailed
+    expr: ooni_clickhouse_backup_last_attempt_failed == 1
+    annotations:
+      summary: 'ClickHouse backup of {{ $labels.table }} ({{ $labels.backup }}) failed on {{ $labels.instance }}'
+      description: 'Nothing was deleted: the older backups of {{ $labels.table }} are kept. The next run retries it. See `journalctl --identifier ooni-clickhouse-backup` on {{ $labels.instance }}.'
+
+  # A table on the cluster that no backup job covers, and that isn't excluded:
+  # e.g. a new table created on one node only (data3 checks after each run)
+  - alert: ClickHouseTableNotBackedUp
+    expr: ooni_clickhouse_backup_table_not_backed_up == 1
+    annotations:
+      summary: 'ClickHouse table {{ $labels.table }} is not backed up (on {{ $labels.hosts }})'
+      description: 'No backup job covers it and it is not excluded. Add it to a job (clickhouse_backup_data1_tables if only data1 holds its data, clickhouse_backup_private_tables if private), or to clickhouse_backup_exclude_tables if it needs no backup. See the runbook, "ClickHouse backup".'
+
+  # A job's run finished with every table backed up. One Slack message per
+  # job run, no email and no "resolved" message: Alertmanager routes it to
+  # slack-notice. It clears an hour after the run.
+  - alert: ClickHouseBackupCompleted
+    expr: |
+      ooni_clickhouse_backup_last_run_duration_seconds
+        and on (instance, backup) (time() - ooni_clickhouse_backup_last_run_finished_timestamp_seconds < 3600)
+        and on (instance, backup) ooni_clickhouse_backup_last_run_failed_tables == 0
+    annotations:
+      summary: 'ClickHouse backup {{ $labels.backup }} on {{ $labels.instance }} completed in {{ $value | humanizeDuration }}'
+      description: 'Tables backed up: {{ with printf `ooni_clickhouse_backup_last_run_tables{instance="%s",backup="%s"}` $labels.instance $labels.backup | query }}{{ . | first | value }}{{ end }}, in full: {{ with printf `ooni_clickhouse_backup_last_run_full_backups{instance="%s",backup="%s"}` $labels.instance $labels.backup | query }}{{ . | first | value }}{{ end }}. Nothing to do.'
+
+  # Backups run weekly, so a table's last one is normally up to a week old. Not
+  # while any job on the host is in a run that started in the last 5 days: on
+  # the first Sunday of a month the full backups take more than a day, and the
+  # jobs run one after the other. A run older than that is stuck (e.g. the
+  # script was killed and left running at 1), and no longer holds the alert back.
+  - alert: ClickHouseBackupStale
+    expr: |
+      time() - ooni_clickhouse_backup_last_success_timestamp_seconds > 8 * 86400
+        unless on (instance) (time() - (ooni_clickhouse_backup_running_since_timestamp_seconds != 0) < 5 * 86400)
+    annotations:
+      summary: 'No ClickHouse backup of {{ $labels.table }} ({{ $labels.backup }}) for {{ $value | humanizeDuration }}'
+      description: 'Check `systemctl status ooni-clickhouse-backup.timer` and the journal on {{ $labels.instance }}.'
+
+  # The job hasn't written any result since its host was set up, or the file is
+  # gone. Backups run weekly, so a new host has none until its first Sunday.
+  - alert: ClickHouseBackupMissing
+    expr: absent(ooni_clickhouse_backup_running{instance="data3.htz-fsn.prod.ooni.nu"}) or absent(ooni_clickhouse_backup_running{instance="data1.htz-fsn.prod.ooni.nu"})
+    for: 8d
+    annotations:
+      summary: 'No ClickHouse backup results from {{ $labels.instance }} for over a week'
+
+  # data3 takes the backups. Above half its CPUs (load per CPU > 0.5), the
+  # backup or merges are crowding the node; lower clickhouse_backup_max_bandwidth
+  # or move the backup to another replica.
+  - alert: ClickHouseBackupHostLoadHigh
+    expr: node_load5{instance="data3.htz-fsn.prod.ooni.nu"} / on (instance) count by (instance) (node_cpu_seconds_total{mode="idle", instance="data3.htz-fsn.prod.ooni.nu"}) > 0.5
+    for: 15m
+    annotations:
+      summary: 'data3 load is {{ $value | humanizePercentage }} of its CPUs'
+      description: 'If a backup is running (ooni_clickhouse_backup_running on data3), lower clickhouse_backup_max_bandwidth; otherwise check merges and queries.'
 ...
diff --git a/ansible/roles/prometheus_alertmanager/files/templates/ooni.tmpl b/ansible/roles/prometheus_alertmanager/files/templates/ooni.tmpl
index c8ad084a..1268b5f2 100644
--- a/ansible/roles/prometheus_alertmanager/files/templates/ooni.tmpl
+++ b/ansible/roles/prometheus_alertmanager/files/templates/ooni.tmpl
@@ -4,3 +4,7 @@
 {{ define "slack.ooni.title" }}
 {{ range $i, $el := .Alerts }}{{ if gt $i 0 }}
 {{ end }}{{ if eq $el.Status "resolved" }}[RESOLVED] {{ else }}[FIRING] {{ end }}{{ $el.Annotations.summary }}{{ end }}{{ end }}
+
+{{ define "slack.ooni.notice.title" }}
+{{ range $i, $el := .Alerts }}{{ if gt $i 0 }}
+{{ end }}{{ $el.Annotations.summary }}{{ end }}{{ end }}
diff --git a/ansible/roles/prometheus_alertmanager/templates/alertmanager.yml b/ansible/roles/prometheus_alertmanager/templates/alertmanager.yml
index eb36cf18..6d0136f4 100755
--- a/ansible/roles/prometheus_alertmanager/templates/alertmanager.yml
+++ b/ansible/roles/prometheus_alertmanager/templates/alertmanager.yml
@@ -42,6 +42,11 @@ route:
   # All the above attributes are inherited by all child routes and can
   # overwritten on each.
   routes:
+  # Notices of things that went right: Slack only, posted once
+  - receiver: slack-notice
+    match: {alertname: ClickHouseBackupCompleted}
+    group_by: ['alertname', 'instance', 'backup']
+    repeat_interval: 24h
   - receiver: team-email
     repeat_interval: 24h
     match: {severity: info}
@@ -69,6 +74,12 @@ receivers:
     title: '{% raw %}{{ template "slack.ooni.title" . }}{% endraw %}'
     channel: '#ooni-bots'
 
+- name: 'slack-notice' # no email, no "resolved" message
+  slack_configs:
+  - send_resolved: false
+    text: '{% raw %}{{ template "slack.ooni.text" . }}{% endraw %}'
+    title: '{% raw %}{{ template "slack.ooni.notice.title" . }}{% endraw %}'
+    channel: '#ooni-bots'
 - name: 'team-email' # no slack
   email_configs:
 {% for u in ['arturo', 'mehul'] %}
diff --git a/ansible/roles/prometheus_node_exporter/defaults/main.yml b/ansible/roles/prometheus_node_exporter/defaults/main.yml
index 0cdc0011..bb75dd55 100644
--- a/ansible/roles/prometheus_node_exporter/defaults/main.yml
+++ b/ansible/roles/prometheus_node_exporter/defaults/main.yml
@@ -14,6 +14,9 @@ node_exporter_host: 'localhost'
 node_exporter_port: 8100
 node_exporter_htpasswd: '/etc/ooni/prometheus_passwd'
 node_exporter_options: ''
+# Directory for the textfile collector (*.prom files written by other jobs,
+# e.g. clickhouse_backup). Empty: no textfile collector directory.
+node_exporter_textfile_dir: ''
 
 node_exporter_state: started
 node_exporter_enabled: true
diff --git a/ansible/roles/prometheus_node_exporter/tasks/install.yml b/ansible/roles/prometheus_node_exporter/tasks/install.yml
index 1ff74425..6e042260 100644
--- a/ansible/roles/prometheus_node_exporter/tasks/install.yml
+++ b/ansible/roles/prometheus_node_exporter/tasks/install.yml
@@ -68,6 +68,18 @@
     - node_exporter
     - secrets
 
+- name: Create the textfile collector directory.
+  file:
+    path: "{{ node_exporter_textfile_dir }}"
+    state: directory
+    owner: root
+    group: root
+    mode: 0755
+  when: node_exporter_textfile_dir | length > 0
+  tags:
+    - prometheus_node_exporter
+    - config
+
 - name: Copy the node_exporter systemd unit file.
   template:
     src: node_exporter.service.j2
diff --git a/ansible/roles/prometheus_node_exporter/templates/node_exporter.service.j2 b/ansible/roles/prometheus_node_exporter/templates/node_exporter.service.j2
index 42cb98cc..31a822b0 100644
--- a/ansible/roles/prometheus_node_exporter/templates/node_exporter.service.j2
+++ b/ansible/roles/prometheus_node_exporter/templates/node_exporter.service.j2
@@ -4,7 +4,7 @@ Description=NodeExporter
 [Service]
 TimeoutStartSec=0
 User=node_exporter
-ExecStart={{ node_exporter_bin_path }} --web.listen-address={{ node_exporter_host }}:{{ node_exporter_port }} {{ node_exporter_options }}
+ExecStart={{ node_exporter_bin_path }} --web.listen-address={{ node_exporter_host }}:{{ node_exporter_port }} {{ node_exporter_options }}{% if node_exporter_textfile_dir %} --collector.textfile.directory={{ node_exporter_textfile_dir }}{% endif %}
 Restart={{ node_exporter_restart }}
 
 [Install]
diff --git a/docs/Runbooks.md b/docs/Runbooks.md
index c1b14ffa..aa0fa228 100644
--- a/docs/Runbooks.md
+++ b/docs/Runbooks.md
@@ -176,6 +176,198 @@ Future implementation is tracked in:
 [Implement Grafana dashboard and alarms backup](#implement-grafana-dashboard-and-alarms-backup)&thinsp;🐞
 
 
+## ClickHouse backup runbook
+The `ooni` database on the ClickHouse cluster is backed up to Cloudflare R2
+by the `clickhouse_backup` Ansible role, using ClickHouse's native `BACKUP`.
+Each host runs the jobs set in its host_vars, one after the other:
+
+| Host | Job | Bucket | Tables |
+|---|---|---|---|
+| data3 | `measurements` | `ooni-clickhouse-backup` | every replicated table, minus the private ones and data1's |
+| data3 | `private` | `ooni-clickhouse-backup-private` | `clickhouse_backup_private_tables`: `msmt_feedback`, `faulty_measurements`, `url_priorities` |
+| data1 | `data1` | `ooni-clickhouse-backup` | `clickhouse_backup_data1_tables`: plain MergeTree tables whose data only data1 holds |
+
+data3 takes the replicated tables because it isn't behind clickhouseproxy
+and doesn't run Airflow; replicated tables hold the same data on every node.
+Private tables have their own bucket because R2 tokens grant access per
+bucket, and researchers may be given read access to the measurement tables.
+When a data1-only table becomes replicated, remove it from
+`clickhouse_backup_data1_tables` and data3 picks it up.
+
+- `ooni-clickhouse-backup.timer` runs `/usr/local/bin/ooni-clickhouse-backup run`
+  weekly, on Sundays at 02:30 UTC (`clickhouse_backup_on_calendar`).
+- Each table has its own chain: a full backup, then weekly incrementals, each
+  on top of the one before. A table starts a new chain with its first backup
+  of each calendar month (UTC): the first Sunday of the month.
+  Once a new chain is complete, older chains are deleted unless pinned.
+- EmbeddedRocksDB and Join tables (`test_groups`): `BACKUP` saves only their
+  schema, so their rows are also exported to `data.native.zst` inside each
+  backup.
+- Not backed up: `fingerprints_dns` and `fingerprints_http`, which are only on
+  data1 and data2 and which the hourly updater rebuilds, and `session_expunge`,
+  which is empty and no longer used. They're in
+  `clickhouse_backup_exclude_tables`, with the other tables that are never
+  backed up.
+- After each run, data3 lists the tables on every node of the cluster
+  (`clickhouse_backup_check_cluster`) and reports any that no job backs up and
+  that aren't excluded, e.g. a new table created on data1 only.
+- R2 rejects multipart uploads whose parts (all but the last) differ in
+  size, so `clickhouse_backup_s3_settings` fixes ClickHouse's part size at
+  128 MiB (`s3_upload_part_size_multiply_factor: 1`). R2 allows 10,000 parts,
+  so one backup file can reach 1.25 TiB.
+- Two complete chains are kept per table (`clickhouse_backup_keep_chains`),
+  so a month or two of restore points.
+- Bandwidth is capped with `clickhouse_backup_max_bandwidth`, at 50 MB/s. A
+  new full backup of everything (~5.6 TB) takes about 31 hours at that rate.
+  Against MinIO in testing, 24.8 uploaded at only about a third of the cap,
+  which would make it about 4 days: check the rate of the first backups
+  (`ooni_clickhouse_backup_last_size_bytes / ooni_clickhouse_backup_last_duration_seconds`)
+  and raise the cap if they're that slow.
+
+Layout in the bucket:
+
+```
+prod/ooni/<table>/<chain>/<backup>/          native backup (first one is full)
+prod/ooni/<table>/<chain>/KEEP-<label>       pinned chain, never pruned
+```
+
+Never put S3 keys in `BACKUP`/`RESTORE`/`s3()` queries: an incremental
+backup stores its `base_backup` argument, keys included, in plain text in
+its `.backup` file. ClickHouse gets the keys from
+`/etc/clickhouse-server/config.d/ooni_backup_s3.xml`, which it reloads
+without a restart.
+
+### Setting up
+1. In the Cloudflare dashboard, create the R2 buckets `ooni-clickhouse-backup`
+   and `ooni-clickhouse-backup-private` (location hint: Western Europe), and
+   an R2 API token with Object Read & Write on both.
+2. Store the S3 endpoint (`https://<account id>.r2.cloudflarestorage.com`) and
+   the token's keys in SSM as
+   `/oonidevops/secrets/clickhouse_backup_r2_endpoint`,
+   `/oonidevops/secrets/clickhouse_backup_r2_access_key_id` and
+   `/oonidevops/secrets/clickhouse_backup_r2_secret_access_key`.
+3. Deploy node_exporter first, for its textfile directory, then the backups:
+   ```bash
+   ./play -i inventory deploy-clickhouse.yml -l data1.htz-fsn.prod.ooni.nu,data3.htz-fsn.prod.ooni.nu -t node_exporter --diff
+   ./play -i inventory deploy-clickhouse.yml -l data1.htz-fsn.prod.ooni.nu,data3.htz-fsn.prod.ooni.nu -t clickhouse_backup --diff
+   ```
+   and the alert rules with `deploy-monitoring.yml`.
+4. The timer takes the first (full) backup at its next run, on Sunday at
+   02:30 UTC. A full backup of everything takes about 31 hours, or longer if
+   the upload rate is below the cap. To start it on deploy day instead, e.g. a
+   Friday so that it runs over the weekend:
+   ```bash
+   sudo systemctl start --no-block ooni-clickhouse-backup.service
+   sudo journalctl -f --identifier ooni-clickhouse-backup
+   ```
+
+### Monitoring
+After each table, the script writes its result to
+`/var/lib/node_exporter/textfile_collector/ooni_clickhouse_backup.prom`
+(metrics `ooni_clickhouse_backup_*`, labels `backup` = job and `table`).
+Alerts, in `alert_rules.yml`:
+
+- `ClickHouseBackupFailed`: a table's last attempt failed. Nothing is
+  deleted, and the next run retries it.
+- `ClickHouseTableNotBackedUp`: a table on the cluster is in no backup job and
+  not excluded. Add it to a job: replicated tables are backed up by data3's
+  `measurements` job unless they're private (`clickhouse_backup_private_tables`),
+  and a table whose data only data1 holds goes in
+  `clickhouse_backup_data1_tables`. Or, if it needs no backup, add it to
+  `clickhouse_backup_exclude_tables`.
+- `ClickHouseBackupCompleted`: a job's run backed up every table. One Slack
+  message per run in `#ooni-bots`, no email.
+- `ClickHouseBackupStale`: a table hasn't been backed up for 8 days (backups
+  are weekly), and no job on its host is in a run started in the last 5 days.
+- `ClickHouseBackupMissing`: data1 or data3 hasn't written any results for 8
+  days, e.g. since the role was deployed and before its first Sunday run.
+- `ClickHouseBackupHostLoadHigh`: data3's 5-minute load is over half its
+  CPUs for 15 minutes.
+
+```bash
+sudo systemctl status ooni-clickhouse-backup.service
+sudo journalctl --identifier ooni-clickhouse-backup --since yesterday
+sudo ooni-clickhouse-backup list
+```
+
+The unit fails if any table fails; the others are still backed up. If the
+script is stopped, a `BACKUP` already sent keeps running inside ClickHouse.
+The next run refuses to start until it finishes:
+
+```sql
+SELECT id, status, start_time, error FROM system.backups ORDER BY start_time DESC LIMIT 10
+```
+
+### Before upgrading ClickHouse
+A server can restore backups taken by the same or an older version, but not
+by a newer one. On 24.8, restoring a backup taken on 26.8 crashed the server
+(segfault in `restorePartFromBackup`). After an upgrade step that changes the
+on-disk format (25.8, 26.3), the only way back is restoring a backup taken
+before that step, so:
+
+1. Disable the timer with `clickhouse_backup_enabled: false` (or
+   `sudo systemctl stop ooni-clickhouse-backup.timer`) and check that
+   `system.backups` shows nothing running. ClickHouse waits for running
+   backups when it stops (`shutdown_wait_backups_and_restores`, and the
+   unit's `TimeoutStopSec=infinity`), so restarting a node during a full
+   backup hangs until the backup is done, a day or more.
+2. Take a pinned backup (incrementals, minutes):
+   ```bash
+   sudo ooni-clickhouse-backup run --keep pre-25.8.29.51
+   ```
+3. Upgrade all three nodes. Don't run backups while nodes are on different
+   versions.
+4. Once the whole upgrade is confirmed good, and with the timer still
+   disabled (once enabled, it may start a run straight away for a missed
+   Sunday), start new chains on the new version. That takes a day or more,
+   so run it as a service rather than in a shell:
+   ```bash
+   sudo systemd-run --unit=ooni-clickhouse-backup-full -p TimeoutStartSec=infinity \
+     -p SyslogIdentifier=ooni-clickhouse-backup /usr/local/bin/ooni-clickhouse-backup run --full
+   sudo journalctl -f --identifier ooni-clickhouse-backup
+   ```
+   When the journal shows `all tables backed up` for every job, unpin the
+   old chains and re-enable the timer (`clickhouse_backup_enabled: true`):
+   ```bash
+   sudo ooni-clickhouse-backup release pre-25.8.29.51
+   ```
+
+Data written after a pinned backup isn't in it. If you roll back, that
+window has to be filled again by reprocessing those days.
+
+### Restoring
+The target server needs `ooni_backup_s3.xml` in `config.d` (the role
+installs it). `ooni-clickhouse-backup list` prints the latest backup URL of
+each table. Restoring the latest incremental restores the whole chain.
+
+1. The database isn't part of a table backup:
+   ```sql
+   CREATE DATABASE IF NOT EXISTS ooni
+   ```
+2. Restore tables before the materialized views that read from them, on
+   one node, and a view's `TO` table before the view: while a view exists
+   without its table, every insert into the view's source fails (fastpath's,
+   for `counters_asn_test_list`). A replicated table that still exists makes `RESTORE` fail with
+   `REPLICA_ALREADY_EXISTS`: drop it (`DROP TABLE ... SYNC`) or restore on
+   another server.
+   ```sql
+   RESTORE TABLE ooni.obs_web FROM S3('https://<account id>.r2.cloudflarestorage.com/ooni-clickhouse-backup/prod/ooni/obs_web/<chain>/<backup>')
+   ```
+3. For EmbeddedRocksDB and Join tables (`test_groups`), load the rows too:
+   ```sql
+   INSERT INTO ooni.test_groups
+   SELECT * FROM s3('<backup URL>/data.native.zst', 'Native')
+   ```
+4. On the other replicas, create replicated tables with the same
+   `SHOW CREATE TABLE` statement; they fetch the data from the restored node.
+   Non-replicated tables whose data only data1 holds were backed up from
+   data1 (job `data1`); restore them there.
+5. Compare row counts and checksums with what you expect, e.g.
+   ```sql
+   SELECT count(), sum(cityHash64(toString(tuple(*)))) FROM ooni.obs_web
+   ```
+
+
 ## Grafana editing
 This runbook describes adding new dashboards, panels and alerts in
 [Grafana](#grafana)&thinsp;🔧
diff --git a/scripts/dump-tables-ch.sh b/scripts/dump-tables-ch.sh
deleted file mode 100755
index efd11d70..00000000
--- a/scripts/dump-tables-ch.sh
+++ /dev/null
@@ -1,46 +0,0 @@
-#!/bin/bash
-set -euxo pipefail
-# This is to be run manually on the clickhouse host to dump schemas and table dumps
-# You may want to make some tweaks to the dumping rules in order avoid dumping
-# too much data (eg. fastpath)
-# You should then scp the data over to the target host manually, by running:
-# $ scp * clickhouse-instance2:/var/lib/clickhouse/ooni-dumps/
-TABLES=(
-"fastpath"
-"jsonl"
-"url_priorities"
-"citizenlab"
-"citizenlab_flip"
-"test_groups"
-"accounts"
-"session_expunge"
-"msmt_feedback"
-"fingerprints_dns"
-"fingerprints_http"
-"asnmeta"
-"counters_test_list"
-"counters_asn_test_list"
-"incidents"
-"oonirun"
-)
-
-dump_dir="./dumps"
-current_date=$(date +%Y%m%d)
-
-# Directory to store the dumps
-mkdir -p "$dump_dir"
-
-# Iterate over each table
-for table in "${TABLES[@]}"; do
-    # Define file names for schema and data dump
-    schema_file="${dump_dir}/${current_date}-${table}_schema.sql"
-    data_file="${dump_dir}/${current_date}-${table}_dump.clickhouse"
-
-    # Dump the table schema
-    echo "[+] dumping schema $schema_file"
-    clickhouse-client --query="SHOW CREATE TABLE ${table} FORMAT TabSeparatedRaw" > "$schema_file"
-
-    # Dump the table data in ClickHouse native format
-    echo "[+] dumping table data $data_file"
-    clickhouse-client --query="SELECT * FROM ${table} INTO OUTFILE '${data_file}' FORMAT Native"
-done
diff --git a/scripts/restore-dumps.sh b/scripts/restore-dumps.sh
deleted file mode 100755
index df67126e..00000000
--- a/scripts/restore-dumps.sh
+++ /dev/null
@@ -1,35 +0,0 @@
-#!/bin/bash
-set -euxo pipefail
-# Restore schema and sampled dumps to running clickhouse instance
-# This script is to be run manually from with CWD set to contains the dumps and
-# schema files generated from `dump-tables-ch.sh`
-# See dump-tables-ch.sh for instruction on it's usage
-for schema_file in *schema.sql;do
-    cat $schema_file | clickhouse-client;
-done
-
-dump_ts="20240202"
-TABLES=(
-"jsonl"
-"url_priorities"
-"citizenlab"
-"citizenlab_flip"
-"test_groups"
-"accounts"
-"session_expunge"
-"msmt_feedback"
-"fingerprints_dns"
-"fingerprints_http"
-"asnmeta"
-"counters_test_list"
-"counters_asn_test_list"
-"incidents"
-"oonirun"
-)
-for table in "${TABLES[@]}"; do
-    echo "Restoring ${table}"
-    cat ${dump_ts}-${table}_dump.clickhouse | clickhouse-client --query="INSERT INTO ${table} FORMAT Native"
-done
-
-echo "Restoring fastpath"
-gzip -cd 20240109T1314-fastpath.clickhouse.gz | clickhouse-client --query="INSERT INTO fastpath FORMAT Native"