diff --git a/scripts/ci_shards.py b/scripts/ci_shards.py index f7e52060..d6fd86a5 100644 --- a/scripts/ci_shards.py +++ b/scripts/ci_shards.py @@ -24,7 +24,7 @@ def partition(nodeids, index, count): def evidence(nodeids, index, count): all_nodes = sorted(nodeids) - return {'version': 1, 'index': index, 'count': count, 'all': all_nodes, + return {'version': 2, 'index': index, 'count': count, 'all': all_nodes, 'selected': partition(all_nodes, index, count)} @@ -48,6 +48,14 @@ def verify(directory, count): cases = root.findall('.//testcase') if len(cases) != len(selected) or root.findall('.//error') or root.findall('.//failure'): raise ValueError('JUnit result is incomplete or failed') + observed = [] + for case in cases: + identities = case.findall('./properties/property[@name="msg.nodeid"]') + if len(identities) != 1 or identities[0].get('value') is None: + raise ValueError('JUnit test identity is missing or ambiguous') + observed.append(identities[0].get('value')) + if sorted(observed) != sorted(selected): + raise ValueError('JUnit test identities do not match the selected shard') if not expected or seen != expected: raise ValueError('test coverage is incomplete') return len(seen) @@ -66,6 +74,10 @@ def pytest_collection_modifyitems(self, config, items): chosen = set(manifest['selected']) rejected = [item for item in items if item.nodeid not in chosen] items[:] = [item for item in items if item.nodeid in chosen] + for item in items: + # Preserve the exact pytest identity, including class/parameter + # punctuation, instead of reverse-engineering JUnit display names. + item.user_properties.append(('msg.nodeid', item.nodeid)) config.hook.pytest_deselected(items=rejected) (directory / f'shard-{index}.json').write_text( json.dumps(manifest, ensure_ascii=False, indent=2) + '\n') diff --git a/src/msg/extensions/repositories.py b/src/msg/extensions/repositories.py index 37c25914..c5d83b1a 100644 --- a/src/msg/extensions/repositories.py +++ b/src/msg/extensions/repositories.py @@ -708,6 +708,7 @@ async def push(ctx,request,tx): async def execute_push(app,job): from msg.workers.effects import current_principal,worker_context,effect_request + from msg.workers.leases import current_attempt store=NativeGitStore(app) app.settings.server.staging_dir.mkdir(parents=True,exist_ok=True) with tempfile.TemporaryDirectory(prefix='git-push-',dir=app.settings.server.staging_dir) as temp: @@ -729,8 +730,9 @@ async def execute_push(app,job): context,request=worker_context(app,job,principal),effect_request(app,job,principal) await check_access(app,context,request,tx,job.arguments['id'],'write') require(tx.setting('runtime_config',{}).get('accept_writes',True),'writes_paused') - current=await tx.job(job.id) - require(current.state=='running' and current.attempts==job.attempts,'job_lease_lost') + current=await current_attempt(app,tx,job) + if current is None: + return await store.update_refs(job.arguments['id'],job.arguments['changes']) moved=True resource=await tx.resource(job.arguments['id']) diff --git a/src/msg/workers/effects.py b/src/msg/workers/effects.py index 1bb4eb3f..09bdbad4 100644 --- a/src/msg/workers/effects.py +++ b/src/msg/workers/effects.py @@ -23,6 +23,7 @@ from msg.plugins.common import check_access, create_resource from msg.security.policy import scope_subset from msg.security.network import intersect_policy +from msg.workers.leases import current_attempt @dataclass(frozen=True, slots=True) @@ -131,28 +132,18 @@ async def _claim(self): await tx.save_job(job) return job, True - async def _live_attempt(self, tx, job): - """Fence completions even when no other worker has swept expired leases.""" - current = await tx.job(job.id) - if current.state != 'running' or current.attempts != job.attempts: - return None - if current.lease_until is None or current.lease_until <= self.app.clock(): - await tx.save_job(replace(current, state='uncertain', lease_until=None)) - tx.set_setting('job_status:' + job.id, {'code': 'expired_execution_lease'}) - return None - return current - - async def _finish(self, job, state, code): + async def _finish(self, job, state, code, *, status=None): async with self.app.metadata.transaction(write=True) as tx: - current = await self._live_attempt(tx, job) + current = await current_attempt(self.app, tx, job) if current is None: return await tx.save_job(replace(current, state=state, lease_until=None)) - tx.set_setting('job_status:' + job.id, {'code': code}) + tx.set_setting('job_status:' + job.id, + {'code': code} if status is None else status) async def _retry(self, job, retry_code, exhausted_code): async with self.app.metadata.transaction(write=True) as tx: - current = await self._live_attempt(tx, job) + current = await current_attempt(self.app, tx, job) if current is None: return exhausted = current.attempts >= 8 @@ -223,8 +214,9 @@ async def pieces(): blob = await self.app.contents.put(pieces(), result.media_type) await self.app.contents.pin(blob, 'job:'+job.id) async with self.app.metadata.transaction(write=True) as tx: - current = await tx.job(job.id) - require(current.state == 'running' and current.attempts == job.attempts, 'job_lease_lost') + current = await current_attempt(self.app, tx, job) + if current is None: + return principal = await current_principal(self.app, job.principal, tx) context, request = worker_context(self.app, job, principal), effect_request(self.app, job, principal) await check_access(self.app, context, request, tx, ref.id, 'tool_use') @@ -357,9 +349,7 @@ async def run_once(self): elif job.kind == 'maintenance': from msg.workers.maintenance import run_maintenance result=await run_maintenance(self.app,job.arguments['action'],principal=job.principal) - await self._finish(job,'done','ok') - async with self.app.metadata.transaction(write=True) as tx: - tx.set_setting('job_status:'+job.id,result) + await self._finish(job,'done','ok',status=result) elif job.kind == 'gc.resource': # Purge permission was consumed by the committed tombstone. This # idempotent cleanup cannot change another resource's retention. diff --git a/src/msg/workers/leases.py b/src/msg/workers/leases.py new file mode 100644 index 00000000..6217fcbd --- /dev/null +++ b/src/msg/workers/leases.py @@ -0,0 +1,19 @@ +"""One transaction-local fence for publishing any durable effect's result.""" +from dataclasses import replace + + +async def current_attempt(app, tx, job): + """Return the live attempt, or persist uncertainty without resending work. + + Call within the same write transaction as the protected state/ref update. + A stale callback must not overwrite another attempt's status, even when no + separate worker has swept the expired lease yet. + """ + current = await tx.job(job.id) + if current.state != 'running' or current.attempts != job.attempts: + return None + if current.lease_until is None or current.lease_until <= app.clock(): + await tx.save_job(replace(current, state='uncertain', lease_until=None)) + tx.set_setting('job_status:' + job.id, {'code': 'expired_execution_lease'}) + return None + return current diff --git a/tests/test_ci_shards.py b/tests/test_ci_shards.py index d62c9582..bbd7eab2 100644 --- a/tests/test_ci_shards.py +++ b/tests/test_ci_shards.py @@ -1,6 +1,9 @@ """The CI gate must reject missing, duplicate, changed or failing evidence.""" import importlib.util import json +import os +import subprocess +import sys from pathlib import Path import xml.etree.ElementTree as ET @@ -20,7 +23,9 @@ def write_evidence(directory, count=4): (directory / f'shard-{index}.json').write_text(json.dumps(record)) suite = ET.Element('testsuite') for node in record['selected']: - ET.SubElement(suite, 'testcase', name=node) + case = ET.SubElement(suite, 'testcase', name=node) + properties = ET.SubElement(case, 'properties') + ET.SubElement(properties, 'property', name='msg.nodeid', value=node) ET.ElementTree(suite).write(directory / f'tests-{index}.xml') @@ -73,3 +78,45 @@ def test_gate_rejects_incomplete_or_inconsistent_evidence(tmp_path, damage): tree.write(report) with pytest.raises((ValueError, FileNotFoundError)): ci.verify(tmp_path, 4) + + +@pytest.mark.parametrize('damage', ['renamed_node', 'duplicate_node', 'missing_identity', 'duplicate_identity']) +def test_equal_test_counts_do_not_hide_wrong_execution_identities(tmp_path, damage): + write_evidence(tmp_path) + report = tmp_path / 'tests-1.xml' + tree = ET.parse(report) + cases = tree.getroot().findall('testcase') + if damage == 'renamed_node': + cases[0].find('./properties/property').set('value', 'tests/test_other.py::wrong') + elif damage == 'duplicate_node': + cases[0].find('./properties/property').set('value', + cases[1].find('./properties/property').get('value')) + elif damage == 'missing_identity': + cases[0].remove(cases[0].find('properties')) + else: + ET.SubElement(cases[0].find('properties'), 'property', name='msg.nodeid', value='extra') + tree.write(report) + with pytest.raises(ValueError): + ci.verify(tmp_path, 4) + + +def test_real_pytest_reports_exact_nodeids_for_parameters_and_classes(tmp_path): + tests = tmp_path / 'tests' + tests.mkdir() + (tests / 'test_sample.py').write_text( + "import pytest\n" + "@pytest.mark.parametrize('value', ['normal', '::[x]', '中文'])\n" + "def test_parameter(value):\n assert value\n" + "class TestNested:\n def test_child(self):\n assert True\n") + directory = tmp_path / 'reports' + completed = subprocess.run( + [sys.executable, ci.__file__, 'run', '0', '1', '--directory', str(directory)], + cwd=tmp_path, env={**os.environ, 'PYTEST_DISABLE_PLUGIN_AUTOLOAD': '1'}, + capture_output=True, text=True, timeout=30) + assert completed.returncode == 0, completed.stdout + completed.stderr + assert ci.verify(directory, 1) == 4 + manifest = json.loads((directory / 'shard-0.json').read_text()) + tree = ET.parse(directory / 'tests-0.xml') + observed = [case.find('./properties/property[@name="msg.nodeid"]').get('value') + for case in tree.getroot().findall('.//testcase')] + assert sorted(observed) == manifest['selected'] diff --git a/tests/test_effect_completion_fencing.py b/tests/test_effect_completion_fencing.py new file mode 100644 index 00000000..93942432 --- /dev/null +++ b/tests/test_effect_completion_fencing.py @@ -0,0 +1,160 @@ +"""Every completion uses the same stored lease/attempt fence. + +Maintenance doubles below isolate state transitions; the tool/Git tests exercise +signed Operations, real PostgreSQL and local Git/CAS, without external traffic. +""" +from dataclasses import replace +from datetime import timedelta + +import pytest + +from msg.core.models import EffectJob, Principal +from msg.workers.effects import EffectWorker, ToolResult + + +@pytest.mark.asyncio +@pytest.mark.parametrize('outcome', ['expired', 'missing', 'superseded', 'uncertain', 'live']) +async def test_maintenance_result_is_atomic_with_the_current_attempt(installed, monkeypatch, outcome): + from msg.workers import maintenance + app, _ = installed + now = app.clock() + job = EffectJob(id='job_maintenance_fence', event_id='e_maintenance_fence', + kind='maintenance', dedupe_key='maintenance-fence', + principal=Principal(actor=None, subject=None, credential_id=None, + method='anonymous', certificates=(), ceiling=()), + operation='system.maintenance', arguments={'action': 'rebuild_search'}, + state='pending', attempts=0, next_attempt_at=now, lease_until=None) + async with app.metadata.transaction(write=True) as tx: + await tx.enqueue(job) + report = {'indexed_resources': 2} + observed = [] + + async def completed_work(app, action, *, principal): + assert action == 'rebuild_search' + async with app.metadata.transaction(write=True) as tx: + current = await tx.job(job.id) + if outcome == 'expired': + current = replace(current, lease_until=now) + elif outcome == 'missing': + current = replace(current, lease_until=None) + elif outcome == 'superseded': + current = replace(current, attempts=current.attempts + 1) + elif outcome == 'uncertain': + current = replace(current, state='uncertain', lease_until=None) + await tx.save_job(current) + tx.set_setting('job_status:' + job.id, {'code': 'existing_decision'}) + observed.append(current) + return report + + monkeypatch.setattr(maintenance, 'run_maintenance', completed_work) + assert await EffectWorker(app).run_once() + async with app.metadata.transaction(write=False) as tx: + current = await tx.job(job.id) + status = tx.setting('job_status:' + job.id) + if outcome == 'live': + assert current == replace(observed[0], state='done', lease_until=None) + assert status == report + elif outcome in {'expired', 'missing'}: + assert current == replace(observed[0], state='uncertain', lease_until=None) + assert status == {'code': 'expired_execution_lease'} + else: + assert current == observed[0] + assert status == {'code': 'existing_decision'} + + +async def expire(app, job_id, deadline): + async with app.metadata.transaction(write=True) as tx: + current = await tx.job(job_id) + until = {'exact': app.clock(), 'past': app.clock() - timedelta(seconds=1), + 'missing': None}[deadline] + await tx.save_job(replace(current, lease_until=until)) + + +@pytest.mark.asyncio +@pytest.mark.parametrize('deadline', ['exact', 'past', 'missing']) +async def test_real_tool_expiry_does_not_publish_a_file_without_a_sweep(installed, deadline): + from test_service import register, call + from test_authorization import approve, scoped + app, root = installed + key, uid, _ = await register(app, 'tool-completion-fence') + grant = scoped(app, 'tool.use', 'tool_dns', app.registry.capability('tool.use').operations) + certificate = await approve(app, root, uid, key, (grant,)) + accepted = await call(app, 'tool.run', {'id': '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/tools/dns', + 'arguments': {'name': 'example.org', 'type': 'A'}}, key=key, + subject=uid, certs=(certificate.resource_id,)) + assert accepted.status == 'accepted' + job_id = accepted.data['job_id'] + executions = [] + + async def runner(tool, arguments, policies, directory): + executions.append(tool.executor_key) + await expire(app, job_id, deadline) + path = directory / 'result.json' + path.write_bytes(b'[]') + return ToolResult(path=path, media_type='application/json', metadata={}) + + worker = EffectWorker(app, tool_runner=runner) + assert await worker.run_once() + async with app.metadata.transaction(write=False) as tx: + current = await tx.job(job_id) + assert current.state == 'uncertain' and current.result is None + assert tx.setting('job_status:' + job_id) == {'code': 'expired_execution_lease'} + assert tx.one('SELECT COUNT(*) FROM resources WHERE name=?', ('tool-' + job_id,))[0] == 0 + assert executions == ['dns'] + assert not await worker.run_once() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('deadline', ['exact', 'past', 'missing']) +async def test_real_git_expiry_does_not_move_refs_without_a_sweep(installed, tmp_path, monkeypatch, deadline): + import subprocess + from msg.core.codec import b64, wire + from msg.extensions.repositories import NativeGitStore + from test_service import register, call + app, _ = installed + key, uid, _ = await register(app, 'git-completion-fence') + repo = await call(app, 'git.create', {'parent': '/@git-completion-fence', + 'name': 'code.git'}, key=key, subject=uid) + assert repo.status == 'ok' + work = tmp_path / 'work' + work.mkdir() + + def git(*arguments): + return subprocess.run(['git', '-C', str(work), *arguments], check=True, + capture_output=True).stdout.decode().strip() + + git('init', '--initial-branch=main') + git('config', 'user.name', 'Test') + git('config', 'user.email', 'test@example.invalid') + (work / 'README.md').write_text('public code\n') + git('add', '.') + git('commit', '-m', 'first') + oid = git('rev-parse', 'HEAD') + bundle = tmp_path / 'input.bundle' + git('bundle', 'create', str(bundle), 'main') + upload = await call(app, 'content.file_put', {'parent': '/@git-completion-fence/files', + 'name': 'input.bundle', 'data': b64(bundle.read_bytes()), + 'media_type': 'application/octet-stream'}, key=key, subject=uid) + assert upload.status == 'ok' + accepted = await call(app, 'git.push', {'id': repo.resources[0].id, + 'bundle': wire(upload.resources[0]), + 'changes': [{'ref': 'refs/heads/main', 'old': None, 'new': oid}]}, + key=key, subject=uid) + assert accepted.status == 'accepted' + job_id = accepted.data['job_id'] + import_bundle = NativeGitStore.import_bundle + + async def import_then_expire(store, *args, **kwargs): + await import_bundle(store, *args, **kwargs) + await expire(app, job_id, deadline) + + monkeypatch.setattr(NativeGitStore, 'import_bundle', import_then_expire) + worker = EffectWorker(app) + assert await worker.run_once() + async with app.metadata.transaction(write=False) as tx: + assert (await tx.job(job_id)).state == 'uncertain' + assert tx.setting('job_status:' + job_id) == {'code': 'expired_execution_lease'} + assert (await tx.resource(repo.resources[0].id)).generation == repo.data['generation'] + refs = await call(app, 'git.refs', {'id': repo.resources[0].id}) + assert refs.status == 'ok' and list(refs.data['refs']) == [] + assert not await worker.run_once()