Compare commits

...
9 Commits
23 changed files with 976 additions and 59 deletions
+1
View File
@@ -7,6 +7,7 @@ RUN pip wheel --no-cache-dir --wheel-dir /wheels .
FROM builder AS test
RUN pip install --no-cache-dir /wheels/*.whl
COPY tests ./tests
COPY scripts ./scripts
CMD ["python", "-m", "unittest", "discover", "-s", "tests", "-v"]
FROM python:3.11-slim
+7
View File
@@ -77,6 +77,13 @@ four eager-mesh routes, transfer/eviction/unarchive workflows, and the
adversarial matrix without live infrastructure or Telegram. See
`e2e/README.md`.
## Deployment Preflight Tests
Check deployment configuration immediately before bringing a new node online.
The reusable read-only preflight validates the deployed image's filesystem and
API view against its Docker bind mounts. See details in
[docs/deployment-preflight.md](docs/deployment-preflight.md).
The complete cross-project design, protocol, workflow, safety, deployment,
Telegram UX, and testing documentation is published under [`docs/`](docs/).
+28 -11
View File
@@ -1,12 +1,15 @@
# Production deployment
These files are the non-secret, host-specific deployment manifests for the
initial x1/x2/lithium topology.
initial x1/x2/lithium topology and the later helium archive node.
- Install the x1 and x2 files as
`~/compose/ArchiveControl-cache/{compose.yaml,client.toml}`.
- Install the lithium files as
`~/compose/ArchiveControl-archive/{compose.yaml,client.toml}`.
- The helium directory contains three isolated compose-project manifests. Install
them under `~/Repositories/compose/{qbittorrent-helium,syncthing-helium,ArchiveControl-archive}`
as described in `helium/README.md`.
- Create sibling `state`, `backups`, and `secrets` directories owned by the
configured container UID/GID.
- Secret files are never committed. Each `secrets` directory contains
@@ -22,10 +25,17 @@ have different mount IDs and `link(2)` may return `EXDEV` across them. The
client deliberately treats that case as copy-only and performs a full payload
free-space check.
When a Syncthing route is physically nested below the qB root, mount the qB
root once and map the exact Syncthing API folder through it:
Automatic routes always use the Syncthing API path `routes/<route-id>`. Bind
that path from a dedicated directory beneath the qB data root, then map the
same path through the qB client mount:
```yaml
# Syncthing compose project
volumes:
- /srv/syncthing-config:/var/syncthing
- /srv/downloads/.archive-control-routes:/var/syncthing/routes
# Archive Control client compose project
volumes:
- /srv/downloads:/data/qb
- /srv/syncthing-config:/data/sync
@@ -35,23 +45,30 @@ volumes:
[syncthing]
api_root = "/var/syncthing"
local_root = "/data/sync"
local_path_overrides = { "/var/syncthing/Downloads/Sync" = "/data/qb/Sync" }
local_path_overrides = { "/var/syncthing/routes" = "/data/qb/.archive-control-routes" }
```
Do **not** additionally mount `/srv/downloads/Sync` at a path beneath
`/data/sync`. The override is the authoritative mapping for that folder and
keeps qB source files and staging destinations in one mount namespace. Use
the folder's normalized API-visible **path** as the override key (for example,
Syncthing `~/Downloads/Sync` becomes `/var/syncthing/Downloads/Sync`); do not
use the folder ID.
Do **not** mount the route directory separately into the client. The override
is the authoritative mapping and keeps qB source files and automatic route
folders in one mount namespace. Existing manually configured Syncthing folders
below the qB tree may retain their own exact overrides. Use API-visible paths,
not Syncthing folder IDs, as override keys.
Before starting a stack, validate it with:
Before starting a stack, validate its config and then run the generic host
preflight with that machine's own paths and container names:
```sh
docker compose config
docker compose run --rm archive-client --check-config
python3 scripts/preflight-deployment.py --help
```
The check is fail-fast and performs local permission, filesystem, sparse-file,
hard-link, and reflink probes. Normal startup additionally probes the local
qBittorrent and Syncthing APIs before registration.
`scripts/preflight-deployment.py` is deliberately topology-agnostic: pass the
host's `client.toml` and its client, qBittorrent, and Syncthing container names.
It resolves Docker mounts rather than assuming project names or host paths. It
also catches a common fatal error: a Syncthing `api_root` that names its config
volume rather than the mounted shared-data tree.
+34
View File
@@ -0,0 +1,34 @@
# Helium archive-node deployment
This directory contains the non-secret manifests for the dedicated helium
archive node. Install the files into these separate compose projects:
- `~/Repositories/compose/qbittorrent-helium/compose.yaml`
- `~/Repositories/compose/syncthing-helium/compose.yaml`
- `~/Repositories/compose/ArchiveControl-archive/{compose.yaml,client.toml}`
The qBittorrent API and Syncthing GUI/API bind only to loopback. Syncthing
transport/discovery ports are published normally. qBittorrent data and
Syncthing route folders both live under `/media/Data2`; the Archive Control
container therefore mounts `/media/Data2` once at `/data/storage` so archive
placements can hardlink rather than make a full copy.
`/media/Data2/Downloading` already contains unrelated files. The fresh qB
instance must not import or manage them; only Archive Control-created torrents
are managed.
Before the first client start, run the repository preflight on the target host
(replace container names if that host uses different compose project names):
```sh
python3 scripts/preflight-deployment.py \
--client-config ~/Repositories/compose/ArchiveControl-archive/client.toml \
--client-container archive-control-helium-archive-client-1 \
--syncthing-container helium-syncthing-syncthing-1 \
--qbittorrent-container helium-qbittorrent-qbittorrent-1
```
It is read-only: it checks the configured API roots resolve to the same host
paths as the client roots, that both data roots share one client bind mount for
hardlinks, secret-file permissions, filesystem capabilities, and the local qB
and Syncthing APIs. It does not send the control token or modify any container.
@@ -0,0 +1,19 @@
name: archive-control-helium
services:
archive-client:
image: sodium/archive-clients:v0.1.19
user: "1000:1000"
restart: unless-stopped
network_mode: host
command: ["--config", "/etc/archive-control/client.toml"]
environment:
QB_USER: admin
volumes:
- ./client.toml:/etc/archive-control/client.toml:ro
- ./secrets:/run/secrets:ro
- ./state:/var/lib/archive-control
- ./backups:/var/backups/archive-control
# Do not split this into separate qB and Sync bind mounts: hardlinks
# require both trees to resolve within this one mount namespace.
- /media/Data2:/data/storage
+42
View File
@@ -0,0 +1,42 @@
client_id = "helium-archive"
display_name = "helium archive"
role = "archive"
control_endpoint = "ws://bot.everdream.xyz:8766/archive_control"
shared_token_file = "/run/secrets/archive_control_token"
state_db = "/var/lib/archive-control/client.db"
backup_dir = "/var/backups/archive-control"
[connection]
registration_timeout = "10s"
heartbeat_interval = "15s"
offline_timeout = "45s"
reconnect_initial = "1s"
reconnect_max = "60s"
reconnect_reset_after = "60s"
reconnect_jitter = true
[jobs]
stall_after = "30m"
verification_timeout = "30m"
poll_interval = "1s"
free_space_reserve_bytes = 33554432
[backup]
interval = "6h"
recent = 12
daily = 14
weekly = 8
[qbittorrent]
endpoint = "http://127.0.0.1:8081"
username = "${QB_USER}"
password_file = "/run/secrets/qb_password"
api_root = "/downloads/Downloading"
local_root = "/data/storage/Downloading"
[syncthing]
endpoint = "http://127.0.0.1:8384"
api_key_file = "/run/secrets/syncthing_api_key"
api_root = "/var/syncthing/Sync"
local_root = "/data/storage/Sync"
advertised_addresses = ["dynamic"]
@@ -0,0 +1,24 @@
name: helium-qbittorrent
services:
qbittorrent:
image: sodium/qbittorrent-nox:5.2.3-lt2-1-proxy-listen-amd64
environment:
PUID: "1000"
PGID: "1000"
TZ: Etc/UTC
WEBUI_PORT: "8081"
TORRENTING_PORT: "6881"
command: ["--webui-port=8081", "--confirm-legal-notice"]
volumes:
- ./config:/config
# This is deliberately one parent mount. Archive Control maps both the
# qB and Syncthing subtrees through the same mount namespace.
- /media/Data2:/downloads
ports:
# Keep the published and internal WebUI ports identical: qBittorrent
# rejects a forwarded request whose Host header carries a different port.
- "127.0.0.1:8081:8081"
- "6881:6881/tcp"
- "6881:6881/udp"
restart: unless-stopped
+18
View File
@@ -0,0 +1,18 @@
#!/bin/sh
set -eu
home=/var/syncthing
api_key=$(cat /run/secrets/syncthing_api_key)
if [ ! -f "$home/config.xml" ]; then
syncthing generate --home="$home" --no-port-probing
fi
exec syncthing serve \
--home="$home" \
--gui-address=http://0.0.0.0:8384 \
--gui-apikey="$api_key" \
--no-browser \
--no-port-probing \
--no-restart \
--no-upgrade
@@ -0,0 +1,19 @@
name: helium-syncthing
services:
syncthing:
image: syncthing/syncthing:2.1.2@sha256:4464f4161dd0251e20d46bb3aec83363db75d80cef1abdd5d5fd4054b04a004d
user: "1000:1000"
hostname: helium
entrypoint: ["/bootstrap/start-syncthing.sh"]
volumes:
- ./start-syncthing.sh:/bootstrap/start-syncthing.sh:ro
- ./secrets:/run/secrets:ro
- ./config:/var/syncthing
- /media/Data2/Sync:/var/syncthing/Sync
ports:
- "127.0.0.1:8384:8384"
- "22000:22000/tcp"
- "22000:22000/udp"
- "21027:21027/udp"
restart: unless-stopped
+1
View File
@@ -39,4 +39,5 @@ endpoint = "http://syncthing:8384"
api_key_file = "/run/secrets/syncthing_api_key"
api_root = "/var/syncthing"
local_root = "/data/sync"
local_path_overrides = { "/var/syncthing/routes" = "/data/qb/.archive-control-routes" }
advertised_addresses = ["dynamic"]
+1 -1
View File
@@ -41,5 +41,5 @@ api_root = "/var/syncthing"
local_root = "/data/sync"
# This existing folder is physically inside the qB data tree. Map it through
# that same client bind mount so source staging can use hardlinks.
local_path_overrides = { "/var/syncthing/DownloadsSync" = "/data/qb/Sync" }
local_path_overrides = { "/var/syncthing/DownloadsSync" = "/data/qb/Sync", "/var/syncthing/routes" = "/data/qb/.archive-control-routes" }
advertised_addresses = ["dynamic"]
+1 -1
View File
@@ -42,5 +42,5 @@ local_root = "/data/sync"
# `DownloadsSync-X2` is ~/Downloads/Sync on the host, nested below the qB
# root. Resolve it through /data/qb rather than a second nested bind mount so
# source staging can hardlink it.
local_path_overrides = { "/var/syncthing/Downloads/Sync" = "/data/qb/Sync" }
local_path_overrides = { "/var/syncthing/Downloads/Sync" = "/data/qb/Sync", "/var/syncthing/routes" = "/data/qb/.archive-control-routes" }
advertised_addresses = ["dynamic"]
+3 -2
View File
@@ -16,8 +16,9 @@ credentials and host-specific details.
6. [Storage safety and recovery](storage-safety.md)
7. [qBittorrent and Syncthing integration](service-apis.md)
8. [Configuration, deployment, and usage](deployment-and-usage.md)
9. [Telegram UX](telegram-ux.md)
10. [Testing strategy](testing.md)
9. [Deployment preflight tests](deployment-preflight.md)
10. [Telegram UX](telegram-ux.md)
11. [Testing strategy](testing.md)
The independent protobuf source of truth lives in the
`cabbage/archive-control-proto` repository. Generated bindings in this
+1
View File
@@ -95,6 +95,7 @@ backup_dir = "/var/backups/archive-control"
registration_timeout = "10s" # First response must arrive within this window.
heartbeat_interval = "15s" # Server may negotiate a different effective value.
offline_timeout = "45s"
outbound_enqueue_timeout = "15s" # Reconnect rather than wedging if sends stop draining.
reconnect_initial = "1s"
reconnect_max = "60s" # Retry forever, never wait longer than this.
reconnect_reset_after = "60s"
+121
View File
@@ -0,0 +1,121 @@
# Deployment preflight tests
Run the deployment preflight on every new node after its qBittorrent and
Syncthing stacks are running, but before starting the Archive Control daemon
for normal operation or creating any routes/jobs. It does not contact the
control daemon, alter Syncthing/qBittorrent state, or print secret contents.
It does create and remove two unique, zero-byte probe files while proving that
the live client mount topology permits hard-link staging; no resource data or
configuration is modified.
The script is [`scripts/preflight-deployment.py`](../scripts/preflight-deployment.py).
It deliberately takes paths and container names as arguments rather than
assuming hostnames, compose project names, mount paths, credentials, or roles.
It can therefore be used unchanged on cache and archive nodes.
## Requirements
- Run it on the Docker host being checked.
- Use Python 3.8 or newer. The script reads the mounted configuration through
the client image, so it does not depend on the host Python TOML library.
- Have Docker CLI access to the daemon.
- Have the checkout containing the script available on that host, or copy only
this script into the node's compose directory.
- Bring up qBittorrent and Syncthing first. Do not start the normal Archive
Control daemon yet.
The disposable client container below uses the exact `archive-client` service
image, mounts, environment, and network specified by that node's compose file.
Consequently its qB/Syncthing API probes validate the same runtime environment
the daemon will use, not the host shell's network namespace.
## New-node procedure
The example uses generic names. Replace paths, compose project directories,
service names, and real container names for the node being deployed.
```sh
# 1. Bring up only the local dependencies.
cd /srv/compose/syncthing-node && docker compose up -d
cd /srv/compose/qbittorrent-node && docker compose up -d
# 2. Create a temporary client from the exact production image/config, without
# running the daemon or registering it with control.
cd /srv/compose/ArchiveControl-archive
docker compose run -d --no-deps --name archive-control-preflight \
--entrypoint sleep archive-client infinity
# 3. Discover the real dependency container names if needed.
docker ps --format '{{.Names}}'
# 4. Run the deployment checks. They include a disposable hard-link probe.
python3 /path/to/archive-clients/scripts/preflight-deployment.py \
--client-config /srv/compose/ArchiveControl-archive/client.toml \
--client-container archive-control-preflight \
--syncthing-container syncthing-node-syncthing-1 \
--qbittorrent-container qbittorrent-node-qbittorrent-1
# 5. Remove the temporary container, then start the real daemon only after a
# successful result.
docker rm -f archive-control-preflight
docker compose up -d archive-client
```
If the real client configuration is mounted somewhere other than the standard
`/etc/archive-control/client.toml`, pass that in-container location explicitly:
```sh
python3 scripts/preflight-deployment.py ... \
--container-config /custom/path/client.toml
```
## Required arguments
| Argument | Meaning |
| --- | --- |
| `--client-config` | Host path to the exact `client.toml` that will be mounted into the daemon. |
| `--client-container` | Running disposable or already-running client container to inspect and probe. |
| `--syncthing-container` | Running Syncthing container for this node. |
| `--qbittorrent-container` | Running qBittorrent container for this node. |
| `--container-config` | Optional `client.toml` path inside the client container; defaults to `/etc/archive-control/client.toml`. |
## What it checks
- The future automatic `routes/<route-id>` path resolves through Docker mounts
to the same host path as its client-side mapping.
- That future route path and qBittorrent content root use one client bind
mount, so hard-link staging remains possible rather than silently falling
back to a space-consuming copy.
- A real `link(2)` operation between a unique zero-byte file in the qB root
and one in the future automatic-route root. The probe verifies that both
names refer to the same inode and removes them unconditionally.
- Every existing `archive-control:*` Syncthing folder beneath `routes/` still
has its local directory. This catches a route-root bind-mount migration that
would otherwise hide an already configured folder and later fail a job with
`No such file or directory`.
- The token, qB password, and Syncthing API-key files are non-empty regular
files with no group/world permissions.
- The client image can read its configuration and reports usable permissions,
sparse-file support, and filesystem capabilities for both roots.
- qBittorrent authentication/version compatibility and Syncthing
authentication/device identity are healthy from the client container.
In particular, if `syncthing.api_root` is the Syncthing configuration root,
bind its `routes` subdirectory from a dedicated directory beneath qB's data
root and add a `local_path_overrides` mapping for that exact API path. This
prevents automatic route folders being created on a small configuration
filesystem while the client expects to hardlink from the qB data mount.
When converting an existing node, stop its client and Syncthing containers,
move each existing `routes/<route-id>` directory from the old Syncthing config
tree into the new qB-backed route-root directory, then recreate Syncthing and
run this preflight. Do not discard existing route directories: they can contain
route handshake state or an in-progress transfer namespace.
## Failure handling
Treat a nonzero exit status as a deployment blocker. Correct the compose
mounts, `client.toml`, secret permissions, service credentials, or service
image/version that the message identifies, then rerun the same command. Do not
work around a mapping failure by creating route folders manually: the route
provisioner relies on those API/local path mappings being exact.
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "archive-clients"
version = "0.1.18"
version = "0.1.19"
requires-python = ">=3.11"
dependencies = ["protobuf==7.35.1", "websockets==16.0"]
+327
View File
@@ -0,0 +1,327 @@
#!/usr/bin/env python3
"""Validate a host's Archive Control Docker deployment before it is used.
This intentionally relies only on the config file and the named containers on
the host where it runs. It never reads secret contents or calls a remote
control daemon. The hard-link probe creates uniquely named, empty files in
the configured qB and automatic-route directories and removes them afterward.
"""
from __future__ import annotations
import argparse
import json
import os
import stat
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any
class CheckFailure(RuntimeError):
pass
@dataclass(frozen=True)
class Mapping:
source: Path
destination: PurePosixPath
def docker_inspect(container: str) -> dict[str, Any]:
result = subprocess.run(
["docker", "inspect", container], check=False, text=True,
capture_output=True,
)
if result.returncode:
raise CheckFailure(f"cannot inspect container {container!r}")
try:
value = json.loads(result.stdout)
return value[0]
except (json.JSONDecodeError, IndexError, TypeError) as exc:
raise CheckFailure(f"invalid Docker inspection for {container!r}") from exc
def map_path(container: dict[str, Any], path: str) -> Mapping:
candidate = PurePosixPath(path)
matches: list[tuple[int, Mapping]] = []
for mount in container.get("Mounts", []):
if mount.get("Type") not in {"bind", "volume"}:
continue
source, destination = mount.get("Source"), mount.get("Destination")
if not isinstance(source, str) or not isinstance(destination, str):
continue
destination_path = PurePosixPath(destination)
try:
relative = candidate.relative_to(destination_path)
except ValueError:
continue
matches.append((len(destination_path.parts), Mapping(
Path(source).joinpath(*relative.parts), destination_path
)))
if not matches:
raise CheckFailure(f"{path} is not backed by a Docker bind/volume mount")
return max(matches, key=lambda item: item[0])[1]
def container_user(container: dict[str, Any]) -> str | None:
value = container.get("Config", {}).get("User")
return value if isinstance(value, str) and value else None
def require_same_path(label: str, left: Mapping, right: Mapping) -> None:
if left.source.resolve(strict=False) != right.source.resolve(strict=False):
raise CheckFailure(
f"{label} host paths differ: {left.source} != {right.source}"
)
def require_regular_secret(path: Path) -> None:
try:
mode = path.stat().st_mode
except OSError as exc:
raise CheckFailure(f"secret is unavailable: {path}") from exc
if not stat.S_ISREG(mode) or stat.S_IMODE(mode) & 0o077 or path.stat().st_size == 0:
raise CheckFailure(f"secret must be a non-empty mode-0600 regular file: {path}")
def route_local_path(syncthing: dict[str, Any]) -> str:
"""Resolve the fixed, future automatic ``routes/`` path in the client."""
api_root = PurePosixPath(syncthing["api_root"])
candidate = api_root / "routes"
overrides = syncthing.get("local_path_overrides", {})
if not isinstance(overrides, dict):
raise CheckFailure("syncthing.local_path_overrides must be a table")
matches: list[tuple[int, PurePosixPath, str]] = []
for raw_api, raw_local in overrides.items():
if not isinstance(raw_api, str) or not isinstance(raw_local, str):
raise CheckFailure("syncthing.local_path_overrides entries are invalid")
root = PurePosixPath(raw_api)
try:
relative = candidate.relative_to(root)
except ValueError:
continue
matches.append((len(root.parts), relative, raw_local))
if matches:
_, relative, local_root = max(matches, key=lambda item: item[0])
return str(Path(local_root).joinpath(*relative.parts))
try:
relative = candidate.relative_to(api_root)
except ValueError as exc:
raise CheckFailure("future route path is outside syncthing.api_root") from exc
return str(Path(syncthing["local_root"]).joinpath(*relative.parts))
def run_client_check(container: str, config: str, flag: str) -> None:
result = subprocess.run(
["docker", "exec", container, "archive-client", "--config", config, flag],
check=False, text=True, capture_output=True,
)
if result.returncode:
detail = result.stderr.strip() or result.stdout.strip() or "failed"
raise CheckFailure(f"client {flag} failed: {detail}")
print(result.stdout.strip())
def mounted_config(container: str, config: str) -> dict[str, Any]:
"""Read normalized non-secret config through the client image itself."""
program = '''import json,sys
from pathlib import Path
from archive_clients.config import ClientConfig
config=ClientConfig.load(Path(sys.argv[1]))
value={"shared_token_file":str(config.shared_token_file)}
value["qbittorrent"]={"api_root":str(config.qbittorrent.api_root),"local_root":str(config.qbittorrent.local_root),"password_file":str(config.qbittorrent.password_file)}
value["syncthing"]={"api_root":str(config.syncthing.api_root),"local_root":str(config.syncthing.local_root),"api_key_file":str(config.syncthing.api_key_file),"local_path_overrides":{str(api):str(local) for api,local in config.syncthing.local_path_overrides}}
print(json.dumps(value,sort_keys=True))'''
result = subprocess.run(
["docker", "exec", container, "python", "-c", program, config],
check=False, text=True, capture_output=True,
)
if result.returncode:
detail = result.stderr.strip() or result.stdout.strip() or "failed"
raise CheckFailure(f"cannot load mounted client config: {detail}")
try:
value = json.loads(result.stdout)
except json.JSONDecodeError as exc:
raise CheckFailure("mounted client config output is invalid") from exc
if not isinstance(value, dict):
raise CheckFailure("mounted client config is invalid")
return value
def run_service_check(container: str, config: str) -> None:
program = '''import json,sys
from pathlib import Path
from archive_clients.config import ClientConfig
from archive_clients.probes import probe_root
from archive_clients.services import probe_qbittorrent,probe_syncthing
config=ClientConfig.load(Path(sys.argv[1]))
sync_probe=probe_root(config.syncthing.local_root)
probes=[probe_qbittorrent(config.qbittorrent),probe_syncthing(config.syncthing,sparse_supported=sync_probe.sparse_files)]
print(json.dumps({"services":[{"service":p.service,"state":p.state,"detail":p.detail,"version":p.version,"device_id":p.device_id} for p in probes]},sort_keys=True))
raise SystemExit(0 if all(p.state == 1 for p in probes) else 1)'''
result = subprocess.run(
["docker", "exec", container, "python", "-c", program, config],
check=False, text=True, capture_output=True,
)
if result.returncode:
detail = result.stderr.strip() or result.stdout.strip() or "failed"
raise CheckFailure(f"client local-service probe failed: {detail}")
print(result.stdout.strip())
def run_hardlink_probe(
container: str, qb_root: str, route_root: str, user: str | None = None,
) -> None:
"""Prove that the daemon can hard-link from qB data into automatic routes."""
program = r'''import os,sys,tempfile
source_root,route_root=sys.argv[1:]
source_path=None
destination_path=None
try:
source_fd,source_path=tempfile.mkstemp(prefix=".archive-control-preflight-",dir=source_root)
os.close(source_fd)
destination_fd,destination_path=tempfile.mkstemp(prefix=".archive-control-preflight-",dir=route_root)
os.close(destination_fd)
os.unlink(destination_path)
os.link(source_path,destination_path)
source_stat=os.stat(source_path)
destination_stat=os.stat(destination_path)
if source_stat.st_dev != destination_stat.st_dev or source_stat.st_ino != destination_stat.st_ino:
raise RuntimeError("link(2) did not produce the same filesystem inode")
print("hard-link staging probe passed")
finally:
for path in (destination_path,source_path):
if path:
try:
os.unlink(path)
except FileNotFoundError:
pass
'''
command = ["docker", "exec"]
if user:
command.extend(["--user", user])
command.extend([container, "python", "-c", program, qb_root, route_root])
result = subprocess.run(
command,
check=False, text=True, capture_output=True,
)
if result.returncode:
detail = result.stderr.strip() or result.stdout.strip() or "failed"
raise CheckFailure(f"hard-link staging probe failed: {detail}")
print(result.stdout.strip())
def run_existing_route_directory_check(container: str, config: str) -> None:
"""Reject a mount migration that hides an already configured route folder."""
program = r'''import json,os,sys
from pathlib import Path,PurePosixPath
from archive_clients.config import ClientConfig
from archive_clients.syncthing import SyncthingHttp
config=ClientConfig.load(Path(sys.argv[1]))
transport=SyncthingHttp(config.syncthing)
route_root=config.syncthing.api_root / "routes"
missing=[]
checked=[]
for folder in transport.get_json("/rest/config").get("folders",[]):
if not isinstance(folder,dict) or not str(folder.get("label","")).startswith("archive-control:"):
continue
raw=folder.get("path")
if not isinstance(raw,str) or not raw:
continue
api=PurePosixPath(raw)
if api.parts and api.parts[0] == "~":
api=config.syncthing.api_root.joinpath(*api.parts[1:])
try:
api.relative_to(route_root)
except ValueError:
continue
local=config.syncthing.roots.api_to_local(api.as_posix())
checked.append(str(local))
if not local.is_dir():
missing.append({"folder_id":folder.get("id",""),"path":str(local)})
print(json.dumps({"checked":checked,"missing":missing},sort_keys=True))
raise SystemExit(1 if missing else 0)
'''
result = subprocess.run(
["docker", "exec", container, "python", "-c", program, config],
check=False, text=True, capture_output=True,
)
if result.returncode:
detail = result.stderr.strip() or result.stdout.strip() or "failed"
raise CheckFailure(
"an existing Archive Control Syncthing route directory is missing; "
f"complete the route-directory migration before starting jobs: {detail}"
)
print(result.stdout.strip())
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(
description="Read-only Archive Control Docker deployment preflight"
)
parser.add_argument("--client-config", type=Path, required=True,
help="host path to client.toml")
parser.add_argument("--client-container", required=True)
parser.add_argument("--syncthing-container", required=True)
parser.add_argument("--qbittorrent-container", required=True)
parser.add_argument("--container-config", default="/etc/archive-control/client.toml",
help="client.toml path inside the client container")
args = parser.parse_args(argv)
try:
client = docker_inspect(args.client_container)
syncthing = docker_inspect(args.syncthing_container)
qbittorrent = docker_inspect(args.qbittorrent_container)
configured_host_path = map_path(client, args.container_config).source
if configured_host_path.resolve(strict=False) != args.client_config.resolve(strict=False):
raise CheckFailure(
"--client-config does not match the file mounted into the client"
)
config = mounted_config(args.client_container, args.container_config)
qb = config["qbittorrent"]
sync = config["syncthing"]
client_qb = map_path(client, qb["local_root"])
client_route = map_path(client, route_local_path(sync))
syncthing_route = map_path(
syncthing, str(PurePosixPath(sync["api_root"]) / "routes")
)
qb_api = map_path(qbittorrent, qb["api_root"])
require_same_path(
"future Syncthing routes/local route mapping",
syncthing_route, client_route,
)
require_same_path("qBittorrent api_root/local_root", qb_api, client_qb)
if client_qb.destination != client_route.destination:
raise CheckFailure(
"qBittorrent and future route roots use separate client bind "
"mounts; hard-link staging would be unavailable"
)
run_hardlink_probe(
args.client_container, qb["local_root"], route_local_path(sync),
container_user(client),
)
run_existing_route_directory_check(
args.client_container, args.container_config
)
for key in ("shared_token_file",):
require_regular_secret(map_path(client, config[key]).source)
for service, key in ((qb, "password_file"), (sync, "api_key_file")):
require_regular_secret(map_path(client, service[key]).source)
run_client_check(args.client_container, args.container_config, "--check-config")
run_service_check(args.client_container, args.container_config)
except (CheckFailure, KeyError, OSError) as exc:
print(f"preflight failed: {exc}", file=sys.stderr)
return 1
print("preflight passed: bind mappings, hard-link staging, secrets, filesystem capabilities, and local APIs are healthy")
return 0
if __name__ == "__main__":
raise SystemExit(main())
+11 -1
View File
@@ -168,7 +168,7 @@ def _v1_files(info: dict[bytes, Any]) -> list[MetaFile]:
components = [name] + [_component(part) for part in raw_path]
result.append(MetaFile(
"/".join(components), _length(item.get(b"length")),
_padding(item),
_padding(item) or _bitcomet_padding_name(components[-1]),
))
return result
@@ -223,3 +223,13 @@ def _length(value: Any) -> int:
def _padding(value: dict[bytes, Any]) -> bool:
attributes = value.get(b"attr", b"")
return isinstance(attributes, bytes) and b"p" in attributes
def _bitcomet_padding_name(component: str) -> bool:
"""Recognize BitComet's legacy padding-file convention.
Such v1 torrents often omit the standard ``attr=p`` flag, but
qBittorrent/libtorrent still hides these synthetic entries from its file
API. The exact reserved prefix is the interoperable marker.
"""
return component.startswith("_____padding_file_")
+7
View File
@@ -85,6 +85,7 @@ class ConnectionConfig:
registration_timeout: float = 10
heartbeat_interval: float = 15
offline_timeout: float = 45
outbound_enqueue_timeout: float = 15
reconnect_initial: float = 1
reconnect_max: float = 60
reconnect_reset_after: float = 60
@@ -249,6 +250,7 @@ def _connection(value: Any) -> ConnectionConfig:
raise ConfigError("connection must be a table")
_keys(value, {
"registration_timeout", "heartbeat_interval", "offline_timeout",
"outbound_enqueue_timeout",
"reconnect_initial", "reconnect_max", "reconnect_reset_after",
"reconnect_jitter",
}, "connection")
@@ -256,6 +258,9 @@ def _connection(value: Any) -> ConnectionConfig:
registration_timeout=_duration(value.get("registration_timeout", "10s")),
heartbeat_interval=_duration(value.get("heartbeat_interval", "15s")),
offline_timeout=_duration(value.get("offline_timeout", "45s")),
outbound_enqueue_timeout=_duration(
value.get("outbound_enqueue_timeout", "15s")
),
reconnect_initial=_duration(value.get("reconnect_initial", "1s")),
reconnect_max=_duration(value.get("reconnect_max", "60s")),
reconnect_reset_after=_duration(value.get("reconnect_reset_after", "60s")),
@@ -265,6 +270,8 @@ def _connection(value: Any) -> ConnectionConfig:
raise ConfigError("reconnect_initial cannot exceed reconnect_max")
if result.offline_timeout <= result.heartbeat_interval:
raise ConfigError("offline_timeout must exceed heartbeat_interval")
if result.outbound_enqueue_timeout <= 0:
raise ConfigError("outbound_enqueue_timeout must be positive")
if not isinstance(result.reconnect_jitter, bool):
raise ConfigError("reconnect_jitter must be boolean")
return result
+105 -41
View File
@@ -229,16 +229,11 @@ class ArchiveClientDaemon:
# detects the missing application heartbeats and reconnects.
while True:
try:
frame = await asyncio.wait_for(
websocket.recv(),
self.config.connection.offline_timeout,
frame = await self._next_frame(
websocket, writer
)
except ConnectionClosedOK:
return
except asyncio.TimeoutError as exc:
raise RuntimeError(
"control heartbeat timed out"
) from exc
await self._handle(decode(frame), outbound, command_tasks)
finally:
writer.cancel()
@@ -323,6 +318,63 @@ class ArchiveClientDaemon:
while True:
await websocket.send(await outbound.get())
async def _next_frame(self, websocket: Any, writer: asyncio.Task[None]):
"""Receive one frame while supervising the connection writer.
The previous receive-only wait let a failed writer go unnoticed until
the remote side happened to close or a heartbeat timeout elapsed.
Waiting for both directions makes a failed send an immediate,
reconnectable connection failure.
"""
receiver = asyncio.create_task(websocket.recv())
try:
done, _ = await asyncio.wait(
{receiver, writer},
timeout=self.config.connection.offline_timeout,
return_when=asyncio.FIRST_COMPLETED,
)
if writer in done:
if not receiver.done():
receiver.cancel()
await asyncio.gather(receiver, return_exceptions=True)
if writer.cancelled():
raise RuntimeError("control writer was cancelled")
error = writer.exception()
if error is not None:
raise RuntimeError("control writer failed") from error
raise RuntimeError("control writer stopped")
if receiver not in done:
receiver.cancel()
await asyncio.gather(receiver, return_exceptions=True)
raise RuntimeError("control heartbeat timed out")
return receiver.result()
except BaseException:
if not receiver.done():
receiver.cancel()
await asyncio.gather(receiver, return_exceptions=True)
raise
async def _enqueue(
self, outbound: asyncio.Queue[str], message: str
) -> None:
"""Bound connection-local backpressure so a dead writer cannot wedge I/O.
A job event is durable before it is sent, so dropping this particular
connection after a bounded wait is safe: reconciliation/redelivery on
the next session will recover it. In contrast, indefinitely waiting
for a full queue can prevent heartbeat acknowledgements from being
read or sent, which prevents the reconnect supervisor from running.
"""
try:
await asyncio.wait_for(
outbound.put(message),
timeout=self.config.connection.outbound_enqueue_timeout,
)
except asyncio.TimeoutError as exc:
raise RuntimeError("control outbound queue is blocked") from exc
async def _handle(
self,
envelope: Any,
@@ -334,13 +386,20 @@ class ArchiveClientDaemon:
response = new_envelope()
response.correlation_id = envelope.message_id
response.heartbeat_ack.sequence = envelope.heartbeat.sequence
await outbound.put(encode(response))
await self._enqueue(outbound, encode(response))
elif payload == "command":
await self._accept_command(envelope, outbound, command_tasks)
elif payload == "protocol_error":
logger.warning(
"control_reported_protocol_error",
extra={"error_code": envelope.protocol_error.error.code},
extra={
"error_code": envelope.protocol_error.error.code,
"error_detail": envelope.protocol_error.error.message,
"offending_message_id": (
envelope.protocol_error.offending_message_id
),
"retryable": envelope.protocol_error.error.retryable,
},
)
async def _accept_command(
@@ -398,7 +457,7 @@ class ArchiveClientDaemon:
response = new_envelope()
response.correlation_id = envelope.message_id
response.command_ack.CopyFrom(acknowledgement)
await outbound.put(encode(response))
await self._enqueue(outbound, encode(response))
if (
accepted is not None
@@ -419,7 +478,7 @@ class ArchiveClientDaemon:
job_snapshot.last_event_sequence = int(
row["last_event_sequence"]
)
await outbound.put(encode(snapshot))
await self._enqueue(outbound, encode(snapshot))
elif (
accepted is not None
and accepted_for_execution
@@ -472,7 +531,6 @@ class ArchiveClientDaemon:
outbound: asyncio.Queue[str],
command_tasks: set[asyncio.Task[None]],
) -> None:
job_commands: list[control_pb2.Command] = []
for row in await asyncio.to_thread(self.store.list_accepted_commands):
acknowledgement = decode_message(
str(row["acknowledgement_json"]), control_pb2.CommandAck()
@@ -484,28 +542,14 @@ class ArchiveClientDaemon:
)
if command.WhichOneof("payload") == "ensure_route":
self._schedule_route_command(
command, "", outbound, command_tasks
command, "", outbound, command_tasks, restore_ready=True
)
elif (
command.WhichOneof("payload") in {
"assign_job", "execute_step", "cancel_job"
}
and self.jobs is not None
):
if command.WhichOneof("payload") == "cancel_job":
self.jobs.request_cancel(command.cancel_job.job_id)
job_commands.append(command)
if job_commands:
task = asyncio.create_task(
self._resume_job_commands(job_commands, outbound),
name="resume-job-commands",
)
command_tasks.add(task)
task.add_done_callback(
lambda completed: self._command_finished(
completed, command_tasks
)
)
# Job commands are intentionally not replayed here. A command
# acknowledgement is durable on both sides; the control daemon
# redelivers an unacknowledged command, while registration
# reconciliation requests snapshots for active jobs. Replaying
# every historical command re-emits overlapping event ranges and
# can flood the single connection's bounded outbound queue.
async def _resume_route_commands(
self,
@@ -604,10 +648,13 @@ class ArchiveClientDaemon:
response.correlation_id = correlation_id
response.job_event.CopyFrom(event)
future = asyncio.run_coroutine_threadsafe(
outbound.put(encode(response)), loop
self._enqueue(outbound, encode(response)), loop
)
try:
future.result(timeout=30)
future.result(
timeout=self.config.connection.outbound_enqueue_timeout
+ 1
)
except Exception:
# The event is already durable in the client DB and will
# be replayed after reconnect.
@@ -627,7 +674,7 @@ class ArchiveClientDaemon:
response = new_envelope()
response.correlation_id = correlation_id
response.job_event.CopyFrom(event)
await outbound.put(encode(response))
await self._enqueue(outbound, encode(response))
def _schedule_route_command(
self,
@@ -635,12 +682,15 @@ class ArchiveClientDaemon:
correlation_id: str,
outbound: asyncio.Queue[str],
command_tasks: set[asyncio.Task[None]],
restore_ready: bool = False,
) -> None:
if command.command_id in self._active_route_commands:
return
self._active_route_commands.add(command.command_id)
task = asyncio.create_task(
self._execute_route(command, correlation_id, outbound),
self._execute_route(
command, correlation_id, outbound, restore_ready=restore_ready
),
name=f"route-{command.ensure_route.route.route_id}",
)
command_tasks.add(task)
@@ -671,13 +721,14 @@ class ArchiveClientDaemon:
response = new_envelope()
response.correlation_id = correlation_id
response.inventory_chunk.CopyFrom(chunk)
await outbound.put(encode(response))
await self._enqueue(outbound, encode(response))
async def _execute_route(
self,
command: Any,
correlation_id: str,
outbound: asyncio.Queue[str],
restore_ready: bool = False,
) -> None:
assert self.routes is not None
spec = command.ensure_route.route
@@ -696,8 +747,21 @@ class ArchiveClientDaemon:
response.route_update.CopyFrom(
decode_message(str(row["update_json"]), control_pb2.RouteUpdate())
)
await outbound.put(encode(response))
if attempt["state"] in {"ready", "failed"}:
await self._enqueue(outbound, encode(response))
if attempt["state"] == "ready":
if restore_ready:
# Route readiness is durable, but this process-local lookup is
# not. Reconfigure (which validates the existing Syncthing
# folder and recreates its local directory if necessary) after
# a daemon restart, before replaying stored READY updates.
configured = await asyncio.to_thread(
self.routes.configure,
spec,
time.monotonic() + spec.setup_timeout_seconds,
)
self._known_route_paths[spec.route_id] = configured.local_path
return
if attempt["state"] == "failed":
return
deadline = time.monotonic() + spec.setup_timeout_seconds
@@ -848,7 +912,7 @@ class ArchiveClientDaemon:
response = new_envelope()
response.correlation_id = correlation_id
response.route_update.CopyFrom(update)
await outbound.put(encode(response))
await self._enqueue(outbound, encode(response))
def _local_route(
self, spec: route_pb2.EnsureRouteSpec, state: int
+121
View File
@@ -110,6 +110,123 @@ class DaemonTransportTests(unittest.IsolatedAsyncioTestCase):
):
await asyncio.wait_for(daemon._connection(), 1)
async def test_failed_writer_ends_connection_without_waiting_for_heartbeat(self):
"""A send failure must immediately reach the reconnect supervisor."""
async def control(websocket):
registration = decode(await websocket.recv())
response = new_envelope()
response.correlation_id = registration.message_id
response.register_response.status = (
client_pb2.REGISTRATION_STATUS_ACCEPTED
)
response.register_response.negotiated_version.major = 1
await websocket.send(encode(response))
await websocket.wait_closed()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
token = root / "token"
token.write_text("shared-secret", encoding="utf-8")
os.chmod(token, 0o600)
async with serve(control, "127.0.0.1", 0, ping_interval=None) as server:
port = server.sockets[0].getsockname()[1]
service = ServiceConfig("http://local", PurePosixPath("/api"), root)
config = ClientConfig(
"cache-1", "Cache 1", "cache",
f"ws://127.0.0.1:{port}", token,
root / "state.db", root / "backups", service, service,
)
probe = FilesystemProbe(root, True, True, True, True, True)
daemon = ArchiveClientDaemon(config, [probe, probe], [])
await asyncio.to_thread(daemon.store.initialize)
async def failed_writer(websocket, outbound):
raise OSError("simulated broken socket")
daemon._writer = failed_writer
with self.assertRaisesRegex(RuntimeError, "writer failed"):
await asyncio.wait_for(daemon._connection(), 1)
async def test_full_outbound_queue_aborts_connection_instead_of_blocking_heartbeats(self):
"""Bulk output cannot indefinitely block the receive/heartbeat loop."""
async def control(websocket):
registration = decode(await websocket.recv())
response = new_envelope()
response.correlation_id = registration.message_id
response.register_response.status = (
client_pb2.REGISTRATION_STATUS_ACCEPTED
)
response.register_response.negotiated_version.major = 1
await websocket.send(encode(response))
for sequence in range(1, 102):
heartbeat = new_envelope()
heartbeat.heartbeat.sequence = sequence
await websocket.send(encode(heartbeat))
await websocket.wait_closed()
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
token = root / "token"
token.write_text("shared-secret", encoding="utf-8")
os.chmod(token, 0o600)
async with serve(control, "127.0.0.1", 0, ping_interval=None) as server:
port = server.sockets[0].getsockname()[1]
service = ServiceConfig("http://local", PurePosixPath("/api"), root)
config = ClientConfig(
"cache-1", "Cache 1", "cache",
f"ws://127.0.0.1:{port}", token,
root / "state.db", root / "backups", service, service,
ConnectionConfig(outbound_enqueue_timeout=0.01),
)
probe = FilesystemProbe(root, True, True, True, True, True)
daemon = ArchiveClientDaemon(config, [probe, probe], [])
await asyncio.to_thread(daemon.store.initialize)
async def stopped_writer(websocket, outbound):
await asyncio.Event().wait()
daemon._writer = stopped_writer
with self.assertRaisesRegex(RuntimeError, "outbound queue is blocked"):
await asyncio.wait_for(daemon._connection(), 2)
async def test_reconnect_resume_skips_historical_job_commands(self):
"""Registration reconciliation, not command replay, recovers job state."""
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
token = root / "token"
token.write_text("shared-secret", encoding="utf-8")
os.chmod(token, 0o600)
service = ServiceConfig("http://local", PurePosixPath("/api"), root)
config = ClientConfig(
"cache-1", "Cache 1", "cache", "ws://control", token,
root / "state.db", root / "backups", service, service,
)
probe = FilesystemProbe(root, True, True, True, True, True)
daemon = ArchiveClientDaemon(config, [probe, probe], [])
await asyncio.to_thread(daemon.store.initialize)
command = control_pb2.Command(command_id=str(uuid4()))
command.execute_step.job_id = str(uuid4())
command.execute_step.expected_last_event_sequence = 1
acknowledgement = control_pb2.CommandAck(
command_id=command.command_id,
status=control_pb2.COMMAND_ACK_STATUS_ACCEPTED,
)
await asyncio.to_thread(
daemon.store.accept_command,
command.command_id,
encode_message(command),
encode_message(acknowledgement),
)
daemon.jobs = Mock()
outbound = asyncio.Queue()
tasks = set()
await daemon._resume_commands(outbound, tasks)
self.assertEqual(tasks, set())
self.assertTrue(outbound.empty())
async def test_eviction_assignment_and_steps_are_admitted(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
@@ -280,6 +397,10 @@ class DaemonTransportTests(unittest.IsolatedAsyncioTestCase):
for _ in range(3)
]
self.assertEqual([item.sequence for item in resumed], [1, 2, 3])
self.assertEqual(manager.configure.call_count, 2)
self.assertEqual(
restarted._route_path("route-1"), root / "routes/route-1"
)
async def test_slow_inventory_does_not_block_heartbeat(self):
with tempfile.TemporaryDirectory() as directory:
+83
View File
@@ -0,0 +1,83 @@
import importlib.util
import sys
import unittest
from unittest import mock
from pathlib import Path
_SCRIPT = Path(__file__).parents[1] / "scripts" / "preflight-deployment.py"
_SPEC = importlib.util.spec_from_file_location("deployment_preflight", _SCRIPT)
assert _SPEC is not None and _SPEC.loader is not None
preflight = importlib.util.module_from_spec(_SPEC)
sys.modules[_SPEC.name] = preflight
_SPEC.loader.exec_module(preflight)
class DeploymentPreflightTests(unittest.TestCase):
def test_maps_nested_container_path_to_host_source(self):
container = {
"Mounts": [
{"Type": "bind", "Source": "/srv/data", "Destination": "/data"},
{"Type": "bind", "Source": "/srv/sync", "Destination": "/data/Sync"},
]
}
mapping = preflight.map_path(container, "/data/Sync/routes/route-1")
self.assertEqual(mapping.source, Path("/srv/sync/routes/route-1"))
self.assertEqual(str(mapping.destination), "/data/Sync")
def test_rejects_api_and_client_roots_from_different_host_paths(self):
with self.assertRaisesRegex(preflight.CheckFailure, "host paths differ"):
preflight.require_same_path(
"Syncthing api_root/local_root",
preflight.Mapping(Path("/srv/syncthing-config/routes"), Path("/var/syncthing")),
preflight.Mapping(Path("/srv/sync/routes"), Path("/data/storage")),
)
def test_rejects_unmounted_container_path(self):
with self.assertRaisesRegex(preflight.CheckFailure, "not backed"):
preflight.map_path({"Mounts": []}, "/missing")
def test_maps_future_route_through_syncthing_override(self):
config = {
"api_root": "/var/syncthing",
"local_root": "/data/sync",
"local_path_overrides": {
"/var/syncthing/routes": "/data/qb/.archive-control-routes",
},
}
self.assertEqual(
preflight.route_local_path(config),
"/data/qb/.archive-control-routes",
)
def test_runs_hardlink_probe_with_qb_and_route_roots(self):
completed = __import__("subprocess").CompletedProcess(
args=[], returncode=0, stdout="hard-link staging probe passed\n", stderr=""
)
with mock.patch.object(preflight.subprocess, "run", return_value=completed) as run:
preflight.run_hardlink_probe(
"client", "/data/qb", "/data/qb/routes", "1001:1001"
)
args = run.call_args.args[0]
self.assertEqual(args[:6], ["docker", "exec", "--user", "1001:1001", "client", "python"])
self.assertEqual(args[-2:], ["/data/qb", "/data/qb/routes"])
def test_hardlink_probe_failure_is_actionable(self):
completed = __import__("subprocess").CompletedProcess(
args=[], returncode=1, stdout="", stderr="[Errno 18] Invalid cross-device link"
)
with mock.patch.object(preflight.subprocess, "run", return_value=completed):
with self.assertRaisesRegex(preflight.CheckFailure, "hard-link staging probe failed"):
preflight.run_hardlink_probe("client", "/data/qb", "/data/qb/routes")
def test_existing_route_directory_failure_is_actionable(self):
completed = __import__("subprocess").CompletedProcess(
args=[], returncode=1, stdout='{"missing":[{"folder_id":"route-1"}]}', stderr=""
)
with mock.patch.object(preflight.subprocess, "run", return_value=completed):
with self.assertRaisesRegex(preflight.CheckFailure, "route directory is missing"):
preflight.run_existing_route_directory_check("client", "/config.toml")
if __name__ == "__main__":
unittest.main()
+1 -1
View File
@@ -123,7 +123,7 @@ class ResourceTests(unittest.TestCase):
b"files": [
{b"length": 3, b"path": [b"first.bin"]},
{
b"attr": b"p", b"length": 5,
b"length": 5,
b"path": [b"_____padding_file_5"],
},
{b"length": 7, b"path": [b"last.bin"]},