Cargo.lock

@@ -4277,6 +4277,23 @@ dependencies = [
  "thiserror 2.0.19",
 ]
 
+[[package]]
+name = "harmony-tenant-backup"
+version = "0.1.0"
+dependencies = [
+ "anyhow",
+ "chrono",
+ "clap",
+ "harmony",
+ "harmony_config",
+ "k8s-openapi",
+ "kube",
+ "schemars 0.8.22",
+ "serde",
+ "serde_json",
+ "tokio",
+]
+
 [[package]]
 name = "harmony_agent"
 version = "0.1.0"

examples/harmony_tenant_backup/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "harmony-tenant-backup"
+edition = "2024"
+version.workspace = true
+readme.workspace = true
+license.workspace = true
+publish = false
+
+[[bin]]
+name = "harmony-tenant-backup"
+path = "src/main.rs"
+
+[[bin]]
+name = "harmony-tenant-backup-setup"
+path = "src/setup.rs"
+
+[dependencies]
+harmony = { path = "../../harmony" }
+harmony_config = { path = "../../harmony_config" }
+tokio = { workspace = true, features = ["process", "time"] }
+kube = { workspace = true }
+k8s-openapi = { workspace = true }
+clap = { workspace = true }
+anyhow = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+schemars = "0.8"
+chrono = { workspace = true }

examples/harmony_tenant_backup/DESIGN.md

@@ -0,0 +1,301 @@
+# Tenant backup design
+
+Status: this is the current plan. The code in this folder is an earlier
+prototype (v0), kept for reference only. It does not implement this design;
+see "What changes from v0".
+
+A tenant on a shared cluster opts in to offsite backups of their own data. The
+backup job runs in the tenant's namespace with an identity that can write
+backups and can never delete them through the S3 API.
+
+## Backup instance
+
+The unit is a backup instance. It is created when a tenant enables backup and
+owns one config, one bucket, one restic password and one writer identity. A
+second instance, or a replacement, gets all four new. Nothing is shared
+between instances or between tenants.
+
+| Who | Does |
+|---|---|
+| Platform | `BackupInstanceScore`: bucket on the offsite RGW with versioning and Object Lock, a writer scoped to it, writer credentials into OpenBao at the tenant's path. Run by the platform team on request today, triggered by tenant self-service later. |
+| Tenant | `TenantBackupScore` in their deploy code: schedule, CNPG clusters, buckets, PVCs. Data they can regenerate stays out. Generates the restic password in their own password manager. |
+| Job | CronJob in the tenant namespace. Reads its instance config from OpenBao (see "Job identity"). |
+
+Tenants are admins of their namespace. They can read the job's secrets, edit
+or delete the CronJob. That only affects their own backups, which is
+acceptable because backup is their choice.
+
+## Storage and security
+
+- The bucket is owned by the platform's RGW account. The writer is an IAM
+  user in that account with an allow-only policy: `GetObject`, `PutObject`,
+  `ListBucket`, `GetBucketLocation`, and `DeleteObject` on `locks/*`.
+  Everything else is denied by default. The bucket owner is never the writer:
+  an owner keeps full control through ACLs whatever the policy says.
+- `PutObject` can overwrite, so the bucket has versioning and Object Lock in
+  compliance mode, 90 days by default. An overwrite or delete only adds a
+  version or a delete marker; the locked version stays. restic never rewrites
+  a key except `locks/*`. The 2025 median intrusion dwell time is 14 days, 26
+  when an outside party reports it (Mandiant M-Trends 2026); 90 days leaves
+  room for the slow tail. Cost: every locked version is stored for 90 days,
+  and a bucket can only be deleted 90 days after its last write. Compliance
+  retention can be extended, never shortened.
+- One lifecycle rule, on `locks/` only, expires noncurrent lock versions and
+  stale delete markers after the retention period. No lifecycle rule ever
+  touches other prefixes; their old versions are what a rollback reads.
+- A bucket quota caps growth. A full bucket fails the job loudly.
+- Object Lock binds the S3 API only. The Admin Ops bucket purge,
+  `radosgw-admin bucket rm --purge-objects`, `radosgw-admin object rm` and raw
+  `rados rm` all delete without checking retention (only `rgw_op.cc`'s S3
+  delete paths call `verify_object_lock`). On the offsite cluster, Ceph
+  keyrings that reach the RGW pools and RGW users with `buckets=write` caps,
+  including the one Rook creates, are held outside routine operator access.
+  Object Lock protects against tenants and stolen S3 keys, not against the
+  storage operator.
+- RGW accounts and IAM users need Ceph Squid (19.2) or later. On Reef the
+  writer is a plain RGW user with an inline policy on a bucket owned by
+  another user.
+- Writer credentials reach one bucket. A tenant who reads them learns nothing
+  about another tenant.
+- An offsite breach alone yields ciphertext without the key.
+- The job never runs `restic forget` or `prune`.
+- TLS is never skipped: `sslmode=verify-full` to Postgres, the RGW CA mounted
+  if it is private.
+
+`BackupInstanceScore` proves its own policy before returning: the writer can
+put an object, and cannot delete an object version or change the bucket
+policy.
+
+Not covered: whoever can read the instance's OpenBao path (OpenBao root, or
+cluster-admin acting as the job) gets the password and the writer
+credentials, and with them all history. They could already read the live
+data. restic is symmetric (no write-only key exists in restic, rustic, kopia
+or borg), so the job necessarily holds a key that reads all history.
+
+## Restic password
+
+The tenant generates the password in their own password manager, so the
+offline copy exists before the backup does. Nothing on the platform side
+generates it.
+
+The password is a `#[config(secret)]` field of the instance's config, stored
+by `harmony_config` like any other secret. On first apply, `TenantBackupScore`
+calls `get_or_prompt` and the tenant pastes it. In CI a missing password
+fails with "apply once interactively to set the restic password". On every
+apply the Score checks that the password opens the existing repo, so a wrong
+paste never creates a second repo or orphans history.
+
+| Event | Outcome |
+|---|---|
+| Password lost from the tenant's vault, OpenBao alive | Read it back with `harmony_config`. Same instance. |
+| Lost from both | Backups are unreadable, permanently. |
+| Password leaked | Read exposure only. New instance; the old bucket is deleted once the new one has a verified snapshot. |
+| Writer credentials reissued by the platform | Update the tenant's vault copy; the Score picks them up on next apply. |
+
+The tenant's vault entry holds endpoint, bucket, region, writer credentials
+and password: everything the restore below needs. The password is never
+rotated in place and never escrowed by the platform.
+
+## Job identity (open)
+
+The job reads its config from OpenBao at run time, using its own short-lived
+token rather than a stored secret. All auth goes through Zitadel today, and
+Zitadel cannot exchange a Kubernetes ServiceAccount token for its own: token
+exchange only accepts Zitadel-issued subject tokens, and the JWT IdP is a
+browser flow. So one of these:
+
+| | Holds | OpenBao trusts | Revocation | Cost |
+|---|---|---|---|---|
+| A. Second OpenBao JWT mount on the cluster's service-account issuer, static `jwt_validation_pubkeys` | Projected token, audience `openbao`, about 1 h | Zitadel, plus each cluster issuer limited to backup roles | Token TTL | Mount per cluster; one role per instance bound to `system:serviceaccount:<ns>:<sa>` and one path; update keys when the cluster rotates its signing key. `harmony_secret` reads the token file instead of signing with a Zitadel key. |
+| B. Zitadel machine key per instance (v0) | Long-lived key in a namespace Secret | Zitadel only | Manual | None, but the stored secret returns |
+| C. OpenBao `kubernetes` auth (TokenReview) | Projected token | Zitadel, plus each cluster API server | Immediate | OpenBao reaches each API server; new login method |
+
+Leaning A. Revisit if Zitadel ships workload identity federation. Rejected:
+copying the config into a namespace Secret at apply time, or External Secrets
+Operator; both leave the secret in etcd, and ESO adds a cluster-wide
+operator.
+
+## The job
+
+Two CronJobs, so a PVC that cannot mount never blocks the databases:
+
+- `pg+s3`: Postgres dumps and tenant buckets.
+- `pvc`: declared RWX volumes. Only created when the Score lists any.
+
+Settings: `concurrencyPolicy: Forbid`, `startingDeadlineSeconds: 3600`,
+`backoffLimit: 0` (a retry duplicates snapshots), `restartPolicy: Never`,
+`activeDeadlineSeconds`, `failedJobsHistoryLimit: 3`. Scratch is an `emptyDir`
+with `sizeLimit` and a matching `ephemeral-storage` request, staged one item
+at a time. Memory leaves room for the restic index, which grows forever
+without prune.
+
+The Score emits one NetworkPolicy: egress from the job to the CNPG pods on
+5432, the RGW on 443 and DNS; matching ingress on the CNPG pods. Shared OKD
+projects often default-deny.
+
+Every run ends with `restic check --read-data-subset=<weekday>/7`, so the
+whole repo is read once a week. It is the only thing that notices a pack
+overwritten with the writer's credentials. A killed run leaves a stale
+non-exclusive lock that does not block the next backup, and orphan packs that
+stay forever (bounded per run).
+
+Exit is zero or non-zero; the log line says fatal or partial and why.
+
+## Surfaces
+
+- CNPG: the Score names the credential Secret per cluster, `<cluster>-app` by
+  default (CNPG disables superuser access by default). `pg_dumpall
+  --globals-only --no-role-passwords`, then `pg_dump --format=custom
+  --compress=0 --lock-wait-timeout=60s` per database over `<cluster>-rw`, then
+  `pg_restore --list` to reject an unreadable archive. No compression because
+  restic compresses and compressed dumps defeat deduplication. Superuser is
+  required only for a cluster with several databases owned by different
+  roles; the Score checks this at apply time. The image's `pg_dump` is the
+  highest Postgres major on the platform. RPO is the last successful run;
+  there is no PITR.
+- Tenant S3 buckets: `rclone copy` to scratch with read-only source
+  credentials, then restic. Old versions on a versioned source bucket are not
+  copied.
+- PVCs: RWX only, decided at apply time. A declared RWO PVC is a Score error;
+  on Ceph most PVCs are RBD RWO, so this surface mostly covers CephFS. The
+  volume is `readOnly` at the `persistentVolumeClaim` source, not just the
+  mount, with `fsGroupChangePolicy: OnRootMismatch`, so kubelet never chowns
+  the tenant's live data. Files the job's UID cannot read make the run
+  partial, not fatal. CNPG data PVCs are never listed; the dump covers them.
+
+Each instance is one restic repo at the root of its bucket, with `--host`
+pinned to `<cluster-id>-<ns>`. Tags on every snapshot: `run:`, `ns:`,
+`surface:pg|s3|pvc`, and `cluster:`, `bucket:` or `pvc:`.
+
+## Deploy and failure visibility
+
+`TenantBackupScore` returns only after one Job created from the CronJob
+template exits zero and its snapshot is listed. NetworkPolicy, region,
+path-style and TLS mistakes fail the deploy with an actionable message instead
+of a 3 AM surprise.
+
+The Score emits a PrometheusRule for user-workload monitoring: alert when
+`time() - kube_cronjob_status_last_successful_time` exceeds twice the
+schedule interval. An optional dead-man's-switch URL is pinged on success
+with the snapshot IDs, which also catches a deleted CronJob or a burned
+cluster.
+
+## Schedule, growth and retention
+
+The schedule is a typed Score field: `Daily { at }` (default), `Hourly`, or
+`Cron(String)` for anything else. There is no pruning, so the bucket only
+grows; restic deduplication keeps each run to the chunks that changed, which
+is why dumps are uncompressed. Each run logs the bytes it added, so the
+tenant sees the real growth rate instead of an estimate.
+
+To drop old backups the tenant opens a ticket. The platform disables the
+writer key; the grace period, no shorter than the Object Lock retention,
+starts then. The tenant provisions a new instance, and the platform deletes
+the old bucket after the grace period.
+
+## Restore
+
+Restore needs only the tenant's vault entry, restic and rclone, never the
+burned cluster. Pick
+the snapshot by the S3 `LastModified` of `snapshots/<id>`, which the server
+sets and Object Lock preserves, never by `latest` or restic's own timestamp:
+someone holding the writer credentials and the password can write a fake
+newer or backdated snapshot. restic 0.18 or later:
+
+```bash
+export RESTIC_REPOSITORY="s3:https://<target-endpoint>/<instance-bucket>"
+export RESTIC_PASSWORD='<instance restic password>'
+export AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_DEFAULT_REGION=default
+R=(restic -o s3.bucket-lookup=path)
+
+"${R[@]}" snapshots --tag "ns:tenant-x"
+
+# one database, into a NEW CNPG cluster, never over the broken one
+"${R[@]}" restore <snapshot-id> --target /restore
+psql -h new-cluster-rw -U postgres -f /restore/pg/app/globals.sql   # first; "role exists" is fine
+pg_restore -h new-cluster-rw -U postgres -C -d postgres /restore/pg/app/app.dump
+# then reset app role passwords from the new cluster's -app Secret
+
+"${R[@]}" restore <snapshot-id> --target /restore   # bucket or PVC snapshot
+```
+
+The restore target needs superuser enabled on the new cluster. Dumps are
+staged at `pg/<cluster>/<db>.dump` and `pg/<cluster>/globals.sql`; the job
+keeps that layout fixed.
+
+If the repo was overwritten, read it as it was before the incident without
+changing anything in the bucket:
+
+```bash
+RCLONE_S3_VERSION_AT='<time before incident>' \
+  restic --no-lock -r rclone:offsite:<instance-bucket> restore <snapshot-id> --target /restore
+```
+
+Tell the tenant the cut time before starting: everything after the last dump
+is lost. A Zitadel database is useless without its master key, which must
+come from outside the cluster.
+
+Drill: once before the schedule is trusted, then yearly or when the vault
+owner changes. On a machine with no cluster access, using only the vault
+entry:
+`snapshots`, restore one dump, restore it into a scratch CNPG cluster, run the
+app against it.
+
+## Code layout
+
+```
+tenant_backup/          runtime binary, no harmony dep: dump, copy, restic, check
+tenant_backup_deploy/   TenantBackupScore<T: K8sclient>, via harmony_cli
+harmony modules/storage BackupInstanceScore<T: ObjectStorageAdmin>
+harmony infra/rgw       ObjectStorageAdmin for Ceph RGW (accounts, IAM users, Object Lock)
+```
+
+`ObjectStorageAdmin` is a capability, not an RGW binding. The existing
+`ObjectBucketScore` cannot be reused: an ObjectBucketClaim cannot set Object
+Lock. Estimated size 1.3 to 1.6k lines; v0's restic/rclone/pg wrappers and
+naming helpers carry over.
+
+## What changes from v0
+
+The prototype gets the job shape right (stage locally, restic offsite, no
+delete on the happy path). These parts go:
+
+| v0 | This design |
+|---|---|
+| Writer credentials shared by every tenant on a cluster (`<cluster-id>/*`) | One writer per instance, IAM user in the platform account |
+| Append-only by IAM alone | Versioning and Object Lock |
+| `-setup` binary from a workstation | `TenantBackupScore` in the tenant's deploy code |
+| Password minted by setup | Generated in the tenant's password manager, pasted once through `harmony_config` |
+| Job logs in to OpenBao with a long-lived Zitadel key in the namespace | Short-lived projected token (pending) |
+| `<cluster>-superuser` assumed | Named Secret, `<cluster>-app` by default |
+| Compressed `-Fc` dumps | `--compress=0`, restic compresses |
+| Monthly repo rotation | New instance, old bucket deleted after the grace period |
+| Planner/worker split, snapshot-to-clone for RWO | Two CronJobs; RWX only |
+| Runtime binary depends on harmony | Runtime and deploy crates split |
+
+## Rejected
+
+- One cluster-wide job with one repo and one password, reading every tenant's
+  secrets through a ClusterRole. It fits a single organisation, not tenants
+  who must not see each other's data, and one failing tenant fails everyone.
+- An operator watching a backup CRD in each namespace. It reimplements what
+  CronJob already schedules. It may come back with tenant self-service or
+  per-tenant coverage status, behind the same Score.
+- `age` per artifact to a tenant public key, instead of restic. The job would
+  hold no key that reads history, but every run is a full copy and hourly runs
+  become unaffordable. Anyone who controls the cluster already has the live
+  data.
+- Password escrow, Shamir splits, HSM wrapping. The job must hold the
+  plaintext key at runtime, and each adds a key or a ceremony to lose.
+- Dumping from the `-ro` replica, co-mounting RWO volumes by pod affinity,
+  `rclone check` after `rclone copy`: more failure modes than they remove at
+  this size.
+
+## Open questions
+
+1. Job identity (above).
+
+CNPG's generated server certificate lists `<cluster>-rw`, `.<ns>`, `.<ns>.svc`
+and `.<ns>.svc.<cluster-domain>` (`api/v1/cluster_funcs.go`
+`GetClusterAltDNSNames`), so `verify-full` against `<cluster>-rw.<ns>.svc`
+with `<cluster>-ca`/`ca.crt` works without extra configuration.

examples/harmony_tenant_backup/Dockerfile

@@ -0,0 +1,22 @@
+# Build first: cargo build --release -p harmony-tenant-backup
+# Build context = target/release (COPY paths), e.g.
+#   docker build -f examples/harmony_tenant_backup/Dockerfile target/release
+FROM docker.io/library/debian:13-slim
+
+# The real pin is this image's digest (setup refuses tags). These two have
+# upstream release cycles outside apt's: the forum bug was rclone 1.60->1.71.
+ARG RESTIC_VERSION=0.18.1
+ARG RCLONE_VERSION=v1.71.0
+
+RUN apt-get update && apt-get install -y --no-install-recommends \
+      ca-certificates curl unzip bzip2 postgresql-client tini \
+ && curl -fsSL "https://github.com/restic/restic/releases/download/v${RESTIC_VERSION}/restic_${RESTIC_VERSION}_linux_amd64.bz2" \
+      | bunzip2 > /usr/local/bin/restic && chmod +x /usr/local/bin/restic \
+ && curl -fsSL "https://downloads.rclone.org/${RCLONE_VERSION}/rclone-${RCLONE_VERSION}-linux-amd64.zip" -o /tmp/r.zip \
+ && unzip -j /tmp/r.zip '*/rclone' -d /usr/local/bin && chmod +x /usr/local/bin/rclone && rm /tmp/r.zip \
+ && apt-get clean && rm -rf /var/lib/apt/lists/*
+
+COPY harmony-tenant-backup /usr/local/bin/harmony-tenant-backup
+
+USER 65534:65534
+ENTRYPOINT ["/usr/bin/tini","--","/usr/local/bin/harmony-tenant-backup"]

examples/harmony_tenant_backup/README.md

@@ -0,0 +1,136 @@
+# harmony-tenant-backup
+
+> **Reference only.** This is the v0 prototype. The current plan is
+> [DESIGN.md](DESIGN.md), which replaces the setup binary, shared credentials
+> and monthly rotation described below.
+
+Backup one tenant namespace. Not restore. Not retention. Not PITR.
+
+Nightly CronJob covering CNPG dumps, opt-in S3 buckets, and PVCs (direct
+mount or snapshot-clone). One restic repo per namespace on the off-site RGW.
+
+Two binaries, same split as `harmony-fleet` vs `harmony-fleet-deploy`:
+
+- `harmony-tenant-backup run` — operate (in-cluster, or a laptop dry-run)
+- `harmony-tenant-backup-setup` — install RBAC + CronJob (workstation)
+
+Out of scope: OpenBao, Ceph, K8s objects, Zitadel.
+
+## Data layout
+
+- Repo: `s3:https://<target-endpoint>/<target-bucket>/<cluster-id>/<ns>`, or
+  `.../<YYYY-MM>` with `rotate_repo_monthly` (expiry = delete a whole prefix).
+- Host pinned to `<cluster-id>-<ns>`. Tags: `run:`, `ns:`, `surface:pg|s3|pvc`,
+  plus `cluster:` / `bucket:` / `pvc:`.
+
+## Runbook
+
+1. Build and push the image:
+
+   ```bash
+   cargo build --release -p harmony-tenant-backup
+   docker build -f examples/harmony_tenant_backup/Dockerfile \
+     -t harbor.example/harmony/ns-backup:v0 target/release
+   docker push harbor.example/harmony/ns-backup:v0
+   ```
+
+2. Config at `<mount>/data/harmony/<ns>/TenantBackup` (`NAMESPACE` maps to
+   store path `harmony/<ns>`):
+
+   ```bash
+    bao kv put -mount=secret harmony/tenant-x/TenantBackup \
+      target_endpoint=https://rgw.site-b target_bucket=tenant-backups \
+      target_access_key=... target_secret_key=... target_region=default \
+      source_endpoint=https://rgw.site-A source_access_key=... \
+      source_secret_key=... rotate_repo_monthly=true
+   ```
+
+   `target_*`: append-only at the target. `source_*`: read-only site-A, only
+   for buckets. `target_ca_pem`: private-CA target only. Never skip TLS.
+   Do not set `restic_password`: setup mints it into OpenBao for the job
+   and prints it once for the offsite kit (KeepassX). Never rotate it.
+
+3. One vault credential in the namespace, plus the bucket opt-in:
+
+   ```bash
+   kubectl create secret generic harmony-tenant-backup-zitadel-key \
+     -n tenant-x --from-file=key.json=<machine-key.json>
+   kubectl annotate ns tenant-x backup.platform/buckets=tenant-x-assets,tenant-x-uploads
+   ```
+
+4. Install:
+
+   ```bash
+   OPENBAO_URL=... OPENBAO_JWT_ROLE=... HARMONY_SSO_URL=... HARMONY_ZITADEL_AUDIENCE=... \
+   HARMONY_ZITADEL_KEY_JSON="$(cat machine-key.json)" \
+   cargo run -p harmony-tenant-backup --bin harmony-tenant-backup-setup -- \
+     --namespace tenant-x --cluster-id site-a-prod \
+     --schedule "17 3 * * *" \
+     --image harbor.example/harmony/ns-backup@sha256:<digest>
+   ```
+
+   Image must be digest-pinned. Stagger schedules per namespace.
+
+5. Run now / watch:
+
+   ```bash
+   kubectl create job backup-now --from=cronjob/harmony-tenant-backup -n tenant-x
+   kubectl logs -n tenant-x job/backup-now -f
+   ```
+
+    Laptop rehearsal needs the same OpenBao login as setup, plus the
+    machine key as `HARMONY_ZITADEL_KEY_JSON` (or `HARMONY_ZITADEL_KEY_PATH`):
+
+    ```bash
+    HARMONY_ZITADEL_KEY_JSON="$(cat machine-key.json)" \
+    harmony-tenant-backup run --namespace tenant-x --cluster-id site-a-prod \
+      --dry-run --scratch /tmp/htb
+    ```
+
+Exit codes: 0 clean, 2 fatal (no repo, no config), 3 partial. Setup exits 1.
+
+## Restore
+
+Fire restore uses the offsite kit (KeepassX), never OpenBao or the burned
+cluster. Same kit as OpenBao unseal keys. Target S3 creds + restic ≥ 0.18:
+
+```bash
+export RESTIC_REPOSITORY="s3:https://rgw.site-b/tenant-backups/site-a-prod/tenant-x/2026-08"
+export RESTIC_PASSWORD='<restic_password from KeepassX>'
+export AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... AWS_DEFAULT_REGION=default
+R=(restic -o s3.bucket-lookup=path)
+
+"${R[@]}" snapshots --tag "ns:tenant-x"
+
+# one database, into a NEW CNPG cluster. Monthly rotation: repo path ends /YYYY-MM;
+# "latest" is only that month. --tag cluster:<cnpg-cluster-name>
+"${R[@]}" restore latest --tag surface:pg --tag cluster:zito --target /restore/pg/zito
+pg_restore -h new-cluster-rw -U postgres -d app /restore/pg/zito/app.dump
+psql -h new-cluster-rw -U postgres -f /restore/pg/zito/globals.sql
+
+"${R[@]}" restore latest --tag bucket:tenant-x-assets --target /restore/assets
+"${R[@]}" restore latest --tag pvc:media --target /restore/media
+```
+
+`globals.sql` is `pg_dumpall --globals-only`: without it, app roles are gone.
+State the RPO to the tenant before starting.
+
+## Retention
+
+No `restic forget`/`prune`. Monthly rotation expires a whole prefix, by hand:
+`rclone delete --rmdirs ":s3:<bucket>/<cluster>/<ns>/2026-05"`. A human on
+the delete button is the design.
+
+## Pilot checklist
+
+1. `harmony-tenant-backup run --namespace <ns> --cluster-id ... --dry-run --scratch /tmp/htb`
+2. NetworkPolicy on CNPG: planner cannot reach `<cluster>-rw`. First failure.
+3. SigV4 = region mismatch; `SecondLevelDomainDenied` = path-style stripped.
+4. emptyDir peak = largest dump or bucket (CronJob ships 5Gi).
+5. `restic unlock` if a killed Job left a lock (`--retry-lock 5m` covers blips).
+6. Restore one database and one PVC into a scratch namespace. Run the app.
+
+## Development
+
+`cargo test -p harmony-tenant-backup`. Acceptance is the pilot.
+`src/common.rs` is the names both binaries share.

examples/harmony_tenant_backup/src/common.rs

@@ -0,0 +1,351 @@
+//! Shared contract: every name, label, and env var setup writes is what
+//! the runtime reads.
+
+use chrono::{DateTime, Utc};
+use k8s_openapi::api::core::v1::{EnvVar, EnvVarSource, SecretKeySelector};
+use std::collections::BTreeMap;
+
+/// Both binaries construct pod env; one place defines the shape.
+pub fn env_var(name: &str, value: &str) -> EnvVar {
+    EnvVar {
+        name: name.into(),
+        value: Some(value.into()),
+        ..Default::default()
+    }
+}
+
+/// Secret-as-env, not a volume: OKD restricted-v2 rejects a pinned
+/// fsGroup, and without fsGroup a 0400 secret mount is unreadable.
+pub fn env_from_secret(name: &str, secret: &str, key: &str) -> EnvVar {
+    EnvVar {
+        name: name.into(),
+        value_from: Some(EnvVarSource {
+            secret_key_ref: Some(SecretKeySelector {
+                name: secret.into(),
+                key: key.into(),
+                optional: Some(false),
+            }),
+            ..Default::default()
+        }),
+        ..Default::default()
+    }
+}
+
+pub const MANAGED_BY: &str = "app.kubernetes.io/managed-by";
+pub const MANAGED_BY_VALUE: &str = "harmony-tenant-backup";
+pub const SA_NAME: &str = "harmony-tenant-backup";
+pub const CRONJOB_NAME: &str = "harmony-tenant-backup";
+
+/// Comma-separated bucket list opt-in, set by the tenant/platform on the
+/// namespace. Absent = "no S3 surface here" (and a coverage gap worth
+/// reporting, never a silent skip).
+pub const A_BUCKETS: &str = "backup.platform/buckets";
+/// Per-PVC opt-out. `true` skips backup, on the record.
+pub const A_PVC_DISABLED: &str = "backup.platform/disabled";
+
+/// Non-secret OpenBao/Zitadel coordinates forwarded verbatim from the planner
+/// environment into worker Jobs. The machine key is env-from-secret
+/// (`HARMONY_ZITADEL_KEY_JSON`), not forwarded and not a volume.
+pub const FORWARD_ENV: [&str; 4] = [
+    "OPENBAO_URL",
+    "OPENBAO_JWT_ROLE",
+    "HARMONY_SSO_URL",
+    "HARMONY_ZITADEL_AUDIENCE",
+];
+pub const ENV_NAMESPACE: &str = "NAMESPACE";
+pub const ENV_CLUSTER_ID: &str = "CLUSTER_ID";
+pub const ENV_WORKER_IMAGE: &str = "WORKER_IMAGE";
+pub const ENV_RUN_ID: &str = "RUN_ID";
+pub const ENV_SCRATCH: &str = "SCRATCH";
+pub const DEFAULT_SCRATCH: &str = "/scratch";
+pub const WORK_MOUNT: &str = "/mnt/data";
+
+pub const ZITADEL_SECRET: &str = "harmony-tenant-backup-zitadel-key";
+pub const ENV_ZITADEL_SECRET: &str = "ZITADEL_SECRET";
+pub const ZITADEL_KEY_FILE: &str = "key.json";
+pub const ENV_ZITADEL_KEY_JSON: &str = "HARMONY_ZITADEL_KEY_JSON";
+
+pub const A_SCC_UID_RANGE: &str = "openshift.io/sa.scc.uid-range";
+
+/// OKD `restricted-v2` injects fsGroup from this annotation and rejects
+/// a pin outside the range. k3d has no SCC: pin 65534 to match the
+/// image USER so emptyDir and PVC clones are readable.
+pub fn volume_fsgroup(annotations: Option<&BTreeMap<String, String>>) -> Option<i64> {
+    match annotations.and_then(|a| a.get(A_SCC_UID_RANGE)) {
+        Some(_) => None,
+        None => Some(65534),
+    }
+}
+
+/// The planner's own CronJob budget; setup writes it, the runtime bounds its
+/// patience by WORKER_DEADLINE_SECS.
+pub const PLANNER_DEADLINE_SECS: i64 = 6 * 3600;
+
+// One worker push may run the full 4h; the planner's per-job patience is the
+// worker deadline plus slack, and the planner's own CronJob deadline must
+// exceed snapshot + clone + worker budgets. Mismatching these orphans
+// run-id-named snapshots when the pod is killed mid-cleanup.
+pub const WORKER_DEADLINE_SECS: i64 = 4 * 3600;
+
+/// Job/snapshot/clone names are DNS-1123: lowercase, digits, hyphen.
+pub fn default_run_id(now: DateTime<Utc>) -> String {
+    now.format("%Y%m%dt%H%M%Sz").to_string()
+}
+
+pub fn dns1123_run_id(s: &str) -> Option<String> {
+    let s = s.to_ascii_lowercase();
+    let ok = (1..=32).contains(&s.len())
+        && s.chars()
+            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-');
+    ok.then_some(s)
+}
+
+/// restic `--host`: never the pod name, or "restore latest" goes ambiguous
+/// across rebuilt Jobs.
+pub fn restic_host(cluster_id: &str, namespace: &str) -> String {
+    format!("{cluster_id}-{namespace}")
+}
+
+pub fn repo_prefix(cluster_id: &str, namespace: &str) -> String {
+    format!("{cluster_id}/{namespace}")
+}
+
+/// Tokens of the `backup.platform/buckets` annotation. Tenant-authored, so
+/// hold them to S3's own grammar: a token escaping it could climb out of the
+/// staging dir (`..`), or parse as an rclone connection-string option that
+/// redirects the source credentials somewhere else.
+pub fn valid_bucket_token(b: &str) -> bool {
+    let alnum = |c: char| c.is_ascii_lowercase() || c.is_ascii_digit();
+    !b.is_empty()
+        && b.len() <= 63
+        && b.chars().all(|c| alnum(c) || c == '.' || c == '-')
+        && !b.contains("..")
+        && b.starts_with(alnum)
+        && b.ends_with(alnum)
+}
+
+/// restic repo URL. Monthly rotation puts each run in a fresh repo per
+/// period; expiry then means deleting a whole prefix, never partial-repo
+/// surgery on immutable shared packs.
+pub fn repo_path(
+    endpoint: &str,
+    bucket: &str,
+    cluster_id: &str,
+    namespace: &str,
+    monthly: bool,
+    now: DateTime<Utc>,
+) -> String {
+    // `s3:` prefix mandatory: a bare https:// location parses as the REST
+    // backend, silently writing nothing.
+    let base = format!(
+        "s3:{endpoint}/{bucket}/{}",
+        repo_prefix(cluster_id, namespace)
+    );
+    if monthly {
+        format!("{base}/{}", now.format("%Y-%m"))
+    } else {
+        base
+    }
+}
+
+/// restic backup args: one pinned host + tag scheme for every push, so
+/// restores address snapshots exactly per the README's restore interface.
+pub fn backup_tags(run_id: &str, namespace: &str, surface: &str) -> Vec<String> {
+    vec![
+        format!("run:{run_id}"),
+        format!("ns:{namespace}"),
+        format!("surface:{surface}"),
+    ]
+}
+
+/// K8s names are DNS-1123 and 63 chars max: `<prefix>-<pvc>-<run_id>`,
+/// truncated front-first on the volatile parts to stay collision-free
+/// within a run.
+pub fn run_object_name(prefix: &str, pvc: &str, run_id: &str) -> String {
+    let full = format!("{prefix}-{pvc}-{run_id}");
+    if full.len() <= 63 {
+        return full;
+    }
+    let keep = 63usize
+        .saturating_sub(prefix.len())
+        .saturating_sub(run_id.len())
+        .saturating_sub(2);
+    let mut cut = pvc.chars().take(keep).collect::<String>();
+    cut = cut.trim_matches(|c| c == '-' || c == '.').to_string();
+    format!("{prefix}-{cut}-{run_id}")
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum PvcDecision {
+    /// Mount the PVC directly (RWX and unattached).
+    Direct,
+    /// RWO: VolumeSnapshot -> clone -> mount the clone.
+    SnapshotFirst,
+    Skip(&'static str),
+    /// Cannot be backed up by this mechanism; must surface as a failure.
+    Fail(&'static str),
+}
+
+/// Classify one PVC for backup. `cnpg_owner` is the `cnpg.io/cluster` label.
+pub fn decide_pvc(
+    phase: &str,
+    cnpg_owner: Option<&str>,
+    disabled: bool,
+    block: bool,
+    access_mode: &str,
+) -> PvcDecision {
+    if phase != "Bound" {
+        return PvcDecision::Skip("phase is not Bound");
+    }
+    if cnpg_owner.is_some() {
+        return PvcDecision::Skip("owned by a CNPG cluster (covered by pg_dump)");
+    }
+    if disabled {
+        return PvcDecision::Skip("backup.platform/disabled=true");
+    }
+    if block {
+        return PvcDecision::Fail(
+            "volumeMode=Block: restic cannot read it, needs app-native backup",
+        );
+    }
+    if access_mode == "ReadWriteMany" {
+        PvcDecision::Direct
+    } else {
+        // RWO/RWX-but-attached volumes block multi-attach: snapshot -> clone.
+        PvcDecision::SnapshotFirst
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::TimeZone;
+
+    fn t() -> DateTime<Utc> {
+        Utc.with_ymd_and_hms(2026, 9, 17, 3, 17, 0).unwrap()
+    }
+
+    #[test]
+    fn repo_path_plain_and_monthly() {
+        assert_eq!(
+            repo_path(
+                "https://rgw.site-b",
+                "tenant-backups",
+                "site-a-prod",
+                "tenant-x",
+                false,
+                t()
+            ),
+            "s3:https://rgw.site-b/tenant-backups/site-a-prod/tenant-x"
+        );
+        assert_eq!(
+            repo_path(
+                "https://rgw.site-b",
+                "tenant-backups",
+                "site-a-prod",
+                "tenant-x",
+                true,
+                t()
+            ),
+            "s3:https://rgw.site-b/tenant-backups/site-a-prod/tenant-x/2026-09"
+        );
+    }
+
+    #[test]
+    fn fsgroup_k3d_vs_okd() {
+        assert_eq!(volume_fsgroup(None), Some(65534));
+        let mut okd = BTreeMap::new();
+        okd.insert(A_SCC_UID_RANGE.into(), "1000700000/10000".into());
+        assert_eq!(volume_fsgroup(Some(&okd)), None);
+    }
+
+    #[test]
+    fn host_is_pinned_not_pod_name() {
+        assert_eq!(
+            restic_host("site-a-prod", "tenant-x"),
+            "site-a-prod-tenant-x"
+        );
+    }
+
+    #[test]
+    fn run_object_name_stays_dns_safe() {
+        let long_pvc = "a".repeat(80);
+        let n = run_object_name("htb-wk", &long_pvc, "20260917t031700z");
+        assert!(n.len() <= 63, "{n}");
+        assert!(n.starts_with("htb-wk-") && n.ends_with("-20260917t031700z"));
+        assert_eq!(
+            run_object_name("htb-snap", "media", "R1"),
+            "htb-snap-media-R1"
+        );
+        let long_run = "r".repeat(70);
+        let n = run_object_name("htb-wk", "media", &long_run);
+        assert!(
+            n.len() <= 70 + 20,
+            "no underflow, api rejects long ones: {n}"
+        );
+    }
+
+    #[test]
+    fn bucket_tokens_hold_to_s3_grammar() {
+        assert!(valid_bucket_token("tenant-x.assets-1"));
+        for bad in [
+            "",
+            "-lead",
+            "UPPER",
+            "../etc",
+            "a..b",
+            "bucket,opt=evil",
+            "x/y",
+        ] {
+            assert!(!valid_bucket_token(bad), "{bad}");
+        }
+    }
+
+    #[test]
+    fn classification() {
+        assert_eq!(
+            decide_pvc("Bound", None, false, false, "ReadWriteOnce"),
+            PvcDecision::SnapshotFirst
+        );
+        assert_eq!(
+            decide_pvc("Bound", None, false, false, "ReadWriteMany"),
+            PvcDecision::Direct
+        );
+        assert!(matches!(
+            decide_pvc("Pending", None, false, false, "ReadWriteOnce"),
+            PvcDecision::Skip(_)
+        ));
+        assert!(matches!(
+            decide_pvc("Bound", Some("zito"), false, false, "ReadWriteOnce"),
+            PvcDecision::Skip(_)
+        ));
+        assert!(matches!(
+            decide_pvc("Bound", None, true, false, "ReadWriteOnce"),
+            PvcDecision::Skip(_)
+        ));
+        assert!(matches!(
+            decide_pvc("Bound", None, false, true, "ReadWriteOnce"),
+            PvcDecision::Fail(_)
+        ));
+    }
+
+    #[test]
+    fn run_id_is_dns1123() {
+        let id = default_run_id(t());
+        assert_eq!(id, "20260917t031700z");
+        assert_eq!(
+            dns1123_run_id("20260917T031700Z").as_deref(),
+            Some("20260917t031700z")
+        );
+        assert!(dns1123_run_id("Nope_Underscore").is_none());
+        assert!(dns1123_run_id(&"a".repeat(33)).is_none());
+    }
+
+    #[test]
+    fn tags_are_the_restore_addressing_scheme() {
+        assert_eq!(
+            backup_tags("20260917T031700Z", "tenant-x", "pg"),
+            vec!["run:20260917T031700Z", "ns:tenant-x", "surface:pg"]
+        );
+    }
+}

examples/harmony_tenant_backup/src/exec.rs

@@ -0,0 +1,345 @@
+//! Subprocess wrappers: restic, rclone, postgres client. Credentials travel
+//! via environment only, never argv.
+
+use anyhow::{Context, Result, anyhow};
+use std::path::{Path, PathBuf};
+use std::process::Stdio;
+use tokio::process::Command;
+
+/// Single place where the restic repo path + credentials are assembled, so
+/// planner and worker can never disagree about where data lands.
+pub fn build_restic(
+    cfg: &crate::TenantBackup,
+    cluster_id: &str,
+    ns: &str,
+    ca_file: Option<PathBuf>,
+    scratch: &Path,
+) -> Restic {
+    Restic::new(
+        crate::common::repo_path(
+            &cfg.target_endpoint,
+            &cfg.target_bucket,
+            cluster_id,
+            ns,
+            cfg.rotate_repo_monthly,
+            chrono::Utc::now(),
+        ),
+        cfg.restic_password.clone(),
+        cfg.target_access_key.clone(),
+        cfg.target_secret_key.clone(),
+        cfg.target_region.clone(),
+        ca_file,
+        scratch,
+    )
+}
+
+pub fn logline(msg: &str) {
+    eprintln!("[{}] {msg}", chrono::Utc::now().format("%H:%M:%S"));
+}
+
+pub async fn run_capture(program: &str, args: &[&str], env: &[(&str, &str)]) -> Result<String> {
+    let out = Command::new(program)
+        .args(args)
+        .envs(env.iter().copied())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .output()
+        .await
+        .with_context(|| format!("spawn {program}"))?;
+    if !out.status.success() {
+        return Err(anyhow!(
+            "{program} failed ({})\nstderr: {}",
+            out.status,
+            tail(&String::from_utf8_lossy(&out.stderr))
+        ));
+    }
+    Ok(String::from_utf8_lossy(&out.stdout).into_owned())
+}
+
+async fn run_quiet(program: &str, args: &[&str], env: &[(&str, &str)]) -> Result<()> {
+    run_capture(program, args, env).await.map(|_| ())
+}
+
+fn strs(v: &[String]) -> Vec<&str> {
+    v.iter().map(String::as_str).collect()
+}
+
+fn tail(s: &str) -> String {
+    s.trim()
+        .lines()
+        .rev()
+        .take(8)
+        .collect::<Vec<_>>()
+        .into_iter()
+        .rev()
+        .collect::<Vec<_>>()
+        .join("\n")
+}
+
+pub struct Restic {
+    repo: String,
+    password: String,
+    access_key: String,
+    secret_key: String,
+    region: Option<String>,
+    ca_file: Option<PathBuf>,
+    cache_dir: PathBuf,
+}
+
+impl Restic {
+    pub fn repo(&self) -> &str {
+        &self.repo
+    }
+
+    pub fn new(
+        repo: String,
+        password: String,
+        access_key: String,
+        secret_key: String,
+        region: Option<String>,
+        ca_file: Option<PathBuf>,
+        scratch: &Path,
+    ) -> Self {
+        Self {
+            repo,
+            password,
+            access_key,
+            secret_key,
+            region,
+            ca_file,
+            cache_dir: scratch.join("cache"),
+        }
+    }
+
+    fn global_args(&self) -> Vec<String> {
+        let mut a = vec![
+            "-r".into(),
+            self.repo.clone(),
+            "-o".into(),
+            "s3.bucket-lookup=path".into(),
+            "--cache-dir".into(),
+            self.cache_dir.display().to_string(),
+            "--retry-lock".into(),
+            "5m".into(),
+        ];
+        if let Some(ca) = &self.ca_file {
+            a.push(format!("--cacert={}", ca.display()));
+        }
+        a
+    }
+
+    fn env(&self) -> Vec<(&'static str, &str)> {
+        // restic's child-process contract. The value is TenantBackup.restic_password
+        // from harmony_config (OpenBao), never a RESTIC_PASSWORD on the pod.
+        let mut e = vec![
+            ("RESTIC_PASSWORD", self.password.as_str()),
+            ("AWS_ACCESS_KEY_ID", self.access_key.as_str()),
+            ("AWS_SECRET_ACCESS_KEY", self.secret_key.as_str()),
+        ];
+        if let Some(r) = &self.region {
+            // restic reads AWS_DEFAULT_REGION, not AWS_REGION.
+            e.push(("AWS_DEFAULT_REGION", r.as_str()));
+        }
+        e
+    }
+
+    pub async fn ensure_repo(&self, dry_run: bool) -> Result<()> {
+        let mut args = self.global_args();
+        args.extend(["cat".into(), "config".into()]);
+        if run_quiet("restic", &strs(&args), &self.env()).await.is_ok() {
+            return Ok(());
+        }
+        logline(&format!("initialising repo {}", self.repo));
+        if dry_run {
+            return Ok(());
+        }
+        let mut args = self.global_args();
+        args.push("init".into());
+        run_quiet("restic", &strs(&args), &self.env())
+            .await
+            .context("restic init")
+    }
+
+    /// The one push path: pinned host + `run:`/`ns:`/`surface:` tags.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn backup(
+        &self,
+        host: &str,
+        run_id: &str,
+        namespace: &str,
+        surface: &str,
+        extra_tags: &[String],
+        path: &str,
+        dry_run: bool,
+    ) -> Result<()> {
+        if dry_run {
+            logline(&format!("DRY-RUN: would push {path} ({surface})"));
+            return Ok(());
+        }
+        let mut args = self.global_args();
+        args.push("backup".into());
+        args.extend(["--host".into(), host.into()]);
+        for tag in crate::common::backup_tags(run_id, namespace, surface)
+            .into_iter()
+            .chain(extra_tags.iter().cloned())
+        {
+            args.extend(["--tag".into(), tag]);
+        }
+        args.push(path.into());
+        let out = run_capture("restic", &strs(&args), &self.env()).await?;
+        logline(&format!(
+            "pushed {path} ({surface}): {}",
+            tail(&out).lines().last().unwrap_or("ok")
+        ));
+        Ok(())
+    }
+
+    pub async fn snapshots_tagged(&self, tag: &str) -> Result<()> {
+        let mut args = self.global_args();
+        args.extend(["snapshots".into(), "--tag".into(), tag.into()]);
+        let out = run_capture("restic", &strs(&args), &self.env()).await?;
+        eprint!("{out}");
+        Ok(())
+    }
+}
+
+/// Read-only tenant source on the site-A RGW (bucket credentials).
+pub struct SrcS3 {
+    pub endpoint: String,
+    pub region: String,
+    pub access_key: String,
+    pub secret_key: String,
+}
+
+impl SrcS3 {
+    fn args(&self) -> Vec<String> {
+        vec![
+            "--s3-endpoint".into(),
+            self.endpoint.clone(),
+            "--s3-provider=Ceph".into(),
+            "--s3-region".into(),
+            self.region.clone(),
+            "--s3-force-path-style".into(),
+            "--s3-env-auth".into(),
+            "--s3-no-check-bucket".into(),
+            "--disable-http2".into(),
+        ]
+    }
+
+    fn env(&self) -> Vec<(&'static str, &str)> {
+        vec![
+            ("AWS_ACCESS_KEY_ID", self.access_key.as_str()),
+            ("AWS_SECRET_ACCESS_KEY", self.secret_key.as_str()),
+        ]
+    }
+
+    fn cmd<'a>(&self, sub: &'a str, paths_and_flags: &'a [String]) -> Vec<&'a str> {
+        let mut argv: Vec<&str> = vec![sub];
+        argv.extend(paths_and_flags.iter().map(String::as_str));
+        argv
+    }
+
+    /// `copy`, never `sync`: add/overwrite only, deletes never propagate.
+    pub async fn copy(&self, bucket: &str, dst: &Path) -> Result<()> {
+        let mut args = vec![":s3:".to_string() + bucket, dst.display().to_string()];
+        args.extend(self.args());
+        args.extend([
+            "--transfers".into(),
+            "8".into(),
+            "--checkers".into(),
+            "16".into(),
+            "--stats-one-line".into(),
+            "-v".into(),
+            "--ignore-checksum".into(),
+        ]);
+        run_quiet("rclone", &self.cmd("copy", &args), &self.env())
+            .await
+            .context("rclone copy")
+    }
+
+    /// Count-only assertion against the staged copy (multi-site ETags differ
+    /// by construction, so content hashes would always "mismatch").
+    pub async fn check(&self, bucket: &str, dst: &Path) -> Result<()> {
+        let mut args = vec![":s3:".to_string() + bucket, dst.display().to_string()];
+        args.extend(self.args());
+        args.extend(["--one-way".into(), "--count".into()]);
+        run_quiet("rclone", &self.cmd("check", &args), &self.env())
+            .await
+            .context("rclone check mismatch: staged copy not trustworthy")
+    }
+
+    pub async fn list_dirs(&self, bucket: &str) -> Result<()> {
+        let mut args = vec![":s3:".to_string() + bucket];
+        args.extend(self.args());
+        let out = run_capture("rclone", &self.cmd("lsd", &args), &self.env()).await?;
+        eprint!("{out}");
+        Ok(())
+    }
+}
+
+/// All postgres wire calls go through the CNPG `<cluster>-rw` service with
+/// TLS verified against the cluster CA secret.
+pub struct Pg {
+    conn_base: String,
+    password: String,
+}
+
+impl Pg {
+    pub fn new(cluster: &str, namespace: &str, ca_file: &Path, password: String) -> Self {
+        let conn_base = format!(
+            "host={cluster}-rw.{namespace}.svc port=5432 user=postgres sslmode=verify-ca sslrootcert={}",
+            ca_file.display()
+        );
+        Self {
+            conn_base,
+            password,
+        }
+    }
+
+    fn conn(&self, db: &str) -> String {
+        format!("{} dbname={db}", self.conn_base)
+    }
+
+    fn env(&self) -> Vec<(&'static str, &str)> {
+        vec![("PGPASSWORD", self.password.as_str())]
+    }
+
+    pub async fn query(&self, db: &str, sql: &str) -> Result<String> {
+        run_capture("psql", &[&self.conn(db), "-Atc", sql], &self.env()).await
+    }
+
+    pub async fn dump_all_globals(&self, out_file: &Path) -> Result<()> {
+        run_quiet(
+            "pg_dumpall",
+            &[
+                &self.conn("postgres"),
+                "--globals-only",
+                "-f",
+                &out_file.display().to_string(),
+            ],
+            &self.env(),
+        )
+        .await
+        .context("pg_dumpall --globals-only: a DB-only dump loses the app roles")
+    }
+
+    pub async fn dump_db(&self, db: &str, out_file: &Path) -> Result<()> {
+        run_quiet(
+            "pg_dump",
+            &[
+                &self.conn(db),
+                "--format=custom",
+                "--file",
+                &out_file.display().to_string(),
+            ],
+            &self.env(),
+        )
+        .await
+        .context("pg_dump")
+    }
+
+    /// The artifact must be a readable archive, not just a file that exists.
+    pub async fn verify_dump(path: &Path) -> Result<()> {
+        run_quiet("pg_restore", &["--list", &path.display().to_string()], &[]).await
+    }
+}

examples/harmony_tenant_backup/src/k8s.rs

@@ -0,0 +1,475 @@
+//! Cluster mechanics: discovery reads, VolumeSnapshot/clone lifecycle,
+//! worker-Job creation and completion polling.
+
+use crate::common::{
+    ENV_ZITADEL_KEY_JSON, MANAGED_BY, MANAGED_BY_VALUE, WORKER_DEADLINE_SECS, ZITADEL_KEY_FILE,
+    env_from_secret, env_var,
+};
+use anyhow::{Result, anyhow};
+use k8s_openapi::api::core::v1::{
+    Capabilities, EmptyDirVolumeSource, PersistentVolumeClaim, PersistentVolumeClaimSpec,
+    PodSecurityContext, SeccompProfile, TypedLocalObjectReference, Volume, VolumeMount,
+    VolumeResourceRequirements,
+};
+use k8s_openapi::api::{batch::v1::Job, core::v1::SecurityContext};
+use k8s_openapi::apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::ObjectMeta};
+use kube::api::{
+    Api, ApiResource, DeleteParams, DynamicObject, ListParams, PostParams, PropagationPolicy,
+};
+use kube::{Client, ResourceExt};
+use std::collections::BTreeMap;
+use std::time::Duration;
+use tokio::time::interval;
+
+pub fn api_resource(group: &str, version: &str, kind: &str, plural: &str) -> ApiResource {
+    ApiResource {
+        group: group.into(),
+        version: version.into(),
+        api_version: format!("{group}/{version}"),
+        kind: kind.into(),
+        plural: plural.into(),
+    }
+}
+
+pub fn cnpg_clusters() -> ApiResource {
+    api_resource("postgresql.cnpg.io", "v1", "Cluster", "clusters")
+}
+
+pub fn volume_snapshots() -> ApiResource {
+    api_resource(
+        "snapshot.storage.k8s.io",
+        "v1",
+        "VolumeSnapshot",
+        "volumesnapshots",
+    )
+}
+
+pub fn volume_snapshot_classes() -> ApiResource {
+    api_resource(
+        "snapshot.storage.k8s.io",
+        "v1",
+        "VolumeSnapshotClass",
+        "volumesnapshotclasses",
+    )
+}
+
+fn sorted_names(list: kube::core::ObjectList<DynamicObject>) -> Vec<String> {
+    let mut names: Vec<String> = list
+        .iter()
+        .filter_map(|i| i.metadata.name.clone())
+        .collect();
+    names.sort();
+    names
+}
+
+pub async fn list_namespaced_names(
+    client: &Client,
+    ns: &str,
+    ar: &ApiResource,
+) -> Result<Vec<String>> {
+    let api: Api<DynamicObject> = Api::namespaced_with(client.clone(), ns, ar);
+    Ok(sorted_names(api.list(&ListParams::default()).await?))
+}
+
+/// The one cluster-scoped name, if any. Picking silently when more than one
+/// exists can route a snapshot to the wrong CSI driver.
+pub async fn first_cluster_scoped_name(
+    client: &Client,
+    ar: &ApiResource,
+) -> Result<Option<String>> {
+    let api: Api<DynamicObject> = Api::all_with(client.clone(), ar);
+    let names = sorted_names(api.list(&ListParams::default()).await?);
+    match names.as_slice() {
+        [] => Ok(None),
+        [one] => Ok(Some(one.clone())),
+        many => Err(anyhow!(
+            "{} has {} classes and none is configured; set snapshot_class (found {})",
+            ar.kind,
+            many.len(),
+            many.join(", ")
+        )),
+    }
+}
+
+pub async fn ns_annotations(client: &Client, ns: &str) -> Result<BTreeMap<String, String>> {
+    Ok(
+        Api::<k8s_openapi::api::core::v1::Namespace>::all(client.clone())
+            .get(ns)
+            .await?
+            .metadata
+            .annotations
+            .unwrap_or_default(),
+    )
+}
+
+pub async fn annotation(client: &Client, ns: &str, key: &str) -> Result<Option<String>> {
+    Ok(ns_annotations(client, ns).await?.get(key).cloned())
+}
+
+/// Restricted-v2 compatible container lock-down. No runAsUser: OKD assigns
+/// the namespace UID; k3d uses the image USER.
+pub fn container_security() -> SecurityContext {
+    SecurityContext {
+        allow_privilege_escalation: Some(false),
+        read_only_root_filesystem: Some(true),
+        run_as_non_root: Some(true),
+        capabilities: Some(Capabilities {
+            drop: Some(vec!["ALL".into()]),
+            ..Default::default()
+        }),
+        seccomp_profile: Some(SeccompProfile {
+            type_: "RuntimeDefault".into(),
+            ..Default::default()
+        }),
+        ..Default::default()
+    }
+}
+
+pub fn pod_security(fs_group: Option<i64>) -> PodSecurityContext {
+    PodSecurityContext {
+        run_as_non_root: Some(true),
+        fs_group,
+        seccomp_profile: Some(SeccompProfile {
+            type_: "RuntimeDefault".into(),
+            ..Default::default()
+        }),
+        ..Default::default()
+    }
+}
+
+pub async fn secret_bytes(client: &Client, ns: &str, name: &str, key: &str) -> Result<Vec<u8>> {
+    let secret = Api::<k8s_openapi::api::core::v1::Secret>::namespaced(client.clone(), ns)
+        .get(name)
+        .await
+        .map_err(|e| anyhow!("secret {ns}/{name}: {e}"))?;
+    secret
+        .data
+        .as_ref()
+        .and_then(|d| d.get(key))
+        .map(|b| b.0.clone())
+        .ok_or_else(|| anyhow!("secret {ns}/{name} has no key '{key}'"))
+}
+
+fn managed_labels(run_id: &str) -> BTreeMap<String, String> {
+    BTreeMap::from([
+        (MANAGED_BY.to_string(), MANAGED_BY_VALUE.to_string()),
+        ("run".to_string(), run_id.to_string()),
+    ])
+}
+
+pub async fn create_volume_snapshot(
+    client: &Client,
+    ns: &str,
+    name: &str,
+    class: &str,
+    pvc: &str,
+    run_id: &str,
+) -> Result<()> {
+    let ar = volume_snapshots();
+    let mut obj = DynamicObject::new(name, &ar);
+    obj.metadata.namespace = Some(ns.into());
+    obj.metadata.labels = Some(managed_labels(run_id));
+    obj.data["spec"] = serde_json::json!({
+        "volumeSnapshotClassName": class,
+        "source": { "persistentVolumeClaimName": pvc },
+    });
+    Api::<DynamicObject>::namespaced_with(client.clone(), ns, &ar)
+        .create(&PostParams::default(), &obj)
+        .await?;
+    Ok(())
+}
+
+pub async fn create_clone_pvc(
+    client: &Client,
+    ns: &str,
+    name: &str,
+    source: &PersistentVolumeClaim,
+    snap: &str,
+    run_id: &str,
+) -> Result<()> {
+    let size = source
+        .status
+        .as_ref()
+        .and_then(|s| s.capacity.as_ref())
+        .and_then(|c| c.get("storage"))
+        .cloned()
+        .unwrap_or_else(|| Quantity("1Gi".into()));
+    let mut labels = managed_labels(run_id);
+    labels.insert("backup.platform/source-pvc".into(), source.name_any());
+    let pvc = PersistentVolumeClaim {
+        metadata: ObjectMeta {
+            name: Some(name.into()),
+            namespace: Some(ns.into()),
+            labels: Some(labels),
+            ..Default::default()
+        },
+        spec: Some(PersistentVolumeClaimSpec {
+            access_modes: Some(vec!["ReadWriteOnce".into()]),
+            volume_mode: Some("Filesystem".into()),
+            storage_class_name: source
+                .spec
+                .as_ref()
+                .and_then(|s| s.storage_class_name.clone()),
+            resources: Some(VolumeResourceRequirements {
+                requests: Some(BTreeMap::from([("storage".into(), size)])),
+                ..Default::default()
+            }),
+            data_source: Some(TypedLocalObjectReference {
+                api_group: Some("snapshot.storage.k8s.io".into()),
+                kind: "VolumeSnapshot".into(),
+                name: snap.into(),
+            }),
+            ..Default::default()
+        }),
+        ..Default::default()
+    };
+    Api::<PersistentVolumeClaim>::namespaced(client.clone(), ns)
+        .create(&PostParams::default(), &pvc)
+        .await?;
+    Ok(())
+}
+
+pub struct WorkerSpec<'a> {
+    pub image: &'a str,
+    pub ns: &'a str,
+    pub name: &'a str,
+    pub pvc_mount: &'a str,
+    pub pvc_logical: &'a str,
+    pub run_id: &'a str,
+    pub cluster_id: &'a str,
+    /// Same in-cluster Zitadel secret the planner uses (env-from, not a mount).
+    pub zitadel_secret: &'a str,
+    /// OpenBao/Zitadel coordinates forwarded from this process's environment.
+    pub forward_env: Vec<(&'a str, &'a str)>,
+    /// None on OKD (SCC assigns); Some(65534) on k3d so PVC clones are readable.
+    pub fs_group: Option<i64>,
+}
+
+pub fn worker_job(w: &WorkerSpec) -> Job {
+    let mut env = vec![
+        env_var(crate::common::ENV_NAMESPACE, w.ns),
+        env_var(crate::common::ENV_RUN_ID, w.run_id),
+        env_var(crate::common::ENV_CLUSTER_ID, w.cluster_id),
+        env_from_secret(ENV_ZITADEL_KEY_JSON, w.zitadel_secret, ZITADEL_KEY_FILE),
+    ];
+    env.extend(w.forward_env.iter().map(|(k, v)| env_var(k, v)));
+
+    Job {
+        metadata: ObjectMeta {
+            name: Some(w.name.into()),
+            namespace: Some(w.ns.into()),
+            labels: Some(managed_labels(w.run_id)),
+            ..Default::default()
+        },
+        spec: Some(k8s_openapi::api::batch::v1::JobSpec {
+            backoff_limit: Some(0),
+            active_deadline_seconds: Some(WORKER_DEADLINE_SECS),
+            // Keep finished worker pods long enough to read their logs the
+            // morning after, when the planner only reports FAILURES by name.
+            ttl_seconds_after_finished: Some(86400),
+            template: k8s_openapi::api::core::v1::PodTemplateSpec {
+                metadata: Some(ObjectMeta {
+                    labels: Some(managed_labels(w.run_id)),
+                    ..Default::default()
+                }),
+                spec: Some(k8s_openapi::api::core::v1::PodSpec {
+                    restart_policy: Some("Never".into()),
+                    service_account_name: Some(crate::common::SA_NAME.into()),
+                    security_context: Some(pod_security(w.fs_group)),
+                    containers: vec![k8s_openapi::api::core::v1::Container {
+                        name: "worker".into(),
+                        image: Some(w.image.into()),
+                        args: Some(vec!["worker".into(), "--pvc".into(), w.pvc_logical.into()]),
+                        env: Some(env),
+                        volume_mounts: Some(vec![
+                            VolumeMount {
+                                name: "data".into(),
+                                mount_path: crate::common::WORK_MOUNT.into(),
+                                read_only: Some(true),
+                                ..Default::default()
+                            },
+                            VolumeMount {
+                                name: "scratch".into(),
+                                mount_path: crate::common::DEFAULT_SCRATCH.into(),
+                                ..Default::default()
+                            },
+                        ]),
+                        resources: Some(k8s_openapi::api::core::v1::ResourceRequirements {
+                            requests: Some(BTreeMap::from([
+                                ("cpu".into(), Quantity("100m".into())),
+                                ("memory".into(), Quantity("256Mi".into())),
+                            ])),
+                            limits: Some(BTreeMap::from([(
+                                "memory".into(),
+                                Quantity("1Gi".into()),
+                            )])),
+                            ..Default::default()
+                        }),
+                        security_context: Some(container_security()),
+                        ..Default::default()
+                    }],
+                    volumes: Some(vec![
+                        Volume {
+                            name: "data".into(),
+                            persistent_volume_claim: Some(
+                                k8s_openapi::api::core::v1::PersistentVolumeClaimVolumeSource {
+                                    claim_name: w.pvc_mount.into(),
+                                    read_only: Some(true),
+                                },
+                            ),
+                            ..Default::default()
+                        },
+                        Volume {
+                            name: "scratch".into(),
+                            empty_dir: Some(EmptyDirVolumeSource {
+                                size_limit: Some(Quantity("2Gi".into())),
+                                ..Default::default()
+                            }),
+                            ..Default::default()
+                        },
+                    ]),
+                    ..Default::default()
+                }),
+            },
+            ..Default::default()
+        }),
+        ..Default::default()
+    }
+}
+
+pub async fn spawn_and_wait_job(
+    client: &Client,
+    ns: &str,
+    job: &Job,
+    patience: Duration,
+) -> Result<()> {
+    let api = Api::<Job>::namespaced(client.clone(), ns);
+    let name = job.metadata.name.clone().unwrap_or_default();
+    api.create(&PostParams::default(), job).await?;
+    wait_job(client, ns, &name, patience).await
+}
+
+pub async fn wait_job(client: &Client, ns: &str, name: &str, patience: Duration) -> Result<()> {
+    let api = Api::<Job>::namespaced(client.clone(), ns);
+    let mut tick = interval(Duration::from_secs(15));
+    let deadline = tokio::time::Instant::now() + patience;
+    loop {
+        tick.tick().await;
+        let job = api.get(name).await?;
+        let status = job.status.unwrap_or_default();
+        if status.succeeded.unwrap_or(0) >= 1 {
+            return Ok(());
+        }
+        if let Some(f) = status.conditions.as_ref().and_then(|cs| {
+            cs.iter()
+                .find(|c| c.type_ == "Failed" && c.status == "True")
+        }) {
+            return Err(anyhow!(
+                "job {name} failed: {}",
+                f.message
+                    .as_deref()
+                    .or(f.reason.as_deref())
+                    .unwrap_or("unknown")
+            ));
+        }
+        if tokio::time::Instant::now() > deadline {
+            return Err(anyhow!("job {name} did not finish within {patience:?}"));
+        }
+    }
+}
+
+pub async fn wait_snapshot_ready(
+    client: &Client,
+    ns: &str,
+    name: &str,
+    patience: Duration,
+) -> Result<()> {
+    let ar = volume_snapshots();
+    let api: Api<DynamicObject> = Api::namespaced_with(client.clone(), ns, &ar);
+    let mut tick = interval(Duration::from_secs(10));
+    let deadline = tokio::time::Instant::now() + patience;
+    loop {
+        tick.tick().await;
+        let obj = api.get(name).await?;
+        if obj
+            .data
+            .pointer("/status/readyToUse")
+            .and_then(|v| v.as_bool())
+            == Some(true)
+        {
+            return Ok(());
+        }
+        if let Some(err) = obj.data.pointer("/status/error") {
+            return Err(anyhow!("volumesnapshot {name} failed: {err}"));
+        }
+        if tokio::time::Instant::now() > deadline {
+            return Err(anyhow!(
+                "volumesnapshot {name} not readyToUse within {patience:?}"
+            ));
+        }
+    }
+}
+
+pub async fn wait_pvc_bound(
+    client: &Client,
+    ns: &str,
+    name: &str,
+    patience: Duration,
+) -> Result<()> {
+    let api = Api::<PersistentVolumeClaim>::namespaced(client.clone(), ns);
+    let mut tick = interval(Duration::from_secs(10));
+    let deadline = tokio::time::Instant::now() + patience;
+    loop {
+        tick.tick().await;
+        let pvc = api.get(name).await?;
+        if pvc.status.as_ref().and_then(|s| s.phase.as_deref()) == Some("Bound") {
+            return Ok(());
+        }
+        if tokio::time::Instant::now() > deadline {
+            return Err(anyhow!("pvc {name} not Bound within {patience:?}"));
+        }
+    }
+}
+
+pub async fn delete_gone(
+    client: &Client,
+    ns: &str,
+    name: &str,
+    snapshot: bool,
+    patience: Duration,
+) -> Result<()> {
+    let del = DeleteParams {
+        propagation_policy: Some(PropagationPolicy::Foreground),
+        ..Default::default()
+    };
+    let outcome = if snapshot {
+        let ar = volume_snapshots();
+        let api: Api<DynamicObject> = Api::namespaced_with(client.clone(), ns, &ar);
+        let _ = api.delete(name, &del).await;
+        wait_gone(&api, name, patience).await
+    } else {
+        let api = Api::<PersistentVolumeClaim>::namespaced(client.clone(), ns);
+        let _ = api.delete(name, &del).await;
+        wait_gone(&api, name, patience).await
+    };
+    outcome.map_err(|e| anyhow!("pvc/snapshot {ns}/{name} not gone: {e}"))
+}
+
+async fn wait_gone<T>(api: &Api<T>, name: &str, patience: Duration) -> Result<()>
+where
+    T: kube::Resource + Clone + serde::de::DeserializeOwned + std::fmt::Debug,
+{
+    let mut tick = interval(Duration::from_secs(5));
+    let deadline = tokio::time::Instant::now() + patience;
+    loop {
+        tick.tick().await;
+        match api.get(name).await {
+            Err(kube::Error::Api(e)) if e.code == 404 => return Ok(()),
+            Err(e) => return Err(e.into()),
+            Ok(_) if tokio::time::Instant::now() > deadline => {
+                return Err(anyhow!("still present"));
+            }
+            Ok(_) => {}
+        }
+    }
+}

examples/harmony_tenant_backup/src/lib.rs

@@ -0,0 +1,144 @@
+//! Shared contract + engine for the two binaries.
+//! `harmony-tenant-backup` operates a namespace; `harmony-tenant-backup-setup`
+//! installs the CronJob. Names live in `common` so they cannot drift.
+
+pub mod common;
+pub mod exec;
+pub mod k8s;
+pub mod planner;
+pub mod worker;
+
+pub use common::*;
+
+use anyhow::{Context, Result};
+use harmony_config::{Config, ConfigClient, EnvSource};
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use std::path::Path;
+use std::sync::Arc;
+
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq, Config)]
+pub struct TenantBackup {
+    pub target_endpoint: String,
+    pub target_bucket: String,
+    #[config(secret)]
+    pub target_access_key: String,
+    #[config(secret)]
+    pub target_secret_key: String,
+    pub target_region: Option<String>,
+    /// Minted by setup. Nightly job reads OpenBao; fire restore uses the
+    /// offsite kit, never this vault.
+    #[serde(default)]
+    #[config(secret)]
+    pub restic_password: String,
+    /// Read-only tenant source on the site-A RGW; needed only when the
+    /// namespace declares buckets.
+    pub source_endpoint: Option<String>,
+    pub source_region: Option<String>,
+    #[config(secret)]
+    pub source_access_key: Option<String>,
+    #[config(secret)]
+    pub source_secret_key: Option<String>,
+    pub snapshot_class: Option<String>,
+    /// One restic repo per month; expiry = delete a whole prefix.
+    #[serde(default)]
+    pub rotate_repo_monthly: bool,
+    /// PEM, only when the target RGW serves a private CA cert. Never
+    /// disable TLS verification instead.
+    pub target_ca_pem: Option<String>,
+}
+
+const OPENBAO_HINT: &str = concat!(
+    "OpenBao unavailable: set OPENBAO_URL, OPENBAO_JWT_ROLE, HARMONY_SSO_URL, ",
+    "HARMONY_ZITADEL_AUDIENCE and HARMONY_ZITADEL_KEY_JSON (or HARMONY_ZITADEL_KEY_PATH)"
+);
+
+async fn openbao_store(namespace: &str) -> Result<Arc<dyn harmony_config::ConfigSource>> {
+    harmony_config::openbao_source(
+        &format!("harmony/{namespace}"),
+        None,
+        None,
+        None,
+        None,
+        None,
+        None,
+    )
+    .await
+    .context(OPENBAO_HINT)
+}
+
+/// No prompt fallback: a Job has no TTY. Store path: `harmony/<tenant-ns>`.
+pub async fn load_config(namespace: &str) -> Result<TenantBackup> {
+    let store = openbao_store(namespace).await?;
+    let client = ConfigClient::new(vec![Arc::new(EnvSource), store]);
+    client.get::<TenantBackup>().await.map_err(Into::into)
+}
+
+fn mint_restic_password() -> Result<String> {
+    use std::io::Read;
+    let mut b = [0u8; 32];
+    std::fs::File::open("/dev/urandom")
+        .context("open /dev/urandom")?
+        .read_exact(&mut b)
+        .context("read /dev/urandom")?;
+    Ok(b.iter().fold(String::with_capacity(64), |mut s, x| {
+        use std::fmt::Write;
+        let _ = write!(s, "{x:02x}");
+        s
+    }))
+}
+
+/// If OpenBao has target_* but no restic_password, mint one, persist it for
+/// the CronJob, and return it so setup can print the offsite-kit line.
+/// Already set → None (never rotate; that would brick the repo).
+pub async fn ensure_restic_password(namespace: &str) -> Result<Option<String>> {
+    let store = openbao_store(namespace).await?;
+    let client = ConfigClient::new(vec![store]);
+    let mut cfg = client
+        .get::<TenantBackup>()
+        .await
+        .context("no TenantBackup in OpenBao; bao kv put target_* first (omit restic_password)")?;
+    if !cfg.restic_password.is_empty() {
+        return Ok(None);
+    }
+    let pw = mint_restic_password()?;
+    cfg.restic_password = pw.clone();
+    client
+        .set(&cfg)
+        .await
+        .context("write restic_password to OpenBao")?;
+    Ok(Some(pw))
+}
+
+pub async fn load_runtime(
+    namespace: &str,
+    cluster_id: &str,
+    scratch: &Path,
+) -> Result<(TenantBackup, exec::Restic)> {
+    std::fs::create_dir_all(scratch).with_context(|| {
+        format!(
+            "cannot create scratch dir {} (override with --scratch / SCRATCH=)",
+            scratch.display()
+        )
+    })?;
+    let cfg = load_config(namespace).await?;
+    if cfg.restic_password.is_empty() {
+        anyhow::bail!(
+            "empty restic_password: run harmony-tenant-backup-setup (mints into OpenBao and the offsite kit)"
+        );
+    }
+    let ca_file = match cfg
+        .target_ca_pem
+        .as_deref()
+        .filter(|s| !s.trim().is_empty())
+    {
+        Some(pem) => {
+            let f = scratch.join("target-ca.crt");
+            std::fs::write(&f, pem).with_context(|| format!("writing {}", f.display()))?;
+            Some(f)
+        }
+        None => None,
+    };
+    let restic = exec::build_restic(&cfg, cluster_id, namespace, ca_file, scratch);
+    Ok((cfg, restic))
+}

examples/harmony_tenant_backup/src/main.rs

@@ -0,0 +1,114 @@
+//! Operate one tenant namespace. Not restore. Not retention. Not install
+//! (`harmony-tenant-backup-setup` does that).
+
+use anyhow::{Context, Result};
+use chrono::Utc;
+use clap::{Parser, Subcommand};
+use harmony_tenant_backup::{
+    DEFAULT_SCRATCH, ENV_CLUSTER_ID, ENV_NAMESPACE, ENV_RUN_ID, ENV_SCRATCH, default_run_id,
+    dns1123_run_id, load_runtime, planner, worker,
+};
+use std::path::Path;
+
+#[derive(Parser, Debug)]
+#[command(
+    name = "harmony-tenant-backup",
+    about = "Backup one tenant namespace. Not restore. Not install."
+)]
+struct Cli {
+    #[command(subcommand)]
+    command: Command,
+}
+
+#[derive(Subcommand, Debug)]
+enum Command {
+    /// Discover and push this namespace (pg, buckets, PVCs).
+    Run {
+        #[arg(short, long, env = ENV_NAMESPACE)]
+        namespace: String,
+        #[arg(long, env = ENV_CLUSTER_ID)]
+        cluster_id: String,
+        /// Connectivity checks and the plan, no writes.
+        #[arg(long)]
+        dry_run: bool,
+        #[arg(long, env = ENV_SCRATCH, default_value = DEFAULT_SCRATCH)]
+        scratch: String,
+        #[arg(long, env = ENV_RUN_ID)]
+        run_id: Option<String>,
+    },
+    /// Push the volume mounted at /mnt/data. Spawned by `run`, not operators.
+    #[command(hide = true)]
+    Worker {
+        /// Logical PVC name this worker pushes.
+        #[arg(long)]
+        pvc: String,
+        #[arg(short, long, env = ENV_NAMESPACE)]
+        namespace: String,
+        #[arg(long, env = ENV_CLUSTER_ID)]
+        cluster_id: String,
+        #[arg(long, env = ENV_SCRATCH, default_value = DEFAULT_SCRATCH)]
+        scratch: String,
+        #[arg(long, env = ENV_RUN_ID)]
+        run_id: Option<String>,
+    },
+}
+
+#[tokio::main]
+async fn main() -> std::process::ExitCode {
+    match dispatch().await {
+        Ok(code) => code,
+        Err(e) => {
+            eprintln!("{e:#}");
+            2
+        }
+    }
+    .into()
+}
+
+async fn dispatch() -> Result<u8> {
+    match Cli::parse().command {
+        Command::Run {
+            namespace,
+            cluster_id,
+            dry_run,
+            scratch,
+            run_id,
+        } => {
+            let run_id = resolve_run_id(run_id)?;
+            let (cfg, restic) = load_runtime(&namespace, &cluster_id, Path::new(&scratch)).await?;
+            let client = kube::Client::try_default()
+                .await
+                .context("no Kubernetes access (in-cluster ServiceAccount or KUBECONFIG)")?;
+            planner::Run::new(
+                &client,
+                &cfg,
+                &namespace,
+                &cluster_id,
+                &run_id,
+                Path::new(&scratch),
+                restic,
+                dry_run,
+            )
+            .run()
+            .await
+        }
+        Command::Worker {
+            pvc,
+            namespace,
+            cluster_id,
+            scratch,
+            run_id,
+        } => {
+            let run_id = resolve_run_id(run_id)?;
+            let (_cfg, restic) = load_runtime(&namespace, &cluster_id, Path::new(&scratch)).await?;
+            worker::run(&restic, &cluster_id, &namespace, &run_id, &pvc).await
+        }
+    }
+}
+
+fn resolve_run_id(run_id: Option<String>) -> Result<String> {
+    let run_id = run_id.unwrap_or_else(|| default_run_id(Utc::now()));
+    dns1123_run_id(&run_id).ok_or_else(|| {
+        anyhow::anyhow!("--run-id must be a DNS-1123 label, 1-32 chars [a-z0-9-]; got {run_id}")
+    })
+}

examples/harmony_tenant_backup/src/planner.rs

@@ -0,0 +1,506 @@
+//! Planner: one namespace's three surfaces — CNPG logical dumps, S3
+//! buckets (staged then restic'd), PVCs (direct or snapshot-clone worker).
+
+use crate::common::*;
+use crate::exec::{Pg, Restic, SrcS3, logline};
+use crate::k8s::{self, WorkerSpec};
+use anyhow::{Context, Result, anyhow};
+use k8s_openapi::api::core::v1::PersistentVolumeClaim;
+use kube::{Api, Client, ResourceExt};
+use std::path::Path;
+use std::time::Duration;
+
+pub struct Plan {
+    pub pg: usize,
+    pub buckets: usize,
+    pub pvc: usize,
+    pub pvc_skipped: usize,
+    pub failures: Vec<String>,
+}
+
+const SNAP_WAIT: Duration = Duration::from_secs(30 * 60);
+const CLONE_WAIT: Duration = Duration::from_secs(15 * 60);
+/// Patience = worker's own deadline + slack, so only a genuinely stuck pod
+/// times out here rather than a slow-but-live push.
+const JOB_WAIT: Duration = Duration::from_secs(WORKER_DEADLINE_SECS as u64 + 15 * 60);
+const CLEANUP_WAIT: Duration = Duration::from_secs(5 * 60);
+const DB_LIST_SQL: &str =
+    "select datname from pg_database where not datistemplate and datallowconn order by 1";
+
+/// Everything one backup run needs, threaded as one value so ns/run_id can
+/// never drift between surfaces.
+pub struct Run<'a> {
+    client: &'a Client,
+    cfg: &'a crate::TenantBackup,
+    restic: Restic,
+    ns: &'a str,
+    cluster_id: &'a str,
+    run_id: &'a str,
+    scratch: &'a Path,
+    dry_run: bool,
+    fs_group: Option<i64>,
+    plan: Plan,
+}
+
+impl<'a> Run<'a> {
+    // Whole context of a run; the restic handle is built once by main so
+    // planner and worker share one repo resolution.
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        client: &'a Client,
+        cfg: &'a crate::TenantBackup,
+        ns: &'a str,
+        cluster_id: &'a str,
+        run_id: &'a str,
+        scratch: &'a Path,
+        restic: Restic,
+        dry_run: bool,
+    ) -> Self {
+        Self {
+            client,
+            cfg,
+            restic,
+            ns,
+            cluster_id,
+            run_id,
+            scratch,
+            dry_run,
+            fs_group: None,
+            plan: Plan {
+                pg: 0,
+                buckets: 0,
+                pvc: 0,
+                pvc_skipped: 0,
+                failures: vec![],
+            },
+        }
+    }
+
+    pub async fn run(&mut self) -> Result<u8> {
+        let anns = k8s::ns_annotations(self.client, self.ns).await?;
+        self.fs_group = volume_fsgroup(Some(&anns));
+        self.restic
+            .ensure_repo(self.dry_run)
+            .await
+            .context("restic repo unavailable")?;
+        self.do_pg().await;
+        self.do_buckets().await;
+        self.do_pvcs().await;
+
+        println!(
+            "ns {}  run {}  pg {}  buckets {}  pvc {}  skipped {}  failures {}",
+            self.ns,
+            self.run_id,
+            self.plan.pg,
+            self.plan.buckets,
+            self.plan.pvc,
+            self.plan.pvc_skipped,
+            self.plan.failures.len(),
+        );
+        logline("snapshots this run:");
+        self.restic
+            .snapshots_tagged(&format!("run:{}", self.run_id))
+            .await
+            .ok();
+        if self.plan.failures.is_empty() {
+            Ok(0)
+        } else {
+            eprintln!("FAILURES:");
+            for f in &self.plan.failures {
+                eprintln!(" - {f}");
+            }
+            Ok(3)
+        }
+    }
+
+    fn fail(&mut self, msg: String) {
+        logline(&format!("FAILED: {msg}"));
+        self.plan.failures.push(msg);
+    }
+
+    fn fail_pvc(&mut self, msg: String) {
+        self.fail(msg);
+        self.plan.pvc_skipped += 1;
+    }
+
+    async fn do_pg(&mut self) {
+        let clusters =
+            match k8s::list_namespaced_names(self.client, self.ns, &k8s::cnpg_clusters()).await {
+                Ok(c) => c,
+                // The CRD may legitimately be absent (no CNPG here), but a real
+                // API error must not read as "no databases".
+                Err(e) if is_not_found(&e) => {
+                    logline("no CNPG CRD present, skipping pg surface");
+                    return;
+                }
+                Err(e) => return self.fail(format!("cannot list CNPG clusters: {e}")),
+            };
+        if clusters.is_empty() {
+            logline(&format!("no CNPG clusters in {}", self.ns));
+            return;
+        }
+
+        for c in clusters {
+            logline(&format!("pg: cluster {c}"));
+            let pg = match self.open_pg(&c).await {
+                Ok(pg) => pg,
+                Err(e) => {
+                    self.fail(format!("pg/{c}: {e}"));
+                    continue;
+                }
+            };
+            if let Err(e) = pg.query("postgres", "select 1").await {
+                self.fail(format!("cannot reach {c}-rw (NetworkPolicy? TLS?): {e}"));
+                continue;
+            }
+            if self.dry_run {
+                match pg.query("postgres", DB_LIST_SQL).await {
+                    Ok(dbs) => logline(&format!(
+                        "DRY-RUN: would dump {c}: {}",
+                        dbs.replace('\n', ", ")
+                    )),
+                    Err(e) => self.fail(format!("cannot list databases for {c}: {e}")),
+                }
+                continue;
+            }
+
+            let out = self.scratch.join("pg").join(&c);
+            tokio::fs::create_dir_all(&out).await.ok();
+            // metadata before dumps: without LSN + timestamp a dump is
+            // undateable in an incident.
+            let lsn = pg
+                .query("postgres", "select pg_current_wal_lsn()")
+                .await
+                .unwrap_or_else(|_| "unknown".into());
+            let dumped_at = pg
+                .query("postgres", "select now()")
+                .await
+                .unwrap_or_default();
+            tokio::fs::write(out.join("dumped_at"), &dumped_at)
+                .await
+                .ok();
+            let meta = serde_json::json!({
+                "cluster": c, "ns": self.ns, "run": self.run_id, "lsn": lsn.trim(),
+                "repo": self.restic.repo(),
+            });
+            tokio::fs::write(out.join("metadata.json"), meta.to_string() + "\n")
+                .await
+                .ok();
+
+            let mut cluster_ok = true;
+            if pg.dump_all_globals(&out.join("globals.sql")).await.is_err() {
+                self.fail(format!(
+                    "pg_dumpall --globals-only {c}: a DB-only dump loses the app roles"
+                ));
+                cluster_ok = false;
+            }
+
+            let dbs = match pg.query("postgres", DB_LIST_SQL).await {
+                Ok(dbs) => dbs,
+                Err(e) => {
+                    self.fail(format!("cannot list databases for {c}: {e}"));
+                    tokio::fs::remove_dir_all(&out).await.ok();
+                    continue;
+                }
+            };
+            let mut dumped = 0;
+            for db in dbs.lines().filter(|d| !d.is_empty()) {
+                let f = out.join(format!("{db}.dump"));
+                logline(&format!("pg: {c}/{db} -> {}", f.display()));
+                match pg.dump_db(db, &f).await {
+                    Ok(()) if Pg::verify_dump(&f).await.is_ok() => dumped += 1,
+                    Ok(()) => {
+                        self.fail(format!(
+                            "pg_restore --list rejected {c}/{db} (corrupt dump)"
+                        ));
+                        tokio::fs::remove_file(&f).await.ok();
+                        cluster_ok = false;
+                    }
+                    Err(e) => {
+                        self.fail(format!("pg_dump {c}/{db}: {e}"));
+                        tokio::fs::remove_file(&f).await.ok();
+                        cluster_ok = false;
+                    }
+                }
+            }
+            if cluster_ok {
+                match self
+                    .restic
+                    .backup(
+                        &restic_host(self.cluster_id, self.ns),
+                        self.run_id,
+                        self.ns,
+                        "pg",
+                        &[format!("cluster:{c}")],
+                        &out.display().to_string(),
+                        false,
+                    )
+                    .await
+                {
+                    Ok(()) => self.plan.pg += dumped,
+                    Err(e) => self.fail(e.to_string()),
+                }
+            }
+            tokio::fs::remove_dir_all(&out).await.ok();
+        }
+    }
+
+    async fn open_pg(&self, cluster: &str) -> Result<Pg> {
+        let password = String::from_utf8_lossy(
+            &k8s::secret_bytes(
+                self.client,
+                self.ns,
+                &format!("{cluster}-superuser"),
+                "password",
+            )
+            .await?,
+        )
+        .trim_end()
+        .to_string();
+        let ca_dir = self.scratch.join("tls");
+        tokio::fs::create_dir_all(&ca_dir).await.ok();
+        let ca_file = ca_dir.join(format!("{cluster}-ca.crt"));
+        let ca =
+            k8s::secret_bytes(self.client, self.ns, &format!("{cluster}-ca"), "ca.crt").await?;
+        tokio::fs::write(&ca_file, ca).await?;
+        Ok(Pg::new(cluster, self.ns, &ca_file, password))
+    }
+
+    async fn do_buckets(&mut self) {
+        let list = match k8s::annotation(self.client, self.ns, A_BUCKETS).await {
+            Ok(Some(l)) if !l.trim().is_empty() => l,
+            // Coverage gap, not a skip: say so loudly.
+            Ok(_) => {
+                logline(&format!(
+                    "no '{A_BUCKETS}' annotation on ns/{} — skipping buckets (THIS IS A COVERAGE GAP)",
+                    self.ns
+                ));
+                return;
+            }
+            Err(e) => return self.fail(format!("cannot read ns/{} annotations: {e}", self.ns)),
+        };
+        let (Some(endpoint), Some(access), Some(secret)) = (
+            &self.cfg.source_endpoint,
+            &self.cfg.source_access_key,
+            &self.cfg.source_secret_key,
+        ) else {
+            return self.fail(format!(
+                "bucket source credentials incomplete in vault for {} (source_endpoint/source_access_key/source_secret_key)",
+                self.ns
+            ));
+        };
+        let src = SrcS3 {
+            endpoint: endpoint.clone(),
+            region: self
+                .cfg
+                .source_region
+                .clone()
+                .unwrap_or_else(|| "default".into()),
+            access_key: access.clone(),
+            secret_key: secret.clone(),
+        };
+
+        for b in list.split(',').map(str::trim).filter(|b| !b.is_empty()) {
+            // Tenant-authored string reaching filesystem paths and rclone
+            // connection strings: reject anything outside S3's own grammar.
+            if !valid_bucket_token(b) {
+                self.fail(format!(
+                    "invalid bucket token {b:?} in {A_BUCKETS} (want ^[a-z0-9][a-z0-9.-]{{0,61}}[a-z0-9]$)"
+                ));
+                continue;
+            }
+            let dst = self.scratch.join("s3").join(b);
+            if self.dry_run {
+                logline(&format!("DRY-RUN: would stage and push s3 bucket {b}"));
+                if let Err(e) = src.list_dirs(b).await {
+                    self.fail(format!("bucket {b} not reachable: {e}"));
+                }
+                continue;
+            }
+            logline(&format!("s3: staging {b} -> {}", dst.display()));
+            tokio::fs::create_dir_all(&dst).await.ok();
+            let staged = match src.copy(b, &dst).await {
+                Ok(()) => src.check(b, &dst).await,
+                Err(e) => Err(e),
+            };
+            let pushed = match staged {
+                Ok(()) => {
+                    self.restic
+                        .backup(
+                            &restic_host(self.cluster_id, self.ns),
+                            self.run_id,
+                            self.ns,
+                            "s3",
+                            &[format!("bucket:{b}")],
+                            &dst.display().to_string(),
+                            false,
+                        )
+                        .await
+                }
+                Err(e) => Err(e),
+            };
+            match pushed {
+                Ok(()) => self.plan.buckets += 1,
+                Err(e) => self.fail(e.to_string()),
+            }
+            tokio::fs::remove_dir_all(&dst).await.ok();
+        }
+    }
+
+    async fn do_pvcs(&mut self) {
+        let pvcs = match Api::<PersistentVolumeClaim>::namespaced(self.client.clone(), self.ns)
+            .list(&Default::default())
+            .await
+        {
+            Ok(l) => l,
+            Err(e) => return self.fail(format!("cannot list PVCs: {e}")),
+        };
+
+        let mut class: Option<String> = self.cfg.snapshot_class.clone();
+        for pvc in pvcs {
+            let name = pvc.name_any();
+            let ann = pvc.metadata.annotations.clone().unwrap_or_default();
+            let labels = pvc.metadata.labels.clone().unwrap_or_default();
+            let spec = pvc.spec.clone().unwrap_or_default();
+            let phase = pvc
+                .status
+                .as_ref()
+                .and_then(|s| s.phase.clone())
+                .unwrap_or_else(|| "Unknown".into());
+            let access = spec
+                .access_modes
+                .as_ref()
+                .and_then(|a| a.first())
+                .cloned()
+                .unwrap_or_else(|| "?".into());
+
+            let outcome = match decide_pvc(
+                &phase,
+                labels.get("cnpg.io/cluster").map(String::as_str),
+                ann.get(A_PVC_DISABLED).map(|v| v.as_str()) == Some("true"),
+                spec.volume_mode.as_deref() == Some("Block"),
+                &access,
+            ) {
+                PvcDecision::Skip(why) => {
+                    logline(&format!("pvc/{name}: {why}, skipping"));
+                    self.plan.pvc_skipped += 1;
+                    continue;
+                }
+                PvcDecision::Fail(why) => {
+                    self.fail_pvc(format!("pvc/{name}: {why}"));
+                    continue;
+                }
+                PvcDecision::Direct => {
+                    logline(&format!("pvc/{name} ({access}): mounting directly"));
+                    if self.dry_run {
+                        logline(&format!("DRY-RUN: worker on {name}"));
+                        continue;
+                    }
+                    self.spawn_worker(&name, &name).await
+                }
+                PvcDecision::SnapshotFirst => {
+                    if class.is_none() {
+                        // A swallowed 403 here reads as "no class exists":
+                        // propagate so the RBAC gap is visible in the log.
+                        match k8s::first_cluster_scoped_name(
+                            self.client,
+                            &k8s::volume_snapshot_classes(),
+                        )
+                        .await
+                        {
+                            Ok(c) => class = c,
+                            Err(e) => {
+                                self.fail_pvc(format!("pvc/{name}: {e}"));
+                                continue;
+                            }
+                        }
+                    }
+                    let Some(cls) = class.clone() else {
+                        self.fail_pvc(format!(
+                            "pvc/{name}: no VolumeSnapshotClass found (set snapshot_class in vault)"
+                        ));
+                        continue;
+                    };
+                    if self.dry_run {
+                        logline(&format!(
+                            "DRY-RUN: snapshot + clone via {cls} + worker for {name}"
+                        ));
+                        continue;
+                    }
+                    self.snapshot_worker(&pvc, &name, &cls).await
+                }
+            };
+            match outcome {
+                Ok(()) => self.plan.pvc += 1,
+                Err(e) => self.fail_pvc(format!("pvc/{name} ({access}): {e}")),
+            }
+        }
+    }
+
+    /// Snapshot -> clone -> worker push -> cleanup. The clone must be deleted
+    /// before the snapshot it references.
+    async fn snapshot_worker(
+        &mut self,
+        source: &PersistentVolumeClaim,
+        logical: &str,
+        class: &str,
+    ) -> Result<()> {
+        let snap = run_object_name("htb-snap", logical, self.run_id);
+        let clone = run_object_name("htb-clone", logical, self.run_id);
+        k8s::create_volume_snapshot(self.client, self.ns, &snap, class, logical, self.run_id)
+            .await?;
+        let outcome = async {
+            k8s::wait_snapshot_ready(self.client, self.ns, &snap, SNAP_WAIT).await?;
+            k8s::create_clone_pvc(self.client, self.ns, &clone, source, &snap, self.run_id).await?;
+            k8s::wait_pvc_bound(self.client, self.ns, &clone, CLONE_WAIT).await?;
+            self.spawn_worker(&clone, logical).await
+        }
+        .await;
+        // Left-over snapshots and clones run into CSI quotas and bill; a
+        // failed cleanup is a run failure, not a shrug.
+        for res in [
+            k8s::delete_gone(self.client, self.ns, &clone, false, CLEANUP_WAIT).await,
+            k8s::delete_gone(self.client, self.ns, &snap, true, CLEANUP_WAIT).await,
+        ] {
+            if let Err(e) = res {
+                self.fail(format!("pvc/{logical}: cleanup: {e}"));
+            }
+        }
+        outcome
+    }
+
+    async fn spawn_worker(&self, pvc_mount: &str, logical: &str) -> Result<()> {
+        let image = std::env::var(ENV_WORKER_IMAGE)
+            .context(format!("${ENV_WORKER_IMAGE} required to back up PVCs"))?;
+        let fwd: Vec<(String, String)> = FORWARD_ENV
+            .iter()
+            .filter_map(|k| std::env::var(k).ok().map(|v| ((*k).to_string(), v)))
+            .collect();
+        if fwd.len() != FORWARD_ENV.len() {
+            return Err(anyhow!(
+                "need all of {} in the planner env to spawn a PVC worker",
+                FORWARD_ENV.join(", ")
+            ));
+        }
+        let fwd_refs: Vec<(&str, &str)> =
+            fwd.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect();
+        let job = k8s::worker_job(&WorkerSpec {
+            image: &image,
+            ns: self.ns,
+            name: &run_object_name("htb-wk", logical, self.run_id),
+            pvc_mount,
+            pvc_logical: logical,
+            run_id: self.run_id,
+            cluster_id: self.cluster_id,
+            zitadel_secret: &std::env::var(ENV_ZITADEL_SECRET)
+                .unwrap_or_else(|_| ZITADEL_SECRET.into()),
+            forward_env: fwd_refs,
+            fs_group: self.fs_group,
+        });
+        k8s::spawn_and_wait_job(self.client, self.ns, &job, JOB_WAIT).await
+    }
+}
+
+fn is_not_found(e: &anyhow::Error) -> bool {
+    matches!(e.downcast_ref::<kube::Error>(), Some(kube::Error::Api(a)) if a.code == 404)
+}

examples/harmony_tenant_backup/src/setup.rs

@@ -0,0 +1,410 @@
+//! Install RBAC + CronJob into a tenant namespace. Not the nightly run.
+
+use anyhow::{Context, Result, anyhow};
+use clap::Parser;
+use harmony::inventory::Inventory;
+use harmony::modules::k8s::K8sBareTopology;
+use harmony::modules::k8s::resource::K8sResourceScore;
+use harmony::score::Score;
+use harmony_tenant_backup::k8s::{container_security, pod_security};
+use harmony_tenant_backup::{env_from_secret, env_var, *};
+use k8s_openapi::api::batch::v1::{CronJob, CronJobSpec, JobSpec, JobTemplateSpec};
+use k8s_openapi::api::core::v1::{
+    Container, EmptyDirVolumeSource, PodSpec, PodTemplateSpec, ResourceRequirements,
+    ServiceAccount, Volume, VolumeMount,
+};
+use k8s_openapi::api::rbac::v1::{
+    ClusterRole, ClusterRoleBinding, PolicyRule, Role, RoleBinding, RoleRef, Subject,
+};
+use k8s_openapi::apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::ObjectMeta};
+use kube::{Api, Client};
+use std::collections::BTreeMap;
+
+#[derive(Parser, Debug)]
+#[command(
+    name = "harmony-tenant-backup-setup",
+    about = "Install backup RBAC + CronJob into a tenant namespace. Not the nightly run."
+)]
+struct Cli {
+    /// Tenant namespace to install into.
+    #[arg(short, long)]
+    namespace: String,
+    /// Cron schedule, e.g. "17 3 * * *" (stagger per namespace, off-peak).
+    #[arg(long)]
+    schedule: String,
+    /// Backup image. MUST be digest-pinned (harbor.example/ns-backup@sha256:...).
+    #[arg(long, env = "BACKUP_IMAGE")]
+    image: String,
+    /// Stable cluster identifier used in the restic repo path and --host.
+    #[arg(long, env = ENV_CLUSTER_ID)]
+    cluster_id: String,
+    #[arg(long, default_value = ZITADEL_SECRET)]
+    zitadel_secret: String,
+    #[arg(long, env = "OPENBAO_URL")]
+    openbao_url: Option<String>,
+    #[arg(long, env = "OPENBAO_JWT_ROLE")]
+    openbao_jwt_role: Option<String>,
+    #[arg(long, env = "HARMONY_SSO_URL")]
+    sso_url: Option<String>,
+    #[arg(long, env = "HARMONY_ZITADEL_AUDIENCE")]
+    audience: Option<String>,
+}
+
+const KEY_FILE: &str = ZITADEL_KEY_FILE;
+
+macro_rules! apply {
+    ($topo:expr, $ns:expr, $obj:expr, $what:expr) => {{
+        let score = K8sResourceScore::single($obj, Some($ns.to_string()));
+        let outcome = Score::<K8sBareTopology>::create_interpret(&score)
+            .execute(&Inventory::empty(), $topo)
+            .await
+            .map_err(|e| anyhow!("apply {}: {e}", $what))?;
+        println!("{:<12} {}: {:?}", $what, $ns, outcome.status);
+    }};
+}
+
+macro_rules! apply_cluster {
+    ($topo:expr, $obj:expr, $what:expr) => {{
+        let score = K8sResourceScore::single($obj, None);
+        let outcome = Score::<K8sBareTopology>::create_interpret(&score)
+            .execute(&Inventory::empty(), $topo)
+            .await
+            .map_err(|e| anyhow!("apply {}: {e}", $what))?;
+        println!("{:<12} cluster: {:?}", $what, outcome.status);
+    }};
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let cli = Cli::parse();
+    if !cli.image.contains("@sha256:") {
+        return Err(anyhow!(
+            "--image must be digest-pinned; tag-following is exactly the \
+             \"worked last week\" bug class this tool exists to avoid"
+        ));
+    }
+    for (name, val) in [
+        ("OPENBAO_URL", &cli.openbao_url),
+        ("OPENBAO_JWT_ROLE", &cli.openbao_jwt_role),
+        ("HARMONY_SSO_URL", &cli.sso_url),
+        ("HARMONY_ZITADEL_AUDIENCE", &cli.audience),
+    ] {
+        if val.is_none() {
+            return Err(anyhow!(
+                "{name} is required (flag or env): the backup workload reaches its vault through it"
+            ));
+        }
+    }
+    if std::env::var("HARMONY_ZITADEL_KEY_JSON").is_err()
+        && std::env::var("HARMONY_ZITADEL_KEY_PATH").is_err()
+    {
+        return Err(anyhow!(
+            "HARMONY_ZITADEL_KEY_JSON or HARMONY_ZITADEL_KEY_PATH required: \
+             setup mints restic_password into OpenBao for the CronJob"
+        ));
+    }
+
+    let client = Client::try_default()
+        .await
+        .context("no kubeconfig / cluster access")?;
+    let ns_obj = Api::<k8s_openapi::api::core::v1::Namespace>::all(client.clone())
+        .get(&cli.namespace)
+        .await
+        .map_err(|e| {
+            if matches!(e, kube::Error::Api(ref a) if a.code == 404) {
+                anyhow!("namespace {} does not exist", cli.namespace)
+            } else {
+                anyhow!(
+                    "cannot reach the cluster to check namespace {}: {e}",
+                    cli.namespace
+                )
+            }
+        })?;
+    let fs_group = volume_fsgroup(ns_obj.metadata.annotations.as_ref());
+    let secret = Api::<k8s_openapi::api::core::v1::Secret>::namespaced(client, &cli.namespace)
+        .get(&cli.zitadel_secret)
+        .await
+        .with_context(|| {
+            format!(
+                "secret {}/{} not found. Create it with:\n  \
+                 kubectl create secret generic {} -n {} --from-file={KEY_FILE}=<zitadel machine key file>",
+                cli.namespace, cli.zitadel_secret, cli.zitadel_secret, cli.namespace
+            )
+        })?;
+    if secret
+        .data
+        .as_ref()
+        .is_none_or(|d| !d.contains_key(KEY_FILE))
+    {
+        return Err(anyhow!(
+            "secret {}/{} must contain key '{KEY_FILE}'",
+            cli.namespace,
+            cli.zitadel_secret
+        ));
+    }
+
+    let topology = K8sBareTopology::from_kubeconfig("tenant-backup-setup")
+        .await
+        .map_err(|e| anyhow!(e))?;
+    let ns = &cli.namespace;
+    apply!(&topology, ns, service_account(), "ServiceAccount");
+    apply!(&topology, ns, role(), "Role");
+    apply!(&topology, ns, role_binding(ns), "RoleBinding");
+    apply_cluster!(&topology, cluster_role(ns), "ClusterRole");
+    apply_cluster!(&topology, cluster_role_binding(ns), "ClusterRoleBinding");
+    apply!(&topology, ns, cron_job(&cli, fs_group), "CronJob");
+
+    match ensure_restic_password(ns).await? {
+        Some(pw) => {
+            println!(
+                "\nOFFSITE KIT (KeepassX). Break-glass. Restore never uses OpenBao or this cluster."
+            );
+            println!("  ns={ns}  restic_password={pw}");
+            println!(
+                "  Put this in the same kit as OpenBao unseal keys. You will not see it again."
+            );
+        }
+        None => println!(
+            "\nrestic_password already in OpenBao; the offsite kit must already have it (not rotated)."
+        ),
+    }
+    println!(
+        "installed: CronJob {CRONJOB_NAME} in {ns}, schedule {:?}, image {}",
+        cli.schedule, cli.image
+    );
+    println!("run now:   kubectl create job backup-now --from=cronjob/{CRONJOB_NAME} -n {ns}");
+    println!("watch:     kubectl logs -n {ns} job/backup-now -f");
+    Ok(())
+}
+
+fn managed_labels() -> BTreeMap<String, String> {
+    BTreeMap::from([(MANAGED_BY.to_string(), MANAGED_BY_VALUE.to_string())])
+}
+
+fn service_account() -> ServiceAccount {
+    ServiceAccount {
+        metadata: ObjectMeta {
+            name: Some(SA_NAME.into()),
+            labels: Some(managed_labels()),
+            ..Default::default()
+        },
+        ..Default::default()
+    }
+}
+
+fn role() -> Role {
+    Role {
+        metadata: ObjectMeta {
+            name: Some(SA_NAME.into()),
+            labels: Some(managed_labels()),
+            ..Default::default()
+        },
+        rules: Some(vec![
+            PolicyRule {
+                // CNPG <cluster>-superuser password + <cluster>-ca cert, both
+                // by known name. get-only: a backup identity must not sweep
+                // the tenant's secrets.
+                api_groups: Some(vec!["".into()]),
+                resources: Some(vec!["secrets".into()]),
+                verbs: vec!["get".into()],
+                ..Default::default()
+            },
+            PolicyRule {
+                api_groups: Some(vec!["".into()]),
+                resources: Some(vec!["persistentvolumeclaims".into()]),
+                verbs: vec!["get".into(), "list".into()],
+                ..Default::default()
+            },
+            PolicyRule {
+                // The RWO clone only; delete is scoped here, never on secrets.
+                api_groups: Some(vec!["".into()]),
+                resources: Some(vec!["persistentvolumeclaims".into()]),
+                verbs: vec!["create".into(), "delete".into()],
+                ..Default::default()
+            },
+            PolicyRule {
+                // VolumeSnapshotClass is cluster-scoped: a namespaced Role
+                // can never grant it (see the ClusterRole below).
+                api_groups: Some(vec!["snapshot.storage.k8s.io".into()]),
+                resources: Some(vec!["volumesnapshots".into()]),
+                verbs: vec![
+                    "get".into(),
+                    "list".into(),
+                    "create".into(),
+                    "delete".into(),
+                ],
+                ..Default::default()
+            },
+            PolicyRule {
+                api_groups: Some(vec!["batch".into()]),
+                resources: Some(vec!["jobs".into()]),
+                verbs: vec!["get".into(), "list".into(), "create".into()],
+                ..Default::default()
+            },
+            PolicyRule {
+                api_groups: Some(vec!["postgresql.cnpg.io".into()]),
+                resources: Some(vec!["clusters".into()]),
+                verbs: vec!["get".into(), "list".into()],
+                ..Default::default()
+            },
+        ]),
+    }
+}
+
+/// Cluster-scoped reads a namespaced Role can never grant: this
+/// namespace (bucket annotation) and VolumeSnapshotClass list.
+fn cluster_role(ns: &str) -> ClusterRole {
+    ClusterRole {
+        metadata: ObjectMeta {
+            name: Some(format!("{SA_NAME}-{ns}")),
+            labels: Some(managed_labels()),
+            ..Default::default()
+        },
+        rules: Some(vec![
+            PolicyRule {
+                api_groups: Some(vec!["".into()]),
+                resources: Some(vec!["namespaces".into()]),
+                resource_names: Some(vec![ns.into()]),
+                verbs: vec!["get".into()],
+                ..Default::default()
+            },
+            PolicyRule {
+                api_groups: Some(vec!["snapshot.storage.k8s.io".into()]),
+                resources: Some(vec!["volumesnapshotclasses".into()]),
+                verbs: vec!["get".into(), "list".into()],
+                ..Default::default()
+            },
+        ]),
+        ..Default::default()
+    }
+}
+
+fn cluster_role_binding(ns: &str) -> ClusterRoleBinding {
+    ClusterRoleBinding {
+        metadata: ObjectMeta {
+            name: Some(format!("{SA_NAME}-{ns}")),
+            labels: Some(managed_labels()),
+            ..Default::default()
+        },
+        role_ref: RoleRef {
+            api_group: "rbac.authorization.k8s.io".into(),
+            kind: "ClusterRole".into(),
+            name: format!("{SA_NAME}-{ns}"),
+        },
+        subjects: Some(vec![Subject {
+            api_group: Some("".into()),
+            kind: "ServiceAccount".into(),
+            name: SA_NAME.into(),
+            namespace: Some(ns.into()),
+        }]),
+    }
+}
+
+fn role_binding(ns: &str) -> RoleBinding {
+    RoleBinding {
+        metadata: ObjectMeta {
+            name: Some(SA_NAME.into()),
+            labels: Some(managed_labels()),
+            ..Default::default()
+        },
+        role_ref: RoleRef {
+            api_group: "rbac.authorization.k8s.io".into(),
+            kind: "Role".into(),
+            name: SA_NAME.into(),
+        },
+        subjects: Some(vec![Subject {
+            api_group: Some("".into()),
+            kind: "ServiceAccount".into(),
+            name: SA_NAME.into(),
+            namespace: Some(ns.into()),
+        }]),
+    }
+}
+
+fn cron_job(cli: &Cli, fs_group: Option<i64>) -> CronJob {
+    let mut env = vec![
+        env_var(ENV_NAMESPACE, &cli.namespace),
+        env_var(ENV_CLUSTER_ID, &cli.cluster_id),
+        env_var(ENV_WORKER_IMAGE, &cli.image),
+        env_var(ENV_ZITADEL_SECRET, &cli.zitadel_secret),
+        env_from_secret(ENV_ZITADEL_KEY_JSON, &cli.zitadel_secret, KEY_FILE),
+    ];
+    for (k, v) in [
+        ("OPENBAO_URL", &cli.openbao_url),
+        ("OPENBAO_JWT_ROLE", &cli.openbao_jwt_role),
+        ("HARMONY_SSO_URL", &cli.sso_url),
+        ("HARMONY_ZITADEL_AUDIENCE", &cli.audience),
+    ] {
+        if let Some(v) = v {
+            env.push(env_var(k, v));
+        }
+    }
+
+    CronJob {
+        metadata: ObjectMeta {
+            name: Some(CRONJOB_NAME.into()),
+            labels: Some(managed_labels()),
+            ..Default::default()
+        },
+        spec: Some(CronJobSpec {
+            schedule: cli.schedule.clone(),
+            concurrency_policy: Some("Forbid".into()),
+            successful_jobs_history_limit: Some(3),
+            failed_jobs_history_limit: Some(3),
+            job_template: JobTemplateSpec {
+                metadata: Some(ObjectMeta {
+                    labels: Some(managed_labels()),
+                    ..Default::default()
+                }),
+                spec: Some(JobSpec {
+                    backoff_limit: Some(0),
+                    active_deadline_seconds: Some(PLANNER_DEADLINE_SECS),
+                    template: PodTemplateSpec {
+                        spec: Some(PodSpec {
+                            restart_policy: Some("Never".into()),
+                            service_account_name: Some(SA_NAME.into()),
+                            security_context: Some(pod_security(fs_group)),
+                            containers: vec![Container {
+                                name: "planner".into(),
+                                image: Some(cli.image.clone()),
+                                args: Some(vec!["run".into()]),
+                                env: Some(env),
+                                volume_mounts: Some(vec![VolumeMount {
+                                    name: "scratch".into(),
+                                    mount_path: DEFAULT_SCRATCH.into(),
+                                    ..Default::default()
+                                }]),
+                                resources: Some(ResourceRequirements {
+                                    requests: Some(BTreeMap::from([
+                                        ("cpu".into(), Quantity("200m".into())),
+                                        ("memory".into(), Quantity("512Mi".into())),
+                                    ])),
+                                    limits: Some(BTreeMap::from([(
+                                        "memory".into(),
+                                        Quantity("2Gi".into()),
+                                    )])),
+                                    ..Default::default()
+                                }),
+                                security_context: Some(container_security()),
+                                ..Default::default()
+                            }],
+                            volumes: Some(vec![Volume {
+                                name: "scratch".into(),
+                                empty_dir: Some(EmptyDirVolumeSource {
+                                    size_limit: Some(Quantity("5Gi".into())),
+                                    ..Default::default()
+                                }),
+                                ..Default::default()
+                            }]),
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    },
+                    ..Default::default()
+                }),
+            },
+            ..Default::default()
+        }),
+        status: None,
+    }
+}

examples/harmony_tenant_backup/src/worker.rs

@@ -0,0 +1,39 @@
+//! Worker role: push one mounted, read-only volume into the restic repo.
+//! Runs as a Job spawned by the planner, with the clone (or RWX PVC)
+//! mounted at WORK_MOUNT and the OpenBao coordinates forwarded by the
+//! planner.
+
+use crate::common::{WORK_MOUNT, restic_host};
+use crate::exec::{Restic, logline};
+use anyhow::{Context, Result, anyhow};
+
+pub async fn run(
+    restic: &Restic,
+    cluster_id: &str,
+    ns: &str,
+    run_id: &str,
+    logical_pvc: &str,
+) -> Result<u8> {
+    if !tokio::fs::try_exists(WORK_MOUNT).await.unwrap_or(false) {
+        return Err(anyhow!("worker: {WORK_MOUNT} not mounted"));
+    }
+    restic
+        .ensure_repo(false)
+        .await
+        .context("restic repo unavailable")?;
+    logline(&format!(
+        "worker: pushing {WORK_MOUNT} as pvc/{logical_pvc}"
+    ));
+    restic
+        .backup(
+            &restic_host(cluster_id, ns),
+            run_id,
+            ns,
+            "pvc",
+            &[format!("pvc:{logical_pvc}")],
+            WORK_MOUNT,
+            false,
+        )
+        .await?;
+    Ok(0)
+}