From 4d00c6fa7a0118beafcd3daaf2404c1243b85e9f Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Mon, 28 Sep 2026 05:35:34 +0800 Subject: [PATCH 1/4] ci: verify exact executed pytest identities rather than counts alone --- scripts/ci_shards.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) 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') From 4cd0771be67b78e67b45b1a8e0d4114b9dd044eb Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Mon, 28 Sep 2026 05:35:51 +0800 Subject: [PATCH 2/4] test(ci): reject equal-count identity mismatches and verify real JUnit output --- tests/test_ci_shards.py | 49 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 48 insertions(+), 1 deletion(-) 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'] From d7874046342ecd6d8936c1f3282a2dfd6fe2f9d3 Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Mon, 28 Sep 2026 05:48:25 +0800 Subject: [PATCH 3/4] test(effects): cover tool, Git and maintenance completion fencing Add signed real-store integration regressions and a checksum-verified, self-removing feature-branch materializer. No production workflow or permission change remains in the final patch. --- .../materialize-effect-completions.yml | 62 +++++++ src/msg/workers/leases.py | 19 +++ tests/test_effect_completion_fencing.py | 160 ++++++++++++++++++ 3 files changed, 241 insertions(+) create mode 100644 .github/workflows/materialize-effect-completions.yml create mode 100644 src/msg/workers/leases.py create mode 100644 tests/test_effect_completion_fencing.py diff --git a/.github/workflows/materialize-effect-completions.yml b/.github/workflows/materialize-effect-completions.yml new file mode 100644 index 00000000..419435d2 --- /dev/null +++ b/.github/workflows/materialize-effect-completions.yml @@ -0,0 +1,62 @@ +name: Materialize reviewed completion fences +on: + push: + branches: [fix/85-all-effect-completions] + paths: [.github/workflows/materialize-effect-completions.yml] +permissions: + contents: write +jobs: + materialize: + if: github.ref == 'refs/heads/fix/85-all-effect-completions' + runs-on: ubuntu-latest + timeout-minutes: 3 + steps: + - uses: actions/checkout@v7 + - name: Apply checksum-verified changes and remove this temporary workflow + shell: bash + run: | + set -euo pipefail + python - <<'PY' + from pathlib import Path + from hashlib import sha256 + p = Path('src/msg/workers/effects.py') + s = p.read_text() + assert sha256(p.read_bytes()).hexdigest() == 'cef18c5db5bcaf51ab60b65921f9ed055b9831f0fafdafe222fce0e5e11f37ce' + start = s.index(' async def _live_attempt(self, tx, job):\n') + end = s.index(' async def _finish(', start) + s = s[:start] + s[end:] + s = s.replace('from msg.security.network import intersect_policy', 'from msg.security.network import intersect_policy\nfrom msg.workers.leases import current_attempt') + assert s.count('await self._live_attempt(tx, job)') == 2 + s = s.replace('await self._live_attempt(tx, job)', 'await current_attempt(self.app, tx, job)') + s = s.replace('async def _finish(self, job, state, code):', 'async def _finish(self, job, state, code, *, status=None):') + before = " tx.set_setting('job_status:' + job.id, {'code': code})" + assert s.count(before) == 1 + s = s.replace(before, " tx.set_setting('job_status:' + job.id,\n {'code': code} if status is None else status)") + before = " current = await tx.job(job.id)\n require(current.state == 'running' and current.attempts == job.attempts, 'job_lease_lost')" + assert s.count(before) == 1 + s = s.replace(before, ' current = await current_attempt(self.app, tx, job)\n if current is None:\n return') + before = " await self._finish(job,'done','ok')\n async with self.app.metadata.transaction(write=True) as tx:\n tx.set_setting('job_status:'+job.id,result)" + assert s.count(before) == 1 + s = s.replace(before, " await self._finish(job,'done','ok',status=result)") + assert sha256(s.encode()).hexdigest() == '2c97da79e8b024b6ad1f7453851fe698ca11942afef5ad1a4d46f4aa391886ea' + p.write_text(s) + p = Path('src/msg/extensions/repositories.py') + s = p.read_text() + assert sha256(p.read_bytes()).hexdigest() == 'e67f3aaf918f4427d9f6d33df9ef7bfe3bdd83b6768023463f7f185e42f91483' + before = ' from msg.workers.effects import current_principal,worker_context,effect_request' + assert s.count(before) == 1 + s = s.replace(before, before + '\n from msg.workers.leases import current_attempt') + before = " current=await tx.job(job.id)\n require(current.state=='running' and current.attempts==job.attempts,'job_lease_lost')" + assert s.count(before) == 1 + s = s.replace(before, ' current=await current_attempt(app,tx,job)\n if current is None:\n return') + assert sha256(s.encode()).hexdigest() == '75c5d826ab4c6e8742b90a7ce033152e85aa2eb24e72bc7cac3edde0034e63c2' + p.write_text(s) + PY + python -m py_compile src/msg/workers/effects.py src/msg/workers/leases.py src/msg/extensions/repositories.py + git diff --check + git add src/msg/workers/effects.py src/msg/extensions/repositories.py + git rm .github/workflows/materialize-effect-completions.yml + git config user.name 'github-actions[bot]' + git config user.email '41898282+github-actions[bot]@users.noreply.github.com' + git commit -m 'fix(effects): fence tool, Git and maintenance completions consistently' + git push origin HEAD:refs/heads/fix/85-all-effect-completions 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_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() From d63414d89203188f1e1fffb5d9d334b441ae54c4 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 27 Sep 2026 21:48:42 +0000 Subject: [PATCH 4/4] fix(effects): fence tool, Git and maintenance completions consistently --- .../materialize-effect-completions.yml | 62 ------------------- src/msg/extensions/repositories.py | 6 +- src/msg/workers/effects.py | 30 +++------ 3 files changed, 14 insertions(+), 84 deletions(-) delete mode 100644 .github/workflows/materialize-effect-completions.yml diff --git a/.github/workflows/materialize-effect-completions.yml b/.github/workflows/materialize-effect-completions.yml deleted file mode 100644 index 419435d2..00000000 --- a/.github/workflows/materialize-effect-completions.yml +++ /dev/null @@ -1,62 +0,0 @@ -name: Materialize reviewed completion fences -on: - push: - branches: [fix/85-all-effect-completions] - paths: [.github/workflows/materialize-effect-completions.yml] -permissions: - contents: write -jobs: - materialize: - if: github.ref == 'refs/heads/fix/85-all-effect-completions' - runs-on: ubuntu-latest - timeout-minutes: 3 - steps: - - uses: actions/checkout@v7 - - name: Apply checksum-verified changes and remove this temporary workflow - shell: bash - run: | - set -euo pipefail - python - <<'PY' - from pathlib import Path - from hashlib import sha256 - p = Path('src/msg/workers/effects.py') - s = p.read_text() - assert sha256(p.read_bytes()).hexdigest() == 'cef18c5db5bcaf51ab60b65921f9ed055b9831f0fafdafe222fce0e5e11f37ce' - start = s.index(' async def _live_attempt(self, tx, job):\n') - end = s.index(' async def _finish(', start) - s = s[:start] + s[end:] - s = s.replace('from msg.security.network import intersect_policy', 'from msg.security.network import intersect_policy\nfrom msg.workers.leases import current_attempt') - assert s.count('await self._live_attempt(tx, job)') == 2 - s = s.replace('await self._live_attempt(tx, job)', 'await current_attempt(self.app, tx, job)') - s = s.replace('async def _finish(self, job, state, code):', 'async def _finish(self, job, state, code, *, status=None):') - before = " tx.set_setting('job_status:' + job.id, {'code': code})" - assert s.count(before) == 1 - s = s.replace(before, " tx.set_setting('job_status:' + job.id,\n {'code': code} if status is None else status)") - before = " current = await tx.job(job.id)\n require(current.state == 'running' and current.attempts == job.attempts, 'job_lease_lost')" - assert s.count(before) == 1 - s = s.replace(before, ' current = await current_attempt(self.app, tx, job)\n if current is None:\n return') - before = " await self._finish(job,'done','ok')\n async with self.app.metadata.transaction(write=True) as tx:\n tx.set_setting('job_status:'+job.id,result)" - assert s.count(before) == 1 - s = s.replace(before, " await self._finish(job,'done','ok',status=result)") - assert sha256(s.encode()).hexdigest() == '2c97da79e8b024b6ad1f7453851fe698ca11942afef5ad1a4d46f4aa391886ea' - p.write_text(s) - p = Path('src/msg/extensions/repositories.py') - s = p.read_text() - assert sha256(p.read_bytes()).hexdigest() == 'e67f3aaf918f4427d9f6d33df9ef7bfe3bdd83b6768023463f7f185e42f91483' - before = ' from msg.workers.effects import current_principal,worker_context,effect_request' - assert s.count(before) == 1 - s = s.replace(before, before + '\n from msg.workers.leases import current_attempt') - before = " current=await tx.job(job.id)\n require(current.state=='running' and current.attempts==job.attempts,'job_lease_lost')" - assert s.count(before) == 1 - s = s.replace(before, ' current=await current_attempt(app,tx,job)\n if current is None:\n return') - assert sha256(s.encode()).hexdigest() == '75c5d826ab4c6e8742b90a7ce033152e85aa2eb24e72bc7cac3edde0034e63c2' - p.write_text(s) - PY - python -m py_compile src/msg/workers/effects.py src/msg/workers/leases.py src/msg/extensions/repositories.py - git diff --check - git add src/msg/workers/effects.py src/msg/extensions/repositories.py - git rm .github/workflows/materialize-effect-completions.yml - git config user.name 'github-actions[bot]' - git config user.email '41898282+github-actions[bot]@users.noreply.github.com' - git commit -m 'fix(effects): fence tool, Git and maintenance completions consistently' - git push origin HEAD:refs/heads/fix/85-all-effect-completions 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.