Compare commits
9
Commits
v0.1.18
...
62a0f24437
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
62a0f24437 | ||
|
|
eabbda3c3a | ||
|
|
a7bc2bf087 | ||
|
|
25d40daea6 | ||
|
|
f3517bd21b | ||
|
|
300210c17d | ||
|
|
1f97faeab5 | ||
|
|
1c2590c3c8 | ||
|
|
515599f2d4 |
@@ -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
|
||||
|
||||
@@ -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
@@ -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.
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
Executable
+18
@@ -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
|
||||
@@ -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"]
|
||||
|
||||
@@ -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"]
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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
@@ -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"]
|
||||
|
||||
|
||||
Executable
+327
@@ -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())
|
||||
@@ -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_")
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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()
|
||||
@@ -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"]},
|
||||
|
||||
Reference in New Issue
Block a user