386 lines
23 KiB
Python
Executable File
386 lines
23 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""WJ-021: real packaged application and H2, production deployment file operations."""
|
|
import argparse
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import zipfile
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
spec = importlib.util.spec_from_file_location('deployment', ROOT / 'scripts/deploy-update-remote.py')
|
|
deployment = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(deployment)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('jar', type=Path)
|
|
parser.add_argument('--report', type=Path, required=True)
|
|
parser.add_argument('--failures', type=int, choices=[1, 2], default=1)
|
|
parser.add_argument('--fail-rollback', action='store_true')
|
|
parser.add_argument('--failure-worker', action='store_true', help=argparse.SUPPRESS)
|
|
parser.add_argument('--upload-only', action='store_true')
|
|
args = parser.parse_args()
|
|
jar = args.jar.resolve(strict=True)
|
|
java = (Path(os.environ['JAVA_HOME']) / 'bin/java').resolve(strict=True)
|
|
args.report.parent.mkdir(parents=True, exist_ok=True)
|
|
args.report.unlink(missing_ok=True)
|
|
if args.fail_rollback and args.failures != 1:
|
|
parser.error('--fail-rollback requires one candidate')
|
|
if args.upload_only and (args.fail_rollback or args.failures != 1):
|
|
parser.error('--upload-only requires one candidate and no rollback failure')
|
|
if args.fail_rollback and not args.failure_worker:
|
|
output = args.report.with_name(args.report.stem + '-failure.log')
|
|
with output.open('wb') as log:
|
|
worker = subprocess.Popen([sys.executable, str(Path(__file__).resolve()), str(jar),
|
|
'--report', str(args.report), '--fail-rollback', '--failure-worker'],
|
|
stdout=log, stderr=subprocess.STDOUT, start_new_session=True)
|
|
try:
|
|
# Above the readiness waits this worker performs, or raising those would only
|
|
# move the failure here, where it reads as a hung fixture instead of a slow host.
|
|
worker.wait(timeout=600)
|
|
finally:
|
|
if worker.poll() is None:
|
|
os.killpg(worker.pid, signal.SIGTERM)
|
|
try:
|
|
worker.wait(timeout=5)
|
|
finally:
|
|
# The worker and every JVM share this private process group.
|
|
try:
|
|
os.killpg(worker.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
worker.wait()
|
|
assert worker.returncode != 0, 'Failed rollback worker incorrectly returned success'
|
|
failure_log = output.read_text()
|
|
assert 'Expected candidate migration failure' in failure_log
|
|
assert 'Expected rollback readiness failure' in failure_log
|
|
report = json.loads(args.report.read_text())
|
|
assert report['result'] == 'EXPECTED_FAILURE'
|
|
assert report['maintenanceRetained'] and report['journalRetained'] and report['rollbackStopped']
|
|
report.update(result='PASS', failureExitCode=worker.returncode, bothFailuresLogged=True)
|
|
args.report.write_text(json.dumps(report, indent=2) + '\n')
|
|
print(json.dumps(report))
|
|
return
|
|
logs = args.report.parent / (args.report.stem + '-logs')
|
|
logs.mkdir(exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix='werkjournal-rollback-') as temporary:
|
|
home = Path(temporary)
|
|
app = home / 'opt/werkjournal'
|
|
app.mkdir(parents=True)
|
|
with zipfile.ZipFile(jar) as archive:
|
|
h2_member, = [name for name in archive.namelist() if name.startswith('BOOT-INF/lib/h2-')]
|
|
h2 = home / 'h2.jar'
|
|
h2.write_bytes(archive.read(h2_member))
|
|
properties = archive.read('META-INF/build-info.properties').decode()
|
|
version = next(line.split('=', 1)[1] for line in properties.splitlines() if line.startswith('build.version='))
|
|
old = 'a' * 40
|
|
candidates = [letter * 40 for letter in 'bc'[:args.failures]]
|
|
for commit in [old, *candidates]:
|
|
release = app / 'releases' / commit
|
|
release.mkdir(parents=True)
|
|
target = release / 'werkjournal.jar'
|
|
if commit == old:
|
|
shutil.copyfile(jar, target)
|
|
else:
|
|
with zipfile.ZipFile(jar) as source, zipfile.ZipFile(target, 'w') as dest:
|
|
for entry in source.infolist():
|
|
data = source.read(entry.filename)
|
|
if entry.filename == 'BOOT-INF/classes/db/changelog.sql':
|
|
data += b'\n--changeset rollback-test:partial-failure dbms:h2 runInTransaction:false\nCREATE TABLE rollback_probe (id INTEGER PRIMARY KEY);\nINSERT INTO rollback_probe VALUES (1);\nSELECT * FROM intentionally_missing_rollback_table;\n'
|
|
dest.writestr(entry, data)
|
|
digest = deployment.checksum(target)
|
|
manifest = dict(schemaVersion=1, runtime='java-25', version=version, commit=commit,
|
|
sha256=digest, files={'werkjournal.jar': digest}, ancestors=[] if commit == old else [old])
|
|
(release / 'manifest.json').write_text(json.dumps(manifest))
|
|
(app / 'current').symlink_to(Path('releases') / old)
|
|
shutil.copyfile(app / 'current/manifest.json', app / 'deployed.json')
|
|
packages = {}
|
|
for candidate in candidates:
|
|
package = home / f'{candidate}.tar'
|
|
with tarfile.open(package, 'w') as archive:
|
|
for name in ['werkjournal.jar', 'manifest.json']:
|
|
archive.add(app / 'releases' / candidate / name, arcname=name)
|
|
packages[candidate] = package
|
|
# stage() must actually unpack and validate the candidate.
|
|
shutil.rmtree(app / 'releases' / candidate)
|
|
env = {'PATH': '/usr/bin:/bin', 'LANG': 'C.UTF-8', 'HOME': str(home),
|
|
'WERKJOURNAL_GUEST_MODE': 'true', 'WERKJOURNAL_PUSH_ENABLED': 'false'}
|
|
trap = home / 'sendmail'
|
|
trap.write_text('#!/bin/sh\ntouch "' + str(home / 'unexpected-mail') + '"\nexit 1\n')
|
|
trap.chmod(0o700)
|
|
env['WERKJOURNAL_SENDMAIL'] = str(trap)
|
|
with socket.socket() as sock:
|
|
sock.bind(('127.0.0.1', 0))
|
|
port = sock.getsockname()[1]
|
|
base = f'http://127.0.0.1:{port}'
|
|
|
|
def sql(statement):
|
|
result = subprocess.run([str(java), '-cp', str(h2), 'org.h2.tools.Shell', '-url',
|
|
'jdbc:h2:file:' + str(app / 'data/werkjournal'), '-user', 'sa', '-password', '',
|
|
'-sql', statement], env=env, capture_output=True, text=True, timeout=60, check=True)
|
|
if 'Error:' in result.stdout:
|
|
raise AssertionError(result.stdout)
|
|
return result.stdout
|
|
|
|
class LocalDeployment(deployment.Deployment):
|
|
process = None
|
|
log = None
|
|
starts = 0
|
|
migration_failed = False
|
|
matched_backup = False
|
|
ready_with_maintenance = False
|
|
blocked_port = None
|
|
|
|
def preflight(self, check_data=True):
|
|
# systemd configuration is covered by operations tests; this fixture owns its process.
|
|
assert self.home == home and self.app == app
|
|
assert (self.app / 'data/werkjournal.mv.db').is_file()
|
|
|
|
def stop(self):
|
|
if self.process is not None:
|
|
if self.process.poll() is None:
|
|
self.process.terminate()
|
|
try:
|
|
# Generous, because this runs beside the browser suites on a shared
|
|
# host: the same release closed its database in 0.2 s when it had the
|
|
# machine to itself. Only a starved host waits anywhere near this.
|
|
self.process.wait(timeout=60)
|
|
except subprocess.TimeoutExpired:
|
|
self.process.kill()
|
|
self.process.wait(timeout=20)
|
|
raise AssertionError('Application failed to close its database gracefully')
|
|
assert self.process.poll() is not None
|
|
self.log.close()
|
|
self.process = None
|
|
|
|
def start(self):
|
|
if self.migration_failed:
|
|
assert (self.app / 'current').resolve().name == old
|
|
state = json.loads(self.pending.read_text())
|
|
inventory = json.loads((self.app / 'rollbacks' / state['backup'] / 'inventory.json').read_text())
|
|
assert all(deployment.checksum(self.app / 'data' / name) == digest for name, digest in inventory.items())
|
|
self.matched_backup = True
|
|
if args.fail_rollback:
|
|
self.blocked_port = socket.socket()
|
|
self.blocked_port.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
self.blocked_port.bind(('127.0.0.1', port))
|
|
self.blocked_port.listen()
|
|
self.starts += 1
|
|
self.logpath = logs / f'start-{self.starts}.log'
|
|
self.log = self.logpath.open('wb')
|
|
self.process = subprocess.Popen([str(java), '-Xmx256m', '-XX:ActiveProcessorCount=2',
|
|
'-jar', str(self.app / 'current/werkjournal.jar'), '--server.address=127.0.0.1',
|
|
f'--server.port={port}'], cwd=self.app, env=env, stdout=self.log, stderr=subprocess.STDOUT)
|
|
|
|
def ready(self, expected):
|
|
# Three minutes for a start that takes one to three seconds on an idle machine.
|
|
# On 2026-10-12 this host needed 17.1 s for one start and had not finished the
|
|
# migration of the next after 25 s, with Firefox and its own applications on the
|
|
# same three threads; the branch build failed with nothing wrong in the release.
|
|
# The proof is unchanged: the application must become ready, and this only grants
|
|
# a starved host the time to get there.
|
|
deadline = time.monotonic() + 180
|
|
while time.monotonic() < deadline:
|
|
if self.process.poll() is not None:
|
|
if args.fail_rollback and self.migration_failed and expected['commit'] == old:
|
|
assert self.process.returncode != 0
|
|
assert 'already in use' in self.logpath.read_text().lower()
|
|
raise RuntimeError('Expected rollback readiness failure')
|
|
assert expected['commit'] in candidates and self.process.returncode != 0
|
|
assert 'intentionally_missing_rollback_table' in self.logpath.read_text()
|
|
assert 'PARTIAL_MIGRATION_CONFIRMED' in sql("SELECT 'PARTIAL_MIGRATION_CONFIRMED' FROM rollback_probe WHERE id=1")
|
|
self.migration_failed = True
|
|
raise RuntimeError('Expected candidate migration failure')
|
|
try:
|
|
with urllib.request.urlopen(base + '/api/v1/info', timeout=2) as response:
|
|
assert json.load(response) == {'application': 'werkjournal', 'version': version}
|
|
with urllib.request.urlopen(base + '/?v-r=init&location=&query=', timeout=3) as response:
|
|
assert isinstance(json.load(response).get('appConfig'), dict)
|
|
if self.migration_failed:
|
|
assert self.flag.exists()
|
|
try:
|
|
urllib.request.urlopen(self.url, timeout=2)
|
|
raise AssertionError('Public access opened before readiness')
|
|
except urllib.error.HTTPError as error:
|
|
assert error.code == 503
|
|
self.ready_with_maintenance = True
|
|
return
|
|
except (urllib.error.URLError, TimeoutError):
|
|
time.sleep(.2)
|
|
raise AssertionError('Application readiness timed out')
|
|
|
|
runner = LocalDeployment(home)
|
|
runner.site.mkdir(parents=True)
|
|
|
|
class Gateway(BaseHTTPRequestHandler):
|
|
def do_GET(self):
|
|
body = runner.flag.read_bytes() if runner.flag.exists() else b'public access available'
|
|
self.send_response(503 if runner.flag.exists() else 200)
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *unused):
|
|
pass
|
|
|
|
gateway = ThreadingHTTPServer(('127.0.0.1', 0), Gateway)
|
|
threading.Thread(target=gateway.serve_forever, daemon=True).start()
|
|
runner.url = f'http://127.0.0.1:{gateway.server_port}'
|
|
try:
|
|
runner.start()
|
|
runner.ready({'commit': old})
|
|
runner.stop()
|
|
ids = [f'10000000-0000-0000-0000-{n:012d}' for n in range(1, 5)]
|
|
user, customer, project, entry = ids
|
|
sql(f"INSERT INTO app_user(id,display_name) VALUES('{user}','Rollback fixture');"
|
|
f"INSERT INTO customer(id,owner_id,name) VALUES('{customer}','{user}','Rollback customer');"
|
|
f"INSERT INTO work_project(id,owner_id,customer_id,name) VALUES('{project}','{user}','{customer}','Rollback project');"
|
|
f"INSERT INTO work_entry(id,owner_id,customer_id,project_id,work_date,hours,description) VALUES('{entry}','{user}','{customer}','{project}',DATE '2026-09-01',3,'Preserve this booking');")
|
|
runner.start()
|
|
runner.ready({'commit': old})
|
|
if args.upload_only:
|
|
client = home / 'client'
|
|
(client / 'tools').mkdir(parents=True)
|
|
(client / 'scripts').mkdir()
|
|
for name in ['tools/remote', 'scripts/lib-env.sh', 'scripts/deploy-prototype.py',
|
|
'scripts/deploy-update-remote.py']:
|
|
shutil.copy2(ROOT / name, client / name)
|
|
artifact = client / 'artifact'
|
|
artifact.mkdir()
|
|
with tarfile.open(packages[candidates[0]]) as archive:
|
|
for name in ['manifest.json', 'werkjournal.jar']:
|
|
with archive.extractfile(name) as source, (artifact / name).open('wb') as target:
|
|
shutil.copyfileobj(source, target)
|
|
transport = home / 'transport'
|
|
transport.mkdir()
|
|
captured = transport / 'upload.tar'
|
|
ssh = transport / 'ssh'
|
|
ssh.write_text("#!/usr/bin/env python3\nimport pathlib, shlex, sys\n"
|
|
"assert sys.argv[1:4] == ['-o', 'BatchMode=yes', 'fixture@invalid']\n"
|
|
"assert len(sys.argv) == 5 and shlex.split(sys.argv[4])[-1] == '--upload'\n"
|
|
"target = pathlib.Path(__file__).with_name('upload.tar')\n"
|
|
"with target.open('xb') as output:\n"
|
|
" import shutil\n shutil.copyfileobj(sys.stdin.buffer, output)\n")
|
|
ssh.chmod(0o700)
|
|
client_env = dict(env, PATH=str(transport) + ':/usr/bin:/bin', BACKEND_SSH='fixture@invalid')
|
|
active_pid = runner.process.pid
|
|
selected = (app / 'current').readlink()
|
|
deployed_bytes = (app / 'deployed.json').read_bytes()
|
|
subprocess.run([str(client / 'tools/remote'), 'backend', 'upload', '-y',
|
|
'--artifact', str(artifact)], env=client_env, cwd=client,
|
|
check=True, capture_output=True, timeout=30)
|
|
# Execute the real server upload path on the exact bytes sent by remote.
|
|
with captured.open('rb') as stream:
|
|
uploaded = runner.upload(stream)
|
|
assert uploaded['commit'] == candidates[0]
|
|
assert deployment.manifest(app / 'releases' / candidates[0]) == json.loads((artifact / 'manifest.json').read_text())
|
|
assert runner.process.pid == active_pid and runner.process.poll() is None
|
|
assert (app / 'current').readlink() == selected
|
|
assert (app / 'deployed.json').read_bytes() == deployed_bytes
|
|
assert not runner.flag.exists() and not runner.pending.exists()
|
|
# Use the actual restart control with only the process adapter replaced.
|
|
controls = dict(vars(deployment), __name__='service_control_fixture')
|
|
exec(compile((ROOT / 'scripts/service-control-remote.py').read_text(),
|
|
'service-control-remote.py', 'exec'), controls)
|
|
assert controls['control'](runner, 'restart') == 0
|
|
assert runner.process.pid != active_pid and runner.process.poll() is None
|
|
assert (app / 'current').readlink() == selected
|
|
assert (app / 'deployed.json').read_bytes() == deployed_bytes
|
|
assert not runner.flag.exists() and not runner.pending.exists()
|
|
runner.stop()
|
|
assert 'BOOKING_PRESERVED' in sql(f"SELECT 'BOOKING_PRESERVED' FROM work_entry WHERE id='{entry}' AND hours=3")
|
|
assert 'CANDIDATE_NOT_RUN' in sql("SELECT 'CANDIDATE_NOT_RUN' WHERE NOT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_name='ROLLBACK_PROBE')")
|
|
assert not (home / 'unexpected-mail').exists()
|
|
report = dict(result='PASS', uploadedCommit=candidates[0], activeCommit=old,
|
|
uploadKeptProcess=True, restartKeptRelease=True, bookingPreserved=True,
|
|
candidateMigrationNotRun=True, mailInvocations=0, pushDisabled=True)
|
|
args.report.write_text(json.dumps(report, indent=2) + '\n')
|
|
print(json.dumps(report))
|
|
return
|
|
rounds = []
|
|
failed_rollback = None
|
|
for index, candidate in enumerate(candidates):
|
|
runner.migration_failed = runner.matched_backup = runner.ready_with_maintenance = False
|
|
with packages[candidate].open('rb') as archive:
|
|
try:
|
|
runner.deploy(archive)
|
|
raise AssertionError('Failed migration was reported as successful')
|
|
except RuntimeError as error:
|
|
if args.fail_rollback:
|
|
assert str(error) == 'Expected rollback readiness failure'
|
|
assert str(error.__context__) == 'Expected candidate migration failure'
|
|
failed_rollback = error
|
|
else:
|
|
assert str(error) == 'Expected candidate migration failure'
|
|
if args.fail_rollback:
|
|
assert runner.migration_failed and runner.matched_backup
|
|
assert runner.process is None, 'Failed rollback left a running process'
|
|
assert runner.flag.exists() and runner.pending.exists()
|
|
state = json.loads(runner.pending.read_text())
|
|
assert state['old']['commit'] == old and state['new']['commit'] == candidate
|
|
assert state['phase'] == 'switched'
|
|
assert (app / 'current').resolve().name == old
|
|
assert 'BOOKING_PRESERVED' in sql(f"SELECT 'BOOKING_PRESERVED' FROM work_entry WHERE id='{entry}' AND hours=3")
|
|
try:
|
|
urllib.request.urlopen(runner.url, timeout=2)
|
|
raise AssertionError('Failed rollback exposed the application')
|
|
except urllib.error.HTTPError as error:
|
|
assert error.code == 503 and 'Wartungsmodus' in error.read().decode()
|
|
break
|
|
assert runner.migration_failed and runner.matched_backup and runner.ready_with_maintenance
|
|
assert not runner.flag.exists() and not runner.pending.exists()
|
|
assert json.loads((app / 'deployed.json').read_text())['commit'] == old
|
|
assert (app / 'current').resolve().name == old
|
|
with urllib.request.urlopen(runner.url, timeout=2) as response:
|
|
assert response.status == 200
|
|
runner.stop()
|
|
assert 'BOOKING_PRESERVED' in sql(f"SELECT 'BOOKING_PRESERVED' FROM work_entry WHERE id='{entry}' AND hours=3 AND description='Preserve this booking'")
|
|
assert 'SCHEMA_RESTORED' in sql("SELECT 'SCHEMA_RESTORED' WHERE NOT EXISTS(SELECT 1 FROM information_schema.tables WHERE table_name='ROLLBACK_PROBE')")
|
|
assert 'HISTORY_RESTORED' in sql("SELECT 'HISTORY_RESTORED' WHERE NOT EXISTS(SELECT 1 FROM databasechangelog WHERE author='rollback-test')")
|
|
if index:
|
|
assert 'INTERMEDIATE_BOOKING_PRESERVED' in sql("SELECT 'INTERMEDIATE_BOOKING_PRESERVED' FROM work_entry WHERE description='Written after first rollback' AND hours=2")
|
|
rounds.append(dict(candidate=candidate, restoredCommit=old, bookingPreserved=True,
|
|
intermediateBookingPreserved=index > 0))
|
|
if index + 1 < len(candidates):
|
|
# New data after B's rollback must survive C as well. Reusing
|
|
# the first deployment's snapshot would lose this entry.
|
|
sql(f"INSERT INTO work_entry(id,owner_id,customer_id,project_id,work_date,hours,description) VALUES('10000000-0000-0000-0000-000000000005','{user}','{customer}','{project}',DATE '2026-09-02',2,'Written after first rollback')")
|
|
runner.migration_failed = False
|
|
runner.start()
|
|
runner.ready({'commit': old})
|
|
assert not (home / 'unexpected-mail').exists()
|
|
if args.fail_rollback:
|
|
args.report.write_text(json.dumps(dict(result='EXPECTED_FAILURE', maintenanceRetained=True,
|
|
journalRetained=True, rollbackStopped=True, bookingPreserved=True,
|
|
mailInvocations=0, pushDisabled=True)) + '\n')
|
|
assert failed_rollback is not None
|
|
raise failed_rollback
|
|
report = dict(result='PASS', partialMigration=True, candidateStopped=True,
|
|
matchingBackupRestored=True, bookingPreserved=True, schemaRestored=True,
|
|
maintenanceUntilReady=True, mailInvocations=0, pushDisabled=True, rounds=rounds)
|
|
args.report.write_text(json.dumps(report, indent=2) + '\n')
|
|
print(json.dumps(report))
|
|
finally:
|
|
runner.stop()
|
|
if runner.blocked_port is not None:
|
|
runner.blocked_port.close()
|
|
gateway.shutdown()
|
|
gateway.server_close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|