Prevent production disk exhaustion from retained deployments

code/production.py

@@ -163,6 +163,85 @@ def environment(self, code, docroot):
                    NO_PROXY='localhost,127.0.0.1,::1', no_proxy='localhost,127.0.0.1,::1')
         return env
 
+    def storage_status(self):
+        disk = shutil.disk_usage(self.root)
+        fs = os.statvfs(self.root)
+        minimum = int(self.config.get('min_free_bytes', 5 * 1024**3))
+        used_percent = round(100 * disk.used / disk.total, 1)
+        inode_percent = round(100 * (1 - fs.f_favail / fs.f_files), 1) if fs.f_files else 0
+        return {'total_bytes': disk.total, 'free_bytes': disk.free,
+                'used_percent': used_percent, 'inode_used_percent': inode_percent,
+                'min_free_bytes': minimum,
+                'healthy': disk.free >= minimum and used_percent < 80 and inode_percent < 80}
+
+    def storage(self):
+        status = self.storage_status()
+        print(json.dumps(status, indent=2))
+        if not status['healthy']:
+            raise SystemExit(1)
+
+    def require_space(self, publish_data=False):
+        # Reserve headroom in addition to the candidate's code, website and data copy.
+        def size(path):
+            return sum(p.stat().st_size for p in path.rglob('*')
+                       if not p.is_symlink() and p.is_file()) if path.exists() else 0
+        needed = 2 * 1024**3 + 2 * (size(self.current / 'code') + size(self.current / 'site'))
+        if publish_data:
+            needed += size(self.state)
+        status = self.storage_status()
+        needed = max(needed, status['min_free_bytes'])
+        if status['free_bytes'] < needed or status['inode_used_percent'] >= 90:
+            raise RuntimeError(f'Insufficient deployment storage: need {needed} free bytes; '
+                               f"have {status['free_bytes']}; inode usage {status['inode_used_percent']}%")
+
+    def prune_locked(self):
+        """Caller holds operation.lock. Only disposable releases/sites are collected."""
+        releases = self.root / 'releases'
+        sites = self.root / 'sites'
+        if releases.is_symlink() or sites.is_symlink():
+            raise RuntimeError('Retention roots must not be symlinks')
+        protected = set()
+        for name in ('current', 'previous'):
+            pointer = self.root / name
+            if pointer.exists() or pointer.is_symlink():
+                target = pointer.resolve(strict=True)
+                if target.parent != releases or not target.is_dir():
+                    raise RuntimeError('Release pointer escapes retention root')
+                protected.add(target)
+        candidates = sorted((p for p in releases.iterdir()
+                             if not p.is_symlink() and p.is_dir()
+                             and re.fullmatch(r'\d{8}T\d{6}-[0-9a-f]{12}', p.name)), reverse=True)
+        count = max(2, int(self.config.get('keep_releases', 3)))
+        retained = protected | set([p for p in candidates if p not in protected][:max(0, count - len(protected))])
+        # Keep sites referenced by every retained or unrecognized release, including job publications.
+        removed_releases = set(candidates) - retained
+        referenced = {sites / p.name for p in retained}
+        for r in releases.iterdir():
+            if r in removed_releases:
+                continue
+            site = r / 'site'
+            if site.exists() or site.is_symlink():
+                target = site.resolve(strict=True)
+                if target.parent == sites:
+                    referenced.add(target)
+        disposable_sites = sorted((p for p in sites.iterdir()
+                                   if not p.is_symlink() and p.is_dir() and p not in referenced
+                                   and re.fullmatch(r'(?:\d{8}T\d{6}-[0-9a-f]{12}|job-[a-zA-Z0-9_-]+)', p.name)),
+                                  key=lambda p: p.stat().st_mtime, reverse=True)
+        # Preserve the newest unreferenced job publication for diagnosis; receipts/logs never expire here.
+        orphan_jobs = [p for p in disposable_sites if p.name.startswith('job-')]
+        keep_jobs = set(orphan_jobs[:1])
+        removed = []
+        for path in sorted(removed_releases) + [p for p in disposable_sites if p not in keep_jobs]:
+            shutil.rmtree(path)
+            removed.append(str(path.relative_to(self.root)))
+        return removed
+
+    def prune(self):
+        with self.lock():
+            removed = self.prune_locked()
+        print(json.dumps({'removed': removed, 'storage': self.storage_status()}, indent=2))
+
     def merged_sha(self, requested=None):
         if not self.mirror.exists():
             run(['git', 'clone', '--bare', self.config['repository'], self.mirror])
@@ -290,6 +369,8 @@ def deploy(self, requested=None, keep_publication=False, publish_data=False):
         if keep_publication and publish_data:
             raise ValueError('Choose either preserved publication or new data publication')
         with self.lock():
+            self.prune_locked()
+            self.require_space(publish_data=publish_data)
             sha = self.merged_sha(requested)
             name = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S') + '-' + sha[:12]
             release = self.root / 'releases' / name
@@ -330,6 +411,10 @@ def deploy(self, requested=None, keep_publication=False, publish_data=False):
                 write_json(release / 'release.json', metadata)
                 self.attach_state(code)
                 self.activate(release)
+                try:
+                    self.prune_locked()
+                except (OSError, RuntimeError) as exc:
+                    print(f'Retention needs attention: {exc}', file=sys.stderr)
                 print(json.dumps({'deployed': name, 'commit': sha}))
             except Exception:
                 # Keep the candidate and logs for diagnosis; never switch on failure.
@@ -350,6 +435,8 @@ def execute_job(self, key, job):
             receipt = self.root / 'runs' / (key + '.json')
             if receipt.exists():
                 return False  # Includes interrupted/failed runs: no surprise paid retries.
+            if job['kind'] != 'smoke':
+                self.require_space(publish_data=job['kind'] == 'forecast')
             release = self.current.resolve(strict=True)
             code = release / 'code'
             started = {'key': key, 'job': job['id'], 'release': release.name, 'status': 'running', 'started': datetime.now(timezone.utc).isoformat()}
@@ -400,7 +487,8 @@ def tick(self):
             self.execute_job(key, job)
 
     def status(self):
-        print(json.dumps({'release': json.loads((self.current / 'release.json').read_text()) if self.current.exists() else None,
+        print(json.dumps({'storage': self.storage_status(),
+                          'release': json.loads((self.current / 'release.json').read_text()) if self.current.exists() else None,
                           'jobs_enabled': (self.root / 'jobs-enabled').exists(),
                           'jobs': json.loads((self.current / 'code/deploy/jobs.json').read_text()) if self.current.exists() else [],
                           'recent_runs': [json.loads(p.read_text()) for p in sorted((self.root / 'runs').glob('*.json'))[-10:]]}, indent=2))
@@ -428,7 +516,7 @@ def log_message(self, *args):
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument('--config', default=os.environ.get('AIRO_RUNTIME_CONFIG', str(Path.home() / '.config/airo/runtime.json')))
-    parser.add_argument('command', choices=['deploy', 'rollback', 'status', 'tick', 'serve', 'smoke'])
+    parser.add_argument('command', choices=['deploy', 'rollback', 'status', 'tick', 'serve', 'smoke', 'prune', 'storage'])
     parser.add_argument('--commit')
     publication = parser.add_mutually_exclusive_group()
     publication.add_argument('--keep-publication', action='store_true', help='Retain the entire website while deploying code and schedules')

deploy/README.md

@@ -1,4 +1,4 @@
-# Production releases in the development cell
+# Production releases
 
 On a configured cell, say **deploy** after merging the work. The agent runs
 `airo deploy`, then `airo status` and checks the configured public URL. Deployment
@@ -62,8 +62,29 @@ including licensed fonts. Repository results seed missing files, but never
 replace accrued data. Corrections to existing production data require a separate,
 reviewed data migration. Dependency versions resolved by pip are recorded per
 release; npm uses its checked-in lockfile. Rollback reuses installed environments.
-Retained releases and data need disk monitoring and backups; automatic deletion
-is deliberately absent.
+Development and production should use separate VMs/filesystems so development
+cannot exhaust production storage. Separate directories on the same filesystem
+do not provide that isolation. Keep development checkouts and coding credentials
+off the production VM; copy only the runtime, forecasting credentials and data.
+
+Deployment prunes disposable releases before building and after activation.
+The current and previous releases are always preserved, plus recent candidates
+up to `keep_releases` (default 3, minimum 2). Sites referenced by retained releases
+are preserved, including publications produced by jobs. The latest unreferenced
+job publication is also retained for diagnosis. Unknown directory names and
+symlinks are left alone. `state/`, `home/`, run receipts and logs are never pruned.
+Run `airo prune` to apply the same retention policy explicitly. Keep backups of
+accrued data independently; release retention is not a backup policy.
+
+Before installing dependencies or starting paid jobs, the runtime checks free
+bytes and inodes. Deployment requires at least `min_free_bytes` (default 5 GiB),
+or estimated staging space plus a 2 GiB reserve, whichever is larger. This is a
+preflight estimate, not a quota; unexpected dependency or forecast growth can
+still exceed it. `airo status` includes filesystem usage. `airo storage` prints
+usage and exits nonzero below the free-space floor or at 80% byte/inode usage.
+Run it from a five-minute systemd timer and monitor failures. Monitor development
+storage independently on its own VM; a healthy production disk says nothing
+about the development disk.
 
 ## Scheduling
 

tests/test_production.py

@@ -14,7 +14,9 @@
 
 
 @pytest.fixture
-def runtime(tmp_path):
+def runtime(tmp_path, monkeypatch):
+    monkeypatch.setattr(p.shutil, 'disk_usage', lambda _: type('Disk', (), {
+        'total': 50*1024**3, 'used': 10*1024**3, 'free': 40*1024**3})())
     return p.Runtime({'root': str(tmp_path), 'path': '/usr/bin:/bin', 'socket': str(tmp_path / 'site.sock')})
 
 
@@ -345,3 +347,74 @@ def publish(argv, **kw):
     assert meta['data_snapshot']['sha256'] == p.data_snapshot(old / 'site')['sha256']
     assert meta['data_snapshot']['source_release'] == 'job-fixture-new-data'
     assert b'old dot layout' in (saved / 'index.html').read_bytes()
+
+
+def test_retention_keeps_rollback_job_site_and_accrued_data(runtime):
+    names = [f'20300101T00000{i}-' + 'a' * 12 for i in range(5)]
+    versions = [release(runtime, name) for name in names]
+    for r in versions:
+        site = runtime.root / 'sites' / r.name
+        site.mkdir()
+        (site / 'index.html').write_text(r.name)
+        p.link(site, r / 'site')
+    p.link(versions[1], runtime.current)
+    p.link(versions[0], runtime.root / 'previous')
+    job = runtime.root / 'sites/job-active'
+    job.mkdir()
+    p.link(job, versions[1] / 'site')
+    runtime.state.mkdir()
+    (runtime.state / 'raw').write_text('do not remove')
+    (runtime.root / 'runs/job.json').write_text('{}')
+    (runtime.root / 'logs/job.log').write_text('diagnostic')
+    # A failed candidate without its final site symlink retains its partial site.
+    (versions[-1] / 'site').unlink()
+    runtime.prune()
+    assert all(r.exists() for r in [versions[0], versions[1], versions[4]])
+    assert not versions[2].exists() and not versions[3].exists()
+    assert job.exists()
+    assert (runtime.root / 'sites' / names[-1]).exists()
+    assert not (runtime.root / 'sites' / names[2]).exists()
+    assert (runtime.state / 'raw').read_text() == 'do not remove'
+    assert (runtime.root / 'runs/job.json').exists()
+    assert (runtime.root / 'logs/job.log').exists()
+
+
+def test_retention_never_follows_external_symlinks(runtime, tmp_path):
+    external = tmp_path / 'external'
+    external.mkdir()
+    (external / 'precious').write_text('keep')
+    (runtime.root / 'releases/20300101T000000-aaaaaaaaaaaa').symlink_to(external)
+    (runtime.root / 'sites/job-external').symlink_to(external)
+    runtime.prune()
+    assert (external / 'precious').read_text() == 'keep'
+    p.link(external, runtime.current)
+    with pytest.raises(RuntimeError, match='escapes'):
+        runtime.prune()
+
+
+def test_low_disk_preflight_does_not_fetch_install_or_change_current(runtime, monkeypatch):
+    old = release(runtime, 'old')
+    p.link(old, runtime.current)
+    monkeypatch.setattr(p.shutil, 'disk_usage', lambda _: type('Disk', (), {'total': 10*1024**3, 'used': 9*1024**3, 'free': 1024**3})())
+    monkeypatch.setattr(runtime, 'merged_sha', lambda _: pytest.fail('must not fetch'))
+    with pytest.raises(RuntimeError, match='Insufficient deployment storage'):
+        runtime.deploy()
+    assert runtime.current.resolve() == old
+    assert list((runtime.root / 'releases').iterdir()) == [old]
+    assert not runtime.storage_status()['healthy']
+
+
+def test_low_disk_blocks_paid_work_before_receipt_or_provider_call(runtime, monkeypatch):
+    def no_space(**kw):
+        raise RuntimeError('Insufficient deployment storage')
+    monkeypatch.setattr(runtime, 'require_space', no_space)
+    monkeypatch.setattr(p, 'run', lambda *a, **kw: pytest.fail('must not run paid work'))
+    with pytest.raises(RuntimeError, match='Insufficient'):
+        runtime.execute_job('paid', {'id': 'forecast', 'kind': 'forecast'})
+    assert not (runtime.root / 'runs/paid.json').exists()
+
+
+def test_inode_exhaustion_blocks_deployment(runtime, monkeypatch):
+    monkeypatch.setattr(runtime, 'storage_status', lambda: {'free_bytes': 100*1024**3, 'min_free_bytes': 5*1024**3, 'inode_used_percent': 99})
+    with pytest.raises(RuntimeError, match='inode usage 99'):
+        runtime.require_space()