diff --git a/Changelog.md b/Changelog.md index 42784ba6..eed62375 100644 --- a/Changelog.md +++ b/Changelog.md @@ -4,6 +4,16 @@ All notable changes are recorded here, newest release first. Versions follow [Semantic Versioning](https://semver.org/). Dates use America/Los_Angeles. The history before this initial versioned snapshot remains in Git. +## [1.0.3] - 2026-10-17 + +### Fixed + +- Release video temporary files before returning pipeline slots, including + oversized rejections and interrupted uploads. A sender waiting on later API + work can no longer retain old media outside the two-pipeline bound. +- Verify immediate cleanup with retained senders and inspect audio/video tracks + in the file delivered by Telegram after local upload files have been removed. + ## [1.0.2] - 2026-10-16 ### Fixed diff --git a/README.md b/README.md index b7a891e8..062038f2 100644 --- a/README.md +++ b/README.md @@ -97,7 +97,8 @@ Tests The consumer keeps at most `consumer.max_pending` submitted jobs (default: pool size). Two video jobs may download, mux, and upload concurrently. Uploads stream -from disk, and each sender removes its private temporary files on close. Media +from disk, and every attempt removes its private files before returning a +pipeline slot, including oversized rejections and interrupted uploads. Media HTTP sessions close after each job; verified TLS contexts are shared across connections. On glibc systems, `pool.malloc_arenas` defaults to four to reduce memory retained by native diff --git a/VERSION b/VERSION index 6d7de6e6..21e8796a 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.0.2 +1.0.3 diff --git a/reddit2telegram/utils/__init__.py b/reddit2telegram/utils/__init__.py index 0638e57e..e980bb50 100644 --- a/reddit2telegram/utils/__init__.py +++ b/reddit2telegram/utils/__init__.py @@ -509,6 +509,7 @@ def _run_async(self, coro): def close(self): loop = getattr(self, '_loop', None) if loop is None or loop.is_closed(): + self._cleanup_temp_files() return request = getattr(self, '_telegram_request', None) try: @@ -523,9 +524,13 @@ def close(self): logging.exception('Failed to shutdown Telegram HTTP client cleanly.') finally: loop.close() - files = getattr(self, '_temp_files', None) - if files is not None: - files.cleanup() + self._cleanup_temp_files() + + def _cleanup_temp_files(self): + files = getattr(self, '_temp_files', None) + if files is not None: + files.cleanup() + self._temp_files = None def __enter__(self): return self @@ -722,7 +727,13 @@ def _get_dash_audio_url(self, dash_url): def send_video(self, url, text, parse_mode=None, dash_url=None): with _VIDEO_UPLOAD_SLOTS: - return self._send_video(url, text, parse_mode, dash_url) + try: + return self._send_video(url, text, parse_mode, dash_url) + finally: + # A rejected candidate can be followed by a long API retry on + # the same sender. Release its file cache before the next video + # takes this slot, on success, rejection, and upload failure. + self._cleanup_temp_files() def _send_video(self, url, text, parse_mode=None, dash_url=None): filename = self._get_file_name('.mp4') diff --git a/tests/test_live_resources.py b/tests/test_live_resources.py index ed1b1b05..4b995284 100644 --- a/tests/test_live_resources.py +++ b/tests/test_live_resources.py @@ -90,6 +90,7 @@ async def send_video(self, **kwargs): class RecordingSender(utils.Reddit2TelegramSender): def __init__(self, config, root, state): self.messages = [] + self.media_paths = set() self.root = root / uuid.uuid4().hex self.root.mkdir() super().__init__(TEST_CHANNEL, config) @@ -99,8 +100,10 @@ def __init__(self, config, root, state): ) def _get_file_name(self, ext='file'): - # Isolate files from production's cleaner and concurrent test senders. - return str(self.root / ('video' + ext)) + # Exercise production's actual private directory and cleanup lifecycle. + filename = super()._get_file_name(ext) + self.media_paths.add(Path(filename)) + return filename def _run_async(self, coro): result = super()._run_async(coro) @@ -238,9 +241,16 @@ def test_real_video_download_audio_merge_and_delivery(self): message = sender.messages[-1] self.assert_delivery(message) self.assertIsNotNone(message.video) + self.assertTrue(sender.media_paths) + self.assertTrue(all(not path.parent.exists() for path in sender.media_paths), + 'A completed upload retained its production temporary directory') + # Inspect what Telegram actually received after local media cleanup. + delivered = sender.root / 'delivered.mp4' + remote = sender._run_async(sender.telegram_bot.get_file(message.video.file_id)) + sender._run_async(remote.download_to_drive(custom_path=delivered)) probe = subprocess.run([ 'ffprobe', '-v', 'error', '-show_entries', 'stream=codec_type', '-of', 'json', - sender._get_file_name('.1.mp4'), + str(delivered), ], check=True, capture_output=True, text=True, timeout=10) self.assertEqual({stream['codec_type'] for stream in json.loads(probe.stdout)['streams']}, {'audio', 'video'}) diff --git a/tests/test_memory_budget.py b/tests/test_memory_budget.py index 17275414..31e5520a 100644 --- a/tests/test_memory_budget.py +++ b/tests/test_memory_budget.py @@ -121,6 +121,34 @@ def stream(size): response.__exit__.assert_called_once() +def test_rejected_videos_leave_no_files_while_senders_remain_alive(monkeypatch, tmp_path): + monkeypatch.setattr(utils, 'TEMP_FOLDER', str(tmp_path)) + + def download(url, filename): + size = (2 if filename.endswith('.aac') else 49) * 1024 * 1024 + with open(filename, 'wb') as handle: + handle.truncate(size) # Sparse files exercise limits without allocating GB of RAM/disk. + return True + + def mux(args, **kwargs): + with open(args[-1], 'wb') as handle: + handle.truncate(51 * 1024 * 1024) + return subprocess.CompletedProcess(args, 0) + + monkeypatch.setattr(utils, 'download_file', download) + monkeypatch.setattr(utils.subprocess, 'run', mux) + senders = [object.__new__(utils.Reddit2TelegramSender) for _ in range(12)] + for i, sender in enumerate(senders): + sender.t_channel = f'@test{i}' + with ThreadPoolExecutor(12) as workers: + results = list(workers.map(lambda sender: sender.send_video('https://local/video', 'test'), senders)) + assert all(result == utils.SupplyResult.DO_NOT_WANT_THIS_SUBMISSION for result in results) + # A rejected candidate may be followed by a long Reddit retry on the same + # sender. Its files must be gone before another video gets the pipeline slot. + leftover = sum(path.stat().st_size for path in tmp_path.rglob('*') if path.is_file()) + assert leftover == 0, f'Rejected videos retained {leftover / 2**20:.0f} MiB of media files' + + def test_large_video_reaches_http_backend_without_full_file_buffer(monkeypatch, tmp_path): size = 49 * 1024 * 1024 fixture = tmp_path / 'fixture.mp4' @@ -154,11 +182,13 @@ def log_message(self, *args): monkeypatch.setattr(utils, 'short_sleep', lambda: None) monkeypatch.setattr(utils.Reddit2TelegramSender, '_make_mongo_connections', lambda self: None) monkeypatch.setattr(utils, 'TEMP_FOLDER', str(tmp_path)) + downloaded = [] def download(url, filename): if filename.endswith('.aac'): return False shutil.copyfile(fixture, filename) + downloaded.append(Path(filename)) return True monkeypatch.setattr(utils, 'download_file', download) @@ -173,8 +203,8 @@ def download(url, filename): peak = tracemalloc.get_traced_memory()[1] finally: tracemalloc.stop() - directory = Path(sender._get_file_name('.mp4')).parent - assert directory.exists() + directory = downloaded[0].parent + assert not directory.exists(), 'Finished upload retained media until sender close' assert received and received[0] > size assert peak < 12 * 1024 * 1024, f'49 MiB upload allocated {peak / 2**20:.1f} MiB' assert not directory.exists(), 'Closed sender retained downloaded media' @@ -184,6 +214,29 @@ def download(url, filename): thread.join(2) +def test_failed_upload_releases_files_before_sender_close(monkeypatch, tmp_path): + monkeypatch.setattr(utils, 'TEMP_FOLDER', str(tmp_path)) + + def download(url, filename): + if filename.endswith('.aac'): + return False + Path(filename).write_bytes(b'video') + return True + + async def upload(**kwargs): + assert not kwargs['video'].input_file_content.closed + raise RuntimeError('upload interrupted') + + monkeypatch.setattr(utils, 'download_file', download) + sender = object.__new__(utils.Reddit2TelegramSender) + sender.t_channel = '@test' + sender.telegram_bot = mock.Mock(send_video=upload) + sender._run_async = asyncio.run + with pytest.raises(RuntimeError, match='upload interrupted'): + sender.send_video('https://local/video', 'test') + assert not list(tmp_path.iterdir()), 'Failed upload retained its temporary directory' + + def test_maintenance_reclaims_dead_job_cycles_without_waiting_for_automatic_gc(monkeypatch): class Job: pass