Spaces:
Running
Running
File size: 7,705 Bytes
89e57b7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 | """Cross-process admission, accepted-work drain and explicit durable handoff.
Only IDs/states are kept locally; this is not an auth/session registry. The hold
is also carried in the verified manifest so replacement containers require resume.
"""
import fcntl
import json
import os
import time
import uuid
from contextlib import contextmanager
from contextvars import ContextVar
from .git import BackupError
from .state import enabled, locked, state_root, write_status, public_status
_admitted = ContextVar('backup_admitted', default=False)
def _read(name, default):
path = state_root() / name
try:
return json.loads(path.read_text())
except FileNotFoundError:
return default
except (OSError, ValueError):
raise BackupError('BACKUP_MAINTENANCE_STATE_INVALID') from None
def _write(name, value):
path = state_root() / name
temp = path.with_name(path.name + '.' + uuid.uuid4().hex + '.tmp')
temp.write_text(json.dumps(value, sort_keys=True))
temp.replace(path)
def status():
if not enabled():
return {'state': 'disabled', 'quiesced': False, 'pending_tasks': 0}
with locked(state_root() / 'control.lock'):
control = _read('maintenance.json', {'state': 'active', 'quiesced': False})
tasks = _read('work.json', {})
return {**control, 'pending_tasks': sum(v == 'pending' for v in tasks.values())}
def initialize(hold=None, preserve_work=False):
with locked(state_root() / 'control.lock'):
if not preserve_work:
_write('work.json', {})
prior = _read('maintenance.json', {}) if preserve_work else {}
if prior.get('quiesced'):
return
_write('maintenance.json', {'state': 'quiesced' if hold else 'active',
'quiesced': bool(hold), 'sealed': bool(hold), 'handoff': hold or {}})
def hold_manifest():
value = status()
return value.get('handoff', {}) if value.get('quiesced') else {}
@contextmanager
def admission():
"""Hold a shared lease until an admitted mutation request has finished."""
if not enabled() or _admitted.get():
yield
return
path = state_root() / 'admission.lock'
path.parent.mkdir(parents=True, exist_ok=True)
with path.open('a+') as lease:
if os.fstat(lease.fileno()).st_uid == os.geteuid():
os.fchmod(lease.fileno(), 0o660)
try:
fcntl.flock(lease, fcntl.LOCK_SH | fcntl.LOCK_NB)
except BlockingIOError:
raise BackupError('BACKUP_MAINTENANCE_QUIESCED') from None
try:
with locked(state_root() / 'control.lock'):
if _read('maintenance.json', {}).get('quiesced'):
raise BackupError('BACKUP_MAINTENANCE_QUIESCED')
token = _admitted.set(True)
try:
yield
finally:
_admitted.reset(token)
finally:
fcntl.flock(lease, fcntl.LOCK_UN)
def register_task(task_id, parent=None):
if not enabled():
return
with locked(state_root() / 'control.lock'):
control = _read('maintenance.json', {})
tasks = _read('work.json', {})
accepted_parent = parent is not None and tasks.get(parent) == 'pending'
if control.get('sealed') or (control.get('quiesced') and not (_admitted.get() or accepted_parent)):
raise BackupError('BACKUP_MAINTENANCE_QUIESCED')
tasks[task_id] = 'pending'
_write('work.json', tasks)
def begin_task(task_id):
"""Count running deliveries separately, including broker redeliveries."""
with locked(state_root() / 'control.lock'):
tasks = _read('work.json', {})
if not task_id or tasks.get(task_id) != 'pending':
raise BackupError('BACKUP_UNREGISTERED_TASK')
execution = uuid.uuid4().hex
tasks[execution] = 'pending'
_write('work.json', tasks)
return execution
def finish_task(task_id, execution=None):
if enabled():
with locked(state_root() / 'control.lock'):
tasks = _read('work.json', {})
tasks.pop(task_id, None)
tasks.pop(execution, None)
_write('work.json', tasks)
def quiesce():
if not enabled():
raise BackupError('BACKUP_DISABLED')
with locked(state_root() / 'control.lock'):
control = _read('maintenance.json', {})
if not control.get('quiesced'):
control = {'state': 'draining', 'quiesced': True, 'sealed': False,
'handoff': {'id': uuid.uuid4().hex, 'quiesced': True}}
_write('maintenance.json', control)
return status()
def fail(code):
quiesce()
with locked(state_root() / 'control.lock'):
control = _read('maintenance.json', {})
control.update(state='degraded', error=code)
_write('maintenance.json', control)
write_status(state='degraded', error=code)
def _seal(timeout):
deadline = time.monotonic() + timeout
path = state_root() / 'admission.lock'
with path.open('a+') as lease:
while True:
try:
fcntl.flock(lease, fcntl.LOCK_EX | fcntl.LOCK_NB)
break
except BlockingIOError:
if time.monotonic() >= deadline:
raise BackupError('BACKUP_DRAIN_TIMEOUT')
time.sleep(.05)
try:
while True:
with locked(state_root() / 'control.lock'):
tasks = _read('work.json', {})
if not any(v == 'pending' for v in tasks.values()):
control = _read('maintenance.json', {})
control.update(state='quiesced', sealed=True)
_write('maintenance.json', control)
return
if time.monotonic() >= deadline:
raise BackupError('BACKUP_DRAIN_TIMEOUT')
time.sleep(.05)
finally:
fcntl.flock(lease, fcntl.LOCK_UN)
def checkpoint(manager, timeout=120):
with locked(state_root() / 'barrier.lock'):
quiesce()
try:
_seal(timeout)
head, _ = manager.backup()
manager.verify_checkpoint(head)
if public_status().get('backup_dirty'):
raise BackupError('BACKUP_DRAIN_WRITES_DETECTED')
with locked(state_root() / 'control.lock'):
control = _read('maintenance.json', {})
control.update(state='durable', checkpoint=head, remote_verified=True)
control.pop('error', None)
_write('maintenance.json', control)
return {'state': 'durable', 'checkpoint': head, 'dirty': False,
'remote_verified': True, 'quiesced': True}
except Exception as exc:
code = str(exc) if isinstance(exc, BackupError) else 'BACKUP_BARRIER_FAILED'
fail(code)
raise BackupError(code) from None
def resume(manager):
with locked(state_root() / 'barrier.lock'):
if not status().get('quiesced'):
return status()
try:
_seal(0)
# Publish release while local admission is still closed. Failure leaves
# both this container and replacements safely held.
head, _ = manager.backup(maintenance={})
manager.verify_checkpoint(head)
initialize()
return {'state': 'active', 'quiesced': False, 'checkpoint': head}
except Exception as exc:
code = str(exc) if isinstance(exc, BackupError) else 'BACKUP_RESUME_FAILED'
fail(code)
raise BackupError(code) from None
|