"""Rebind installed Worker roles through the existing control-plane CAS.""" from __future__ import annotations import hashlib import json import os from pathlib import Path import re import tempfile from serviceification_d3_engine import DeployError, canonical_bytes, read_worker_ownership from standalone_l0_workers import configured_workers def atomic_role(path, value): descriptor, temporary = tempfile.mkstemp(prefix=path.name + '.', dir=path.parent) try: with os.fdopen(descriptor, 'wb') as stream: stream.write(canonical_bytes(value)) stream.flush() os.fsync(stream.fileno()) os.fchmod(stream.fileno(), 0o400) os.replace(temporary, path) finally: if os.path.exists(temporary): os.unlink(temporary) def upgrade_worker_bindings(*, state_root, target_digest, target_source, envv, activate, read_ownership=read_worker_ownership, write_role=atomic_role): """Caller holds the stopped L0 boundary across this call and replacement. Role files retain all installation facts. The DB owns the epoch and release binding. A retry after CAS succeeds but before the role is replaced accepts only this exact deterministic operation's successor, never another writer. Preflight every role before advancing the first ownership epoch. """ if not re.fullmatch(r'sha256:[0-9a-f]{64}', target_digest): raise DeployError('target Worker digest is malformed') if not isinstance(target_source, str) or not target_source.strip(): raise DeployError('target Worker source revision is empty') roles = state_root / 'config/workers' if not roles.is_dir(): raise DeployError('L0 upgrade requires installed Worker roles') ownership = read_ownership(envv) plans = [] sources_by_digest = {} for stem in configured_workers(state_root): path = roles / (stem + '.json') role = json.loads(path.read_text()) family = 'runtime.' + stem + '.v1' if role.get('work_family') != family: raise DeployError('Worker role family disagrees with its installed path') digest = role.get('worker_release_digest', '') source = role.get('worker_source_revision', '') epoch = role.get('ownership_epoch') if (not re.fullmatch(r'sha256:[0-9a-f]{64}', digest) or not source or isinstance(epoch, bool) or not str(epoch).isdecimal() or int(epoch) < 1): raise DeployError('installed Worker binding is malformed') if sources_by_digest.setdefault(digest, source) != source: raise DeployError('installed Worker source revisions disagree') current = ownership.get(family) if current is None or not current.active or current.deployment_id != role.get('deployment_id'): raise DeployError('installed Worker no longer owns its active deployment') if digest == target_digest: if (source != target_source or current.worker_release_digest != digest or current.ownership_epoch != int(epoch)): raise DeployError('target Worker binding disagrees with authority') continue request = { 'schema': 'nexuscrm.runtime-work-ownership-activation/v1', 'workFamily': family, 'deploymentId': current.deployment_id, 'workerReleaseDigest': target_digest, 'expectedOwnershipEpoch': str(epoch), } request['operationId'] = 'l0-upgrade-' + hashlib.sha256(canonical_bytes({ **request, 'previousDigest': digest, 'targetSource': target_source, })).hexdigest() expected = { 'schema': 'nexuscrm.runtime-work-ownership-activation-result/v1', 'operationId': request['operationId'], 'workFamily': family, 'deploymentId': current.deployment_id, 'workerReleaseDigest': target_digest, 'ownershipEpoch': str(int(epoch) + 1), } predecessor = current.worker_release_digest == digest and current.ownership_epoch == int(epoch) resumed = (current.worker_release_digest == target_digest and current.ownership_epoch == int(epoch) + 1 and current.operation_id == request['operationId']) if not (predecessor or resumed): raise DeployError('Worker ownership changed outside this L0 upgrade') plans.append((path, role, request, expected, predecessor)) if len(set(sources_by_digest) - {target_digest}) > 1: raise DeployError('installed Worker predecessor releases disagree') for path, role, request, expected, predecessor in plans: if predecessor and activate(path.stem, request) != expected: raise DeployError('Worker activation result disagrees with selected L0 release') current = read_ownership(envv).get(request['workFamily']) if (current is None or not current.active or current.deployment_id != request['deploymentId'] or current.worker_release_digest != target_digest or current.ownership_epoch != int(expected['ownershipEpoch']) or current.operation_id != request['operationId']): raise DeployError('Worker activation readback disagrees with authority') write_role(path, {**role, 'worker_release_digest': target_digest, 'worker_source_revision': target_source, 'ownership_epoch': int(expected['ownershipEpoch'])}) return len(plans) def catalog_activator(*, image, database_env, evidence, network, execute): evidence.mkdir(parents=True, mode=0o700, exist_ok=True) def activate(stem, request): config = evidence / (stem + '-request.json') output = evidence / (stem + '-result.json') atomic_role(config, request) output.unlink(missing_ok=True) execute(('docker', 'run', '--rm', '--network', network, '--user', f'{os.getuid()}:{os.getgid()}', '--env-file', str(database_env), '--mount', f'type=bind,src={config},dst=/config.json,readonly', '--mount', f'type=bind,src={evidence},dst=/output', image, 'work-ownership', '-config', '/config.json', '-output', '/output/' + output.name)) return json.loads(output.read_text()) return activate def main(): import argparse import subprocess from common import load_env_file parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--state-root', type=Path, required=True) parser.add_argument('--worker-image', required=True) parser.add_argument('--source', required=True) parser.add_argument('--runtime-container', required=True) parser.add_argument('--worker-container-prefix', required=True) args = parser.parse_args() if not re.fullmatch(r'[0-9a-f]{40}', args.source): raise DeployError('remote L0 source must be an exact commit SHA') if not re.fullmatch(r'ghcr\.io/pingcap-inc/nexuscrm-runtime-worker@sha256:[0-9a-f]{64}', args.worker_image): raise DeployError('remote Worker image must use the exact official repository and digest') image = (args.state_root / 'config/catalog-controlplane-image').read_text().strip() if not re.fullmatch(r'ghcr\.io/pingcap-inc/nexuscrm-controlplane-catalog-reconcile@sha256:[0-9a-f]{64}', image): raise DeployError('remote Catalog image must use the exact official repository and digest') # The workflow pulls the selected Worker; prove it exists and matches the # declared source before accessing database credentials or changing bindings. worker_source = subprocess.run(('docker', 'image', 'inspect', '--format', '{{index .Config.Labels "org.opencontainers.image.revision"}}', args.worker_image), check=True, capture_output=True, text=True, timeout=30).stdout.strip() if worker_source != args.source: raise DeployError('remote Worker image source differs from the selected L0 source') names = [args.runtime_container] + [args.worker_container_prefix + stem for stem in configured_workers(args.state_root)] # Inspect the inventory once; missing containers are already quiesced. running = subprocess.run(('docker', 'ps', '--format', '{{.Names}}'), check=True, capture_output=True, text=True, timeout=30) if set(names) & set(running.stdout.splitlines()): raise DeployError('stop L0 API and all Workers before upgrading their ownership') database_env = args.state_root / 'config/runtime.env' envv = load_env_file(database_env) activate = catalog_activator(image=image, database_env=database_env, evidence=args.state_root / 'data/l0-worker-upgrade', network='host', execute=lambda command: subprocess.run(command, check=True, timeout=300)) count = upgrade_worker_bindings(state_root=args.state_root, target_digest=args.worker_image.split('@')[1], target_source=args.source, envv=envv, activate=activate) print(f'L0 Worker ownership upgrade complete: {count} bindings advanced or recovered') if __name__ == '__main__': main()