diff --git a/flake.lock b/flake.lock index 1151556..45f7426 100644 --- a/flake.lock +++ b/flake.lock @@ -42,17 +42,17 @@ "fractal-fixture": { "flake": false, "locked": { - "lastModified": 1793341662, - "narHash": "sha256-TG9HJx2WE941Bv6crfWIm2IJDyOR8WBKHkic3mZ0MeE=", + "lastModified": 1794265163, + "narHash": "sha256-pGJRiUHcongVBrYEWT2w3TZTS8UdcZJjHRJvTDhrpYg=", "owner": "compoundingtech", "repo": "fractal", - "rev": "e83871ec919f95b3b3e2a9a1cd211c2ac37babda", + "rev": "e2721e52463428b4ff655addee129cecd2839763", "type": "github" }, "original": { "owner": "compoundingtech", "repo": "fractal", - "rev": "e83871ec919f95b3b3e2a9a1cd211c2ac37babda", + "rev": "e2721e52463428b4ff655addee129cecd2839763", "type": "github" } }, diff --git a/flake.nix b/flake.nix index dc66ce0..2545d9b 100644 --- a/flake.nix +++ b/flake.nix @@ -7,7 +7,7 @@ inputs.effect-utils.inputs.nixpkgs.url = "github:NixOS/nixpkgs/4ba99f3209788ed04f01af38f77ddb803ad6ec63"; inputs.effect-utils.inputs.tsgo.inputs.nixpkgsUnstable.url = "github:NixOS/nixpkgs/44a91898084f46797b5fac650c7e8c9ac38c43d4"; inputs.fractal-fixture = { - url = "github:compoundingtech/fractal/e83871ec919f95b3b3e2a9a1cd211c2ac37babda"; + url = "github:compoundingtech/fractal/e2721e52463428b4ff655addee129cecd2839763"; flake = false; }; inputs.python-oracle = { diff --git a/scripts/PROVENANCE b/scripts/PROVENANCE index b5b010c..b10409d 100644 --- a/scripts/PROVENANCE +++ b/scripts/PROVENANCE @@ -28,6 +28,11 @@ # resync-blob scripts/fractal-test-report-test b0dc50163986760343d7eb67c701cb4b775e9945 # resync-blob scripts/lane-definition-closure 492b3741b117092395855d9790dc96e6ce3612e6 # resync-blob scripts/lane-definition-closure-test 635945d2a01f18e9c4716c50a2eb9803896cbb08 +# resync-blob scripts/evidence-flip-ready ce94216c0435711419e1b2bc5acc996e2ac926b9 +# resync-blob scripts/evidence-flip-ready-test 16e505355c4092c880c3f5e283baa8c0996cea4b +# resync-blob scripts/evidence-verify-test 08d11cc7ba5e8a3d5ffb59fadb9eb49238aec95f +# resync-blob scripts/st-test 087ae7d9f3d1106a38eedc1e911caeeaf46a96ae +# resync-blob scripts/fractal-test-test decb42b9d442807c9601b473c190b3743b24c4c0 # scripts/fractal-test-test: apply the fixture maintenance-isolation patch from # https://github.com/compoundingtech/fractal/pull/4398 (5f995f26e5). # The source_blob column remains the original import; mergemaster_blob includes applied patches. diff --git a/scripts/ci-evidence-gate b/scripts/ci-evidence-gate index 4400fdf..2952f89 100755 --- a/scripts/ci-evidence-gate +++ b/scripts/ci-evidence-gate @@ -45,7 +45,7 @@ VERIFIER_TIMEOUT = int(os.environ.get('GATE_VERIFIER_TIMEOUT', '30')) # A change that touches anything that decides a skip is not skipped on a note alone: those files are what a forged skip would # have to edit. Compared between the accepted base and this run's tree. One exception (decide): a pull_request run covered by a note # of one of OWN_LEVELS, our dev hosts, which ran the PR's own scripts. It is judged by THIS gate and verifier as the base has them -# (the workflow fetches both from BASE_SHA), never the PR's copies, and the merge queue runs every check in full before it lands. +# (the workflow fetches both from BASE_SHA), never the PR's copies. Merge groups reuse only exact-tree evidence enabled at BASE. PROTECTED = ( 'scripts/ci-evidence-gate', 'scripts/evidence-verify', '.st/evidence-skip-policy.json', '.st/evidence-keys', '.github/workflows/fleet.yml', '.github/workflows/fleet.yml.genie.ts', '.github/workflows/workspace-ci.ts', @@ -158,6 +158,11 @@ def decide(context, outcome, result, tree): return 'run', 'event does not allow skipping' if context.get('tree_of') == 'head' and context['event'] != 'pull_request': return 'run', 'a head-tree note covers pull request runs only' + if context['event'] == 'merge_group': + if context.get('merge_group_mode') != 'same-tree': + return 'run', 'merge-group tree reuse is not enabled at the accepted base' + if result.get('match') != 'tree': + return 'run', 'merge groups require exact-tree evidence, not input-key reuse' protected = context.get('protected_changed') if protected != [] and not (own_pull_request(context, result) and isinstance(protected, list) and all(isinstance(p, str) and p and not p.startswith('<') for p in protected)): @@ -190,6 +195,20 @@ def own_pull_request(context, result): return context['event'] == 'pull_request' and result.get('trust_level') in OWN_LEVELS +def merge_group_mode(context): + """Queue reuse is opt-in at BASE; absent, malformed or unknown modes keep hosted CI running.""" + path = config_path(context['base'], SIGNED_ONLY) + if path is None: + return None + shown = run('git', 'show', f"{context['base']}:{path}") + if shown.returncode != 0: + return None + try: + return json.loads(shown.stdout).get('merge_group') + except (ValueError, AttributeError): + return None + + def signed_only(context): """Whether this opener is signed-only, using the accepted base's selected layout. @@ -736,6 +755,8 @@ def main(): except Exception: context['protected_changed'] = [''] try: + if context['event'] == 'merge_group': + context['merge_group_mode'] = merge_group_mode(context) decision, why = decide(context, outcome, result, tree) except Exception: decision, why = 'run', 'decision error' diff --git a/scripts/ci-evidence-gate-test b/scripts/ci-evidence-gate-test index 9927e0b..38a34e7 100755 --- a/scripts/ci-evidence-gate-test +++ b/scripts/ci-evidence-gate-test @@ -208,10 +208,11 @@ class Decide(unittest.TestCase): def good(self): record = 'a' * 40 now = gate.datetime.now(gate.timezone.utc).isoformat() - context = {'check': 'linux-gate', 'fork': 'false', 'event': 'merge_group', 'protected_changed': []} + context = {'check': 'linux-gate', 'fork': 'false', 'event': 'merge_group', 'protected_changed': [], + 'merge_group_mode': 'same-tree'} result = {'satisfied': True, 'policy': True, 'enforce': True, 'tier': 'full', 'kind': 'observed', 'trust_level': 'fleet', 'tree': 'b' * 40, 'check': 'linux-gate', 'record': record, 'signer': 'fleet/ci', 'observed_at': now, - 'sample_one_in': 1} + 'sample_one_in': 1, 'match': 'tree'} return context, result, 'b' * 40 def test_all_conditions_together_skip(self): @@ -585,6 +586,9 @@ class EndToEnd(unittest.TestCase): (repo / '.st/evidence-keys').write_text(self.f.keys.read_text()) if policy is not False: (repo / '.st/evidence-skip-policy.json').write_text(policy or skip_policy()) + signed = repo / '.st/evidence-signed-only.json' + if not signed.exists(): + signed.write_text(json.dumps({'openers': [], 'merge_group': 'same-tree'})) self.f.git('add', '-A') self.f.git('commit', '-q', '--allow-empty', '-m', 'base') return self.f.git('rev-parse', 'HEAD') @@ -630,6 +634,68 @@ class EndToEnd(unittest.TestCase): self.assertEqual((fields['decision'], outputs['skip'], outputs['record']), ('run', 'false', '-'), why) return fields + def test_queue_reuses_an_equal_tree_on_a_different_commit_then_runs_a_changed_tree(self): + base = self.commit_base() + commit, tree = self.head() + self.note(commit, tree) + self.f.git('commit', '-q', '--allow-empty', '-m', 'queue commit') + self.assertNotEqual(commit, self.f.git('rev-parse', 'HEAD')) + self.assertEqual(tree, self.f.git('rev-parse', 'HEAD^{tree}')) + fields, outputs = self.run_gate(base) + self.assertEqual((fields['decision'], fields['match'], outputs['refuse']), ('skip', 'tree', 'false')) + self.f.commit('another queued change') + self.assertEqual(self.assertRuns(base)['outcome'], 'no_note') + + def test_queue_invalid_signature_or_principal_runs_and_never_refuses(self): + for kwargs in ({'key': self.stranger}, {'signer': 'fleet/not-allowed'}): + with self.subTest(kwargs): + base = self.commit_base() + commit, tree = self.f.commit('invalid ' + str(kwargs)) + self.note(commit, tree, **kwargs) + fields, outputs = self.run_gate(base) + self.assertEqual((fields['decision'], fields['outcome'], outputs['refuse']), + ('run', 'rejected', 'false')) + + def test_queue_mode_comes_only_from_base_and_off_or_malformed_runs(self): + for text in ('{"openers": [], "merge_group": "off"}', '{"openers": []}', '{bad', + '{"openers": [], "merge_group": "unknown"}'): + with self.subTest(text): + path = self.f.repo / '.st/evidence-signed-only.json' + path.parent.mkdir(exist_ok=True) + path.write_text(text) + base = self.commit_base() + path.write_text('{"openers": [], "merge_group": "same-tree"}') + commit, tree = self.head() + self.note(commit, tree) + fields, outputs = self.run_gate(base) + self.assertEqual((fields['decision'], fields['outcome'], outputs['refuse']), + ('run', 'satisfied', 'false')) + self.assertEqual(json.loads(root_config('evidence-signed-only.json').read_text())['merge_group'], 'off') + + def test_queue_agent_note_requires_the_base_lane_digest(self): + agent = self.f.key('agent', 'agent/larch') + policy = skip_policy(trust_levels=['agent'], kinds={'wrapper': { + 'command': ['make', 'test'], 'by_level': {'agent': { + 'command': ['make', 'test'], 'events': ['pull_request', 'merge_group'], 'lane': True}}}}) + repo = self.f.repo + (repo / '.st').mkdir(exist_ok=True) + (repo / '.st/required-lanes.json').write_text(json.dumps(self.LANES)) + (repo / 'scripts').mkdir(exist_ok=True) + (repo / 'scripts/lane.sh').write_text('test\n') + (repo / 'scripts/lane.json').write_text('{}\n') + base = self.commit_base(policy=policy) + commit, tree = self.head() + rules = fixtures.evidence_verify.lane_rules(json.dumps(self.LANES), 'linux-gate') + digest = fixtures.evidence_verify.lane_definition(str(repo), base, tree, rules)['sha256'] + record = self.note(commit, tree, key=agent, signer='agent/larch', runner='wrapper', + command=('make', 'test'), inputs={'lane_definition_sha256': digest}) + self.f.git('commit', '-q', '--allow-empty', '-m', 'queue') + self.assertEqual(self.run_gate(base)[0]['decision'], 'skip') + self.f.git('update-ref', '-d', self.f.ref_of[record]) + self.note(commit, tree, key=agent, signer='agent/larch', runner='wrapper', + command=('make', 'test'), inputs={'lane_definition_sha256': '0' * 64}, seconds=1) + self.assertEqual(self.assertRuns(base)['outcome'], 'rejected') + # ---- the happy path and its progression ------------------------------------------------- def test_no_note_then_degraded_then_a_trusted_note_skips_then_a_stranger_does_not(self): @@ -1505,9 +1571,8 @@ class EndToEnd(unittest.TestCase): for check, entry in policy['checks'].items(): self.assertIn(entry.get('enforce', False), (True, False), check) - def test_the_committed_policy_skips_an_enforced_check_on_pull_requests_and_never_in_the_merge_queue(self): - # The committed policy as it is (only sampling and inputs neutralised): an enforced check skips a pull request run on a - # trusted note and still runs in the merge queue, which keeps testing the combined tree; a check that is not enforced never skips. + def test_the_committed_policy_can_skip_exact_queue_trees_when_the_mode_is_enabled(self): + # Fixture opts in; the deployed mode remains off until post-merge hosted proof is deployed. doc = json.loads(root_config('evidence-skip-policy.json').read_text()) for entry in doc['checks'].values(): entry['sample_one_in'] = 1 @@ -1527,9 +1592,7 @@ class EndToEnd(unittest.TestCase): queue, _ = self.run_gate(base, check=check, EVENT_NAME='merge_group') self.assertEqual((pull['outcome'], queue['outcome']), ('satisfied', 'satisfied'), check) self.assertEqual(pull['decision'], 'skip' if enforced else 'run', check) - self.assertEqual(queue['decision'], 'run', check) - if enforced: - self.assertIn('cover_this_event', queue['reason'], check) + self.assertEqual(queue['decision'], 'skip' if enforced else 'run', check) def input_scenario(self, **per_check): """A base whose policy declares `inputs.txt` as the only input of every check; head 1 has a note.""" @@ -1551,7 +1614,9 @@ class EndToEnd(unittest.TestCase): def test_a_record_of_another_tree_with_the_same_input_key_skips_and_says_so(self): base, key, record = self.input_scenario() self.f.commit('an undeclared file changed') - fields, outputs = self.run_gate(base) + queue, outputs = self.run_gate(base) + self.assertEqual((queue['decision'], queue['outcome'], outputs['refuse']), ('run', 'satisfied', 'false')) + fields, outputs = self.run_gate(base, EVENT_NAME='pull_request') self.assertEqual((fields['decision'], fields['outcome'], fields['match'], fields['source_inputs'], outputs['record']), ('skip', 'satisfied', 'inputs', key, record)) (self.f.repo / 'inputs.txt').write_text('a declared file changed') @@ -1562,7 +1627,7 @@ class EndToEnd(unittest.TestCase): def test_sampling_keeps_running_a_deterministic_share_of_matching_runs(self): base, key, record = self.input_scenario(sample_one_in=1) self.f.commit('two') - self.assertEqual(self.run_gate(base)[0]['decision'], 'skip') + self.assertEqual(self.run_gate(base, EVENT_NAME='pull_request')[0]['decision'], 'skip') # The sample is a function of (check, input key, head). The heads here are FIXED object ids, never the checkout's real commit # (whose id changes with the clock), so the outcome of every run below is the same on every machine and every day. Twelve # fixed heads at one in three include both outcomes for this scenario's key; the test says so before it relies on it. @@ -1573,15 +1638,15 @@ class EndToEnd(unittest.TestCase): for head in heads} self.assertEqual(set(expected.values()), {'skip', 'run'}) for head in heads: - first = self.run_gate(base, HEAD_SHA=head)[0] + first = self.run_gate(base, EVENT_NAME='pull_request', HEAD_SHA=head)[0] self.assertEqual(first['decision'], expected[head], head) - self.assertEqual(first['decision'], self.run_gate(base, HEAD_SHA=head)[0]['decision']) # a retry behaves the same + self.assertEqual(first['decision'], self.run_gate(base, EVENT_NAME='pull_request', HEAD_SHA=head)[0]['decision']) if first['decision'] == 'run': self.assertEqual(first['outcome'], 'satisfied') self.assertTrue(first['reason'].startswith('satisfied_but_not_skipped:_sampled:'), first['reason']) # Control: the real commit does not matter. The same fixed head gives the same decision after the checkout moved on. self.f.commit('the checkout moved') - again = self.run_gate(base, HEAD_SHA=heads[0])[0] + again = self.run_gate(base, EVENT_NAME='pull_request', HEAD_SHA=heads[0])[0] self.assertEqual(again['decision'], expected[heads[0]]) self.assertTrue(gate.sampled({'check': 'c', 'head': 'h'}, {'sample_one_in': 'x'}, 't')) # unreadable: sample self.assertFalse(gate.sampled({'check': 'c', 'head': 'h'}, {'sample_one_in': 1}, 't')) @@ -1759,28 +1824,23 @@ class EndToEnd(unittest.TestCase): self.wrapper_note(commit, tree, lock, inputs={'toolchain_sha256': ''}) self.assertRuns(base) - def test_one_fractal_web_shard_note_cannot_satisfy_the_check_and_the_whole_check_note_covers_pull_requests_only(self): - """Each fractal-web-execution shard signs its own note; the policy accepts only the whole-check note the evidence note job - signs from the lane's jobs. One shard's note, even shard 1 of 2 or a pull request's only shard, never satisfies the - check. A pull request's one-shard whole-check note covers a pull request run (it ran everything a pull request runs); - the policy keeps it out of the merge queue with its events (pull_request only).""" + def test_one_fractal_web_shard_note_cannot_satisfy_the_whole_check(self): + """A shard cannot substitute for the signed whole-check aggregate, on either event.""" entry = json.loads(root_config('evidence-skip-policy.json').read_text())['checks']['fractal-web'] - self.assertIs(entry['enforce'], True) # turned on for pull requests only; the merge queue keeps running it - self.assertEqual(entry['events'], ['pull_request']) + self.assertIs(entry['enforce'], True) + self.assertEqual(entry['events'], ['pull_request', 'merge_group']) self.assertEqual(entry.get('require', 'off'), 'off') self.assertEqual(entry['trust_levels'], ['fleet']) self.assertEqual(sorted(entry['kinds']), ['wrapper']) # the always-emitted fractal-web job passes without running the lane - # Host and agent levels admit only the local one-shard whole-check note (scripts/ci-local-web), for pull requests, bound to its - # lane; CI's own notes stay fleet-signed, and a single CI shard's note still never satisfies the check (below). + # Host and agent notes retain the exact local command and lane binding on both events. wrapper = entry['kinds']['wrapper'] self.assertEqual(wrapper['trust_levels'], ['fleet', 'host', 'agent', 'queue']) self.assertEqual(sorted(wrapper['by_level']), ['agent', 'host', 'queue']) for name, level in wrapper['by_level'].items(): self.assertEqual(level['command'], ['nix', 'develop', '.#web', '-c', 'bash', 'scripts/ci-local-web']) - # host and agent cover pull requests only; the (inert) queue level covers the merge queue only, like the other host checks'. - # A host note is held to the lane as its own tree defines it (a change may edit the lane: our hosts sign it too). - self.assertEqual((level['events'], level['lane']), (['merge_group' if name == 'queue' else 'pull_request'], - 'tree' if name == 'host' else True), name) + self.assertEqual((level['events'], level['lane']), + (['merge_group'] if name == 'queue' else ['pull_request', 'merge_group'], + 'tree' if name == 'host' else True), name) self.assertRegex((ROOT / '.github/workflows/fleet.yml').read_text(), re.compile(r'^ evidence-gate-fractal-web:', re.M)) shard_inputs = {'job': 'fractal-web', 'steps': 'run', 'step_outcomes': 'run=success'} for shard in ({'kit_shard': '1', 'kit_shard_count': '2', 'kit_play_set': 'all', 'app_test_set': 'all'}, @@ -2107,6 +2167,65 @@ class EndToEnd(unittest.TestCase): self.assertEqual((fields['log'], fields['decision']), ('unlogged_pass', plain['decision']), 'shadow only') +class HistoryFetch(unittest.TestCase): + """The generated "Fetch the whole history without evidence tags" step, run on a one-commit clone whose remote drops + transfers: a dropped transfer is tried once more, and a fetch that still fails leaves the checkout as it was.""" + + @staticmethod + def script(): + text = (ROOT / '.github/workflows/fleet.yml').read_text() + step = re.search(r'- name: Fetch the whole history without evidence tags\n(?:\s+(?!run:)[a-z-]+:.*\n|\s+[A-Z_]+:.*\n)*' + r'(\s+)run: \|\n((?:\1 .*\n|\n)+)', text) + indent = len(step.group(1)) + 2 + return ''.join(line[indent:] if line.strip() else '\n' for line in step.group(2).splitlines(keepends=True)) + + def fetched(self, fails, message): + tmp = Path(tempfile.mkdtemp(prefix='history-fetch-')) + self.addCleanup(shutil.rmtree, tmp, ignore_errors=True) + git = lambda *args, cwd=tmp: subprocess.run(['git', *args], cwd=cwd, check=True, capture_output=True, text=True).stdout + git('init', '-q', '--bare', 'origin.git') + git('init', '-q', 'seed') + for number in range(3): + (tmp / 'seed/file').write_text(f'{number}\n') + git('add', 'file', cwd=tmp / 'seed') + git('-c', 'user.name=t', '-c', 'user.email=t@example.com', 'commit', '-q', '-m', f'c{number}', cwd=tmp / 'seed') + git('push', '-q', str(tmp / 'origin.git'), 'HEAD:refs/heads/main', cwd=tmp / 'seed') + git('clone', '-q', '--depth', '1', '--branch', 'main', f'file://{tmp}/origin.git', 'clone') + # The remote answers through a command that drops the first `fails` transfers with `message`, as GitHub does. + (tmp / 'remote').write_text('n=$(cat "$COUNT" 2>/dev/null || echo 0); echo $((n + 1)) > "$COUNT"\n' + 'if [ "$n" -lt "$FAILS" ]; then echo "$MESSAGE" >&2; exit 128; fi\nexec "$1" "$2"\n') + clone = tmp / 'clone' + git('config', 'protocol.ext.allow', 'always', cwd=clone) + git('remote', 'set-url', 'origin', f'ext::sh {tmp}/remote %S {tmp}/origin.git', cwd=clone) + env = {**os.environ, 'FETCH_TOKEN': 'unused', 'COUNT': str(tmp / 'count'), 'FAILS': str(fails), 'MESSAGE': message} + run = subprocess.run(['bash', '--noprofile', '--norc', '-eo', 'pipefail', '-c', self.script()], cwd=clone, env=env, + capture_output=True, text=True) + attempts = int((tmp / 'count').read_text()) if (tmp / 'count').exists() else 0 + intact = subprocess.run(['git', 'fsck', '--connectivity-only', '--no-dangling'], cwd=clone, capture_output=True).returncode == 0 + # A checkout left without its marker but with only one commit's objects cannot even count its history: None. + listed = subprocess.run(['git', 'rev-list', '--count', 'HEAD'], cwd=clone, capture_output=True, text=True) + commits = int(listed.stdout) if listed.returncode == 0 else None + shallow = Path(git('rev-parse', '--git-path', 'shallow', cwd=clone).strip()) + return run, attempts, intact, commits, (clone / shallow).exists() + + def test_a_dropped_transfer_is_fetched_again_once(self): + run, attempts, intact, commits, shallow = self.fetched(1, 'error: RPC failed; HTTP 408 curl 22 The requested URL returned error: 408') + self.assertEqual(run.returncode, 0, run.stderr) + self.assertEqual((attempts, intact, commits, shallow), (2, True, 3, False)) + self.assertIn('trying once more', run.stderr) + + def test_a_fetch_that_keeps_failing_fails_red_and_leaves_the_one_commit_checkout_intact(self): + run, attempts, intact, commits, shallow = self.fetched(2, 'error: RPC failed; HTTP 408 curl 22 The requested URL returned error: 408') + self.assertNotEqual(run.returncode, 0) + self.assertEqual((attempts, intact, commits, shallow), (2, True, 1, True)) + + def test_a_failure_that_is_not_the_transport_is_not_tried_again(self): + run, attempts, intact, commits, shallow = self.fetched(1, "fatal: Authentication failed for 'https://github.com/example/repo/'") + self.assertNotEqual(run.returncode, 0) + self.assertEqual((attempts, intact, commits, shallow), (1, True, 1, True)) + self.assertNotIn('trying once more', run.stderr) + + class WorkflowStructure(unittest.TestCase): """The loops factory named, checked on the workflow files themselves.""" @@ -2237,11 +2356,13 @@ class WorkflowStructure(unittest.TestCase): expression = expression[3:-2].strip() for gate, value in gates.items(): expression = expression.replace(f"needs.evidence-gate-{gate}.outputs.skip == 'true'", str(value)) + expression = expression.replace(f"needs.evidence-gate-{gate}.outputs.skip != 'true'", str(not value)) for gate in ('linux-gate', 'rust-tests', 'typescript-client', 'genie-freshness', 'ci-script-tests', 'fractal-web'): expression = expression.replace(f"needs.evidence-gate-{gate}.outputs.refuse == 'true'", str(gate in refused)) expression = expression.replace("github.event_name != 'pull_request' || github.event.pull_request.draft == false", 'True') expression = expression.replace('!cancelled()', 'True') expression = expression.replace("github.event_name == 'merge_group'", str(event == 'merge_group')) + expression = expression.replace("github.event_name", repr(event)) expression = expression.replace('!(', 'not (').replace('&&', ' and ').replace('||', ' or ') return bool(eval(expression, {'__builtins__': {}})) @@ -2393,7 +2514,7 @@ class WorkflowStructure(unittest.TestCase): with tempfile.TemporaryDirectory() as scratch: work = Path(scratch) (work / '.github/workflows').mkdir(parents=True) - for name in ('queue-watch.ts', 'ready-pr.ts', 'build-snapshot.ts', 'cache-audit.ts', 'nix-sandbox-scope.ts', 'workspace-ci.ts'): + for name in ('compat-archive.ts', 'queue-watch.ts', 'ready-pr.ts', 'build-snapshot.ts', 'cache-audit.ts', 'nix-sandbox-scope.ts', 'workspace-ci.ts', 'swift-client.ts'): shutil.copy(ROOT / '.github/workflows' / name, work / '.github/workflows' / name) (work / '.github/workflows/fleet.yml.genie.ts').write_text(source.replace(anchor, f'{probe}\n{anchor}', 1)) for name in ('repos', 'scripts'): @@ -2673,24 +2794,40 @@ class WorkflowStructure(unittest.TestCase): self.assertEqual(json.loads(shard_env['STEP_COMMAND'].strip().strip("'")), FRACTAL_WEB_SHARD_ARGV) self.assertNotEqual(FRACTAL_WEB_SHARD_ARGV, rule['command']) - def test_the_fractal_web_lane_skips_its_jobs_only_for_trusted_pull_request_evidence(self): - """The semantic check remains emitted; only PR execution avoids runner allocation. - Queue and manual execution retain their complete proof and shard coverage.""" + def test_the_fractal_web_lane_honors_trusted_pr_and_queue_skips_but_runs_on_main_and_manual(self): + """Evidence suppresses both execution and receipts on PR/queue events. + Push and manual runs cannot receive a trusted skip and retain full hosted proof.""" blocks = self.blocks() execution, verdict = blocks['fractal-web-execution'], blocks['fractal-web'] self.assertIn('evidence-gate-fractal-web', self.needs_of(execution)) self.assertIn('evidence-gate-fractal-web', self.needs_of(verdict)) condition = self.job_condition(execution) self.assertIn("github.event_name == 'pull_request'", condition) - self.assertIn("needs.evidence-gate-fractal-web.outputs.skip == 'true'", condition) + self.assertIn("needs.evidence-gate-fractal-web.outputs.skip != 'true'", condition) self.assertIn("!(", condition) self.assertNotIn('evidence-gate', self.job_condition(verdict)) self.assertNotIn('evidence covered by a trusted note', execution) self.assertIn("EVIDENCE_SKIP: ${{ needs.evidence-gate-fractal-web.outputs.skip }}", verdict) self.assertIn("EVENT_NAME: ${{ github.event_name }}", verdict) + ready = condition.replace("needs.fractal-web-changes.result == 'success'", 'True') \ + .replace("needs.fractal-web-changes.outputs.relevant == 'true'", 'True') \ + .replace("needs.genie-freshness.result == 'success'", 'True') + for event in ('pull_request', 'merge_group', 'push', 'workflow_dispatch'): + context, result, tree = Decide().good() + context['event'] = event + allowed = gate.decide(context, 'satisfied', result, tree)[0] == 'skip' + self.assertEqual(allowed, event in ('pull_request', 'merge_group')) + self.assertEqual(self.truth(ready, {'fractal-web': allowed}, event), not allowed) + self.assertTrue(self.truth(ready, {'fractal-web': False}, event)) for name, condition in self.steps_of('fractal-web'): if name == 'Require every kit shard and exact full-index coverage': - self.assertNotIn('evidence-gate', condition) + self.assertIn("needs.evidence-gate-fractal-web.outputs.skip != 'true'", condition) + sharded = condition.replace("needs.fractal-web-changes.outputs.kit-inputs == 'true'", 'True') \ + .replace("needs.fractal-web-changes.outputs.relevant == 'true'", 'True') + for event in ('merge_group', 'push', 'workflow_dispatch'): + skip = event == 'merge_group' + self.assertEqual(self.truth(sharded, {'fractal-web': skip}, event), not skip) + self.assertTrue(self.truth(sharded, {'fractal-web': False}, event)) elif name == 'Require detection and, for web changes, execution': self.assertIsNone(condition, 'the verdict always runs') diff --git a/scripts/ci-signed-tree b/scripts/ci-signed-tree index 0939b74..77a5606 100755 --- a/scripts/ci-signed-tree +++ b/scripts/ci-signed-tree @@ -10,6 +10,13 @@ It asks `evidence-verify --require-all` about the head tree and prints ONE decis signed-tree v1 decision=ok|would-reject|reject|skipped|error mode=shadow pr=N head=SHA tree=SHA base=SHA reason=... lanes=name:state,... +followed, when a fetch failed, by one `signed-tree v1 error WHAT: STDERR` line each, so `notes-fetch-failed` says why. + +Like the enforced gate (scripts/ci-evidence-gate), it holds the head to each lane's definition at the pull request's BRANCH POINT +(`--lane-base`), not at the base's tip: a branch that only lags main changed no lane file itself (docs/st3/gate-evidence.md, "Which +lane definition a pull request is held to"). Finding the branch point deepens the shallow checkout's history without blobs, as the +gate does, and fetches only the lane file's blob at that commit. Without a branch point the tip is used and the reason says so. + `decision` mirrors the verifier (`reject` only if a check whose policy says `require: reject` is unsatisfied) but this job is SHADOW ONLY: it exits 0 whatever it finds, is not a required check, and reads no secret (read-only token, no credentials in the checkout). It is a no-op (no notes fetched) when no check in the base policy says `require`, and for forks and drafts. @@ -37,12 +44,55 @@ REPOSITORY = re.compile(r'^[A-Za-z0-9._-]+/[A-Za-z0-9._-]+$') VERIFIER_TIMEOUT = int(os.environ.get('SIGNED_TREE_VERIFIER_TIMEOUT', '60')) -def run(*command, timeout=120): +def run(*command, timeout=120, env=None): + return run_err(*command, timeout=timeout, env=env)[:2] + + +def run_err(*command, timeout=120, env=None): + """(exit status, stdout, stderr); a command that cannot start or times out is 124 with the reason as its stderr.""" try: - done = subprocess.run(command, capture_output=True, timeout=timeout) - return done.returncode, done.stdout - except (OSError, subprocess.TimeoutExpired): - return 124, b'' + done = subprocess.run(command, capture_output=True, timeout=timeout, env=dict(os.environ, **env) if env else None) + return done.returncode, done.stdout, done.stderr + except subprocess.TimeoutExpired: + return 124, b'', f'timed out after {timeout} s'.encode() + except OSError as error: + return 124, b'', str(error).encode() + + +def stderr_text(stderr): + """A failed command's stderr on one bounded line.""" + text = ' '.join(stderr.decode(errors='replace').split()) + return (text or 'no stderr')[:400] + + +def branch_point(base, head, pr): + """(the pull request's branch point, None), or (None, why) when it cannot be read. + + The checkout and the head are depth 1. Their history is deepened without blobs (commits and trees: what `merge-base` and the + verifier's definition paths read), and only the lane file's blob at the branch point is fetched if it is missing, with no + negotiation: a server would otherwise leave out a blob an advertised commit reaches.""" + _, shallow = run('git', 'rev-parse', '--is-shallow-repository') + if shallow.strip() == b'true': + code, _, stderr = run_err('git', 'fetch', '--no-tags', '-q', '--filter=blob:none', '--unshallow', 'origin', base, + f'refs/pull/{pr}/head', timeout=180) + if code != 0: + return None, ('history-fetch-failed', stderr) + code, point = run('git', 'merge-base', base, head) + point = point.decode().strip() + if code != 0 or not OID.match(point): + return None, ('no-branch-point', b'') + path = config_path(point, LANES) + if path is None: + return point, None + code, blob = run('git', 'rev-parse', '--verify', f'{point}:{path}') + blob = blob.decode().strip() + if code != 0 or not OID.match(blob): + return None, ('no-lane-file-at-the-branch-point', b'') + if run('git', 'cat-file', '-e', blob, env={'GIT_NO_LAZY_FETCH': '1'})[0] != 0: + code, _, stderr = run_err('git', '-c', 'fetch.negotiationAlgorithm=noop', 'fetch', '--no-tags', '-q', 'origin', blob) + if code != 0: + return None, ('lane-file-fetch-failed', stderr) + return point, None # arn:lmig:smalltalk:2026-10-27-dot-fractal-config — no candidate or sibling fallback. @@ -60,10 +110,11 @@ def field(value): return text[:200] -def emit(context, decision, reason, tree='-', base='-', lanes=None, seconds=0.0): +def emit(context, decision, reason, tree='-', base='-', lanes=None, seconds=0.0, errors=()): names = ','.join(f'{name}:{state}' for name, state in sorted((lanes or {}).items())) or '-' text = (f"signed-tree v1 decision={decision} mode=shadow pr={field(context.get('pr'))} head={field(context.get('head'))} " f"tree={field(tree)} base={field(base)} reason={field(reason)} lanes={field(names)} seconds={seconds:.1f}") + text = '\n'.join([text, *(f'signed-tree v1 error {what}: {stderr_text(stderr)}' for what, stderr in errors)]) print(text) for variable, body in (('GITHUB_STEP_SUMMARY', text + '\n'), ('GITHUB_OUTPUT', f'decision={decision}\n')): target = os.environ.get(variable) @@ -113,9 +164,11 @@ def main(): if code != 0 or not OID.match(tree): return emit(context, 'error', 'no-head-tree', base=base, seconds=elapsed()) # Only this tree's records (one prefix), plus the input-keyed records of each required check that declares inputs: never the namespace. - code, _ = run('git', 'fetch', '--no-tags', '--depth=1', '--no-filter', '-q', 'origin', f'+{TAGS}/{tree}/*:{TAGS}/{tree}/*', - f'+{NOTES}/{tree}/*:{NOTES}/{tree}/*') - notes = 'ok' if code == 0 else 'failed' + errors = [] + code, _, stderr = run_err('git', 'fetch', '--no-tags', '--depth=1', '--no-filter', '-q', 'origin', f'+{TAGS}/{tree}/*:{TAGS}/{tree}/*', + f'+{NOTES}/{tree}/*:{NOTES}/{tree}/*') + if code != 0: + errors.append(('notes-fetch', stderr)) for check, entry in sorted(checks.items()): if isinstance(entry, dict) and entry.get('require', 'off') != 'off' and entry.get('inputs') is not None and re.fullmatch(r'[A-Za-z0-9._-]+', check): code, out = run(sys.executable, VERIFIER, '--repo', '.', '--revision', context['head'], '--check', check, '--project', '--policy-from', base, @@ -126,24 +179,35 @@ def main(): key = None if isinstance(key, str) and OID.match(key): prefix, tags = f'{INPUT_NOTES}/{check}/{key}', f'{TAGS}/input/{check}/{key}' - if run('git', 'fetch', '--no-tags', '--depth=1', '--no-filter', '-q', 'origin', f'+{tags}/*:{tags}/*', f'+{prefix}/*:{prefix}/*')[0] != 0: - notes = 'failed' + code, _, stderr = run_err('git', 'fetch', '--no-tags', '--depth=1', '--no-filter', '-q', 'origin', f'+{tags}/*:{tags}/*', + f'+{prefix}/*:{prefix}/*') + if code != 0: + errors.append((f'notes-fetch:{check}', stderr)) + notes_failed = bool(errors) command = [sys.executable, VERIFIER, '--repo', '.', '--tree', tree, '--require-all', '--keys-from', base, '--policy-from', base, '--repository', repository] + lane_base = None if config_path(base, LANES) is not None: command += ['--lanes-from', base] + lane_base, missing = branch_point(base, context['head'], context['pr']) + if lane_base is not None: + command += ['--lane-base', lane_base] + elif missing[1]: + errors.append(missing) code, out = run(*command, timeout=VERIFIER_TIMEOUT) try: result = json.loads(out) except ValueError: - return emit(context, 'error', f'verifier-exit-{code}-without-json', tree=tree, base=base, seconds=elapsed()) + return emit(context, 'error', f'verifier-exit-{code}-without-json', tree=tree, base=base, seconds=elapsed(), errors=errors) if result.get('decision') == 'error': - return emit(context, 'error', f'verifier:{result.get("reason", "error")}', tree=tree, base=base, seconds=elapsed()) + return emit(context, 'error', f'verifier:{result.get("reason", "error")}', tree=tree, base=base, seconds=elapsed(), errors=errors) lanes = {name: ('ok' if lane.get('satisfied') else lane.get('reason', 'unsatisfied')) for name, lane in result.get('lanes', {}).items()} reason = 'satisfied' if result.get('satisfied') else 'unsatisfied' - if notes != 'ok': + if notes_failed: reason += ',notes-fetch-failed' - return emit(context, result.get('decision', 'error'), reason, tree=tree, base=base, lanes=lanes, seconds=elapsed()) + if lane_base is None and config_path(base, LANES) is not None: + reason += f',lanes-at-the-base-tip:{missing[0]}' + return emit(context, result.get('decision', 'error'), reason, tree=tree, base=base, lanes=lanes, seconds=elapsed(), errors=errors) except Exception as error: # a shadow job never fails return emit(context, 'error', f'{type(error).__name__}', seconds=elapsed()) diff --git a/scripts/ci-signed-tree-test b/scripts/ci-signed-tree-test index 078b3eb..9161297 100755 --- a/scripts/ci-signed-tree-test +++ b/scripts/ci-signed-tree-test @@ -60,11 +60,18 @@ class SignedTree(unittest.TestCase): defaults.update(kwargs) return self.f.note(head, tree, self.ci, 'machine/ci1', **defaults) - def run_job(self, head, number=1, base_changes=None, **env): - """Run the script in a fresh clone of the repository at main, the way the workflow's checkout of the base branch is.""" + def run_job(self, head, number=1, base_changes=None, shallow=False, **env): + """Run the script in a fresh clone of the repository at main, the way the workflow's checkout of the base branch is: with + `shallow`, at depth 1 as actions/checkout makes it, from a server that serves partial and by-object fetches as GitHub does.""" checkout = self.f.root / f'base-checkout-{len(list(self.f.root.glob("base-checkout-*")))}' # --no-tags: actions/checkout fetches no tags at its default depth, so an evidence tag is here only if the job fetched it. - subprocess.run(['git', 'clone', '-q', '--no-tags', str(self.f.repo), str(checkout)], check=True, capture_output=True) + if shallow: + for key in ('uploadpack.allowFilter', 'uploadpack.allowAnySHA1InWant'): + self.f.git('config', key, 'true') + source = ['--depth=1', f'file://{self.f.repo}'] + else: + source = [str(self.f.repo)] + subprocess.run(['git', 'clone', '-q', '--no-tags', *source, str(checkout)], check=True, capture_output=True) self.last_checkout = checkout summary, output = self.f.root / 'summary', self.f.root / 'output' summary.write_text('') @@ -79,8 +86,10 @@ class SignedTree(unittest.TestCase): status = subprocess.run(['git', '-C', str(checkout), 'status', '--porcelain'], capture_output=True).stdout self.assertEqual(status, b'', 'the pull request must never be checked out into the base checkout') self.assertEqual(subprocess.run(['git', '-C', str(checkout), 'rev-parse', 'HEAD'], capture_output=True).stdout, before) - line = done.stdout.decode().strip() - self.assertEqual(line, summary.read_text().strip()) + text = done.stdout.decode().strip() + self.assertEqual(text, summary.read_text().strip()) + line, *self.errors = text.splitlines() + self.assertTrue(all(error.startswith('signed-tree v1 error ') for error in self.errors), text) fields = dict(part.split('=', 1) for part in line.split()[2:]) self.assertTrue(line.startswith('signed-tree v1 '), line) self.assertEqual(output.read_text().strip(), f'decision={fields["decision"]}') @@ -229,6 +238,52 @@ class SignedTree(unittest.TestCase): fields = self.run_job(head) self.assertEqual((fields['decision'], fields['reason']), ('error', 'verifier:error')) # a malformed require is refused, not ignored + def lagging_pull_request(self): + """A head that changes no lane file, branched before main changed its lane's definition, with a note carrying the lane digest + at its branch point: what scripts/st-test signs for a branch that only lags main.""" + point = self.f.git('rev-parse', 'main') + head, tree = self.pull_request({'src.txt': 'behind main\n'}) + (self.f.repo / 'scripts/lane.sh').write_text('lane, changed on main\n') + self.f.git('commit', '-q', '-am', 'main changes the lane definition') + rules = support.evidence_verify.lane_rules((self.f.repo / '.st/required-lanes.json').read_text(), 'unit') + digest = support.evidence_verify.lane_definition(str(self.f.repo), self.f.git('rev-parse', f'{point}^{{tree}}'), tree, rules)['sha256'] + self.note(head, tree, inputs={'lane_definition_sha256': digest}) + return head, tree + + def test_a_head_that_only_lags_a_lane_change_on_main_is_held_to_its_branch_point(self): + head, _ = self.lagging_pull_request() + for shallow in (False, True): + fields = self.run_job(head, shallow=shallow) + self.assertEqual((fields['decision'], fields['reason'], fields['lanes']), ('ok', 'satisfied', 'unit:ok'), shallow) + self.assertEqual(self.errors, []) + + def test_without_its_branch_point_the_head_is_held_to_the_tip_and_the_reason_says_so(self): + _, tree = self.lagging_pull_request() + # The same tree, and so the same note, as a commit with no history in common with main: it has no branch point. + head = self.f.git('commit-tree', tree, '-m', 'unrelated history') + self.f.git('update-ref', 'refs/pull/2/head', head) + fields = self.run_job(head, number=2, shallow=True) + self.assertEqual((fields['decision'], fields['lanes']), ('would-reject', 'unit:definition_changed')) + self.assertTrue(fields['reason'].endswith(',lanes-at-the-base-tip:no-branch-point'), fields['reason']) + + def test_a_failed_notes_fetch_says_why(self): + head, tree = self.pull_request({'src.txt': 'unfetchable\n'}) + # A record ref the server advertises but cannot send. + ref = self.f.repo / '.git' / 'refs' / 'notes' / 'evidence' / tree / ('0' * 40) + ref.parent.mkdir(parents=True) + ref.write_text('1' * 40 + '\n') + fields = self.run_job(head) + self.assertEqual(fields['decision'], 'would-reject') + self.assertTrue(fields['reason'].endswith(',notes-fetch-failed'), fields['reason']) + self.assertEqual(len(self.errors), 1, self.errors) + self.assertTrue(self.errors[0].startswith('signed-tree v1 error notes-fetch: '), self.errors) + self.assertNotIn('no stderr', self.errors[0]) + # A verifier that then gives no answer still leaves the fetch's error beside its own. + fields = self.run_job(head, SIGNED_TREE_VERIFIER_TIMEOUT='0') + self.assertEqual((fields['decision'], fields['reason']), ('error', 'verifier-exit-124-without-json')) + self.assertEqual(len(self.errors), 1, self.errors) + self.assertTrue(self.errors[0].startswith('signed-tree v1 error notes-fetch: '), self.errors) + if __name__ == '__main__': unittest.main() diff --git a/scripts/evidence-flip-ready b/scripts/evidence-flip-ready index 37b23f2..ce94216 100755 --- a/scripts/evidence-flip-ready +++ b/scripts/evidence-flip-ready @@ -140,14 +140,14 @@ def fields_of(text): return dict(part.split('=', 1) for part in text.split(' ') if '=' in part) -def pr_runs(github, since, until, tries=4): - """The window's pull request runs, newest first, page by page until a page reaches back before the window. The created= +def pr_runs(github, since, until, tries=4, workflow='fleet.yml'): + """The window's pull request runs of `workflow` (Workspace CI by default), newest first, page by page until a page reaches back before the window. The created= filter is not used: it answers an empty listing for about one request in four (2026-10-22: 205, 205, 0, ... for the same request), which would read as missing lines. A page that is empty or short of what the listing's total says it holds is asked again, then the run is unreadable; so is a window of more than the listing's 1,000 runs.""" runs, page = {}, 1 while True: - path = f'repos/{github.repo}/actions/workflows/fleet.yml/runs?event=pull_request&per_page=100&page={page}' + path = f'repos/{github.repo}/actions/workflows/{workflow}/runs?event=pull_request&per_page=100&page={page}' for attempt in range(tries): out = json.loads(github.gh('api', path)) batch, total = out.get('workflow_runs') or [], min(out.get('total_count') or 0, 1000) diff --git a/scripts/evidence-flip-ready-test b/scripts/evidence-flip-ready-test index 675ec53..16e5053 100755 --- a/scripts/evidence-flip-ready-test +++ b/scripts/evidence-flip-ready-test @@ -17,6 +17,7 @@ import sys import tempfile import time import unittest +from unittest import mock sys.dont_write_bytecode = True ROOT = Path(__file__).resolve().parent.parent @@ -154,6 +155,24 @@ def remove_tree(tmp, attempts=5): time.sleep(0.2 * (attempt + 1)) +class Cleanup(unittest.TestCase): + def test_cleanup_records_backoff_until_success_or_exhaustion(self): + for failures in (0, 2, 4, 5): + with self.subTest(failures=failures): + tmp = mock.Mock() + tmp.cleanup.side_effect = [OSError("busy")] * failures + [None] + sleeps = [] + # Override the shared no-op only here; record backoff without real waiting. + with mock.patch.object(time, 'sleep', side_effect=sleeps.append): + if failures == 5: + with self.assertRaisesRegex(OSError, "busy"): + remove_tree(tmp) + else: + remove_tree(tmp) + self.assertEqual(tmp.cleanup.call_count, min(failures + 1, 5)) + self.assertEqual(sleeps, [0.2 * step for step in range(1, min(failures, 4) + 1)]) + + class FlipReady(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() diff --git a/scripts/evidence-gc b/scripts/evidence-gc index a314f2d..1340a79 100755 --- a/scripts/evidence-gc +++ b/scripts/evidence-gc @@ -4,6 +4,7 @@ scripts/evidence-gc [--repo .] [--remote origin] [--fetch] [--apply] [--main origin/main] [--retention FILE] [--as-of UNIX] [--head COMMIT ...] [--heads-from FILE] [--open-heads-from FILE] [--plan-out FILE] [--before-delete JSON_ARGV] + [--resume-state FILE [--resume-only]] Dry run by default: it reads the records in --repo and prints the plan, changing nothing. With --fetch it first fetches every `refs/tags/evidence/*`, `refs/notes/evidence/*` and `refs/notes/evidence-input/*`, main's recent history and the tip @@ -46,6 +47,13 @@ anything is deleted or migrated, with the plan file's path as its last argument, here (scripts/evidence-gc-timer runs `evidence-log-timer gc --plan FILE`), so the log says GC removed a tree before its records disappear. If it fails, nothing is deleted (`refused`, exit 1). +RESUMING (--resume-state FILE, with --apply --remote): persist the complete plan before its hook can run. While that file exists, +finish that plan before fetching or planning another; its original repository, remote and hook must still match. Each attempt +reconfirms the hook, then lists the remote to skip deletions already completed. Every remaining operation is leased to the original +object id, and whole-group atomic deletion is unchanged. The journal stays until remote and local cleanup finish; a failure leaves +it for a later invocation. Protection-kept groups finish as reported outcomes, so the next run can plan again. A file lock prevents +two runs using the same journal. --resume-only refuses to start a new plan if another run completed this one meanwhile. Neither a retry loop nor a longer unit or hook timeout is needed. + OPEN PULL REQUEST HEADS (--open-heads-from FILE, one commit id a line, an empty file for none): their trees are live however old the head is, so a pull request that waits longer than branch_days keeps its records. With --fetch the commits are fetched by id (a fork's head is on no branch here). A missing FILE is refused. @@ -59,6 +67,7 @@ command fails, --apply refuses to delete anything. import argparse import base64 import collections +import fcntl import hashlib import importlib.machinery import importlib.util @@ -440,9 +449,11 @@ def header_for_token(): PROTECTION_REFUSED = re.compile(r'GH013: Repository rule violations') -def push_deletions(repo, remote, refs, header, token, atomic): +def push_deletions(repo, remote, refs, header, token, atomic, expected=None): """(refs the remote deleted, error text or None) of one push deleting REFS: with `atomic`, all of them or none.""" - done = git(repo, 'push', '--porcelain', *(['--atomic'] if atomic else []), remote, *[f':{ref}' for ref in refs], header=header, check=False) + leases = [f'--force-with-lease={ref}:{expected[ref]}' for ref in refs] if expected is not None else [] + done = git(repo, 'push', '--porcelain', *(['--atomic'] if atomic else []), *leases, remote, + *[f':{ref}' for ref in refs], header=header, check=False) deleted = set() for line in done.stdout.decode(errors='replace').splitlines(): parts = line.split('\t') @@ -451,10 +462,10 @@ def push_deletions(repo, remote, refs, header, token, atomic): if done.returncode == 0: return deleted, None text = (done.stdout + done.stderr).decode(errors='replace') - return deleted, (text.replace(token, '***') if token else text).strip()[-300:] + return deleted, (text.replace(token, '***') if token else text).strip() -def apply_remote(repo, remote, migrate, doomed, groups): +def apply_remote(repo, remote, migrate, doomed, groups, expected=None, replacements=None): """Make the remote match the plan without ever splitting a group: (failures, refs gone from the remote, refs its protection kept, whether the migrations' new names exist there). @@ -467,12 +478,19 @@ def apply_remote(repo, remote, migrate, doomed, groups): token, header = header_for_token() failed, deleted, protected = [], set(), set() listed = git(repo, 'ls-remote', remote, f'{TAGS}*', f'{NOTES}*', f'{INPUT_NOTES}*', header=header).stdout.decode().split('\n') - there = {line.split()[1] for line in listed if len(line.split()) == 2} + advertised = {line.split()[1]: line.split()[0] for line in listed if len(line.split()) == 2} + there = set(advertised) + if expected is not None: + changed = sorted(ref for ref, oid in expected.items() if ref in advertised and advertised[ref] != oid) + if changed: + raise Refused(f'a ref in the unfinished plan changed: {changed[0]}: nothing was deleted') deleted |= {ref for ref in doomed if ref not in there} # already gone (a run stopped halfway): nothing to push, only to forget deleted |= {old for old, _ in migrate if old not in there} - creates = [f'{old}:{new}' for old, new in migrate if new not in there] + creates = [(f'{expected[old] if expected is not None else old}:{new}', new) for old, new in migrate if new not in there] for start in range(0, len(creates), BATCH): - done = git(repo, 'push', '--porcelain', remote, *creates[start:start + BATCH], header=header, check=False) + batch = creates[start:start + BATCH] + leases = [f'--force-with-lease={new}:' for _, new in batch] if expected is not None else [] + done = git(repo, 'push', '--porcelain', *leases, remote, *[refspec for refspec, _ in batch], header=header, check=False) if done.returncode != 0: text = (done.stdout + done.stderr).decode(errors='replace') failed.append((text.replace(token, '***') if token else text).strip()[-300:]) @@ -485,10 +503,18 @@ def apply_remote(repo, remote, migrate, doomed, groups): redundant += [ref for ref in group if not ref.startswith(TAGS) and ref.rsplit('/', 1)[1] in tagged] redundant = sorted(ref for ref in set(redundant) if ref in there) for start in range(0, len(redundant), BATCH): - gone, error = push_deletions(repo, remote, redundant[start:start + BATCH], header, token, atomic=False) + batch = redundant[start:start + BATCH] + keepers = sorted({replacements[old] for old in batch if old in replacements}) if replacements else [] + if keepers: + # A successful migration push is not proof that its replacement still exists when the old name is deleted. + lines = git(repo, 'ls-remote', remote, *keepers, header=header).stdout.decode().splitlines() + current = {line.split()[1]: line.split()[0] for line in lines if len(line.split()) == 2} + if any(current.get(ref) != expected[ref] for ref in keepers): + raise Refused('a live replacement is missing or changed: old names remain pending') + gone, error = push_deletions(repo, remote, batch, header, token, atomic=False, expected=expected) deleted |= gone if error: - failed.append(error) + failed.append(error[-300:]) batches = [] for tagged in (False, True): # never in one push: a refused tag would take a tagless group down with it batch = [] @@ -502,12 +528,12 @@ def apply_remote(repo, remote, migrate, doomed, groups): if batch: batches.append(batch) for batch in batches: - gone, error = push_deletions(repo, remote, batch, header, token, atomic=True) + gone, error = push_deletions(repo, remote, batch, header, token, atomic=True, expected=expected) deleted |= gone if error and PROTECTION_REFUSED.search(error) and any(ref.startswith(TAGS) for ref in batch): protected |= set(batch) - gone elif error: - failed.append(error) + failed.append(error[-300:]) return failed, deleted, protected, True @@ -523,13 +549,175 @@ def before_delete(argv, plan_path, summary): return done.returncode == 0 -def apply_local(repo, migrate, doomed): - commands = ''.join(f'create {new} {git(repo, "rev-parse", old).stdout.decode().strip()}\n' for old, new in migrate) - commands += ''.join(f'delete {ref}\n' for ref in sorted(doomed)) # a migration's old name among them once the remote dropped it +def apply_local(repo, migrate, doomed, expected=None): + current = {line.split()[1]: line.split()[0] for line in git(repo, 'for-each-ref', '--format=%(objectname) %(refname)', + TAGS, NOTES, INPUT_NOTES).stdout.decode().splitlines()} + commands = '' + for old, new in migrate: + oid = expected[old] if expected is not None else current.get(old) + if oid is None or (new in current and current[new] != oid): + raise Refused('a local migration no longer matches the plan') + if new not in current: + commands += f'create {new} {oid}\n' + for ref in sorted(doomed): # replay after the local transaction committed is a no-op + if ref not in current: + continue + if expected is not None and current[ref] != expected[ref]: + raise Refused(f'a local ref in the unfinished plan changed: {ref}') + commands += f'delete {ref} {current[ref]}\n' git(repo, 'update-ref', '--stdin', stdin=f'start\n{commands}prepare\ncommit\n'.encode()) git(repo, 'pack-refs', '--all') +def sync_directory(path): + fd = os.open(os.path.dirname(os.path.abspath(path)), os.O_RDONLY | os.O_DIRECTORY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +def write_journal(path, value): + """The frozen plan must be durable before its hook can publish any GC lines.""" + value = {**value, 'sha256': digest(value)} + with open(path + '.new', 'w', opener=lambda name, flags: os.open(name, flags, 0o600)) as handle: + json.dump(value, handle, sort_keys=True) + handle.flush() + os.fsync(handle.fileno()) + os.replace(path + '.new', path) + sync_directory(path) + + +def remote_identity(args): + configured = git(args.repo, 'remote', 'get-url', args.remote, check=False) + return [args.remote, configured.stdout.decode().strip() if configured.returncode == 0 else args.remote] + + +def live_replacements(plan): + """Names removed without retiring a record require their original tree-prefixed replacement.""" + replacements = dict(plan['migrate']) + grouped = {ref for group in plan['groups'] for ref in group} + redundant = set(plan['delete']) - grouped + for ref, oid, layout, tree, _ in plan['inputs']['records']: + if ref in redundant and layout == 'legacy' and tree: + replacements[ref] = f'{NOTES}{tree}/{oid}' + if redundant - replacements.keys(): + raise Refused('the unfinished plan has no replacement for a live ref') + return replacements + + +def plan_oids(plan): + """Bind every planned operation to the original record object, including a migration's new name.""" + found = {r[0]: r[1] for r in plan['inputs']['records']} + found.update({r[0]: r[3] for r in plan['inputs']['index']}) + expected = {} + for ref in [*plan['delete'], *(old for old, _ in plan['migrate'])]: + oid = found.get(ref) + if (not isinstance(ref, str) or not isinstance(oid, str) or not OID.fullmatch(oid) + or not valid_evidence_ref(ref, oid)): + raise Refused('the unfinished plan has an invalid evidence ref') + expected[ref] = oid + for old, new in live_replacements(plan).items(): + if old != NOTES + expected[old] or not valid_evidence_ref(new, expected[old]) or not new.startswith(NOTES) or new.count('/') != 4: + raise Refused('the unfinished plan has an invalid migration') + expected[new] = expected[old] + return expected + + +def valid_evidence_ref(ref, oid): + for prefix in (INPUT_TAGS, INPUT_NOTES): + if ref.startswith(prefix): + parts = ref[len(prefix):].split('/') + return len(parts) == 3 and bool(CHECK.fullmatch(parts[0])) and bool(OID.fullmatch(parts[1])) and parts[2] == oid + for prefix in (TAGS, NOTES): + if ref.startswith(prefix): + parts = ref[len(prefix):].split('/') + return ((len(parts) == 1 and prefix == NOTES and parts[0] == oid) + or (len(parts) == 2 and bool(OID.fullmatch(parts[0])) and parts[1] == oid)) + return False + + +def read_journal(args): + try: + with open(args.resume_state) as handle: + value = json.load(handle) + if not isinstance(value, dict): + raise ValueError('journal is not an object') + checksum = value.pop('sha256') + if checksum != digest(value) or value['version'] != 1: + raise ValueError('version or checksum differs') + if value['repo'] != os.path.realpath(git(args.repo, 'rev-parse', '--absolute-git-dir').stdout.decode().strip()) or value['remote'] != remote_identity(args): + raise ValueError('repository or remote differs') + plan = value['plan'] + if not isinstance(plan['operation_id'], str) or not re.fullmatch(r'[0-9a-f]{64}', plan['operation_id']): + raise ValueError('operation id is invalid') + if (plan['inputs_sha256'] != digest(plan['inputs']) + or plan['plan_sha256'] != digest({k: plan[k] for k in ('delete', 'migrate', 'operation_id')})): + raise ValueError('plan digest differs') + if not isinstance(value['before'], (list, type(None))) or (value['before'] is not None and + (not value['before'] or not all(isinstance(a, str) for a in value['before']))): + raise ValueError('hook is not an argv') + if args.before_delete and json.loads(args.before_delete) != value['before']: + raise ValueError('the before-delete hook differs') + plan_oids(plan) + doomed = set(plan['delete']) + if not all(isinstance(g, list) and g and set(g) <= doomed for g in plan['groups']): + raise ValueError('deletion groups differ') + return value + except (OSError, ValueError, KeyError, TypeError, IndexError, AttributeError) as error: + raise Refused(f'cannot resume {args.resume_state}: {error}: nothing was deleted') from error + + +def execute_plan(args, plan, before, resumed=False): + summary = {k: v for k, v in plan.items() if k not in ('inputs', 'delete', 'migrate', 'trees', 'keys', 'groups')} + if args.resume_state: + summary['resumed'] = resumed + if args.plan_out: + with open(args.plan_out, 'w') as handle: + json.dump({**{k: v for k, v in plan.items() if k != 'groups'}, **summary}, handle, indent=1) + if summary['unresolved_heads'] and args.apply: + summary['refused'] = 'a live head could not be resolved to a tree: nothing was changed' + print(json.dumps(summary, indent=2)) + return 1 + doomed, migrate = plan['delete'], plan['migrate'] + status = 0 + if args.apply and (doomed or migrate): + expected = plan_oids(plan) if args.resume_state else None + if args.resume_state and not resumed: + write_journal(args.resume_state, {'version': 1, 'repo': os.path.realpath(git(args.repo, 'rev-parse', '--absolute-git-dir').stdout.decode().strip()), + 'remote': remote_identity(args), 'before': before, 'plan': plan}) + if before and not before_delete(before, args.plan_out, summary): + summary['refused'] = 'the --before-delete command failed: nothing was changed' + print(json.dumps(summary, indent=2)) + return 1 + if args.remote: + failed, gone, protected, migrated = apply_remote(args.repo, args.remote, migrate, doomed, plan['groups'], expected=expected, + replacements=live_replacements(plan) if expected is not None else None) + else: + failed, gone, protected, migrated = [], set(doomed) | {old for old, _ in migrate}, set(), True + if failed: + summary['failed'] = failed + status = 1 + if protected: + summary['kept_by_protection'] = {'refs': len(protected), 'groups': sum(1 for g in plan['groups'] if protected & set(g)), + 'why': 'the remote refused deleting their tags: they stay, each group whole'} + print(f'evidence-gc: the remote\'s protection kept {len(protected)} refs of dead groups whole', file=sys.stderr) + if migrated: + apply_local(args.repo, migrate, gone, expected=expected) + summary['applied'] = True + summary['deleted_refs'] = len(gone) + if args.resume_state: + summary['unfinished'] = bool(failed) # protection-kept groups are a completed outcome, as without a journal + if not summary['unfinished']: + os.unlink(args.resume_state) + sync_directory(args.resume_state) + if status == 0 and plan['protected_records'] > plan['retention']['cap_records']: + summary['alert'] = f'the live set ({plan["protected_records"]} records) is over the cap ({plan["retention"]["cap_records"]}); nothing live was deleted' + status = 3 + print(json.dumps(summary, indent=2)) + return status + + def main(argv): parser = argparse.ArgumentParser(description=__doc__.split('\n\n')[0]) parser.add_argument('--repo', default='.') @@ -546,12 +734,17 @@ def main(argv): parser.add_argument('--policy-from', help='revision whose preferred evidence-skip-policy.json says which checks have inputs (default --main)') parser.add_argument('--plan-out', help='also write the plan, with its inputs digest, to this file') parser.add_argument('--before-delete', help='JSON argv run with the --plan-out path appended, after planning and before deleting') + parser.add_argument('--resume-state', help='with --apply --remote: save the plan here and finish it before planning another') + parser.add_argument('--resume-only', action='store_true', help='refuse a new plan if --resume-state is absent') try: args = parser.parse_args(argv) except SystemExit as error: return 2 if error.code else 0 + journal_lock = None try: before = None + if args.resume_only and not args.resume_state: + raise Usage('--resume-only needs --resume-state') if args.before_delete: try: before = json.loads(args.before_delete) @@ -561,6 +754,26 @@ def main(argv): raise Usage('--before-delete must be a JSON array of strings') if not args.plan_out: raise Usage('--before-delete needs --plan-out: the command is given the plan file') + if args.resume_state: + if not args.apply or not args.remote: + raise Usage('--resume-state needs --apply and --remote') + args.resume_state = os.path.abspath(args.resume_state) + if args.plan_out and (os.path.realpath(args.plan_out) == os.path.realpath(args.resume_state) + or (os.path.exists(args.plan_out) and os.path.exists(args.resume_state) + and os.path.samefile(args.plan_out, args.resume_state))): + raise Usage('--plan-out and --resume-state must be different files') + journal_lock = open(args.resume_state + '.lock', 'a', opener=lambda name, flags: os.open(name, flags, 0o600)) + try: + fcntl.flock(journal_lock, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + raise Refused('another GC run holds the unfinished-plan lock') + if os.path.lexists(args.resume_state): + saved = read_journal(args) + if saved['before'] and not args.plan_out: + raise Usage('resuming a before-delete hook needs --plan-out') + return execute_plan(args, saved['plan'], saved['before'], resumed=True) + if args.resume_only: + raise Refused('the unfinished plan is gone: a new run must read the current open heads') open_heads = [] if args.open_heads_from: try: @@ -648,41 +861,18 @@ def main(argv): 'main_window': sum(1 for r in live.values() if r == 'main-window')}, 'crate_keys': crates, 'trees_deleted': len(plan['trees']), 'keys_deleted': len(plan['keys']), **{k: v for k, v in plan.items() if k not in ('delete', 'migrate', 'trees', 'keys', 'groups')}} - if args.plan_out: - with open(args.plan_out, 'w') as handle: - json.dump({**summary, 'inputs': inputs, **{k: plan[k] for k in ('delete', 'migrate', 'trees', 'keys')}}, handle, indent=1) - status = 0 - doomed, migrate = plan['delete'], plan['migrate'] - if unresolved and args.apply: - summary['refused'] = 'a live head could not be resolved to a tree: nothing was changed' - status = 1 - elif args.apply and (doomed or migrate) and before and not before_delete(before, args.plan_out, summary): - summary['refused'] = 'the --before-delete command failed: nothing was changed' - status = 1 - elif args.apply and (doomed or migrate): - if args.remote: - failed, gone, protected, migrated = apply_remote(args.repo, args.remote, migrate, doomed, plan['groups']) - else: - failed, gone, protected, migrated = [], set(doomed) | {old for old, _ in migrate}, set(), True - if failed: - summary['failed'] = failed - status = 1 - if protected: - summary['kept_by_protection'] = {'refs': len(protected), 'groups': sum(1 for g in plan['groups'] if protected & set(g)), - 'why': 'the remote refused deleting their tags: they stay, each group whole'} - print(f'evidence-gc: the remote\'s protection kept {len(protected)} refs of dead groups whole', file=sys.stderr) - if migrated: # the local refs follow exactly what the remote did - apply_local(args.repo, migrate, gone) - summary['applied'] = True - summary['deleted_refs'] = len(gone) - if status == 0 and plan['protected_records'] > retention['cap_records']: - summary['alert'] = f'the live set ({plan["protected_records"]} records) is over the cap ({retention["cap_records"]}); nothing live was deleted' - status = 3 - print(json.dumps(summary, indent=2)) - return status - except Refused as error: + full_plan = {**summary, 'inputs': inputs, **{k: plan[k] for k in ('delete', 'migrate', 'trees', 'keys', 'groups')}} + if args.resume_state: + # Identical refs can be published again after an earlier operation finished. Its GC lines are not this run's receipt. + full_plan['operation_id'] = os.urandom(32).hex() + full_plan['plan_sha256'] = digest({k: full_plan[k] for k in ('delete', 'migrate', 'operation_id')}) + return execute_plan(args, full_plan, before) + except (Refused, OSError) as error: print(f'evidence-gc: {error}', file=sys.stderr) return 2 if isinstance(error, Usage) else 1 + finally: + if journal_lock is not None: + journal_lock.close() if __name__ == '__main__': diff --git a/scripts/evidence-gc-test b/scripts/evidence-gc-test index d8b4b2c..3c7482d 100755 --- a/scripts/evidence-gc-test +++ b/scripts/evidence-gc-test @@ -3,6 +3,7 @@ index, absolute grace and the cap, fail-closed on an unresolved head, a remote, and idempotence. An open pull request head is never deleted. gc_plan itself is checked as a pure function: random worlds, any input order, whole trees only, and no verdict of a kept tree changes.""" import collections +import fcntl import importlib.machinery import importlib.util import json @@ -14,6 +15,7 @@ import subprocess import sys import time import unittest +from unittest import mock # Fixture repositories must not inherit a developer's hooks, identity or signing settings, even outside ci-script-tests. os.environ.update(GIT_CONFIG_GLOBAL='/dev/null', GIT_CONFIG_NOSYSTEM='1') @@ -758,8 +760,8 @@ sys.exit({exit_code}) done = subprocess.run(['bash', str(TIMER)], capture_output=True, text=True, env=env) self.assertEqual(done.returncode, 0, done.stderr) before = json.loads(seen.read_text()) - self.assertEqual(before['argv'][:3], ['gc', '--publish', '--plan']) - self.assertTrue(before['argv'][3].endswith('.plan.json')) + self.assertEqual(before['argv'][:4], ['gc', '--publish', '--no-wait', '--plan']) + self.assertTrue(before['argv'][4].endswith('.plan.json')) self.assertTrue(set(doomed) <= {ref.rsplit('/', 1)[1] for ref in before['tags']}, 'the log heard first') self.assertTrue(set(doomed).isdisjoint(ref.rsplit('/', 1)[1] for ref in self.remote_refs(bare, 'refs/tags/evidence'))) @@ -776,6 +778,221 @@ sys.exit({exit_code}) self.assertIn('the log signer failed', done.stderr) self.assertTrue(set(doomed) <= {ref.rsplit('/', 1)[1] for ref in self.remote_refs(bare, 'refs/tags/evidence')}) + def test_resume_after_local_commit_before_ack_keeps_migration_and_finishes_without_planning(self): + pr, head, doomed, bystander = self.lock_world() + legacy = self.record(pr[0], pr[1], 27, layout='legacy') + bare, _ = self.protected_remote('update') + self.push_all() + state, plan = self.f.root / 'resume.json', self.f.root / 'plan.json' + driver = self.f.root / 'local-commit.py' + driver.write_text(f'''import importlib.machinery, importlib.util, os, signal, sys +from pathlib import Path +loader = importlib.machinery.SourceFileLoader('gc', {str(GC)!r}) +gc = importlib.util.module_from_spec(importlib.util.spec_from_loader(loader.name, loader)) +loader.exec_module(gc) +if not Path({str(state)!r}).exists(): + original = gc.apply_local + def commit_then_stop(*args, **kwargs): + original(*args, **kwargs) + os.kill(os.getpid(), signal.SIGKILL) + gc.apply_local = commit_then_stop +else: + def no_planning(*args, **kwargs): + raise AssertionError('completed local transaction must not restart planning') + gc.read_refs = gc.protected_keys = gc.crate_keys = no_planning +sys.exit(gc.main(sys.argv[1:])) +''') + argv = [sys.executable, str(driver), '--repo', str(self.f.repo), '--main', 'main', '--head', pr[0], + '--retention', self.retention(), '--as-of', str(NOW), '--remote', 'origin', '--apply', + '--resume-state', str(state), '--plan-out', str(plan)] + stopped = subprocess.run(argv, capture_output=True, text=True, timeout=30) + self.assertEqual(stopped.returncode, -signal.SIGKILL, stopped.stderr) + self.assertTrue(state.exists()) + self.assertIn(f'refs/notes/evidence/{pr[1]}/{legacy}', self.evidence_refs()) + self.assertNotIn(f'refs/notes/evidence/{legacy}', self.evidence_refs()) + done = subprocess.run(argv, capture_output=True, text=True, timeout=30) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertTrue(json.loads(done.stdout)['resumed']) + self.assertFalse(state.exists()) + self.assertEqual(self.evidence_refs(), self.evidence_refs(bare)) + self.assertEqual({r.rsplit('/', 1)[1] for r in self.evidence_refs()}, {head, bystander, legacy}) + + def test_resume_rechecks_live_replacements_before_deleting_old_names(self): + pr, _, _, _ = self.lock_world() + migrating = self.record(pr[0], pr[1], 27, layout='legacy') + duplicate = self.record(pr[0], pr[1], 26, layout='legacy') + old_m = f'refs/notes/evidence/{migrating}' + new_m = f'refs/notes/evidence/{pr[1]}/{migrating}' + old_d = f'refs/notes/evidence/{duplicate}' + new_d = f'refs/notes/evidence/{pr[1]}/{duplicate}' + self.f.git('update-ref', new_d, duplicate) + bare, _ = self.protected_remote('update') + self.push_all() + state = self.f.root / 'resume.json' + # Lose both live replacements after the migration push, before old-name deletion. + hook = bare / 'hooks/post-receive' + hook.write_text(f'#!/bin/sh\ngit update-ref -d {new_m}\ngit update-ref -d {new_d}\n') + hook.chmod(0o755) + argv = ['--repo', str(self.f.repo), '--main', 'main', '--head', pr[0], '--retention', self.retention(), + '--as-of', str(NOW), '--remote', 'origin', '--apply', '--resume-state', str(state)] + done = subprocess.run([sys.executable, str(GC), *argv], capture_output=True, text=True, timeout=30) + self.assertEqual(done.returncode, 1, done.stderr) + self.assertIn('replacement', done.stderr) + self.assertTrue(state.exists()) + self.assertTrue({old_m, old_d} <= self.evidence_refs(bare)) + hook.unlink() + # Resume may recreate the migration target, but cannot invent the missing duplicate's proof. + done = subprocess.run([sys.executable, str(GC), *argv], capture_output=True, text=True, timeout=30) + self.assertEqual(done.returncode, 1, done.stderr) + self.assertTrue({old_m, old_d} <= self.evidence_refs(bare)) + self.assertTrue(state.exists()) + subprocess.run(['git', '-C', str(bare), 'update-ref', new_d, duplicate], check=True) + done = subprocess.run([sys.executable, str(GC), *argv], capture_output=True, text=True, timeout=30) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertFalse(state.exists()) + self.assertTrue({new_m, new_d} <= self.evidence_refs(bare)) + self.assertFalse({old_m, old_d} & self.evidence_refs(bare)) + + def test_a_resumed_plan_refuses_a_changed_ref_and_leases_the_atomic_push(self): + pr, head, doomed, bystander = self.lock_world() + bare, _ = self.protected_remote('update') + self.push_all() + state = self.f.root / 'resume.json' + changed = f'refs/tags/evidence/{"1" * 40}/{doomed[0]}' + driver = self.f.root / 'race.py' + driver.write_text(f'''import importlib.machinery, importlib.util, subprocess, sys +loader = importlib.machinery.SourceFileLoader('gc', {str(GC)!r}) +gc = importlib.util.module_from_spec(importlib.util.spec_from_loader(loader.name, loader)) +loader.exec_module(gc) +original = gc.git +def git(repo, *args, **kwargs): + if args[0] == 'push' and '--atomic' in args: + subprocess.run(['git', '-C', {str(bare)!r}, 'update-ref', {changed!r}, {bystander!r}], check=True) + return original(repo, *args, **kwargs) +gc.git = git +sys.exit(gc.main(sys.argv[1:])) +''') + argv = ['--repo', str(self.f.repo), '--main', 'main', '--head', pr[0], '--retention', self.retention(), + '--as-of', str(NOW), '--remote', 'origin', '--apply', '--resume-state', str(state)] + done = subprocess.run([sys.executable, str(driver), *argv], capture_output=True, text=True, timeout=30) + self.assertEqual(done.returncode, 1, done.stderr) + self.assertTrue(json.loads(done.stdout)['failed']) + self.assertTrue(state.exists()) + self.assertIn(changed, self.evidence_refs(bare), 'a race must not delete the replacement') + self.assertTrue(all(f'refs/tags/evidence/{str(n) * 40}/{r}' in self.evidence_refs(bare) for n, r in zip((1, 2, 3), doomed)), + 'the other groups in the refused atomic batch stay whole') + before = self.evidence_refs(bare) + done = subprocess.run([sys.executable, str(GC), *argv], capture_output=True, text=True, timeout=30) + self.assertEqual(done.returncode, 1) + self.assertIn('changed', done.stderr) + self.assertEqual(self.evidence_refs(bare), before) + subprocess.run(['git', '-C', str(bare), 'update-ref', changed, doomed[0]], check=True) + done = subprocess.run([sys.executable, str(GC), *argv], capture_output=True, text=True, timeout=30) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertFalse(state.exists()) + + def test_an_unfinished_plan_lock_and_missing_resume_only_state_fail_closed(self): + self.world() + state = self.f.root / 'resume.json' + with open(str(state) + '.lock', 'w') as lock: + fcntl.flock(lock, fcntl.LOCK_EX) + code, _, error = self.run_gc('--remote', 'origin', '--apply', '--resume-state', str(state)) + self.assertEqual(code, 1) + self.assertIn('lock', error) + code, _, error = self.run_gc('--remote', 'origin', '--apply', '--resume-state', str(state), '--resume-only') + self.assertEqual(code, 1) + self.assertIn('new run must read', error) + self.assertFalse(state.exists()) + + def test_a_frozen_plan_must_be_durable_before_its_hook_runs(self): + pr, _, _, _ = self.lock_world() + bare, _ = self.protected_remote('update') + self.push_all() + state, plan = self.f.root / 'resume.json', self.f.root / 'plan.json' + before, seen = self.recorder(bare) + driver = self.f.root / 'sync-failure.py' + driver.write_text(f'''import importlib.machinery, importlib.util, sys +loader = importlib.machinery.SourceFileLoader('gc', {str(GC)!r}) +gc = importlib.util.module_from_spec(importlib.util.spec_from_loader(loader.name, loader)) +loader.exec_module(gc) +def cannot_persist(path): + raise OSError('simulated directory fsync failure') +gc.sync_directory = cannot_persist +sys.exit(gc.main(sys.argv[1:])) +''') + remote_before = self.evidence_refs(bare) + done = subprocess.run([sys.executable, str(driver), '--repo', str(self.f.repo), '--main', 'main', '--head', pr[0], + '--retention', self.retention(), '--as-of', str(NOW), '--remote', 'origin', '--apply', + '--resume-state', str(state), '--plan-out', str(plan), '--before-delete', json.dumps(before)], + capture_output=True, text=True, timeout=30) + self.assertEqual(done.returncode, 1) + self.assertIn('fsync', done.stderr) + self.assertFalse(seen.exists()) + self.assertEqual(self.evidence_refs(bare), remote_before) + + def test_a_bad_journal_or_changed_remote_or_hook_never_authorizes_deletion(self): + pr, _, _, _ = self.lock_world() + bare, _ = self.protected_remote('update') + self.push_all() + state, plan = self.f.root / 'resume.json', self.f.root / 'plan.json' + before, _ = self.recorder(bare, exit_code=3) + base = ['--head', pr[0], '--apply', '--resume-state', str(state), '--plan-out', str(plan)] + code, _, error = self.run_gc(*base, '--remote', 'origin', '--before-delete', json.dumps(before)) + self.assertEqual(code, 1, error) + saved = state.read_text() + original = self.evidence_refs(bare) + for contents in ('{', saved.replace('"version": 1', '"version": 2')): + with self.subTest(contents=contents[:20]): + state.write_text(contents) + code, _, error = self.run_gc(*base, '--remote', 'origin') + self.assertEqual(code, 1) + self.assertIn('cannot resume', error) + self.assertEqual(self.evidence_refs(bare), original) + state.write_text(saved) + for extra in (['--remote', 'elsewhere'], ['--remote', 'origin', '--before-delete', '["true"]']): + code, _, error = self.run_gc(*base, *extra) + self.assertEqual(code, 1) + self.assertIn('differs', error) + self.assertEqual(self.evidence_refs(bare), original) + + def test_a_protection_only_outcome_finishes_the_journal_so_the_timer_plans_again(self): + env, bare, flag, head, doomed, stored, state = self.timer_world() + env['EVIDENCE_GC_LOG_TIMER'] = str(self.f.root / 'no-log-here') + flag.write_text('update deletion') + done = subprocess.run(['bash', str(TIMER)], capture_output=True, text=True, env=env, timeout=30) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertFalse((state / 'pending-plan.json').exists(), 'protection is a completed outcome, not an unfinished operation') + self.assertTrue(set(doomed) <= {r.rsplit('/', 1)[1] for r in self.evidence_refs(bare)}) + # A newly dead tagless group must still be collected on the next daily plan. + later = self.record(self.f.git('rev-parse', 'main'), 'd' * 40, 10) + ref = f'refs/notes/evidence/{"d" * 40}/{later}' + self.f.git('push', '-q', 'origin', f'{ref}:{ref}') + done = subprocess.run(['bash', str(TIMER)], capture_output=True, text=True, env=env, timeout=30) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertFalse((state / 'pending-plan.json').exists()) + self.assertEqual(len((self.f.root / 'curl.urls').read_text().splitlines()), 2) + self.assertNotIn(ref, self.evidence_refs(bare)) + self.assertTrue(set(doomed) <= {r.rsplit('/', 1)[1] for r in self.evidence_refs(bare)}) + + def test_the_timer_resumes_before_reading_new_open_heads(self): + env, bare, flag, head, doomed, stored, state = self.timer_world() + env['EVIDENCE_GC_LOG_TIMER'] = str(self.f.root / 'no-log-here') + hook = bare / 'hooks/pre-receive' + original = hook.read_text() + hook.write_text(original.replace('while read old new ref; do', + 'case "$rules" in *busy*) echo "remote temporarily unavailable" >&2; exit 1;; esac\nwhile read old new ref; do')) + flag.write_text('busy') + first = subprocess.run(['bash', str(TIMER)], capture_output=True, text=True, env=env, timeout=30) + self.assertEqual(first.returncode, 1, first.stderr) + self.assertTrue((state / 'pending-plan.json').exists()) + flag.write_text('update') + (self.f.root / 'bin/curl').write_text('#!/bin/sh\necho "must not read a new live set during resume" >&2\nexit 22\n') + done = subprocess.run(['bash', str(TIMER)], capture_output=True, text=True, env=env, timeout=30) + self.assertEqual(done.returncode, 0, done.stderr) + self.assertFalse((state / 'pending-plan.json').exists()) + self.assertTrue(set(doomed).isdisjoint(r.rsplit('/', 1)[1] for r in self.remote_refs(bare, 'refs/tags/evidence'))) + self.assertEqual(len((self.f.root / 'curl.urls').read_text().splitlines()), 1) + def test_bad_retention_and_bad_arguments_are_refused(self): self.world() for bad in ({'soft_records': 5, 'cap_records': 3}, {'main_days': 0}, {'main_days': 31}, {'grace_hours': -1}, {'extra': 1}, @@ -801,6 +1018,42 @@ sys.exit({exit_code}) self.assertEqual(done.returncode, 2) self.assertIn('has no .st/evidence-retention.json', done.stderr) + def test_this_repository_s_cap_holds_the_measured_live_set_and_the_cap_never_changes_the_plan(self): + """The 2026-11-05 dry run's live set (73,782 records, docs/st3/gate-evidence.md) is under this repository's cap_records, so a + nightly run exits 0; one record over the cap exits 3 with the alert. The cap is only the alert: the same world plans the same + deletions whether its cap alerts or not.""" + # arn:lmig:smalltalk:2026-10-27-dot-fractal-config — checkout missing-only bridge. + retention_path = ROOT / '.fractal/evidence-retention.json' + try: + retention_path.lstat() + except FileNotFoundError: + retention_path = ROOT / '.st/evidence-retention.json' + main_file = evidence_gc.load_retention(retention_path.read_text()) + + def status(protected): + args = mock.Mock(resume_state=None, plan_out=None, apply=False, remote=None) + plan = {'protected_records': protected, 'retention': main_file, 'unresolved_heads': [], 'delete': [], 'migrate': [], 'groups': []} + with mock.patch('builtins.print') as printed: + code = evidence_gc.execute_plan(args, plan, None) + return code, json.loads(printed.call_args.args[0]) + + self.assertEqual(status(73782), (0, {'protected_records': 73782, 'retention': main_file, 'unresolved_heads': []})) + code, summary = status(main_file['cap_records'] + 1) + self.assertEqual(code, 3) + self.assertIn('is over the cap', summary['alert']) + old, recent, pr = self.world() + for n in range(2): # two live records: over a cap of 1, not over one of 1000 + self.both(pr[0], pr[1], 28 - n / 10) + self.both(old[0], 'a' * 40, 10) + plans = [] + for cap in (1, 1000): + out = self.f.root / f'plan-{cap}.json' + code, result, _ = self.run_gc('--head', pr[0], '--retention', self.retention(soft_records=1, cap_records=cap), '--plan-out', str(out)) + self.assertEqual(code, 3 if cap == 1 else 0, result) + plans.append(json.loads(out.read_text())['delete']) + self.assertEqual(plans[0], plans[1]) + self.assertTrue(plans[0], 'the world has a dead tree to delete') + def test_the_retention_file_is_read_from_main(self): old, recent, pr = self.world() kept = self.record(old[0], 'a' * 40, 10) @@ -954,6 +1207,19 @@ sys.exit({exit_code}) self.assertEqual((replayed['delete'], replayed['migrate']), (written['delete'], [tuple(m) for m in written['migrate']])) +class PushClassification(unittest.TestCase): + def test_protected_deletion_is_classified_before_long_output_is_truncated(self): + ref = f'refs/tags/evidence/{"a" * 40}/{"b" * 40}' + output = 'remote: GH013: Repository rule violations found for ' + ref + '\n' + 'long remote path ' * 40 + self.assertNotIn('GH013:', output[-300:]) + listed = subprocess.CompletedProcess([], 0, f'{"b" * 40}\t{ref}\n'.encode(), b'') + refused = subprocess.CompletedProcess([], 1, b'', output.encode()) + with mock.patch.object(evidence_gc, 'git', side_effect=[listed, refused]) as git: + failed, deleted, protected, created = evidence_gc.apply_remote(Path('.'), 'origin', [], [ref], [[ref]]) + self.assertEqual((failed, deleted, protected, created), ([], set(), {ref}, True)) + self.assertIn('--atomic', git.call_args.args) + + class PurePlan(unittest.TestCase): """gc_plan on random worlds: a model of the verifier (the newest record decides, a failure shadows a pass dated within the skew after it; an input key's records count for every tree that projects to it) must give the same verdict for every live tree diff --git a/scripts/evidence-gc-timer b/scripts/evidence-gc-timer index 6a34d07..dd5781a 100755 --- a/scripts/evidence-gc-timer +++ b/scripts/evidence-gc-timer @@ -32,6 +32,12 @@ for tool in evidence-gc evidence-verify; do done stamp=$(date -u +%Y%m%dT%H%M%SZ) open_heads=$state/runs/$stamp.open-heads +resume_state=$state/pending-plan.json +resume_args=(--resume-state "$resume_state") +# Resume before reading a new live set; if another run finishes the plan meanwhile, refuse instead of planning without it. +if [ -e "$resume_state" ]; then + resume_args+=(--resume-only) +else python3 -I - "$api/repos/$repository/pulls?state=open&per_page=100" > "$open_heads" <<'PY' import json, subprocess, sys page = 1 @@ -45,13 +51,14 @@ while True: # curl fails on an HTTP error, and then so does this run: without t break page += 1 PY +fi before_args=() if [ -f "$log_timer" ] && [ -f "$log_config/signer.json" ]; then - before_args=(--before-delete "$(python3 -I -c 'import json, sys; print(json.dumps(sys.argv[1:]))' bash "$log_timer" gc --publish --plan)") + before_args=(--before-delete "$(python3 -I -c 'import json, sys; print(json.dumps(sys.argv[1:]))' bash "$log_timer" gc --publish --no-wait --plan)") fi status=0 python3 -I "$tools/evidence-gc" --repo "$repo" --remote origin --main origin/main --fetch --apply --open-heads-from "$open_heads" "${before_args[@]}" \ - --plan-out "$state/runs/$stamp.plan.json" > "$state/runs/$stamp.json" 2> "$state/runs/$stamp.log" || status=$? + "${resume_args[@]}" --plan-out "$state/runs/$stamp.plan.json" > "$state/runs/$stamp.json" 2> "$state/runs/$stamp.log" || status=$? cat "$state/runs/$stamp.log" >&2 if [ -s "$state/runs/$stamp.json" ]; then st documents put "$state/runs/$stamp.json" --as "$document" >&2 || echo "evidence-gc-timer: could not store the summary in $document" >&2 diff --git a/scripts/evidence-log b/scripts/evidence-log index 3fd406e..73e22e8 100755 --- a/scripts/evidence-log +++ b/scripts/evidence-log @@ -12,16 +12,22 @@ GitHub is only the transport. One `pull` (scripts/evidence-log.timer runs it eve 2. accepts each record it has not logged yet if it is well formed and signed by a key that main's `.st/evidence-keys` trusts for the signer it names, and keeps it under `refs/evidence-copy/` (never pruned: GitHub losing it does not lose it here); 3. appends one line per accepted record and check to the log, and one per accepted input-index ref; - 4. commits the log, ALWAYS, even with nothing new: the commit's message carries its time, and a verifier refuses a head older + 4. commits the log every 500 verified refs or 120 seconds of verification, and ALWAYS even with nothing new: each commit's + message carries its commit time, and a verifier refuses a head older than its bound, so a deleted or rewound log costs skips and never makes an old one count; - 5. with --publish, pushes the head to `refs/evidence-log/main` on the remote; with --restore, pushes back any logged record - the remote no longer has (and reports it: `missing`). + 5. with --publish, pushes the head to `refs/evidence-log/main` on the remote, only as a fast-forward of the published head + this run fetched (then the preceding checkpoint), and leased on it: a host whose chain does not continue the published one + (a second publisher, or a copy not seeded from the remote) refuses loudly and publishes nothing, so two publishers never + overwrite each other's heads; + with --restore, pushes back any logged record the remote no longer has (and reports it: `missing`). `gc` appends a GC line for every tree and input key of an evidence-gc plan (`--plan-out`) and drops their copies: run it BEFORE the plan's deletions (evidence-gc --before-delete), so a pull does not restore what GC removed. It also keeps the names of the plan's refs as PENDING DELETIONS: a pull that runs between the GC lines and the deletions does not log them again after the GC -line, which would make their deletion "missing" and a restore push them back. A pending name is dropped once the remote no longer -has it, or after TOMBSTONE_HOURS if GC never deleted it, and is then logged again like any record. With --publish it pushes the -new head at once, so a verifier does not see the deletions as missing until the next pull. +line, which would make their deletion "missing" and a restore push them back. Names absent remotely are dropped from pending at +once. After TOMBSTONE_HOURS, at most TOMBSTONE_RELEASE_REFS names per pull are rechecked, oldest first, after fresh records have +been checkpointed. Successful checks release their pending names only after the checkpoint; rejected names stay held and wait +another TOMBSTONE_HOURS. Nothing uses a pending name as evidence of trust. With --publish it pushes the new head at once, so a +verifier does not see the deletions as missing until the next pull. THE LOG is a chain of commits signed with the log key (`--signer` signs stdin to an SSH signature on stdout, for example ["ssh-keygen", "-Y", "sign", "-n", "git", "-f", KEY], or the same through `st sekrets`, so no seat reads the key). Its tree: @@ -29,7 +35,8 @@ THE LOG is a chain of commits signed with the log key (`--signer` signs stdin to keys/// record one per input-index ref gc