feat(deploy): add verified image deployments with consistent backups

deploy/compose.image.yml

@@ -0,0 +1,15 @@
+# Copy this file to the instance directory; use a private .env beside it.
+name: ${RAWMARK_PROJECT:?Set RAWMARK_PROJECT}
+services:
+  rawmark:
+    image: ${RAWMARK_IMAGE:?Set a verified registry image digest}
+    ports:
+      - "${RAWMARK_BIND:-127.0.0.1}:${RAWMARK_PORT:?Set RAWMARK_PORT}:8080"
+    volumes:
+      - type: bind
+        source: ${RAWMARK_DATA_DIR:?Set the absolute persistent data directory}
+        target: /data
+        bind:
+          create_host_path: false
+    restart: unless-stopped
+    stop_grace_period: 15s

deploy/nginx/raw-dev.134333.xyz.conf

@@ -0,0 +1,47 @@
+# Keep resource tickets and query strings out of proxy logs.
+map $uri $rawmark_dev_log_path {
+  ~^/api/v1/resources/[^/]+(?<rawmark_dev_suffix>/.*)$ /api/v1/resources/[redacted]$rawmark_dev_suffix;
+  default $uri;
+}
+log_format rawmark_dev '$remote_addr - $remote_user [$time_local] '
+  '"$request_method $rawmark_dev_log_path $server_protocol" $status $body_bytes_sent '
+  '"$http_user_agent"';
+
+server {
+  listen 80;
+  server_name raw-dev.134333.xyz;
+  access_log /var/log/nginx/raw-dev.134333.xyz.log rawmark_dev;
+  error_log /dev/null;
+  return 308 https://$host$request_uri;
+}
+server {
+  listen 443 ssl;
+  http2 on;
+  server_name raw-dev.134333.xyz;
+  include /etc/nginx/snippets/ssl-134333.xyz.conf;
+  include /etc/nginx/snippets/no-spider.conf;
+  access_log /var/log/nginx/raw-dev.134333.xyz.log rawmark_dev;
+  error_log /var/log/nginx/raw-dev.134333.xyz.error.log warn;
+  client_max_body_size 0;
+
+  location / {
+    proxy_pass http://127.0.0.1:21001;
+    proxy_http_version 1.1;
+    proxy_request_buffering off;
+    proxy_buffering off;
+    proxy_set_header Upgrade $http_upgrade;
+    proxy_set_header Connection $connection_upgrade;
+    proxy_set_header Host $http_host;
+    proxy_set_header X-Forwarded-Proto $scheme;
+    proxy_set_header X-Real-IP $remote_addr;
+    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+    proxy_read_timeout 1h;
+    proxy_send_timeout 1h;
+
+    # Error logs cannot redact ticket-bearing request paths.
+    location ^~ /api/v1/resources/ {
+      proxy_pass http://127.0.0.1:21001;
+      error_log /dev/null;
+    }
+  }
+}

scripts/deploy.py

@@ -0,0 +1,162 @@
+#!/usr/bin/env python3
+"""Deploy a verified image to one explicitly configured Compose instance."""
+
+import argparse
+import contextlib
+import fcntl
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import tarfile
+import tempfile
+import time
+import urllib.request
+
+from release import API, GITEA_IMAGE, GITHUB_IMAGE, release_receipt, require, run
+
+
+def settings(directory, channel):
+    values = {}
+    for line in (directory / '.env').read_text().splitlines():
+        if line.strip() and not line.lstrip().startswith('#'):
+            key, value = line.split('=', 1)
+            require(key.startswith('RAWMARK_') and value and '\n' not in value, 'Invalid instance setting')
+            values[key] = value
+    require(values['RAWMARK_CHANNEL'] == channel, 'Instance channel does not match deployment command')
+    require(re.fullmatch(r'rawmark(?:-[a-z0-9-]+)?', values['RAWMARK_PROJECT']), 'Invalid Compose project')
+    require(Path(values['RAWMARK_DATA_DIR']).is_absolute(), 'Persistent data path must be absolute')
+    require(Path(values['RAWMARK_DATA_DIR']).resolve() == directory / 'data', 'Data must belong to this instance')
+    require((directory / 'data').is_dir(), 'Persistent data directory does not exist')
+    require(values['RAWMARK_BIND'] in ('127.0.0.1', '0.0.0.0'), 'Invalid bind address')
+    require(21000 <= int(values['RAWMARK_PORT']) <= 21999, 'Port must follow the personal-project range')
+    return values
+
+
+def save(path, data):
+    with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as target:
+        temporary = Path(target.name)
+        target.write(data)
+        target.flush()
+        os.fsync(target.fileno())
+    try:
+        temporary.replace(path)
+    finally:
+        temporary.unlink(missing_ok=True)
+
+
+def deployment(directory, channel, image, commit, version, client_container=None):
+    directory = Path(directory).resolve(strict=True)
+    cfg = settings(directory, channel)
+    registry = GITEA_IMAGE if channel == 'development' else GITHUB_IMAGE
+    require(re.fullmatch(re.escape(registry) + r'@sha256:[0-9a-f]{64}', image), 'Expected an immutable registry digest')
+    require(re.fullmatch(r'[0-9a-f]{40}', commit), 'Expected a full commit SHA')
+    # Serializes actual deployments, including manually started jobs and CLI use.
+    with (directory / '.deploy.lock').open('a') as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        require(not (directory / 'deployment.pending.json').exists(),
+                'Resolve the previous pending deployment before attempting another update')
+        run('docker', 'pull', image)
+        inspected = json.loads(run('docker', 'image', 'inspect', image))[0]
+        labels = inspected['Config'].get('Labels') or {}
+        require(labels.get('org.opencontainers.image.revision') == commit, 'Image revision mismatch')
+        require(labels.get('org.opencontainers.image.version') == version, 'Image version mismatch')
+        require(inspected['Config']['User'].split(':')[0] not in ('', '0', 'root'), 'Image must run as non-root')
+        require(run('docker', 'run', '--rm', '--network', 'none', image, '--version') == 'RawMark ' + version,
+                'Executable version mismatch')
+        if channel == 'development':
+            api = API(os.environ['GITEA_API_URL'], os.environ['GITEA_RELEASE_TOKEN'])
+            if api.repo('/branches/main')['commit']['id'] != commit:
+                print('Skipped development deployment: main has advanced')
+                return
+        identifiers = run('docker', 'ps', '-aq', '--filter', 'label=com.docker.compose.project=' + cfg['RAWMARK_PROJECT'],
+                          '--filter', 'label=com.docker.compose.service=rawmark').split()
+        require(len(identifiers) <= 1, 'Multiple containers claim this instance')
+        previous = json.loads(run('docker', 'inspect', identifiers[0]))[0] if identifiers else None
+        if previous:
+            mounts = [m for m in previous['Mounts'] if m['Destination'] == '/data']
+            require(len(mounts) == 1 and mounts[0]['Type'] == 'bind'
+                    and Path(mounts[0]['Source']).resolve() == directory / 'data', 'Existing data mount differs')
+        writers = run('docker', 'ps', '-q', '--filter', 'volume=' + str(directory / 'data')).split()
+        require(set(writers) <= {i[:12] for i in identifiers} | set(identifiers), 'Another container is using the data directory')
+        env = {**os.environ, **cfg, 'RAWMARK_IMAGE': image}
+        compose = ['docker', 'compose', '--project-name', cfg['RAWMARK_PROJECT'], '--env-file', str(directory / '.env'),
+                   '-f', str(directory / 'compose.yml')]
+        run(*compose, 'config', '--quiet', env=env)
+        backup = None
+        if previous:
+            run('docker', 'stop', '--time', '15', previous['Id'])
+            try:
+                backups = directory / 'backups'
+                backups.mkdir(mode=0o700, exist_ok=True)
+                backup = backups / (time.strftime('%Y%m%dT%H%M%SZ', time.gmtime()) + '-' + commit[:12] + '.tar.gz')
+                # Exclusively create the backup, including WAL files after the service is stopped.
+                with backup.open('xb') as target:
+                    os.chmod(backup, 0o600)
+                    with tarfile.open(fileobj=target, mode='w:gz') as archive:
+                        archive.add(directory / 'data', arcname='data')
+                        for name in ('.env', 'compose.yml', 'deployment.json'):
+                            if (directory / name).is_file():
+                                archive.add(directory / name, arcname=name)
+            except BaseException:
+                # No new executable has touched the data; restarting the old container is safe here.
+                if previous['State']['Running']:
+                    run('docker', 'start', previous['Id'])
+                raise
+        pending = {'image': image, 'commit': commit, 'version': version, 'channel': channel,
+                   'previous_image': previous['Image'] if previous else None,
+                   'backup': str(backup) if backup else None}
+        save(directory / 'deployment.pending.json', (json.dumps(pending, indent=2) + '\n').encode())
+        try:
+            run(*compose, 'up', '-d', '--no-build', '--pull', 'never', '--wait', '--wait-timeout', '120', env=env)
+            identifier = run(*compose, 'ps', '-q', 'rawmark', env=env)
+            info = json.loads(run('docker', 'inspect', identifier))[0]
+            require(info['Image'] == inspected['Id'], 'Running image differs from selected image')
+            run('docker', 'exec', identifier, '/rawmark', 'healthcheck')
+            with contextlib.ExitStack() as cleanup:
+                network = next(iter(info['NetworkSettings']['Networks'].values()))
+                if client_container:
+                    run('docker', 'network', 'connect', network['NetworkID'], client_container)
+                    cleanup.callback(run, 'docker', 'network', 'disconnect', network['NetworkID'], client_container)
+                origin = 'http://' + network['IPAddress'] + ':8080'
+                opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+                with opener.open(origin + '/', timeout=10) as response:
+                    require(b'<html' in response.read().lower(), 'Embedded frontend is missing')
+                with opener.open(origin + '/api/v1/auth/state', timeout=10) as response:
+                    require(json.load(response)['code'] == 0, 'Authentication state failed')
+        except BaseException:
+            # Data compatibility cannot be inferred from a failed health check. Do not restore or downgrade it automatically.
+            subprocess.run([*compose, 'stop', 'rawmark'], env=env, check=True)
+            print('Deployment failed; inspect deployment.pending.json and the consistent backup before recovery', flush=True)
+            raise
+        cfg['RAWMARK_IMAGE'] = image
+        save(directory / '.env', ''.join(f'{key}={value}\n' for key, value in cfg.items()).encode())
+        save(directory / 'deployment.json', (json.dumps(pending, indent=2) + '\n').encode())
+        (directory / 'deployment.pending.json').unlink()
+        print(json.dumps(pending))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('channel', choices=('development', 'production'))
+    parser.add_argument('--directory', required=True)
+    parser.add_argument('--image')
+    parser.add_argument('--commit')
+    parser.add_argument('--version', required=True)
+    parser.add_argument('--client-container')
+    args = parser.parse_args()
+    if args.channel == 'production':
+        require(not args.image and not args.commit, 'Production accepts a completed version, not an arbitrary image')
+        gitea = API(os.environ['GITEA_API_URL'], os.environ['GITEA_READ_TOKEN'])
+        github = API('https://api.github.com', '', github=True)
+        build, receipt = release_receipt(gitea, github, args.version)
+        image, commit = GITHUB_IMAGE + '@' + receipt['image_digest'], build['commit']
+    else:
+        image, commit = args.image, args.commit
+        require(image and commit, 'Development requires an image digest and commit')
+    deployment(args.directory, args.channel, image, commit, args.version, args.client_container)
+
+
+if __name__ == '__main__':
+    main()

scripts/tests/test_cicd.py

@@ -0,0 +1,305 @@
+"""Exercise artifact integrity and deployment failure boundaries with synthetic data."""
+
+import contextlib
+import hashlib
+import io
+import json
+import os
+import runpy
+from pathlib import Path
+import sys
+import tarfile
+import tempfile
+import unittest
+from unittest.mock import Mock, patch
+import urllib.error
+import urllib.request
+import zipfile
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import deploy
+import release
+cleanup = runpy.run_path(str(Path(__file__).resolve().parents[1] / 'ci-cleanup.py'))['cleanup']
+
+
+SHA = 'a' * 40
+DIGEST = 'sha256:' + 'b' * 64
+
+
+class BundleTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.cwd = Path.cwd()
+        os.chdir(self.tmp.name)
+        self.addCleanup(os.chdir, self.cwd)
+        Path('VERSION').write_text('0.1.1\n')
+        Path('downloads').mkdir()
+        Path('downloads/rawmark_0.1.1_linux_amd64.tar.gz').write_bytes(b'archive')
+        self.environment = patch.dict(os.environ, {'CI_RUN_ID': '42'})
+        self.environment.start()
+        self.addCleanup(self.environment.stop)
+        self.git = patch.object(release, 'run', return_value=SHA)
+        self.git.start()
+        self.addCleanup(self.git.stop)
+
+    def test_development_has_only_one_target_and_exact_checksums(self):
+        release.bundle('development')
+        build = release.verify_bundle()
+        self.assertEqual(build['platforms'], ['linux/amd64'])
+        self.assertEqual(build['run_id'], 42)
+        self.assertEqual(Path('downloads/SHA256SUMS').read_text(),
+                         hashlib.sha256(b'archive').hexdigest() + '  rawmark_0.1.1_linux_amd64.tar.gz\n')
+
+    def test_corrupted_download_is_rejected(self):
+        release.bundle('development')
+        Path('downloads/rawmark_0.1.1_linux_amd64.tar.gz').write_bytes(b'corrupt')
+        with self.assertRaisesRegex(ValueError, 'Checksum mismatch'):
+            release.verify_bundle()
+
+    def test_archive_at_cdn_budget_is_rejected(self):
+        with patch.object(release, 'UPLOAD_LIMIT', 7):
+            with self.assertRaisesRegex(ValueError, 'CDN upload budget'):
+                release.bundle('development')
+
+    def test_receipt_from_other_commit_is_rejected(self):
+        release.bundle('development')
+        with patch.object(release, 'run', return_value='c' * 40):
+            with self.assertRaisesRegex(ValueError, 'commit mismatch'):
+                release.verify_bundle()
+
+    def test_missing_release_platform_is_rejected(self):
+        release.bundle('development')
+        path = Path('downloads/build.json')
+        build = json.loads(path.read_text())
+        build['mode'] = 'release'
+        release.write_json(path, build)
+        with self.assertRaisesRegex(ValueError, 'platforms mismatch'):
+            release.verify_bundle()
+
+
+class PublicationTests(unittest.TestCase):
+    def test_cross_host_redirect_strips_credentials(self):
+        request = urllib.request.Request('https://api.github.com/asset', headers={'Authorization': 'Bearer private'})
+        redirected = release.SafeRedirect().redirect_request(request, None, 302, 'Found', {}, 'https://objects.example/asset')
+        self.assertFalse(redirected.has_header('Authorization'))
+
+    def test_anonymous_client_cannot_write(self):
+        api = release.API('https://api.github.com', '', github=True)
+        with self.assertRaisesRegex(ValueError, 'cannot mutate'):
+            api.repo('/releases', data={}, method='POST')
+
+    def test_existing_asset_conflict_is_not_overwritten(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'archive.zip'
+            path.write_bytes(b'new')
+            api = release.API('https://git.example/api/v1', 'private')
+            api.assets = Mock(return_value=[{'name': path.name}])
+            api.download = Mock(return_value=b'old')
+            api.request = Mock()
+            with self.assertRaisesRegex(ValueError, 'conflicts'):
+                api.upload({'draft': True}, path)
+            api.request.assert_not_called()
+
+    def test_identical_asset_is_reused_even_after_publication(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'archive.zip'
+            path.write_bytes(b'same')
+            api = release.API('https://git.example/api/v1', 'private')
+            api.assets = Mock(return_value=[{'name': path.name}])
+            api.download = Mock(return_value=b'same')
+            api.request = Mock()
+            api.upload({'draft': False}, path)
+            api.request.assert_not_called()
+
+    def test_incomplete_release_cannot_be_deployed(self):
+        gitea, github = Mock(), Mock()
+        gitea.release.return_value = {'draft': True, 'prerelease': False}
+        github.release.return_value = {'draft': False, 'prerelease': False}
+        with self.assertRaisesRegex(ValueError, 'not completely published'):
+            release.release_receipt(gitea, github, '0.1.1')
+
+    def test_registry_permission_error_is_not_treated_as_missing_image(self):
+        result = Mock(returncode=1, stderr='403 Forbidden', stdout='')
+        with patch.object(release.subprocess, 'run', return_value=result):
+            with self.assertRaisesRegex(RuntimeError, '403 Forbidden'):
+                release.registry_digest(release.GITEA_IMAGE, '0.1.1')
+
+    def test_registry_missing_tag_is_distinct_from_other_failures(self):
+        result = Mock(returncode=1, stderr='manifest unknown', stdout='')
+        with patch.object(release.subprocess, 'run', return_value=result):
+            self.assertIsNone(release.registry_digest(release.GITEA_IMAGE, '0.1.1'))
+
+
+class ResumeTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.cwd = Path.cwd()
+        os.chdir(self.tmp.name)
+        self.addCleanup(os.chdir, self.cwd)
+        Path('VERSION').write_text('0.1.1\n')
+        Path('downloads').mkdir()
+        for platform in release.PLATFORMS:
+            Path('downloads', release.archive_name('0.1.1', platform)).write_bytes(platform.encode())
+        Path('web/dist-demo').mkdir(parents=True)
+        Path('web/dist-demo/index.html').write_text('<html>Demo</html>')
+        with patch.object(release, 'run', return_value=SHA), patch.dict(os.environ, {'CI_RUN_ID': '42'}):
+            release.bundle('release')
+        self.original = {p.name: p.read_bytes() for p in Path('downloads').iterdir()}
+        for path in Path('downloads').iterdir():
+            path.unlink()
+        self.gitea = Mock()
+        self.gitea.assets.return_value = [{'name': name} for name in self.original]
+        self.gitea.download.side_effect = lambda asset: self.original[asset['name']]
+        self.draft = {'draft': True, 'body': 'Build run: `42`.'}
+
+    def test_resume_from_release_assets_does_not_require_old_ci_runs(self):
+        with patch.object(release, 'run', return_value=SHA):
+            release.restore(self.gitea, self.draft)
+        self.gitea.repo.assert_not_called()
+        self.gitea.upload.assert_not_called()
+        self.assertEqual({p.name: p.read_bytes() for p in Path('downloads').iterdir()}, self.original)
+
+    def test_expired_artifact_fails_instead_of_rebuilding(self):
+        self.gitea.assets.return_value = [{'name': 'build.json'}]
+        self.gitea.repo.return_value = {'artifacts': []}
+        with patch.object(release, 'run', return_value=SHA) as command:
+            with self.assertRaisesRegex(ValueError, 'refusing to rebuild'):
+                release.restore(self.gitea, self.draft)
+        self.assertFalse(any('build' in str(call) for call in command.call_args_list))
+
+    def test_interrupted_first_receipt_upload_uses_original_run(self):
+        self.gitea.assets.return_value = []
+        names = {'checksums': ['build.json', 'SHA256SUMS'], 'demo': ['demo.zip']}
+        names.update({platform.replace('/', '-'): [release.archive_name('0.1.1', platform)] for platform in release.PLATFORMS})
+        archives = {}
+        for identifier, (name, files) in enumerate(names.items(), 1):
+            data = io.BytesIO()
+            with zipfile.ZipFile(data, 'w') as archive:
+                for filename in files:
+                    archive.writestr(filename, self.original[filename])
+            archives[identifier] = (name, data.getvalue())
+        def request(path, **kwargs):
+            if path == '/actions/runs/42/artifacts':
+                return {'artifacts': [{'id': identifier, 'name': name, 'expired': False}
+                                      for identifier, (name, _) in archives.items()]}
+            return archives[int(path.split('/')[-2])][1]
+        self.gitea.repo.side_effect = request
+        with patch.object(release, 'run', return_value=SHA):
+            release.restore(self.gitea, self.draft)
+        self.gitea.upload.assert_called_once_with(self.draft, Path('downloads/build.json'))
+        self.assertEqual({p.name: p.read_bytes() for p in Path('downloads').iterdir()}, self.original)
+
+
+class DeploymentTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.directory = Path(self.tmp.name).resolve()
+        (self.directory / 'data').mkdir()
+        (self.directory / 'data/rawmark.db').write_bytes(b'database')
+        (self.directory / 'data/rawmark.db-wal').write_bytes(b'wal')
+        (self.directory / 'compose.yml').write_text('services: {}\n')
+        (self.directory / '.env').write_text(
+            f'RAWMARK_CHANNEL=development\nRAWMARK_PROJECT=rawmark-ci-deploy\n'
+            f'RAWMARK_BIND=127.0.0.1\nRAWMARK_PORT=21899\nRAWMARK_DATA_DIR={self.directory}/data\n')
+        self.image = release.GITEA_IMAGE + '@' + DIGEST
+        self.previous = {'Id': 'old-id', 'Image': 'old-image', 'State': {'Running': True},
+                         'Mounts': [{'Destination': '/data', 'Type': 'bind', 'Source': str(self.directory / 'data')}]}
+        self.image_info = {'Id': 'new-image', 'Config': {'User': '65532:65532', 'Labels': {
+            'org.opencontainers.image.revision': SHA, 'org.opencontainers.image.version': '0.1.1'}}}
+        self.commands = []
+        self.api = patch.object(deploy, 'API')
+        self.api.start().return_value.repo.return_value = {'commit': {'id': SHA}}
+        self.addCleanup(self.api.stop)
+        self.env = patch.dict(os.environ, {'GITEA_API_URL': 'https://git.example/api/v1', 'GITEA_RELEASE_TOKEN': 'private'})
+        self.env.start()
+        self.addCleanup(self.env.stop)
+
+    def command(self, *args, **kwargs):
+        self.commands.append(args)
+        if args[:3] == ('docker', 'image', 'inspect'):
+            return json.dumps([self.image_info])
+        if args[:3] == ('docker', 'run', '--rm'):
+            return 'RawMark 0.1.1'
+        if args[:3] == ('docker', 'ps', '-aq'):
+            return 'old-id'
+        if args[:3] == ('docker', 'ps', '-q'):
+            return 'old-id'
+        if args[:3] == ('docker', 'inspect', 'old-id'):
+            return json.dumps([self.previous])
+        if args[0:2] == ('docker', 'compose') and 'ps' in args:
+            return 'new-id'
+        if args[:3] == ('docker', 'inspect', 'new-id'):
+            return json.dumps([{'Image': 'new-image', 'NetworkSettings': {'Networks': {
+                'test': {'NetworkID': 'network', 'IPAddress': '127.0.0.1'}}}}])
+        return ''
+
+    def test_channel_mismatch_cannot_touch_docker(self):
+        with patch.object(deploy, 'run') as docker:
+            with self.assertRaisesRegex(ValueError, 'channel'):
+                deploy.deployment(self.directory, 'production', self.image, SHA, '0.1.1')
+            docker.assert_not_called()
+
+    def test_unresolved_pending_deployment_cannot_be_overwritten(self):
+        (self.directory / 'deployment.pending.json').write_text('{"backup":"keep"}')
+        with patch.object(deploy, 'run') as docker:
+            with self.assertRaisesRegex(ValueError, 'pending deployment'):
+                deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1')
+            docker.assert_not_called()
+
+    def test_stale_main_skips_without_stopping_existing_service(self):
+        with patch.object(deploy, 'API') as api, patch.object(deploy, 'run', side_effect=self.command):
+            api.return_value.repo.return_value = {'commit': {'id': 'c' * 40}}
+            deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1')
+        self.assertFalse(any(c[:2] == ('docker', 'stop') for c in self.commands))
+        self.assertFalse((self.directory / 'deployment.json').exists())
+
+    def test_wrong_mount_is_rejected_before_stopping(self):
+        self.previous['Mounts'][0]['Source'] = '/some-other-data'
+        with patch.object(deploy, 'run', side_effect=self.command):
+            with self.assertRaisesRegex(ValueError, 'mount differs'):
+                deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1')
+        self.assertFalse(any(c[:2] == ('docker', 'stop') for c in self.commands))
+
+    def test_backup_includes_wal_and_precedes_replacement(self):
+        response = contextlib.nullcontext(io.BytesIO(b'<html>RawMark</html>'))
+        opener = Mock()
+        opener.open.side_effect = [response, contextlib.nullcontext(io.BytesIO(b'{"code":0}'))]
+        with patch.object(deploy, 'run', side_effect=self.command), patch.object(deploy.urllib.request, 'build_opener', return_value=opener):
+            deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1')
+        receipt = json.loads((self.directory / 'deployment.json').read_text())
+        with tarfile.open(receipt['backup']) as archive:
+            self.assertEqual(archive.extractfile('data/rawmark.db-wal').read(), b'wal')
+        self.assertEqual(Path(receipt['backup']).stat().st_mode & 0o777, 0o600)
+        stop = next(i for i,c in enumerate(self.commands) if c[:2] == ('docker', 'stop'))
+        start = next(i for i,c in enumerate(self.commands) if c[:2] == ('docker', 'compose') and 'up' in c)
+        self.assertLess(stop, start)
+        self.assertIn('RAWMARK_IMAGE=' + self.image, (self.directory / '.env').read_text())
+
+    def test_failed_replacement_keeps_backup_and_does_not_downgrade_data(self):
+        def failure(*args, **kwargs):
+            if args[:2] == ('docker', 'compose') and 'up' in args:
+                raise RuntimeError('candidate failed')
+            return self.command(*args, **kwargs)
+        with patch.object(deploy, 'run', side_effect=failure), patch.object(deploy.subprocess, 'run') as stop:
+            with self.assertRaisesRegex(RuntimeError, 'candidate failed'):
+                deploy.deployment(self.directory, 'development', self.image, SHA, '0.1.1')
+        stop.assert_called_once()
+        pending = json.loads((self.directory / 'deployment.pending.json').read_text())
+        self.assertTrue(Path(pending['backup']).exists())
+        self.assertFalse(any(c[:2] == ('docker', 'start') for c in self.commands))
+        self.assertNotIn('RAWMARK_IMAGE=', (self.directory / '.env').read_text())
+
+
+class CleanupTests(unittest.TestCase):
+    def test_cleanup_rejects_broad_filter(self):
+        with patch('subprocess.check_output') as docker:
+            with self.assertRaises(ValueError):
+                cleanup('')
+            docker.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()