159 lines
4.8 KiB
Python
159 lines
4.8 KiB
Python
"""One process/thread-safe transaction boundary for board state and queue outbox.
|
|
|
|
Models never access this module. A committed outbox is replayed before the
|
|
next transaction; consumers may only observe jobs after their state commits.
|
|
"""
|
|
import copy
|
|
import fcntl
|
|
import functools
|
|
import json
|
|
import os
|
|
import threading
|
|
import uuid
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
_lock = threading.RLock()
|
|
_local = threading.local()
|
|
|
|
|
|
def atomic_json(path, value):
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
|
|
try:
|
|
with temporary.open('x', encoding='utf-8') as output:
|
|
os.chmod(temporary, 0o600)
|
|
json.dump(value, output, ensure_ascii=False, indent=2)
|
|
output.write('\n')
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
temporary.replace(path)
|
|
fd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
finally:
|
|
temporary.unlink(missing_ok=True)
|
|
|
|
|
|
def _read(path):
|
|
try:
|
|
data = json.loads(path.read_text(encoding='utf-8'))
|
|
except FileNotFoundError:
|
|
return {'cards': {}}
|
|
if not isinstance(data, dict) or not isinstance(data.get('cards', {}), dict):
|
|
raise ValueError('Board state is invalid; restore a verified backup before continuing.')
|
|
data.setdefault('cards', {})
|
|
return data
|
|
|
|
|
|
def load():
|
|
if getattr(_local, 'depth', 0):
|
|
return copy.deepcopy(_local.data)
|
|
return _read(Path(os.environ['BOARD_STATE_FILE']))
|
|
|
|
|
|
def save(data):
|
|
if getattr(_local, 'depth', 0):
|
|
_local.data = copy.deepcopy(data)
|
|
_local.dirty = True
|
|
else:
|
|
raise RuntimeError('Board state writes require a transaction')
|
|
|
|
|
|
def write_payload(path, payload):
|
|
if getattr(_local, 'depth', 0):
|
|
_local.outbox[str(path)] = copy.deepcopy(payload)
|
|
else:
|
|
atomic_json(path, payload)
|
|
|
|
|
|
def _flush(path, data):
|
|
pending = data.get('_pending_writes', {})
|
|
if not pending:
|
|
return
|
|
for target, payload in pending.items():
|
|
# Claims and publication share a queue-local lock. Durable claim
|
|
# markers close the crash window between publication and receipt.
|
|
receipt = path.parent / 'outbox-receipts' / (Path(target).name + '.json')
|
|
destination = Path(target)
|
|
with delivery_lock(destination.parent):
|
|
if not receipt.exists():
|
|
if not (destination.parent / '.claimed' / destination.name).exists():
|
|
atomic_json(destination, payload)
|
|
atomic_json(receipt, {'target': target})
|
|
data.pop('_pending_writes', None)
|
|
atomic_json(path, data)
|
|
|
|
|
|
@contextmanager
|
|
def delivery_lock(directory):
|
|
directory.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
with (directory / '.delivery.lock').open('a+') as handle:
|
|
fcntl.flock(handle, fcntl.LOCK_EX)
|
|
yield
|
|
|
|
|
|
def claim_job(source, target):
|
|
with delivery_lock(source.parent):
|
|
if not source.exists():
|
|
return False
|
|
atomic_json(source.parent / '.claimed' / source.name, {'claimed': True})
|
|
source.replace(target)
|
|
return True
|
|
|
|
|
|
@contextmanager
|
|
def transaction():
|
|
# Some pure helpers/tests have no state configured.
|
|
if not os.environ.get('BOARD_STATE_FILE'):
|
|
yield
|
|
return
|
|
with _lock:
|
|
if getattr(_local, 'depth', 0):
|
|
_local.depth += 1
|
|
try:
|
|
yield
|
|
finally:
|
|
_local.depth -= 1
|
|
return
|
|
path = Path(os.environ['BOARD_STATE_FILE'])
|
|
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
with path.with_suffix('.transaction.lock').open('a+') as handle:
|
|
fcntl.flock(handle, fcntl.LOCK_EX)
|
|
data = _read(path)
|
|
_flush(path, data)
|
|
_local.depth, _local.data = 1, data
|
|
_local.dirty, _local.outbox = False, {}
|
|
try:
|
|
yield
|
|
if _local.dirty or _local.outbox:
|
|
if _local.outbox:
|
|
_local.data['_pending_writes'] = _local.outbox
|
|
atomic_json(path, _local.data)
|
|
_flush(path, _local.data)
|
|
finally:
|
|
_local.depth = 0
|
|
|
|
|
|
class StateLock:
|
|
def __enter__(self):
|
|
context = transaction()
|
|
result = context.__enter__()
|
|
contexts = getattr(_local, 'contexts', [])
|
|
contexts.append(context)
|
|
_local.contexts = contexts
|
|
return result
|
|
|
|
def __exit__(self, *args):
|
|
return _local.contexts.pop().__exit__(*args)
|
|
|
|
|
|
def atomic(function):
|
|
@functools.wraps(function)
|
|
def wrapped(*args, **kwargs):
|
|
with transaction():
|
|
return function(*args, **kwargs)
|
|
return wrapped
|