Fix vlt backtest flakes on artifact preflight fetch drops

scripts/backtest-vlt.py

@@ -993,6 +993,7 @@ def __init__(self, ctx, version, mode, shape_name):
         self.envelopes = []
         self.fresh_patched = {}
         self.cli_failures = []
+        self.preflight_failures = []
 
     # CLI -------------------------------------------------------------------
     def cli(self, args, cwd=None):
@@ -1011,7 +1012,11 @@ def patch_run(self, mode, cwd=None):
         cwd = cwd or self.project
         verb = ['get', UUID] if self.spec.get('driver') == 'get' else ['scan']
         code, out, err = self.cli([*verb, '--mode', mode, '--yes', '--cwd', cwd], cwd)
-        return code, parse_envelope(out), err
+        envelope = parse_envelope(out)
+        # A successful run's envelope is not on the row; keep the preflight's
+        # reasons there so the transport retry sees them.
+        self.preflight_failures.extend(envelope_details(envelope, BLOCKED_CODES[0]))
+        return code, envelope, err
 
     def revert(self):
         if self.mode == 'hosted':
@@ -1204,6 +1209,8 @@ def blocked_refusal(self, row, envelopes, reference):
     def finish(self, row, checks, started):
         if self.cli_failures:
             row['cliFailures'] = self.cli_failures
+        if self.preflight_failures:
+            row['preflightFailures'] = self.preflight_failures
         row['failingChecks'] = [k for k, v in checks.items() if v is False]
         row['notEvaluated'] = [k for k, v in checks.items() if v is None]
         row.setdefault('codes', [])
@@ -1469,6 +1476,19 @@ def check_agent(self, row, checks, before_lock, vlt, original):
 # The CLI's own report of a transport failure: a request error, or a 5xx from
 # the patch API (e.g. "API request failed with status 504: error code: 504").
 CLI_TRANSPORT_FAILURE = re.compile(r'error sending request for url \(|API request failed with status 5\d\d\b')
+# The hosted vlt artifact preflight's fetch of the artifact failed in transit
+# (vlt_preflight.rs `ArtifactProbe::failure`: "fetch error <e>", "http <5xx>").
+# The CLI withholds the dep, writes nothing and exits 0, so the cell reads as
+# a safe refusal although the cell's own probe of the same URL verified. The
+# fetch errors that are deterministic (an over-cap body, a non-http URL) and
+# every other reason (content-encoding, sha512 mismatch, 4xx) stay final.
+PREFLIGHT_TRANSPORT_FAILURE = re.compile(
+    r'vlt would fail to verify \S+: (?:http 5\d\d\b'
+    r'|fetch error (?!refusing a non-http)(?![^;]*too large))')
+# vlt's own registry fetch dropped (Node's errno codes for a connect or read
+# timeout, a reset connection and a temporary DNS failure). ENOTFOUND and
+# ECONNREFUSED stay final: a wrongly rewritten URL fails that way too.
+VLT_TRANSPORT_FAILURE = re.compile(r'\b(?:ETIMEDOUT|ECONNRESET|EAI_AGAIN)\b|socket hang up')
 
 
 def transient(row):
@@ -1481,7 +1501,9 @@ def transient(row):
     if probe and (probe.get('curlExit') or (probe.get('status') or 0) >= 500
                   or probe.get('status') is None):
         return True
-    return bool(CLI_TRANSPORT_FAILURE.search(json.dumps(row)))
+    text = json.dumps(row)
+    return any(p.search(text) for p in (CLI_TRANSPORT_FAILURE, PREFLIGHT_TRANSPORT_FAILURE,
+                                         VLT_TRANSPORT_FAILURE))
 
 
 def run_with_retries(cell_factory, job, attempts=3):

scripts/tests/test_backtest_harnesses.py

@@ -914,6 +914,50 @@ def test_only_transport_failures_retry(self):
         self.assertFalse(vlt.transient({'serveProbe': {'curlExit': 0, 'status': 200},
                                         'failingChecks': ['freshCi']}))
 
+    def test_a_preflight_transport_failure_retries(self):
+        def refusal(reason):
+            return {'serveProbe': {'curlExit': 0, 'status': 200}, 'safeRefusal': True,
+                    'preflightFailures': [f'vlt would fail to verify https://h/<redacted>/u/a.tgz: '
+                                          f'{reason}; nothing was written for pkg:npm/a@1']}
+        for reason in ('http 502', 'http 504', 'fetch error error sending request',
+                       'fetch error no response within 60s', 'fetch error no response',
+                       'fetch error hosted artifact body not received within 300s'):
+            with self.subTest(reason=reason):
+                self.assertTrue(vlt.transient(refusal(reason)))
+        for reason in ('http 404', 'http 403', 'content-encoding gzip', 'sha512 mismatch',
+                       'offline', 'fetch error refusing a non-http(s) artifact URL',
+                       'fetch error hosted artifact too large (5 bytes > 4)'):
+            with self.subTest(reason=reason):
+                self.assertFalse(vlt.transient(refusal(reason)))
+
+    def test_a_vlt_network_drop_retries(self):
+        for err in ("vlt install exited 1: Timeout { code: 'ETIMEDOUT', syscall: 'connect' }",
+                    'vlt install exited 1: read ECONNRESET', 'getaddrinfo EAI_AGAIN registry',
+                    'vlt install exited 1: socket hang up'):
+            with self.subTest(err=err):
+                self.assertTrue(vlt.transient({'error': f'RuntimeError: {err}'}))
+        for err in ('getaddrinfo ENOTFOUND h', 'connect ECONNREFUSED 127.0.0.1:1', 'EINTEGRITY'):
+            with self.subTest(err=err):
+                self.assertFalse(vlt.transient({'error': f'RuntimeError: {err}'}))
+
+    def test_a_successful_runs_preflight_reasons_reach_the_row(self):
+        with tempfile.TemporaryDirectory() as temp:
+            cell = vlt.Cell({'out': Path(temp), 'record': {}, 'cli': 'socket-patch',
+                             'cli_env': {}}, '1.2.0', 'hosted', 'direct')
+            cell.project.mkdir(parents=True)
+            row = dict(cell=cell.name, expectedVerdict='patched', passed=False, checks={},
+                       safeRefusal=True)
+            detail = 'vlt would fail to verify https://h/a.tgz: http 503; nothing was written for x'
+            envelope = {'status': 'success', 'redirect': {'warnings': [
+                {'code': 'redirect_vlt_artifact_unverifiable', 'detail': detail}]}}
+            with patch.object(vlt, 'run', return_value=(0, json.dumps(envelope), '')):
+                cell.patch_run('hosted')
+            with patch('sys.stdout'):
+                cell.finish(row, row['checks'], vlt.time.time())
+            captured = json.loads((cell.case / 'result.json').read_text())
+            self.assertEqual(captured['preflightFailures'], [detail])
+            self.assertTrue(vlt.transient(captured))
+
     def test_a_transport_failure_reruns_from_scratch(self):
         with tempfile.TemporaryDirectory() as temp:
             rows = [{'cell': 'c', 'error': 'error sending request for url (https://x)'},